serve: a session's end stops its bash jobs, in a daemon too

crates/agent/ARCHITECTURE.md

@@ -1396,8 +1396,12 @@ so a job that ends while the session is busy is never missed. A client prompt th
 arm is last in a biased `select!`): its run delivers the notice at its first tool results or its stop,
 and the arm then finds nothing to take, so no second turn runs. A transcript switch (`new_session`,
 `switch_session`, `fork`, …) stops the jobs — they belong to the old history — and the session's own
-end (stdin EOF, the daemon's reaper, a closed socket) kills them with its cell. Jobs do not survive a
-restart; `bash_job` on an id from before one says so.
+end (stdin EOF, the daemon's reaper, a closed socket, a supersede) stops them explicitly
+(`BashJobs::stop_all` on the blocking pool as `serve_session` returns, reported in a
+`bash_jobs_stopped` frame), so in a daemon, whose process lives on, nothing outlives its session. The
+table is kept out of what the cell's runner and backend clone (`CellRoute`): every running job's task
+holds the runner, and a runner that held the table would be a cycle that kept an ended session's jobs
+alive, unreachable. Jobs do not survive a restart; `bash_job` on an id from before one says so.
 
 ---
 

crates/agent/src/exec_endpoint.rs

@@ -1219,15 +1219,44 @@ impl ExecTarget {
 /// fall back to.
 #[derive(Clone, Default)]
 pub struct ExecCell {
+    route: CellRoute,
+    /// The session's `bash` jobs. Here because they share the cell's lifetime and reach: one per
+    /// session, handed to every registry rebuild (`build_tools` takes the cell), so a `set_model`
+    /// keeps them running.
+    ///
+    /// Deliberately *not* in [`CellRoute`], the part [`Self::runner`] and [`Self::backend`] clone:
+    /// every running job's task holds the runner, so a runner that held the table would make a
+    /// cycle (task → runner → table → job → task), and a session that ended without stopping its
+    /// jobs would leak them, unreachable, for the life of the process.
+    bash_jobs: std::sync::Arc<crate::tools::bash_jobs::BashJobs>,
+}
+
+/// Where an [`ExecCell`] dispatches: the part its runner and filesystem backend share.
+#[derive(Clone, Default)]
+struct CellRoute {
     target: std::sync::Arc<std::sync::RwLock<Option<ExecTarget>>>,
     /// Fail closed: an empty cell errors instead of running on this host. Service mode
     /// (`serve --service`) sets it, because there the host is the *replica* — a tenant's `bash`
     /// landing on it is not a stale-config bug but a breach. Fixed at construction.
     strict: bool,
-    /// The session's `bash` jobs. Here because they share the cell's lifetime and reach: one per
-    /// session, handed to every registry rebuild (`build_tools` takes the cell), so a `set_model`
-    /// keeps them running.
-    bash_jobs: std::sync::Arc<crate::tools::bash_jobs::BashJobs>,
+}
+
+impl CellRoute {
+    fn get(&self) -> Option<ExecTarget> {
+        self.target
+            .read()
+            .unwrap_or_else(|e| e.into_inner())
+            .clone()
+    }
+
+    /// The target to dispatch to, or — in strict mode with an empty cell — the refusal.
+    fn resolve(&self) -> Result<Option<ExecTarget>, &'static str> {
+        match (self.get(), self.strict) {
+            (Some(t), _) => Ok(Some(t)),
+            (None, false) => Ok(None),
+            (None, true) => Err(NO_TARGET),
+        }
+    }
 }
 
 /// What a strict cell answers when nothing is configured. Static text: it reaches the model as a
@@ -1245,7 +1274,10 @@ impl ExecCell {
     /// stale-config bug but a breach.
     pub fn strict() -> Self {
         Self {
-            strict: true,
+            route: CellRoute {
+                strict: true,
+                ..CellRoute::default()
+            },
             ..Self::default()
         }
     }
@@ -1259,39 +1291,27 @@ impl ExecCell {
     pub fn set(&self, target: Option<ExecTarget>) {
         // Recover a poisoned lock rather than panicking: the cell holds a single value with no
         // invariant a panicked writer could leave half-updated, and the workspace forbids `unwrap`.
-        *self.target.write().unwrap_or_else(|e| e.into_inner()) = target;
+        *self.route.target.write().unwrap_or_else(|e| e.into_inner()) = target;
     }
 
     pub fn get(&self) -> Option<ExecTarget> {
-        self.target
-            .read()
-            .unwrap_or_else(|e| e.into_inner())
-            .clone()
-    }
-
-    /// The target to dispatch to, or — in strict mode with an empty cell — the refusal.
-    fn resolve(&self) -> Result<Option<ExecTarget>, &'static str> {
-        match (self.get(), self.strict) {
-            (Some(t), _) => Ok(Some(t)),
-            (None, false) => Ok(None),
-            (None, true) => Err(NO_TARGET),
-        }
+        self.route.get()
     }
 
     /// A [`CommandRunner`] that always dispatches to whatever this cell currently holds.
     pub fn runner(&self) -> std::sync::Arc<dyn CommandRunner> {
-        std::sync::Arc::new(CellRunner(self.clone()))
+        std::sync::Arc::new(CellRunner(self.route.clone()))
     }
 
     /// An [`FsBackend`](crate::tools::fs::FsBackend) that always dispatches to whatever this cell
     /// currently holds.
     pub fn backend(&self) -> std::sync::Arc<dyn crate::tools::fs::FsBackend> {
-        std::sync::Arc::new(CellFs(self.clone()))
+        std::sync::Arc::new(CellFs(self.route.clone()))
     }
 }
 
 /// Dispatches each command to the cell's current target, or this host when it is empty.
-struct CellRunner(ExecCell);
+struct CellRunner(CellRoute);
 
 #[async_trait]
 impl CommandRunner for CellRunner {
@@ -1393,7 +1413,7 @@ impl CommandRunner for CellRunner {
 }
 
 /// The filesystem half of the same indirection.
-struct CellFs(ExecCell);
+struct CellFs(CellRoute);
 
 impl CellFs {
     fn inner(

crates/agent/src/serve.rs

@@ -2873,7 +2873,8 @@ pub(crate) async fn serve_session(
     };
     // `bash` jobs outlive the run that started them here: ending a turn leaves them running, and the
     // idle loop below starts a turn when one ends (see the `bash_jobs_wake` arm). They end with the
-    // session (its cell drops), and a transcript switch stops them (they belong to its history).
+    // session (stopped explicitly when this function returns), and a transcript switch stops them
+    // (they belong to its history).
     let bash_jobs = exec_cell.bash_jobs().clone();
     bash_jobs.set_policy(crate::tools::bash_jobs::EndPolicy::KeepRunning);
     let mut bash_jobs_wake = bash_jobs.subscribe();
@@ -8584,6 +8585,24 @@ pub(crate) async fn serve_session(
             }
         }
     }
+    // The session is over, however it ended (EOF, a reap, a supersede, shutdown): its `bash` jobs
+    // stop with it. Explicit rather than left to the table's drop: in a daemon the process lives
+    // on, and nothing else would ever reach these jobs again. Off the runtime thread, because each
+    // kill blocks until its processes are gone (on the daemon's current-thread runtime that would
+    // stall every other session).
+    let stopped_jobs = {
+        let jobs = bash_jobs.clone();
+        tokio::task::spawn_blocking(move || jobs.stop_all())
+            .await
+            .unwrap_or_default()
+    };
+    if !stopped_jobs.is_empty() {
+        // Best effort: the client may already be gone.
+        let _ = out_tx.send(OutFrame::Value(json!({
+            "type": "bash_jobs_stopped",
+            "jobs": stopped_jobs,
+        })));
+    }
     // Unsubscribe (webhook) / cancel (push) every MCP Events subscription before the session goes:
     // best effort and bounded, so a dead server cannot hold the teardown.
     if let Some(hub) = mcp_events.take() {

crates/agent/tests/bash_jobs_serve_daemon.rs

@@ -0,0 +1,120 @@
+//! `serve --listen` e2e: a daemon session's `bash` jobs end with the session, not with the process.
+//! A reaped session stops the job it left running; before the fix the job's own task kept the job
+//! table alive (task → runner → cell → table → job → task), so it ran on, unreachable, until the
+//! daemon exited.
+#![cfg(target_os = "linux")]
+#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
+
+mod common;
+
+use std::process::{Command, Stdio};
+use std::time::{Duration, Instant};
+
+use common::{
+    ChildGuard, ISOLATED_HOME, spawn_listening, spawn_model_server, turn_text, turn_tool_use,
+    ws_connect, ws_read_until_response, ws_send,
+};
+use serde_json::json;
+
+fn daemon(base: &str, dir: &str, idle_secs: u64) -> (ChildGuard, u16) {
+    spawn_listening(
+        Command::new(env!("CARGO_BIN_EXE_beyond-ai-agent"))
+            .args([
+                "serve",
+                "--gateway-url",
+                base,
+                "--key",
+                "bai_v1.test",
+                "--model",
+                "claude-test",
+                "--session-dir",
+                dir,
+                "--session-idle-timeout",
+                &idle_secs.to_string(),
+            ])
+            .env("HOME", ISOLATED_HOME)
+            .stdin(Stdio::null())
+            .stdout(Stdio::null()),
+    )
+}
+
+fn alive(pid: u32) -> bool {
+    let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).unwrap_or_default();
+    let state = stat
+        .rsplit_once(')')
+        .and_then(|(_, rest)| rest.split_whitespace().next())
+        .unwrap_or("");
+    !stat.is_empty() && state != "Z" && state != "X"
+}
+
+fn read_pid(path: &std::path::Path) -> u32 {
+    let deadline = Instant::now() + Duration::from_secs(10);
+    while Instant::now() < deadline {
+        if let Ok(text) = std::fs::read_to_string(path)
+            && let Ok(pid) = text.trim().parse()
+        {
+            return pid;
+        }
+        std::thread::sleep(Duration::from_millis(50));
+    }
+    panic!("{} was never written", path.display());
+}
+
+/// Kills the job's process if the test fails before the daemon did, so a failure leaks nothing.
+struct PidGuard(u32);
+
+impl Drop for PidGuard {
+    fn drop(&mut self) {
+        if alive(self.0) {
+            let _ = Command::new("kill")
+                .args(["-9", &self.0.to_string()])
+                .status();
+        }
+    }
+}
+
+#[tokio::test]
+async fn a_reaped_session_stops_its_jobs() {
+    let tmp = tempfile::tempdir().unwrap();
+    let pidfile = tmp.path().join("pid");
+    // `nohup … &`: the command exits at once, leaving the sleep running as an adopted job.
+    let command = format!(
+        "nohup sh -c 'echo $$ > {}; exec sleep 89' > /dev/null 2>&1 & echo up",
+        pidfile.display()
+    );
+    let (base, _bodies) = spawn_model_server(vec![
+        turn_tool_use("t1", "bash", &json!({ "command": command }).to_string()),
+        turn_text("it is up"),
+        // The one stop reminder.
+        turn_text("ok"),
+    ]);
+    let dir = tempfile::tempdir().unwrap();
+    let (mut child, port) = daemon(&base, dir.path().to_str().unwrap(), 1);
+    {
+        let mut ws = ws_connect(port, Some("reapjobs01")).await;
+        ws_send(
+            &mut ws,
+            json!({ "type": "prompt", "id": "p", "message": "start it" }),
+        )
+        .await;
+        let frames = ws_read_until_response(&mut ws, "prompt").await;
+        assert_eq!(frames.last().unwrap()["success"], true, "{frames:?}");
+    }
+    let pid = read_pid(&pidfile);
+    let _guard = PidGuard(pid);
+    assert!(alive(pid), "the job runs past its turn");
+
+    // Detached: the reaper ends the session after 1 s (plus a tick); its job goes with it, while
+    // the daemon itself keeps running.
+    let deadline = Instant::now() + Duration::from_secs(15);
+    while alive(pid) && Instant::now() < deadline {
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    }
+    assert!(
+        child.try_wait().unwrap().is_none(),
+        "the daemon is still up: the session's end, not the process's, stopped the job"
+    );
+    assert!(!alive(pid), "the reaped session's job is still running");
+    let _ = child.kill();
+    let _ = child.wait();
+}