From 111d4a9d8c5d5f011182694a23cfb358e5082936 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Thu, 29 Oct 2026 20:29:09 +0000 Subject: [PATCH] fix(acpmux): bound automatic agent steering by original turn age Agent-Session: https://cmux.com/deeplink/agent?id=aid-e919bf8ec2a9fcb5155e --- cmux-tui/crates/acpmux/README.md | 7 + cmux-tui/crates/acpmux/src/config.rs | 11 ++ .../src/hub/agent_steer_window_tests.rs | 163 ++++++++++++++++++ cmux-tui/crates/acpmux/src/hub/turns.rs | 36 +++- 4 files changed, 209 insertions(+), 8 deletions(-) create mode 100644 cmux-tui/crates/acpmux/src/hub/agent_steer_window_tests.rs diff --git a/cmux-tui/crates/acpmux/README.md b/cmux-tui/crates/acpmux/README.md index 644c5bf22226..7a9a329bc10c 100644 --- a/cmux-tui/crates/acpmux/README.md +++ b/cmux-tui/crates/acpmux/README.md @@ -379,6 +379,13 @@ flight whose model has shown no sign for that long has stalled: `lane.stalledSin `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. +`agentSteerWindowMs` (default 600000, 10 min): automatic agent messages steer only inside +this window from the turn's original start. Later messages wait in the durable queue, +behind prompts already waiting; they never interrupt the running tool or renew the +window. A steer-only request is refused explicitly. Human steering and explicit Send +now are unchanged. Zero queues automatic agent messages from the start of every turn. +This bounds admission, not tool execution: work already running finishes normally. +The original start survives daemon adoption; elapsed time uses the system clock. `stalledTurnEndMs` (default 21600000, 6 h; 0 turns it off): a running turn whose model has shown no sign of work, counted the same way, for that long (four times that while one of its tool calls has not ended; also across a daemon restart: an adopted diff --git a/cmux-tui/crates/acpmux/src/config.rs b/cmux-tui/crates/acpmux/src/config.rs index 75cb4f1e5cc0..2e603f2c7eef 100644 --- a/cmux-tui/crates/acpmux/src/config.rs +++ b/cmux-tui/crates/acpmux/src/config.rs @@ -939,6 +939,12 @@ pub struct Config { /// call never stalls. Default 600000; 0 turns it off. #[serde(default, skip_serializing_if = "Option::is_none")] pub stalled_turn_ms: Option, + /// `agentSteerWindowMs`: how long automatic agent messages may steer into one turn, + /// measured from its original start, never renewed by activity. Later messages queue + /// durably without interrupting tools. Human steering and explicit Send now are unchanged. + /// Default 600000 (10 min); 0 queues automatic messages from the start of every turn. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub agent_steer_window_ms: Option, /// `stalledTurnEndMs`: how long a running turn may go with no sign of its model at work (as /// `silentTurnMs` counts them, also across a daemon restart) before acpmux ends it as failed, /// `detail` `stalled` (`hub/stalled_turn.rs`); four times that inside a tool call that has @@ -1128,6 +1134,11 @@ impl Config { } } + /// Original-turn window for automatic agent steering; zero queues it immediately. + pub fn agent_steer_window(&self) -> u64 { + self.agent_steer_window_ms.unwrap_or(600_000) + } + /// `stalledTurnEndMs`, or its 6 h default; None when it is 0 (off). pub fn stalled_turn_end(&self) -> Option { match self.stalled_turn_end_ms.unwrap_or(6 * 60 * 60 * 1000) { diff --git a/cmux-tui/crates/acpmux/src/hub/agent_steer_window_tests.rs b/cmux-tui/crates/acpmux/src/hub/agent_steer_window_tests.rs new file mode 100644 index 000000000000..0e3a6d631f59 --- /dev/null +++ b/cmux-tui/crates/acpmux/src/hub/agent_steer_window_tests.rs @@ -0,0 +1,163 @@ +//! Automatic messages stop extending an aged turn; its tools finish and its queue drains. + +use super::*; +use crate::config::{Config, StoreMode}; +use std::time::Duration; + +const WAIT: Duration = Duration::from_secs(20); + +fn blocks(text: &str) -> Vec { + vec![json!({"type": "text", "text": text})] +} + +async fn admitted( + hub: &Arc, + session: &Arc, + text: &'static str, + steer: bool, + mut opts: PromptOptions, +) -> (tokio::task::JoinHandle>, Value) { + let (tx, rx) = tokio::sync::oneshot::channel(); + opts.prompt_id = Some(text.to_owned()); + opts.on_accepted = Some(Box::new(move |v| { + let _ = tx.send(v); + })); + let (hub, session) = (hub.clone(), session.clone()); + let task = + tokio::spawn( + async move { hub.prompt_with(&session, blocks(text), "test", steer, opts).await }, + ); + let accepted = tokio::time::timeout(WAIT, rx).await.unwrap().unwrap(); + (task, accepted) +} + +fn agent() -> PromptOptions { + PromptOptions { agent: Some("worker".to_owned()), ..Default::default() } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn agent_steer_window_drains_mixed_queue_without_cancelling_work() { + let temp = tempfile::tempdir().unwrap(); + let fifo = temp.path().join("work-gate"); + assert!(std::process::Command::new("mkfifo").arg(&fifo).status().unwrap().success()); + let fake = concat!(env!("CARGO_MANIFEST_DIR"), "/tests/fake_agent.py"); + let mut cfg: Config = serde_json::from_value(json!({ + "harnesses": {"fake": {"argv": ["/usr/bin/env", "FAKE_CLAUDE_STEER=1", "python3", fake]}}, + "defaultHarness": "fake", + })) + .unwrap(); + cfg.store.mode = StoreMode::Memory; + let store = crate::store::open(&cfg.store, temp.path()).unwrap(); + let hub = Hub::new(cfg, store); + let session = hub + .new_session(NewRequest { cwd: Some(temp.path().to_owned()), ..Default::default() }) + .await + .unwrap(); + let mut events = hub.subscribe(); + let original = { + let (hub, session) = (hub.clone(), session.clone()); + let text = format!("work-gate: {}", fifo.display()); + tokio::spawn(async move { hub.prompt(&session, blocks(&text), "test", false).await }) + }; + tokio::time::timeout(WAIT, async { + loop { + let event = events.recv().await.unwrap(); + if event.record.msg["params"]["update"]["toolCallId"] == "wg-1" { + break; + } + } + }) + .await + .unwrap(); + // Closing this writer lets the fixture's tool finish, even if an assertion unwinds. + let gate = tokio::task::spawn_blocking(move || { + std::fs::OpenOptions::new().write(true).open(fifo).unwrap() + }) + .await + .unwrap(); + let original_turn = session.turn().unwrap().turn_id; + let (fresh, accepted) = admitted(&hub, &session, "fresh-agent", true, agent()).await; + assert_eq!(accepted["steer"], true); + assert_eq!( + tokio::time::timeout(WAIT, fresh).await.unwrap().unwrap().unwrap()["stopReason"], + "steered" + ); + + // No sleep: age the original turn while its tool remains blocked. Activity and steers + // must not renew the window, including on an adopted turn with an old started_at. + session.turn.lock().unwrap().as_mut().unwrap().started_at = now_ms() - 60 * 60 * 1000; + let (human, human_accepted) = + admitted(&hub, &session, "human-steer", true, Default::default()).await; + let human_answer = tokio::time::timeout(WAIT, human).await.unwrap().unwrap().unwrap(); + let (explicit, explicit_accepted) = admitted( + &hub, + &session, + "explicit-send-now", + true, + PromptOptions { steer_now: true, ..agent() }, + ) + .await; + let explicit_answer = tokio::time::timeout(WAIT, explicit).await.unwrap().unwrap().unwrap(); + let refused = hub + .prompt_with( + &session, + blocks("steer-only-refused"), + "test", + true, + PromptOptions { steer_only: true, ..agent() }, + ) + .await; + let (older, accepted) = + admitted(&hub, &session, "older-unsteered", false, Default::default()).await; + assert_eq!(accepted["queued"], true); + let (late, late_accepted) = + admitted(&hub, &session, "late-agent", true, PromptOptions { interrupt: true, ..agent() }) + .await; + let (later, later_accepted) = admitted(&hub, &session, "later-agent", true, agent()).await; + // Release before checking the regression so a red run cleans up its fixture too. + let pending = session.queue(); + let still_running = session.turn().map(|t| t.turn_id); + drop(gate); + let original_answer = tokio::time::timeout(WAIT, original).await.unwrap().unwrap().unwrap(); + let older_answer = tokio::time::timeout(WAIT, older).await.unwrap().unwrap().unwrap(); + let late_answer = tokio::time::timeout(WAIT, late).await.unwrap().unwrap().unwrap(); + let later_answer = tokio::time::timeout(WAIT, later).await.unwrap().unwrap().unwrap(); + assert_eq!(late_accepted["queued"], true, "aged agent steer must enter the durable queue"); + assert_eq!(later_accepted["queued"], true, "a newer delivery cannot renew the turn"); + assert_eq!(human_accepted["steer"], true); + assert_eq!(human_answer["stopReason"], "steered"); + assert_eq!(explicit_accepted["steer"], true); + assert_eq!(explicit_answer["stopReason"], "steered"); + assert!(refused.is_err_and(|e| e.message.starts_with("steer.unavailable"))); + assert_eq!(still_running.as_deref(), Some(original_turn.as_str())); + assert_eq!( + pending.iter().map(|q| q.prompt_id.as_str()).collect::>(), + ["older-unsteered", "late-agent", "later-agent"] + ); + assert_eq!(original_answer["stopReason"], "end_turn", "admission must not cancel the tool"); + for (id, answer) in [ + ("older-unsteered", older_answer), + ("late-agent", late_answer), + ("later-agent", later_answer), + ] { + assert_eq!(answer["stopReason"], "end_turn"); + assert_eq!(answer["_meta"]["acpmux"]["promptId"], id); + assert_ne!(answer["_meta"]["acpmux"]["turnId"], original_turn); + } + assert!(session.turn().is_none()); + assert!(session.queue().is_empty()); + let records = hub.events(&session.id, 0, 100_000).unwrap(); + let sent: Vec<_> = + records.iter().filter(|r| r.dir == "out" && r.kind == "session/prompt").collect(); + for text in ["fresh-agent", "older-unsteered", "late-agent", "later-agent"] { + assert_eq!( + sent.iter().filter(|r| r.msg["params"]["prompt"].to_string().contains(text)).count(), + 1, + "{text} must be delivered once" + ); + } + assert!(!records.iter().any(|r| r.dir == "out" && r.kind == "session/cancel")); + let starts = records.iter().filter(|r| r.kind == "turn_started").count(); + let ends = records.iter().filter(|r| r.kind == "turn_result").count(); + assert_eq!(starts, ends, "each actual turn has exactly one final outcome"); +} diff --git a/cmux-tui/crates/acpmux/src/hub/turns.rs b/cmux-tui/crates/acpmux/src/hub/turns.rs index ef9b7e0cd091..bd78c53de624 100644 --- a/cmux-tui/crates/acpmux/src/hub/turns.rs +++ b/cmux-tui/crates/acpmux/src/hub/turns.rs @@ -4,6 +4,10 @@ use super::exit_reason::explains_exit; use super::*; use std::sync::atomic::Ordering; +#[cfg(all(test, unix))] +#[path = "agent_steer_window_tests.rs"] +mod agent_steer_window_tests; + /// What the agent of `session` is sent for `blocks`: model-sized copies of its large images /// (`model_copies.rs`), then [`acp_prompt_blocks`]. No copy is made, or kept, for a turn a remote /// device controls: it is read from this machine's disk as the prompt's sender, who was local. @@ -270,13 +274,30 @@ impl Hub { None => false, }; let text = prompt_text(&blocks); + // Automatic messages may extend a turn only inside its original steering window. + // Once it expires, let existing work finish and drain the durable queue. Human steers + // and an explicit Send now remain available; neither activity nor new messages renew it. + let agent = super::digest::agent_sender(opts.provenance.as_ref(), opts.agent.as_deref()); + let agent_window = if steer && agent.is_some() && !opts.steer_now { + Some(self.config.read().await.agent_steer_window()) + } else { + None + }; let running = session.turn(); - let steer_now = steer && session.steering.load(Ordering::SeqCst) && running.is_some(); + let defer_agent = agent_window + .zip(running.as_ref()) + .is_some_and(|(window, turn)| now_ms().saturating_sub(turn.started_at) >= window); + let steer_now = + steer && !defer_agent && session.steering.load(Ordering::SeqCst) && running.is_some(); if steer && opts.steer_only && !steer_now { - return Err(RpcError::invalid_params( - "steer.unavailable: the session has no running turn that its agent can steer", - ) - .with_data(json!({"reason": "steer.unavailable"}))); + let message = if defer_agent { + "steer.unavailable: the automatic agent steering window has ended" + } else { + "steer.unavailable: the session has no running turn that its agent can steer" + }; + return Err( + RpcError::invalid_params(message).with_data(json!({"reason": "steer.unavailable"})) + ); } if !live { self.cold_prompt_check(session).await?; @@ -284,8 +305,6 @@ impl Hub { if steer_now && session.moving().is_some() { return Err(super::queue::moving_error()); } - // Who sent it, as the queue knows its prompts' senders (`digest.rs`). - let agent = super::digest::agent_sender(opts.provenance.as_ref(), opts.agent.as_deref()); // What it asks of this chat: another agent's only, `result` when it named none. let class = crate::prompt_class::of_message(opts.class, agent.is_some()); if steer_now { @@ -306,7 +325,8 @@ impl Hub { prompt_id: &prompt_id, control, trust_gate, - interrupt: opts.interrupt, + // Budget deferral is cooperative: never use the non-steering interrupt fallback. + interrupt: opts.interrupt && !defer_agent, interrupt_turn: opts.interrupt_turn.as_deref(), provenance: opts.provenance.as_ref(), agent: agent.as_deref(),