fix(exec): bounded post-kill reaps — D-state children can't wedge a turn

Port of upstream grok-build 1.0.1 (landed upstream 2026-08-06).

A child wedged in an uninterruptible kernel syscall (D-state) never
observes SIGKILL, and a background descendant can hold the output pipes
open past the kill; both previously stalled tool futures without bound.

- kigi-tty-utils: KILL_REAP_TIMEOUT (5s) + reap_killed_bounded — wait
  for an already-killed child at most the bound, then abandon the corpse
  to tokio's orphan reaper.
- local_terminal: pipe readers share their buffer with the spawner and
  join under the reap bound on both exit arms (aborted readers are
  awaited so no tail bytes race the take); the whole tree is enrolled in
  a ProcessGroup and group-killed on timeout; the timeout arm returns
  the synthetic result instead of waiting on the corpse. Grandchild
  kill + prompt-return pinned by a ported unix regression test.
- kigi-tools: reap_killed_search_child wrapper; grep's four kill sites
  reap bounded; glob's stderr goes to null (an unread pipe blocks rg on
  warning backpressure) and its truncation kill reaps bounded.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KSy9S31pZNWXtRZNVJ1sDf

crates/codegen/kigi-shell/src/terminal/local_terminal.rs

@@ -8,13 +8,66 @@ use tokio::time;
 use crate::terminal::runner::{
     AsyncTerminalRunner, TerminalError, TerminalRunRequest, TerminalRunResult,
 };
+use kigi_tty_utils::KILL_REAP_TIMEOUT;
 
 pub struct LocalTerminalRunner;
 
-async fn read_stream(mut stream: impl AsyncReadExt + Unpin) -> Vec<u8> {
-    let mut buffer = Vec::new();
-    let _ = stream.read_to_end(&mut buffer).await;
-    buffer
+/// A pipe reader whose buffer is shared with the spawner, so bytes read
+/// before an abort are not lost with the task.
+struct PipeReader {
+    task: tokio::task::JoinHandle<()>,
+    buffer: std::sync::Arc<std::sync::Mutex<Vec<u8>>>,
+}
+
+impl PipeReader {
+    fn spawn(mut stream: impl AsyncReadExt + Unpin + Send + 'static) -> Self {
+        let buffer = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
+        let task_buffer = std::sync::Arc::clone(&buffer);
+        let task = tokio::spawn(async move {
+            let mut chunk = [0u8; 8192];
+            loop {
+                match stream.read(&mut chunk).await {
+                    Ok(0) | Err(_) => break,
+                    Ok(n) => Self::lock_buffer(&task_buffer).extend_from_slice(&chunk[..n]),
+                }
+            }
+        });
+        Self { task, buffer }
+    }
+
+    /// Never panics: a poisoned lock just means a holder panicked mid-append —
+    /// the buffer is plain bytes and stays consistent, so recover it.
+    fn lock_buffer(buffer: &std::sync::Mutex<Vec<u8>>) -> std::sync::MutexGuard<'_, Vec<u8>> {
+        buffer
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+    }
+
+    fn take_buffer(self) -> Vec<u8> {
+        std::mem::take(&mut Self::lock_buffer(&self.buffer))
+    }
+
+    /// Wait for EOF (pipe closed by all writers) at most [`KILL_REAP_TIMEOUT`],
+    /// then return everything read so far.
+    ///
+    /// Callers only join after the child was reaped or killed, so EOF is
+    /// normally immediate; the bound is a backstop for writers that outlive
+    /// the command and keep the pipe open — a background descendant
+    /// (`printf hi; sleep 30 &`) or a process wedged in uninterruptible
+    /// kernel I/O (D-state), which not even SIGKILL moves.
+    async fn join_bounded(mut self) -> Vec<u8> {
+        if time::timeout(KILL_REAP_TIMEOUT, &mut self.task)
+            .await
+            .is_err()
+        {
+            self.task.abort();
+            // Await the cancellation: abort() is asynchronous, and a reader
+            // past its await point could otherwise append between the take
+            // below and its own completion, losing those bytes.
+            let _ = (&mut self.task).await;
+        }
+        self.take_buffer()
+    }
 }
 
 /// Truncate buffer to keep only the last `limit` bytes (drops oldest bytes).
