fix(pool): force-evict hung sessions so zombie slots cannot exhaust the pool (#1300)

* refactor(pool): extract idle and eviction classification helpers

Pure helpers for cleanup_idle stale detection and get_or_create eviction
candidate selection, with characterization tests for current behavior.

* feat(pool): add lock-free session activity tracking

SessionActivity exposes last-active and in-flight state without the
connection mutex; pool state mirrors activity handles alongside cancel handles.

* fix(pool): force-evict hung sessions past hard-timeout grace

cleanup_idle detects in-flight sessions stuck beyond
prompt_hard_timeout_secs + hung_grace_secs and evicts them lock-free.

* test(pool): cover hung classification and eviction bookkeeping

Extract classify_hung and apply_hung_eviction helpers with unit tests;
add SessionActivity touch/in-flight tests.

* docs(config): document pool.hung_grace_secs

Document the hung-session grace period and apply rustfmt to touched acp files.

* fix(pool)!: make hung eviction reset-like and harden cancel path

Hung sessions are purged from suspended/persisted (not suspended for
resume): the old streaming task may still own an in-flight turn, so
resuming the same session id is unsafe. Eviction uses the connection
handle captured at classification time via remove_if_same_handle, and
session/cancel is sent from a detached 5s-timeout task outside the
state lock. SessionActivity gains Release/Acquire ordering, a
saturating age() helper, and a test-only last_active setter.

* feat(pool): wire hung threshold from config into SessionPool::new

SessionPool::new now takes hung_threshold_secs; main.rs passes
prompt_hard_timeout_secs + hung_grace_secs and the builder setter is
removed. Document hung_grace_secs in config.toml.example.

* fix(pool): never await existing connection mutex under the creating gate

Awaiting a hung connection's mutex in get_or_create while holding the
per-thread creating gate would permanently jam all future messages for
that thread. try_lock and treat lock-held as busy/alive (F1).

* fix(pool): use monotonic clock for session activity timestamps

A wall-clock step (NTP, manual change) could make an active session
look hours stale and trigger a false hung eviction. Track activity as
milliseconds since a process-boot Instant instead of SystemTime (F2).

* fix(pool): self-heal stale in_flight flag during cleanup sweep

If cleanup_idle can try_lock the connection, no turn is streaming, so a
true in_flight flag means the turn aborted without prompt_done. Reset it
to prevent a future false hung classification (F3).

* fix(pool): kill hung agent process group after eviction

The hung task never unwinds, so AcpConnection::Drop never fires and the
ACP child would leak forever. Track child pgids lock-free in PoolState
and, after the 5s session/cancel attempt, SIGTERM then SIGKILL the
process group from the detached task (F4). Also enrich the eviction
warn with session id, age_secs and threshold_secs (F7).

* fix(acp): bound stdin writes in send_raw with a 10s timeout

A hung agent can stop draining stdin; an unbounded write_all would then
block the caller (and any mutex it holds) forever (F5).

* fix(pool): keep the creating gate when purging a hung session

The creating gate is concurrency control, not session state: removing
it while a holder still owns the old gate Arc lets a concurrent
get_or_create mint a fresh gate and run two creations for the same key (F6).

* test(dispatch): derive hung threshold from config defaults

Replace the magic 30*60+120 literal with the actual serde default fns
so the test constant cannot drift from config defaults (F8).

Author: frankekn

config.toml.example

@@ -180,6 +180,9 @@ session_ttl_hours = 24
 # prompt_hard_timeout_secs = 1800
 # Liveness-check cadence (sec) for the recv loop; see #732. Default: 30.
 # liveness_check_secs = 30
+# Grace (sec) past prompt_hard_timeout_secs before a hung in-flight session is
+# force-evicted from the pool. Default: 120.
+# hung_grace_secs = 120
 
 [markdown]
 tables = "code"  # "code" (default) | "bullets" | "off"

crates/openab-core/src/acp/connection.rs

@@ -4,7 +4,7 @@ use crate::acp::protocol::{
 use anyhow::{anyhow, Result};
 use serde_json::{json, Value};
 use std::collections::HashMap;
-use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::Arc;
 use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader};
 use tokio::process::{Child, ChildStdin};
@@ -107,6 +107,68 @@ impl ContentBlock {
     }
 }
 
