brokerd: start pipe threads safely and collect output within a grace period

Implemented-By: OpenCode session (model recorded in docs/implementer-log.md)

crates/brokerd/src/container.rs

@@ -11,7 +11,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::{Duration, Instant};
 
 use crate::config::Runner;
-use crate::pipes::Io;
+use crate::pipes::{GRACE, Io};
 use crate::podman;
 use crate::runner::{RunError, RunOutput, RunSpec, Runtime};
 
@@ -20,6 +20,7 @@ pub const COULD_NOT_RUN: &str = "the tool could not run";
 pub const CANNOT_START: &str = "the tool runner could not start the container";
 pub const KILLED: &str = "the tool was stopped: it ran out of memory or was killed";
 pub const TIMED_OUT: &str = "the tool ran past its time limit";
+pub const OUTPUT_OPEN: &str = "the tool left its output open";
 pub const UNEXPECTED: &str = "the tool failed with an unexpected status";
 /// How often a running container is checked.
 pub const POLL: Duration = Duration::from_millis(50);
@@ -144,10 +145,29 @@ impl Podman {
         };
         let started = Instant::now();
         let cap = usize::try_from(self.runner.output_cap).unwrap_or(usize::MAX);
-        let io = Io::start(&mut child, input, cap);
+        let io = match Io::start(&mut child, input, cap, STDERR_KEPT) {
+            Ok(io) => io,
+            Err(e) => {
+                // The container may be running: stop it before answering.
+                self.podman(&["kill", name]);
+                self.podman(&["rm", "-f", name]);
+                let _ = child.kill();
+                let _ = child.wait();
+                (self.log)(&format!(
+                    "brokerd: cannot start a thread for {name}: {e}\n{RUNBOOK}"
+                ));
+                return Err(RunError::Unavailable(CANNOT_START.to_string()));
+            }
+        };
         let status = self.wait(&mut child, name, started, limit);
-        let (out, truncated, err) = io.finish();
-        self.answer(name, status, out, truncated, &err)
+        let done = io.finish(GRACE);
+        if done.open && status.is_some() {
+            (self.log)(&format!(
+                "brokerd: {name} ended but something still holds its output; it was abandoned"
+            ));
+            return Err(RunError::Failed(OUTPUT_OPEN.to_string()));
+        }
+        self.answer(name, status, done.out, done.truncated, &done.err)
     }
 
     /// Step 3: spawn `podman` with `args` and all three standard streams piped. On failure, log

crates/brokerd/src/pipes.rs

@@ -1,65 +1,107 @@
 //! The three pipes of one container: its arguments go in on standard input, its output and errors
-//! come back, each on its own thread so no pipe can block another.
+//! come back, each on its own thread so no pipe can block another. The threads start with
+//! `Builder`, which returns an error instead of panicking, and the output is collected within a
+//! grace period so a pipe held open by something outside the container cannot hold `brokerd` (M3b
+//! review findings 3 and 4).
 
 use std::io::{Read, Write};
 use std::process::Child;
-use std::thread::JoinHandle;
+use std::sync::mpsc::{self, Receiver, RecvTimeoutError};
+use std::thread::Builder;
+use std::time::{Duration, Instant};
+
+/// How long, after the container has ended, its output may take to reach its end.
+pub const GRACE: Duration = Duration::from_secs(2);
+
+/// What came back.
+pub(crate) struct Finished {
+    pub out: Vec<u8>,
+    pub truncated: bool,
+    pub err: String,
+    /// A pipe had not ended by the grace deadline: something outside the container holds it.
+    pub open: bool,
+}
 
