acpmux: carried turns arm their own cap, hold their own output and count lane reports

cmux-tui/crates/acpmux/src/agent.rs

@@ -832,8 +832,8 @@ impl ChildAgent {
                 // open; watch it so its exit is reported regardless.
                 let mut leader_check = tokio::time::interval(LEADER_CHECK);
                 let mut leader_gone: Option<tokio::time::Instant> = None;
-                // A carry's hold was waited out (`claude_stdio/carry.rs`): once per hold.
-                let mut held = false;
+                // The carry whose hold was waited out (`claude_stdio/carry.rs`): once per carry.
+                let mut held: Option<u64> = None;
                 loop {
                     let next = tokio::select! {
                         next = lines.next_line() => next,
@@ -990,14 +990,16 @@ impl ChildAgent {
                     // Claude Code ended its turn and runs on with steers it had not read: its next
                     // line waits until the hub logged that turn's end and the next one's start
                     // (the next turn's prompt), bounded (`claude_stdio/carry.rs`).
-                    let holding = agent_for_exit.translator.as_ref().filter(|tr| tr.holding());
+                    // Each carry holds once, also one made while an earlier one still holds.
+                    let tr = agent_for_exit.translator.as_ref();
+                    let holding = tr.and_then(|tr| tr.holding().map(|carry| (tr, carry)));
                     match holding {
-                        Some(tr) if !held => {
+                        Some((tr, carry)) if held != Some(carry) => {
                             tr.wait_unheld(crate::claude_stdio::CARRY_HOLD_MAX).await;
-                            held = true;
+                            held = Some(carry);
                         }
                         Some(_) => {}
-                        None => held = false,
+                        None => held = None,
                     }
                 }
                 // Claude's background shells ended with it (cx-ncc.84).

cmux-tui/crates/acpmux/src/agent_host/host.rs

@@ -195,7 +195,7 @@ enum Event {
     DrainOver(u64),
     /// [`POOLED_OWNER_WAIT`] passed since the host started serving.
     OwnerWaitOver,
-    /// The output held after a carry (its hold's number) is read now, its prompt or not
+    /// The output held after a carry (its carry's number) is read now, its prompt or not
     /// (`claude_stdio/carry.rs`).
     HoldOver(u64),
 }
@@ -282,8 +282,11 @@ struct State {
     /// Claude's output lines after a carry, read once the next turn's prompt came
     /// (`claude_stdio/carry.rs` `holding`), oldest first.
     held: VecDeque<String>,
-    /// The current hold's number, and whether its bound passed.
+    /// The carry that holds the output now (its number, `claude_stdio/carry.rs` `holding`), and
+    /// whether its bound passed: each carry holds once, with a bound of its own.
     hold: (u64, bool),
+    /// The output ended: what a carry held is read, and no carry holds more.
+    read_all: bool,
 }
 
 /// The debug builds' hello-stall seam's variable ([`claim_hello_stall`]):
@@ -492,6 +495,7 @@ impl Started {
             stall_hello: claim_hello_stall(),
             held: VecDeque::new(),
             hold: (0, false),
+            read_all: false,
         };
         // The app-server's initialize goes out at spawn, before the hub's.
         if let Some(cx) = state.codex.clone() {
@@ -535,7 +539,7 @@ impl Started {
                 Event::Stdout(None) => {
                     permit_out = false;
                     // What a carry held is read before the output's end.
-                    state.hold.1 = true;
+                    state.read_all = true;
                     state.read_held(&events_tx).await;
                     if state.exit_h.is_none() {
                         state.end_background_shells().await;
@@ -755,30 +759,31 @@ impl State {
     }
 
     /// Reads the held output lines in order, until a carry holds the rest (`claude_stdio/carry.rs`
-    /// `holding`): a hold starts its bound ([`crate::claude_stdio::CARRY_HOLD_MAX`]).
+    /// `holding`): each carry's hold starts its own bound ([`crate::claude_stdio::CARRY_HOLD_MAX`]),
+    /// also one made while an earlier carry still holds.
     async fn read_held(&mut self, events: &mpsc::Sender<Event>) {
-        loop {
-            let holding = self.translator.as_ref().is_some_and(|tr| tr.holding());
-            if !holding {
-                self.hold.1 = false;
-            } else if !self.hold.1 {
-                return;
-            }
+        while !self.holds(events) {
             let Some(line) = self.held.pop_front() else { return };
             self.on_stdout(line).await;
-            let held_now = self.translator.as_ref().is_some_and(|tr| tr.holding());
-            if held_now && !holding {
-                self.hold = (self.hold.0 + 1, false);
-                let (events, hold) = (events.clone(), self.hold.0);
-                // task-owner: one bounded wait; the runtime ends it with the host.
-                tokio::spawn(async move {
-                    tokio::time::sleep(crate::claude_stdio::CARRY_HOLD_MAX).await;
-                    let _ = events.send(Event::HoldOver(hold)).await;
-                });
-            }
         }
     }
 
+    /// Whether a carry holds the output now: one that holds anew starts its bound.
+    fn holds(&mut self, events: &mpsc::Sender<Event>) -> bool {
+        let carry = self.translator.as_ref().and_then(|tr| tr.holding());
+        let Some(carry) = carry.filter(|_| !self.read_all) else { return false };
+        if self.hold.0 != carry {
+            self.hold = (carry, false);
+            let events = events.clone();
+            // task-owner: one bounded wait; the runtime ends it with the host.
+            tokio::spawn(async move {
+                tokio::time::sleep(crate::claude_stdio::CARRY_HOLD_MAX).await;
+                let _ = events.send(Event::HoldOver(carry)).await;
+            });
+        }
+        !self.hold.1
+    }
+
     async fn on_stdout(&mut self, line: String) {
         if line.trim().is_empty() {
             return;

cmux-tui/crates/acpmux/src/claude_stdio/carry.rs

@@ -138,7 +138,7 @@ impl Translator {
         let mut carry = self.carry.lock().await;
         if let Some(answer) = carry.answers.pop_front() {
             carry.owed = carry.owed.saturating_sub(1);
-            self.carry_owed.send_replace(carry.owed);
+            self.carry_owed.send_replace((carry.owed, carry.seq));
             return Outbound::Reply(match answer {
                 Ok(v) => Message::ok(id.clone(), v),
                 Err(e) => Message::err(id.clone(), e),
@@ -150,19 +150,22 @@ impl Translator {
         carry.owed -= 1;
         // Taken while the carry lock is held: the result that answers it takes that lock first.
         self.pending.lock().await.insert(id.to_string(), Pending::Prompt);
-        self.carry_owed.send_replace(carry.owed);
+        self.carry_owed.send_replace((carry.owed, carry.seq));
         Outbound::Lines(Vec::new())
     }
 
-    /// The output after a carry is held: the next turn's prompt has not come yet.
-    pub fn holding(&self) -> bool {
-        *self.carry_owed.borrow() > 0
+    /// The output after a carry is held while a carried turn's prompt has not come yet: the
+    /// newest carry's number, so a reader gives each carry a hold of its own, also one made
+    /// while an earlier carry's prompt is still to come.
+    pub fn holding(&self) -> Option<u64> {
+        let (owed, seq) = *self.carry_owed.borrow();
+        (owed > 0).then_some(seq)
     }
 
     /// Waits, at most `max`, until no carried turn waits for its prompt.
     pub async fn wait_unheld(&self, max: std::time::Duration) {
         let mut owed = self.carry_owed.subscribe();
-        let _ = tokio::time::timeout(max, owed.wait_for(|n| *n == 0)).await;
+        let _ = tokio::time::timeout(max, owed.wait_for(|(n, _)| *n == 0)).await;
     }
 
     /// A result carried (`carried`) or not: the carry state after it, under its lock.
@@ -172,7 +175,7 @@ impl Translator {
         } else {
             carry.ended();
         }
-        self.carry_owed.send_replace(carry.owed);
+        self.carry_owed.send_replace((carry.owed, carry.seq));
     }
 
     /// A new prompt of the hub's: no carried turn is owed any more.
@@ -182,6 +185,6 @@ impl Translator {
         carry.current = None;
         carry.answers.clear();
         carry.steers.clear();
-        self.carry_owed.send_replace(0);
+        self.carry_owed.send_replace((0, carry.seq));
     }
 }

cmux-tui/crates/acpmux/src/claude_stdio/carry_tests.rs

@@ -196,11 +196,11 @@ async fn a_carried_turn_whose_result_was_lost_fails_its_prompt() {
     let t = Translator::new("acp-1".into(), "default", "haiku", "default");
     let line = turn_with_a_steer(&t).await;
     assert!(answer(&t.inbound(&done("first")).await, 1).is_some());
-    assert!(t.holding(), "Claude Code's output waits for the carried turn's prompt");
+    assert!(t.holding().is_some(), "Claude Code's output waits for the carried turn's prompt");
     t.inbound(&echo(&line)).await;
     t.fail_pending("agent answer was over the size limit").await;
     let Outbound::Reply(reply) = t.outbound(&carried(3)).await else { panic!("it waited") };
     let Message::Response { error, .. } = &reply else { panic!("{reply:?}") };
     assert!(error.as_ref().is_some_and(|e| e.message.contains("size limit")), "{reply:?}");
-    assert!(!t.holding());
+    assert!(t.holding().is_none());
 }

cmux-tui/crates/acpmux/src/claude_stdio/mod.rs

@@ -447,8 +447,9 @@ pub struct Translator {
     /// Turns Claude Code runs on steers it read after its turn ended (`carry.rs`). Taken before
     /// `pending`, and after `batches`, `steers` and `steer_lines`.
     carry: Mutex<carry::Carry>,
-    /// Carried turns whose prompt has not come yet (`carry.rs` `holding`).
-    carry_owed: tokio::sync::watch::Sender<usize>,
+    /// Carried turns whose prompt has not come yet, and the newest carry's number (`carry.rs`
+    /// `holding`).
+    carry_owed: tokio::sync::watch::Sender<(usize, u64)>,
     /// Text streamed so far in the current turn, to build the prompt result.
     pub cancelled: AtomicBool,
     pub slash_commands: Mutex<Vec<Value>>,
@@ -603,7 +604,7 @@ impl Translator {
             steer_lines: Mutex::new(HashMap::new()),
             batches: Mutex::new(Default::default()),
             carry: Mutex::new(carry::Carry::new()),
-            carry_owed: tokio::sync::watch::channel(0).0,
+            carry_owed: tokio::sync::watch::channel((0, 0)).0,
             cancelled: AtomicBool::new(false),
             slash_commands: Mutex::new(Vec::new()),
             skills: Mutex::new(Vec::new()),