bash: a cancelled bash_job wait parks the job again instead of killing it

crates/agent/ARCHITECTURE.md

@@ -2455,8 +2455,11 @@ and `serve`'s own `bash`/`abort_bash` RPC commands reuse `RealRunner` directly,
   carries the signal alongside the code (`exec::exit_signal`, Unix-only) and `bash` reports it as a failure
   naming the signal. Previously that `None` shared an arm with `Some(0)` and an OOM-killed build, a SIGSEGV,
   or a SIGTERM'd server all reported **success** to the model — silently, with `(no output)` standing in for
-  the output the command never got to produce. A cancellation kills the job like a `bash_job` kill
-  (dropping the `Job`), so a backgrounded grandchild process can't outlive it. Cancellation is
+  the output the command never got to produce. A cancellation kills the call's own fresh command like
+  a `bash_job` kill (dropping the `Job`), so a backgrounded grandchild process can't outlive it. A job
+  that already has a number is different: cancelling a `bash_job` wait on it (an `abort`, or a
+  command that ends the run) parks it again and returns "Wait cancelled; job N is still running.",
+  because the model was told how long that job lives and a cancelled wait must not end it. Cancellation is
   cooperative, not just an external `Drop`: `attach` races the running job against the call's own
   `ToolProgress::cancelled()` (the same `CancellationToken` a caller like `serve`'s `abort_bash` trips),
   and on a trip flushes whatever the always-current `OutputAccumulator` already captured before

crates/agent/src/tools/bash.rs

@@ -561,8 +561,8 @@ impl Bash {
 
     /// Wait on `job` until it ends (its final result), its shell exits leaving processes running
     /// (adopt them as a job, if the call was a foreground one), `hold` says to hand control back
-    /// (park it and report the output since the last report), or the call is cancelled (kill it and
-    /// return what it printed, tagged "Command cancelled").
+    /// (park it and report the output since the last report), or the call is cancelled (kill a fresh
+    /// command and return what it printed, tagged "Command cancelled"; park a numbered job again).
     async fn attach(
         &self,
         job: Job,
@@ -625,12 +625,24 @@ impl Bash {
                 }
                 Woke::Cancelled => {
                     let snap = job.shared.take_segment();
-                    // Dropping the job kills its processes (see `Job`'s `Drop`).
-                    drop(job);
                     if let Some(p) = progress {
                         emit_update(p, &snap);
                     }
                     let text = clean(format_output(&snap, ""));
+                    // A job with a number is a `bash_job` wait being cancelled (an abort, or a
+                    // command that ends the run): the model was told how long that job lives, and
+                    // cancelling a wait on it must not end it. It goes back in the table; only a
+                    // call's own fresh command dies with the call.
+                    if let Some(id) = job.id {
+                        // An evicted yielded job is dropped (killed) here, outside the table lock.
+                        let _evicted = self.jobs.park(job);
+                        return Err(ToolError::Execution(append_status(
+                            &text,
+                            &format!("Wait cancelled; job {id} is still running."),
+                        )));
+                    }
+                    // Dropping the job kills its processes (see `Job`'s `Drop`).
+                    drop(job);
                     return Err(ToolError::Execution(append_status(
                         &text,
                         "Command cancelled",

crates/agent/tests/bash_jobs_serve_abort.rs

@@ -0,0 +1,147 @@
+//! `serve` e2e: an `abort` while the model waits on a job with `bash_job` cancels the wait, not the
+//! job. The model was told a background job "keeps running after your turn ends"; the cancelled wait
+//! used to drop the job it had taken out of the table, killing it with no `bash_jobs_stopped` frame,
+//! and the next `bash_job` found nothing.
+#![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, turn_tool_use,
+};
+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 job's 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 an_abort_during_a_bash_job_wait_keeps_the_job() {
+    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 start = json!({
+        "command": format!("echo $$ > {}; exec sleep 60", pidfile.display()),
+        "run_in_background": true,
+    });
+    let (base, _bodies) = spawn_model_server(vec![
+        turn_tool_use("t1", "bash", &start.to_string()),
+        turn_text("started"),
+        // The one stop reminder.
+        turn_text("ok, it runs"),
+        // p2: wait on it (aborted).
+        turn_tool_use("t2", "bash_job", &json!({ "job": 1 }).to_string()),
+        // p3: peek at it.
+        turn_tool_use(
+            "t3",
+            "bash_job",
+            &json!({ "job": 1, "timeout_ms": 0 }).to_string(),
+        ),
+        turn_text("peeked"),
+        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": "prompt", "id": "p1", "message": "start it" }),
+    );
+    let frames = read_until_response(&mut stdout, "prompt");
+    assert_eq!(frames.last().unwrap()["success"], true, "{frames:?}");
+    let pid = read_pid(&pidfile);
+    let _guard = PidGuard(pid);
+    assert!(alive(pid));
+
+    send(
+        &mut stdin,
+        json!({ "type": "prompt", "id": "p2", "message": "wait for it" }),
+    );
+    loop {
+        let frame = stdout.next_frame(Duration::from_secs(20)).unwrap();
+        if frame["event"]["kind"] == "tool_start" && frame["event"]["name"] == "bash_job" {
+            break;
+        }
+    }
+    std::thread::sleep(Duration::from_millis(300));
+    send(&mut stdin, json!({ "type": "abort", "id": "a1" }));
+    let frames = read_until_response(&mut stdout, "prompt");
+    assert!(
+        !frames.iter().any(|f| f["type"] == "bash_jobs_stopped"),
+        "nothing was stopped: {frames:?}"
+    );
+    let wait_end = frames
+        .iter()
+        .find(|f| f["event"]["kind"] == "tool_end")
+        .expect("the cancelled wait ends")
+        .to_string();
+    assert!(
+        wait_end.contains("Wait cancelled; job 1 is still running."),
+        "{wait_end}"
+    );
+    std::thread::sleep(Duration::from_millis(300));
+    assert!(alive(pid), "the abort killed the background job");
+
+    // The job is still the model's: a peek finds it running.
+    send(
+        &mut stdin,
+        json!({ "type": "prompt", "id": "p3", "message": "peek" }),
+    );
+    let frames = read_until_response(&mut stdout, "prompt");
+    assert_eq!(frames.last().unwrap()["success"], true, "{frames:?}");
+    let peek = frames
+        .iter()
+        .find(|f| f["event"]["kind"] == "tool_end")
+        .expect("the peek ends")
+        .to_string();
+    assert!(!peek.contains("no job 1"), "{peek}");
+    assert!(peek.contains("Job 1 is still running"), "{peek}");
+    assert!(alive(pid));
+    drop(stdin);
+    let _ = child.wait();
+}