fix(tools): review follow-ups for shell retention and contained child processes

- Shell job age: undelivered completions age out again, counted from
  finish. Jobs launched over the Runtime API, or owned by an inactive
  session, are never drained, so the earlier exemption kept them until the
  count cap.
- Count/byte eviction orders finished jobs by finish time, the same clock
  as the age rule, so a long job that just finished is no longer evicted
  first.
- Synchronous shell path: stdin is closed (not inherited) when no input
  is given, so `cat` or a prompt gets EOF instead of reading Codewhale's
  own stdin until the timeout.
- process_tree: contained_output_until returns partial output when a stop
  future fires; run_tests and tasks gate_run use it, so a timed-out run
  still reports what it wrote. A command that exits on its own is released
  instead of having its process group killed, as with a bare output().
  In-flight contained groups are killed on the signal-exit path, which
  runs no destructors.
- Omitted-bytes notice in bash deltas names the spill file.
- Tests wait for fixture pid files instead of fixed sleeps, treat zombies
  as exited, and the delta test gates its second chunk on a file.

Tests: 233 passed / 0 failed / 1 ignored across process_tree, tools::shell,
js_execution, tasks, test_runner, terminal_session, run_tool; broader
filter (runtime_api jobs shell bash stdin gate terminal test_runner)
1501 passed / 0 failed. 8 new regression tests fail with the fixes reverted.
clippy (CI flags) and fmt clean.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_014ZwqatxgVFxHvovngywnks

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

@@ -17,7 +17,8 @@
 //!   any plugin code.
 //!
 //! 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 +131,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)]
@@ -152,6 +169,34 @@ impl Drop for ProcessTree {
 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> {
     use std::process::Stdio;
     cmd.stdin(Stdio::null())
         .stdout(Stdio::piped())
@@ -159,10 +204,134 @@ pub(crate) async fn contained_output(
         .kill_on_drop(true);
     #[cfg(unix)]
     cmd.process_group(0);
-    let child = cmd.spawn()?;
+    let mut child = cmd.spawn()?;
     // Best effort: without the tree guard, `kill_on_drop` still ends the child.
-    let _tree = ProcessTree::attach_tokio(&child).ok();
-    child.wait_with_output().await
+    let tree = ProcessTree::attach_tokio(&child).ok();
+    // 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 mut stdout = Vec::new();
+    let mut stderr = Vec::new();
+    let exited = {
+        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(),
+            );
+            out?;
+            err?;
+            status
+        };
+        tokio::select! {
+            status = run => Some(status?),
+            () = stop => None,
+        }
+    };
+    let (status, stopped) = match exited {
+        Some(status) => {
+            if let Some(tree) = tree {
+                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)]
@@ -193,6 +362,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) }
@@ -215,12 +398,14 @@ pub(crate) fn windows_io_error(error: windows::core::Error) -> std::io::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 {
+        if unsafe { libc::kill(pid, 0) } != 0 || is_zombie(pid) {
             return true;
         }
         if std::time::Instant::now() >= deadline {
@@ -233,15 +418,24 @@ pub(crate) fn wait_for_pid_exit(pid: libc::pid_t, within: std::time::Duration) -
     }
 }
 
+/// 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) = std::fs::read_to_string(path)
-            .ok()
-            .and_then(|text| text.trim().parse().ok())
-        {
+        if let Some(pid) = parse_pid_file(path) {
             return pid;
         }
         assert!(
@@ -253,6 +447,39 @@ pub(crate) fn read_pid_file(path: &std::path::Path, within: std::time::Duration)
     }
 }
 
+#[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::*;
@@ -266,10 +493,7 @@ mod tests {
         cmd.arg("-c")
             .arg("sleep 300 & echo $! > grandchild.pid; wait")
             .current_dir(tmp.path());
-        let timed_out =
-            tokio::time::timeout(Duration::from_millis(500), contained_output(&mut cmd)).await;
-        assert!(timed_out.is_err(), "fixture must still be running");
-        let grandchild = read_pid_file(&pid_file, Duration::from_secs(5));
+        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"
@@ -287,4 +511,78 @@ mod tests {
         assert!(output.status.success());
         assert_eq!(String::from_utf8_lossy(&output.stdout), "done\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));
+        assert!(wait_for_pid_exit(leader, Duration::from_secs(5)));
+    }
 }

crates/tui/src/tools/js_execution.rs

@@ -290,16 +290,12 @@ mod tests {
                     const child = spawn('sleep', ['300'], { stdio: 'ignore' });\n\
                     require('fs').writeFileSync('grandchild.pid', String(child.pid));\n\
                     setInterval(() => {}, 1000);";
-        let dropped = tokio::time::timeout(
-            Duration::from_secs(3),
-            execute_js_execution_tool(&json!({ "code": code }), tmp.path()),
+        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!(dropped.is_err(), "fixture must still be running");
-        let grandchild = crate::process_tree::read_pid_file(
-            &tmp.path().join("grandchild.pid"),
-            Duration::from_secs(5),
-        );
         assert!(
             crate::process_tree::wait_for_pid_exit(grandchild, Duration::from_secs(5)),
             "a process started by the dropped script is still running"

crates/tui/src/tools/shell.rs

@@ -1526,11 +1526,14 @@ impl BackgroundShell {
             }
             self.last_output_at = Instant::now();
             self.last_observed_output_len = total_bytes;
-            if omitted > 0 {
-                delta.insert_str(
-                    0,
-                    &format!("[{omitted} bytes of earlier output not retained in memory]\n"),
-                );
+            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);
         }
@@ -2393,9 +2396,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);
@@ -3649,11 +3657,12 @@ impl ShellManager {
     ///
     /// 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. A completion nobody has received yet is never aged out; only
-    /// the count and byte ceilings below may drop one.
+    /// 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 || !shell.completion_reported {
+            if shell.status == ShellStatus::Running {
                 return true;
             }
             shell.mark_finished();
@@ -3692,7 +3701,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(),
                 )
             })
