Merge pull request #6759 from Hmbown/fix/bh2-tools-shell-process-lifecycle

Verified original PR #6759 at 4c31a5314f: exact-head Linux, macOS and Windows tests and doctests, required checks and resolved review threads. Contributor authorship and existing history are preserved.

Signed-off-by: CodeWhale Bot <[email protected]>

crates/tui/src/core/engine/tool_catalog.rs

@@ -1637,10 +1637,16 @@ pub(super) async fn execute_code_execution_tool(
     })?;
     cmd.arg(&script_path).current_dir(workspace);
 
-    let output = tokio::time::timeout(Duration::from_secs(120), cmd.output())
-        .await
-        .map_err(|_| ToolError::Timeout { seconds: 120 })
-        .and_then(|res| res.map_err(|e| ToolError::execution_failed(e.to_string())))?;
+    // Contained: at the timeout (or a cancelled turn) the interpreter and
+    // anything the script started die together; a bare `output()` left both
+    // running.
+    let output = tokio::time::timeout(
+        Duration::from_secs(120),
+        crate::process_tree::contained_output(&mut cmd),
+    )
+    .await
+    .map_err(|_| ToolError::Timeout { seconds: 120 })
+    .and_then(|res| res.map_err(|e| ToolError::execution_failed(e.to_string())))?;
 
     let stdout = String::from_utf8_lossy(&output.stdout).to_string();
     let stderr = String::from_utf8_lossy(&output.stderr).to_string();

crates/tui/src/core/engine/tool_catalog/tests.rs

@@ -425,3 +425,26 @@ fn execute_tools_is_eager_in_code_mode_and_deferred_in_direct() {
             .all(|definition| definition.name != "execute_tools")
     );
 }
+
+/// A timed-out or cancelled `code_execution` kills the interpreter and what the
+/// script started; a bare `output()` left both running.
+#[cfg(unix)]
+#[tokio::test]
+async fn dropped_code_execution_kills_the_interpreter_tree() {
+    if crate::dependencies::resolve_python_interpreter().is_none() {
+        return;
+    }
+    let tmp = tempfile::tempdir().expect("tempdir");
+    let pid_file = tmp.path().join("grandchild.pid");
+    let code = format!(
+        "import subprocess\np = subprocess.Popen(['sleep', '300'])\nopen({:?}, 'w').write(str(p.pid))\np.wait()\n",
+        pid_file.display().to_string()
+    );
+    let input = json!({ "code": code });
+    let run = super::execute_code_execution_tool(&input, tmp.path());
+    let grandchild = crate::process_tree::drop_once_pid_written(run, &pid_file).await;
+    assert!(
+        crate::process_tree::wait_for_pid_exit(grandchild, std::time::Duration::from_secs(5)),
+        "a process the script started outlived the dropped code_execution call"
+    );
+}

crates/tui/src/lib.rs

