serve: a host bash command's leftovers are the client's, never the model's

crates/agent/ARCHITECTURE.md

@@ -2599,7 +2599,14 @@ and `serve`'s own `bash`/`abort_bash` RPC commands reuse `RealRunner` directly,
     things that stop a job: there is no per-command kill deadline (see "`bash` timeouts").
   - **One table per owner.** `ToolConfig::bash_jobs` hands `bash` the host's table: `run`'s, or the
     `serve` session's (kept on its `ExecCell`, so a `set_model` rebuild keeps the jobs). `None` (a
-    subagent) is a table of the registry's own.
+    subagent) is a table of the registry's own. `serve`'s host `bash` command (the client's, not the
+    model's) gets a third: `host_bash_jobs`, a table of the session's that nothing takes notices from.
+    What a client's `cmd &` leaves running is parked there, so it never wakes the model (a paid turn
+    nobody asked for, quoting output the client may have excluded from context), never appears in a
+    notice, reminder or carry, and the result says "it runs until the session ends" instead of
+    `bash_job` instructions (an instance that never detaches has no `bash_job`). It survives a
+    transcript switch (the processes are the client's, not a history's) and is stopped, unreported,
+    when the session ends; ended entries are dropped at the next host command (`forget_ended`).
   - **Compaction carries the table** (`Background::carry`): running and ended-unread jobs (id,
     command, state, age) as a `<background-jobs>` block on every summary, so a compacted model does not
     start a job again. Rendered from the live table, not persisted in `CompactionProvenance`: after a

crates/agent/src/serve.rs

@@ -2914,6 +2914,15 @@ pub(crate) async fn serve_session(
     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();
+    // What the host `bash` command (the client's own, not the model's) leaves running (`cmd &`).
+    // A table of its own, so those processes never reach the model: no finish notice, no wake (a
+    // paid turn the client never asked for, quoting output it may have excluded from context), no
+    // `<background-jobs>` carry. Nothing reads notices from it; they live until they exit or the
+    // session ends, and a transcript switch leaves them alone (they belong to the client, not to a
+    // history).
+    let host_bash_jobs = Arc::new(crate::tools::bash_jobs::BashJobs::new(
+        crate::tools::bash_jobs::EndPolicy::KeepRunning,
+    ));
     let mut bash_jobs_transcript = persistence.transcript();
     // Per-session MCP kit gate (which configured servers' tools are advertised). Defaults to all
     // enabled — matching prior behavior — and is cleared on session switch so tenants cannot inherit
@@ -3078,7 +3087,7 @@ pub(crate) async fn serve_session(
         &exec_cell,
         &mcp_enabled,
         &skills,
-        false,
+        None,
     );
     let has_read = startup_tools.get("read").is_some();
     let has_todo = startup_tools.get(crate::tools::todo::NAME).is_some();
@@ -3644,7 +3653,7 @@ pub(crate) async fn serve_session(
         &exec_cell,
         &mcp_enabled,
         &skills,
-        true,
+        Some(&host_bash_jobs),
     )
     .get("bash");
     // The same deny-list `build_agent` installs as an `AgentHooks` gate on the model's own tool calls —
@@ -6182,7 +6191,7 @@ pub(crate) async fn serve_session(
                     &exec_cell,
                     &mcp_enabled,
                     &skills,
-                    false,
+                    None,
                 )
                 .names()
                 .into_iter()
@@ -6713,7 +6722,7 @@ pub(crate) async fn serve_session(
                     &exec_cell,
                     &mcp_enabled,
                     &skills,
-                    false,
+                    None,
                 );
                 let mut tool_defs = registry.definitions();
                 tool_defs.extend(registry.deferred_definitions());
@@ -8111,6 +8120,9 @@ pub(crate) async fn serve_session(
                     ));
                     continue;
                 };