@@ -4837,10 +4847,11 @@ async fn execute_foreground_via_background(
         manager.attach_heavy_permit(&task_id, permit)?;
     }
 
-    // A foreground pipe command gets EOF on stdin, like the synchronous
-    // path: 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 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

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

@@ -244,6 +244,27 @@ impl BoundedOutputAccumulator {
         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()));
@@ -835,6 +856,27 @@ mod tests {
         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

@@ -3811,7 +3811,7 @@ 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_and_keeps_undelivered_completions() {
+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);
@@ -3838,6 +3838,86 @@ fn cleanup_ages_jobs_from_finish_and_keeps_undelivered_completions() {
     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)]
@@ -3847,7 +3927,7 @@ fn bounded_job_delta_returns_only_new_output() {
     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'; sleep 1; printf 'second-chunk\\n'",
+            "printf 'first-chunk\\n'; while [ ! -e go ]; do sleep 0.05; done; printf 'second-chunk\\n'",
             None,
             60_000,
             true,
@@ -3882,6 +3962,9 @@ fn bounded_job_delta_returns_only_new_output() {
         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)

crates/tui/src/tools/tasks.rs

@@ -675,30 +675,34 @@ impl TasksTool {
         let started = Instant::now();
         let mut cmd = build_gate_command(&command, &cwd);
         // Contained: when the timeout elapses the gate's whole process tree
-        // is killed, instead of leaving it running behind a "timeout" result.
-        let output = tokio::time::timeout(
-            std::time::Duration::from_millis(timeout_ms),
-            crate::process_tree::contained_output(&mut cmd),
+        // 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!(
@@ -1805,8 +1809,8 @@ mod tests {
                 json!({
                     "action": "gate_run",
                     "gate": "test",
-                    "command": "sleep 300 & echo $! > gate-child.pid; wait",
-                    "timeout_ms": 1_000
+                    "command": "echo gate-started; sleep 300 & echo $! > gate-child.pid; wait",
+                    "timeout_ms": 5_000
                 }),
                 &context,
             )
@@ -1814,6 +1818,12 @@ mod tests {
             .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),

crates/tui/src/tools/test_runner.rs

@@ -28,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]
@@ -101,17 +105,26 @@ impl ToolSpec for RunTestsTool {
         }
 
         let command_str = format_command(&workdir, &args);
-        let output = run_cargo(&workdir, &args, context, RUN_TESTS_TIMEOUT).await?;
+        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,
         })
     }
 }
@@ -140,13 +153,15 @@ fn run_tests_result(result: RunTestsOutput) -> Result<ToolResult, ToolError> {
 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).
+/// 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<std::process::Output, ToolError> {
+) -> 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",
@@ -156,17 +171,16 @@ async fn run_cargo(
     // `cargo test` builds and runs workspace code; do not hand it parent
     // credentials.
     crate::child_env::apply_to_tokio_command(&mut cmd, std::iter::empty::<(&str, &str)>());
-    let run = tokio::time::timeout(timeout, crate::process_tree::contained_output(&mut cmd));
+    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.map_err(|_| ToolError::Timeout {
-            seconds: timeout.as_secs(),
-        })?,
+        output = run => output,
         () = cancelled => return Err(ToolError::cancelled("cargo test cancelled")),
     };
     output.map_err(|e| {
@@ -348,6 +362,7 @@ mod tests {
             stdout: stdout.clone(),
             stderr: String::new(),
             command: "(cd /repo && cargo test)".to_string(),
+            timed_out: false,
         })
         .expect("result");
 
@@ -449,7 +464,7 @@ mod tests {
         let workspace = tempdir().expect("workspace");
 
         let ctx = ToolContext::new(workspace.path());
-        let output = run_cargo(
+        let run = run_cargo(
             workspace.path(),
             &["envprobe".to_string()],
             &ctx,
@@ -457,6 +472,8 @@ mod tests {
         )
         .await
         .expect("cargo runs");
+        assert!(!run.stopped);
+        let output = run.output;
         let stdout = String::from_utf8_lossy(&output.stdout);
         assert!(
             output.status.success(),
@@ -468,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"
+        );
+    }
 }