@@ -859,6 +859,8 @@ fn spawn_signal_cleanup_task() {
         if !CLEANED_UP.swap(true, std::sync::atomic::Ordering::SeqCst) {
             #[cfg(unix)]
             crate::tools::shell::abort_pending_persistent_process_groups_for_exit();
+            #[cfg(unix)]
+            crate::process_tree::kill_contained_trees_for_exit();
             crate::tui::ui::emergency_restore_terminal();
             // Nothing async survives the `exit` below, so this is the last
             // chance to say how the session ended. `record_blocking` is one

crates/tui/src/process_tree.rs

@@ -11,13 +11,13 @@
 //! * **Windows:** the child is assigned to a Job Object configured with
 //!   `JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE`, so closing the job kills every
 //!   process in it. Hooks are created suspended and resumed only after the
-//!   assignment, so no descendant can escape; the extension host is not
-//!   (tokio's `Command` does not expose a suspended spawn), which leaves a
-//!   window of microseconds before assignment in which Node has not yet run
-//!   any plugin code.
+//!   assignment. Tokio children are assigned after spawning and can start
+//!   descendants before assignment. This bounds ordinary process lifetimes;
+//!   it is not a security sandbox.
 //!
 //! Dropping the guard kills the tree. That is deliberate: the guard's lifetime
-//! is the tree's lifetime.
+//! is the tree's lifetime, unless [`ProcessTree::release`] explicitly lets the
+//! tree outlive it.
 
 #[cfg(windows)]
 use windows::Win32::Foundation::{CloseHandle, HANDLE};
@@ -130,6 +130,22 @@ impl ProcessTree {
     }
 }
 
+impl ProcessTree {
+    /// Give up containment without killing anything: the tree outlives the
+    /// guard. For a command that exited on its own and may have deliberately
+    /// left something running.
+    pub(crate) fn release(self) {
+        #[cfg(windows)]
+        {
+            // Closing the handle kills the job unless the limit is cleared
+            // first. If clearing fails, the close still kills: the safe side.
+            let _ = self.job.clear_kill_on_close();
+        }
+        #[cfg(not(windows))]
+        std::mem::forget(self);
+    }
+}
+
 impl Drop for ProcessTree {
     fn drop(&mut self) {
         #[cfg(unix)]
@@ -144,6 +160,222 @@ impl Drop for ProcessTree {
     }
 }
 
+/// `Command::output()` for a child whose whole process tree dies with the
+/// returned future. Dropping it — a caller's `timeout` elapsing, or a
+/// cancelled tool call — kills the child and everything it started, where a
+/// bare `output()` left them running. Stdin is closed: callers run
+/// non-interactive work that must never wait on input.
+pub(crate) async fn contained_output(
+    cmd: &mut tokio::process::Command,
+) -> std::io::Result<std::process::Output> {
+    contained_output_until(cmd, std::future::pending())
+        .await
+        .map(|run| run.output)
+}
+
+/// What [`contained_output_until`] collected.
+pub(crate) struct ContainedOutput {
+    pub(crate) output: std::process::Output,
+    /// `stop` fired before the command exited: the tree was killed, and
+    /// `output` holds everything it wrote until then.
+    pub(crate) stopped: bool,
+}
+
+/// [`contained_output`], plus a `stop` future (a deadline, a cancel token).
+/// When `stop` fires first the tree is killed and the output written so far is
+/// still returned, so a hung command leaves evidence of where it hung.
+///
+/// A command that exits on its own is released, not killed: whatever it
+/// deliberately left running with its output redirected (`nohup server >log
+/// &`) keeps running, as it did with a bare `output()`. Only a dropped future
+/// or `stop` kills the tree. Until then the group is also listed for
+/// [`kill_contained_trees_for_exit`], because a terminating signal ends
+/// Codewhale through `process::exit`, where no destructor runs, and a child in
+/// its own process group no longer receives the terminal's Ctrl+C itself.
+pub(crate) async fn contained_output_until(
+    cmd: &mut tokio::process::Command,
+    stop: impl std::future::Future<Output = ()>,
+) -> std::io::Result<ContainedOutput> {
+    contained_run(cmd, None, stop, ProcessTree::attach_tokio).await
+}
+
+/// [`contained_output`] for a command that reads a request on stdin (a plugin
+/// tool's JSON input): `input` is written while the output pipes drain, then
+/// stdin is closed.
+pub(crate) async fn contained_output_with_input(
+    cmd: &mut tokio::process::Command,
+    input: Vec<u8>,
+) -> std::io::Result<std::process::Output> {
+    contained_run(
+        cmd,
+        Some(input),
+        std::future::pending(),
+        ProcessTree::attach_tokio,
+    )
+    .await
+    .map(|run| run.output)
+}
+
+async fn contained_run(
+    cmd: &mut tokio::process::Command,
+    input: Option<Vec<u8>>,
+    stop: impl std::future::Future<Output = ()>,
+    attach: impl FnOnce(&tokio::process::Child) -> std::io::Result<ProcessTree>,
+) -> std::io::Result<ContainedOutput> {
+    use std::process::Stdio;
+    cmd.stdin(if input.is_some() {
+        Stdio::piped()
+    } else {
+        Stdio::null()
+    })
+    .stdout(Stdio::piped())
+    .stderr(Stdio::piped())
+    .kill_on_drop(true);
+    #[cfg(unix)]
+    cmd.process_group(0);
+    let mut child = cmd.spawn()?;
+    // Never report a contained run when attachment failed. The child's
+    // kill-on-drop guard still cleans up the direct child on this error path.
+    let tree = attach(&child)?;
+    // Declared after `tree`, so a dropped future unlists the group first and
+    // the tree guard then kills it.
+    #[cfg(unix)]
+    let _listed = child.id().map(ContainedExitListing::new);
+    let mut stdout_pipe = child.stdout.take();
+    let mut stderr_pipe = child.stderr.take();
+    let stdin_pipe = child.stdin.take();
+    let mut stdout = Vec::new();
+    let mut stderr = Vec::new();
+    let exited = {
+        // Written alongside the drains: a child that answers before it has
+        // read all of its input would otherwise deadlock on a full pipe.
+        let feed = async move {
+            use tokio::io::AsyncWriteExt;
+            if let (Some(mut pipe), Some(input)) = (stdin_pipe, input) {
+                // A child that exits without reading its input closes the
+                // pipe; that is its answer, not a run failure.
+                if pipe.write_all(&input).await.is_ok() {
+                    let _ = pipe.shutdown().await;
+                }
+            }
+        };
+        let run = async {
+            let (out, err, status, ()) = tokio::join!(
+                drain_pipe(stdout_pipe.as_mut(), &mut stdout),
+                drain_pipe(stderr_pipe.as_mut(), &mut stderr),
+                child.wait(),
+                feed,
+            );
+            out?;
+            err?;
+            status
+        };
+        tokio::select! {
+            status = run => Some(status?),
+            () = stop => None,
+        }
+    };
+    let (status, stopped) = match exited {
+        Some(status) => {
+            tree.release();
+            (status, false)
+        }
+        None => {
+            drop(tree);
+            let _ = child.start_kill();
+            let status = child.wait().await?;
+            // Collect what is still buffered in the pipes. Bounded: a process
+            // that escaped the tree may still hold one open.
+            let _ = tokio::time::timeout(std::time::Duration::from_secs(1), async {
+                let _ = tokio::join!(
+                    drain_pipe(stdout_pipe.as_mut(), &mut stdout),
+                    drain_pipe(stderr_pipe.as_mut(), &mut stderr),
+                );
+            })
+            .await;
+            (status, true)
+        }
+    };
+    Ok(ContainedOutput {
+        output: std::process::Output {
+            status,
+            stdout,
+            stderr,
+        },
+        stopped,
+    })
+}
+
+/// Read `pipe` to EOF into `into`. Each chunk lands in `into` as soon as it is
+/// read, so a caller that stops polling keeps everything read so far.
+async fn drain_pipe<R: tokio::io::AsyncRead + Unpin>(
+    pipe: Option<&mut R>,
+    into: &mut Vec<u8>,
+) -> std::io::Result<()> {
+    use tokio::io::AsyncReadExt;
+    let Some(pipe) = pipe else {
+        return Ok(());
+    };
+    let mut chunk = [0_u8; 8 * 1024];
+    loop {
+        let read = pipe.read(&mut chunk).await?;
+        if read == 0 {
+            return Ok(());
+        }
+        into.extend_from_slice(&chunk[..read]);
+    }
+}
+
+/// Process groups of [`contained_output_until`] runs still in flight.
+#[cfg(unix)]
+static CONTAINED_EXIT_GROUPS: std::sync::OnceLock<
+    std::sync::Mutex<std::collections::HashSet<u32>>,
+> = std::sync::OnceLock::new();
+
+#[cfg(unix)]
+fn contained_exit_groups() -> std::sync::MutexGuard<'static, std::collections::HashSet<u32>> {
+    CONTAINED_EXIT_GROUPS
+        .get_or_init(Default::default)
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+}
+
+/// Lists one in-flight process group until dropped.
+#[cfg(unix)]
+struct ContainedExitListing(u32);
+
+#[cfg(unix)]
+impl ContainedExitListing {
+    fn new(process_group_id: u32) -> Self {
+        contained_exit_groups().insert(process_group_id);
+        Self(process_group_id)
+    }
+}
+
+#[cfg(unix)]
+impl Drop for ContainedExitListing {
+    fn drop(&mut self) {
+        contained_exit_groups().remove(&self.0);
+    }
+}
+
+/// Kill every in-flight contained tree. The process-wide signal path calls
+/// this immediately before `process::exit`, where Rust destructors cannot run.
+#[cfg(unix)]
+pub(crate) fn kill_contained_trees_for_exit() {
+    let groups = contained_exit_groups().drain().collect::<Vec<_>>();
+    for process_group_id in groups {
+        if let Ok(process_group_id) = libc::pid_t::try_from(process_group_id) {
+            // SAFETY: the id is a child spawned with `process_group(0)` whose
+            // run is still in flight. A negative pid names that group, never
+            // Codewhale's own.
+            unsafe {
+                libc::kill(-process_group_id, libc::SIGKILL);
+            }
+        }
+    }
+}
+
 #[cfg(windows)]
 struct WindowsJob {
     handle: HANDLE,
@@ -172,6 +404,20 @@ impl WindowsJob {
         Ok(job)
     }
 
+    fn clear_kill_on_close(&self) -> std::io::Result<()> {
+        let limits = JOBOBJECT_EXTENDED_LIMIT_INFORMATION::default();
+        // SAFETY: `limits` is live with matching size; the handle is live.
+        unsafe {
+            SetInformationJobObject(
+                self.handle,
+                JobObjectExtendedLimitInformation,
+                &limits as *const _ as *const core::ffi::c_void,
+                std::mem::size_of::<JOBOBJECT_EXTENDED_LIMIT_INFORMATION>() as u32,
+            )
+            .map_err(windows_io_error)
+        }
+    }
+
     fn terminate(&self) -> std::io::Result<()> {
         // SAFETY: `self.handle` is a live owned job handle.
         unsafe { TerminateJobObject(self.handle, 1).map_err(windows_io_error) }
@@ -192,3 +438,249 @@ impl Drop for WindowsJob {
 pub(crate) fn windows_io_error(error: windows::core::Error) -> std::io::Error {
     std::io::Error::other(error)
 }
+
+/// Test support: wait for `pid` to stop existing. `true` once it is gone.
+/// A zombie counts as gone: it was killed and only awaits a reaper, which a
+/// container whose PID 1 does not reap orphans may never provide.
+#[cfg(all(test, unix))]
+pub(crate) fn wait_for_pid_exit(pid: libc::pid_t, within: std::time::Duration) -> bool {
+    let deadline = std::time::Instant::now() + within;
+    loop {
+        // SAFETY: signal 0 only checks that the process exists.
+        if unsafe { libc::kill(pid, 0) } != 0 || is_zombie(pid) {
+            return true;
+        }
+        if std::time::Instant::now() >= deadline {
+            // Do not leak the fixture when the assertion is about to fail.
+            // SAFETY: kill(2) dereferences no pointers.
+            unsafe { libc::kill(pid, libc::SIGKILL) };
+            return false;
+        }
+        std::thread::sleep(std::time::Duration::from_millis(50));
+    }
+}
+
+/// Linux only (`/proc`); elsewhere launchd/init reaps orphans promptly.
+#[cfg(all(test, unix))]
+fn is_zombie(pid: libc::pid_t) -> bool {
+    std::fs::read_to_string(format!("/proc/{pid}/stat"))
+        .ok()
+        .and_then(|stat| {
+            stat.rsplit_once(')')
+                .map(|(_, rest)| rest.trim_start().starts_with('Z'))
+        })
+        .unwrap_or(false)
+}
+
+/// Test support: read the pid a fixture wrote to `path`, waiting for it.
+#[cfg(all(test, unix))]
+pub(crate) fn read_pid_file(path: &std::path::Path, within: std::time::Duration) -> libc::pid_t {
+    let deadline = std::time::Instant::now() + within;
+    loop {
+        if let Some(pid) = parse_pid_file(path) {
+            return pid;
+        }
+        assert!(
+            std::time::Instant::now() < deadline,
+            "fixture never wrote {}",
+            path.display()
+        );
+        std::thread::sleep(std::time::Duration::from_millis(20));
+    }
+}
+
+#[cfg(all(test, unix))]
+fn parse_pid_file(path: &std::path::Path) -> Option<libc::pid_t> {
+    std::fs::read_to_string(path)
+        .ok()
+        .and_then(|text| text.trim().parse().ok())
+}
+
+/// Test support: drive `run` until the fixture has written its pid to `path`,
+/// then drop `run`. No wall-clock guess about how long the fixture needs to
+/// start; `run` finishing first is a fixture bug.
+#[cfg(all(test, unix))]
+pub(crate) async fn drop_once_pid_written<F: std::future::Future>(
+    run: F,
+    path: &std::path::Path,
+) -> libc::pid_t {
+    let written = async {
+        loop {
+            if let Some(pid) = parse_pid_file(path) {
+                return pid;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+        }
+    };
+    tokio::time::timeout(std::time::Duration::from_secs(60), async {
+        tokio::select! {
+            _ = run => panic!("fixture exited before writing {}", path.display()),
+            pid = written => pid,
+        }
+    })
+    .await
+    .unwrap_or_else(|_| panic!("fixture never wrote {}", path.display()))
+}
+
+#[cfg(all(test, unix))]
+mod tests {
+    use super::*;
+    use std::time::Duration;
+
+    #[tokio::test]
+    async fn failed_containment_attachment_is_reported_and_kills_the_child() {
+        let pid = std::cell::Cell::new(None);
+        let mut cmd = tokio::process::Command::new("/bin/sleep");
+        cmd.arg("300");
+        let result = tokio::time::timeout(
+            Duration::from_secs(1),
+            contained_run(&mut cmd, None, std::future::pending(), |child| {
+                pid.set(child.id());
+                Err(std::io::Error::new(
+                    std::io::ErrorKind::PermissionDenied,
+                    "injected containment attachment failure",
+                ))
+            }),
+        )
+        .await;
+        let pid = libc::pid_t::try_from(pid.get().expect("child spawned")).expect("pid");
+        // Let Tokio reap while checking cleanup; the helper kills the fixture
+        // itself if the child ever outlives this failed run.
+        assert!(
+            tokio::task::spawn_blocking(move || wait_for_pid_exit(pid, Duration::from_secs(5)))
+                .await
+                .expect("cleanup probe"),
+            "the direct child outlived failed containment attachment"
+        );
+        let Err(error) = result
+            .expect("attachment failure must return without running the command to completion")
+        else {
+            panic!("a containment failure must not be accepted as a successful run");
+        };
+        assert_eq!(error.kind(), std::io::ErrorKind::PermissionDenied);
+        assert_eq!(error.to_string(), "injected containment attachment failure");
+    }
+
+    #[tokio::test]
+    async fn dropping_contained_output_kills_the_whole_tree() {
+        let tmp = tempfile::tempdir().expect("tempdir");
+        let pid_file = tmp.path().join("grandchild.pid");
+        let mut cmd = tokio::process::Command::new("/bin/sh");
+        cmd.arg("-c")
+            .arg("sleep 300 & echo $! > grandchild.pid; wait")
+            .current_dir(tmp.path());
+        let grandchild = drop_once_pid_written(contained_output(&mut cmd), &pid_file).await;
+        assert!(
+            wait_for_pid_exit(grandchild, Duration::from_secs(5)),
+            "a process started by the dropped command is still running"
+        );
+    }
+
+    #[tokio::test]
+    async fn contained_output_captures_output_and_closes_stdin() {
+        let mut cmd = tokio::process::Command::new("/bin/sh");
+        cmd.arg("-c").arg("cat; echo done");
+        let output = tokio::time::timeout(Duration::from_secs(10), contained_output(&mut cmd))
+            .await
+            .expect("stdin must be closed, not inherited")
+            .expect("run");
+        assert!(output.status.success());
+        assert_eq!(String::from_utf8_lossy(&output.stdout), "done\n");
+    }
+
+    /// The input reaches stdin, and stdin is closed after it.
+    #[tokio::test]
+    async fn contained_output_with_input_feeds_then_closes_stdin() {
+        let mut cmd = tokio::process::Command::new("/bin/sh");
+        cmd.arg("-c").arg("cat; echo done");
+        let output = tokio::time::timeout(
+            Duration::from_secs(10),
+            contained_output_with_input(&mut cmd, b"request\n".to_vec()),
+        )
+        .await
+        .expect("stdin must be closed after the input")
+        .expect("run");
+        assert!(output.status.success());
+        assert_eq!(String::from_utf8_lossy(&output.stdout), "request\ndone\n");
+    }
+
+    /// A command that exits on its own is not reaped along with what it
+    /// deliberately left running, as with a bare `output()`.
+    #[tokio::test]
+    async fn contained_output_leaves_a_detached_daemon_of_a_clean_exit_running() {
+        let tmp = tempfile::tempdir().expect("tempdir");
+        let mut cmd = tokio::process::Command::new("/bin/sh");
+        cmd.arg("-c")
+            .arg("sleep 300 >/dev/null 2>&1 & echo $! > daemon.pid")
+            .current_dir(tmp.path());
+        let output = contained_output(&mut cmd).await.expect("run");
+        assert!(output.status.success());
+        let daemon = read_pid_file(&tmp.path().join("daemon.pid"), Duration::from_secs(5));
+        // `wait_for_pid_exit` kills the fixture when it times out.
+        assert!(
+            !wait_for_pid_exit(daemon, Duration::from_millis(500)),
+            "the daemon a clean command left behind was killed"
+        );
+    }
+
+    /// `stop` kills the tree but keeps what the command already wrote.
+    #[tokio::test]
+    async fn stopped_contained_output_keeps_partial_output_and_kills_the_tree() {
+        let tmp = tempfile::tempdir().expect("tempdir");
+        let mut cmd = tokio::process::Command::new("/bin/sh");
+        cmd.arg("-c")
+            .arg("echo before-stop; sleep 300 & echo $! > grandchild.pid; wait")
+            .current_dir(tmp.path());
+        let pid_file = tmp.path().join("grandchild.pid");
+        let stop = async {
+            while parse_pid_file(&pid_file).is_none() {
+                tokio::time::sleep(Duration::from_millis(20)).await;
+            }
+        };
+        let run = tokio::time::timeout(
+            Duration::from_secs(60),
+            contained_output_until(&mut cmd, stop),
+        )
+        .await
+        .expect("stop must end the run")
+        .expect("run");
+        assert!(run.stopped);
+        assert_eq!(String::from_utf8_lossy(&run.output.stdout), "before-stop\n");
+        let grandchild = read_pid_file(&pid_file, Duration::from_secs(5));
+        assert!(
+            wait_for_pid_exit(grandchild, Duration::from_secs(5)),
+            "a process started by the stopped command is still running"
+        );
+    }
+
+    /// An in-flight run is listed for the signal-exit kill, and unlisted once
+    /// the future is gone.
+    #[tokio::test]
+    async fn in_flight_contained_run_is_listed_for_signal_exit() {
+        let tmp = tempfile::tempdir().expect("tempdir");
+        let mut cmd = tokio::process::Command::new("/bin/sh");
+        cmd.arg("-c")
+            .arg("echo $$ > leader.pid; sleep 300")
+            .current_dir(tmp.path());
+        let pid_file = tmp.path().join("leader.pid");
+        let mut run = Box::pin(contained_output(&mut cmd));
+        let leader = drop_once_pid_written(
+            async {
+                run.as_mut().await.ok();
+            },
+            &pid_file,
+        )
+        .await;
+        let leader_group = u32::try_from(leader).expect("pid");
+        assert!(contained_exit_groups().contains(&leader_group));
+        drop(run);
+        assert!(!contained_exit_groups().contains(&leader_group));
+        // Let this current-thread runtime poll Tokio's orphan reaper while the
+        // cleanup probe waits; kill(pid, 0) still sees an unreaped macOS child.
+        assert!(
+            tokio::task::spawn_blocking(move || wait_for_pid_exit(leader, Duration::from_secs(5)))
+                .await
+                .expect("cleanup probe")
+        );
+    }
+}

crates/tui/src/runtime_api/git.rs

@@ -42,7 +42,6 @@
 
 use std::collections::{BTreeMap, BTreeSet};
 use std::path::{Path as FsPath, PathBuf};
-use std::process::Stdio;
 use std::time::Duration;
 
 use axum::Json;
@@ -112,8 +111,12 @@ async fn git_read(workspace: &FsPath, args: &[&str]) -> Result<GitRun, ApiError>
         .map_err(|_| ApiError::internal("git read setup failed"))?
         .map_err(|error| ApiError::internal(format!("git is unavailable: {error}")))?;
     let mut command = tokio::process::Command::from(command);
-    command.args(args).kill_on_drop(true);
-    finish_git(command.output(), GIT_READ_TIMEOUT).await
+    command.args(args);
+    finish_git(
+        crate::process_tree::contained_output(&mut command),
+        GIT_READ_TIMEOUT,
+    )
+    .await
 }
 
 /// Write path for operator-driven mutations. Non-interactive by contract:
@@ -124,12 +127,14 @@ async fn git_read(workspace: &FsPath, args: &[&str]) -> Result<GitRun, ApiError>
 async fn git_write(workspace: &FsPath, args: Vec<String>) -> Result<GitRun, ApiError> {
     let mut command = Git::tokio_command()
         .ok_or_else(|| ApiError::internal("git is not installed or not in PATH"))?;
-    command
-        .args(&args)
-        .current_dir(workspace)
-        .stdin(Stdio::null())
-        .kill_on_drop(true);
-    finish_git(command.output(), GIT_WRITE_TIMEOUT).await
+    command.args(&args).current_dir(workspace);
+    // Contained: hooks and filters run here, and a timeout or a dropped
+    // request must end them too, not only the `git` process itself.
+    finish_git(
+        crate::process_tree::contained_output(&mut command),
+        GIT_WRITE_TIMEOUT,
+    )
+    .await
 }
 
 async fn finish_git(
@@ -2692,6 +2697,27 @@ mod tests {
         assert_eq!(ok.files.unwrap().keys().collect::<Vec<_>>(), vec!["a.txt"]);
     }
 
+    /// A dropped (or timed-out) write takes down what git started — a hook, a
+    /// filter, an alias — not just the `git` process.
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn dropped_git_write_kills_what_git_started() {
+        let tmp = repo();
+        let pid_file = tmp.path().join("hang.pid");
+        let args = vec![
+            "-c".to_string(),
+            "alias.hang=!sleep 300 & echo $! > hang.pid; wait".to_string(),
+            "hang".to_string(),
+        ];
+        let grandchild =
+            crate::process_tree::drop_once_pid_written(git_write(tmp.path(), args), &pid_file)
+                .await;
+        assert!(
+            crate::process_tree::wait_for_pid_exit(grandchild, Duration::from_secs(5)),
+            "a process git started outlived the dropped request"
+        );
+    }
+
     #[test]
     fn file_diff_reads_the_path_literally_inside_a_subdirectory_workspace() {
         let tmp = repo();

crates/tui/src/tools/git_history.rs

@@ -891,11 +891,11 @@ async fn run_git_command_async(
 /// Run git under a hard deadline. `Ok(None)` means the deadline passed and the
 /// child was killed, so a timed-out fetch never lingers.
 ///
-/// On unix git runs in its own process group and the whole group is killed at
-/// the deadline: git hands the network to a transport child (`git-remote-http`,
-/// `ssh`) that survives a SIGKILL to git alone and would otherwise keep the
-/// stalled connection open indefinitely. `kill_on_drop` still covers the
-/// leader everywhere else.
+/// git runs contained ([`crate::process_tree::contained_output`]) and the whole
+/// tree is killed at the deadline or when the call is cancelled: git hands the
+/// network to a transport child (`git-remote-http`, `ssh`) that survives a
+/// SIGKILL to git alone and would otherwise keep the stalled connection open
+/// indefinitely.
 async fn run_git_command_bounded(
     working_dir: &Path,
     args: &[String],
@@ -906,49 +906,18 @@ async fn run_git_command_bounded(
             "git is not installed or not in PATH",
         ));
     };
-    cmd.args(args)
-        .current_dir(working_dir)
-        .stdin(std::process::Stdio::null())
-        .stdout(std::process::Stdio::piped())
-        .stderr(std::process::Stdio::piped())
-        .kill_on_drop(true);
-    #[cfg(unix)]
-    cmd.process_group(0);
-    let child = match cmd.spawn() {
-        Ok(child) => child,
-        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
-            return Err(ToolError::not_available(
-                "git is not installed or not in PATH",
-            ));
-        }
-        Err(e) => {
-            return Err(ToolError::execution_failed(format!(
-                "Failed to run git: {e}"
-            )));
-        }
-    };
-    let process_group = child.id();
-    match tokio::time::timeout(timeout, child.wait_with_output()).await {
+    cmd.args(args).current_dir(working_dir);
+    match tokio::time::timeout(timeout, crate::process_tree::contained_output(&mut cmd)).await {
         Ok(Ok(output)) => Ok(Some(output)),
+        Ok(Err(e)) if e.kind() == std::io::ErrorKind::NotFound => Err(ToolError::not_available(
+            "git is not installed or not in PATH",
+        )),
         Ok(Err(e)) => Err(ToolError::execution_failed(format!(
             "Failed to run git: {e}"
         ))),
-        Err(_) => {
-            #[cfg(unix)]
-            if let Some(pgid) = process_group
-                .and_then(|id| libc::pid_t::try_from(id).ok())
-                .filter(|pgid| *pgid > 0)
-            {
-                // SAFETY: kill(2) dereferences no pointers; a negative pid
-                // targets the group this call created with process_group(0).
-                unsafe {
-                    libc::kill(-pgid, libc::SIGKILL);
-                }
-            }
-            #[cfg(not(unix))]
-            let _ = process_group;
-            Ok(None)
-        }
+        // The elapsed deadline dropped the contained run, which killed git's
+        // whole tree.
+        Err(_) => Ok(None),
     }
 }
 
@@ -1486,4 +1455,27 @@ mod tests {
         assert!(output.status.success());
         String::from_utf8_lossy(&output.stdout).trim().to_string()
     }
+
+    /// A cancelled call kills git's whole tree, not only at the deadline.
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn cancelled_bounded_git_run_kills_what_git_started() {
+        if !git_available() {
+            return;
+        }
+        let tmp = tempdir().expect("tempdir");
+        init_git_repo(tmp.path());
+        let pid_file = tmp.path().join("hang.pid");
+        let args = [
+            "-c".to_string(),
+            "alias.hang=!sleep 300 & echo $! > hang.pid; wait".to_string(),
+            "hang".to_string(),
+        ];
+        let run = run_git_command_bounded(tmp.path(), &args, std::time::Duration::from_secs(600));
+        let grandchild = crate::process_tree::drop_once_pid_written(run, &pid_file).await;
+        assert!(
+            crate::process_tree::wait_for_pid_exit(grandchild, std::time::Duration::from_secs(5)),
+            "a process git started outlived the cancelled call"
+        );
+    }
 }

crates/tui/src/tools/js_execution.rs

@@ -156,10 +156,15 @@ pub async fn execute_js_execution_tool(
         cmd.env("NODE_USE_ENV_PROXY", "1");
     }
 
-    let output = tokio::time::timeout(Duration::from_secs(120), cmd.output())
-        .await
-        .map_err(|_| ToolError::Timeout { seconds: 120 })
-        .and_then(|res| res.map_err(|e| ToolError::execution_failed(e.to_string())))?;
+    // Contained: a timeout or a cancelled call ends the interpreter and
+    // anything it started (`setInterval`, a server), not only this future.
+    let output = tokio::time::timeout(
+        Duration::from_secs(120),
+        crate::process_tree::contained_output(&mut cmd),
+    )
+    .await
+    .map_err(|_| ToolError::Timeout { seconds: 120 })
+    .and_then(|res| res.map_err(|e| ToolError::execution_failed(e.to_string())))?;
 
     let stdout = String::from_utf8_lossy(&output.stdout).to_string();
     let stderr = String::from_utf8_lossy(&output.stderr).to_string();
@@ -272,6 +277,31 @@ mod tests {
         );
     }
 
+    /// A timed-out or cancelled call drops the future; the interpreter and
+    /// what it started must end with it instead of running on.
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn dropped_js_execution_kills_the_interpreter_tree() {
+        if !node_present() {
+            return;
+        }
+        let tmp = tempdir().expect("tempdir");
+        let code = "const { spawn } = require('child_process');\n\
+                    const child = spawn('sleep', ['300'], { stdio: 'ignore' });\n\
+                    require('fs').writeFileSync('grandchild.pid', String(child.pid));\n\
+                    setInterval(() => {}, 1000);";
+        let input = json!({ "code": code });
+        let grandchild = crate::process_tree::drop_once_pid_written(
+            execute_js_execution_tool(&input, tmp.path()),
+            &tmp.path().join("grandchild.pid"),
+        )
+        .await;
+        assert!(
+            crate::process_tree::wait_for_pid_exit(grandchild, Duration::from_secs(5)),
+            "a process started by the dropped script is still running"
+        );
+    }
+
     #[tokio::test]
     async fn execute_js_surfaces_runtime_error_with_nonzero_exit() {
         if !node_present() {

crates/tui/src/tools/plugin.rs

@@ -25,7 +25,6 @@ use std::time::Duration;
 
 use async_trait::async_trait;
 use serde_json::Value;
-use tokio::io::AsyncWriteExt;
 
 use super::spec::{
     ApprovalRequirement, ToolCapability, ToolContext, ToolError, ToolResult, ToolSpec,
@@ -300,32 +299,17 @@ async fn run_plugin_child_raw(
     let input_bytes = serde_json::to_vec(&input)
         .map_err(|e| ToolError::invalid_input(format!("failed to serialize input: {e}")))?;
 
-    cmd.stdin(std::process::Stdio::piped());
-    cmd.stdout(std::process::Stdio::piped());
-    cmd.stderr(std::process::Stdio::piped());
-
-    let mut child = cmd
-        .spawn()
-        .map_err(|e| ToolError::execution_failed(format!("failed to spawn {label}: {e}")))?;
-
-    let stdin_writer = child.stdin.take().map(|mut stdin| {
-        tokio::spawn(async move {
-            if stdin.write_all(&input_bytes).await.is_ok() {
-                let _ = stdin.shutdown().await;
-            }
-        })
-    });
-
-    let output = tokio::time::timeout(PLUGIN_EXECUTION_TIMEOUT, child.wait_with_output())
-        .await
-        .map_err(|_| ToolError::Timeout {
-            seconds: PLUGIN_EXECUTION_TIMEOUT.as_secs(),
-        })?
-        .map_err(|e| ToolError::execution_failed(format!("process error: {e}")))?;
-
-    if let Some(stdin_writer) = stdin_writer {
-        let _ = stdin_writer.await;
-    }
+    // Contained: a timed-out or cancelled plugin takes everything it started
+    // down with it, not just the interpreter that `kill_on_drop` would reach.
+    let output = tokio::time::timeout(
+        PLUGIN_EXECUTION_TIMEOUT,
+        crate::process_tree::contained_output_with_input(cmd, input_bytes),
+    )
+    .await
+    .map_err(|_| ToolError::Timeout {
+        seconds: PLUGIN_EXECUTION_TIMEOUT.as_secs(),
+    })?
+    .map_err(|e| ToolError::execution_failed(format!("failed to run {label}: {e}")))?;
 
     if output.status.success() {
         let stdout = String::from_utf8_lossy(&output.stdout).to_string();
@@ -728,6 +712,25 @@ echo hello
         assert!(result.content.len() > 64 * 1024);
     }
 
+    /// A cancelled (or timed-out) plugin call kills what the plugin started,
+    /// not just the interpreter.
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn dropped_plugin_call_kills_the_plugin_process_tree() {
+        let tmp = TempDir::new().unwrap();
+        let pid_file = tmp.path().join("grandchild.pid");
+        let mut cmd = tokio::process::Command::new("/bin/sh");
+        cmd.arg("-c")
+            .arg("sleep 300 & echo $! > grandchild.pid; wait")
+            .current_dir(tmp.path());
+        let run = run_plugin_child_raw(&mut cmd, "hanging plugin", serde_json::json!({}));
+        let grandchild = crate::process_tree::drop_once_pid_written(run, &pid_file).await;
+        assert!(
+            crate::process_tree::wait_for_pid_exit(grandchild, Duration::from_secs(5)),
+            "a process started by the cancelled plugin is still running"
+        );
+    }
+
     #[test]
     fn plugin_deadlock_child_process() {
         if std::env::var_os(DEADLOCK_CHILD_ENV).is_none() {

crates/tui/src/tools/shell.rs

@@ -1512,23 +1512,34 @@ impl BackgroundShell {
     }
 
     fn take_delta(&mut self) -> (String, String, usize, usize, usize, usize) {
-        if let Some(snapshot) = self.bounded_output_snapshot(false).ok().flatten() {
-            let changed = snapshot.total_bytes != self.stdout_cursor;
-            self.stdout_cursor = snapshot.total_bytes;
-            if changed {
-                self.last_output_at = Instant::now();
-                self.last_observed_output_len = snapshot.total_bytes;
-                let delta_len = snapshot.content.len();
-                return (
-                    snapshot.content,
-                    String::new(),
-                    delta_len,
-                    0,
-                    snapshot.total_bytes,
-                    0,
-                );
+        // Only the bytes after the cursor: returning the whole retained tail
+        // made every `wait` poll repeat output the caller already holds.
+        let bounded_delta = self.bounded_output.as_ref().and_then(|output| {
+            let output = output.lock().unwrap_or_else(|error| error.into_inner());
+            let total = output.total_bytes();
+            output
+                .delta_since(self.stdout_cursor)
+                .ok()
+                .map(|delta| (delta, total))
+        });
+        if let Some(((mut delta, omitted), total_bytes)) = bounded_delta {
+            let delta_len = total_bytes.saturating_sub(self.stdout_cursor);
+            self.stdout_cursor = total_bytes;
+            if delta_len == 0 {
+                return (String::new(), String::new(), 0, 0, total_bytes, 0);
             }
-            return (String::new(), String::new(), 0, 0, snapshot.total_bytes, 0);
+            self.last_output_at = Instant::now();
+            self.last_observed_output_len = total_bytes;
+            if omitted > 0
+                && let Some(output) = self.bounded_output.as_ref()
+            {
+                let notice = output
+                    .lock()
+                    .unwrap_or_else(|error| error.into_inner())
+                    .omitted_notice(omitted);
+                delta.insert_str(0, &notice);
+            }
+            return (delta, String::new(), delta_len, 0, total_bytes, 0);
         }
         let (stdout_delta, stdout_total) =
             take_delta_from_buffer(&self.stdout_buffer, &mut self.stdout_cursor);
@@ -2390,9 +2401,14 @@ impl ShellManager {
         }
         install_parent_death_signal(&mut cmd);
 
-        if stdin_data.is_some() {
-            cmd.stdin(Stdio::piped());
-        }
+        // Without input, stdin is closed rather than inherited: an unexpected
+        // read (`cat`, a prompt) gets EOF instead of blocking on, or reading,
+        // the operator's terminal until the timeout.
+        cmd.stdin(if stdin_data.is_some() {
+            Stdio::piped()
+        } else {
+            Stdio::null()
+        });
 
         child_env::apply_to_command(&mut cmd, child_env::string_map_env(&exec_env.env));
         remove_readonly_redirect_env(&mut cmd, &exec_env.env);
@@ -3644,13 +3660,21 @@ impl ShellManager {
     /// Age alone is not a bound: it only fired from `list_jobs()`, so a session
     /// that never opened the jobs panel evicted nothing, and 500 finished
     /// records inside one hour were all retained regardless of size (#5472).
+    ///
+    /// Age counts from when a job finished, not when it started: a job that
+    /// ran longer than `max_age` would otherwise be dropped the moment it
+    /// exited, before its completion was delivered. Undelivered completions
+    /// age out too: jobs launched over the Runtime API, or owned by a session
+    /// that is no longer active, are never drained.
     pub fn cleanup(&mut self, max_age: Duration) {
         self.processes.retain(|_, shell| {
             if shell.status == ShellStatus::Running {
-                true
-            } else {
-                shell.started_at.elapsed() < max_age
+                return true;
             }
+            shell.mark_finished();
+            shell
+                .finished_at
+                .is_none_or(|finished| finished.elapsed() < max_age)
         });
         self.enforce_finished_job_bounds();
     }
@@ -3683,7 +3707,8 @@ impl ShellManager {
                 (
                     id.clone(),
                     shell.completion_reported,
-                    shell.started_at,
+                    // Oldest by finish, like the age rule in `cleanup`.
+                    shell.finished_at.unwrap_or(shell.started_at),
                     shell.retained_output_bytes(),
                 )
             })
@@ -4870,7 +4895,12 @@ async fn execute_foreground_via_background(
         manager.attach_heavy_permit(&task_id, permit)?;
     }
 
-    if stdin_data.is_some() {
+    // A foreground pipe command gets EOF on stdin: an unexpected read (`cat`,
+    // `read x`, a confirmation prompt) then fails at once instead of blocking
+    // until the timeout kills it. A TTY keeps its terminal input. A command
+    // later moved to /jobs keeps the closed stdin; interactive input needs
+    // `background: true` from the start.
+    if stdin_data.is_some() || !tty {
         let mut manager = context
             .shell_manager
             .lock()

crates/tui/src/tools/shell/output.rs

@@ -218,6 +218,53 @@ impl BoundedOutputAccumulator {
         self.total_bytes
     }
 
+    /// Decoded output appended after `cursor` (a previous `total_bytes`), for
+    /// incremental readers such as `wait`. The second value counts bytes of
+    /// that range the memory bound already discarded; the complete output is
+    /// still in the spill file.
+    pub(super) fn delta_since(&self, cursor: usize) -> io::Result<(String, usize)> {
+        if let Some(error) = self.stream_error.as_ref() {
+            return Err(io::Error::other(error.clone()));
+        }
+        let new_bytes = self.total_bytes.saturating_sub(cursor);
+        let available = new_bytes.min(self.tail.len());
+        let mut bytes = self
+            .tail
+            .range(self.tail.len() - available..)
+            .copied()
+            .collect::<Vec<_>>();
+        // A cursor is always a character boundary; a clipped tail's front may
+        // not be, so drop any continuation bytes it starts with.
+        let partial = bytes
+            .iter()
+            .take_while(|byte| (**byte & 0xC0) == 0x80)
+            .count();
+        bytes.drain(..partial);
+        let omitted = new_bytes - available + partial;
+        Ok((String::from_utf8_lossy(&bytes).into_owned(), omitted))
+    }
+
+    /// First line of a delta that lost `omitted` bytes to the memory bound:
+    /// how many, and where the complete output is (or why it is not kept).
+    pub(super) fn omitted_notice(&self, omitted: usize) -> String {
+        let location = self
+            .full_output_path
+            .as_deref()
+            .or_else(|| self.temp.as_ref().map(tempfile::NamedTempFile::path));
+        match (location, self.spill_unavailable.as_deref()) {
+            (Some(path), _) => format!(
+                "[{omitted} bytes of earlier output not retained in memory. Full output: {}]\n",
+                path.display()
+            ),
+            (None, Some(reason)) => format!(
+                "[{omitted} bytes of earlier output not retained in memory. Full output was not persisted: {reason}]\n"
+            ),
+            (None, None) => {
+                format!("[{omitted} bytes of earlier output not retained in memory]\n")
+            }
+        }
+    }
+
     pub(super) fn snapshot(&mut self, finalize: bool) -> io::Result<BoundedOutputSnapshot> {
         if let Some(error) = self.stream_error.as_ref() {
             return Err(io::Error::other(error.clone()));
@@ -562,9 +609,9 @@ pub(super) fn tail_text(text: &str, max_chars: usize) -> String {
 #[cfg(test)]
 mod tests {
     use super::{
-        BOUNDED_OUTPUT_MAX_BYTES, BOUNDED_OUTPUT_MAX_LINES, BoundedOutputAccumulator,
-        RAW_STREAM_MAX_BYTES, RawOutputBuffer, SharedRawOutput, tail_from_buffer,
-        take_delta_from_buffer,
+        BOUNDED_OUTPUT_MAX_BYTES, BOUNDED_OUTPUT_MAX_LINES, BOUNDED_OUTPUT_RETAIN_BYTES,
+        BoundedOutputAccumulator, RAW_STREAM_MAX_BYTES, RawOutputBuffer, SharedRawOutput,
+        tail_from_buffer, take_delta_from_buffer,
     };
     use std::sync::{Arc, Mutex};
 
@@ -778,6 +825,58 @@ mod tests {
         std::fs::remove_file(path).expect("remove full output");
     }
 
+    #[test]
+    fn bounded_output_delta_since_returns_only_new_bytes() {
+        let mut output = BoundedOutputAccumulator::new_in(None);
+        output.append(b"first\n").expect("append");
+        let cursor = output.total_bytes();
+        assert_eq!(output.delta_since(0).expect("delta").0, "first\n");
+        output.append(b"second\n").expect("append");
+        assert_eq!(
+            output.delta_since(cursor).expect("delta"),
+            ("second\n".to_string(), 0)
+        );
+        assert_eq!(
+            output.delta_since(output.total_bytes()).expect("delta"),
+            (String::new(), 0)
+        );
+
+        // Past the memory bound: the dropped span is counted, and a clipped
+        // multi-byte character at the front is not emitted as garbage.
+        let mut output = BoundedOutputAccumulator::new_in(None);
+        // One ASCII byte first puts the retained tail's front mid-character.
+        output.append(b"a").expect("append");
+        output
+            .append("é".repeat(BOUNDED_OUTPUT_MAX_BYTES).as_bytes())
+            .expect("append");
+        let total = output.total_bytes();
+        let (delta, omitted) = output.delta_since(0).expect("delta");
+        assert!(!delta.contains('\u{FFFD}'));
+        assert_eq!(delta.len() + omitted, total);
+        assert!(delta.len() <= BOUNDED_OUTPUT_RETAIN_BYTES);
+    }
+
+    /// The notice for dropped delta bytes says where the full output is.
+    #[test]
+    fn bounded_output_omitted_notice_points_at_the_spill_file() {
+        let output = BoundedOutputAccumulator::new_in(None);
+        let spill = output
+            .temp
+            .as_ref()
+            .expect("spill file")
+            .path()
+            .display()
+            .to_string();
+        let notice = output.omitted_notice(42);
+        assert!(notice.starts_with("[42 bytes"), "{notice}");
+        assert!(notice.contains(&spill), "{notice}");
+
+        let missing = tempfile::tempdir().expect("tempdir").path().join("gone");
+        let unspilled = BoundedOutputAccumulator::new_in(Some(&missing));
+        let notice = unspilled.omitted_notice(7);
+        assert!(notice.contains("not persisted"), "{notice}");
+    }
+
     #[test]
     fn bounded_output_huge_terminal_line_matches_upstream_notice() {
         let mut source = vec![b'x'; BOUNDED_OUTPUT_MAX_BYTES + 1_024];

crates/tui/src/tools/shell/tests.rs

@@ -4053,7 +4053,8 @@ fn test_list_jobs_cleans_up_completed_old_processes() {
     // Both the completed job and any tracking state should be present.
     assert!(!manager.processes.is_empty());
 
-    // cleanup(ZERO) removes all completed processes immediately.
+    // cleanup(ZERO) removes every delivered completed process immediately.
+    manager.drain_finished_jobs_with_evidence();
     manager.cleanup(Duration::ZERO);
     assert!(
         manager.processes.is_empty(),
@@ -4061,6 +4062,203 @@ fn test_list_jobs_cleans_up_completed_old_processes() {
     );
 }
 
+/// A job that ran longer than the retention age must survive until its
+/// completion is delivered: age counts from the finish, not the start.
+#[test]
+fn cleanup_ages_jobs_from_finish() {
+    let tmp = tempdir().expect("tempdir");
+    let mut manager = ShellManager::new(tmp.path().to_path_buf());
+    let long_run = FINISHED_SHELL_MAX_AGE + Duration::from_secs(600);
+    manager.seed_finished_record_for_test("long-job", long_run);
+
+    // Started 70 minutes ago, finished just now, not yet delivered.
+    manager.list_jobs();
+    assert!(
+        manager.inspect_job("long-job").is_ok(),
+        "a long job must not be evicted the moment it finishes"
+    );
+    let delivered = manager.drain_finished_jobs_with_evidence();
+    assert_eq!(delivered.len(), 1, "the completion must still be delivered");
+    assert_eq!(delivered[0].event.task_id, "long-job");
+
+    // Delivered and recently finished: still inside the retention window.
+    manager.cleanup(FINISHED_SHELL_MAX_AGE);
+    assert!(manager.inspect_job("long-job").is_ok());
+
+    // Delivered and finished longer ago than the window: evicted.
+    let shell = manager.processes.get_mut("long-job").expect("record");
+    shell.finished_at = Instant::now().checked_sub(long_run);
+    manager.cleanup(FINISHED_SHELL_MAX_AGE);
+    assert!(manager.inspect_job("long-job").is_err());
+}
+
+/// Nothing ever drains a completion owned by a Runtime API scope or by a
+/// session that is no longer active, so an undelivered job must still age out
+/// (counted from when it finished) instead of staying until the count cap.
+#[test]
+fn cleanup_ages_out_undelivered_completions_after_finish() {
+    let tmp = tempdir().expect("tempdir");
+    let mut manager = ShellManager::new(tmp.path().to_path_buf());
+    let max_age = Duration::from_secs(30);
+    manager.seed_finished_record_for_test("api-job", Duration::from_secs(50));
+    let shell = manager.processes.get_mut("api-job").expect("record");
+    shell.owner_session_id = "api:thread-1".to_string();
+    shell.finished_at = Instant::now().checked_sub(Duration::from_secs(40));
+    assert!(!shell.completion_reported);
+
+    manager.cleanup(max_age);
+    assert!(
+        manager.inspect_job("api-job").is_err(),
+        "an undelivered completion finished longer ago than max_age must age out"
+    );
+}
+
+/// The count ceiling evicts the records that finished longest ago, the same
+/// clock the age rule uses: a long job that just finished is not "oldest".
+#[test]
+fn finished_job_bounds_evict_by_finish_time() {
+    let tmp = tempdir().expect("tempdir");
+    let mut manager = ShellManager::new(tmp.path().to_path_buf());
+    let now = Instant::now();
+    for index in 0..MAX_FINISHED_SHELL_RECORDS {
+        let id = format!("short-{index:03}");
+        // Started 20 s ago, finished 10 s ago.
+        manager.seed_finished_record_for_test(id.clone(), Duration::from_secs(20));
+        let shell = manager.processes.get_mut(&id).expect("record");
+        shell.finished_at = now.checked_sub(Duration::from_secs(10));
+        shell.completion_reported = true;
+    }
+    // Started 60 s ago, finished just now.
+    manager.seed_finished_record_for_test("long-job", Duration::from_secs(60));
+    manager
+        .processes
+        .get_mut("long-job")
+        .expect("record")
+        .completion_reported = true;
+
+    manager.cleanup(FINISHED_SHELL_MAX_AGE);
+    assert_eq!(manager.tracked_job_count(), MAX_FINISHED_SHELL_RECORDS);
+    assert!(
+        manager.inspect_job("long-job").is_ok(),
+        "the most recently finished job was evicted first"
+    );
+}
+
+/// The synchronous path closes stdin when there is no input: `cat` gets EOF
+/// instead of reading, or blocking on, Codewhale's own stdin.
+#[cfg(unix)]
+#[test]
+fn sync_command_without_input_gets_eof_not_inherited_stdin() {
+    let tmp = tempdir().expect("tempdir");
+    let mut manager = ShellManager::new(tmp.path().to_path_buf());
+    let started = Instant::now();
+    let result = manager
+        .execute_with_options_env(
+            "cat; echo after-eof",
+            None,
+            15_000,
+            false,
+            None,
+            false,
+            None,
+            HashMap::new(),
+        )
+        .expect("run");
+    assert!(
+        started.elapsed() < Duration::from_secs(10),
+        "stdin read blocked for {:?}",
+        started.elapsed()
+    );
+    assert!(result.stdout.contains("after-eof"), "{}", result.stdout);
+}
+
+/// A backgrounded lowercase `bash` job: each read returns only output the
+/// caller has not seen, not the whole retained tail again.
+#[cfg(unix)]
+#[test]
+fn bounded_job_delta_returns_only_new_output() {
+    let tmp = tempdir().expect("tempdir");
+    let mut manager = ShellManager::new(tmp.path().to_path_buf());
+    let started = manager
+        .execute_with_options_env_for_owner_and_work(
+            "printf 'first-chunk\\n'; while [ ! -e go ]; do sleep 0.05; done; printf 'second-chunk\\n'",
+            None,
+            60_000,
+            true,
+            None,
+            false,
+            None,
+            HashMap::new(),
+            None,
+            String::new(),
+            None,
+            None,
+            None,
+            None,
+            false,
+            (1, BASH_MAX_TIMEOUT_MS),
+        )
+        .expect("spawn bounded job");
+    let task_id = started.task_id.expect("task id");
+    assert!(
+        manager.processes[&task_id].bounded_output.is_some(),
+        "fixture must exercise the bounded accumulator"
+    );
+
+    let deadline = Instant::now() + Duration::from_secs(10);
+    let mut seen = String::new();
+    while !seen.contains("first-chunk") {
+        assert!(Instant::now() < deadline, "first chunk never arrived");
+        let delta = manager
+            .get_output_delta(&task_id, false, 0)
+            .expect("first delta");
+        seen.push_str(&delta.result.stdout);
+        std::thread::sleep(Duration::from_millis(20));
+    }
+    assert_eq!(seen.matches("first-chunk").count(), 1, "{seen}");
+    // The fixture writes its second chunk only now, so the first delta cannot
+    // have raced past it.
+    std::fs::write(tmp.path().join("go"), "").expect("release fixture");
+
+    let rest = manager
+        .get_output_delta(&task_id, true, 10_000)
+        .expect("remaining delta");
+    assert_ne!(rest.result.status, ShellStatus::Running);
+    assert!(
+        rest.result.stdout.contains("second-chunk"),
+        "{}",
+        rest.result.stdout
+    );
+    assert!(
+        !rest.result.stdout.contains("first-chunk"),
+        "a delta must not repeat output already returned: {:?}",
+        rest.result.stdout
+    );
+}
+
+/// A foreground command that reads stdin gets EOF instead of blocking until
+/// the timeout kills it.
+#[cfg(unix)]
+#[tokio::test]
+async fn foreground_command_reading_stdin_gets_eof() {
+    let tmp = tempdir().expect("tempdir");
+    let ctx = ToolContext::new(tmp.path());
+    let started = Instant::now();
+    let result = BashTool::new("Bash")
+        .execute(
+            json!({"command": "cat; echo after-eof", "timeout_ms": 15_000}),
+            &ctx,
+        )
+        .await
+        .expect("execute");
+    assert!(
+        started.elapsed() < Duration::from_secs(10),
+        "stdin read blocked for {:?}",
+        started.elapsed()
+    );
+    assert!(result.content.contains("after-eof"), "{}", result.content);
+}
+
 /// Regression for #1691: a `git commit -m "feat: complete sub-pages"` shell
 /// command must reach the OS shell with its quoted message intact (one argv
 /// slot), never split into `feat:` / `complete` / `sub-pages"`.

crates/tui/src/tools/tasks.rs

@@ -674,26 +674,35 @@ impl TasksTool {
 
         let started = Instant::now();
         let mut cmd = build_gate_command(&command, &cwd);
-        let output =
-            tokio::time::timeout(std::time::Duration::from_millis(timeout_ms), cmd.output()).await;
+        // Contained: when the timeout elapses the gate's whole process tree
+        // is killed, instead of leaving it running behind a "timeout" result,
+        // and what it wrote until then still reaches the log.
+        let output = crate::process_tree::contained_output_until(
+            &mut cmd,
+            tokio::time::sleep(std::time::Duration::from_millis(timeout_ms)),
+        )
+        .await;
 
         let duration_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX);
         let (exit_code, stdout, stderr, timed_out, spawn_error) = match output {
-            Ok(Ok(out)) => (
-                out.status.code(),
-                String::from_utf8_lossy(&out.stdout).to_string(),
-                String::from_utf8_lossy(&out.stderr).to_string(),
-                false,
+            Ok(run) => (
+                if run.stopped {
+                    None
+                } else {
+                    run.output.status.code()
+                },
+                String::from_utf8_lossy(&run.output.stdout).to_string(),
+                String::from_utf8_lossy(&run.output.stderr).to_string(),
+                run.stopped,
                 None,
             ),
-            Ok(Err(err)) => (
+            Err(err) => (
                 None,
                 String::new(),
                 String::new(),
                 false,
                 Some(err.to_string()),
             ),
-            Err(_) => (None, String::new(), String::new(), true, None),
         };
 
         let full_log = format!(
@@ -1789,6 +1798,42 @@ mod tests {
         );
     }
 
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn timed_out_gate_run_kills_the_gate_process_tree() {
+        let workspace = tempfile::tempdir().expect("workspace");
+        let context = ToolContext::new(workspace.path())
+            .with_shell_policy(crate::worker_profile::ShellPolicy::Full);
+        let result = TasksTool::new("tasks")
+            .execute(
+                json!({
+                    "action": "gate_run",
+                    "gate": "test",
+                    "command": "echo gate-started; sleep 300 & echo $! > gate-child.pid; wait",
+                    "timeout_ms": 5_000
+                }),
+                &context,
+            )
+            .await
+            .expect("gate runs");
+        let metadata = result.metadata.expect("metadata");
+        assert_eq!(metadata["timed_out"], true, "{metadata}");
+        // What the gate wrote before the timeout is kept.
+        assert!(
+            result.content.contains("gate-started"),
+            "{}",
+            result.content
+        );
+        let child = crate::process_tree::read_pid_file(
+            &workspace.path().join("gate-child.pid"),
+            std::time::Duration::from_secs(5),
+        );
+        assert!(
+            crate::process_tree::wait_for_pid_exit(child, std::time::Duration::from_secs(5)),
+            "the timed-out gate's process is still running"
+        );
+    }
+
     #[test]
     fn gate_command_uses_non_login_shell_invocation() {
         // A login shell would source profile files that can re-export

crates/tui/src/tools/terminal_session.rs

@@ -595,15 +595,17 @@ fn prune_output(input: &str) -> String {
         .char_indices()
         .find(|(index, _)| *index >= head)
         .map_or(input.len(), |(index, _)| index);
+    // The earliest boundary that still fits: the tail is where a build's
+    // error and the command's completion marker are.
     let tail_start = input
         .char_indices()
-        .rev()
-        .find(|(index, _)| input.len() - *index <= tail)
-        .map_or(0, |(index, _)| index);
+        .map(|(index, _)| index)
+        .find(|index| *index >= head_end && input.len() - *index <= tail)
+        .unwrap_or(input.len());
     format!(
         "{}\n… [output truncated: {} bytes omitted] …\n{}",
         &input[..head_end],
-        input.len() - OUTPUT_LIMIT,
+        tail_start - head_end,
         &input[tail_start..]
     )
 }
@@ -1109,6 +1111,24 @@ mod tests {
         session_result(&mut session, done, timed_out)
     }
 
+    #[test]
+    fn prune_output_keeps_the_tail_and_counts_what_it_dropped() {
+        let input = format!(
+            "first-line\n{}\nerror: the last line é\n",
+            "x".repeat(OUTPUT_LIMIT * 2)
+        );
+        let pruned = prune_output(&input);
+        let (head, rest) = pruned
+            .split_once("\n… [output truncated: ")
+            .expect("truncation marker");
+        let (omitted, tail) = rest.split_once(" bytes omitted] …\n").expect("count");
+        let omitted: usize = omitted.parse().expect("omitted count");
+        assert!(head.starts_with("first-line"));
+        assert!(tail.ends_with("error: the last line é\n"), "{tail:?}");
+        assert!(tail.len() > OUTPUT_LIMIT / 2, "tail kept only {tail:?}");
+        assert_eq!(head.len() + omitted + tail.len(), input.len());
+    }
+
     #[test]
     #[cfg(unix)]
     fn cd_persists_between_runs() {

crates/tui/src/tools/test_runner.rs

@@ -4,6 +4,7 @@
 //! approval policy as the other code-executing tools.
 
 use std::path::Path;
+use std::time::Duration;
 
 use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
@@ -27,6 +28,10 @@ struct RunTestsOutput {
     stdout: String,
     stderr: String,
     command: String,
+    /// The run hit [`RUN_TESTS_TIMEOUT`] and was killed; stdout/stderr are
+    /// what it wrote until then.
+    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
+    timed_out: bool,
 }
 
 #[async_trait]
@@ -100,17 +105,26 @@ impl ToolSpec for RunTestsTool {
         }
 
         let command_str = format_command(&workdir, &args);
-        let output = run_cargo(&workdir, &args)?;
+        let run = run_cargo(&workdir, &args, context, RUN_TESTS_TIMEOUT).await?;
+        let output = run.output;
+        let mut stderr = String::from_utf8_lossy(&output.stderr).into_owned();
+        if run.stopped {
+            stderr.push_str(&format!(
+                "\n[run_tests: stopped after {} s; the output above is everything cargo wrote before it was killed]",
+                RUN_TESTS_TIMEOUT.as_secs()
+            ));
+        }
 
         // The whole output: the end of a cargo run is where the failures and
         // the `test result:` line are. Size is the engine's one recoverable
         // budget (#6508), so the failure summary sees everything.
         run_tests_result(RunTestsOutput {
-            success: output.status.success(),
+            success: !run.stopped && output.status.success(),
             exit_code: output.status.code().unwrap_or(-1),
             stdout: String::from_utf8_lossy(&output.stdout).into_owned(),
-            stderr: String::from_utf8_lossy(&output.stderr).into_owned(),
+            stderr,
             command: command_str,
+            timed_out: run.stopped,
         })
     }
 }
@@ -134,8 +148,21 @@ fn run_tests_result(result: RunTestsOutput) -> Result<ToolResult, ToolError> {
 
 // === Helpers ===
 
-fn run_cargo(workspace: &Path, args: &[String]) -> Result<std::process::Output, ToolError> {
-    let Some(mut cmd) = crate::dependencies::Cargo::command() else {
+/// Ceiling for one `cargo test` run. Long suites fit; a hung test does not
+/// hold the turn forever.
+const RUN_TESTS_TIMEOUT: Duration = Duration::from_secs(30 * 60);
+
+/// Run cargo without blocking a runtime worker. Stop and the timeout both end
+/// the whole process tree (cargo and the test binaries it started). A timeout
+/// still returns the output written so far (`stopped`): it is the only
+/// evidence of which test hung.
+async fn run_cargo(
+    workspace: &Path,
+    args: &[String],
+    context: &ToolContext,
+    timeout: Duration,
+) -> Result<crate::process_tree::ContainedOutput, ToolError> {
+    let Some(mut cmd) = crate::dependencies::Cargo::tokio_command() else {
         return Err(ToolError::not_available(
             "cargo is not installed or not in PATH",
         ));
@@ -143,8 +170,20 @@ fn run_cargo(workspace: &Path, args: &[String]) -> Result<std::process::Output,
     cmd.args(args).current_dir(workspace);
     // `cargo test` builds and runs workspace code; do not hand it parent
     // credentials.
-    crate::child_env::apply_to_command(&mut cmd, std::iter::empty::<(&str, &str)>());
-    cmd.output().map_err(|e| {
+    crate::child_env::apply_to_tokio_command(&mut cmd, std::iter::empty::<(&str, &str)>());
+    let run = crate::process_tree::contained_output_until(&mut cmd, tokio::time::sleep(timeout));
+    let cancelled = async {
+        match context.cancel_token.as_ref() {
+            Some(token) => token.cancelled().await,
+            None => std::future::pending::<()>().await,
+        }
+    };
+    // Dropping `run` on cancel kills the tree.
+    let output = tokio::select! {
+        output = run => output,
+        () = cancelled => return Err(ToolError::cancelled("cargo test cancelled")),
+    };
+    output.map_err(|e| {
         if e.kind() == std::io::ErrorKind::NotFound {
             ToolError::not_available("cargo is not installed or not in PATH")
         } else {
@@ -214,6 +253,24 @@ mod tests {
         );
     }
 
+    /// Stop must reach `cargo test`: the call returns as cancelled instead of
+    /// blocking a runtime worker until cargo exits on its own.
+    #[tokio::test]
+    async fn run_tests_honors_cancellation() {
+        if !cargo_available() {
+            return;
+        }
+        let tmp = tempdir().expect("tempdir");
+        let token = tokio_util::sync::CancellationToken::new();
+        token.cancel();
+        let ctx = ToolContext::new(tmp.path()).with_cancel_token(token);
+        let result = RunTestsTool.execute(json!({}), &ctx).await;
+        assert!(
+            matches!(result, Err(ToolError::Cancelled { .. })),
+            "cancelled run_tests must not run to completion: {result:?}"
+        );
+    }
+
     #[tokio::test]
     async fn run_tests_succeeds_on_fresh_project() {
         if !cargo_available() {
@@ -305,6 +362,7 @@ mod tests {
             stdout: stdout.clone(),
             stderr: String::new(),
             command: "(cd /repo && cargo test)".to_string(),
+            timed_out: false,
         })
         .expect("result");
 
@@ -378,8 +436,8 @@ mod tests {
     }
 
     #[cfg(unix)]
-    #[test]
-    fn run_cargo_does_not_inherit_parent_secret_env() {
+    #[tokio::test]
+    async fn run_cargo_does_not_inherit_parent_secret_env() {
         use crate::test_support::{EnvVarGuard, lock_test_env};
         use std::os::unix::fs::PermissionsExt;
         if !cargo_available() {
@@ -405,7 +463,17 @@ mod tests {
         let _target = EnvVarGuard::set("CARGO_TARGET_DIR", "/tmp/codewhale-fixture-target");
         let workspace = tempdir().expect("workspace");
 
-        let output = run_cargo(workspace.path(), &["envprobe".to_string()]).expect("cargo runs");
+        let ctx = ToolContext::new(workspace.path());
+        let run = run_cargo(
+            workspace.path(),
+            &["envprobe".to_string()],
+            &ctx,
+            RUN_TESTS_TIMEOUT,
+        )
+        .await
+        .expect("cargo runs");
+        assert!(!run.stopped);
+        let output = run.output;
         let stdout = String::from_utf8_lossy(&output.stdout);
         assert!(
             output.status.success(),
@@ -417,4 +485,52 @@ mod tests {
             "secret=unset target=/tmp/codewhale-fixture-target"
         );
     }
+
+    /// A run that hits the timeout is killed, tree and all, and still returns
+    /// what it wrote: the only evidence of which test hung.
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn timed_out_run_cargo_keeps_partial_output_and_kills_the_tree() {
+        use crate::test_support::{EnvVarGuard, lock_test_env};
+        use std::os::unix::fs::PermissionsExt;
+        if !cargo_available() {
+            return;
+        }
+        let _env_lock = lock_test_env();
+        let bin = tempdir().expect("bin dir");
+        let probe = bin.path().join("cargo-hangprobe");
+        fs::write(
+            &probe,
+            "#!/bin/sh\necho 'test slow_case ... started'\nsleep 300 &\necho $! > hang.pid\nwait\n",
+        )
+        .expect("write probe");
+        fs::set_permissions(&probe, fs::Permissions::from_mode(0o755)).expect("chmod probe");
+        let path = std::env::var_os("PATH").unwrap_or_default();
+        let mut paths = vec![bin.path().to_path_buf()];
+        paths.extend(std::env::split_paths(&path));
+        let _path = EnvVarGuard::set("PATH", std::env::join_paths(paths).expect("join PATH"));
+        let workspace = tempdir().expect("workspace");
+        let pid_file = workspace.path().join("hang.pid");
+
+        let ctx = ToolContext::new(workspace.path());
+        let run = run_cargo(
+            workspace.path(),
+            &["hangprobe".to_string()],
+            &ctx,
+            Duration::from_secs(10),
+        )
+        .await
+        .expect("a timed-out run still returns its output");
+        assert!(run.stopped);
+        assert!(
+            String::from_utf8_lossy(&run.output.stdout).contains("test slow_case ... started"),
+            "{:?}",
+            run.output
+        );
+        let hung = crate::process_tree::read_pid_file(&pid_file, Duration::from_secs(5));
+        assert!(
+            crate::process_tree::wait_for_pid_exit(hung, Duration::from_secs(5)),
+            "the timed-out run's process is still running"
+        );
+    }
 }

crates/tui/tests/cucumber/automations_editor_pty.rs

@@ -245,7 +245,14 @@ completion_sound = "off"
         click(&mut reopened, "Confirm");
         wait(&mut reopened, "y/Enter");
         click(&mut reopened, "Confirm");
-        wait(&mut reopened, "deleted");
+        // The review already says "Nothing was deleted"; wait for the durable
+        // mutation and its receipt after the armed review has closed.
+        reopened
+            .wait_for(
+                |frame| !path.exists() && frame.contains("deleted") && !frame.contains("y/Enter"),
+                WAIT,
+            )
+            .unwrap_or_else(|error| panic!("deletion did not complete at {cols}x{rows}: {error}"));
         assert!(
             !path.exists(),
             "confirmed mouse control deletes the reviewed definition"