fix(acp): clean up pending + cancel agent on abandoned prompts (#732)

The flat 600s recv_timeout in adapter.rs:386 fires "Agent stopped responding"
without removing pending[id] or sending session/cancel. The agent keeps
running the abandoned prompt and eventually emits its final response with
the original id. The reader at connection.rs:284 looks up pending[id], sees
the now-stale entry, and forwards the message to the *current* notify_tx
subscriber — which belongs to the next prompt. The next prompt's loop sees
notification.id.is_some() and breaks immediately with empty text_buf,
returning "(no response)". Each new prompt sent before the agent drains its
backlog inherits the previous prompt's stale id and the cascade persists.

Fix follows the issue's recommended A+B+C:

(A) Replace flat 600s timeout with a tokio::select! loop in stream_prompt_blocks.
    Recv arm + 30s liveness arm. Liveness arm checks conn.alive() (cheap,
    just !reader_handle.is_finished()) and a configurable hard ceiling.
    Default ceiling is 30 min via pool.prompt_hard_timeout_secs. Long-running
    tools no longer trip the timeout — only a dead reader task or the hard
    ceiling abandon the prompt.

(B) Add AcpConnection::abandon_request(request_id) called on every abandon
    path: drops pending[request_id] so a late response cannot route to a
    future subscriber, and best-effort writes session/cancel so the agent
    stops working on a request the broker has given up on.

(C) Capture request_id from session_prompt() (was discarded as `_`) and
    skip notifications whose id doesn't match. Defense-in-depth at the
    routing layer; complements (B)'s cleanup if any future abandon path
    forgets to call abandon_request.

No unit test for abandon_request — the connection has no test seam without
spawning a real subprocess. Behavior is exercised end-to-end via the
adapter loop on real ACP backends.

Refs:
- #76 (Assumption 2: prompts always complete)
- #307 (sibling: same family, different visible symptom)
- #470 (added the 600s recv timeout this issue exposes)

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: 2 people

config.toml.example

@@ -104,6 +104,8 @@ working_dir = "/home/agent"
 [pool]
 max_sessions = 10
 session_ttl_hours = 24
+# Hard ceiling (sec) per prompt; see #732. Default: 1800 (30 min).
+# prompt_hard_timeout_secs = 1800
 
 [markdown]
 tables = "code"  # "code" (default) | "bullets" | "off"

src/acp/connection.rs

@@ -557,6 +557,24 @@ impl AcpConnection {
         self.last_active = Instant::now();
     }
 
+    /// Drop the pending entry for `request_id` and best-effort send
+    /// `session/cancel`. Errors are swallowed: the agent process may already
+    /// be dead, in which case the stdin write fails harmlessly. See #732.
+    pub async fn abandon_request(&self, request_id: u64) {
+        self.pending.lock().await.remove(&request_id);
+        let Some(session_id) = self.acp_session_id.as_deref() else {
+            return;
+        };
+        let req = json!({
+            "jsonrpc": "2.0",
+            "method": "session/cancel",
+            "params": {"sessionId": session_id},
+        });
+        if let Ok(data) = serde_json::to_string(&req) {
+            let _ = self.send_raw(&data).await;
+        }
+    }
+
     /// Return a clone of the stdin handle for lock-free cancel.
     pub fn cancel_handle(&self) -> Arc<Mutex<ChildStdin>> {
         Arc::clone(&self.stdin)

src/adapter.rs

@@ -153,24 +153,30 @@ pub trait ChatAdapter: Send + Sync + 'static {
 
 // --- AdapterRouter ---
 
+/// Polling cadence for the recv-loop liveness check (#732).
+const LIVENESS_CHECK_INTERVAL: std::time::Duration = std::time::Duration::from_secs(30);
+
 /// Shared logic for routing messages to ACP agents, managing sessions,
 /// streaming edits, and controlling reactions. Platform-independent.
 pub struct AdapterRouter {
     pool: Arc<SessionPool>,
     reactions_config: ReactionsConfig,
     table_mode: TableMode,
+    prompt_hard_timeout: std::time::Duration,
 }
 
 impl AdapterRouter {
     pub fn new(
         pool: Arc<SessionPool>,
         reactions_config: ReactionsConfig,
         table_mode: TableMode,
+        prompt_hard_timeout_secs: u64,
     ) -> Self {
         Self {
             pool,
             reactions_config,
             table_mode,
+            prompt_hard_timeout: std::time::Duration::from_secs(prompt_hard_timeout_secs),
         }
     }
 
@@ -335,6 +341,7 @@ impl AdapterRouter {
         let streaming = adapter.use_streaming(other_bot_present);
         let table_mode = self.table_mode;
         let tool_display = self.reactions_config.tool_display;
+        let prompt_hard_timeout = self.prompt_hard_timeout;
 
         self.pool
             .with_connection(thread_key, |conn| {
@@ -343,7 +350,7 @@ impl AdapterRouter {
                     let reset = conn.session_reset;
                     conn.session_reset = false;
 
-                    let (mut rx, _) = conn.session_prompt(content_blocks).await?;
+                    let (mut rx, request_id) = conn.session_prompt(content_blocks).await?;
                     reactions.set_thinking().await;
 
                     let mut text_buf = String::new();
@@ -396,20 +403,40 @@ impl AdapterRouter {
                         (None, None)
                     };
 
-                    // Process ACP notifications
+                    // (#732) Liveness-aware recv loop. Filters stale id-bearing
+                    // messages and abandons cleanly on dead agent / hard ceiling
+                    // so late responses cannot leak into the next prompt.
                     let mut response_error: Option<String> = None;
-                    let recv_timeout = std::time::Duration::from_secs(600);
+                    let prompt_start = std::time::Instant::now();
                     loop {
-                        let notification = match tokio::time::timeout(recv_timeout, rx.recv()).await
-                        {
-                            Ok(Some(n)) => n,
-                            Ok(None) => break, // channel closed
-                            Err(_) => {
-                                response_error = Some("Agent stopped responding".into());
-                                break;
+                        let notification = tokio::select! {
+                            msg = rx.recv() => match msg {
+                                Some(n) => n,
+                                // Reader saw EOF and already drained pending; nothing to abandon.
+                                None => break,
+                            },
+                            _ = tokio::time::sleep(LIVENESS_CHECK_INTERVAL) => {
+                                if !conn.alive() {
+                                    response_error = Some("Agent process died".into());
+                                    conn.abandon_request(request_id).await;
+                                    break;
+                                }
+                                if prompt_start.elapsed() > prompt_hard_timeout {
+                                    response_error = Some(format!(
+                                        "Agent exceeded hard timeout ({}m)",
+                                        prompt_hard_timeout.as_secs() / 60,
+                                    ));
+                                    conn.abandon_request(request_id).await;
+                                    break;
+                                }
+                                continue;
                             }
                         };
-                        if notification.id.is_some() {
+                        if let Some(notification_id) = notification.id {
+                            if notification_id != request_id {
+                                // Stale response from a previously-abandoned prompt.
+                                continue;
+                            }
                             if let Some(ref err) = notification.error {
                                 response_error = Some(format_coded_error(err.code, &err.message));
                             }

src/config.rs

@@ -313,6 +313,12 @@ pub struct PoolConfig {
     pub max_sessions: usize,
     #[serde(default = "default_ttl_hours")]
     pub session_ttl_hours: u64,
+    /// Hard ceiling for a single prompt (#732). Once exceeded, the broker
+    /// abandons the in-flight request, sends `session/cancel` to the agent,
+    /// and clears the pending entry so late responses cannot leak into the
+    /// next prompt's subscriber.
+    #[serde(default = "default_prompt_hard_timeout_secs")]
+    pub prompt_hard_timeout_secs: u64,
 }
 
 #[derive(Debug, Clone, Deserialize)]
@@ -434,6 +440,9 @@ fn default_max_sessions() -> usize {
 fn default_ttl_hours() -> u64 {
     4
 }
+pub(crate) fn default_prompt_hard_timeout_secs() -> u64 {
+    30 * 60
+}
 fn default_true() -> bool {
     true
 }
@@ -481,6 +490,7 @@ impl Default for PoolConfig {
         Self {
             max_sessions: default_max_sessions(),
             session_ttl_hours: default_ttl_hours(),
+            prompt_hard_timeout_secs: default_prompt_hard_timeout_secs(),
         }
     }
 }

src/dispatch.rs

@@ -1072,6 +1072,7 @@ mod tests {
             pool,
             crate::config::ReactionsConfig::default(),
             crate::markdown::TableMode::Off,
+            crate::config::default_prompt_hard_timeout_secs(),
         ));
         Dispatcher::with_idle_timeout(router, 10, 24_000, grouping, DEFAULT_CONSUMER_IDLE_TIMEOUT)
     }

src/main.rs

@@ -147,6 +147,7 @@ async fn main() -> anyhow::Result<()> {
         pool.clone(),
         cfg.reactions,
         cfg.markdown.tables,
+        cfg.pool.prompt_hard_timeout_secs,
     ));
 
     // Shutdown signal for Slack adapter