From d01aaa781888195aaa3d22c890f3cdcdcbb9ef64 Mon Sep 17 00:00:00 2001 From: Matt Toohey Date: Wed, 19 Aug 2026 15:31:28 +1000 Subject: [PATCH 1/5] feat(doctor): add opt-in piped stdin for streaming fix execution MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fix commands run through the streaming executor always inherited the host process's stdin. In a GUI host that stdin is never writable, so an interactive fix like `claude-agent-acp --cli auth login` — which prints an OAuth URL and then blocks reading the auth code — hangs forever (block/berd#99). Add an opt-in pipe the host can feed: - New public types `FixStdin` / `FixStdinWriter`: `FixStdin::pipe()` returns a cloneable line writer (`send_line`, which appends `\n` and flushes) plus the `FixStdin` to place in the options. Lines sent before spawn are buffered; dropping every writer delivers EOF. - `ExecuteFixOptions` gains `pub stdin: Option` and a `with_stdin` builder; the existing `Debug`/`Clone`/`Default` derives are preserved via an `Arc>>` around the non-cloneable channel receiver. - The option threads through `execute_fix_streaming_with_env_options` → `run_command_streaming` → `run_command_streaming_blocking`, which only then sets `Stdio::piped()` on stdin and feeds the child from a detached writer thread. The thread is deliberately never joined — a writer outliving the child would park it in `rx.iter()` and hang the fix; it exits on channel close or on the post-exit EPIPE write error. - `stdin: None` keeps today's inherited-stdin behavior byte-for-byte, so terminal hosts with legitimately interactive fixes are untouched. Tests: cat echo round-trip (write + EOF + pre-spawn buffering), `read -r` prompt shape (the paste-an-auth-code flow), and no-hang-when-writer-outlives-child including post-exit `send_line` erroring instead of panicking. Existing streaming tests cover the default path. Verified with `cargo test` in crates/doctor (118 passed) plus `cargo fmt` and `cargo clippy --all-targets`. Signed-off-by: Matt Toohey --- apps/staged/src-tauri/src/doctor.rs | 3 + crates/doctor/src/lib.rs | 194 +++++++++++++++++++++++++++- 2 files changed, 194 insertions(+), 3 deletions(-) diff --git a/apps/staged/src-tauri/src/doctor.rs b/apps/staged/src-tauri/src/doctor.rs index 6feae12ca..84b79afb6 100644 --- a/apps/staged/src-tauri/src/doctor.rs +++ b/apps/staged/src-tauri/src/doctor.rs @@ -59,6 +59,9 @@ fn execute_fix_options( command_override, npm_registry: crate::managed_acp_tools::npm_registry().map(str::to_string), env: None, + // Staged's fixes are non-interactive: nothing here feeds a prompt, so + // the child keeps inheriting stdin rather than getting a piped one. + stdin: None, } .with_env_snapshot(env_vars) } diff --git a/crates/doctor/src/lib.rs b/crates/doctor/src/lib.rs index abdd2325f..94a2d7683 100644 --- a/crates/doctor/src/lib.rs +++ b/crates/doctor/src/lib.rs @@ -673,6 +673,55 @@ struct FreshnessTarget { version_args: Option<&'static [&'static str]>, } +/// Opt-in piped stdin for a fix subprocess. Create with [`FixStdin::pipe`]; +/// keep the [`FixStdinWriter`], put the `FixStdin` in +/// [`ExecuteFixOptions::stdin`]. Cloning shares the underlying receiver: the +/// first execution to spawn takes it, so a cloned options struct cannot feed +/// two children. +#[derive(Debug, Clone)] +pub struct FixStdin { + rx: Arc>>>, +} + +impl FixStdin { + /// Create a connected pair: a cloneable writer for the caller to keep and + /// the `FixStdin` to place in [`ExecuteFixOptions::stdin`]. Lines sent + /// before the fix subprocess spawns are buffered and written once it does; + /// dropping every writer clone closes the child's stdin (EOF). + pub fn pipe() -> (FixStdinWriter, FixStdin) { + let (tx, rx) = std::sync::mpsc::channel(); + ( + FixStdinWriter { tx }, + FixStdin { + rx: Arc::new(Mutex::new(Some(rx))), + }, + ) + } + + /// Take the receiving end for a spawned child. First caller wins; `None` + /// on later calls (a clone already fed an execution). + fn take_receiver(&self) -> Option> { + self.rx.lock().ok().and_then(|mut rx| rx.take()) + } +} + +/// Cloneable handle for feeding lines to a fix subprocess's stdin. +#[derive(Debug, Clone)] +pub struct FixStdinWriter { + tx: std::sync::mpsc::Sender, +} + +impl FixStdinWriter { + /// Queue one line for the fix's stdin; a trailing `\n` is appended and the + /// pipe is flushed. `Err` when the fix has already finished (its stdin + /// pipe is closed). + pub fn send_line(&self, line: impl Into) -> Result<(), String> { + self.tx + .send(line.into()) + .map_err(|_| "Fix is no longer accepting input".to_string()) + } +} + /// Options for executing a doctor fix command. #[derive(Debug, Clone, Default)] pub struct ExecuteFixOptions { @@ -682,6 +731,10 @@ pub struct ExecuteFixOptions { pub npm_registry: Option, /// Optional caller-provided environment snapshot for the fix subprocess. pub env: Option, + /// Opt-in piped stdin for the fix subprocess (see [`FixStdin::pipe`]). + /// `None` keeps the child inheriting the host process's stdin, so + /// terminal hosts can still run interactive fixes directly. + pub stdin: Option, } impl ExecuteFixOptions { @@ -689,6 +742,11 @@ impl ExecuteFixOptions { self.env = Some(DoctorEnv::new(vars)); self } + + pub fn with_stdin(mut self, stdin: FixStdin) -> Self { + self.stdin = Some(stdin); + self + } } /// Run a fix command for a doctor check, identified by check ID and fix type. @@ -723,6 +781,7 @@ pub async fn execute_fix_with_options( command_override, npm_registry: npm_registry.map(str::to_string), env: None, + stdin: None, }, ) .await @@ -780,6 +839,7 @@ where command_override, npm_registry: npm_registry.map(str::to_string), env: None, + stdin: None, }, on_line, ) @@ -814,20 +874,21 @@ where // Fixes are intentionally not routed through the bounded probe runner: // these are user-triggered install/auth/update actions and can reasonably // be interactive or long-running. - run_command_streaming(command, opts.env, on_line).await + run_command_streaming(command, opts.env, opts.stdin, on_line).await } /// Async wrapper that runs `run_command_streaming_blocking` on the blocking pool. pub(crate) async fn run_command_streaming( command: String, env: Option, + stdin: Option, on_line: F, ) -> Result<(), String> where F: FnMut(&str) + Send + 'static, { tokio::task::spawn_blocking(move || { - run_command_streaming_blocking(&command, env.as_ref(), on_line) + run_command_streaming_blocking(&command, env.as_ref(), stdin, on_line) }) .await .unwrap_or_else(|e| Err(format!("Task failed: {e}"))) @@ -1020,22 +1081,57 @@ pub(crate) fn execute_command_with_path_prefix_with_env( fn run_command_streaming_blocking( command: &str, env: Option<&DoctorEnv>, + stdin: Option, mut on_line: F, ) -> Result<(), String> where F: FnMut(&str), { - use std::io::{BufRead, BufReader}; + use std::io::{BufRead, BufReader, Write}; let mut command = build_shell_command(command, &[], env); command .stdout(std::process::Stdio::piped()) .stderr(std::process::Stdio::piped()); + // Opt-in only: without a `FixStdin` the child keeps inheriting the host + // process's stdin, so interactive fixes in terminal hosts are untouched. + if stdin.is_some() { + command.stdin(std::process::Stdio::piped()); + } command::configure_command(&mut command); let mut child = command .spawn() .map_err(|e| format!("Failed to run command: {e}"))?; + if let Some(fix_stdin) = stdin { + let mut child_stdin = child.stdin.take().expect("stdin was piped"); + match fix_stdin.take_receiver() { + Some(stdin_rx) => { + // Detached on purpose: joining would hang the fix whenever a + // caller still holds a writer after the child exits (the + // thread would be parked in `iter()`). It exits on its own + // when every writer drops (channel closed) or a write fails + // once the child is gone (Rust ignores SIGPIPE, so EPIPE + // surfaces as a clean `Err`); dropping `child_stdin` then + // delivers EOF. + std::thread::spawn(move || { + for line in stdin_rx.iter() { + if child_stdin + .write_all(format!("{line}\n").as_bytes()) + .and_then(|()| child_stdin.flush()) + .is_err() + { + break; + } + } + }); + } + // A clone of this `FixStdin` already fed another execution; no + // line can ever arrive, so close the pipe now (immediate EOF). + None => drop(child_stdin), + } + } + let stdout = child.stdout.take().expect("stdout was piped"); let stderr = child.stderr.take().expect("stderr was piped"); @@ -1198,6 +1294,7 @@ mod tests { let result = run_command_streaming( "echo doctor-streaming-marker-hello && echo doctor-streaming-marker-world".to_string(), None, + None, move |line| { lines_clone.lock().unwrap().push(line.to_string()); }, @@ -1220,6 +1317,95 @@ mod tests { ); } + /// A line sent through the `FixStdin` pipe must reach the child's stdin + /// and dropping the last writer must deliver EOF: `cat` echoes the line + /// and exits 0 only when its stdin closes. Sending before the child + /// spawns also exercises the pre-spawn buffering guarantee. + #[tokio::test] + async fn run_command_streaming_piped_stdin_round_trips_through_cat() { + let lines: Arc>> = Arc::new(Mutex::new(Vec::new())); + let lines_clone = lines.clone(); + let (writer, stdin) = FixStdin::pipe(); + + writer.send_line("doctor-stdin-marker-echo").unwrap(); + drop(writer); + + let result = run_command_streaming("cat".to_string(), None, Some(stdin), move |line| { + lines_clone.lock().unwrap().push(line.to_string()); + }) + .await; + + assert!(result.is_ok(), "cat should exit 0 on EOF; got {result:?}"); + let captured = lines.lock().unwrap().clone(); + assert!( + captured.iter().any(|l| l == "doctor-stdin-marker-echo"), + "cat should echo the line written to its piped stdin; captured: {captured:?}", + ); + } + + /// The paste-an-auth-code shape: the command prompts by blocking on a + /// line read, and the caller feeds the answer through the writer while + /// the fix is running. + #[tokio::test] + async fn run_command_streaming_piped_stdin_feeds_prompt_style_read() { + let lines: Arc>> = Arc::new(Mutex::new(Vec::new())); + let lines_clone = lines.clone(); + let (writer, stdin) = FixStdin::pipe(); + + let handle = tokio::spawn(run_command_streaming( + "read -r line && echo \"got-$line\"".to_string(), + None, + Some(stdin), + move |line| { + lines_clone.lock().unwrap().push(line.to_string()); + }, + )); + + writer.send_line("doctor-stdin-auth-code").unwrap(); + drop(writer); + + let result = handle.await.unwrap(); + assert!(result.is_ok(), "read/echo should exit 0; got {result:?}"); + let captured = lines.lock().unwrap().clone(); + assert!( + captured.iter().any(|l| l == "got-doctor-stdin-auth-code"), + "prompt-style read should see the sent line; captured: {captured:?}", + ); + } + + /// A writer held across the fix's completion must not hang the run — the + /// stdin writer thread is detached, never joined. Afterwards, `send_line` + /// must fail cleanly (never panic): the first post-exit send may still + /// queue, but it wakes the writer thread, whose write fails with EPIPE + /// and drops the receiver, so sends error from then on. + #[tokio::test] + async fn run_command_streaming_piped_stdin_no_hang_when_writer_outlives_child() { + let (writer, stdin) = FixStdin::pipe(); + + let result = run_command_streaming( + "echo doctor-stdin-done".to_string(), + None, + Some(stdin), + |_| {}, + ) + .await; + + assert!(result.is_ok(), "echo fix should complete; got {result:?}"); + + let mut saw_error = false; + for _ in 0..100 { + if writer.send_line("late-line").is_err() { + saw_error = true; + break; + } + std::thread::sleep(Duration::from_millis(10)); + } + assert!( + saw_error, + "send_line after child exit should eventually return Err", + ); + } + /// `execute_fix(|_| {})` and `execute_fix_streaming(.., |_| {})` must /// produce identical results for the same fix lookup — `execute_fix` is /// supposed to be a thin delegate. @@ -1622,6 +1808,7 @@ mod tests { command_override: Some(script_name.to_string()), npm_registry: None, env: Some(env), + stdin: None, }, move |line| { lines_clone.lock().unwrap().push(line.to_string()); @@ -1676,6 +1863,7 @@ mod tests { command_override: Some(command.to_string()), npm_registry: None, env: Some(env), + stdin: None, }, move |line| { lines_clone.lock().unwrap().push(line.to_string()); From 696c61d6f074fdfd89f712cf837547388f5340b8 Mon Sep 17 00:00:00 2001 From: Matt Toohey Date: Wed, 19 Aug 2026 15:58:49 +1000 Subject: [PATCH 2/5] fix(doctor): error on a reused FixStdin instead of spawning a doomed fix MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `ExecuteFixOptions` and `FixStdin` are both `Clone`, so a host that caches its options struct and retries a failed login got a second child whose stdin was EOF'd at spawn, while `send_line` calls either vanished or landed in the *first* child's stdin. The retry hung or died with nothing in the output explaining why — block/berd#99 re-created by the mechanism meant to fix it. Claim the receiver before anything is launched and turn the already-consumed case into an error: - `run_command_streaming_blocking` hoists `take_receiver` to the top of the body, ahead of `build_shell_command`, and maps `None` to "FixStdin already consumed by a previous fix execution". Pre-spawn matters beyond tidiness: returning `Err` after `spawn` would drop the `Child`, and `std::process::Child`'s `Drop` neither kills nor reaps, so the doomed login subprocess would keep running and become a zombie. - `Stdio::piped()` now keys off the claimed receiver, and the post-spawn block collapses to the writer thread — no `match`, no `drop(child_stdin)` EOF arm. - `stdin: None` is untouched byte-for-byte; terminal hosts with legitimately interactive fixes keep inheriting stdin. A spawn failure now consumes the receiver even though no child got the pipe, so a retry with the same cached options reports the reuse rather than the underlying spawn error. That is strictly better than a hang, and a `restore_receiver` escape hatch would add a branch no test in this crate can exercise (the shell path isn't injectable). Docs for `FixStdin`, `ExecuteFixOptions::stdin`, `with_stdin`, and `take_receiver` are corrected to the single-use contract they now have. Tests: new reuse case asserts the second run errors naming the reuse and that nothing reaches `on_line`, which pins the pre-spawn ordering — `run_command_streaming` emits no preamble of its own, so any output would mean the child ran. Verified with `cargo test` in crates/doctor (119 passed) plus `cargo fmt`, `cargo clippy --all-targets`, and `cargo check` in apps/staged/src-tauri, which is workspace-excluded and so never built by crate-local commands. Signed-off-by: Matt Toohey --- crates/doctor/src/lib.rs | 115 ++++++++++++++++++++++++++++----------- 1 file changed, 84 insertions(+), 31 deletions(-) diff --git a/crates/doctor/src/lib.rs b/crates/doctor/src/lib.rs index 94a2d7683..2e95d24e6 100644 --- a/crates/doctor/src/lib.rs +++ b/crates/doctor/src/lib.rs @@ -675,9 +675,12 @@ struct FreshnessTarget { /// Opt-in piped stdin for a fix subprocess. Create with [`FixStdin::pipe`]; /// keep the [`FixStdinWriter`], put the `FixStdin` in -/// [`ExecuteFixOptions::stdin`]. Cloning shares the underlying receiver: the -/// first execution to spawn takes it, so a cloned options struct cannot feed -/// two children. +/// [`ExecuteFixOptions::stdin`]. +/// +/// Single-use: the first execution claims the underlying receiver, and any +/// later execution handed the same `FixStdin` — or a clone of it, including one +/// carried along by a cloned [`ExecuteFixOptions`] — fails with an error +/// instead of spawning. Retrying a fix needs a fresh pipe. #[derive(Debug, Clone)] pub struct FixStdin { rx: Arc>>>, @@ -698,8 +701,9 @@ impl FixStdin { ) } - /// Take the receiving end for a spawned child. First caller wins; `None` - /// on later calls (a clone already fed an execution). + /// Take the receiving end for a child about to be spawned. First caller + /// wins; `None` on later calls (a clone already fed an execution), which + /// the caller turns into an error rather than an immediately-EOF'd pipe. fn take_receiver(&self) -> Option> { self.rx.lock().ok().and_then(|mut rx| rx.take()) } @@ -734,6 +738,10 @@ pub struct ExecuteFixOptions { /// Opt-in piped stdin for the fix subprocess (see [`FixStdin::pipe`]). /// `None` keeps the child inheriting the host process's stdin, so /// terminal hosts can still run interactive fixes directly. + /// + /// A `FixStdin` feeds exactly one execution, so a cached options struct + /// must have this field refreshed (or be rebuilt) before a fix is retried; + /// reusing it fails the run. pub stdin: Option, } @@ -743,6 +751,9 @@ impl ExecuteFixOptions { self } + /// Attach an opt-in stdin pipe (see [`FixStdin::pipe`]). The `FixStdin` + /// feeds exactly one execution: call this again with a fresh pipe for + /// every retry rather than reusing a built options struct. pub fn with_stdin(mut self, stdin: FixStdin) -> Self { self.stdin = Some(stdin); self @@ -1089,13 +1100,27 @@ where { use std::io::{BufRead, BufReader, Write}; + // Claim the write end before anything is launched: a `FixStdin` whose + // receiver a previous execution already took can never deliver a line, so + // the child would block forever on a pipe nobody writes — the exact hang + // this option exists to fix. Always a caller bug, so surface it at the call + // site rather than spawning a doomed subprocess. + let stdin_rx = match stdin { + Some(fix_stdin) => Some(fix_stdin.take_receiver().ok_or_else(|| { + "FixStdin already consumed by a previous fix execution; \ + create a fresh pipe with FixStdin::pipe() for each run" + .to_string() + })?), + None => None, + }; + let mut command = build_shell_command(command, &[], env); command .stdout(std::process::Stdio::piped()) .stderr(std::process::Stdio::piped()); // Opt-in only: without a `FixStdin` the child keeps inheriting the host // process's stdin, so interactive fixes in terminal hosts are untouched. - if stdin.is_some() { + if stdin_rx.is_some() { command.stdin(std::process::Stdio::piped()); } command::configure_command(&mut command); @@ -1103,33 +1128,25 @@ where .spawn() .map_err(|e| format!("Failed to run command: {e}"))?; - if let Some(fix_stdin) = stdin { + if let Some(stdin_rx) = stdin_rx { let mut child_stdin = child.stdin.take().expect("stdin was piped"); - match fix_stdin.take_receiver() { - Some(stdin_rx) => { - // Detached on purpose: joining would hang the fix whenever a - // caller still holds a writer after the child exits (the - // thread would be parked in `iter()`). It exits on its own - // when every writer drops (channel closed) or a write fails - // once the child is gone (Rust ignores SIGPIPE, so EPIPE - // surfaces as a clean `Err`); dropping `child_stdin` then - // delivers EOF. - std::thread::spawn(move || { - for line in stdin_rx.iter() { - if child_stdin - .write_all(format!("{line}\n").as_bytes()) - .and_then(|()| child_stdin.flush()) - .is_err() - { - break; - } - } - }); + // Detached on purpose: joining would hang the fix whenever a caller + // still holds a writer after the child exits (the thread would be + // parked in `iter()`). It exits on its own when every writer drops + // (channel closed) or a write fails once the child is gone (Rust + // ignores SIGPIPE, so EPIPE surfaces as a clean `Err`); dropping + // `child_stdin` then delivers EOF. + std::thread::spawn(move || { + for line in stdin_rx.iter() { + if child_stdin + .write_all(format!("{line}\n").as_bytes()) + .and_then(|()| child_stdin.flush()) + .is_err() + { + break; + } } - // A clone of this `FixStdin` already fed another execution; no - // line can ever arrive, so close the pipe now (immediate EOF). - None => drop(child_stdin), - } + }); } let stdout = child.stdout.take().expect("stdout was piped"); @@ -1406,6 +1423,42 @@ mod tests { ); } + /// Reusing a `FixStdin` (or a clone) for a second execution must fail + /// loudly rather than hand the child an immediately-EOF'd stdin — the + /// receiver lives with the first run, so a second could only hang. The + /// second run must also never spawn: nothing reaches `on_line`. + #[tokio::test] + async fn run_command_streaming_piped_stdin_errors_when_reused() { + let (writer, stdin) = FixStdin::pipe(); + let reused = stdin.clone(); + writer.send_line("doctor-stdin-reuse-first").unwrap(); + drop(writer); + + let first = run_command_streaming("cat".to_string(), None, Some(stdin), |_| {}).await; + assert!(first.is_ok(), "first run should succeed; got {first:?}"); + + let lines: Arc>> = Arc::new(Mutex::new(Vec::new())); + let lines_clone = lines.clone(); + let second = run_command_streaming( + "echo doctor-stdin-reuse-second".to_string(), + None, + Some(reused), + move |line| lines_clone.lock().unwrap().push(line.to_string()), + ) + .await; + + let err = second.expect_err("reusing a consumed FixStdin should fail"); + let captured = lines.lock().unwrap().clone(); + assert!( + err.contains("already consumed"), + "error should name the reuse; got {err:?}", + ); + assert!( + captured.is_empty(), + "second run must not spawn; captured: {captured:?}", + ); + } + /// `execute_fix(|_| {})` and `execute_fix_streaming(.., |_| {})` must /// produce identical results for the same fix lookup — `execute_fix` is /// supposed to be a thin delegate. From 2ab5e4b82947374d038679d1ae58aa2d221b7d7c Mon Sep 17 00:00:00 2001 From: Matt Toohey Date: Wed, 19 Aug 2026 16:07:35 +1000 Subject: [PATCH 3/5] fix(doctor): bound the fix runner's two unbounded waits MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `run_command_streaming_blocking` could park forever in two places, and the two were entangled: the detached stdin writer thread's only "am I done?" signal is the main loop finishing, and the main loop can't finish while the child is alive. Fixing either alone leaves the other in place. Reclaim the writer thread: - The thread polls `recv_timeout(STDIN_WRITER_POLL_INTERVAL)` (250 ms) against a shared `AtomicBool` instead of parking in `stdin_rx.iter()`, so a host that holds its `FixStdinWriter` and simply stops sending no longer leaks an OS thread and a `ChildStdin` fd per fix run. - A `FixFinishedFlag` drop guard sets that flag on every exit path — including the `wait` error return and a panic in `on_line` — so worst-case thread lifetime is "fix completion + one poll interval" regardless of caller discipline. Breaking out still drops `child_stdin`, delivering EOF. - Dropping the receiver on the way out makes `send_line` fail deterministically from then on, which lets the writer-outlives-child test replace its 100 x 10 ms polling loop with a fixed wait. Give the fix itself a deadline: - New `FixTimeout` (`Standard` / `After` / `Unbounded`) plus `DEFAULT_FIX_TIMEOUT = 600s`, added to `ExecuteFixOptions` with a `with_timeout` builder and threaded through `execute_fix_streaming_with_env_options` -> `run_command_streaming` -> `run_command_streaming_blocking` exactly as `stdin` was. An enum rather than `Option` because `None` reads as both "use the default" and "no timeout"; `Unbounded` also keeps the plain `rx.recv()` path, sidestepping the `recv_timeout(Duration::MAX)` instant-overflow hazard. - 10 minutes clears a cold-cache `npm install -g` behind the corporate proxy and a human doing SSO in a browser. The 10s/15s probe timeouts in `command.rs` are wildly wrong for this path. An idle timeout would suit installs better but is precisely wrong for `auth login`, which prints its URL and then goes deliberately silent; `FixTimeout` can gain `Idle` later. - Enforcement replaces `rx.iter()` with `recv_timeout` against the deadline and `child.wait()` with `wait_timeout` — a process can close both pipes and keep running. On expiry: drain queued lines with `try_recv` so the last real output survives, emit one greppable notice through `on_line`, kill, reap, and return `Err`. The reader threads are deliberately *not* joined, because a descendant that escaped the process group can hold the inherited stdout open indefinitely (`command.rs`'s module docs; the escaped-descendant test). - `kill_child_process_group_or_child` is promoted to `pub(crate)`, and the child gets `process_group(0)` only when `stdin.is_some()`. Group-kill is available precisely when doctor owns the child's stdin and so the child can't touch the tty; setting it unconditionally would give a terminal host's tty-reading fix a SIGTTIN stop. `stdin: None` stays byte-for-byte unchanged. The in-crate `ExecuteFixOptions` literals switch to `..Default::default()` so the next option field is genuinely additive. Tests: `After(100ms)` vs `sleep 60` returns `Err` naming the timeout and the command in well under a second, with the notice line visible to `on_line`; the group kill takes a backgrounded grandchild with it (verified non-vacuous by disabling `process_group(0)`, which makes it fail); the timeout returns promptly even when a `setsid`-escaped descendant holds the pipes open; the writer thread has retired a few poll intervals after the fix returns; and a guard test pins `DEFAULT_FIX_TIMEOUT` at 600s, `FixTimeout::Standard` as the `ExecuteFixOptions` default, and the default at >= 30x probe scale. Verified with `cargo test` in crates/doctor (123 passed) plus `cargo fmt` and `cargo clippy --all-targets`. Signed-off-by: Matt Toohey --- crates/doctor/src/command.rs | 6 +- crates/doctor/src/lib.rs | 429 +++++++++++++++++++++++++++++------ 2 files changed, 362 insertions(+), 73 deletions(-) diff --git a/crates/doctor/src/command.rs b/crates/doctor/src/command.rs index 4ccc3a2a8..c2eb1b4a4 100644 --- a/crates/doctor/src/command.rs +++ b/crates/doctor/src/command.rs @@ -193,7 +193,11 @@ fn clean_up_after_incomplete_wait(child: &mut Child) { let _ = child.wait(); } -fn kill_child_process_group_or_child(child: &mut Child) { +/// Best-effort kill: target the child's process group first so a shell's whole +/// command tree goes with it, falling back to the direct child when the group +/// lookup fails (the child wasn't spawned with `process_group(0)`, or it isn't +/// Unix). Callers must still reap afterwards. +pub(crate) fn kill_child_process_group_or_child(child: &mut Child) { if kill_child_process_group(child) { return; } diff --git a/crates/doctor/src/lib.rs b/crates/doctor/src/lib.rs index 2e95d24e6..d26d4a644 100644 --- a/crates/doctor/src/lib.rs +++ b/crates/doctor/src/lib.rs @@ -19,7 +19,9 @@ pub use types::{AgentVersionInfo, CheckStatus, DoctorCheck, DoctorReport, FixTyp use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; use agents::{ bundled_version_probe_args, check_single_ai_agent, derive_update_command, lookup_fix_command, @@ -691,6 +693,11 @@ impl FixStdin { /// the `FixStdin` to place in [`ExecuteFixOptions::stdin`]. Lines sent /// before the fix subprocess spawns are buffered and written once it does; /// dropping every writer clone closes the child's stdin (EOF). + /// + /// Dropping the writers is how you say "no more input", but it is not what + /// bounds the machinery: the writer thread also retires shortly after the + /// fix ends, so a writer a host forgets to drop can't strand a thread or + /// the child's stdin handle. pub fn pipe() -> (FixStdinWriter, FixStdin) { let (tx, rx) = std::sync::mpsc::channel(); ( @@ -717,8 +724,10 @@ pub struct FixStdinWriter { impl FixStdinWriter { /// Queue one line for the fix's stdin; a trailing `\n` is appended and the - /// pipe is flushed. `Err` when the fix has already finished (its stdin - /// pipe is closed). + /// pipe is flushed. Delivery is best-effort — `Ok` means the line reached + /// the writer thread's queue, not that the child read it. `Err` once that + /// thread has retired, which happens a fraction of a second after the fix + /// finishes, or immediately when a write hits the closed pipe. pub fn send_line(&self, line: impl Into) -> Result<(), String> { self.tx .send(line.into()) @@ -726,6 +735,50 @@ impl FixStdinWriter { } } +/// How often the stdin writer thread wakes to ask whether the fix has +/// finished. Sets the worst-case lag between a fix completing and its writer +/// thread (plus the child's stdin handle) being reclaimed, which matters +/// because a host is free to hold its [`FixStdinWriter`] and simply stop +/// sending — nothing else would ever wake a thread parked on the channel. +const STDIN_WRITER_POLL_INTERVAL: Duration = Duration::from_millis(250); + +/// Wall-clock bound on a single fix execution. +/// +/// Fixes are install/auth/update actions, so the bound has to clear a +/// cold-cache `npm install -g` behind a corporate proxy and a human doing SSO +/// in a browser — orders of magnitude above the probe timeouts in +/// [`crate::command`]. This is an enum rather than `Option` because +/// `None` reads as both "use the default" and "no timeout"; here every literal +/// has to say which it means, and `Unbounded` stays reachable for a caller +/// that genuinely wants the old forever-wait. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub enum FixTimeout { + /// [`DEFAULT_FIX_TIMEOUT`]. + #[default] + Standard, + /// A caller-chosen bound. + After(Duration), + /// No bound at all: the fix runs until it exits on its own. + Unbounded, +} + +impl FixTimeout { + /// The wall-clock bound, or `None` for [`FixTimeout::Unbounded`]. + fn duration(self) -> Option { + match self { + FixTimeout::Standard => Some(DEFAULT_FIX_TIMEOUT), + FixTimeout::After(duration) => Some(duration), + FixTimeout::Unbounded => None, + } + } +} + +/// Deadline applied by [`FixTimeout::Standard`]. Deliberately generous: it +/// exists to stop a wedged fix from pinning a blocking worker and a process +/// tree for the lifetime of the host, not to police slow-but-honest installs +/// or a leisurely browser login. +pub const DEFAULT_FIX_TIMEOUT: Duration = Duration::from_secs(600); + /// Options for executing a doctor fix command. #[derive(Debug, Clone, Default)] pub struct ExecuteFixOptions { @@ -743,6 +796,8 @@ pub struct ExecuteFixOptions { /// must have this field refreshed (or be rebuilt) before a fix is retried; /// reusing it fails the run. pub stdin: Option, + /// Wall-clock bound on the fix. Defaults to [`FixTimeout::Standard`]. + pub timeout: FixTimeout, } impl ExecuteFixOptions { @@ -758,6 +813,12 @@ impl ExecuteFixOptions { self.stdin = Some(stdin); self } + + /// Override the wall-clock bound on the fix (see [`FixTimeout`]). + pub fn with_timeout(mut self, timeout: FixTimeout) -> Self { + self.timeout = timeout; + self + } } /// Run a fix command for a doctor check, identified by check ID and fix type. @@ -791,8 +852,7 @@ pub async fn execute_fix_with_options( ExecuteFixOptions { command_override, npm_registry: npm_registry.map(str::to_string), - env: None, - stdin: None, + ..Default::default() }, ) .await @@ -849,8 +909,7 @@ where ExecuteFixOptions { command_override, npm_registry: npm_registry.map(str::to_string), - env: None, - stdin: None, + ..Default::default() }, on_line, ) @@ -884,8 +943,9 @@ where // Fixes are intentionally not routed through the bounded probe runner: // these are user-triggered install/auth/update actions and can reasonably - // be interactive or long-running. - run_command_streaming(command, opts.env, opts.stdin, on_line).await + // be interactive or long-running, so they get the far more generous + // `FixTimeout` bound instead of a probe timeout. + run_command_streaming(command, opts.env, opts.stdin, opts.timeout, on_line).await } /// Async wrapper that runs `run_command_streaming_blocking` on the blocking pool. @@ -893,13 +953,14 @@ pub(crate) async fn run_command_streaming( command: String, env: Option, stdin: Option, + timeout: FixTimeout, on_line: F, ) -> Result<(), String> where F: FnMut(&str) + Send + 'static, { tokio::task::spawn_blocking(move || { - run_command_streaming_blocking(&command, env.as_ref(), stdin, on_line) + run_command_streaming_blocking(&command, env.as_ref(), stdin, timeout, on_line) }) .await .unwrap_or_else(|e| Err(format!("Task failed: {e}"))) @@ -1083,22 +1144,54 @@ pub(crate) fn execute_command_with_path_prefix_with_env( } } +/// Sets the shared "the fix is over" flag when `run_command_streaming_blocking` +/// leaves its body — normal return, error return, or a panic in `on_line`. That +/// is the only signal the detached stdin writer thread has, so it has to fire on +/// every path or the thread outlives the fix. +struct FixFinishedFlag(Arc); + +impl Drop for FixFinishedFlag { + fn drop(&mut self) { + self.0.store(true, Ordering::Relaxed); + } +} + /// Spawn `command` through a login shell, stream stdout/stderr lines to -/// `on_line`, and return based on the process exit status. This path is -/// deliberately unbounded: fix commands are user-triggered install/auth/update -/// actions and may prompt or run package managers. Stderr lines are also -/// accumulated so a non-zero exit can surface a useful error message (matching -/// the non-streaming behavior of the previous `execute_command`). +/// `on_line`, and return based on the process exit status. Bounded by +/// `timeout`, which is generous rather than tight: fix commands are +/// user-triggered install/auth/update actions and may prompt or run package +/// managers. Stderr lines are also accumulated so a non-zero exit can surface a +/// useful error message (matching the non-streaming behavior of the previous +/// `execute_command`). fn run_command_streaming_blocking( command: &str, env: Option<&DoctorEnv>, stdin: Option, + timeout: FixTimeout, mut on_line: F, ) -> Result<(), String> where F: FnMut(&str), { use std::io::{BufRead, BufReader, Write}; + use std::sync::mpsc::RecvTimeoutError; + + use wait_timeout::ChildExt; + + fn consume(msg: StreamLine, on_line: &mut F, stderr_accum: &mut String) { + match msg { + StreamLine::Stdout(s) => { + on_line(&s); + } + StreamLine::Stderr(s) => { + on_line(&s); + if !stderr_accum.is_empty() { + stderr_accum.push('\n'); + } + stderr_accum.push_str(&s); + } + } + } // Claim the write end before anything is launched: a `FixStdin` whose // receiver a previous execution already took can never deliver a line, so @@ -1114,37 +1207,60 @@ where None => None, }; - let mut command = build_shell_command(command, &[], env); - command + let mut shell_command = build_shell_command(command, &[], env); + shell_command .stdout(std::process::Stdio::piped()) .stderr(std::process::Stdio::piped()); // Opt-in only: without a `FixStdin` the child keeps inheriting the host // process's stdin, so interactive fixes in terminal hosts are untouched. if stdin_rx.is_some() { - command.stdin(std::process::Stdio::piped()); + shell_command.stdin(std::process::Stdio::piped()); + // Own the whole tree so a timeout can kill more than the login shell: + // `kill(-pid)` only reaches an `npm install` under `zsh -lc` if the + // shell leads its own group. Gated on piped stdin because a child in + // its own group that reads the controlling terminal gets SIGTTIN and + // stops — impossible here precisely because doctor owns its stdin, but + // a real regression for a terminal host on the inherited-stdin path. + #[cfg(unix)] + { + use std::os::unix::process::CommandExt; + shell_command.process_group(0); + } } - command::configure_command(&mut command); - let mut child = command + command::configure_command(&mut shell_command); + let mut child = shell_command .spawn() .map_err(|e| format!("Failed to run command: {e}"))?; + let finished = Arc::new(AtomicBool::new(false)); + let _finished_guard = FixFinishedFlag(finished.clone()); + if let Some(stdin_rx) = stdin_rx { let mut child_stdin = child.stdin.take().expect("stdin was piped"); // Detached on purpose: joining would hang the fix whenever a caller - // still holds a writer after the child exits (the thread would be - // parked in `iter()`). It exits on its own when every writer drops - // (channel closed) or a write fails once the child is gone (Rust - // ignores SIGPIPE, so EPIPE surfaces as a clean `Err`); dropping - // `child_stdin` then delivers EOF. - std::thread::spawn(move || { - for line in stdin_rx.iter() { - if child_stdin - .write_all(format!("{line}\n").as_bytes()) - .and_then(|()| child_stdin.flush()) - .is_err() - { - break; + // still holds a writer after the child exits. Instead of parking on the + // channel, the thread polls, so it retires on any of three signals — + // every writer dropped (channel closed), a write failing once the child + // is gone (Rust ignores SIGPIPE, so EPIPE surfaces as a clean `Err`), + // or the fix finishing while a host still holds its writer. Dropping + // `child_stdin` on the way out delivers EOF. + std::thread::spawn(move || loop { + match stdin_rx.recv_timeout(STDIN_WRITER_POLL_INTERVAL) { + Ok(line) => { + if child_stdin + .write_all(format!("{line}\n").as_bytes()) + .and_then(|()| child_stdin.flush()) + .is_err() + { + break; + } } + Err(RecvTimeoutError::Timeout) => { + if finished.load(Ordering::Relaxed) { + break; + } + } + Err(RecvTimeoutError::Disconnected) => break, } }); } @@ -1172,28 +1288,75 @@ where } }); + let limit = timeout.duration(); + let deadline = limit.map(|limit| Instant::now() + limit); let mut stderr_accum = String::new(); - for msg in rx.iter() { - match msg { - StreamLine::Stdout(s) => { - on_line(&s); - } - StreamLine::Stderr(s) => { - on_line(&s); - if !stderr_accum.is_empty() { - stderr_accum.push('\n'); + let mut expired = false; + + loop { + let msg = match deadline { + Some(deadline) => { + match rx.recv_timeout(deadline.saturating_duration_since(Instant::now())) { + Ok(msg) => msg, + Err(RecvTimeoutError::Timeout) => { + expired = true; + break; + } + Err(RecvTimeoutError::Disconnected) => break, } - stderr_accum.push_str(&s); } - } + // `recv_timeout(Duration::MAX)` overflows instantly, so the + // unbounded case keeps the plain blocking receive. + None => match rx.recv() { + Ok(msg) => msg, + Err(_) => break, + }, + }; + consume(msg, &mut on_line, &mut stderr_accum); } - let _ = stdout_thread.join(); - let _ = stderr_thread.join(); + let status = if expired { + None + } else { + // Both pipes hit EOF, so the readers are already done and joining is + // immediate. The process can still outlive its pipes, though, so the + // reap is bounded by the same deadline. + let _ = stdout_thread.join(); + let _ = stderr_thread.join(); + match deadline { + Some(deadline) => child + .wait_timeout(deadline.saturating_duration_since(Instant::now())) + .map_err(|e| format!("Failed to wait for command: {e}"))?, + None => Some( + child + .wait() + .map_err(|e| format!("Failed to wait for command: {e}"))?, + ), + } + }; - let status = child - .wait() - .map_err(|e| format!("Failed to wait for command: {e}"))?; + let Some(status) = status else { + let limit = limit.expect("a deadline only exists when the fix is bounded"); + // Anything the readers already queued is real output the user should + // see before the notice explaining why it stopped. + while let Ok(msg) = rx.try_recv() { + consume(msg, &mut on_line, &mut stderr_accum); + } + on_line(&format!( + "doctor: fix timed out after {} — terminating", + format_duration(limit) + )); + command::kill_child_process_group_or_child(&mut child); + let _ = child.wait(); + // The reader threads are deliberately not joined: a descendant that + // escaped the process group can hold the inherited stdout open long + // after the fix is dead, and waiting on that is the hang this timeout + // exists to end. Dropping `rx` retires them at their next send. + return Err(format!( + "Fix timed out after {} without finishing: {command}", + format_duration(limit) + )); + }; if status.success() { Ok(()) @@ -1213,7 +1376,7 @@ mod tests { use std::path::Path; use std::sync::{Arc, Mutex}; - use std::time::Duration; + use std::time::{Duration, Instant}; fn timeout(label: &str, command: &str) -> CommandTimeout { CommandTimeout::new(label, command, Duration::from_secs(15)) @@ -1312,6 +1475,7 @@ mod tests { "echo doctor-streaming-marker-hello && echo doctor-streaming-marker-world".to_string(), None, None, + FixTimeout::Standard, move |line| { lines_clone.lock().unwrap().push(line.to_string()); }, @@ -1347,9 +1511,15 @@ mod tests { writer.send_line("doctor-stdin-marker-echo").unwrap(); drop(writer); - let result = run_command_streaming("cat".to_string(), None, Some(stdin), move |line| { - lines_clone.lock().unwrap().push(line.to_string()); - }) + let result = run_command_streaming( + "cat".to_string(), + None, + Some(stdin), + FixTimeout::Standard, + move |line| { + lines_clone.lock().unwrap().push(line.to_string()); + }, + ) .await; assert!(result.is_ok(), "cat should exit 0 on EOF; got {result:?}"); @@ -1373,6 +1543,7 @@ mod tests { "read -r line && echo \"got-$line\"".to_string(), None, Some(stdin), + FixTimeout::Standard, move |line| { lines_clone.lock().unwrap().push(line.to_string()); }, @@ -1391,35 +1562,31 @@ mod tests { } /// A writer held across the fix's completion must not hang the run — the - /// stdin writer thread is detached, never joined. Afterwards, `send_line` - /// must fail cleanly (never panic): the first post-exit send may still - /// queue, but it wakes the writer thread, whose write fails with EPIPE - /// and drops the receiver, so sends error from then on. + /// stdin writer thread is detached, never joined — and the thread must not + /// leak either. It polls `STDIN_WRITER_POLL_INTERVAL` for the finished + /// flag, so within one interval of the fix returning it has retired and + /// dropped the receiver, which makes `send_line` fail cleanly (never + /// panic) rather than queueing into a void. Waiting several intervals + /// keeps this a fixed assertion instead of a poll loop. #[tokio::test] - async fn run_command_streaming_piped_stdin_no_hang_when_writer_outlives_child() { + async fn run_command_streaming_piped_stdin_reclaims_writer_thread_after_child_exits() { let (writer, stdin) = FixStdin::pipe(); let result = run_command_streaming( "echo doctor-stdin-done".to_string(), None, Some(stdin), + FixTimeout::Standard, |_| {}, ) .await; assert!(result.is_ok(), "echo fix should complete; got {result:?}"); - let mut saw_error = false; - for _ in 0..100 { - if writer.send_line("late-line").is_err() { - saw_error = true; - break; - } - std::thread::sleep(Duration::from_millis(10)); - } + tokio::time::sleep(STDIN_WRITER_POLL_INTERVAL * 4).await; assert!( - saw_error, - "send_line after child exit should eventually return Err", + writer.send_line("late-line").is_err(), + "writer thread should have retired within one poll interval of the fix finishing", ); } @@ -1434,7 +1601,14 @@ mod tests { writer.send_line("doctor-stdin-reuse-first").unwrap(); drop(writer); - let first = run_command_streaming("cat".to_string(), None, Some(stdin), |_| {}).await; + let first = run_command_streaming( + "cat".to_string(), + None, + Some(stdin), + FixTimeout::Standard, + |_| {}, + ) + .await; assert!(first.is_ok(), "first run should succeed; got {first:?}"); let lines: Arc>> = Arc::new(Mutex::new(Vec::new())); @@ -1443,6 +1617,7 @@ mod tests { "echo doctor-stdin-reuse-second".to_string(), None, Some(reused), + FixTimeout::Standard, move |line| lines_clone.lock().unwrap().push(line.to_string()), ) .await; @@ -1459,6 +1634,118 @@ mod tests { ); } + /// The default bound must stay at fix scale, not probe scale. A fix is an + /// `npm install -g` behind a corporate proxy or a human doing SSO in a + /// browser; retuning this toward `DEFAULT_PROBE_TIMEOUT` would kill honest + /// work mid-flight. + #[test] + fn default_fix_timeout_stays_at_fix_scale() { + assert_eq!(DEFAULT_FIX_TIMEOUT, Duration::from_secs(600)); + assert_eq!(ExecuteFixOptions::default().timeout, FixTimeout::Standard); + assert_eq!(FixTimeout::Standard.duration(), Some(DEFAULT_FIX_TIMEOUT)); + assert_eq!(FixTimeout::Unbounded.duration(), None); + assert!( + DEFAULT_FIX_TIMEOUT >= DEFAULT_PROBE_TIMEOUT * 30, + "fix timeout must stay far above probe scale", + ); + } + + /// A fix that never finishes must return on its deadline instead of + /// pinning the blocking worker forever — the whole point of the bound. + #[tokio::test] + async fn run_command_streaming_returns_when_the_fix_outlives_its_timeout() { + let lines: Arc>> = Arc::new(Mutex::new(Vec::new())); + let lines_clone = lines.clone(); + let started = Instant::now(); + + let result = run_command_streaming( + "sleep 60".to_string(), + None, + None, + FixTimeout::After(Duration::from_millis(100)), + move |line| lines_clone.lock().unwrap().push(line.to_string()), + ) + .await; + + let err = result.expect_err("a fix past its deadline should fail"); + assert!( + err.contains("timed out") && err.contains("sleep 60"), + "error should name the timeout and the command; got {err:?}", + ); + assert!( + started.elapsed() < Duration::from_secs(2), + "timeout path waited for the fix instead of its deadline", + ); + let captured = lines.lock().unwrap().clone(); + assert!( + captured + .iter() + .any(|l| l.starts_with("doctor: fix timed out")), + "callers should see a notice line explaining the stop; captured: {captured:?}", + ); + } + + /// With piped stdin the shell leads its own process group, so the timeout + /// kill must take the whole tree — not just the login shell, leaving a + /// backgrounded installer running. + #[cfg(unix)] + #[tokio::test] + async fn run_command_streaming_timeout_kills_the_whole_process_tree() { + let tmp = unique_tmp_dir("fix-timeout-tree"); + let marker = tmp.join("grandchild-ran"); + let (_writer, stdin) = FixStdin::pipe(); + + let result = run_command_streaming( + format!("(sleep 2; touch {}) & sleep 60", marker.display()), + None, + Some(stdin), + FixTimeout::After(Duration::from_millis(300)), + |_| {}, + ) + .await; + + assert!(result.is_err(), "timed-out fix should fail; got {result:?}"); + // Past when the backgrounded grandchild would have written its marker + // had it survived the group kill. + tokio::time::sleep(Duration::from_secs(3)).await; + let survived = marker.exists(); + let _ = std::fs::remove_dir_all(&tmp); + assert!( + !survived, + "backgrounded grandchild outlived the timeout kill", + ); + } + + /// A descendant that escaped the process group keeps the inherited + /// stdout/stderr open, so the reader threads never see EOF. The timeout + /// path must not join them — it must return on the deadline regardless + /// (the streaming twin of `command_runner_returns_when_escaped_descendant_ + /// keeps_pipes_open`). + #[cfg(unix)] + #[tokio::test] + async fn run_command_streaming_timeout_returns_when_escaped_descendant_keeps_pipes_open() { + let started = Instant::now(); + + let result = run_command_streaming( + "perl -MPOSIX=setsid -e 'setsid(); sleep 5' & wait".to_string(), + None, + None, + FixTimeout::After(Duration::from_millis(250)), + |_| {}, + ) + .await; + + let err = result.expect_err("a fix past its deadline should fail"); + assert!( + err.contains("timed out"), + "error should name the timeout; got {err:?}", + ); + assert!( + started.elapsed() < Duration::from_secs(2), + "timeout path waited for the escaped descendant to close the pipes", + ); + } + /// `execute_fix(|_| {})` and `execute_fix_streaming(.., |_| {})` must /// produce identical results for the same fix lookup — `execute_fix` is /// supposed to be a thin delegate. @@ -1859,9 +2146,8 @@ mod tests { FixType::UpdateMain, ExecuteFixOptions { command_override: Some(script_name.to_string()), - npm_registry: None, env: Some(env), - stdin: None, + ..Default::default() }, move |line| { lines_clone.lock().unwrap().push(line.to_string()); @@ -1914,9 +2200,8 @@ mod tests { FixType::UpdateMain, ExecuteFixOptions { command_override: Some(command.to_string()), - npm_registry: None, env: Some(env), - stdin: None, + ..Default::default() }, move |line| { lines_clone.lock().unwrap().push(line.to_string()); From e94a488ee8cba424e75307a24dc290deb7ff8e10 Mon Sep 17 00:00:00 2001 From: Matt Toohey Date: Wed, 19 Aug 2026 16:07:49 +1000 Subject: [PATCH 4/5] fix(staged): stop the doctor fix options literal from breaking on new fields MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `apps/staged/src-tauri` is in the workspace `exclude` list, so `cargo check` under `crates/` never compiles it — while `staged-ci.yml` triggers on `crates/**`. Its exhaustive `ExecuteFixOptions` literal therefore turns every new doctor option into a CI-only breakage, twice now: `stdin`, then `timeout`. Switch it to `..Default::default()`. Staged's fixes are non-interactive, so the defaults are what it wanted anyway — inherited stdin and the standard 600s fix timeout, both far more than any install or login this runs needs. Verified with `cargo check` from apps/staged/src-tauri. Signed-off-by: Matt Toohey --- apps/staged/src-tauri/src/doctor.rs | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/apps/staged/src-tauri/src/doctor.rs b/apps/staged/src-tauri/src/doctor.rs index 84b79afb6..e8eb76508 100644 --- a/apps/staged/src-tauri/src/doctor.rs +++ b/apps/staged/src-tauri/src/doctor.rs @@ -55,13 +55,17 @@ fn execute_fix_options( command_override: Option, env_vars: Vec<(String, String)>, ) -> ExecuteFixOptions { + // Everything else stays at doctor's defaults: Staged's fixes are + // non-interactive, so nothing here feeds a prompt and the child keeps + // inheriting stdin rather than getting a piped one; the standard fix + // timeout is far above any install or login this runs. Spelled with + // `..Default::default()` so a new doctor option doesn't break this + // workspace-excluded crate, which `cargo check` under `crates/` never + // compiles but `staged-ci.yml` does. ExecuteFixOptions { command_override, npm_registry: crate::managed_acp_tools::npm_registry().map(str::to_string), - env: None, - // Staged's fixes are non-interactive: nothing here feeds a prompt, so - // the child keeps inheriting stdin rather than getting a piped one. - stdin: None, + ..Default::default() } .with_env_snapshot(env_vars) } From c29da8f1cd93158d4cd57b0e04e36f61ea64905a Mon Sep 17 00:00:00 2001 From: Matt Toohey Date: Wed, 19 Aug 2026 16:18:02 +1000 Subject: [PATCH 5/5] fix(doctor): write fix stdin inline so `Ok` from `send_line` means delivered MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `send_line` was documented as returning `Err` "when the fix has already finished (its stdin pipe is closed)" and could not honour it: it returned `mpsc::Sender::send`'s result, which fails only once the receiver drops, and the receiver lived in the detached writer thread. So the first post-exit `send_line` — plus anything queued behind it — returned `Ok` and was then discarded when the thread's write hit EPIPE. For berd#99 that is the difference between an error and a hang: the login subprocess dies, the user pastes the auth code a beat later, `send_line` says `Ok`, the code goes nowhere, and a host treating `Ok` as delivery waits forever with nothing in the log to explain it. Replace the channel and the thread with a three-state pipe shared behind the `Arc>` `FixStdin` already carried, written inline by the caller: - `FixStdinState` is `Buffered { lines, eof, claimed }` -> `Live(ChildStdin)` -> `Closed`. `send_line` queues on `Buffered`, writes through on `Live` and reports the real `io::Error`, and fails fast on `Closed` — which is latched, so a dead pipe is discovered once. `ChildStdin: Debug`, so the public `Debug`/`Clone` derives on `FixStdin` and `ExecuteFixOptions` survive. - EOF on last-writer-drop, previously free from the mpsc disconnect, comes from a `Drop` on a `FixStdinWriterInner` behind an `Arc`, keeping `FixStdinWriter: Clone`. The `Buffered { eof: true }` arm is load-bearing: a host may queue a line, drop the writer, and only then start the fix, and those lines must still be replayed before the pipe closes. - The runner claims pre-spawn exactly as before (`claim` replaces `take_receiver`), then `attach`es the child's stdin *after* the reader threads start — the replay writes inline on the runner thread, so a queue bigger than the pipe buffer would otherwise deadlock against a child whose output nobody is draining. - `FixStdinCloser` replaces `FixFinishedFlag`, closing the pipe on every exit path (return, error, timeout, spawn failure, a panic in `on_line`). EPIPE cannot carry this alone: a probe confirms that with the runner's `zsh -l -c` shape a backgrounded grandchild inherits stdin and keeps the read end open, so writes into a finished fix's pipe still succeed. Two review comments dissolve rather than getting documented around: there is no detached thread to leak an OS thread and a `ChildStdin` fd for, so `STDIN_WRITER_POLL_INTERVAL` and its worst-case-lag caveat are both gone, and the fd is now reclaimed at fix completion instead of one poll interval later. Public API is unchanged — `FixStdin::pipe`, `send_line`, `ExecuteFixOptions:: stdin`/`with_stdin` keep their signatures, and `stdin: None` is byte-for-byte untouched. One new semantic: `send_line` performs blocking I/O under a mutex on the calling thread. For a one-line auth code against a ~64KB pipe buffer that is instantaneous, but the doc now says it can block if the fix isn't reading, and a host sending anything bulkier should keep it off its async runtime. Docs also state the hazard that predates and survives this change: a fix reading *to EOF* won't exit until every writer clone drops. Tests: the writer-outlives-child test drops its 100x10ms poll loop and asserts the *first* post-fix `send_line` is `Err` naming closed input; a new grandchild test pins the explicit close (verified non-vacuous — disabling `close()` fails it, while the no-grandchild case passes on EPIPE alone); the prompt-style test now sends from inside `on_line`, on the fix's own thread in response to the fix's own prompt, so its `Ok` is the delivery guarantee rather than the pre-spawn queueing one; the `cat` round-trip test is unchanged as the `Buffered { eof: true }` regression test, and was confirmed to hang without that arm. Verified with `cargo test` in crates/doctor (124 passed) plus `cargo fmt`, `cargo clippy --all-targets`, and `cargo check` in apps/staged/src-tauri. Signed-off-by: Matt Toohey --- crates/doctor/src/lib.rs | 385 +++++++++++++++++++++++++++------------ 1 file changed, 273 insertions(+), 112 deletions(-) diff --git a/crates/doctor/src/lib.rs b/crates/doctor/src/lib.rs index d26d4a644..572d47b23 100644 --- a/crates/doctor/src/lib.rs +++ b/crates/doctor/src/lib.rs @@ -19,7 +19,6 @@ pub use types::{AgentVersionInfo, CheckStatus, DoctorCheck, DoctorReport, FixTyp use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; -use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; @@ -679,69 +678,212 @@ struct FreshnessTarget { /// keep the [`FixStdinWriter`], put the `FixStdin` in /// [`ExecuteFixOptions::stdin`]. /// -/// Single-use: the first execution claims the underlying receiver, and any -/// later execution handed the same `FixStdin` — or a clone of it, including one -/// carried along by a cloned [`ExecuteFixOptions`] — fails with an error -/// instead of spawning. Retrying a fix needs a fresh pipe. +/// Single-use: the first execution claims the pipe, and any later execution +/// handed the same `FixStdin` — or a clone of it, including one carried along by +/// a cloned [`ExecuteFixOptions`] — fails with an error instead of spawning. +/// Retrying a fix needs a fresh pipe. #[derive(Debug, Clone)] pub struct FixStdin { - rx: Arc>>>, + state: Arc>, +} + +/// The pipe's whole life cycle: `Buffered` until the fix spawns, `Live` while it +/// runs, then `Closed` — terminal, and reached when the fix ends, when the last +/// writer drops, or when a write finds the read end gone. Holding the child's +/// stdin handle here rather than in a thread of its own is what lets +/// [`FixStdinWriter::send_line`] write through and report the real outcome. +#[derive(Debug)] +enum FixStdinState { + /// Before the fix spawns: lines the host queued, replayed at spawn. + /// `claimed` marks the execution that reserved this pipe, so a second one + /// is rejected before it spawns. `eof` records that every writer dropped + /// pre-spawn, so the replay is followed immediately by closing the pipe. + Buffered { + lines: Vec, + eof: bool, + claimed: bool, + }, + /// Fix running: writes go straight into the child's stdin. + Live(std::process::ChildStdin), + /// Fix finished, every writer gone, or a write hit a dead pipe. + Closed, +} + +/// Rejection for an execution handed a `FixStdin` another one already claimed. +const FIX_STDIN_REUSED: &str = "FixStdin already consumed by a previous fix execution; \ + create a fresh pipe with FixStdin::pipe() for each run"; + +/// Rejection for a line the pipe cannot deliver because it is closed. +const FIX_STDIN_CLOSED: &str = "Fix is no longer accepting input"; + +/// Locking the pipe state recovers from poisoning instead of propagating it: no +/// invariant spans the lock (the state is a plain enum, and the only work done +/// under it is a `Vec` push or a pipe write), while treating a poisoned lock as +/// a failure would cost `send_line` its delivery guarantee and leak the child's +/// stdin handle for the lifetime of the writer. +fn lock_fix_stdin_state(state: &Mutex) -> std::sync::MutexGuard<'_, FixStdinState> { + state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} + +impl FixStdinState { + /// Queue or write `line` — with the trailing newline the caller doesn't + /// supply — according to the current state. A failed write latches `Closed` + /// so later sends fail without re-discovering the dead pipe. + fn send_line(&mut self, line: String) -> Result<(), String> { + match self { + FixStdinState::Buffered { lines, .. } => { + lines.push(line); + Ok(()) + } + FixStdinState::Live(pipe) => { + use std::io::Write; + match pipe + .write_all(format!("{line}\n").as_bytes()) + .and_then(|()| pipe.flush()) + { + Ok(()) => Ok(()), + Err(e) => { + *self = FixStdinState::Closed; + Err(format!("{FIX_STDIN_CLOSED}: {e}")) + } + } + } + FixStdinState::Closed => Err(FIX_STDIN_CLOSED.to_string()), + } + } } impl FixStdin { /// Create a connected pair: a cloneable writer for the caller to keep and /// the `FixStdin` to place in [`ExecuteFixOptions::stdin`]. Lines sent - /// before the fix subprocess spawns are buffered and written once it does; + /// before the fix subprocess spawns are queued and replayed once it does; /// dropping every writer clone closes the child's stdin (EOF). /// - /// Dropping the writers is how you say "no more input", but it is not what - /// bounds the machinery: the writer thread also retires shortly after the - /// fix ends, so a writer a host forgets to drop can't strand a thread or - /// the child's stdin handle. + /// Dropping the writers is the only way to say "no more input", and a fix + /// that reads *to EOF* rather than a fixed number of lines will not exit + /// until that happens — a host that leaves its input UI open pins the fix + /// until [`ExecuteFixOptions::timeout`] fires. Nothing else is at stake in + /// dropping them: the child's stdin handle lives with the fix and is + /// reclaimed when it ends, held writer or not. pub fn pipe() -> (FixStdinWriter, FixStdin) { - let (tx, rx) = std::sync::mpsc::channel(); + let state = Arc::new(Mutex::new(FixStdinState::Buffered { + lines: Vec::new(), + eof: false, + claimed: false, + })); ( - FixStdinWriter { tx }, - FixStdin { - rx: Arc::new(Mutex::new(Some(rx))), + FixStdinWriter { + inner: Arc::new(FixStdinWriterInner { + state: state.clone(), + }), }, + FixStdin { state }, ) } - /// Take the receiving end for a child about to be spawned. First caller - /// wins; `None` on later calls (a clone already fed an execution), which - /// the caller turns into an error rather than an immediately-EOF'd pipe. - fn take_receiver(&self) -> Option> { - self.rx.lock().ok().and_then(|mut rx| rx.take()) + /// Reserve this pipe for a child about to be spawned. First caller wins; + /// `Err` on every later call (a clone already fed an execution), which the + /// caller surfaces instead of spawning a fix whose stdin is already dead. + fn claim(&self) -> Result<(), String> { + match &mut *lock_fix_stdin_state(&self.state) { + FixStdinState::Buffered { claimed, .. } if !*claimed => { + *claimed = true; + Ok(()) + } + _ => Err(FIX_STDIN_REUSED.to_string()), + } + } + + /// Hand the spawned child's stdin to the pipe, replay whatever the host + /// queued before the spawn, and go live. + /// + /// Only ever reached after a successful [`FixStdin::claim`], which is what + /// guarantees the state is still `Buffered`; any other state means another + /// execution owns the pipe, and dropping the handle — an immediate EOF for + /// this child — is the only safe reading of that. A replay write that fails + /// is not the fix's failure (a command is free to exit successfully without + /// reading its stdin), so it only latches `Closed`; the host hears about it + /// from its next `send_line`. + fn attach(&self, child_stdin: std::process::ChildStdin) { + let mut state = lock_fix_stdin_state(&self.state); + let FixStdinState::Buffered { lines, eof, .. } = &mut *state else { + return; + }; + let queued = std::mem::take(lines); + let eof = *eof; + *state = FixStdinState::Live(child_stdin); + for line in queued { + if state.send_line(line).is_err() { + break; + } + } + if eof { + // Every writer was dropped before the spawn, so the queued lines + // above are all the input there will ever be and closing now is the + // EOF the fix is waiting for. + *state = FixStdinState::Closed; + } + } + + /// The fix is over: close the pipe so every later send fails immediately. + /// A write hitting `EPIPE` cannot be the signal on its own — a backgrounded + /// grandchild that inherited the child's stdin keeps the read end open, and + /// writes into it go on succeeding long after the fix is gone. + fn close(&self) { + *lock_fix_stdin_state(&self.state) = FixStdinState::Closed; } } -/// Cloneable handle for feeding lines to a fix subprocess's stdin. +/// Cloneable handle for feeding lines to a fix subprocess's stdin. Dropping +/// every clone closes the fix's stdin (EOF). #[derive(Debug, Clone)] pub struct FixStdinWriter { - tx: std::sync::mpsc::Sender, + inner: Arc, +} + +/// Shared by every [`FixStdinWriter`] clone so EOF is delivered exactly when +/// the last one drops, which is what keeps the writer `Clone`. +#[derive(Debug)] +struct FixStdinWriterInner { + state: Arc>, +} + +impl Drop for FixStdinWriterInner { + fn drop(&mut self) { + match &mut *lock_fix_stdin_state(&self.state) { + // Pre-spawn the queued lines still have to reach the child first, so + // record the EOF for `attach` to deliver after the replay. + FixStdinState::Buffered { eof, .. } => *eof = true, + // Otherwise dropping the state's `ChildStdin` *is* the EOF. + state => *state = FixStdinState::Closed, + } + } } impl FixStdinWriter { - /// Queue one line for the fix's stdin; a trailing `\n` is appended and the - /// pipe is flushed. Delivery is best-effort — `Ok` means the line reached - /// the writer thread's queue, not that the child read it. `Err` once that - /// thread has retired, which happens a fraction of a second after the fix - /// finishes, or immediately when a write hits the closed pipe. + /// Write one line to the fix's stdin; a trailing `\n` is appended and the + /// pipe is flushed. + /// + /// `Ok` means the bytes were handed to the child's stdin pipe — not that the + /// fix read them, since a fix can exit with bytes still buffered. `Err` + /// means the line was *not* delivered: the fix has finished, its stdin is + /// closed, or this pipe was never attached to a spawned fix. + /// + /// Lines sent before the fix spawns are queued and replayed at spawn, so + /// they return `Ok` before any pipe exists; if the fix never spawns they are + /// dropped. + /// + /// Completion is signalled by the fix's own `Result`, never by `send_line`. + /// May block if the fix isn't reading and the pipe buffer fills, so a host + /// sending anything bulkier than a pasted code should call this off its + /// async runtime. pub fn send_line(&self, line: impl Into) -> Result<(), String> { - self.tx - .send(line.into()) - .map_err(|_| "Fix is no longer accepting input".to_string()) + lock_fix_stdin_state(&self.inner.state).send_line(line.into()) } } -/// How often the stdin writer thread wakes to ask whether the fix has -/// finished. Sets the worst-case lag between a fix completing and its writer -/// thread (plus the child's stdin handle) being reclaimed, which matters -/// because a host is free to hold its [`FixStdinWriter`] and simply stop -/// sending — nothing else would ever wake a thread parked on the channel. -const STDIN_WRITER_POLL_INTERVAL: Duration = Duration::from_millis(250); - /// Wall-clock bound on a single fix execution. /// /// Fixes are install/auth/update actions, so the bound has to clear a @@ -1144,15 +1286,17 @@ pub(crate) fn execute_command_with_path_prefix_with_env( } } -/// Sets the shared "the fix is over" flag when `run_command_streaming_blocking` -/// leaves its body — normal return, error return, or a panic in `on_line`. That -/// is the only signal the detached stdin writer thread has, so it has to fire on -/// every path or the thread outlives the fix. -struct FixFinishedFlag(Arc); +/// Closes the fix's stdin pipe when `run_command_streaming_blocking` leaves its +/// body — normal return, error return, timeout, spawn failure, or a panic in +/// `on_line`. Every path has to close it: a host that still holds a +/// [`FixStdinWriter`] would otherwise keep getting `Ok` from `send_line` for a +/// fix that is already over, and the child's stdin handle would live as long as +/// that writer. +struct FixStdinCloser<'a>(&'a FixStdin); -impl Drop for FixFinishedFlag { +impl Drop for FixStdinCloser<'_> { fn drop(&mut self) { - self.0.store(true, Ordering::Relaxed); + self.0.close(); } } @@ -1173,7 +1317,7 @@ fn run_command_streaming_blocking( where F: FnMut(&str), { - use std::io::{BufRead, BufReader, Write}; + use std::io::{BufRead, BufReader}; use std::sync::mpsc::RecvTimeoutError; use wait_timeout::ChildExt; @@ -1193,19 +1337,14 @@ where } } - // Claim the write end before anything is launched: a `FixStdin` whose - // receiver a previous execution already took can never deliver a line, so - // the child would block forever on a pipe nobody writes — the exact hang - // this option exists to fix. Always a caller bug, so surface it at the call - // site rather than spawning a doomed subprocess. - let stdin_rx = match stdin { - Some(fix_stdin) => Some(fix_stdin.take_receiver().ok_or_else(|| { - "FixStdin already consumed by a previous fix execution; \ - create a fresh pipe with FixStdin::pipe() for each run" - .to_string() - })?), - None => None, - }; + // Claim the pipe before anything is launched: a `FixStdin` another execution + // already consumed can never deliver a line, so the child would block + // forever on a pipe nobody writes — the exact hang this option exists to + // fix. Always a caller bug, so surface it at the call site rather than + // spawning a doomed subprocess. + if let Some(fix_stdin) = &stdin { + fix_stdin.claim()?; + } let mut shell_command = build_shell_command(command, &[], env); shell_command @@ -1213,7 +1352,7 @@ where .stderr(std::process::Stdio::piped()); // Opt-in only: without a `FixStdin` the child keeps inheriting the host // process's stdin, so interactive fixes in terminal hosts are untouched. - if stdin_rx.is_some() { + if stdin.is_some() { shell_command.stdin(std::process::Stdio::piped()); // Own the whole tree so a timeout can kill more than the login shell: // `kill(-pid)` only reaches an `npm install` under `zsh -lc` if the @@ -1228,43 +1367,17 @@ where } } command::configure_command(&mut shell_command); + + // Declared ahead of the spawn so a spawn failure closes the pipe too: the + // claim above is already spent, so the host must not keep getting `Ok` for a + // fix that never started. + let _stdin_closer = stdin.as_ref().map(FixStdinCloser); + let mut child = shell_command .spawn() .map_err(|e| format!("Failed to run command: {e}"))?; - let finished = Arc::new(AtomicBool::new(false)); - let _finished_guard = FixFinishedFlag(finished.clone()); - - if let Some(stdin_rx) = stdin_rx { - let mut child_stdin = child.stdin.take().expect("stdin was piped"); - // Detached on purpose: joining would hang the fix whenever a caller - // still holds a writer after the child exits. Instead of parking on the - // channel, the thread polls, so it retires on any of three signals — - // every writer dropped (channel closed), a write failing once the child - // is gone (Rust ignores SIGPIPE, so EPIPE surfaces as a clean `Err`), - // or the fix finishing while a host still holds its writer. Dropping - // `child_stdin` on the way out delivers EOF. - std::thread::spawn(move || loop { - match stdin_rx.recv_timeout(STDIN_WRITER_POLL_INTERVAL) { - Ok(line) => { - if child_stdin - .write_all(format!("{line}\n").as_bytes()) - .and_then(|()| child_stdin.flush()) - .is_err() - { - break; - } - } - Err(RecvTimeoutError::Timeout) => { - if finished.load(Ordering::Relaxed) { - break; - } - } - Err(RecvTimeoutError::Disconnected) => break, - } - }); - } - + let child_stdin = child.stdin.take(); let stdout = child.stdout.take().expect("stdout was piped"); let stderr = child.stderr.take().expect("stderr was piped"); @@ -1288,6 +1401,13 @@ where } }); + // Deliberately after the readers are running: the replay of pre-spawn lines + // writes inline on this thread, so a queue larger than the pipe buffer would + // deadlock against a child whose output nobody is draining yet. + if let (Some(fix_stdin), Some(child_stdin)) = (&stdin, child_stdin) { + fix_stdin.attach(child_stdin); + } + let limit = timeout.duration(); let deadline = limit.map(|limit| Instant::now() + limit); let mut stderr_accum = String::new(); @@ -1530,46 +1650,60 @@ mod tests { ); } - /// The paste-an-auth-code shape: the command prompts by blocking on a - /// line read, and the caller feeds the answer through the writer while - /// the fix is running. + /// The paste-an-auth-code shape: the command prompts by blocking on a line + /// read, and the caller feeds the answer through the writer while the fix is + /// running. Sending from inside `on_line` — on the fix's own thread, in + /// response to the prompt the fix printed — pins the send to a moment when + /// the pipe is provably live, so the `Ok` asserted here is the delivery + /// guarantee and not the pre-spawn queueing one. #[tokio::test] async fn run_command_streaming_piped_stdin_feeds_prompt_style_read() { let lines: Arc>> = Arc::new(Mutex::new(Vec::new())); let lines_clone = lines.clone(); + let live_send: Arc>>> = Arc::new(Mutex::new(None)); + let live_send_clone = live_send.clone(); let (writer, stdin) = FixStdin::pipe(); - let handle = tokio::spawn(run_command_streaming( - "read -r line && echo \"got-$line\"".to_string(), + let result = run_command_streaming( + "echo doctor-stdin-prompt; read -r line && echo \"got-$line\"".to_string(), None, Some(stdin), FixTimeout::Standard, move |line| { lines_clone.lock().unwrap().push(line.to_string()); + if line == "doctor-stdin-prompt" { + *live_send_clone.lock().unwrap() = + Some(writer.send_line("doctor-stdin-auth-code")); + } }, - )); - - writer.send_line("doctor-stdin-auth-code").unwrap(); - drop(writer); + ) + .await; - let result = handle.await.unwrap(); assert!(result.is_ok(), "read/echo should exit 0; got {result:?}"); let captured = lines.lock().unwrap().clone(); + let sent = live_send + .lock() + .unwrap() + .take() + .expect("the fix's prompt line should have reached on_line"); + assert!( + sent.is_ok(), + "a send while the fix is live should report delivery; got {sent:?}", + ); assert!( captured.iter().any(|l| l == "got-doctor-stdin-auth-code"), "prompt-style read should see the sent line; captured: {captured:?}", ); } - /// A writer held across the fix's completion must not hang the run — the - /// stdin writer thread is detached, never joined — and the thread must not - /// leak either. It polls `STDIN_WRITER_POLL_INTERVAL` for the finished - /// flag, so within one interval of the fix returning it has retired and - /// dropped the receiver, which makes `send_line` fail cleanly (never - /// panic) rather than queueing into a void. Waiting several intervals - /// keeps this a fixed assertion instead of a poll loop. + /// A writer held across the fix's completion must not hang the run, and the + /// *first* send after it must fail: the runner closes the pipe as it returns, + /// so `Ok` never means "queued for a fix that is already over". That is the + /// berd#99 shape — the login subprocess dies, the user pastes the auth code + /// a beat later — and a host keying off `Ok` would otherwise wait forever + /// with nothing in the log to explain it. #[tokio::test] - async fn run_command_streaming_piped_stdin_reclaims_writer_thread_after_child_exits() { + async fn run_command_streaming_piped_stdin_rejects_sends_once_the_fix_finishes() { let (writer, stdin) = FixStdin::pipe(); let result = run_command_streaming( @@ -1582,11 +1716,38 @@ mod tests { .await; assert!(result.is_ok(), "echo fix should complete; got {result:?}"); + let err = writer + .send_line("late-line") + .expect_err("the first send after the fix finished should fail"); + assert!( + err.contains("no longer accepting input"), + "error should say the input is closed; got {err:?}", + ); + } + + /// `EPIPE` alone can't carry "the fix is over": a backgrounded grandchild + /// inherits the child's stdin and keeps the read end open, so a write into a + /// finished fix's pipe still succeeds. Only the runner's explicit close on + /// the way out makes this send fail. The grandchild's stdout is redirected so + /// it doesn't also hold the reader threads open — this test is about stdin. + #[cfg(unix)] + #[tokio::test] + async fn run_command_streaming_piped_stdin_rejects_sends_when_a_grandchild_holds_the_pipe() { + let (writer, stdin) = FixStdin::pipe(); - tokio::time::sleep(STDIN_WRITER_POLL_INTERVAL * 4).await; + let result = run_command_streaming( + "sleep 2 >/dev/null 2>&1 & echo doctor-stdin-done".to_string(), + None, + Some(stdin), + FixTimeout::Standard, + |_| {}, + ) + .await; + + assert!(result.is_ok(), "echo fix should complete; got {result:?}"); assert!( writer.send_line("late-line").is_err(), - "writer thread should have retired within one poll interval of the fix finishing", + "a grandchild holding the read end must not make a dead fix look writable", ); }