From ffadcd7b2360157fe00370fe10adf8b8013667f0 Mon Sep 17 00:00:00 2001 From: Jared Lunde Date: Thu, 15 Oct 2026 23:26:49 -0700 Subject: [PATCH] serve: a session's end stops its bash jobs, in a daemon too A reaped (or otherwise ended) daemon session never stopped its bash jobs. The cell's runner cloned the whole ExecCell, job table included, and every running job's task holds that runner: task -> runner -> cell -> table -> job -> task. The table never dropped, so the job ran on, unreachable and silent, until the daemon exited (the stdio test passed only because process exit runs kill_live_jobs). - ExecCell splits its dispatch half (CellRoute: target + strict) from the job table; the runner and filesystem backend clone only the route, so there is no cycle. - serve_session stops its jobs explicitly as it returns (stop_all on the blocking pool, reported in a bash_jobs_stopped frame), whatever ended it. Test: bash_jobs_serve_daemon.rs reaps a detached daemon session and checks its adopted `nohup sleep &` dies while the daemon stays up (fails on main after 15 s). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01JimHGjsfk2Ktm5GxyZJKKk --- crates/agent/ARCHITECTURE.md | 8 +- crates/agent/src/exec_endpoint.rs | 66 ++++++---- crates/agent/src/serve.rs | 21 +++- crates/agent/tests/bash_jobs_serve_daemon.rs | 120 +++++++++++++++++++ 4 files changed, 189 insertions(+), 26 deletions(-) create mode 100644 crates/agent/tests/bash_jobs_serve_daemon.rs diff --git a/crates/agent/ARCHITECTURE.md b/crates/agent/ARCHITECTURE.md index a57c82cc..07368987 100644 --- a/crates/agent/ARCHITECTURE.md +++ b/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. --- diff --git a/crates/agent/src/exec_endpoint.rs b/crates/agent/src/exec_endpoint.rs index 113adb27..728885e2 100644 --- a/crates/agent/src/exec_endpoint.rs +++ b/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, +} + +/// Where an [`ExecCell`] dispatches: the part its runner and filesystem backend share. +#[derive(Clone, Default)] +struct CellRoute { target: std::sync::Arc>>, /// 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, +} + +impl CellRoute { + fn get(&self) -> Option { + 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, &'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) { // 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 { - 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, &'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 { - 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 { - 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( diff --git a/crates/agent/src/serve.rs b/crates/agent/src/serve.rs index f450654a..f8ab2609 100644 --- a/crates/agent/src/serve.rs +++ b/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() { diff --git a/crates/agent/tests/bash_jobs_serve_daemon.rs b/crates/agent/tests/bash_jobs_serve_daemon.rs new file mode 100644 index 00000000..3e526978 --- /dev/null +++ b/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(); +}