+                // Nobody reads the client's table for notices, so its ended jobs are dropped here
+                // rather than kept for a `bash_job` that cannot reach them.
+                host_bash_jobs.forget_ended();
                 let mut input = json!({ "command": command });
                 if let Some(cwd) = cmd.get("cwd").and_then(Value::as_str) {
                     input["cwd"] = json!(cwd);
@@ -8635,9 +8647,14 @@ pub(crate) async fn serve_session(
     // stall every other session).
     let stopped_jobs = {
         let jobs = bash_jobs.clone();
-        tokio::task::spawn_blocking(move || jobs.stop_all())
-            .await
-            .unwrap_or_default()
+        let host_jobs = host_bash_jobs.clone();
+        tokio::task::spawn_blocking(move || {
+            // The client's are stopped too, but not reported as the model's jobs.
+            host_jobs.stop_all();
+            jobs.stop_all()
+        })
+        .await
+        .unwrap_or_default()
     };
     if !stopped_jobs.is_empty() {
         // Best effort: the client may already be gone.
@@ -8863,7 +8880,7 @@ fn build_subagent_ctx(
         exec,
         &crate::tools::mcp::McpEnabledSet::new(),
         skills,
-        false,
+        None,
     );
     crate::tools::apply_filter(
         &mut probe,
@@ -8989,7 +9006,7 @@ fn build_agent(
         ..compaction_config(cfg, model)
     };
 
-    let mut registry = build_tools(cfg, image_auto_resize, exec, mcp_enabled, skills, false);
+    let mut registry = build_tools(cfg, image_auto_resize, exec, mcp_enabled, skills, None);
     // Registered here, not in `build_tools`, because this is where `write_locks` is in scope — the ctx
     // needs the same registry every child shares. Reused across rebuilds; the ctx's factory handles the
     // per-child model itself, so a `set_model` rebuild doesn't invalidate it.
@@ -9302,15 +9319,16 @@ fn tool_root(cfg: &ServeConfig) -> std::path::PathBuf {
         .unwrap_or_default()
 }
 
-/// `bash_waits_for_exit`: see [`tools::ToolConfig::bash_waits_for_exit`] — only the host `bash`
-/// command's own instance sets it.
+/// `host_bash_jobs`: `Some` only for the host `bash` command's own instance, which waits for every
+/// command to exit (see [`tools::ToolConfig::bash_waits_for_exit`]) and parks what a command leaves
+/// running in this table — the client's, not the model's (see `host_bash_jobs` in `serve_session`).
 fn build_tools(
     cfg: &ServeConfig,
     image_auto_resize: bool,
     exec: &crate::exec_endpoint::ExecCell,
     mcp_enabled: &crate::tools::mcp::McpEnabledSet,
     skills: &[crate::skills::Skill],
-    bash_waits_for_exit: bool,
+    host_bash_jobs: Option<&Arc<crate::tools::bash_jobs::BashJobs>>,
 ) -> agent_core::ToolRegistry {
     // Task #34 (pi-parity fix): previously always called the plain `default_registry_with_prefix`,
     // hardcoding `image_auto_resize: true` regardless of `cfg.image_auto_resize` — `run`'s identical
@@ -9327,7 +9345,7 @@ fn build_tools(
     let empty: &[String] = &[];
     let nested_exclude = cfg.exclude_tools.as_deref().unwrap_or(empty);
     let mut registry = tools::default_registry_with_config(&tools::ToolConfig {
-        bash_waits_for_exit,
+        bash_waits_for_exit: host_bash_jobs.is_some(),
         bash_shell_path: cfg.bash_shell_path.as_deref(),
         bash_command_prefix: cfg.bash_command_prefix.as_deref(),
         web_allow_private: cfg.web_allow_private,
@@ -9346,9 +9364,14 @@ fn build_tools(
         fs_backend: Some(exec.backend()),
         command_runner: Some(exec.runner()),
         skills,
-        // The session's, not this registry's: a rebuild (`set_model`) keeps the jobs, the host
-        // `bash` command adopts into the same table, and the idle session wakes on it.
-        bash_jobs: Some(exec.bash_jobs().clone()),
+        // The session's, not this registry's: a rebuild (`set_model`) keeps the jobs, and the idle
+        // session wakes on them. The host `bash` command's instance gets the client's table instead:
+        // what a client's command leaves running is not the model's to be told about.
+        bash_jobs: Some(
+            host_bash_jobs
+                .cloned()
+                .unwrap_or_else(|| exec.bash_jobs().clone()),
+        ),
         // In service mode a relative path the model writes belongs in the tenant's workspace, not
         // wherever this replica's process happens to have been started. Empty (= the process cwd)
         // for every other caller, which is the pre-service behavior.

crates/agent/src/tools/bash.rs

@@ -825,7 +825,24 @@ impl Bash {
         let shell_status = exit_status_line(shell, job.started.elapsed().as_secs());
         job.kind = JobKind::Adopted;
         let lifetime = self.jobs.lifetime(job.kind);
+        // An instance that never detaches has no `bash_job` beside it and nobody to tell when the
+        // processes exit: `serve`'s host `bash`, whose caller is a client, not the model.
         let note = match self.jobs.park_intended(job) {
+            Ok(_) if !self.detaches() => {
+                format!(
+                    "The command exited but left {left} running; it runs until the session ends."
+                )
+            }
+            Err((job, _)) if !self.detaches() => {
+                if let Some(group) = job.shared.untrack() {
+                    let _ = tokio::task::spawn_blocking(move || group.kill()).await;
+                }
+                drop(job);
+                format!(
+                    "The command exited but left {left} running. It was stopped: {MAX_INTENDED_JOBS} \
+                     commands' leftover processes are already running in this session."
+                )
+            }
             Ok(id) => format!(
                 "The command exited but left {left} running; it is tracked as job {id} \
                  ({lifetime}). You will be told when it exits. bash_job {{\"job\": {id}, \"kill\": \

crates/agent/src/tools/bash_jobs.rs

@@ -398,6 +398,21 @@ impl BashJobs {
         lock(&self.table).remove(&id).map(|p| p.job)
     }
 
+    /// Drop every ended job. For a table whose notices nobody takes (`serve`'s host `bash`): this
+    /// is the only retention it gets, so ended jobs do not pile up for the life of the session.
+    pub fn forget_ended(&self) {
+        let ended: Vec<Parked> = {
+            let mut table = lock(&self.table);
+            let ids: Vec<u64> = table
+                .iter()
+                .filter(|(_, p)| p.job.shared.ended())
+                .map(|(id, _)| *id)
+                .collect();
+            ids.iter().filter_map(|id| table.remove(id)).collect()
+        };
+        drop(ended);
+    }
+
     /// Stop every running job, oldest first, and say which. Ended jobs are dropped too: nobody will
     /// read them now. For the end of a `run` and a transcript switch in `serve`.
     pub fn stop_all(&self) -> Vec<StoppedJob> {

crates/agent/tests/bash_jobs_serve_host.rs

@@ -0,0 +1,135 @@
+//! `serve` e2e: what a client's host `bash` command leaves running (`cmd &`) is the client's, not the
+//! model's. It used to be adopted into the model's job table, so its end woke the idle session — an
+//! unrequested, paid model turn whose input quoted the command's output, even with
+//! `exclude_from_context: true`. Now it lives in a table of the session's own that nothing wakes on,
+//! and ends with the session.
+#![cfg(target_os = "linux")]
+#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
+
+mod common;
+
+use std::io::Write;
+use std::process::{ChildStdin, Command};
+use std::time::{Duration, Instant};
+
+use common::{SpawnGuarded, read_until_response, serve_cmd, spawn_model_server, turn_text};
+use serde_json::{Value, json};
+
+const BIN: &str = env!("CARGO_BIN_EXE_beyond-ai-agent");
+
+fn send(stdin: &mut ChildStdin, cmd: Value) {
+    writeln!(stdin, "{cmd}").unwrap();
+    stdin.flush().unwrap();
+}
+
+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 process if the test fails before `serve` 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();
+        }
+    }
+}
+
+#[test]
+fn a_host_bash_leftover_never_wakes_the_model() {
+    let dir = tempfile::tempdir().unwrap();
+    let tmp = tempfile::tempdir().unwrap();
+    let pidfile = tmp.path().join("pid");
+    let session_file = dir.path().join("s.jsonl").to_string_lossy().into_owned();
+    let (base, bodies) = spawn_model_server(vec![turn_text("hello"), turn_text("spare")]);
+    let mut child = serve_cmd(BIN, &base, &session_file).spawn_guarded();
+    let mut stdin = child.stdin.take().unwrap();
+    let mut stdout = common::child_frames(&mut child);
+
+    send(
+        &mut stdin,
+        json!({
+            "type": "bash",
+            "id": "b1",
+            "command": "(sleep 1; echo CLIENT-SECRET-OUTPUT) &",
+            "exclude_from_context": true,
+        }),
+    );
+    let frames = read_until_response(&mut stdout, "bash");
+    let result = frames.last().unwrap()["data"]["result"].to_string();
+    assert!(result.contains("runs until the session ends"), "{result}");
+    assert!(
+        !result.contains("bash_job"),
+        "no model-facing instructions in a client's result: {result}"
+    );
+
+    // The leftover exits after 1 s: nothing wakes (a wake would settle 200 ms later).
+    let deadline = Instant::now() + Duration::from_secs(3);
+    while let Some(left) = deadline.checked_duration_since(Instant::now()) {
+        if let Ok(frame) = stdout.next_frame(left) {
+            assert!(
+                frame["command"] != "prompt" && frame["event"]["kind"] != "background",
+                "the client's command woke the model: {frame}"
+            );
+        }
+    }
+    assert_eq!(bodies.lock().unwrap().len(), 0, "no model call");
+
+    // Nor does the model's next turn hear of it.
+    send(
+        &mut stdin,
+        json!({ "type": "prompt", "id": "p1", "message": "hi" }),
+    );
+    let frames = read_until_response(&mut stdout, "prompt");
+    assert_eq!(frames.last().unwrap()["success"], true, "{frames:?}");
+    let body = bodies.lock().unwrap()[0].clone();
+    assert!(!body.contains("CLIENT-SECRET-OUTPUT"), "{body}");
+    assert!(!body.contains("[job "), "{body}");
+
+    // A leftover still running ends with the session.
+    send(
+        &mut stdin,
+        json!({
+            "type": "bash",
+            "id": "b2",
+            "command": format!("sh -c 'echo $$ > {}; exec sleep 60' &", pidfile.display()),
+        }),
+    );
+    let frames = read_until_response(&mut stdout, "bash");
+    assert_eq!(frames.last().unwrap()["success"], true, "{frames:?}");
+    let pid = read_pid(&pidfile);
+    let _guard = PidGuard(pid);
+    assert!(alive(pid));
+    drop(stdin);
+    let _ = child.wait();
+    let deadline = Instant::now() + Duration::from_secs(5);
+    while alive(pid) && Instant::now() < deadline {
+        std::thread::sleep(Duration::from_millis(50));
+    }
+    assert!(
+        !alive(pid),
+        "the session's end leaves the client's leftover running"
+    );
+}