-/// Step 4: the three threads that feed and read one container, so no pipe can block another.
 pub(crate) struct Io {
-    writer: Option<JoinHandle<()>>,
-    stdout: Option<JoinHandle<(Vec<u8>, bool)>>,
-    stderr: Option<JoinHandle<(Vec<u8>, bool)>>,
+    stdout: Option<Receiver<(Vec<u8>, bool)>>,
+    stderr: Option<Receiver<(Vec<u8>, bool)>>,
 }
 
 impl Io {
     /// Take `child`'s three pipes (`child.stdin.take()` and so on) and start one thread for each:
-    /// write `input` to standard input and then drop it; `read_capped` standard output with `cap`;
-    /// `read_capped` standard error with `STDERR_KEPT`. A pipe that is `None` gets no thread.
-    pub(crate) fn start(child: &mut Child, input: String, cap: usize) -> Io {
-        let writer = child.stdin.take().map(|mut stdin| {
-            std::thread::spawn(move || {
+    /// write `input` to standard input and then drop it (its handle is not kept, it is never
+    /// joined); `read_capped` standard output with `cap`; `read_capped` standard error with
+    /// `err_cap`. The reader threads are detached, keeping only the channel receiver. A pipe that
+    /// is `None` gets no thread; a thread that cannot be started is the returned error.
+    pub(crate) fn start(
+        child: &mut Child,
+        input: String,
+        cap: usize,
+        err_cap: usize,
+    ) -> std::io::Result<Io> {
+        if let Some(mut stdin) = child.stdin.take() {
+            Builder::new().spawn(move || {
                 let _ = stdin.write_all(input.as_bytes());
-            })
-        });
-        let stdout = child
-            .stdout
-            .take()
-            .map(|stdout| std::thread::spawn(move || read_capped(stdout, cap)));
-        let stderr = child.stderr.take().map(|stderr| {
-            std::thread::spawn(move || read_capped(stderr, crate::container::STDERR_KEPT))
-        });
-        Io {
-            writer,
-            stdout,
-            stderr,
+            })?;
+        }
+        let stdout = if let Some(stdout) = child.stdout.take() {
+            Some(reader(stdout, cap)?)
+        } else {
+            None
+        };
+        let stderr = if let Some(stderr) = child.stderr.take() {
+            Some(reader(stderr, err_cap)?)
+        } else {
+            None
+        };
+        Ok(Io { stdout, stderr })
+    }
+
+    /// Collect the output within `grace`: one deadline for both readers, after which a pipe that has
+    /// not ended counts as open (something outside the container holds it) and is abandoned.
+    pub(crate) fn finish(self, grace: Duration) -> Finished {
+        let until = Instant::now() + grace;
+        let (out, truncated, stdout_open) = collect(self.stdout, until);
+        let (err_bytes, _, stderr_open) = collect(self.stderr, until);
+        let err = String::from_utf8_lossy(&err_bytes).into_owned();
+        Finished {
+            out,
+            truncated,
+            err,
+            open: stdout_open || stderr_open,
+        }
     }
 }
 
-    /// Step 6: join the three threads. A thread that is missing or panicked counts as empty
-    /// (`join().ok()`, `unwrap_or_default()`). Returns the kept standard output, whether there was
-    /// more, and the kept standard error decoded with `from_utf8_lossy`.
-    pub(crate) fn finish(self) -> (Vec<u8>, bool, String) {
-        if let Some(handle) = self.writer {
-            let _ = handle.join();
+/// One detached reader: `read_capped` `pipe` and hand the result back on a channel. The handle is
+/// kept only long enough to start the thread; a spawn error is returned, a send error ignored.
+fn reader(
+    pipe: impl Read + Send + 'static,
+    cap: usize,
+) -> std::io::Result<Receiver<(Vec<u8>, bool)>> {
+    let (tx, rx) = mpsc::channel();
+    Builder::new().spawn(move || {
+        let _ = tx.send(read_capped(pipe, cap));
+    })?;
+    Ok(rx)
 }
-        let (out, truncated) = match self.stdout {
-            Some(handle) => handle.join().ok().unwrap_or_default(),
-            None => (Vec::new(), false),
-        };
-        let err = match self.stderr {
-            Some(handle) => {
-                let (bytes, _) = handle.join().ok().unwrap_or_default();
-                String::from_utf8_lossy(&bytes).into_owned()
-            }
-            None => String::new(),
-        };
-        (out, truncated, err)
+
+/// Read from `rx` until `until`, classifying the outcome: `None` (no pipe) is empty and closed, a
+/// timeout is empty and open, and a disconnected channel (the reader panicked) is empty and closed.
+fn collect(rx: Option<Receiver<(Vec<u8>, bool)>>, until: Instant) -> (Vec<u8>, bool, bool) {
+    match rx {
+        None => (Vec::new(), false, false),
+        Some(rx) => match rx.recv_timeout(until.saturating_duration_since(Instant::now())) {
+            Ok((bytes, truncated)) => (bytes, truncated, false),
+            Err(RecvTimeoutError::Timeout) => (Vec::new(), false, true),
+            Err(RecvTimeoutError::Disconnected) => (Vec::new(), false, false),
+        },
     }
 }
 
 /// Everything `from` gives, keeping the first `cap` bytes; `true` if there was more. Reads on past
-/// the cap, so the writer is never blocked or broken by a closed pipe. No indexing: `buf.get(..n)`.
+/// the cap, so the writer is never blocked or broken by a closed pipe. No indexing and no
+/// subtraction that could wrap.
 pub(crate) fn read_capped(from: impl Read, cap: usize) -> (Vec<u8>, bool) {
     let mut from = from;
     let mut kept: Vec<u8> = Vec::new();
@@ -71,13 +113,13 @@ pub(crate) fn read_capped(from: impl Read, cap: usize) -> (Vec<u8>, bool) {
             Ok(n) => n,
             Err(_) => break,
         };
-        let remaining = cap - kept.len();
+        let remaining = cap.saturating_sub(kept.len());
         if remaining == 0 {
             truncated = true;
         } else {
-            let take = remaining.min(n);
-            kept.extend_from_slice(&chunk[..take]);
-            if n > take {
+            let take = n.min(remaining);
+            kept.extend_from_slice(chunk.get(..take).unwrap_or_default());
+            if take < n {
                 truncated = true;
             }
         }

crates/brokerd/tests/container_grace.rs

@@ -0,0 +1,86 @@
+//! After a container ends, its output is collected within a grace period, never waited on for
+//! ever: something outside the container that still holds a pipe must not hold `brokerd` (M3b
+//! review finding 4). Against a fake `podman` whose shell leaves a background `sleep` holding the
+//! pipes, which real Podman does not do. Do not edit.
+
+#[path = "support/build.rs"]
+mod build;
+#[path = "support/fake_podman.rs"]
+mod fake_podman;
+
+use std::time::{Duration, Instant};
+
+use brokerd::container::{OUTPUT_OPEN, Podman, TIMED_OUT};
+use brokerd::pipes::GRACE;
+use brokerd::policy::{Outcome, SessionState, decide};
+use brokerd::runner::run;
+use build::{grant, now, read, set};
+use fake_podman::{Fake, Lines, serial};
+use proto::{Mode, ToolResponse};
+
+fn call(fake: &Fake, extra: &str, log: &Lines) -> (ToolResponse, Duration) {
+    let podman = Podman::new(fake.runner(extra), fake.dir.join("egress"), log.sink());
+    let grants = set(vec![grant("notes", "read_file", Mode::Auto).paths(&["/n"])]);
+    let decision = match decide(read("/n/a"), &grants, SessionState::default(), now()) {
+        Outcome::Allowed(d) => d,
+        other => panic!("not allowed: {other:?}"),
+    };
+    let started = Instant::now();
+    let got = run(decision, &podman);
+    (got, started.elapsed())
+}
+
+#[test]
+fn the_grace_period_is_two_seconds() {
+    assert_eq!(GRACE, Duration::from_secs(2));
+}
+
+#[test]
+fn output_held_open_after_the_container_ended_is_abandoned_after_the_grace_period() {
+    let _s = serial();
+    // The shell exits at once; the background sleep keeps standard output and error open.
+    let fake = Fake::new("grace-open", "cat > /dev/null; printf ok; sleep 6 & exit 0");
+    let log = Lines::default();
+    let (got, took) = call(&fake, "", &log);
+    assert_eq!(
+        got,
+        ToolResponse::Failed {
+            message: OUTPUT_OPEN.to_string()
+        }
+    );
+    assert!(took >= GRACE, "{took:?}");
+    assert!(took < GRACE + Duration::from_secs(2), "{took:?}");
+    assert!(log.all().contains("abandoned"), "{}", log.all());
+}
+
+#[test]
+fn a_tool_past_its_limit_is_answered_within_the_grace_period_even_if_its_pipes_stay_open() {
+    let _s = serial();
+    // No `exec`: killing the shell leaves the sleep holding the pipes.
+    let fake = Fake::new("grace-slow", "cat > /dev/null; sleep 6");
+    let log = Lines::default();
+    let (got, took) = call(&fake, "read_file_ms = 300", &log);
+    assert_eq!(
+        got,
+        ToolResponse::Failed {
+            message: TIMED_OUT.to_string()
+        }
+    );
+    assert!(
+        took < Duration::from_millis(300) + GRACE + Duration::from_secs(2),
+        "{took:?}"
+    );
+}
+
+#[test]
+fn a_tool_that_ends_normally_is_not_slowed_by_the_grace_period() {
+    let _s = serial();
+    let fake = Fake::new("grace-ok", "cat > /dev/null; printf done; exit 0");
+    let log = Lines::default();
+    let (got, took) = call(&fake, "", &log);
+    assert!(
+        matches!(&got, ToolResponse::Result { content, .. } if content == "done"),
+        "{got:?}"
+    );
+    assert!(took < Duration::from_secs(1), "{took:?}");
+}

docs/implementer-log.md

@@ -6,6 +6,7 @@ reviewer adds findings under "Reviews" once per milestone.
 
 | Task | Date | Status | Gate runs | First gate | Deviations | Notes | Model |
 |---|---|---|---|---|---|---|---|
+| M3b/15-brokerd-pipes-grace | 2026-09-23 | done | 1 | pass | none | Fixed the two M3b review findings in `pipes.rs`. Added `pub const GRACE: Duration = Duration::from_secs(2)`; replaced the `JoinHandle`-holding `Io` with one holding `Option<Receiver<(Vec<u8>, bool)>>` for stdout/stderr and added the `Finished { out, truncated, err, open }` struct. `Io::start` now returns `std::io::Result<Io>`: the stdin writer is started with `Builder::new().spawn(...)` and its handle dropped (never joined, a spawn error returned with `?`); a new private `reader(pipe, cap) -> io::Result<Receiver<...>>` starts one detached reader per pipe and returns a spawn error, used for both stdout (`cap`) and stderr (`err_cap`). `Io::finish(grace)` sets one `until = Instant::now() + grace` and calls a private `collect(rx, until)` per receiver: `recv_timeout(until.saturating_duration_since(now))` → `Ok` keeps bytes, `Timeout` sets `open`, `Disconnected` (panicked reader) counts closed; `open` is `stdout_open || stderr_open`. `read_capped` unchanged in behavior, now `cap.saturating_sub(kept.len())` and `chunk.get(..take).unwrap_or_default()` (no indexing, no wrapping subtraction). `container.rs`: added `pub const OUTPUT_OPEN` after `TIMED_OUT`, `use crate::pipes::{GRACE, Io}`, and replaced the `run_container` tail with the task's `Io::start` match (on spawn error, `podman kill`/`rm -f`, `child.kill()`/`wait()`, log + `Unavailable(CANNOT_START)`), then `io.finish(GRACE)` and the `done.open && status.is_some()` check returning `Failed(OUTPUT_OPEN)` (a timed-out call, `status` None, still answers `TIMED_OUT`). All `std::thread::spawn` gone from `crates/brokerd/src/`. Copied `tests/container_grace.rs`; 4 passed in ~4.4s, container 11, container_egress 6, all five runs; `make gate` prints `gate: ok` first run. | ? |
 | M3b/14-brokerd-pipes-module | 2026-09-23 | done | 1 | pass | none | Pure move: cut `struct Io`, `impl Io { start, finish }`, and `fn read_capped` from `container.rs` and pasted them into `crates/brokerd/src/pipes.rs` with bodies unchanged and `pub(crate)` visibility; `Io::start` now calls `crate::container::STDERR_KEPT` (the constant stays in `container.rs` since the tests import it). Added the module doc comment and the `use std::io::{Read, Write}`, `use std::process::Child`, `use std::thread::JoinHandle` lines. `lib.rs` gained `pub mod pipes;` between `ledger` and `podman`. `container.rs` gained `use crate::pipes::Io;` and lost the `Read, Write` and `JoinHandle` imports the compiler reported unused; nothing else changed. `cargo fmt --all` reflowed the `STDERR_KEPT` call line in `start` to wrap. `cargo check`/clippy clean; container 11, container_egress 6, serve_runner 2 pass; `grep "struct Io\|fn read_capped" container.rs` prints nothing and `container.rs` is 350 lines. `make gate` prints `gate: ok` on the first run. | ? |
 | M3b/13-brokerd-serve-runner | 2026-09-23 | done | 1 | pass | none | The `Podman` runtime was already written in task 12 (`crates/brokerd/src/container.rs`), so this task only wired it into `serve`. In `main.rs`: after the config loads, added `let log: Arc<dyn Fn(&str) + Send + Sync> = Arc::new(|line| eprintln!("{line}"))`; chose the runtime and its notice with a match on `&cfg.runner` — `Some(runner)` captures `runner.image` first then `Box::new(Podman::new(runner.clone(), cfg.egress_dir(), Arc::clone(&log)))` with notice `tools run in containers from {image}`; `None` → `Box::new(Refusing)` with notice `no [runner] section: every tool call is refused`. Passed `runtime` and `log` to `serve::start` (its signature needed no change — it already took `Box<dyn Runtime>` and `Arc<dyn Fn(&str)+Send+Sync>`), and printed the notice with `eprintln!` after the `serving tools on … and approvals on …` line. Added `use brokerd::container::Podman` and `Runtime` to the imports. Copied `tests/serve_runner.rs`; the 2 new tests plus the 9 serve tests pass; `make gate` prints `gate: ok` on the first run. | ? |
 | M3b/12-brokerd-egress | 2026-09-23 | done | 1 | pass | none | Wrote `crates/brokerd/src/container.rs` (432 lines). Added consts `EGRESS_WAIT` (5s) and `EGRESS_POLL` (20ms); `Podman` gained private `egress_wait` (set to `EGRESS_WAIT` by `new`) and the public `with_egress_wait`. Added private `cannot_launch(podman, e)` (the task-11 step 3 log `brokerd: cannot start {path}: {e}\n{RUNBOOK}`), which `spawn` now calls, and `cannot_make(dir, e)` returning `Err(Unavailable(CANNOT_START))` with `brokerd: cannot make {dir}: {e}\n{RUNBOOK}`. `Runtime::run` now branches on `spec.egress()`: `None` runs `tool_args(spec, &runner, &name, None)` as before; `Some(hosts)` computes `dir = egress_dir.join(&name)`, calls `start_egress`, then `tool_args(spec, &runner, &name, Some(&dir))`. `start_egress` creates the `EgressGuard` first (so every return cleans up), then step 1: `DirBuilder::new().recursive(true).mode(0o700).create(egress_dir)`, `set_permissions(egress_dir, 0o700)` anyway, `remove_dir_all(dir)` if it exists (NotFound ok), `DirBuilder::new().mode(0o700).create(dir)` non-recursively, each failure via `cannot_make`; step 2: `Command::new(&runner.podman).args(egress_args).stdin(null).stdout(null).stderr(piped()).output()`, non-success logs `brokerd: podman could not start {name}-egress: {stderr}\n{RUNBOOK}` and returns Unavailable, launch error uses `cannot_launch`; step 3: wait for `dir.join("egress.sock")` every 20 ms until `egress_wait`, then log `brokerd: {name}-egress did not make its socket within {ms} ms\n{RUNBOOK}` and return Unavailable; step 4 `Ok(_guard)`. `EgressGuard<'a>` holds `&'a Podman`, the container name `<name>-egress` and the directory; its `Drop` runs `podman rm -f <name>-egress` (the task-11 helper) then `remove_dir_all(dir)` (NotFound ok, else log `brokerd: cannot remove {dir}: {e}`). Added `use std::os::unix::fs::{DirBuilderExt, PermissionsExt};`. `cargo fmt` put the new `impl Podman` block after `impl Runtime`. 6 egress + 11 container tests pass ten runs in a row. First gate failed on `start_egress` being inside `impl Runtime` (not a trait member) and missing `DirBuilderExt`/`PermissionsExt`; then on the guard being created at the end instead of the start (the two "nothing left" tests need the `rm -f <name>-egress` call on a failed proxy), fixed by moving `EgressGuard::new` to the top and returning it; then on `unused_variable` for the drop guard and `mismatched_lifetime_syntaxes` on the return type (fixed to `EgressGuard<'_>`), the guard renamed `_guard` (1.98 still lints drop-only bindings). `make gate` prints `gate: ok`. | ? |