From 6e36dcafca1916b8497b96d1d17783c036ce5ca9 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Sat, 24 Oct 2026 16:33:26 -0700 Subject: [PATCH 1/6] acpmux: a working lane whose model went quiet with no tool call running says it has stalled (lane.stalledSince) Co-Authored-By: Claude Opus 5.5 Agent-Session: https://cmux.com/deeplink/agent?harness=claude&session=01a16c8c-54ec-71b2-a0fa-125d1404dd72&host=big-red --- cmux-tui/crates/acpmux/src/cli/task.rs | 9 +- cmux-tui/crates/acpmux/src/config.rs | 14 +++ cmux-tui/crates/acpmux/src/hub/activity.rs | 100 +++++++++++++++++- .../crates/acpmux/src/hub/catalog_reload.rs | 1 + cmux-tui/crates/acpmux/src/hub/mod.rs | 1 + cmux-tui/crates/acpmux/src/hub/session.rs | 3 + cmux-tui/crates/acpmux/src/hub/subsessions.rs | 2 + cmux-tui/crates/acpmux/src/hub/tap.rs | 1 + cmux-tui/crates/acpmux/src/hub/views.rs | 17 ++- .../crates/acpmux/src/protocol/session.rs | 10 ++ .../acpmux/tests/hub_integration/lanes.rs | 46 ++++++++ 11 files changed, 197 insertions(+), 7 deletions(-) diff --git a/cmux-tui/crates/acpmux/src/cli/task.rs b/cmux-tui/crates/acpmux/src/cli/task.rs index 330fbac43f4c..554cee66f61b 100644 --- a/cmux-tui/crates/acpmux/src/cli/task.rs +++ b/cmux-tui/crates/acpmux/src/cli/task.rs @@ -130,11 +130,16 @@ fn report(state: &str, r: ResultArgs) -> (&'static str, Value) { (method::MUX_REPORT, params) } -/// One task: id, state, its lane and the lane's state now, the title, and the result's summary. +/// One task: id, state, its lane and the lane's state now (and how long it has stalled), the +/// title, and the result's summary. fn line(row: &Value) -> String { let s = |v: &Value| v.as_str().unwrap_or("?").to_owned(); - let lane = + let mut lane = row["lane"]["state"].as_str().map_or_else(|| s(&row["sessionStatus"]), str::to_owned); + if let Some(since) = row["lane"]["stalledSince"].as_u64() { + let minutes = crate::store::now_ms().saturating_sub(since) / 60_000; + lane.push_str(&format!(", stalled {minutes}m")); + } let mut text = format!( "{:<11} {:<8} {} ({lane}) {}", s(&row["taskId"]), diff --git a/cmux-tui/crates/acpmux/src/config.rs b/cmux-tui/crates/acpmux/src/config.rs index d436934989b8..89e7a2dae721 100644 --- a/cmux-tui/crates/acpmux/src/config.rs +++ b/cmux-tui/crates/acpmux/src/config.rs @@ -812,6 +812,12 @@ pub struct Config { /// Default 90000; 0 turns it off. #[serde(default, skip_serializing_if = "Option::is_none")] pub silent_turn_ms: Option, + /// `stalledTurnMs`: how long a running turn may go with no tool call in + /// flight and no sign of its model at work before its lane says it has + /// stalled (`lane.stalledSince`, `hub/views.rs`). A long build in one tool + /// call never stalls. Default 600000; 0 turns it off. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub stalled_turn_ms: Option, /// `autoResumeDelaysMs`: the wait before each automatic continuation of a turn a /// connection error cut off (`hub/auto_resume.rs`), one entry per attempt. Default /// `[10000, 60000, 300000, 1200000]`; `[]` turns it off. Checked on load: at most 4 @@ -936,6 +942,14 @@ impl Config { } } + /// `stalledTurnMs`, or its 10 minute default; None when it is 0 (off). + pub fn stalled_turn(&self) -> Option { + match self.stalled_turn_ms.unwrap_or(600_000) { + 0 => None, + ms => Some(ms), + } + } + /// Family of a configured profile. pub fn family(&self, profile: &str) -> Option { self.harnesses.get(profile).map(|p| derive_family(profile, p)) diff --git a/cmux-tui/crates/acpmux/src/hub/activity.rs b/cmux-tui/crates/acpmux/src/hub/activity.rs index 96c835326fa1..401bdd3ea692 100644 --- a/cmux-tui/crates/acpmux/src/hub/activity.rs +++ b/cmux-tui/crates/acpmux/src/hub/activity.rs @@ -9,11 +9,18 @@ //! still doing its last tool. Each is kept twice: raw for the owner's views, //! and masked before its cut for the sub-session tools (`subsession_row`), so //! a secret across the cut leaves no head there. +//! +//! The turn also keeps the ids of its tool calls in flight (started, not yet +//! completed or failed): a turn with none, whose model has been quiet for +//! `stalledTurnMs`, has stalled (`stalled_since`). A long build in one tool +//! call is in flight, so it never stalls. use super::*; /// Longest activity or plan step kept, in characters. const ACTIVITY_CHARS: usize = 160; +/// Most tool calls a turn keeps in flight; one past it is still counted as in flight. +const MAX_OPEN_TOOLS: usize = 256; /// What a running turn does now (`TurnInfo::doing`), one line each. #[derive(Debug, Clone, Default)] @@ -26,9 +33,41 @@ pub struct TurnActivity { pub masked_activity: Option, /// `plan_step` with its secrets masked before its cut. pub masked_plan_step: Option, + /// The ids of its tool calls in flight. + pub open_tools: std::collections::BTreeSet, + /// A tool call started past [`MAX_OPEN_TOOLS`]: the turn counts one in flight until it ends. + pub open_overflow: bool, } impl TurnActivity { + /// Notes a `tool_call` or `tool_call_update` in `update`: in flight until it completes or fails. + fn note_tool(&mut self, update: &Value) { + let kind = update.get("sessionUpdate").and_then(Value::as_str); + let Some(id) = update.get("toolCallId").and_then(Value::as_str) else { return }; + if !matches!(kind, Some("tool_call" | "tool_call_update")) { + return; + } + let status = update.get("status").and_then(Value::as_str); + if matches!(status, Some("completed" | "failed")) { + self.open_tools.remove(id); + return; + } + // A new call, or an update that says it runs; an update with no status changes nothing. + if status.is_none() && kind != Some("tool_call") { + return; + } + if self.open_tools.len() < MAX_OPEN_TOOLS { + self.open_tools.insert(id.to_owned()); + } else if !self.open_tools.contains(id) { + self.open_overflow = true; + } + } + + /// Whether a tool call of the turn is in flight. + pub(super) fn tool_in_flight(&self) -> bool { + !self.open_tools.is_empty() || self.open_overflow + } + /// Puts the masked lines in `turn`, a summary's running turn, for the /// sub-session tools. pub(super) fn mask_into(&self, turn: &mut Value) { @@ -86,18 +125,41 @@ pub(crate) fn doing_line(plan_step: Option<&str>, activity: Option<&str>) -> Opt } } +/// Since when a running turn that started at `started_at` has stalled, or None: its model's last +/// sign was at `heard_at` (0: none; the prompt counts as one), no tool call is in flight +/// (`in_flight`), and it has been quiet `after` ms or more by `now` (module docs). +pub(super) fn stalled_since( + started_at: u64, + heard_at: u64, + in_flight: bool, + now: u64, + after: Option, +) -> Option { + let quiet_since = started_at.max(heard_at); + (!in_flight && now.saturating_sub(quiet_since) >= after?).then_some(quiet_since) +} + impl Hub { - /// Notes `update`'s activity on the running turn. Nothing while no turn + /// Notes `update`'s activity and tool calls on the running turn. Nothing while no turn /// runs: an update after the turn ended does not outlive it. pub(super) fn note_activity(&self, session: &Session, update: &Value) { - let (Some(raw), Some(masked)) = (activity_of(update), masked_activity_of(update)) else { + let tool = matches!( + update.get("sessionUpdate").and_then(Value::as_str), + Some("tool_call" | "tool_call_update") + ); + let activity = activity_of(update).zip(masked_activity_of(update)); + if !tool && activity.is_none() { return; - }; + } let mut turn = session.turn.lock().unwrap_or_else(|e| e.into_inner()); let Some(turn) = turn.as_mut() else { return; }; let doing = &mut turn.doing; + if tool { + doing.note_tool(update); + } + let Some((raw, masked)) = activity else { return }; match (raw, masked) { (Activity::Tool(title), Activity::Tool(masked)) => { doing.activity = Some(title); @@ -143,6 +205,38 @@ mod tests { assert_eq!(activity_of(&done), Some(Activity::PlanStep(None))); } + #[test] + fn a_tool_call_is_in_flight_until_it_completes_or_fails() { + let mut doing = TurnActivity::default(); + doing.note_tool(&json!({"sessionUpdate": "tool_call", "toolCallId": "a", "title": "Build"})); + doing.note_tool(&json!({"sessionUpdate": "tool_call", "toolCallId": "b", "status": "pending"})); + assert!(doing.tool_in_flight()); + doing.note_tool(&json!({"sessionUpdate": "tool_call_update", "toolCallId": "a", "title": "Built"})); + doing.note_tool(&json!({"sessionUpdate": "tool_call_update", "toolCallId": "a", "status": "completed"})); + assert!(doing.tool_in_flight(), "b still runs"); + doing.note_tool(&json!({"sessionUpdate": "tool_call_update", "toolCallId": "b", "status": "failed"})); + assert!(!doing.tool_in_flight()); + // A call that starts already done is never in flight. + doing.note_tool(&json!({"sessionUpdate": "tool_call", "toolCallId": "c", "status": "completed"})); + assert!(!doing.tool_in_flight()); + } + + #[test] + fn a_turn_stalls_when_its_model_is_quiet_with_no_tool_in_flight() { + let after = Some(600_000); + // Heard a minute ago: not stalled. + assert_eq!(stalled_since(1_000, 2_000_000, false, 2_060_000, after), None); + // Quiet ten minutes: stalled since it was last heard. + assert_eq!(stalled_since(1_000, 2_000_000, false, 2_600_000, after), Some(2_000_000)); + // Never heard: quiet since the prompt. + assert_eq!(stalled_since(1_000, 0, false, 601_000, after), Some(1_000)); + // Heard only in an earlier turn: quiet since this turn's prompt. + assert_eq!(stalled_since(5_000, 4_000, false, 300_000, after), None); + // A tool call in flight (a long build) never stalls, and 0 turns it off. + assert_eq!(stalled_since(1_000, 0, true, 99_000_000, after), None); + assert_eq!(stalled_since(1_000, 0, false, 99_000_000, None), None); + } + #[test] fn a_long_title_is_cut() { let u = json!({"sessionUpdate": "tool_call", "title": "x".repeat(500)}); diff --git a/cmux-tui/crates/acpmux/src/hub/catalog_reload.rs b/cmux-tui/crates/acpmux/src/hub/catalog_reload.rs index f89ec2b9fb6d..7c83bad23b5d 100644 --- a/cmux-tui/crates/acpmux/src/hub/catalog_reload.rs +++ b/cmux-tui/crates/acpmux/src/hub/catalog_reload.rs @@ -71,6 +71,7 @@ impl Hub { current.agent_start_timeout_ms = next.agent_start_timeout_ms; current.uploads_max_age_days = next.uploads_max_age_days; current.silent_turn_ms = next.silent_turn_ms; + current.stalled_turn_ms = next.stalled_turn_ms; let start_changed = current.default_permission_mode != next.default_permission_mode || current.last_permission_mode != next.last_permission_mode || current.last_permission_mode_change != next.last_permission_mode_change; diff --git a/cmux-tui/crates/acpmux/src/hub/mod.rs b/cmux-tui/crates/acpmux/src/hub/mod.rs index a1e83385e010..b75ddb84340b 100644 --- a/cmux-tui/crates/acpmux/src/hub/mod.rs +++ b/cmux-tui/crates/acpmux/src/hub/mod.rs @@ -782,6 +782,7 @@ impl Hub { append_errors: AtomicU64::new(0), last_progress: AtomicU64::new(0), heard: AtomicU64::new(0), + last_model_at: AtomicU64::new(0), silent_turn: StdMutex::new(None), last_active: AtomicU64::new(self.clock_now()), web_control_ended: AtomicBool::new(false), diff --git a/cmux-tui/crates/acpmux/src/hub/session.rs b/cmux-tui/crates/acpmux/src/hub/session.rs index 60dc5802ca82..ed74d975dc94 100644 --- a/cmux-tui/crates/acpmux/src/hub/session.rs +++ b/cmux-tui/crates/acpmux/src/hub/session.rs @@ -123,6 +123,9 @@ pub struct Session { /// requests, its harness's model lines; not its own bookkeeping, `tap.rs` /// `from_the_model`); a turn that hears none is silent (`silent_turn.rs`). pub(super) heard: AtomicU64, + /// Unix ms of the agent's last sign of its model at work (as `heard`); 0: none since load. + /// A running turn quiet since then, with no tool call in flight, stalls (`views.rs`). + pub(super) last_model_at: AtomicU64, /// The turn that logged `turn_silent` and has not heard its model since (`silent_turn.rs`). pub(super) silent_turn: StdMutex>, /// The hub clock's time (`Hub::clock_now`) of the last record or diff --git a/cmux-tui/crates/acpmux/src/hub/subsessions.rs b/cmux-tui/crates/acpmux/src/hub/subsessions.rs index 2d9888bd01c7..70b23c913262 100644 --- a/cmux-tui/crates/acpmux/src/hub/subsessions.rs +++ b/cmux-tui/crates/acpmux/src/hub/subsessions.rs @@ -856,6 +856,8 @@ impl Hub { // working, input, failed, lost or idle, and since when (`views::lane`). "state": summary["lane"]["state"].clone(), "stateSince": summary["lane"]["since"].clone(), + // Working, but quiet with no tool call in flight since then (`hub/activity.rs`). + "stalledSince": summary["lane"]["stalledSince"].clone(), "doing": super::subsession_send::doing(&summary), // question or permission: only the person allows a permission, in the app. "waitingOn": pick("waitingOn"), diff --git a/cmux-tui/crates/acpmux/src/hub/tap.rs b/cmux-tui/crates/acpmux/src/hub/tap.rs index 7f28b3b748dd..c60cbd17947c 100644 --- a/cmux-tui/crates/acpmux/src/hub/tap.rs +++ b/cmux-tui/crates/acpmux/src/hub/tap.rs @@ -72,6 +72,7 @@ impl Hub { Arc::new(move |dir: Direction, msg: &Message, host_seq: Option| -> bool { if matches!(dir, Direction::In) && from_the_model(msg) { tap_session.heard.fetch_add(1, Ordering::SeqCst); + tap_session.last_model_at.store(crate::store::now_ms(), Ordering::Relaxed); tap_hub.heard_after_silence(&tap_session); } let (d, kind) = match (dir, msg) { diff --git a/cmux-tui/crates/acpmux/src/hub/views.rs b/cmux-tui/crates/acpmux/src/hub/views.rs index 05c6330b2551..2b5b83a981f1 100644 --- a/cmux-tui/crates/acpmux/src/hub/views.rs +++ b/cmux-tui/crates/acpmux/src/hub/views.rs @@ -83,7 +83,9 @@ impl Hub { /// `lastMessage`, `preview`, `title`) carry the raw text, masked only on the /// way to a peer daemon (`server/peer_text.rs`). /// -/// `{state, since, error, step, title}`; null for a closed session. +/// `stalledSince`: a working lane whose turn has stalled (`hub/activity.rs`), since when. +/// +/// `{state, since, error, step, title, stalledSince}`; null for a closed session. pub fn lane(summary: &Value) -> Value { if summary["status"] == "closed" { return Value::Null; @@ -111,7 +113,9 @@ pub fn lane(summary: &Value) -> Value { _ => Value::Null, }; let error = error.as_str().map_or(Value::Null, |e| Value::String(masked_error(e))); - json!({"state": state, "since": since, "error": error, "step": lane_step(summary), "title": lane_title(summary)}) + let stalled = if state == "working" { summary["turn"]["stalledSince"].clone() } else { Value::Null }; + json!({"state": state, "since": since, "error": error, "step": lane_step(summary), + "title": lane_title(summary), "stalledSince": stalled}) } /// The lane's step, one line: the running turn's prompt, else the agent's @@ -266,6 +270,8 @@ impl Hub { let m = session.meta(); let waiting = super::subsession_send::waiting_on(session); let turn = session.turn(); + let stall_after = self.config.try_read().ok().and_then(|c| c.stalled_turn()); + let heard_at = session.last_model_at.load(Ordering::Relaxed); let summary = SessionSummary { session_id: m.id.clone(), name: m.name.clone(), @@ -307,6 +313,13 @@ impl Hub { // Ctrl-B moves its work: by the kind its harness started as (`processes.rs`). backgrounds_work: session.backgrounds_work.load(Ordering::SeqCst), turn: turn.map(|t| TurnSummary { + stalled_since: super::activity::stalled_since( + t.started_at, + heard_at, + t.doing.tool_in_flight(), + crate::store::now_ms(), + stall_after, + ), started_at: t.started_at, client: t.client, prompt: t.prompt_preview, diff --git a/cmux-tui/crates/acpmux/src/protocol/session.rs b/cmux-tui/crates/acpmux/src/protocol/session.rs index 93a51afe64dc..35ce2736ae0c 100644 --- a/cmux-tui/crates/acpmux/src/protocol/session.rs +++ b/cmux-tui/crates/acpmux/src/protocol/session.rs @@ -234,6 +234,12 @@ pub struct Lane { /// The session's title, else its latest prompt, one line, secrets masked /// (`secret_scan`). pub title: Option, + /// A `working` lane that has stalled: since when its turn has had no tool + /// call in flight and no sign of its model (`turn.stalledSince`); null + /// otherwise. The state stays `working`, so a client that reads only it + /// is unchanged. + #[serde(default)] + pub stalled_since: Option, } #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] @@ -275,6 +281,10 @@ pub struct TurnSummary { pub activity: Option, /// Its plan's entry in progress; null without one. pub plan_step: Option, + /// Since when it has stalled: no tool call in flight and no sign of its + /// model since then, for `stalledTurnMs` or more (`hub/activity.rs`); + /// null while it has not. + pub stalled_since: Option, } /// A prompt waiting for the running turn to end. diff --git a/cmux-tui/crates/acpmux/tests/hub_integration/lanes.rs b/cmux-tui/crates/acpmux/tests/hub_integration/lanes.rs index e6efde51a6e3..599eed2c56ab 100644 --- a/cmux-tui/crates/acpmux/tests/hub_integration/lanes.rs +++ b/cmux-tui/crates/acpmux/tests/hub_integration/lanes.rs @@ -105,3 +105,49 @@ async fn an_agent_gone_without_an_answer_is_lost() { assert_eq!(s["lane"]["state"], "lost", "{s}"); assert!(s["lane"]["since"].as_u64().is_some(), "{s}"); } + +/// Goal 8 (leo-workspace research/2026-10-24-acpmux-tasks-liveness-urgency.md): a lane whose model +/// went quiet mid-turn with no tool call running says it has stalled, since its last sign; one +/// waiting on a long tool call (a build) never does. The state stays `working`. +#[cfg(unix)] +#[tokio::test] +async fn a_quiet_turn_with_no_tool_running_stalls_and_a_long_tool_call_does_not() { + let (hub, mut c) = setup(PermissionPolicy::ApproveAll).await; + hub.config.write().await.stalled_turn_ms = Some(300); + let fifo = |tag: &str| { + let gate = std::env::temp_dir().join(format!("acpmux-stall-{tag}-{}", uuid::Uuid::now_v7())); + assert!(std::process::Command::new("mkfifo").arg(&gate).status().unwrap().success()); + gate + }; + let (quiet_gate, tool_gate) = (fifo("quiet"), fifo("tool")); + let quiet = new_session(&mut c, "quiet").await; + let building = new_session(&mut c, "building").await; + c.send(method::SESSION_PROMPT, prompt(&quiet, &format!("gate: {}", quiet_gate.display()), None)) + .await; + c.wait_for(method::SESSION_UPDATE, |p| p["update"]["content"]["text"] == "before-gate").await; + let tool = format!("work-gate: {} | Run the full build", tool_gate.display()); + c.send(method::SESSION_PROMPT, prompt(&building, &tool, None)).await; + c.wait_for(method::SESSION_UPDATE, |p| p["update"]["title"] == "Run the full build").await; + + let s = settled(&hub, &mut c, &quiet, |s| s["lane"]["stalledSince"].is_u64()).await; + assert_eq!(s["lane"]["state"], "working", "{s}"); + let since = s["lane"]["stalledSince"].as_u64().unwrap(); + assert!(since >= s["turn"]["startedAt"].as_u64().unwrap(), "since its last sign: {s}"); + assert_eq!(s["turn"]["stalledSince"], s["lane"]["stalledSince"], "{s}"); + + tokio::time::sleep(Duration::from_millis(600)).await; + let s = settled(&hub, &mut c, &building, |s| s["status"] == "running").await; + assert_eq!(s["lane"]["state"], "working", "{s}"); + assert!(s["lane"]["stalledSince"].is_null(), "a tool call runs: {s}"); + + for gate in [&quiet_gate, &tool_gate] { + std::fs::write(gate, "go").unwrap(); + } + for id in [&quiet, &building] { + let s = settled(&hub, &mut c, id, |s| s["lastTurn"]["status"] == "completed").await; + assert!(s["lane"]["stalledSince"].is_null(), "an ended turn has not stalled: {s}"); + } + for gate in [quiet_gate, tool_gate] { + let _ = std::fs::remove_file(gate); + } +} From 4c61f764edfd5db4601b2753f62c3a94c7f8b7df Mon Sep 17 00:00:00 2001 From: Leo Li Date: Sat, 24 Oct 2026 16:38:23 -0700 Subject: [PATCH 2/6] acpmux: lane.stalledSince: fmt, protocol spec and generated types Co-Authored-By: Claude Opus 5.5 Agent-Session: https://cmux.com/deeplink/agent?harness=claude&session=01a16c8c-54ec-71b2-a0fa-125d1404dd72&host=big-red --- .../crates/acpmux/spec/acpmux-protocol.json | 21 +++++++++++++++++ cmux-tui/crates/acpmux/src/hub/activity.rs | 23 ++++++++++++++----- cmux-tui/crates/acpmux/src/hub/views.rs | 3 ++- .../crates/acpmux/src/protocol/session.rs | 2 +- .../acpmux/tests/hub_integration/lanes.rs | 10 +++++--- .../acpmux/protocol/generated/types.ts | 13 +++++++++++ 6 files changed, 61 insertions(+), 11 deletions(-) diff --git a/cmux-tui/crates/acpmux/spec/acpmux-protocol.json b/cmux-tui/crates/acpmux/spec/acpmux-protocol.json index b8348451091c..29a6d101521d 100644 --- a/cmux-tui/crates/acpmux/spec/acpmux-protocol.json +++ b/cmux-tui/crates/acpmux/spec/acpmux-protocol.json @@ -1711,6 +1711,16 @@ "null" ] }, + "stalledSince": { + "default": null, + "description": "A `working` lane that has stalled: since when its turn has had no tool\ncall in flight and no sign of its model (`turn.stalledSince`); null\notherwise. The state stays `working`, so a client that reads only it\nis unchanged.", + "format": "uint64", + "minimum": 0, + "type": [ + "integer", + "null" + ] + }, "state": { "$ref": "#/types/LaneState" }, @@ -1732,6 +1742,7 @@ "required": [ "error", "since", + "stalledSince", "state", "step", "title" @@ -3595,6 +3606,15 @@ "promptId": { "type": "string" }, + "stalledSince": { + "description": "Since when it has stalled: no tool call in flight and no sign of its\nmodel since then, for `stalledTurnMs` or more (`hub/activity.rs`);\nnull while it has not.", + "format": "uint64", + "minimum": 0, + "type": [ + "integer", + "null" + ] + }, "startedAt": { "format": "uint64", "minimum": 0, @@ -3615,6 +3635,7 @@ "planStep", "prompt", "promptId", + "stalledSince", "startedAt", "turnId", "turnSeq" diff --git a/cmux-tui/crates/acpmux/src/hub/activity.rs b/cmux-tui/crates/acpmux/src/hub/activity.rs index 401bdd3ea692..4b80010b54be 100644 --- a/cmux-tui/crates/acpmux/src/hub/activity.rs +++ b/cmux-tui/crates/acpmux/src/hub/activity.rs @@ -208,16 +208,27 @@ mod tests { #[test] fn a_tool_call_is_in_flight_until_it_completes_or_fails() { let mut doing = TurnActivity::default(); - doing.note_tool(&json!({"sessionUpdate": "tool_call", "toolCallId": "a", "title": "Build"})); - doing.note_tool(&json!({"sessionUpdate": "tool_call", "toolCallId": "b", "status": "pending"})); + doing + .note_tool(&json!({"sessionUpdate": "tool_call", "toolCallId": "a", "title": "Build"})); + doing.note_tool( + &json!({"sessionUpdate": "tool_call", "toolCallId": "b", "status": "pending"}), + ); assert!(doing.tool_in_flight()); - doing.note_tool(&json!({"sessionUpdate": "tool_call_update", "toolCallId": "a", "title": "Built"})); - doing.note_tool(&json!({"sessionUpdate": "tool_call_update", "toolCallId": "a", "status": "completed"})); + doing.note_tool( + &json!({"sessionUpdate": "tool_call_update", "toolCallId": "a", "title": "Built"}), + ); + doing.note_tool( + &json!({"sessionUpdate": "tool_call_update", "toolCallId": "a", "status": "completed"}), + ); assert!(doing.tool_in_flight(), "b still runs"); - doing.note_tool(&json!({"sessionUpdate": "tool_call_update", "toolCallId": "b", "status": "failed"})); + doing.note_tool( + &json!({"sessionUpdate": "tool_call_update", "toolCallId": "b", "status": "failed"}), + ); assert!(!doing.tool_in_flight()); // A call that starts already done is never in flight. - doing.note_tool(&json!({"sessionUpdate": "tool_call", "toolCallId": "c", "status": "completed"})); + doing.note_tool( + &json!({"sessionUpdate": "tool_call", "toolCallId": "c", "status": "completed"}), + ); assert!(!doing.tool_in_flight()); } diff --git a/cmux-tui/crates/acpmux/src/hub/views.rs b/cmux-tui/crates/acpmux/src/hub/views.rs index 2b5b83a981f1..c3f669e94e33 100644 --- a/cmux-tui/crates/acpmux/src/hub/views.rs +++ b/cmux-tui/crates/acpmux/src/hub/views.rs @@ -113,7 +113,8 @@ pub fn lane(summary: &Value) -> Value { _ => Value::Null, }; let error = error.as_str().map_or(Value::Null, |e| Value::String(masked_error(e))); - let stalled = if state == "working" { summary["turn"]["stalledSince"].clone() } else { Value::Null }; + let stalled = + if state == "working" { summary["turn"]["stalledSince"].clone() } else { Value::Null }; json!({"state": state, "since": since, "error": error, "step": lane_step(summary), "title": lane_title(summary), "stalledSince": stalled}) } diff --git a/cmux-tui/crates/acpmux/src/protocol/session.rs b/cmux-tui/crates/acpmux/src/protocol/session.rs index 35ce2736ae0c..299bf0ea69ac 100644 --- a/cmux-tui/crates/acpmux/src/protocol/session.rs +++ b/cmux-tui/crates/acpmux/src/protocol/session.rs @@ -238,7 +238,7 @@ pub struct Lane { /// call in flight and no sign of its model (`turn.stalledSince`); null /// otherwise. The state stays `working`, so a client that reads only it /// is unchanged. - #[serde(default)] + #[serde(default, rename = "stalledSince")] pub stalled_since: Option, } diff --git a/cmux-tui/crates/acpmux/tests/hub_integration/lanes.rs b/cmux-tui/crates/acpmux/tests/hub_integration/lanes.rs index 599eed2c56ab..5a4a2a6dce2d 100644 --- a/cmux-tui/crates/acpmux/tests/hub_integration/lanes.rs +++ b/cmux-tui/crates/acpmux/tests/hub_integration/lanes.rs @@ -115,15 +115,19 @@ async fn a_quiet_turn_with_no_tool_running_stalls_and_a_long_tool_call_does_not( let (hub, mut c) = setup(PermissionPolicy::ApproveAll).await; hub.config.write().await.stalled_turn_ms = Some(300); let fifo = |tag: &str| { - let gate = std::env::temp_dir().join(format!("acpmux-stall-{tag}-{}", uuid::Uuid::now_v7())); + let gate = + std::env::temp_dir().join(format!("acpmux-stall-{tag}-{}", uuid::Uuid::now_v7())); assert!(std::process::Command::new("mkfifo").arg(&gate).status().unwrap().success()); gate }; let (quiet_gate, tool_gate) = (fifo("quiet"), fifo("tool")); let quiet = new_session(&mut c, "quiet").await; let building = new_session(&mut c, "building").await; - c.send(method::SESSION_PROMPT, prompt(&quiet, &format!("gate: {}", quiet_gate.display()), None)) - .await; + c.send( + method::SESSION_PROMPT, + prompt(&quiet, &format!("gate: {}", quiet_gate.display()), None), + ) + .await; c.wait_for(method::SESSION_UPDATE, |p| p["update"]["content"]["text"] == "before-gate").await; let tool = format!("work-gate: {} | Run the full build", tool_gate.display()); c.send(method::SESSION_PROMPT, prompt(&building, &tool, None)).await; diff --git a/webviews/src/agent-session/acpmux/protocol/generated/types.ts b/webviews/src/agent-session/acpmux/protocol/generated/types.ts index 1b99bc4eb7f8..4441100780b5 100644 --- a/webviews/src/agent-session/acpmux/protocol/generated/types.ts +++ b/webviews/src/agent-session/acpmux/protocol/generated/types.ts @@ -127,6 +127,13 @@ export type Lane = { error: (string | null); /** Unix ms the state began; null when the daemon does not know. */ since: (number | null); + /** + * A `working` lane that has stalled: since when its turn has had no tool + * call in flight and no sign of its model (`turn.stalledSince`); null + * otherwise. The state stays `working`, so a client that reads only it + * is unchanged. + */ + stalledSince: (number | null); state: LaneState; /** * What the session is doing now, one line: the running turn's prompt, @@ -770,6 +777,12 @@ export type TurnSummary = { planStep: (string | null); prompt: string; promptId: string; + /** + * Since when it has stalled: no tool call in flight and no sign of its + * model since then, for `stalledTurnMs` or more (`hub/activity.rs`); + * null while it has not. + */ + stalledSince: (number | null); startedAt: number; turnId: string; turnSeq: number; From 7671f78a5f8109a4a26e0c6dfc26f22caf56f994 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Sat, 24 Oct 2026 16:53:15 -0700 Subject: [PATCH 3/6] acpmux: tool calls past the in-flight cap end too Count calls past the cap instead of a flag that never cleared, so a turn with 257+ tool calls can still stall once they all end (Codex review). Co-Authored-By: Claude Opus 5.5 Agent-Session: https://cmux.com/deeplink/agent?harness=claude&session=01a16c8c-54ec-71b2-a0fa-125d1404dd72&host=big-red --- cmux-tui/crates/acpmux/src/hub/activity.rs | 43 +++++++++++++++++----- 1 file changed, 34 insertions(+), 9 deletions(-) diff --git a/cmux-tui/crates/acpmux/src/hub/activity.rs b/cmux-tui/crates/acpmux/src/hub/activity.rs index 4b80010b54be..b3a47040e196 100644 --- a/cmux-tui/crates/acpmux/src/hub/activity.rs +++ b/cmux-tui/crates/acpmux/src/hub/activity.rs @@ -19,7 +19,7 @@ use super::*; /// Longest activity or plan step kept, in characters. const ACTIVITY_CHARS: usize = 160; -/// Most tool calls a turn keeps in flight; one past it is still counted as in flight. +/// Most tool-call ids a turn keeps in flight; calls past it are counted (`open_past_cap`). const MAX_OPEN_TOOLS: usize = 256; /// What a running turn does now (`TurnInfo::doing`), one line each. @@ -35,8 +35,9 @@ pub struct TurnActivity { pub masked_plan_step: Option, /// The ids of its tool calls in flight. pub open_tools: std::collections::BTreeSet, - /// A tool call started past [`MAX_OPEN_TOOLS`]: the turn counts one in flight until it ends. - pub open_overflow: bool, + /// How many tool calls started past [`MAX_OPEN_TOOLS`] are in flight, by count, not id: the + /// end of a call whose id isn't kept ends one of them. + pub open_past_cap: usize, } impl TurnActivity { @@ -49,23 +50,26 @@ impl TurnActivity { } let status = update.get("status").and_then(Value::as_str); if matches!(status, Some("completed" | "failed")) { - self.open_tools.remove(id); + if !self.open_tools.remove(id) { + self.open_past_cap = self.open_past_cap.saturating_sub(1); + } return; } // A new call, or an update that says it runs; an update with no status changes nothing. - if status.is_none() && kind != Some("tool_call") { + if status.is_none() && kind != Some("tool_call") || self.open_tools.contains(id) { return; } - if self.open_tools.len() < MAX_OPEN_TOOLS { + if self.open_past_cap == 0 && self.open_tools.len() < MAX_OPEN_TOOLS { self.open_tools.insert(id.to_owned()); - } else if !self.open_tools.contains(id) { - self.open_overflow = true; + } else if kind == Some("tool_call") { + // Past the cap only a new call counts: an update may be for a call already counted. + self.open_past_cap += 1; } } /// Whether a tool call of the turn is in flight. pub(super) fn tool_in_flight(&self) -> bool { - !self.open_tools.is_empty() || self.open_overflow + !self.open_tools.is_empty() || self.open_past_cap > 0 } /// Puts the masked lines in `turn`, a summary's running turn, for the @@ -232,6 +236,27 @@ mod tests { assert!(!doing.tool_in_flight()); } + #[test] + fn tool_calls_past_the_cap_stay_in_flight_only_until_they_end() { + let mut doing = TurnActivity::default(); + let n = MAX_OPEN_TOOLS + 3; + for i in 0..n { + let id = format!("t{i}"); + doing.note_tool(&json!({"sessionUpdate": "tool_call", "toolCallId": id})); + doing.note_tool(&json!({ + "sessionUpdate": "tool_call_update", "toolCallId": id, "status": "in_progress" + })); + } + for i in 0..n { + assert!(doing.tool_in_flight(), "t{i} and later still run"); + doing.note_tool(&json!({ + "sessionUpdate": "tool_call_update", "toolCallId": format!("t{i}"), + "status": "completed" + })); + } + assert!(!doing.tool_in_flight()); + } + #[test] fn a_turn_stalls_when_its_model_is_quiet_with_no_tool_in_flight() { let after = Some(600_000); From 55d48118880682948a98601ae4adbb6e95db2c79 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Sat, 24 Oct 2026 17:16:04 -0700 Subject: [PATCH 4/6] acpmux: an adopted turn keeps its tool calls in flight; schema and README say stalledSince The adopting daemon rebuilt the running turn with nothing in flight, so a tool call that ran across a restart read as stalled. open_work now notes the open turn's logged tool_call/plan updates (TurnActivity::note), which also brings back the turn's activity and plan step. _acpmux/schema and the README describe stalledSince and stalledTurnMs (Codex review). Co-Authored-By: Claude Opus 5.5 Agent-Session: https://cmux.com/deeplink/agent?harness=claude&session=01a16c8c-54ec-71b2-a0fa-125d1404dd72&host=big-red --- cmux-tui/crates/acpmux/README.md | 5 ++ .../crates/acpmux/docs/acpmux-schema.json | 2 +- cmux-tui/crates/acpmux/src/hub/activity.rs | 50 ++++++++++--------- cmux-tui/crates/acpmux/src/hub/hosts.rs | 13 ++++- .../crates/acpmux/tests/durable_agents.rs | 45 +++++++++++++++++ 5 files changed, 89 insertions(+), 26 deletions(-) diff --git a/cmux-tui/crates/acpmux/README.md b/cmux-tui/crates/acpmux/README.md index 4aae75566593..76e7e22480fa 100644 --- a/cmux-tui/crates/acpmux/README.md +++ b/cmux-tui/crates/acpmux/README.md @@ -199,6 +199,11 @@ responding model produces, such as Claude's stream and thinking or Codex's moder notice; the agent's own usage and turn notices do not count) for that long after its prompt (a stalled model) logs `turn_silent` {turnId, silentMs} once, so a client can say the model has not answered. The model's first sign after it logs `turn_heard` {turnId}. +`stalledTurnMs` (default 600000; 0 turns it off): a running turn with no tool call in +flight whose model has shown no sign for that long has stalled: `lane.stalledSince` (and +`turn.stalledSince`) say since when, and the lane's state stays `working`. A long build in +one tool call is in flight, so it never stalls. It is checked when a summary is read, with +no timer. Files dropped into a chat (`_acpmux/file_put`, under `/uploads//`) are deleted with the chat (`_acpmux/kill` with purge) and once older than `uploadsMaxAgeDays` (default 30; 0 keeps them), checked at the daemon's start and daily. diff --git a/cmux-tui/crates/acpmux/docs/acpmux-schema.json b/cmux-tui/crates/acpmux/docs/acpmux-schema.json index 9b06a97541bb..df51e4d8975c 100644 --- a/cmux-tui/crates/acpmux/docs/acpmux-schema.json +++ b/cmux-tui/crates/acpmux/docs/acpmux-schema.json @@ -119,7 +119,7 @@ "permission_chat_allowance": {"active":"boolean"}, "SessionDefaults": {"model": "string?", "effort": "string?", "policy": "string?", "prefer": "string[]? (profiles to use for a family, in order; first installed wins)", "env": "{key: value}? (${cwd}, ${home}, ${model}, ~/ expand)", "note": "keyed by family or profile"}, "HarnessProfile": {"kind": "acp|claude-stdio", "argv": "string[] (${model} = the requested model at spawn)", "env": "{key: value}", "family": "string? (derived from argv when absent)", "models": "(string | {id, name})[] declared catalog, shown ahead of the reported one", "model": "string?", "effort": "string?", "policy": "string?", "fallback": "string? (profile to move to on limit or auth errors)", "description": "string?"}, - "SessionSummary": {"sessionId": "string", "name": "string", "harness": "string", "family": "string|null", "cwd": "string", "status": "idle|ready|running|waiting|disconnected|closed", "peer": "string? (host name for remote sessions)", "stateSeq": "number (bumps on status, permission and turn changes)", "unread": "boolean (a turn ended while no attached client showed the chat; cleared when one shows it)", "shown": "boolean (some attached client shows the chat: _acpmux/attach visible, _acpmux/view)", "indicator": "input|working|failed|lost|done|null (the status dot every client draws; first match wins: input = a permission or question waits, working = a turn runs, failed = the last turn failed, lost = the agent process is gone (disconnected or unreachable) and the last turn did not answer, done = the last turn answered (seen is client-side), null = nothing to show)", "lastTurn": "{turnId, promptId, status, stopReason?, errorText?, errorSource?, detail?, endedAt} | null", "lane": "{state: working|input|idle|failed|lost, since: number|null, error: string|null, step: string|null, title: string|null} | null (a coordinator's view of the session; null when closed: working since its turn started, input since it began waiting, idle since its last turn ended, failed with that turn's errorText, lost when the agent process or peer is gone; step is the running turn's prompt, else the agent's last reply line, one line of at most 240 characters; title is the session's title, else its latest prompt, cut the same way; error, step and title have secrets masked, e.g. [GitHub Personal Access Token], and error is cut to 3072 characters; the summary's other text (title, lastPrompt, preview, lastMessage.text, the turn's and queued prompts, lastTurn.errorText) is raw for every client except a proven peer daemon (peer token), which reads it masked and cut to 16384 characters and may not rewrite a queued prompt that holds a secret)", "inputSince": "number|null (when the session began waiting on the person)", "attached": "number (clients attached now)", "pendingPermissions": "number", "queued": "number", "turnCount": "number", "lastSeq": "number", "lastPrompt": "string|null", "preview": "string|null", "model": "string|null", "currentModeId": "string|null", "policy": "string|null", "rules": "boolean", "tools": "ToolRow[] (the chat's tools: cmux's own, then what the harness reported)", "toolsMutable": "boolean (_acpmux/set_tools can turn its MCP servers off: Claude Code only)", "tags": "{key: value}", "createdAt": "number", "updatedAt": "number", "queue": "[{promptId, turnId, client, prompt, queuedAt, held, images: [{id, mimeType, name: string|null}]}] prompts waiting for the running turn (images: what each sends, in order; _acpmux/attachment reads each by id before it runs)", "turn": "{startedAt, client, prompt, turnId, promptId, turnSeq} | null (the running turn)", "enforcement": "Enforcement", "gitBranch": "string|null (the branch the session's cwd has checked out now; null on a detached HEAD or outside a repository. Live while a client holds the chat (a git record on each change); otherwise the last branch this daemon read, else the one the session started on)"}, + "SessionSummary": {"sessionId": "string", "name": "string", "harness": "string", "family": "string|null", "cwd": "string", "status": "idle|ready|running|waiting|disconnected|closed", "peer": "string? (host name for remote sessions)", "stateSeq": "number (bumps on status, permission and turn changes)", "unread": "boolean (a turn ended while no attached client showed the chat; cleared when one shows it)", "shown": "boolean (some attached client shows the chat: _acpmux/attach visible, _acpmux/view)", "indicator": "input|working|failed|lost|done|null (the status dot every client draws; first match wins: input = a permission or question waits, working = a turn runs, failed = the last turn failed, lost = the agent process is gone (disconnected or unreachable) and the last turn did not answer, done = the last turn answered (seen is client-side), null = nothing to show)", "lastTurn": "{turnId, promptId, status, stopReason?, errorText?, errorSource?, detail?, endedAt} | null", "lane": "{state: working|input|idle|failed|lost, since: number|null, error: string|null, step: string|null, title: string|null, stalledSince: number|null} | null (a coordinator's view of the session; null when closed: working since its turn started (stalledSince: since when a working turn has had no tool call in flight and no sign from its model for stalledTurnMs, default 10 minutes, 0 off), input since it began waiting, idle since its last turn ended, failed with that turn's errorText, lost when the agent process or peer is gone; step is the running turn's prompt, else the agent's last reply line, one line of at most 240 characters; title is the session's title, else its latest prompt, cut the same way; error, step and title have secrets masked, e.g. [GitHub Personal Access Token], and error is cut to 3072 characters; the summary's other text (title, lastPrompt, preview, lastMessage.text, the turn's and queued prompts, lastTurn.errorText) is raw for every client except a proven peer daemon (peer token), which reads it masked and cut to 16384 characters and may not rewrite a queued prompt that holds a secret)", "inputSince": "number|null (when the session began waiting on the person)", "attached": "number (clients attached now)", "pendingPermissions": "number", "queued": "number", "turnCount": "number", "lastSeq": "number", "lastPrompt": "string|null", "preview": "string|null", "model": "string|null", "currentModeId": "string|null", "policy": "string|null", "rules": "boolean", "tools": "ToolRow[] (the chat's tools: cmux's own, then what the harness reported)", "toolsMutable": "boolean (_acpmux/set_tools can turn its MCP servers off: Claude Code only)", "tags": "{key: value}", "createdAt": "number", "updatedAt": "number", "queue": "[{promptId, turnId, client, prompt, queuedAt, held, images: [{id, mimeType, name: string|null}]}] prompts waiting for the running turn (images: what each sends, in order; _acpmux/attachment reads each by id before it runs)", "turn": "{startedAt, client, prompt, turnId, promptId, turnSeq, stalledSince} | null (the running turn; stalledSince as lane.stalledSince)", "enforcement": "Enforcement", "gitBranch": "string|null (the branch the session's cwd has checked out now; null on a detached HEAD or outside a repository. Live while a client holds the chat (a git record on each change); otherwise the last branch this daemon read, else the one the session started on)"}, "SessionDetail": "SessionSummary plus modes, configOptions, models, agentInfo, agentCapabilities, usage, pending: [{permissionId, request, groupId, turnId}] (groupId and turnId are string|null), rulesJson, gitHead: string|null (the cwd's full HEAD commit when the session started or forked; null outside a repository; gitBranch is SessionSummary's, the branch checked out now), totals: TurnUsage|null (the sum of this session's own turns; null until a harness reports any; a fork starts its own)", "TurnUsage": {"costUsd": "number? (USD, the harness's estimate)", "inputTokens": "number?", "outputTokens": "number?", "cacheReadTokens": "number?", "cacheWriteTokens": "number?", "note": "Only what the harness reported: Claude Code reports all of them (its tokens cover the main loop, its cost includes subagents); other harnesses report none, so the fields are absent, never 0. A turn's costUsd is the rise in Claude Code's running total, and is absent when that cannot be told: a failed turn, a total that went down, the first turn of a fork through an earlier turn or of a session older than this field."}, "Process": {"id": "string (/: Claude's task id, Codex's processId)", "sessionId": "string", "subagentSessionId": "string|null (the subagent whose call started it)", "toolCallId": "string (Codex: the commandExecution item id)", "taskId": "string (Codex: the processId)", "harness": "string (the session's harness family: claude, codex)", "command": "string", "cwd": "string", "startedAt": "number (unix ms)", "endedAt": "number|null", "state": "running|exited|failed|killed|lost (lost: it ended with its harness, unseen)", "exitCode": "number|null (Codex reports it when the process exits on its own; Claude Code does not)", "signal": "null", "reason": "person|agent|harness_exit|null (why it stopped)", "detail": "string|null (the harness's own words on its end)", "origin": "background|auto_backgrounded|user_backgrounded (user_backgrounded: moved there by _acpmux/process/background, Ctrl-B)", "outputBytes": "number|null (its output file's length when listed; null while the harness named no file, and for Codex)", "outputFile": "string? (the file the harness writes its output to; read it with _acpmux/process/tail)"}, diff --git a/cmux-tui/crates/acpmux/src/hub/activity.rs b/cmux-tui/crates/acpmux/src/hub/activity.rs index b3a47040e196..e1f3177a12e9 100644 --- a/cmux-tui/crates/acpmux/src/hub/activity.rs +++ b/cmux-tui/crates/acpmux/src/hub/activity.rs @@ -67,6 +67,28 @@ impl TurnActivity { } } + /// Notes `update`'s activity and tool calls (`Hub::note_activity`; an adopted host's turn, + /// rebuilt from the log, `hosts.rs`). + pub(super) fn note(&mut self, update: &Value) { + if matches!( + update.get("sessionUpdate").and_then(Value::as_str), + Some("tool_call" | "tool_call_update") + ) { + self.note_tool(update); + } + match activity_of(update).zip(masked_activity_of(update)) { + Some((Activity::Tool(title), Activity::Tool(masked))) => { + self.activity = Some(title); + self.masked_activity = Some(masked); + } + Some((Activity::PlanStep(step), Activity::PlanStep(masked))) => { + self.plan_step = step; + self.masked_plan_step = masked; + } + _ => {} + } + } + /// Whether a tool call of the turn is in flight. pub(super) fn tool_in_flight(&self) -> bool { !self.open_tools.is_empty() || self.open_past_cap > 0 @@ -147,33 +169,13 @@ impl Hub { /// Notes `update`'s activity and tool calls on the running turn. Nothing while no turn /// runs: an update after the turn ended does not outlive it. pub(super) fn note_activity(&self, session: &Session, update: &Value) { - let tool = matches!( - update.get("sessionUpdate").and_then(Value::as_str), - Some("tool_call" | "tool_call_update") - ); - let activity = activity_of(update).zip(masked_activity_of(update)); - if !tool && activity.is_none() { + let kind = update.get("sessionUpdate").and_then(Value::as_str); + if !matches!(kind, Some("tool_call" | "tool_call_update" | "plan")) { return; } let mut turn = session.turn.lock().unwrap_or_else(|e| e.into_inner()); - let Some(turn) = turn.as_mut() else { - return; - }; - let doing = &mut turn.doing; - if tool { - doing.note_tool(update); - } - let Some((raw, masked)) = activity else { return }; - match (raw, masked) { - (Activity::Tool(title), Activity::Tool(masked)) => { - doing.activity = Some(title); - doing.masked_activity = Some(masked); - } - (Activity::PlanStep(step), Activity::PlanStep(masked)) => { - doing.plan_step = step; - doing.masked_plan_step = masked; - } - _ => {} + if let Some(turn) = turn.as_mut() { + turn.doing.note(update); } } } diff --git a/cmux-tui/crates/acpmux/src/hub/hosts.rs b/cmux-tui/crates/acpmux/src/hub/hosts.rs index de5fb0fe7762..c4bf272e7b2b 100644 --- a/cmux-tui/crates/acpmux/src/hub/hosts.rs +++ b/cmux-tui/crates/acpmux/src/hub/hosts.rs @@ -49,6 +49,9 @@ pub(super) struct OpenWork { /// A remote chain's Claude Code session whose current host has no /// passed sandbox canary just before it started (`remote_sandbox.rs`). unsandboxed: bool, + /// What that turn does, from its logged updates under this incarnation: its tool calls in + /// flight, its latest tool and its plan step (`hub/activity.rs`). + doing: super::activity::TurnActivity, } impl Hub { @@ -467,6 +470,7 @@ impl Hub { work.prompt_response = None; work.steers.clear(); work.last_host_seq = 0; + work.doing = Default::default(); } ("mux", "turn_started") => { let field = |k: &str| { @@ -483,6 +487,7 @@ impl Hub { work.prompt_request = None; work.prompt_response = None; work.steers.clear(); + work.doing = Default::default(); } ("mux", "turn_control") => { if let Some(t) = work.turn.as_mut() @@ -496,6 +501,12 @@ impl Hub { work.prompt_request = None; work.prompt_response = None; work.steers.clear(); + work.doing = Default::default(); + } + ("in", "tool_call" | "tool_call_update" | "plan") + if current && work.turn.is_some() => + { + work.doing.note(&e.msg["params"]["update"]); } ("mux", "permission_request") if current => { if let (Some(pid), Some(aid)) = ( @@ -617,7 +628,7 @@ impl Hub { turn_seq, // Who prompted (`turn_started`); an older record without it is Web. control, - doing: Default::default(), + doing: work.doing.clone(), }); session.floor.last_turn_web.store(control == Control::Web, Ordering::SeqCst); // A Web turn in flight in an agent the sandbox did not diff --git a/cmux-tui/crates/acpmux/tests/durable_agents.rs b/cmux-tui/crates/acpmux/tests/durable_agents.rs index 034a768a3319..5ed3803c6edc 100644 --- a/cmux-tui/crates/acpmux/tests/durable_agents.rs +++ b/cmux-tui/crates/acpmux/tests/durable_agents.rs @@ -380,6 +380,51 @@ async fn agent_turn_survives_a_daemon_crash_and_completes_after_adoption() { assert!(alive(harness_pid), "a second agent replaced the adopted one"); } +/// Goal 8 (`lane.stalledSince`, hub/activity.rs): a tool call still running across a daemon +/// restart is still in flight once the next daemon adopts its host, so the turn waiting on it +/// does not read as stalled. The adopting daemon rebuilt the turn with no tool calls in flight. +#[tokio::test] +async fn a_tool_call_running_across_a_restart_keeps_its_turn_from_stalling() { + let mut daemon = Daemon::new("stall-up", "approve-all"); + daemon.sigterm(); + std::fs::write( + daemon.home.join("config.json"), + json!({"harnesses": {"fake": {"argv": ["python3", FAKE]}}, "defaultHarness": "fake", "permissionPolicy": "approve-all", "stalledTurnMs": 300}).to_string(), + ) + .unwrap(); + daemon.start(); + let session = new_session(&daemon).await; + let gate = daemon.home.join("gate"); + make_fifo(&gate); + let mut client = daemon.rpc().await; + client + .send( + "session/prompt", + json!({"sessionId": session, "prompt": [{"type": "text", "text": format!("work-gate: {}", gate.display())}]}), + ) + .await; + daemon.wait_event(&session, "the running tool call", |e| e["kind"] == "tool_call"); + + daemon.sigkill(); + drop(client); + daemon.start(); + until_turn_restored(&daemon, &session).await; + tokio::time::sleep(Duration::from_millis(600)).await; + let all = daemon.rpc().await.call("_acpmux/sessions", json!({})).await; + let entry = all["sessions"] + .as_array() + .and_then(|all| all.iter().find(|s| s["sessionId"] == session)) + .cloned() + .unwrap_or(Value::Null); + assert_eq!(entry["lane"]["state"], "working", "{entry}"); + assert!(entry["lane"]["stalledSince"].is_null(), "its tool call still runs: {entry}"); + assert_eq!(entry["turn"]["activity"], "Run cargo test", "its tool call, from the log: {entry}"); + + std::fs::write(&gate, b"go").unwrap(); + let result = daemon.wait_event(&session, "turn_result", |e| e["kind"] == "turn_result"); + assert_eq!(result["msg"]["status"], "completed", "{result}"); +} + /// An update restarts acpmux while a turn runs that took a steer (the Chief's `tell`). Claude /// Code answers a steer "steered" at once and the turn runs on. The adopting daemon must keep /// the turn open until the turn's own prompt is answered; it settled the turn on the steer's From dc567e8c223565015d8388dd65b90c9059d28965 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Sat, 24 Oct 2026 17:26:52 -0700 Subject: [PATCH 5/6] acpmux: an adopted turn leaves its subagents' tool calls out The rebuild after a restart fed a subagent's logged updates into its parent turn, so a child's running tool kept the parent from stalling and took its activity. Skip records with _meta.acpmux.subagent, as the live path does (Codex review). Co-Authored-By: Claude Opus 5.5 Agent-Session: https://cmux.com/deeplink/agent?harness=claude&session=01a16c8c-54ec-71b2-a0fa-125d1404dd72&host=big-red --- cmux-tui/crates/acpmux/src/hub/hosts.rs | 5 ++- .../crates/acpmux/tests/durable_agents.rs | 45 +++++++++++++++++++ cmux-tui/crates/acpmux/tests/fake_agent.py | 11 +++++ 3 files changed, 60 insertions(+), 1 deletion(-) diff --git a/cmux-tui/crates/acpmux/src/hub/hosts.rs b/cmux-tui/crates/acpmux/src/hub/hosts.rs index c4bf272e7b2b..2377de0e1435 100644 --- a/cmux-tui/crates/acpmux/src/hub/hosts.rs +++ b/cmux-tui/crates/acpmux/src/hub/hosts.rs @@ -503,8 +503,11 @@ impl Hub { work.steers.clear(); work.doing = Default::default(); } + // A subagent's updates are not the turn's own (`permissions.rs` `on_update`). ("in", "tool_call" | "tool_call_update" | "plan") - if current && work.turn.is_some() => + if current + && work.turn.is_some() + && e.msg.pointer("/params/_meta/acpmux/subagent").is_none() => { work.doing.note(&e.msg["params"]["update"]); } diff --git a/cmux-tui/crates/acpmux/tests/durable_agents.rs b/cmux-tui/crates/acpmux/tests/durable_agents.rs index 5ed3803c6edc..5469cb63e252 100644 --- a/cmux-tui/crates/acpmux/tests/durable_agents.rs +++ b/cmux-tui/crates/acpmux/tests/durable_agents.rs @@ -425,6 +425,51 @@ async fn a_tool_call_running_across_a_restart_keeps_its_turn_from_stalling() { assert_eq!(result["msg"]["status"], "completed", "{result}"); } +/// A subagent's tool call is not its parent turn's: rebuilt after a restart, the parent turn +/// still stalls while only its subagent's tool call runs, and does not take its activity. +#[tokio::test] +async fn a_subagents_tool_call_across_a_restart_is_not_its_parent_turns() { + let mut daemon = Daemon::new("stall-sub", "approve-all"); + daemon.sigterm(); + std::fs::write( + daemon.home.join("config.json"), + json!({"harnesses": {"fake": {"argv": ["python3", FAKE]}}, "defaultHarness": "fake", "permissionPolicy": "approve-all", "stalledTurnMs": 300}).to_string(), + ) + .unwrap(); + daemon.start(); + let session = new_session(&daemon).await; + let gate = daemon.home.join("gate"); + make_fifo(&gate); + let mut client = daemon.rpc().await; + client + .send( + "session/prompt", + json!({"sessionId": session, "prompt": [{"type": "text", "text": format!("child-gate: {}", gate.display())}]}), + ) + .await; + daemon.wait_event(&session, "the subagent's tool call", |e| { + e["kind"] == "tool_call" && e["msg"].pointer("/params/_meta/acpmux/subagent").is_some() + }); + + daemon.sigkill(); + drop(client); + daemon.start(); + until_turn_restored(&daemon, &session).await; + tokio::time::sleep(Duration::from_millis(600)).await; + let all = daemon.rpc().await.call("_acpmux/sessions", json!({})).await; + let entry = all["sessions"] + .as_array() + .and_then(|all| all.iter().find(|s| s["sessionId"] == session)) + .cloned() + .unwrap_or(Value::Null); + assert!(entry["lane"]["stalledSince"].is_u64(), "only its subagent's call runs: {entry}"); + assert!(entry["turn"]["activity"].is_null(), "the subagent's activity: {entry}"); + + std::fs::write(&gate, b"go").unwrap(); + let result = daemon.wait_event(&session, "turn_result", |e| e["kind"] == "turn_result"); + assert_eq!(result["msg"]["status"], "completed", "{result}"); +} + /// An update restarts acpmux while a turn runs that took a steer (the Chief's `tell`). Claude /// Code answers a steer "steered" at once and the turn runs on. The adopting daemon must keep /// the turn open until the turn's own prompt is answered; it settled the turn on the steer's diff --git a/cmux-tui/crates/acpmux/tests/fake_agent.py b/cmux-tui/crates/acpmux/tests/fake_agent.py index 7af2fbc142d5..5bfdb892342a 100644 --- a/cmux-tui/crates/acpmux/tests/fake_agent.py +++ b/cmux-tui/crates/acpmux/tests/fake_agent.py @@ -418,6 +418,17 @@ def handle_prompt(rid, params): update(sid, {"sessionUpdate": "agent_message_chunk", "content": {"type": "text", "text": "worked"}}) send({"jsonrpc": "2.0", "id": rid, "result": {"stopReason": "end_turn"}}) return + # "child-gate: PATH" starts subagent child-1, whose tool call "Child build" runs while + # the turn blocks on the FIFO at PATH; then the child's call and the turn end. + if text.startswith("child-gate:"): + update(sid, {"sessionUpdate": "subagent_spawned", "subagentSessionId": "child-1", "name": "Child", "task": "Child build", "capabilities": {}}) + update("child-1", {"sessionUpdate": "tool_call", "toolCallId": "cg-1", "title": "Child build", "kind": "execute", "status": "in_progress"}) + with open(text[len("child-gate:"):].strip()) as gate: + gate.read() + update("child-1", {"sessionUpdate": "tool_call_update", "toolCallId": "cg-1", "status": "completed"}) + update(sid, {"sessionUpdate": "subagent_state_update", "subagentSessionId": "child-1", "state": "completed"}) + send({"jsonrpc": "2.0", "id": rid, "result": {"stopReason": "end_turn"}}) + return # "gate: PATH" streams "before-gate", blocks until the test writes to # the FIFO at PATH (no timers), then streams "after-gate" and ends. if text.startswith("gate:"): From d2d8084a7a7ff0697a19da22af9030a16361c178 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Sat, 24 Oct 2026 17:37:52 -0700 Subject: [PATCH 6/6] acpmux: an adopted turn tells its subagents' updates by the log's own subagents open_work rebuilds the subagent tree from the log (SubagentTree::annotate, as the tap does live) instead of trusting an update's _meta.acpmux, which the agent can write (Codex review). Co-Authored-By: Claude Opus 5.5 Agent-Session: https://cmux.com/deeplink/agent?harness=claude&session=01a16c8c-54ec-71b2-a0fa-125d1404dd72&host=big-red --- cmux-tui/crates/acpmux/src/hub/hosts.rs | 13 ++- .../crates/acpmux/tests/durable_agents.rs | 89 +++++++++---------- cmux-tui/crates/acpmux/tests/fake_agent.py | 12 +++ 3 files changed, 62 insertions(+), 52 deletions(-) diff --git a/cmux-tui/crates/acpmux/src/hub/hosts.rs b/cmux-tui/crates/acpmux/src/hub/hosts.rs index 2377de0e1435..b210dea67b70 100644 --- a/cmux-tui/crates/acpmux/src/hub/hosts.rs +++ b/cmux-tui/crates/acpmux/src/hub/hosts.rs @@ -410,7 +410,14 @@ impl Hub { let remote_claude = self.remote_claude(session); work.unsandboxed = remote_claude; let mut canary_passed = false; - let _ = self.store.scan(&session.id, 0, &mut |e: EventRecord| { + // Which logged updates are a subagent's, as the tap tells them live: by the subagents + // the log itself started, never by an update's own `_meta`. + let mut subagents = crate::subagents::SubagentTree::default(); + subagents.set_own(session.meta().agent_session_id.as_deref()); + let _ = self.store.scan(&session.id, 0, &mut |mut e: EventRecord| { + let subagents_own = e.dir == "in" + && e.msg.get("method").and_then(Value::as_str) == Some(method::SESSION_UPDATE) + && e.msg.get_mut("params").is_some_and(|p| subagents.annotate(p).is_some()); if e.dir == "mux" { let text = |k: &str| e.msg.get(k).and_then(Value::as_str).map(str::to_owned); match e.kind.as_str() { @@ -505,9 +512,7 @@ impl Hub { } // A subagent's updates are not the turn's own (`permissions.rs` `on_update`). ("in", "tool_call" | "tool_call_update" | "plan") - if current - && work.turn.is_some() - && e.msg.pointer("/params/_meta/acpmux/subagent").is_none() => + if current && work.turn.is_some() && !subagents_own => { work.doing.note(&e.msg["params"]["update"]); } diff --git a/cmux-tui/crates/acpmux/tests/durable_agents.rs b/cmux-tui/crates/acpmux/tests/durable_agents.rs index 5469cb63e252..256df778a0ef 100644 --- a/cmux-tui/crates/acpmux/tests/durable_agents.rs +++ b/cmux-tui/crates/acpmux/tests/durable_agents.rs @@ -380,12 +380,15 @@ async fn agent_turn_survives_a_daemon_crash_and_completes_after_adoption() { assert!(alive(harness_pid), "a second agent replaced the adopted one"); } -/// Goal 8 (`lane.stalledSince`, hub/activity.rs): a tool call still running across a daemon -/// restart is still in flight once the next daemon adopts its host, so the turn waiting on it -/// does not read as stalled. The adopting daemon rebuilt the turn with no tool calls in flight. -#[tokio::test] -async fn a_tool_call_running_across_a_restart_keeps_its_turn_from_stalling() { - let mut daemon = Daemon::new("stall-up", "approve-all"); +/// A daemon whose turns stall after 300 ms (`stalledTurnMs`) runs `mode: GATE` (fake_agent.py) +/// until `ready` logs, is killed and started again, and adopts the turn: the chat's summary +/// 600 ms after that, with the daemon, the chat and the gate that ends the turn. +async fn adopted_after( + tag: &str, + mode: &str, + ready: impl Fn(&Value) -> bool, +) -> (Daemon, String, PathBuf, Value) { + let mut daemon = Daemon::new(tag, "approve-all"); daemon.sigterm(); std::fs::write( daemon.home.join("config.json"), @@ -400,10 +403,10 @@ async fn a_tool_call_running_across_a_restart_keeps_its_turn_from_stalling() { client .send( "session/prompt", - json!({"sessionId": session, "prompt": [{"type": "text", "text": format!("work-gate: {}", gate.display())}]}), + json!({"sessionId": session, "prompt": [{"type": "text", "text": format!("{mode}: {}", gate.display())}]}), ) .await; - daemon.wait_event(&session, "the running tool call", |e| e["kind"] == "tool_call"); + daemon.wait_event(&session, "the running tool call", ready); daemon.sigkill(); drop(client); @@ -416,58 +419,48 @@ async fn a_tool_call_running_across_a_restart_keeps_its_turn_from_stalling() { .and_then(|all| all.iter().find(|s| s["sessionId"] == session)) .cloned() .unwrap_or(Value::Null); + (daemon, session, gate, entry) +} + +/// Opens `gate` and waits for the turn to complete. +fn ends_its_turn(daemon: &Daemon, session: &str, gate: &Path) { + std::fs::write(gate, b"go").unwrap(); + let result = daemon.wait_event(session, "turn_result", |e| e["kind"] == "turn_result"); + assert_eq!(result["msg"]["status"], "completed", "{result}"); +} + +/// Goal 8 (`lane.stalledSince`, hub/activity.rs): a tool call still running across a daemon +/// restart is still in flight once the next daemon adopts its host, so the turn waiting on it +/// does not read as stalled. The adopting daemon rebuilt the turn with no tool calls in flight. +#[tokio::test] +async fn a_tool_call_running_across_a_restart_keeps_its_turn_from_stalling() { + let (daemon, session, gate, entry) = + adopted_after("stall-up", "work-gate", |e| e["kind"] == "tool_call").await; assert_eq!(entry["lane"]["state"], "working", "{entry}"); assert!(entry["lane"]["stalledSince"].is_null(), "its tool call still runs: {entry}"); assert_eq!(entry["turn"]["activity"], "Run cargo test", "its tool call, from the log: {entry}"); - - std::fs::write(&gate, b"go").unwrap(); - let result = daemon.wait_event(&session, "turn_result", |e| e["kind"] == "turn_result"); - assert_eq!(result["msg"]["status"], "completed", "{result}"); + ends_its_turn(&daemon, &session, &gate); } /// A subagent's tool call is not its parent turn's: rebuilt after a restart, the parent turn -/// still stalls while only its subagent's tool call runs, and does not take its activity. +/// still stalls while only its subagent's tool call runs, and does not take its activity. The +/// log's own subagents say whose an update is, not the update's `_meta`: a tool call of the +/// turn's own that names a subagent there is still the turn's. #[tokio::test] async fn a_subagents_tool_call_across_a_restart_is_not_its_parent_turns() { - let mut daemon = Daemon::new("stall-sub", "approve-all"); - daemon.sigterm(); - std::fs::write( - daemon.home.join("config.json"), - json!({"harnesses": {"fake": {"argv": ["python3", FAKE]}}, "defaultHarness": "fake", "permissionPolicy": "approve-all", "stalledTurnMs": 300}).to_string(), - ) - .unwrap(); - daemon.start(); - let session = new_session(&daemon).await; - let gate = daemon.home.join("gate"); - make_fifo(&gate); - let mut client = daemon.rpc().await; - client - .send( - "session/prompt", - json!({"sessionId": session, "prompt": [{"type": "text", "text": format!("child-gate: {}", gate.display())}]}), - ) - .await; - daemon.wait_event(&session, "the subagent's tool call", |e| { + let (daemon, session, gate, entry) = adopted_after("stall-sub", "child-gate", |e| { e["kind"] == "tool_call" && e["msg"].pointer("/params/_meta/acpmux/subagent").is_some() - }); - - daemon.sigkill(); - drop(client); - daemon.start(); - until_turn_restored(&daemon, &session).await; - tokio::time::sleep(Duration::from_millis(600)).await; - let all = daemon.rpc().await.call("_acpmux/sessions", json!({})).await; - let entry = all["sessions"] - .as_array() - .and_then(|all| all.iter().find(|s| s["sessionId"] == session)) - .cloned() - .unwrap_or(Value::Null); + }) + .await; assert!(entry["lane"]["stalledSince"].is_u64(), "only its subagent's call runs: {entry}"); assert!(entry["turn"]["activity"].is_null(), "the subagent's activity: {entry}"); + ends_its_turn(&daemon, &session, &gate); - std::fs::write(&gate, b"go").unwrap(); - let result = daemon.wait_event(&session, "turn_result", |e| e["kind"] == "turn_result"); - assert_eq!(result["msg"]["status"], "completed", "{result}"); + let (daemon, session, gate, entry) = + adopted_after("stall-meta", "meta-gate", |e| e["kind"] == "tool_call").await; + assert!(entry["lane"]["stalledSince"].is_null(), "its own call runs: {entry}"); + assert_eq!(entry["turn"]["activity"], "Own build", "{entry}"); + ends_its_turn(&daemon, &session, &gate); } /// An update restarts acpmux while a turn runs that took a steer (the Chief's `tell`). Claude diff --git a/cmux-tui/crates/acpmux/tests/fake_agent.py b/cmux-tui/crates/acpmux/tests/fake_agent.py index 5bfdb892342a..713de7e595a6 100644 --- a/cmux-tui/crates/acpmux/tests/fake_agent.py +++ b/cmux-tui/crates/acpmux/tests/fake_agent.py @@ -418,6 +418,18 @@ def handle_prompt(rid, params): update(sid, {"sessionUpdate": "agent_message_chunk", "content": {"type": "text", "text": "worked"}}) send({"jsonrpc": "2.0", "id": rid, "result": {"stopReason": "end_turn"}}) return + # "meta-gate: PATH" is "work-gate:" with no plan, its own tool call "Own build" marked as a + # subagent's in its params' _meta.acpmux (the hub, not the agent, says whose an update is). + if text.startswith("meta-gate:"): + send({"jsonrpc": "2.0", "method": "session/update", "params": { + "sessionId": sid, "_meta": {"acpmux": {"subagent": "child-1"}}, + "update": {"sessionUpdate": "tool_call", "toolCallId": "mg-1", "title": "Own build", "kind": "execute", + "status": "in_progress"}}}) + with open(text[len("meta-gate:"):].strip()) as gate: + gate.read() + update(sid, {"sessionUpdate": "tool_call_update", "toolCallId": "mg-1", "status": "completed"}) + send({"jsonrpc": "2.0", "id": rid, "result": {"stopReason": "end_turn"}}) + return # "child-gate: PATH" starts subagent child-1, whose tool call "Child build" runs while # the turn blocks on the FIFO at PATH; then the child's call and the turn end. if text.startswith("child-gate:"):