fix(run): bound child output capture and reaper threads (#340)

* fix(run): bound child output capture and reaper threads (#339)

The shell-out helpers redirected child stdout/stderr to tempfiles but then
read both streams fully into heap Vecs, so peak RSS scaled 1:1 with child
output volume and summed across concurrent calls (measured: 128 MiB child
output -> +128 MiB RSS; 8 concurrent captures sum). The doc comments claimed
bounded capture; nothing bounded it.

- Diagnostics capture is now capped at 256 KiB per stream, tail-preserving
  (the last bytes are what error messages need), with a diagnostic line on
  truncation.
- `pty list --json` moves to an explicit full-stdout variant: parsed
  structured output must be whole; stdout is intentionally uncapped there.
- One shared reaper thread replaces detached thread-per-timed-out-child,
  which accumulated without bound under timeout storms.
- Eval run steps stream child output straight into their capture files and
  combined logs instead of buffering whole outputs; agent log dumps stream
  scrollback to file without materializing it; the bash judge uses null
  stdio instead of undrained pipes that could deadlock on chatty judges.

Decision record: docs/vrs/.decisions/0007.

agent-identity: unknown
agent-persona: generalist
agent-supervisor: unavailable
agent-tool: OMP
agent-tool-version: 18.0.3
agent-runtime: OMP 18.0.3
tooling-profile: dotfiles@ffce621

* docs(vrs): carry the bounded child-capture contract (R34)

Decision 0007 landed with the fix, but the requirement and spec did not
state the bound. R32's reaper clause already covered teardown ownership;
R34 now requires capture memory to be cap-bounded, tail-preserving, and
observable, with unbounded reads opt-in at the call site. The supervision
spec gains a child-process execution section pointing at D0007.

agent-identity: unknown
agent-persona: generalist
agent-supervisor: unavailable
agent-tool: OMP
agent-tool-version: 18.0.3
agent-runtime: OMP 18.0.3
tooling-profile: dotfiles@ffce621

Author: schickling-assistant

docs/vrs/.decisions/0007-child-output-capture-is-bounded-and-tail-preserving.md

@@ -0,0 +1,58 @@
+# Child process output capture is bounded and tail-preserving
+
+Status: accepted
+
+Decision made by Johannes on 2026-08-25 (issue #339, independently reproduced
+measurements in the session record; option A of three, cap size 256 KiB).
+
+## Context
+
+The supervisor shell-out helpers (`src/run.rs`
+`output_with_input_timeout_observed`, `src/ding/mod.rs` `output_with_timeout`)
+redirect child stdout/stderr to tempfiles — sound, because files keep an
+escaped descendant from blocking cleanup and make bounded read-back
+deadlock-free — then unconditionally rewind both streams and `read_to_end`
+them into fresh heap buffers. Measured peak RSS scales 1:1 with child output
+(16 MiB child → +16.2 MiB RSS; 8 concurrent × 16 MiB → +103 MiB) despite doc
+comments claiming "bounded output capture". No spec or invariant defined a
+bound; the comment described aspiration, not behavior.
+
+Consumer inventory: only `pty list --json` (parsed JSON, naturally bounded by
+catalog size) and `pty peek` (terminal screen text, consumed for composer
+matching where the tail is what matters) need complete stdout. Every other
+call site uses stderr only trimmed inside error strings. The same audit found
+two sibling hazards: detached reaper threads accumulate without bound under
+timeout storms, and `src/eval_run.rs` holds whole step output plus full
+scrollback in memory before writing log files, with an undrained pipe pair in
+the bash judge that can deadlock.
+
+## Decision
+
+1. **Diagnostics capture is capped at 256 KiB per stream and tail-preserving.**
+   When a stream exceeds the cap, the last `CAPTURE_CAP_BYTES` bytes are kept;
+   the head is dropped. Truncation emits one diagnostic line naming the
+   command, stream, kept/total bytes, and cap.
+2. **Payload capture stays complete and explicit.** Callers that parse
+   structured data (`pty list --json`) use a distinctly named
+   full-stdout variant whose doc comment states that stdout is intentionally
+   uncapped and why. Bounded-tail remains the default so an uncapped read is
+   always a visible, deliberate choice at the call site.
+3. **One shared reaper thread** drains killed children over a channel,
+   replacing per-timeout detached threads.
+4. **Eval run steps stream to their log files** instead of buffering, and the
+   bash judge uses null stdio (only its exit status is consumed).
+
+Rejected alternatives: disk spill references for oversized diagnostics
+(spill-file lifecycle for no demonstrated consumer) and streaming every
+shell-out to log files exec-backend style (changes every error path; revisit
+only if a consumer needs full oversized diagnostics).
+
+## Consequences
+
+- Peak supervisor RSS no longer scales with child output volume; worst case
+  is bounded by calls × 512 KiB regardless of child behavior.
+- Diagnostics for oversized children lose their head. Error messages built
+  from stderr keep their tail, which is where failure text lives.
+- The tempfile redirection is now load-bearing for more than cleanup: it is
+  what makes the bounded read-back deadlock-free. Any future move back to
+  pipes must preserve a bound on buffered bytes.

docs/vrs/requirements.md

@@ -72,6 +72,13 @@ accepted.
   that outlive that child. st2 either reaps the direct child before returning or
   transfers wait ownership to a background reaper; the failure remains bounded
   and reports its originating input error or timeout.
+- **R34 Bounded helper output capture:** Capturing a spawned non-interactive
+  helper's stdout/stderr consumes memory bounded by a fixed per-stream cap
+  independent of the child's output volume and of how many captures run
+  concurrently. When a stream exceeds the cap, the retained bytes are that
+  stream's tail, and truncation is observable. A caller that must consume a
+  stream whole (structured data for parsing) opts in through an explicitly
+  named capture path, so an unbounded read is always visible at its call site.
 - **R22 Quiet coordination after events:** A network with minimal or default
   personas stays quiet while useful work continues. Agents coordinate only after
   an inbox DING, a durable failure, a real blocker, a completion or decision

docs/vrs/spec.md

@@ -710,6 +710,26 @@ validate ──► materialize ──► host-local st2 scheduler/reconciler
   closed instead of hanging reconciliation. The deadline is containment, not
   the mechanism for admitting a larger fleet.
 
+### Child-process execution (R32, R34)
+
+Non-interactive helper shells-outs (`src/run.rs`, `src/ding/mod.rs`) share one
+shape: the child is `setsid` so its pid is its process group, stdout/stderr go
+to unlinked tempfiles (an escaped descendant that inherited them cannot block
+cleanup), and deadline expiry kills the whole group. Wait ownership for a
+killed child transfers to one shared reaper thread draining a channel — never
+one detached thread per timed-out child, which accumulates without bound under
+timeout storms.
+
+Read-back is tail-capped at `CAPTURE_CAP_BYTES` (256 KiB) per stream: over-cap
+streams keep their last 256 KiB and emit one diagnostic line naming the
+command, stream, kept/total bytes, and cap. Memory per capture is therefore
+bounded by calls × 2 × cap regardless of child behavior. `pty list --json`
+parses structured output that must be whole, so it uses the explicitly named
+full-stdout variant; that read is intentionally uncapped and visible at its
+call site. Eval run steps and agent log dumps stream child output straight to
+their catalog log files without buffering it. Rationale and rejected
+alternatives: [decision 0007](.decisions/0007-child-output-capture-is-bounded-and-tail-preserving.md).
+
 ## Message lifecycle
 
 ```text

src/ding/mod.rs

@@ -14,7 +14,6 @@
 //! work into one generic recovery DING. `busy` never suppresses a notification; fresh `dnd` does.
 
 use std::collections::{HashSet, VecDeque};
-use std::io::{Read as _, Seek as _};
 use std::os::unix::process::CommandExt as _;
 use std::path::{Path, PathBuf};
 use std::process::{Command, Output, Stdio};
@@ -27,7 +26,9 @@ mod composer;
 mod harness;
 
 use crate::message::{self, Message};
+use crate::run::{CAPTURE_CAP_BYTES, reap_detached, read_bounded_tail};
 use crate::status;
+
 use composer::{ComposerState, classify_composer, classify_receipt};
 use harness::ReceiptState;
 
@@ -329,6 +330,9 @@ impl PtyPoker {
         Ok(())
     }
 
+    /// Reads the terminal screen of the session. Output capture is tail-capped at
+    /// [`crate::run::CAPTURE_CAP_BYTES`]; semantics are preserved because a terminal screen is
+    /// far below that bound.
     fn peek(&self) -> anyhow::Result<String> {
         let out = output_with_timeout(
             Command::new(&self.bin).args(["peek", self.session.as_str()]),
@@ -391,8 +395,10 @@ impl Poker for PtyPoker {
     }
 }
 
-/// Run a non-interactive child with bounded output capture. Temporary files keep an escaped
-/// descendant that inherited stdout/stderr from blocking cleanup after the direct child times out.
+/// Run a non-interactive child with bounded output capture: each stream keeps at most its last
+/// [`crate::run::CAPTURE_CAP_BYTES`] bytes (tail-preserving, with a diagnostic line on
+/// truncation). Temporary files keep an escaped descendant that inherited stdout/stderr from
+/// blocking cleanup after the direct child times out.
 fn output_with_timeout(command: &mut Command, timeout: Duration) -> anyhow::Result<Output> {
     let mut stdout = tempfile::tempfile()?;
     let mut stderr = tempfile::tempfile()?;
@@ -421,23 +427,27 @@ fn output_with_timeout(command: &mut Command, timeout: Duration) -> anyhow::Resu
                 libc::kill(-pid, libc::SIGKILL);
             }
             let _ = child.kill();
-            thread::spawn(move || {
-                let _ = child.wait();
-            });
+            reap_detached(child);
             anyhow::bail!("timed out after {:.1}s", timeout.as_secs_f64());
         }
         thread::sleep(Duration::from_millis(10));
     };
-    stdout.rewind()?;
-    stderr.rewind()?;
-    let mut stdout_bytes = Vec::new();
-    let mut stderr_bytes = Vec::new();
-    stdout.read_to_end(&mut stdout_bytes)?;
-    stderr.read_to_end(&mut stderr_bytes)?;
+    let stdout_stream = read_bounded_tail(&mut stdout, CAPTURE_CAP_BYTES)?;
+    let stderr_stream = read_bounded_tail(&mut stderr, CAPTURE_CAP_BYTES)?;
+    let program = command.get_program().to_string_lossy();
+    for (stream, name) in [(&stdout_stream, "stdout"), (&stderr_stream, "stderr")] {
+        if stream.truncated() {
+            eprintln!(
+                "st2: truncated {name} capture of `{program}`: keeping last {} of {} bytes (cap {CAPTURE_CAP_BYTES})",
+                stream.bytes.len(),
+                stream.total,
+            );
+        }
+    }
     Ok(Output {
         status,
-        stdout: stdout_bytes,
-        stderr: stderr_bytes,
+        stdout: stdout_stream.bytes,
+        stderr: stderr_stream.bytes,
     })
 }
 

src/eval_run.rs

@@ -911,7 +911,7 @@ fn run_steps(
     catalog: &Path,
     top_env: &BTreeMap<String, String>,
 ) -> (Vec<JudgeResult>, BTreeMap<String, String>) {
-    use std::process::Command;
+    use std::process::{Command, Stdio};
     let mut results = Vec::new();
     // The env the JUDGES also get: $RUNS_DIR + each step's $RUN_<id>_EXIT (so a bash judge can read the
     // captures). Empty when there are no run steps.
@@ -947,7 +947,8 @@ fn run_steps(
             .unwrap_or((1, Duration::ZERO));
 
         let mut exit = -1;
-        let (mut out, mut err) = (Vec::new(), Vec::new());
+        let out_path = runs_dir.join(format!("{}.out", step.id));
+        let err_path = runs_dir.join(format!("{}.err", step.id));
         for attempt in 0..attempts {
             let mut cmd = Command::new("sh");
             cmd.arg("-c")
@@ -961,15 +962,22 @@ fn run_steps(
             for (k, v) in &runtime {
                 cmd.env(k, v);
             }
-            match cmd.output() {
-                Ok(o) => {
-                    exit = o.status.code().unwrap_or(-1);
-                    out = o.stdout;
-                    err = o.stderr;
-                }
+            // The child's stdout/stderr stream STRAIGHT into the capture files: a run step can emit
+            // arbitrarily much output and it must never be buffered wholesale in this process. Each
+            // attempt re-truncates, so the files hold the LAST attempt's output — the same bytes the
+            // old buffer-then-write path produced.
+            let attempt_exit = (|| -> std::io::Result<i32> {
+                let out_file = std::fs::File::create(&out_path)?;
+                let err_file = std::fs::File::create(&err_path)?;
+                cmd.stdout(Stdio::from(out_file))
+                    .stderr(Stdio::from(err_file));
+                Ok(cmd.status()?.code().unwrap_or(-1))
+            })();
+            match attempt_exit {
+                Ok(code) => exit = code,
                 Err(e) => {
                     exit = -1;
-                    err = format!("run step spawn failed: {e}").into_bytes();
+                    let _ = std::fs::write(&err_path, format!("run step spawn failed: {e}"));
                 }
             }
             if exit == 0 {
@@ -980,13 +988,19 @@ fn run_steps(
             }
         }
 
-        let _ = std::fs::write(runs_dir.join(format!("{}.out", step.id)), &out);
-        let _ = std::fs::write(runs_dir.join(format!("{}.err", step.id)), &err);
-        let _ = std::fs::write(runs_dir.join(format!("{}.exit", step.id)), exit.to_string());
-        // Also a unified, judge-greppable combined log (stdout then stderr) named after the run label.
-        let mut combined = out.clone();
-        combined.extend_from_slice(&err);
-        let _ = std::fs::write(logs_dir.join(format!("{}.log", step.id)), &combined);
+        let _ =
+            std::fs::write(runs_dir.join(format!("{}.exit", step.id)), exit.to_string());
+        // Also a unified, judge-greppable combined log (stdout then stderr), copied from the two
+        // capture files so neither ever has to fit in memory.
+        let combined_path = logs_dir.join(format!("{}.log", step.id));
+        let _ = (|| -> std::io::Result<()> {
+            let mut log = std::fs::File::create(&combined_path)?;
+            let mut out = std::fs::File::open(&out_path)?;
+            std::io::copy(&mut out, &mut log)?;
+            let mut err = std::fs::File::open(&err_path)?;
+            std::io::copy(&mut err, &mut log)?;
+            Ok(())
+        })();
         runtime.insert(format!("RUN_{}_EXIT", env_key(&step.id)), exit.to_string());
         eval_log!(
             "== run step {} → exit {}{} ==",
@@ -1020,22 +1034,32 @@ fn run_steps(
 /// continuous plain-text log, so this is the scrollback captured at judge time — enough to inspect a
 /// wedged/finished agent's history. A truly continuous agent log would need a `pty` feature.
 fn dump_agent_logs(pty_task_ids: &[String], catalog: &Path) {
+    use std::process::Stdio;
     if pty_task_ids.is_empty() {
         return;
     }
     let logs_dir = catalog.join("logs");
     let _ = std::fs::create_dir_all(&logs_dir);
     let pty_root = crate::run::effective_pty_root(catalog);
     for task_id in pty_task_ids {
-        let out = std::process::Command::new("pty")
-            .args(["peek", "--full", "--plain", task_id])
-            .env("PTY_ROOT", &pty_root)
-            .output();
-        if let Ok(o) = out
-            && o.status.success()
-            && !o.stdout.is_empty()
-        {
-            let _ = std::fs::write(logs_dir.join(format!("{task_id}.log")), &o.stdout);
+        let log_path = logs_dir.join(format!("{task_id}.log"));
+        // Stream the peek's stdout straight into the log file: the full scrollback can be large and
+        // must never be buffered wholesale in this process. The old contract — keep the log only
+        // when the peek succeeded AND produced output — is kept by deleting the (possibly empty)
+        // file otherwise.
+        let dumped = (|| -> std::io::Result<bool> {
+            let mut child = std::process::Command::new("pty")
+                .args(["peek", "--full", "--plain", task_id])
+                .env("PTY_ROOT", &pty_root)
+                .stdout(Stdio::from(std::fs::File::create(&log_path)?))
+                .stderr(Stdio::null())
+                .spawn()?;
+            child.wait().map(|s| s.success())
+        })();
+        let keep =
+            dumped.unwrap_or(false) && std::fs::metadata(&log_path).is_ok_and(|m| m.len() > 0);
+        if !keep {
+            let _ = std::fs::remove_file(&log_path);
         }
     }
 }
@@ -1552,8 +1576,11 @@ fn run_bash_judge(
         .env("CATALOG", catalog)
         .env("ST_ROOT", bus)
         .env("SPEC_DIR", &physical_spec_dir)
-        .stdout(Stdio::piped())
-        .stderr(Stdio::piped());
+        // Only the status/exit detail is consumed. Piping stdout/stderr and never draining them
+        // deadlocks a chatty judge: once its pipe buffers fill, the child blocks forever while the
+        // parent keeps polling try_wait.
+        .stdout(Stdio::null())
+        .stderr(Stdio::null());
     // $RUNS_DIR + each $RUN_<id>_EXIT, so a judge can read the run steps' captured stdout/stderr/exit.
     for (k, v) in run_env {
         command.env(k, v);
@@ -2930,6 +2957,47 @@ agent "worker" { identity "worker"; host "evalhost"; argv "true" }
         assert!(!run_declarative(&missing, cat.path()).0);
     }
 
+    #[test]
+    fn run_steps_streams_large_step_output_to_files_without_corruption() {
+        let catalog = tempfile::tempdir().unwrap();
+        let cat = catalog.path();
+        // ~64 KiB each of stdout and stderr — far beyond any pipe buffer. This round-trips byte
+        // for byte only if the child streams straight into the capture files, and the combined log
+        // keeps the stdout-then-stderr ordering.
+        let marker_out = "O".repeat(64 * 1024);
+        let marker_err = "E".repeat(64 * 1024);
+        let step = RunStep {
+            id: "big".into(),
+            workspace: None,
+            command: "printf '%s' \"$BIG_OUT\"; printf '%s' \"$BIG_ERR\" >&2".into(),
+            env: BTreeMap::from([
+                ("BIG_OUT".to_string(), marker_out.clone()),
+                ("BIG_ERR".to_string(), marker_err.clone()),
+            ]),
+            unset: vec![],
+            retry: None,
+            allow_nonzero: false,
+        };
+        let (results, judge_env) = run_steps(&[step], cat, &BTreeMap::new());
+        assert_eq!(results.len(), 1);
+        assert!(results[0].passed);
+        assert_eq!(judge_env.get("RUN_big_EXIT").map(String::as_str), Some("0"));
+        let runs = cat.join(".runs");
+        assert_eq!(
+            std::fs::read(runs.join("big.out")).unwrap(),
+            marker_out.as_bytes()
+        );
+        assert_eq!(
+            std::fs::read(runs.join("big.err")).unwrap(),
+            marker_err.as_bytes()
+        );
+        assert_eq!(std::fs::read_to_string(runs.join("big.exit")).unwrap(), "0");
+        // Combined log = stdout fully followed by stderr.
+        let mut expected = marker_out.into_bytes();
+        expected.extend_from_slice(marker_err.as_bytes());
+        assert_eq!(std::fs::read(cat.join("logs/big.log")).unwrap(), expected);
+    }
+
     #[test]
     fn bash_judge_exit_code_and_timeout() {
         let spec = tempfile::tempdir().unwrap();