+/// Lock-free view of session activity, readable without the connection mutex.
+pub struct SessionActivity {
+    /// Milliseconds since process boot (monotonic) of the last observed activity.
+    last_active_ms: AtomicU64,
+    /// True while a prompt turn is in flight (mutex likely held).
+    prompt_in_flight: AtomicBool,
+}
+
+impl Default for SessionActivity {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl SessionActivity {
+    pub fn new() -> Self {
+        Self {
+            last_active_ms: AtomicU64::new(Self::now_ms()),
+            prompt_in_flight: AtomicBool::new(false),
+        }
+    }
+
+    /// Monotonic milliseconds since first use (process boot). SystemTime is
+    /// unsuitable here: a wall-clock step (NTP, manual change) could make an
+    /// active session look hours stale and trigger a false hung eviction.
+    fn now_ms() -> u64 {
+        use std::sync::OnceLock;
+        static BOOT: OnceLock<std::time::Instant> = OnceLock::new();
+        BOOT.get_or_init(std::time::Instant::now)
+            .elapsed()
+            .as_millis() as u64
+    }
+
+    pub fn touch(&self) {
+        self.last_active_ms.store(Self::now_ms(), Ordering::Release);
+    }
+
+    pub fn set_in_flight(&self, in_flight: bool) {
+        self.prompt_in_flight.store(in_flight, Ordering::Release);
+    }
+
+    /// Milliseconds since process boot of the last observed activity.
+    pub fn last_active_ms(&self) -> u64 {
+        self.last_active_ms.load(Ordering::Acquire)
+    }
+
+    /// Elapsed time since the last observed activity (saturating at zero).
+    pub fn age(&self) -> std::time::Duration {
+        let last = self.last_active_ms.load(Ordering::Acquire);
+        std::time::Duration::from_millis(Self::now_ms().saturating_sub(last))
+    }
+
+    pub fn in_flight(&self) -> bool {
+        self.prompt_in_flight.load(Ordering::Acquire)
+    }
+
+    #[cfg(test)]
+    pub(crate) fn set_last_active_ms(&self, ms: u64) {
+        self.last_active_ms.store(ms, Ordering::Release);
+    }
+}
+
 pub struct AcpConnection {
     _proc: Child,
     /// PID of the direct child, used as the process group ID for cleanup.
@@ -119,6 +181,7 @@ pub struct AcpConnection {
     pub supports_load_session: bool,
     pub config_options: Vec<ConfigOption>,
     pub last_active: Instant,
+    pub activity: Arc<SessionActivity>,
     pub session_reset: bool,
     _reader_handle: JoinHandle<()>,
     _stderr_handle: Option<JoinHandle<()>>,
@@ -371,7 +434,8 @@ impl AcpConnection {
                         Ok(_) => {
                             let trimmed = line.trim();
                             if !trimmed.is_empty() {
-                                let sanitized: String = trimmed.chars()
+                                let sanitized: String = trimmed
+                                    .chars()
                                     .filter(|c| !c.is_control() || *c == '\t')
                                     .collect();
                                 if !sanitized.is_empty() {
@@ -399,6 +463,8 @@ impl AcpConnection {
             notify_tx.clone(),
         ));
 
+        let activity = Arc::new(SessionActivity::new());
+
         Ok(Self {
             _proc: proc,
             child_pgid,
@@ -410,6 +476,7 @@ impl AcpConnection {
             supports_load_session: false,
             config_options: Vec::new(),
             last_active: Instant::now(),
+            activity,
             session_reset: false,
             _reader_handle: reader_handle,
             _stderr_handle: stderr_handle,
@@ -422,10 +489,17 @@ impl AcpConnection {
 
     pub(crate) async fn send_raw(&self, data: &str) -> Result<()> {
         debug!(data = data.trim(), "acp_send");
-        let mut w = self.stdin.lock().await;
-        w.write_all(data.as_bytes()).await?;
-        w.write_all(b"\n").await?;
-        w.flush().await?;
+        // A hung agent can stop draining stdin; bound the write so callers
+        // (and the mutexes they hold) can never block on it indefinitely.
+        tokio::time::timeout(std::time::Duration::from_secs(10), async {
+            let mut w = self.stdin.lock().await;
+            w.write_all(data.as_bytes()).await?;
+            w.write_all(b"\n").await?;
+            w.flush().await?;
+            Ok::<(), anyhow::Error>(())
+        })
+        .await
+        .map_err(|_| anyhow!("stdin write timeout"))??;
         Ok(())
     }
 
@@ -572,6 +646,8 @@ impl AcpConnection {
         content_blocks: Vec<ContentBlock>,
     ) -> Result<(mpsc::UnboundedReceiver<JsonRpcMessage>, u64)> {
         self.last_active = Instant::now();
+        self.activity.touch();
+        self.activity.set_in_flight(true);
 
         let session_id = self
             .acp_session_id
@@ -606,6 +682,8 @@ impl AcpConnection {
     /// Call after prompt streaming is done to clean up subscriber.
     pub async fn prompt_done(&mut self) {
         *self.notify_tx.lock().await = None;
+        self.activity.touch();
+        self.activity.set_in_flight(false);
         self.last_active = Instant::now();
     }
 
@@ -634,6 +712,15 @@ impl AcpConnection {
         Arc::clone(&self.stdin)
     }
 
+    pub fn activity_handle(&self) -> Arc<SessionActivity> {
+        Arc::clone(&self.activity)
+    }
+
+    /// Process-group id of the agent child, readable without any lock state.
+    pub fn child_pgid(&self) -> Option<i32> {
+        self.child_pgid
+    }
+
     pub fn alive(&self) -> bool {
         !self._reader_handle.is_finished()
     }
@@ -872,13 +959,10 @@ mod reader_loop_tests {
         agent_stdout_writer.write_all(stale).await.unwrap();
         agent_stdout_writer.flush().await.unwrap();
 
-        let forwarded = tokio::time::timeout(
-            std::time::Duration::from_secs(2),
-            sub_rx.recv(),
-        )
-        .await
-        .expect("subscriber should receive stale message before timeout")
-        .expect("subscriber channel should not be closed");
+        let forwarded = tokio::time::timeout(std::time::Duration::from_secs(2), sub_rx.recv())
+            .await
+            .expect("subscriber should receive stale message before timeout")
+            .expect("subscriber channel should not be closed");
         assert_eq!(forwarded.id, Some(42));
         assert!(pending.lock().await.is_empty());
 
@@ -934,4 +1018,33 @@ mod reader_loop_tests {
         drop(agent_stdout_writer);
         handle.await.unwrap();
     }
+
+    #[test]
+    fn session_activity_touch_advances_last_active() {
+        let activity = SessionActivity::new();
+        // Warm the monotonic clock past zero so a backdated value is older.
+        std::thread::sleep(std::time::Duration::from_millis(10));
+        activity.set_last_active_ms(0);
+        let before = activity.last_active_ms();
+        activity.touch();
+        assert!(activity.last_active_ms() > before);
+        // Backdated last_active yields a positive age; touch resets it near zero.
+        activity.set_last_active_ms(0);
+        assert!(activity.age() >= std::time::Duration::from_millis(10));
+        activity.touch();
+        assert!(activity.age() < std::time::Duration::from_secs(60));
+        // A future timestamp must not underflow: age saturates at zero.
+        activity.set_last_active_ms(u64::MAX);
+        assert_eq!(activity.age(), std::time::Duration::ZERO);
+    }
+
+    #[test]
+    fn session_activity_set_in_flight_round_trips() {
+        let activity = SessionActivity::new();
+        assert!(!activity.in_flight());
+        activity.set_in_flight(true);
+        assert!(activity.in_flight());
+        activity.set_in_flight(false);
+        assert!(!activity.in_flight());
+    }
 }