@@ -66,12 +119,35 @@ impl AsyncTerminalRunner for LocalTerminalRunner {
 
         // Detach from the controlling terminal so child processes
         // (e.g. GPG pinentry) cannot open /dev/tty and corrupt the TUI.
+        // (This also makes the child its own session/group leader, which the
+        // ProcessGroup attach below relies on.)
         kigi_tools::util::detach_command(&mut cmd);
 
+        // process-group-killed on the request timeout; an unreapable (D-state)
+        // child is abandoned to the runtime's orphan reaper
         let mut child = cmd
             .spawn()
             .map_err(|e| TerminalError::Other(format!("Failed to start shell: {e}")))?;
 
+        // Own the whole tree, not just the direct shell: on timeout the group
+        // is killed so grandchildren can't keep running (and can't hold the
+        // output pipes open past the kill). Group teardown is an enhancement
+        // over the direct-kill floor, so both enrollment failures degrade
+        // softly instead of dropping an already-running shell.
+        let process_group = match kigi_tty_utils::ProcessGroup::new() {
+            Ok(mut group) => {
+                if let Err(e) = group.attach(&child) {
+                    // e.g. the child already exited.
+                    tracing::debug!("ProcessGroup::attach failed: {e}");
+                }
+                Some(group)
+            }
+            Err(e) => {
+                tracing::warn!("Failed to create process group: {e}");
+                None
+            }
+        };
+
         let stdout = child
             .stdout
             .take()
@@ -81,28 +157,46 @@ impl AsyncTerminalRunner for LocalTerminalRunner {
             .take()
             .ok_or_else(|| TerminalError::Other("Failed to capture stderr".into()))?;
 
-        let stdout_task = tokio::spawn(read_stream(stdout));
-        let stderr_task = tokio::spawn(read_stream(stderr));
+        let stdout_reader = PipeReader::spawn(stdout);
+        let stderr_reader = PipeReader::spawn(stderr);
 
         let mut timed_out = false;
 
         let wait_result = time::timeout(request.timeout, child.wait()).await;
-        let exit_status = match wait_result {
+        let exit_code = match wait_result {
             Ok(status_res) => status_res
-                .map_err(|e| TerminalError::Other(format!("Failed to wait for process: {e}")))?,
+                .map_err(|e| TerminalError::Other(format!("Failed to wait for process: {e}")))?
+                .code(),
             Err(_) => {
                 timed_out = true;
+                // Kill the direct child AND the whole group (grandchildren),
+                // then return the synthetic timeout result without waiting
+                // for the corpse: `timed_out: true` already tells the caller
+                // everything, and a D-state child would never become reapable
+                // anyway. Both kills are unconditional — `kill()` on a group
+                // whose attach failed is a silent no-op, so the direct
+                // `start_kill` is the guaranteed floor (same as
+                // `util/subprocess.rs`). The kills close the pipes, so the
+                // bounded joins below return immediately in the normal case;
+                // the abandoned child goes to tokio's orphan reaper.
                 if let Err(e) = child.start_kill() {
                     tracing::warn!("Failed to kill timed-out process: {e}");
                 }
-                child.wait().await.map_err(|e| {
-                    TerminalError::Other(format!("Failed to wait for killed process: {e}"))
-                })?
+                if let Some(group) = &process_group
+                    && let Err(e) = group.kill()
+                {
+                    tracing::warn!("Failed to kill timed-out process group: {e}");
+                }
+                None
             }
         };
 
-        let stdout_result = stdout_task.await.unwrap_or_else(|_| Vec::new());
-        let stderr_result = stderr_task.await.unwrap_or_else(|_| Vec::new());
+        // Join concurrently (worst case one reap bound, not one per pipe).
+        // Bounded on the natural-exit arm too: a background descendant that
+        // outlives the shell (`printf hi; sleep 30 &`) holds the pipes open
+        // past the child's exit, and an unbounded join stalls on it.
+        let (stdout_result, stderr_result) =
+            tokio::join!(stdout_reader.join_bounded(), stderr_reader.join_bounded());
 
         let mut combined = stdout_result;
         combined.extend(stderr_result);
@@ -112,7 +206,7 @@ impl AsyncTerminalRunner for LocalTerminalRunner {
 
         Ok(TerminalRunResult {
             combined_output,
-            exit_code: exit_status.code(),
+            exit_code,
             truncated,
             signal: None,
             timed_out,
@@ -181,4 +275,56 @@ mod tests {
         assert_eq!(result.combined_output.trim(), "hello");
         assert_eq!(result.exit_code, Some(0));
     }
+
+    /// Timing out a command whose grandchild inherited the output pipes must
+    /// (a) return promptly — before the group kill, the joins waited for the
+    /// grandchild's EOF (the full sleep here; forever for a writer wedged in
+    /// uninterruptible kernel I/O, the production incident this guards
+    /// against) — and (b) actually kill the grandchild via the group kill,
+    /// not merely stop waiting for it.
+    #[tokio::test]
+    #[cfg(unix)]
+    async fn test_timeout_kills_grandchildren_and_returns_promptly() {
+        let mut request = make_request("sleep 5 & echo bgpid=$!; sleep 5");
+        request.timeout = std::time::Duration::from_millis(300);
+
+        let started = std::time::Instant::now();
+        let result = LocalTerminalRunner.run(request).await.unwrap();
+
+        assert!(result.timed_out, "run should report the timeout");
+        // Prompt return: request timeout (0.3s) + pipe EOF from the group
+        // kill (normally instant; KILL_REAP_TIMEOUT-bounded worst case) —
+        // anything near the 5s sleeps means we waited for the grandchild.
+        assert!(
+            started.elapsed() < std::time::Duration::from_secs(4),
+            "timeout path must not wait for the killed tree (took {:?})",
+            started.elapsed()
+        );
+
+        // Outcome: the background grandchild must actually be dead, proving
+        // the group kill reached it. Poll briefly — the SIGKILL is delivered
+        // synchronously but init may reap the orphan a beat later.
+        let bg_pid = result
+            .combined_output
+            .lines()
+            .find_map(|l| l.trim().strip_prefix("bgpid="))
+            .and_then(|p| p.trim().parse::<u32>().ok())
+            .expect("background pid should have been echoed before the kill");
+        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
+        loop {
+            let alive = std::process::Command::new("kill")
+                .args(["-0", &bg_pid.to_string()])
+                .status()
+                .map(|s| s.success())
+                .unwrap_or(false);
+            if !alive {
+                break;
+            }
+            assert!(
+                std::time::Instant::now() < deadline,
+                "background grandchild {bg_pid} survived the group kill"
+            );
+            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        }
+    }
 }