M3b plan: follow-up tasks 14 to 17 for the review's lower findings

14 moves the pipe handling out of container.rs (a pure move, replayed on its
own); 15 starts threads with Builder and bounds output collection with a 2 s
grace period; 16 escapes container errors in the log and fixes two texts; 17
fixes toolkit's thread start, casts and the egress-proxy form. Each checked
against a reference, which is not kept. Tips T24 to T26 from this run.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

docs/plans/M3b/15-brokerd-pipes-grace.md

@@ -0,0 +1,152 @@
+# M3b task 15: no thread panics, and a grace period for the output
+
+**Branch:** `m3b` (run `git switch m3b`; `git status --short` must be empty, otherwise stop)
+**Commit subject:** `brokerd: start pipe threads safely and collect output within a grace period`
+
+## Goal
+
+Two defects from the M3b review, both in `pipes.rs` (task 14):
+
+- **Finding 3.** `std::thread::spawn` panics when the system cannot make a thread. The panic drops
+  the `Child` without stopping the container, which then runs on without its time limit. Use
+  `std::thread::Builder`, which returns an error instead, and stop the container on that error.
+- **Finding 4.** `Io::finish` joins the readers with no limit. Each reader stops only at its pipe's
+  end, so if any other process still holds a pipe, `brokerd` waits for it: the call's time limit
+  does not bound it. After the container has ended, give the output a **grace period** of 2 s,
+  then abandon what has not ended.
+
+## Files
+
+- Copy: `crates/brokerd/tests/container_grace.rs`
+- Modify: `crates/brokerd/src/pipes.rs`, `crates/brokerd/src/container.rs`,
+  `docs/implementer-log.md`
+
+## `pipes.rs`, new shape
+
+```rust
+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,
+}
+
+pub(crate) struct Io {
+    stdout: Option<Receiver<(Vec<u8>, bool)>>,
+    stderr: Option<Receiver<(Vec<u8>, bool)>>,
+}
+
+impl Io {
+    pub(crate) fn start(child: &mut Child, input: String, cap: usize, err_cap: usize)
+        -> std::io::Result<Io>;
+    pub(crate) fn finish(self, grace: Duration) -> Finished;
+}
+
+pub(crate) fn read_capped(from: impl Read, cap: usize) -> (Vec<u8>, bool);  // as now
+```
+
+Write each function in turn, running `cargo check -p brokerd` after each.
+
+### `Io::start`
+
+1. If `child.stdin.take()` is `Some(stdin)`: start a thread with `Builder::new().spawn(…)` that
+   writes `input` with `write_all` (ignore its error) and then drops `stdin`. **Do not keep its
+   handle and never join it**: it ends when the input is written or the pipe breaks. A spawn error
+   → return it (`?`).
+2. For standard output with `cap`, and then standard error with `err_cap`: if the pipe is
+   `Some`, make a channel (`mpsc::channel()`), start a thread with `Builder::new().spawn(…)` that
+   runs `read_capped(pipe, cap)` and `send`s the result (ignore a send error), and keep the
+   `Receiver`. A spawn error → return it. Write a small private function
+   `fn reader(pipe: impl Read + Send + 'static, cap: usize) -> std::io::Result<Receiver<(Vec<u8>, bool)>>`
+   and use it for both.
+3. `Ok(Io { stdout, stderr })`.
+
+### `Io::finish(grace)`
+
+One deadline for both readers: `let until = Instant::now() + grace;`. For each receiver (standard
+output first), a small private function `collect(rx, until) -> (Vec<u8>, bool, bool)`:
+
+- `None` (no pipe) → `(empty, false, false)`.
+- `rx.recv_timeout(until.saturating_duration_since(Instant::now()))`:
+  - `Ok((bytes, truncated))` → `(bytes, truncated, false)`;
+  - `Err(RecvTimeoutError::Timeout)` → `(empty, false, true)`: **open**;
+  - `Err(RecvTimeoutError::Disconnected)` (the reader panicked) → `(empty, false, false)`.
+
+Return `Finished { out, truncated, err: String::from_utf8_lossy(&err_bytes).into_owned(),
+open: stdout_open || stderr_open }`.
+
+### `read_capped`
+
+Unchanged in what it does, but no indexing and no subtraction that could wrap: keep
+`let take = n.min(cap.saturating_sub(kept.len()));`, copy with
+`kept.extend_from_slice(chunk.get(..take).unwrap_or_default());`, and set `truncated` when
+`take < n`. Retry `ErrorKind::Interrupted`; stop on any other error.
+
+## `container.rs`
+
+1. A new constant after `TIMED_OUT`:
+   `pub const OUTPUT_OPEN: &str = "the tool left its output open";`
+2. `use crate::pipes::{GRACE, Io};`
+3. In `run_container`, replace the lines from `let io = Io::start(…)` to the end of the function
+   with exactly this:
+
+```rust
+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 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)
+```
+
+When the time limit was passed (`status` is `None`), an open pipe changes nothing: the answer is
+still `TIMED_OUT`, from `answer`.
+
+Also: `grep -n "thread::spawn" crates/brokerd/src/` must print nothing when you are done.
+
+## Steps
+
+- [ ] **1. Copy.** `git switch m3b`, then
+  `cp docs/plans/M3b/files/crates/brokerd/tests/container_grace.rs crates/brokerd/tests/`
+- [ ] **2. See it fail.** `cargo test -p brokerd --test container_grace`. Expected: it does not
+  compile (`OUTPUT_OPEN` and `pipes::GRACE` do not exist).
+- [ ] **3. Write `pipes.rs`, one function at a time, then `container.rs`.** Run `cargo fmt --all`.
+- [ ] **4. See it pass.** `cargo test -p brokerd --test container_grace`. Expected: 4 passed, in
+  about 4 s (two tests wait out the grace period on purpose). Then
+  `cargo test -p brokerd --test container --test container_egress`: 11 and 6 passed. Run all three
+  five times.
+- [ ] **5. Run the gate.** `make gate`. Expected last line: `gate: ok`.
+- [ ] **6. Log and commit.** `git add crates/brokerd docs/implementer-log.md Cargo.lock && git commit`
+
+## Done when
+
+- The three suites pass five times running; no `thread::spawn` in `crates/brokerd/src/`; `make
+  gate` prints `gate: ok`.
+
+## Stop and report if
+
+- A test takes much longer than stated, or passes only sometimes.