fix(acpmux): queue automatic messages after the original steering window

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

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<u64>,
+    /// `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<u64>,
     /// `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<std::time::Duration> {
         match self.stalled_turn_end_ms.unwrap_or(6 * 60 * 60 * 1000) {

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<Value> {
+    vec![json!({"type": "text", "text": text})]
+}
+
+async fn admitted(
+    hub: &Arc<Hub>,
+    session: &Arc<Session>,
+    text: &'static str,
+    steer: bool,
+    mut opts: PromptOptions,
+) -> (tokio::task::JoinHandle<Result<Value, RpcError>>, 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::<Vec<_>>(),
+        ["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");
+}

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,22 +274,37 @@ 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?;
         }
         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(),