forge_sdk/transport/

process.rs

//! `tokio::process` wrapping of the `claude` binary.
//!
//! ## I/O architecture
//!
//! The subprocess's stdin and stdout each run inside a dedicated
//! tokio task; the [`Subprocess`] surface talks to them over mpsc
//! channels. Two properties fall out:
//!
//! - **Cancel-safe reads.** [`tokio::io::AsyncBufReadExt::read_line`]
//!   is documented as not cancel-safe, but [`mpsc::Receiver::recv`]
//!   is. Driving stdout through a reader task lets callers
//!   `tokio::select!` over [`Subprocess::read_line`] without losing
//!   already-consumed bytes.
//! - **Concurrent writes.** Cloning the writer-side mpsc is cheap
//!   and `Send + 'static`. The `Subprocess::clone_writer` helper
//!   hands out a clonable writer backed by the same writer task,
//!   so detached control-request dispatch can write concurrently
//!   with the reader without serialising on `&mut self`.

use std::process::Stdio;
use std::sync::Arc;

use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::process::{Child, ChildStderr, ChildStdin, ChildStdout, Command};
use tokio::sync::{mpsc, oneshot};
use tokio::task::JoinHandle;
use tracing::{debug, warn};

use crate::Error;
use crate::argv::build_args;
use crate::options::{Options, WireTee};

/// Default upper bound on `close()` wait-for-exit. After this elapses,
/// the child is SIGKILL'd. 5s is generous for a CLI that's draining
/// in-flight turns; tests that want to verify the kill path drive
/// against a process that ignores stdin EOF. Public so a consumer
/// timing a wait over a session's teardown can derive it from this
/// budget rather than guess.
pub const CLOSE_WAIT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);

/// Name what actually failed. A spawn that cannot find the program and
/// one that cannot enter `current_dir` both surface as `ENOENT`, so the
/// cwd is re-checked here - this is the only point that holds it.
fn spawn_error(err: std::io::Error, options: &Options) -> Error {
    if err.kind() != std::io::ErrorKind::NotFound {
        return Error::Io(err);
    }
    match options.cwd.as_deref() {
        Some(cwd) if !cwd.is_dir() => {
            Error::CwdNotFound { path: cwd.to_string_lossy().into_owned() }
        }
        _ => Error::CliNotFound { binary: options.binary.clone() },
    }
}

/// Run `<binary> --version` synchronously and return the stdout.
///
/// # Errors
///
/// [`Error::CliNotFound`] when the binary isn't on PATH; [`Error::Io`]
/// on any other spawn/wait failure; [`Error::Process`] when the version
/// probe exits non-zero.
pub fn query_cli_version(binary: &str) -> Result<String, Error> {
    let output =
        std::process::Command::new(binary).arg("--version").output().map_err(|e| {
            match e.kind() {
                std::io::ErrorKind::NotFound => Error::CliNotFound { binary: binary.to_string() },
                _ => Error::Io(e),
            }
        })?;
    if !output.status.success() {
        return Err(Error::Process {
            exit_code: output.status.code(),
            stderr: String::from_utf8_lossy(&output.stderr).into_owned(),
        });
    }
    Ok(String::from_utf8_lossy(&output.stdout).trim().to_string())
}

/// Check that the reported `claude` version is at least `min_version`.
/// Compares the parsed (major, minor, patch) components numerically.
///
/// # Errors
///
/// [`Error::Connection`] when the reported version is below the minimum
/// or can't be parsed.
pub fn check_cli_version(reported: &str, min_version: &str) -> Result<(), Error> {
    // `reported` is typically like `2.1.116 (anthropic)` or `claude 2.1.116`.
    let token = reported
        .split_whitespace()
        .find(|t| t.chars().next().is_some_and(|c| c.is_ascii_digit()))
        .ok_or_else(|| Error::Connection {
            reason: format!("could not parse claude version from: {reported}"),
        })?;
    let reported_triple = parse_semver_triple(token).ok_or_else(|| Error::Connection {
        reason: format!("could not parse semver triple from: {token}"),
    })?;
    let min_triple = parse_semver_triple(min_version).ok_or_else(|| Error::Connection {
        reason: format!("could not parse minimum semver triple from: {min_version}"),
    })?;
    if reported_triple < min_triple {
        return Err(Error::Connection {
            reason: format!("claude CLI version {reported} below minimum required {min_version}"),
        });
    }
    Ok(())
}

/// Parse `"<major>.<minor>.<patch>"` (any trailing non-numeric suffix
/// ignored). Missing minor/patch default to 0.
fn parse_semver_triple(s: &str) -> Option<(u32, u32, u32)> {
    let mut parts = s.split('.');
    let major: u32 = parts.next()?.parse().ok()?;
    let minor: u32 = parts.next().and_then(|p| p.parse().ok()).unwrap_or(0);
    let patch_str = parts.next().unwrap_or("0");
    // Strip any non-digit suffix (e.g. "117-rc1" or "117 (build)").
    let patch_digits: String = patch_str.chars().take_while(char::is_ascii_digit).collect();
    let patch: u32 = if patch_digits.is_empty() { 0 } else { patch_digits.parse().ok()? };
    Some((major, minor, patch))
}

/// Every flag [`build_args`] can emit. A user-defined `extra_args`
/// name is not covered, except where it collides with one of these -
/// forge routes `--effort` that way, so the collision is the common
/// case rather than a corner.
const LOGGABLE_FLAGS: [&str; 16] = [
    "output-format",
    "verbose",
    "system-prompt",
    "system-prompt-file",
    "append-system-prompt",
    "allowedTools",
    "max-turns",
    "model",
    "permission-prompt-tool",
    "permission-mode",
    "resume",
    "session-id",
    "mcp-config",
    "plugin-dir",
    "effort",
    "input-format",
];

/// The recognised flag names in an argv, and a count of the `--`
/// tokens that are neither recognised nor a declared `extra_args`
/// name.
///
/// Matching against [`LOGGABLE_FLAGS`] rather than on a `--` prefix is
/// what stops a value reaching the log when it happens to look like a
/// flag - `--mcp-config` alone carries every external MCP server's
/// `env` and `headers` verbatim.
///
/// The count exists to make a silent omission loud, so it reads zero
/// on every shape forge builds today. `extra_args` names are excluded
/// because their own count reports them, which leaves this number
/// meaning "a `--` token nothing accounts for". A repeated recognised
/// flag also raises it - no production path emits one.
fn flag_names(
    args: &[String],
    extra_args: &std::collections::HashMap<String, Option<String>>,
) -> (String, usize) {
    let mut names: Vec<&str> = Vec::new();
    let mut dropped = 0;
    for token in args {
        let Some(name) = token.strip_prefix("--") else { continue };
        let name = name.split_once('=').map_or(name, |(before, _)| before);
        if LOGGABLE_FLAGS.contains(&name) && !names.contains(&name) {
            names.push(name);
        } else if !extra_args.contains_key(name) {
            dropped += 1;
        }
    }
    (names.join(" "), dropped)
}

/// How long one write to the CLI's stdin, and its ack, may take before the
/// write fails.
///
/// **A write had no bound.** A child that stopped reading its stdin - a
/// wedged CLI, a busy one, a full pipe - left the writer task parked in
/// `write_all` and every caller waiting on its ack for ever. The hook
/// callback that writes a control response is one of those callers, so a
/// backed-up child hung the server silently: no error, no panic, no reap.
/// Thirty seconds is longer than a pipe write to a reading child ever takes
/// and far shorter than the 26.7 minutes the 2026-11-09 freeze went
/// unnoticed.
pub(crate) const STDIN_WRITE_BOUND: std::time::Duration = std::time::Duration::from_secs(30);

/// A stdin write, or its ack, that outlived [`STDIN_WRITE_BOUND`].
fn write_stalled(bound: std::time::Duration, what: &str) -> Error {
    Error::Io(std::io::Error::new(
        std::io::ErrorKind::TimedOut,
        format!("the CLI stopped reading stdin: {what} did not complete within {bound:?}"),
    ))
}

/// Why the writer task's stdin is gone, kept beside it so a later write names
/// the cause rather than blaming a call that never ran.
#[derive(Clone, Copy, Debug)]
enum Closed {
    /// `end_input` dropped it on purpose.
    EndInput,
    /// A write outlived [`STDIN_WRITE_BOUND`], so a partial line may be on
    /// the wire.
    Stalled,
    /// A write failed at the pipe.
    Failed,
}

impl Closed {
    fn why(self) -> &'static str {
        match self {
            Self::EndInput => "stdin already closed (end_input)",
            Self::Stalled => "stdin was closed because a write outlived its bound",
            Self::Failed => "stdin was closed because a write failed",
        }
    }
}

/// The error a write gets once the stream is closed, naming why it closed.
fn closed_stdin(reason: Option<Closed>) -> Error {
    Error::Io(std::io::Error::new(
        std::io::ErrorKind::BrokenPipe,
        reason.map_or("stdin already closed", Closed::why),
    ))
}

/// Outcome of one writer-task operation. Sent back over a oneshot the
/// caller provides.
type IoAck = Result<(), Error>;

/// A write-side command dispatched to the writer task.
#[derive(Debug)]
enum WriterCmd {
    /// Append `line` to stdin and flush.
    Write(String, oneshot::Sender<IoAck>),
    /// Drop stdin so the CLI sees EOF.
    EndInput(oneshot::Sender<IoAck>),
}

/// A live `claude` subprocess driven over channels.
///
/// Spawned via [`Subprocess::spawn`]. Every read goes through the
/// reader task → mpsc; every write goes through the writer task ←
/// mpsc. Drop hits `kill_on_drop` cleanup; [`close`](Self::close)
/// gives a graceful exit path with a 5s timeout before SIGKILL.
pub struct Subprocess {
    /// Outbound channel into the writer task. Cloned by
    /// [`clone_writer`](Self::clone_writer) so external
    /// dispatchers can write without contending on `&mut self`.
    writer_tx: mpsc::UnboundedSender<WriterCmd>,
    /// How long one write to stdin may take. The shipping bound is
    /// [`STDIN_WRITE_BOUND`]; a test injects a shorter one so it can watch
    /// the bound fire without waiting thirty seconds out.
    ///
    /// **A caller waits twice this** ([`Subprocess::ack_wait`]), because the
    /// task's verdict is the authoritative one for a line it dequeued: a
    /// writer that started a line and timed out reports it before the caller
    /// gives up, so no line is written after its caller was told it failed.
    /// A line still queued behind two slow *successful* writes can outlast
    /// even that wait; its caller is told the wait's own bound, not the
    /// writer's.
    write_bound: std::time::Duration,
    /// Inbound channel from the reader task. Single-consumer.
    reader_rx: mpsc::UnboundedReceiver<Result<Option<String>, Error>>,
    /// Writer task handle. [`close`](Self::close) aborts this so the
    /// child sees stdin EOF promptly even when external writer clones
    /// still hold `writer_tx` clones - without the abort the writer
    /// task would wait for every clone to drop and the child wouldn't
    /// see EOF on stdin until then.
    writer_task: Option<JoinHandle<()>>,
    /// Stderr drain task - best-effort logged. Joined during
    /// [`close`](Self::close), which takes its returned tail for
    /// [`Error::Process`].
    stderr_task: Option<JoinHandle<String>>,
    /// The child handle. [`close`](Self::close) waits on it (with
    /// timeout) and SIGKILLs on hang.
    child: Option<Child>,
    /// The child's process group, when this transport put it in one of
    /// its own. `None` means no group was claimed and teardown must not
    /// signal one.
    pgid: Option<u32>,
    /// The claude session this child carries, as the container the reader
    /// and the dispatcher fill. **Shared, not copied**: `/new` swaps the
    /// occupant without swapping the process, so a label read at spawn names
    /// a session the record is not about - and the long-lived session is
    /// exactly where a stall happens.
    session: Arc<parking_lot::RwLock<String>>,
    /// Idempotency guard for [`close`](Self::close).
    closed: bool,
}

impl std::fmt::Debug for Subprocess {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Subprocess")
            .field("child", &self.child.is_some())
            .field("closed", &self.closed)
            .finish_non_exhaustive()
    }
}

impl Subprocess {
    /// Spawn the `claude` binary with stream-json flags.
    ///
    /// # Errors
    ///
    /// - [`Error::CliNotFound`] when the binary isn't on PATH or the given path
    ///   doesn't exist.
    /// - [`Error::CwdNotFound`] when `cwd` names a directory that isn't there.
    /// - [`Error::Io`] for other spawn failures.
    pub async fn spawn(options: &Options) -> Result<Self, Error> {
        // Optional CLI-version guard. Runs `<binary> --version` once.
        // The caller asked for a floor - if the probe fails, surface it
        // rather than silently skipping (they won't know the check was
        // bypassed otherwise).
        if let Some(min) = &options.minimum_cli_version {
            // `claude --version` is a fork+exec; wrap in spawn_blocking
            // so the tokio worker thread isn't parked during the probe.
            let binary = options.binary.clone();
            let reported = tokio::task::spawn_blocking(move || query_cli_version(&binary))
                .await
                .map_err(|e| Error::Connection {
                    reason: format!("version probe join failed: {e}"),
                })?
                .map_err(|e| Error::Connection {
                    reason: format!("minimum_cli_version set but --version probe failed: {e}"),
                })?;
            check_cli_version(&reported, min)?;
        }
        let args = build_args(options)?;
        // Seeded with what the caller knows now; the reader and the
        // dispatcher keep writing it as the occupant changes.
        let session = Arc::new(parking_lot::RwLock::new(
            options.session_id.clone().or_else(|| options.resume.clone()).unwrap_or_default(),
        ));
        let mut cmd = Command::new(&options.binary);
        cmd.args(&args);
        if let Some(cwd) = &options.cwd {
            cmd.current_dir(cwd);
        }

        // Env setup. Order matters because later writes win:
        //
        // 1. Start from parent env, filtering out `CLAUDECODE` so SDK-
        //    spawned subprocesses don't think they're nested inside a
        //    Claude Code parent (upstream issue #573).
        // 2. Clear any inherited `CLAUDE_CODE_ENTRYPOINT` so the child
        //    self-classifies instead of reporting a stamp leaked from
        //    the parent shell (e.g. running forge inside a Claude Code
        //    session that itself set `sdk-rs`). Callers that genuinely
        //    want to preset the entrypoint can still do so via
        //    `options.env` - those writes happen after this removal.
        // 3. Let `options.env` override anything above, EXCEPT
        //    `CLAUDE_AGENT_SDK_VERSION` - that one we always stamp last.
        // 4. Stamp `CLAUDE_AGENT_SDK_VERSION` as the final write.
        // 5. Set `PWD` to the chosen cwd when present.
        cmd.env_remove("CLAUDECODE");
        cmd.env_remove("CLAUDE_CODE_ENTRYPOINT");
        for (k, v) in &options.env {
            cmd.env(k, v);
        }
        cmd.env("CLAUDE_AGENT_SDK_VERSION", env!("CARGO_PKG_VERSION"));
        if let Some(cwd) = &options.cwd {
            cmd.env("PWD", cwd);
        }

        cmd.stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::piped());
        cmd.kill_on_drop(true);
        // Its own process group, so teardown can take down whatever the
        // session started: a descendant otherwise outlives the child's
        // death.
        #[cfg(unix)]
        cmd.process_group(0);

        let (flags, flags_dropped) = flag_names(&args, &options.extra_args);
        debug!(
            binary = %options.binary,
            cwd = ?options.cwd,
            %flags,
            flags_dropped,
            extra_args_len = options.extra_args.len(),
            "spawning claude subprocess"
        );
        let mut child = cmd.spawn().map_err(|e| spawn_error(e, options))?;
        // `process_group(0)` makes the child its own group leader, so the
        // group id is its pid. Kept rather than re-read in `close`: a
        // completed wait reaps the child and drops the id.
        #[cfg(unix)]
        let pgid = child.id();
        #[cfg(not(unix))]
        let pgid: Option<u32> = None;

        let stdin = child
            .stdin
            .take()
            .ok_or_else(|| Error::Connection { reason: "stdin pipe missing".into() })?;
        let stdout = child
            .stdout
            .take()
            .ok_or_else(|| Error::Connection { reason: "stdout pipe missing".into() })?;
        let stderr = child
            .stderr
            .take()
            .ok_or_else(|| Error::Connection { reason: "stderr pipe missing".into() })?;

        let stderr_task = tokio::spawn(tracing::Instrument::instrument(
            drain_stderr(stderr),
            tracing::info_span!("forge_sdk::stderr_reader"),
        ));

        let (reader_tx, reader_rx) = mpsc::unbounded_channel();
        spawn_reader_task(stdout, options.tee_inbound.clone(), reader_tx);

        let (writer_tx, writer_rx) = mpsc::unbounded_channel();
        let writer_task = spawn_writer_task(
            stdin,
            options.tee_outbound.clone(),
            writer_rx,
            STDIN_WRITE_BOUND,
            session.clone(),
        );

        Ok(Self {
            writer_tx,
            write_bound: STDIN_WRITE_BOUND,
            reader_rx,
            writer_task: Some(writer_task),
            stderr_task: Some(stderr_task),
            child: Some(child),
            pgid,
            session,
            closed: false,
        })
    }

    /// Read one line (without the trailing `\n`) from the subprocess stdout.
    ///
    /// Operating-system PID of the spawned `claude` child, when one is
    /// still attached. Returns `None` after [`close`](Self::close) or
    /// when the child has already been reaped.
    ///
    /// Used by the Inspector pane's PROCESSES section to anchor an
    /// OS-level walk of the descendant tree (the `forge-agent`
    /// crate's `env::processes` scanner - not linked here because the
    /// dependency direction forbids forge-sdk from naming forge-agent
    /// types). The PID is stable for the lifetime of the subprocess
    /// so the scanner can cache its snapshot across polls keyed off
    /// this value.
    pub fn child_pid(&self) -> Option<u32> {
        self.child.as_ref().and_then(tokio::process::Child::id)
    }

    /// The session this child carries, as the container the reader and the
    /// dispatcher fill: a consumer that records a problem about this child
    /// reads it at the failure rather than at the spawn, because `/new` swaps
    /// the occupant.
    pub(crate) fn session_id(&self) -> Arc<parking_lot::RwLock<String>> {
        Arc::clone(&self.session)
    }

    /// Returns `Ok(None)` at end-of-stream. Cancel-safe.
    ///
    /// # Errors
    ///
    /// [`Error::Io`] on read failure.
    pub async fn read_line(&mut self) -> Result<Option<String>, Error> {
        // Reader task signals EOF by either sending `Ok(None)` or by
        // closing the channel - both collapse to `Ok(None)` here.
        self.reader_rx.recv().await.unwrap_or(Ok(None))
    }

    /// How long a caller waits for a write's ack: **twice the writer's own
    /// bound**, so the task's verdict lands first for a line it dequeued and
    /// no line is written after its caller was told it failed.
    fn ack_wait(&self) -> std::time::Duration {
        self.write_bound * 2
    }

    /// Write one line of stream-json to the subprocess stdin.
    ///
    /// The caller is responsible for including the trailing `\n` - this
    /// matches the contract of
    /// [`codec::encode_user_prompt`](super::codec::encode_user_prompt).
    ///
    /// # Errors
    ///
    /// [`Error::Io`] on write failure, including writes issued after
    /// [`end_input`](Self::end_input) or [`close`](Self::close).
    pub async fn write_line(&mut self, line: &str) -> Result<(), Error> {
        let (ack_tx, ack_rx) = oneshot::channel();
        self.writer_tx.send(WriterCmd::Write(line.to_owned(), ack_tx)).map_err(|_| {
            Error::Io(std::io::Error::new(
                std::io::ErrorKind::BrokenPipe,
                "Subprocess writer task gone",
            ))
        })?;
        match tokio::time::timeout(self.ack_wait(), ack_rx).await {
            Ok(Ok(result)) => result,
            Ok(Err(_)) => Err(Error::Io(std::io::Error::new(
                std::io::ErrorKind::BrokenPipe,
                "Subprocess writer task dropped ack",
            ))),
            Err(_) => {
                Err(write_stalled(self.ack_wait(), "the caller's wait for a stdin write's ack"))
            }
        }
    }

    /// Close the subprocess's stdin. Safe to call multiple times.
    /// Subsequent [`write_line`](Self::write_line) calls will fail
    /// with a `BrokenPipe` `Io` error.
    ///
    /// # Errors
    ///
    /// [`Error::Io`] on flush failure.
    pub async fn end_input(&mut self) -> Result<(), Error> {
        let (ack_tx, ack_rx) = oneshot::channel();
        // Writer task already gone (e.g. previous end_input + drop) →
        // treat as no-op, matching the old direct-stdin behaviour.
        if self.writer_tx.send(WriterCmd::EndInput(ack_tx)).is_err() {
            return Ok(());
        }
        match tokio::time::timeout(self.ack_wait(), ack_rx).await {
            Ok(Ok(result)) => result,
            Ok(Err(_)) => {
                warn!("Subprocess::end_input: ack channel dropped");
                Ok(())
            }
            Err(_) => {
                Err(write_stalled(self.ack_wait(), "the caller's wait for the end-input ack"))
            }
        }
    }

    /// Hand out a clonable [`SharedWriter`] backed by the same writer
    /// task. Multiple clones can write concurrently; the writer task
    /// serialises onto the child's stdin in arrival order. Used by
    /// detached control-request dispatch so a slow callback can't
    /// block the reader / command loop.
    pub(crate) fn clone_writer(&self) -> Arc<SharedWriter> {
        Arc::new(SharedWriter { writer_tx: self.writer_tx.clone(), bound: self.write_bound })
    }

    /// Graceful shutdown: close stdin, wait for the subprocess to
    /// exit (5s timeout, SIGKILL fallback), SIGKILL whatever it left
    /// in its process group, drain the stderr task. Idempotent -
    /// subsequent calls are no-ops.
    ///
    /// # Errors
    ///
    /// [`Error::Process`] when the subprocess exits non-zero or hits
    /// the SIGKILL fallback; [`Error::Io`] on wait failure.
    pub async fn close(&mut self) -> Result<(), Error> {
        if self.closed {
            return Ok(());
        }
        self.closed = true;

        // Drop our held writer_tx and abort the writer task. The abort
        // forces stdin closure even if external `SharedWriter` clones
        // (handed out via `clone_writer`) still hold writer_tx
        // clones - without the abort the writer task would wait for
        // every clone to drop. In-flight write_line acks on cloned
        // writers will resolve with the ack-channel-dropped error,
        // which is correct semantics for a closed transport.
        let (closed_tx, closed_rx) = mpsc::unbounded_channel();
        self.writer_tx = closed_tx;
        // Drop the placeholder receiver immediately so any concurrent
        // SharedWriter clone trying to send on `writer_tx` fails fast
        // with BrokenPipe rather than queuing into a dead channel and
        // hanging for the duration of the close grace period.
        drop(closed_rx);
        if let Some(handle) = self.writer_task.take() {
            handle.abort();
            // Symmetric drain with the stderr task below. abort() then
            // await - the future returns JoinError::Cancelled which is
            // expected; surface panic JoinErrors at debug so the
            // tokio default panic handler isn't the only path that
            // notices.
            if let Err(e) = handle.await
                && !e.is_cancelled()
            {
                debug!(error = %e, "writer task ended abnormally");
            }
        }

        // Wait for child exit, with a SIGKILL timeout so a stuck CLI
        // doesn't pin the disconnect path forever.
        if let Some(mut child) = self.child.take() {
            let waited = tokio::time::timeout(CLOSE_WAIT_TIMEOUT, child.wait()).await;
            // Whatever the child left running in its group goes down with
            // it: a descendant that survived the child is not the
            // session's to leave behind. An empty group is the ordinary
            // answer and stays quiet; anything else leaves processes
            // forge meant to kill, so it is forge's own problem.
            if let Some(pgid) = self.pgid
                && let Some(error) = kill_group(pgid)
            {
                warn!(
                    target: "forge_sdk",
                    event_name = "subprocess_group_signal_failed",
                    session = %self.session.read().as_str(),
                    pgid,
                    error = %error,
                    "the session child's process group could not be signalled; \
                     a descendant may outlive the session"
                );
            }
            if waited.is_err() {
                warn!("Subprocess::close timed out waiting for child; sending SIGKILL");
                if let Err(e) = child.kill().await {
                    warn!(error = %e, "Subprocess::close: kill() failed");
                }
            }
            // Release the child before joining, so `kill_on_drop` still
            // fires on the wait-failed path.
            drop(child);
            // Join before building the error rather than reading a shared
            // buffer at construction, which would race the last line.
            let stderr = label_stderr_tail(&self.join_stderr_tail().await);
            match waited {
                Ok(Ok(status)) if status.success() => Ok(()),
                Ok(Ok(status)) => Err(Error::Process { exit_code: status.code(), stderr }),
                Ok(Err(e)) => Err(Error::Io(e)),
                Err(_elapsed) => {
                    let mut reason = String::from("close timeout - child killed");
                    if !stderr.is_empty() {
                        reason.push('\n');
                        reason.push_str(&stderr);
                    }
                    Err(Error::Process { exit_code: None, stderr: reason })
                }
            }
        } else {
            self.join_stderr_tail().await;
            Ok(())
        }
    }

    /// Join the stderr drain task and take its retained tail. Errors
    /// are best-effort - it's a logging sink, so a dead task costs the
    /// tail rather than the close.
    async fn join_stderr_tail(&mut self) -> String {
        let Some(task) = self.stderr_task.take() else { return String::new() };
        match task.await {
            Ok(tail) => tail,
            Err(e) => {
                debug!(error = %e, "stderr drain task ended abnormally");
                String::new()
            }
        }
    }
}

/// Cloneable writer half of [`Subprocess`] - pushes onto the writer
/// task's mpsc. Multiple clones can write concurrently; the writer
/// task serialises onto the child's stdin in arrival order.
///
/// Returned by [`Subprocess::clone_writer`]. Used by detached
/// `control_request` dispatch.
#[derive(Debug, Clone)]
pub(crate) struct SharedWriter {
    writer_tx: mpsc::UnboundedSender<WriterCmd>,
    /// The writer task's own bound, so this side can derive the wait that
    /// outlasts it (see [`SharedWriter::ack_wait`]).
    bound: std::time::Duration,
}

impl SharedWriter {
    /// How long a caller waits for a write's ack: **twice the writer's own
    /// bound**, so the task's verdict lands first for a line it dequeued and
    /// no line is written after its caller was told it failed.
    fn ack_wait(&self) -> std::time::Duration {
        self.bound * 2
    }
}

#[cfg(test)]
impl SharedWriter {
    /// Test stub: written lines land on the returned receiver; every
    /// write is acked `Ok`.
    pub(crate) fn test_stub() -> (Self, mpsc::UnboundedReceiver<String>) {
        let (writer_tx, mut cmd_rx) = mpsc::unbounded_channel();
        let (line_tx, line_rx) = mpsc::unbounded_channel();
        tokio::spawn(async move {
            while let Some(cmd) = cmd_rx.recv().await {
                match cmd {
                    WriterCmd::Write(line, ack) => {
                        let _ = line_tx.send(line);
                        let _ = ack.send(Ok(()));
                    }
                    WriterCmd::EndInput(ack) => {
                        let _ = ack.send(Ok(()));
                    }
                }
            }
        });
        (Self { writer_tx, bound: STDIN_WRITE_BOUND }, line_rx)
    }
}

impl SharedWriter {
    /// Write one line of stream-json to the transport. Caller supplies
    /// the trailing `\n`.
    ///
    /// # Errors
    ///
    /// [`Error::Io`] on write failure or after the transport has
    /// closed its write half.
    pub(crate) async fn write_line(&self, line: &str) -> Result<(), Error> {
        let (ack_tx, ack_rx) = oneshot::channel();
        self.writer_tx.send(WriterCmd::Write(line.to_owned(), ack_tx)).map_err(|_| {
            Error::Io(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "SharedWriter task gone"))
        })?;
        // **The ack can be as slow as the write ahead of it**, and a writer
        // parked on a full pipe never sends one: an unbounded wait here hangs
        // the caller with the task still alive, which is the shape that went
        // unnoticed for 26.7 minutes.
        match tokio::time::timeout(self.ack_wait(), ack_rx).await {
            Ok(Ok(result)) => result,
            Ok(Err(_)) => Err(Error::Io(std::io::Error::new(
                std::io::ErrorKind::BrokenPipe,
                "SharedWriter task dropped ack",
            ))),
            Err(_) => {
                Err(write_stalled(self.ack_wait(), "the caller's wait for a stdin write's ack"))
            }
        }
    }

    /// Close the write half so the remote sees EOF on stdin. Idempotent.
    ///
    /// # Errors
    ///
    /// [`Error::Io`] on flush failure.
    pub(crate) async fn end_input(&self) -> Result<(), Error> {
        let (ack_tx, ack_rx) = oneshot::channel();
        // Writer task already exited (e.g. previous end_input or close)
        // → treat as no-op, matching `Subprocess::end_input`.
        if self.writer_tx.send(WriterCmd::EndInput(ack_tx)).is_err() {
            return Ok(());
        }
        // The same bound as a write: a command queued behind a stalled write
        // is an ack that never comes.
        match tokio::time::timeout(self.ack_wait(), ack_rx).await {
            Ok(Ok(result)) => result,
            Ok(Err(_)) => {
                warn!("SharedWriter::end_input: ack channel dropped");
                Ok(())
            }
            Err(_) => {
                Err(write_stalled(self.ack_wait(), "the caller's wait for the end-input ack"))
            }
        }
    }
}

fn spawn_reader_task(
    stdout: ChildStdout,
    tee: Option<WireTee>,
    tx: mpsc::UnboundedSender<Result<Option<String>, Error>>,
) {
    use tracing::Instrument;
    let span = tracing::info_span!("forge_sdk::stdout_reader");
    tokio::spawn(
        async move {
            let mut reader = BufReader::new(stdout);
            let mut buf = String::new();
            loop {
                buf.clear();
                match reader.read_line(&mut buf).await {
                    Ok(0) => {
                        // EOF - signal end-of-stream and exit.
                        let _ = tx.send(Ok(None));
                        break;
                    }
                    Ok(_) => {
                        let mut line = std::mem::take(&mut buf);
                        while matches!(line.chars().last(), Some('\n' | '\r')) {
                            line.pop();
                        }
                        if let Some(cb) = tee.as_ref() {
                            cb(&line);
                        }
                        if tx.send(Ok(Some(line))).is_err() {
                            // Receiver gone - caller dropped the transport.
                            break;
                        }
                    }
                    Err(e) => {
                        if tx.send(Err(Error::Io(e))).is_err() {
                            tracing::warn!(
                                target: "forge_sdk::transport",
                                "subprocess stdout I/O error after caller dropped reader"
                            );
                        }
                        break;
                    }
                }
            }
        }
        .instrument(span),
    );
}

/// Write one line and flush it, as the writer task's unit of work.
async fn write_one(stdin: &mut ChildStdin, line: &[u8]) -> std::io::Result<()> {
    stdin.write_all(line).await?;
    stdin.flush().await
}

fn spawn_writer_task(
    stdin: ChildStdin,
    tee: Option<WireTee>,
    mut rx: mpsc::UnboundedReceiver<WriterCmd>,
    write_bound: std::time::Duration,
    session: Arc<parking_lot::RwLock<String>>,
) -> JoinHandle<()> {
    use tracing::Instrument;
    let span = tracing::info_span!("forge_sdk::stdin_writer");
    tokio::spawn(
        async move {
            let mut stdin = Some(stdin);
            // Kept beside the handle so a write after the closure names the
            // cause. Without it every later failure blamed `end_input`, a call
            // that never ran.
            let mut closed: Option<Closed> = None;
            while let Some(cmd) = rx.recv().await {
                match cmd {
                    WriterCmd::Write(line, ack) => {
                        if let Some(cb) = tee.as_ref() {
                            cb(line.trim_end_matches('\n'));
                        }
                        let result = if let Some(s) = stdin.as_mut() {
                            match tokio::time::timeout(write_bound, write_one(s, line.as_bytes()))
                                .await
                            {
                                Ok(Ok(())) => Ok(()),
                                Ok(Err(e)) => {
                                    closed = Some(Closed::Failed);
                                    Err(Error::Io(e))
                                }
                                Err(_) => {
                                    // **The one record a stall gets**, and the
                                    // only one that can name the session: the
                                    // CLI stopped reading and this task is
                                    // parked in a write no caller can see.
                                    tracing::warn!(
                                        target: "forge_sdk",
                                        event_name = "stdin_write_stalled",
                                        session = %session.read().as_str(),
                                        bound_ms = write_bound.as_millis(),
                                        "the CLI stopped reading stdin: a write did not complete \
                                         within the bound, so the stream is closed",
                                    );
                                    closed = Some(Closed::Stalled);
                                    Err(write_stalled(write_bound, "a stdin write"))
                                }
                            }
                        } else {
                            Err(closed_stdin(closed))
                        };
                        if closed.is_some() {
                            // **A closed stream stays closed.** A write that
                            // timed out may have left a partial line on the
                            // wire, and one that failed left a broken pipe;
                            // either way another line cannot follow. Dropping
                            // stdin also lets the child see EOF instead of
                            // waiting for bytes that will not come.
                            stdin = None;
                        }
                        let _ = ack.send(result);
                    }
                    WriterCmd::EndInput(ack) => {
                        // **A stall keeps its reason**: an end-input after a
                        // stalled write must not relabel the stream as a
                        // deliberate close.
                        if closed.is_none() {
                            closed = Some(Closed::EndInput);
                        }
                        drop(stdin.take());
                        let _ = ack.send(Ok(()));
                    }
                }
            }
        }
        .instrument(span),
    )
}

/// Prefix the retained tail so it reads as the last thing the CLI
/// wrote rather than as the cause. Nothing here can tell whether it is
/// causal: a session whose only stderr was a startup warning attaches
/// that warning to an exit hours later.
fn label_stderr_tail(tail: &str) -> String {
    if tail.is_empty() { String::new() } else { format!("last stderr from the CLI: {tail}") }
}

/// Upper bound on the drained stderr retained for [`Error::Process`].
/// Large enough for a CLI failure message and its context, small
/// enough that a chatty child can't pin unbounded memory for the life
/// of the process.
const STDERR_TAIL_CAP: usize = 8 * 1024;

/// The most recent drained stderr lines, capped at [`STDERR_TAIL_CAP`]
/// bytes. Oldest lines are dropped first: a failing CLI puts its
/// reason last.
#[derive(Default)]
struct StderrTail {
    lines: std::collections::VecDeque<String>,
    bytes: usize,
}

impl StderrTail {
    fn push(&mut self, line: &str) {
        // Keep the end of an over-cap line, for the same reason we evict
        // oldest-first.
        let mut start = line.len().saturating_sub(STDERR_TAIL_CAP);
        while !line.is_char_boundary(start) {
            start += 1;
        }
        let kept = &line[start..];
        self.bytes += kept.len() + 1;
        self.lines.push_back(kept.to_owned());
        // Never evict the newest line: on a one-line overflow it is the
        // whole message.
        while self.bytes > STDERR_TAIL_CAP && self.lines.len() > 1 {
            let Some(dropped) = self.lines.pop_front() else { break };
            self.bytes -= dropped.len() + 1;
        }
    }

    fn into_string(self) -> String {
        Vec::from(self.lines).join("\n")
    }
}

/// SIGKILL every process left in `pgid`, reporting a failure a caller
/// should tell someone about.
///
/// `None` covers the ordinary answers: the signal landed, or the group
/// was already empty (`ESRCH`), which is what a clean exit leaves.
#[cfg(unix)]
pub(crate) fn kill_group(pgid: u32) -> Option<std::io::Error> {
    use rustix::io::Errno;
    use rustix::process::{Pid, Signal, kill_process_group};

    let Ok(raw) = i32::try_from(pgid) else {
        debug!(pgid, "process group id does not fit in i32; nothing signalled");
        return None;
    };
    let Some(pgid) = Pid::from_raw(raw) else {
        debug!(raw, "process group id names no process; nothing signalled");
        return None;
    };
    match kill_process_group(pgid, Signal::KILL) {
        Ok(()) | Err(Errno::SRCH) => None,
        Err(error) => Some(std::io::Error::from_raw_os_error(error.raw_os_error())),
    }
}

#[cfg(not(unix))]
fn kill_group(_pgid: u32) -> Option<std::io::Error> {
    None
}

/// Background drain for the subprocess stderr pipe. Reads lines as UTF-8
/// (lossy on invalid bytes) and logs each, so the pipe never blocks.
///
/// Returns the tail of what it drained, which [`Subprocess::close`]
/// attaches to [`Error::Process`] on a non-zero exit.
async fn drain_stderr(stderr: ChildStderr) -> String {
    let mut reader = BufReader::new(stderr);
    let mut raw = Vec::new();
    let mut tail = StderrTail::default();
    loop {
        raw.clear();
        // Bytes, not `read_line`: that returns InvalidData on non-UTF-8
        // and would end the drain at the first bad byte, losing every
        // line after it.
        let n = match reader.read_until(b'\n', &mut raw).await {
            Ok(n) => n,
            Err(e) => {
                debug!(?e, "stderr read failed");
                break;
            }
        };
        if n == 0 {
            break;
        }
        let mut line = String::from_utf8_lossy(&raw).into_owned();
        while matches!(line.chars().last(), Some('\n' | '\r')) {
            line.pop();
        }
        tail.push(&line);
        if line.starts_with("ERROR") || line.starts_with("Error") {
            tracing::warn!(target: "forge_sdk::stderr", line = %line, "claude stderr");
        } else {
            tracing::info!(target: "forge_sdk::stderr", line = %line, "claude stderr");
        }
    }
    tail.into_string()
}

#[cfg(test)]
mod tests {

    use super::*;
    use std::time::Duration;

    /// Every field of every record, formatted, in one buffer.
    #[derive(Clone, Default)]
    struct FieldCapture(Arc<std::sync::Mutex<String>>);

    impl FieldCapture {
        fn text(&self) -> String {
            self.0.lock().expect("capture").clone()
        }
    }

    struct AllFields<'a>(&'a mut String);

    impl tracing::field::Visit for AllFields<'_> {
        // `record_str` forwards here, so every field arrives through
        // this one arm however the macro wrote it.
        fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
            use std::fmt::Write;
            let _ = write!(self.0, " {}={value:?}", field.name());
        }
    }

    impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for FieldCapture {
        fn on_event(
            &self,
            event: &tracing::Event<'_>,
            _ctx: tracing_subscriber::layer::Context<'_, S>,
        ) {
            let mut buf = self.0.lock().expect("capture");
            buf.push_str(event.metadata().target());
            event.record(&mut AllFields(&mut buf));
            buf.push('\n');
        }
    }

    /// An `Options` carrying a sentinel on every surface that reaches
    /// argv or the child env with a value: the env map, all three
    /// external MCP transports, `extra_args`, and the system prompt.
    /// `Preset { append }` is the shape forge builds.
    fn options_with_secrets() -> Options {
        let mut options = crate::OptionsBuilder::new()
            // Never spawns. The record under test is written before
            // `cmd.spawn()`, so a missing binary still exercises it.
            .binary("/nonexistent/claude-for-this-test")
            .model("claude-test")
            .resume("session-abc")
            .system_prompt(crate::SystemPromptKind::preset_append(
                "sentinel-append-prompt-must-never-be-logged",
            ))
            .env("ANTHROPIC_AUTH_TOKEN", "sentinel-auth-must-never-be-logged")
            .env("AIRMAIL_TOKEN", "sentinel-airmail-must-never-be-logged")
            .extra_arg("some-flag", Some("sentinel-extra-arg-must-never-be-logged".to_owned()))
            // A value shaped like a flag. Anything keying on a `--`
            // prefix rather than on a known name renders this one.
            .extra_arg("odd-flag", Some("--sentinel-dashed-must-never-be-logged".to_owned()))
            .build();
        // The probe is a separate fork+exec that would fail first and
        // never reach the record under test.
        options.minimum_cli_version = None;
        options.external_mcp_servers.insert(
            "billing-http".to_owned(),
            forge_primitives::McpServerConfig::Http {
                url: "https://mcp.example.com".to_owned(),
                headers: std::collections::HashMap::from([(
                    "Authorization".to_owned(),
                    "Bearer sentinel-http-header-must-never-be-logged".to_owned(),
                )]),
            },
        );
        options.external_mcp_servers.insert(
            "billing-sse".to_owned(),
            forge_primitives::McpServerConfig::Sse {
                url: "https://sse.example.com".to_owned(),
                headers: std::collections::HashMap::from([(
                    "Authorization".to_owned(),
                    "Bearer sentinel-sse-header-must-never-be-logged".to_owned(),
                )]),
            },
        );
        options.external_mcp_servers.insert(
            "billing-stdio".to_owned(),
            forge_primitives::McpServerConfig::Stdio {
                command: "/bin/true".to_owned(),
                args: Vec::new(),
                env: std::collections::HashMap::from([(
                    "STDIO_MCP_TOKEN".to_owned(),
                    "sentinel-stdio-env-must-never-be-logged".to_owned(),
                )]),
            },
        );
        options
    }

    /// Every declared secret, read back off the `Options` rather than
    /// restated, so a sentinel added to the fixture is asserted
    /// without touching the assertion.
    fn declared_secrets(options: &Options) -> Vec<String> {
        let mut secrets: Vec<String> = options.env.values().cloned().collect();
        secrets.extend(options.extra_args.values().flatten().cloned());
        for cfg in options.external_mcp_servers.values() {
            match cfg {
                forge_primitives::McpServerConfig::Stdio { env, .. } => {
                    secrets.extend(env.values().cloned());
                }
                forge_primitives::McpServerConfig::Sse { headers, .. }
                | forge_primitives::McpServerConfig::Http { headers, .. } => {
                    secrets.extend(headers.values().cloned());
                }
            }
        }
        if let Some(crate::SystemPromptKind::Preset { append: Some(text), .. }) =
            &options.system_prompt
        {
            secrets.push(text.clone());
        }
        secrets
    }

    /// A spawn that cannot enter its `cwd` and one that cannot find the
    /// binary both surface as `ENOENT`, so the error has to say which.
    /// The mirror case pins the other half: a genuinely missing binary
    /// still reports as one, so the cwd arm cannot swallow it.
    #[tokio::test]
    async fn spawn_names_a_missing_cwd_apart_from_a_missing_binary() {
        let missing_cwd = std::env::temp_dir().join("forge-sdk-no-such-working-dir");
        assert!(!missing_cwd.exists(), "fixture precondition: {missing_cwd:?} must not exist");
        let missing_cwd = missing_cwd.to_string_lossy().into_owned();
        let options = Options {
            binary: "claude".to_owned(),
            cwd: Some(missing_cwd.clone().into()),
            minimum_cli_version: None,
            ..Options::default()
        };
        let err = Subprocess::spawn(&options).await.expect_err("a missing cwd must fail the spawn");
        assert!(
            matches!(&err, Error::CwdNotFound { path } if *path == missing_cwd),
            "a spawn that could not enter its working directory reported {err:?} rather than \
             naming that directory"
        );

        let options = Options {
            binary: "forge-no-such-binary-on-path".to_owned(),
            cwd: Some(std::env::temp_dir()),
            minimum_cli_version: None,
            ..Options::default()
        };
        let err =
            Subprocess::spawn(&options).await.expect_err("a missing binary must fail the spawn");
        assert!(
            matches!(err, Error::CliNotFound { .. }),
            "a binary that is not on PATH reported {err:?} rather than CliNotFound"
        );
    }

    /// The spawn record must not render a credential. Two carriers
    /// reach it independently: the child environment, and argv.
    #[test]
    fn spawn_record_renders_no_declared_secret() {
        use tracing_subscriber::layer::SubscriberExt;

        let options = options_with_secrets();
        let capture = FieldCapture::default();
        let subscriber = tracing_subscriber::registry()
            .with(capture.clone())
            .with(tracing_subscriber::filter::LevelFilter::TRACE);
        let result = tracing::subscriber::with_default(subscriber, || {
            tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .expect("runtime")
                .block_on(Subprocess::spawn(&options))
        });
        assert!(
            matches!(result, Err(Error::CliNotFound { .. })),
            "fixture binary must not exist, got {:?}",
            result.as_ref().err()
        );

        let logged = capture.text();
        assert!(
            logged.contains("spawning claude subprocess"),
            "the record under test never fired, so nothing was proved: {logged:?}",
        );

        // Without a floor the loop below is vacuous: emptying the
        // fixture leaves zero secrets and a green test.
        let secrets = declared_secrets(&options);
        assert_eq!(secrets.len(), 8, "fixture lost a sentinel: {secrets:?}");

        let argv = build_args(&options).expect("fixture argv").join("\u{1}");
        for secret in &secrets {
            // A sentinel reaching neither argv nor the child env would
            // prove nothing by being absent from the record.
            assert!(
                argv.contains(secret) || options.env.values().any(|v| v == secret),
                "sentinel {secret:?} reaches neither argv nor the child env",
            );
            // `--sentinel-…` renders with its dashes stripped, so
            // searching for the raw value alone would miss the leak.
            // Pairs with the dashed `extra_args` sentinel: drop either and prefix-widening
            // stops being covered while this test stays green.
            let bare = secret.trim_start_matches('-');
            assert!(
                !logged.contains(secret) && !logged.contains(bare),
                "spawn record leaked {secret:?}: {logged:?}",
            );
        }
    }

    fn spawn_sleep_subprocess(secs: u64) -> Subprocess {
        spawn_sleep_subprocess_with_bound(secs, STDIN_WRITE_BOUND)
    }

    fn spawn_sleep_subprocess_with_bound(secs: u64, write_bound: Duration) -> Subprocess {
        let mut cmd = Command::new("/bin/sleep");
        cmd.arg(secs.to_string());
        spawn_test_subprocess_with_bound(cmd, write_bound)
    }

    fn spawn_test_subprocess(cmd: Command) -> Subprocess {
        spawn_test_subprocess_with_bound(cmd, STDIN_WRITE_BOUND)
    }

    fn spawn_test_subprocess_with_bound(mut cmd: Command, write_bound: Duration) -> Subprocess {
        cmd.stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::piped());
        cmd.kill_on_drop(true);
        let mut child = cmd.spawn().expect("spawn test child");
        let stdin = child.stdin.take().expect("stdin");
        let stdout = child.stdout.take().expect("stdout");
        let stderr = child.stderr.take().expect("stderr");

        let stderr_task = tokio::spawn(drain_stderr(stderr));
        let (reader_tx, reader_rx) = mpsc::unbounded_channel();
        spawn_reader_task(stdout, None, reader_tx);
        let (writer_tx, writer_rx) = mpsc::unbounded_channel();
        let session = Arc::new(parking_lot::RwLock::new(String::new()));
        let writer_task =
            spawn_writer_task(stdin, None, writer_rx, write_bound, Arc::clone(&session));

        Subprocess {
            writer_tx,
            write_bound,
            reader_rx,
            writer_task: Some(writer_task),
            stderr_task: Some(stderr_task),
            child: Some(child),
            pgid: None,
            session,
            closed: false,
        }
    }

    /// `close()` must SIGKILL a subprocess that ignores stdin EOF
    /// rather than hang on `wait()`. Bound documented at 5s; allow ~1s
    /// slack for tokio scheduling.
    #[tokio::test]
    async fn close_kills_unresponsive_child_within_5s() {
        let mut sub = spawn_sleep_subprocess(30);

        let start = tokio::time::Instant::now();
        let result = sub.close().await;
        let elapsed = start.elapsed();

        // The documented bound is 5s; widen the assert to 30s for
        // CI-load tolerance - flagging a one-time-blip scheduler
        // delay as a regression would be noise.
        assert!(elapsed <= Duration::from_secs(30), "close() took {elapsed:?}, expected <= 30s");
        assert!(
            elapsed >= Duration::from_secs(5),
            "close() returned in {elapsed:?}, expected >= 5s (close timeout fired)"
        );
        match result {
            Err(Error::Process { stderr, .. }) => {
                assert!(
                    stderr.contains("close timeout"),
                    "expected stderr to mention close timeout, got: {stderr}"
                );
            }
            other => panic!("expected Process error from kill, got {other:?}"),
        }
    }

    /// **A child that stops reading stdin must fail the write, not hang it.**
    /// The writer task parked in `write_all` on a full pipe, so no ack ever
    /// came and every caller behind it waited for ever. The hook callback
    /// that writes a control response is one of those callers: the server
    /// went silent for 26.7 minutes on 2026-11-09 with no error and no reap.
    ///
    /// The clock is paused so the wait is measured, not raced: the writer's
    /// own bound is the only timer that can fire, so the verdict is the
    /// writer's and the elapsed time is exact.
    #[tokio::test(start_paused = true)]
    async fn a_child_that_stops_reading_stdin_fails_the_write_within_the_bound() {
        // `sleep` inherits the pipe and never reads it, so the buffer fills
        // and the next write has nowhere to go.
        let sub = spawn_sleep_subprocess_with_bound(30, Duration::from_millis(300));
        let writer = sub.clone_writer();
        // Past any pipe buffer, so `write_all` cannot finish in one go.
        let line = format!("{}\n", "x".repeat(1 << 20));

        let started = tokio::time::Instant::now();
        let outcome = tokio::time::timeout(Duration::from_secs(5), writer.write_line(&line))
            .await
            .expect("the write did not fail within its bound");
        let waited = started.elapsed();
        let err = outcome.expect_err("a child that stopped reading stdin must fail the write");
        let Error::Io(io) = err else { panic!("expected an I/O error, got {err:?}") };
        assert_eq!(io.kind(), std::io::ErrorKind::TimedOut, "the bound is what failed it: {io}");
        assert!(
            waited >= Duration::from_millis(300),
            "the writer's own bound is what fired: {waited:?}"
        );
        // **The writer's verdict, not the caller's.** With the write's bound
        // widened, the caller's ack wait is what answers, and its message
        // names the wait instead: this assertion is what tells them apart.
        assert!(
            io.to_string().contains("a stdin write did not complete"),
            "the writer's own failure must be the answer, not the caller's wait: {io}"
        );

        // **The writer task itself gave up, not only the caller.** A partial
        // line may be on the wire, so it dropped stdin: the next write
        // answers at once with a broken pipe rather than queueing for ever
        // behind the write that never finished.
        let again = tokio::time::timeout(Duration::from_secs(5), writer.write_line("more\n"))
            .await
            .expect("a write after a stalled one must answer, not hang");
        let err = again.expect_err("a stalled stream must not accept another line");
        let Error::Io(io) = err else { panic!("expected an I/O error, got {err:?}") };
        assert_eq!(io.kind(), std::io::ErrorKind::BrokenPipe, "the task did not drop stdin: {io}");
        assert!(
            io.to_string().contains("outlived its bound"),
            "a later write must name the stall, not a call that never ran: {io}"
        );
    }

    /// The other half of the same hole: the writer task is alive (or gone),
    /// but the command sits behind a write that never finishes, so its ack
    /// never arrives. **All four ack waits are bounded**, and each used to
    /// wait for ever: the two writes and the two end-inputs. The paused clock
    /// makes the caller's deadline measurable rather than raced.
    #[tokio::test(start_paused = true)]
    async fn every_wait_on_a_writer_that_never_acks_fails_within_its_bound() {
        /// Answers how long the wait took and what it said, so the caller's
        /// deadline can be measured rather than assumed.
        async fn fails_within<F>(what: &str, fut: F) -> (Duration, String)
        where
            F: std::future::Future<Output = Result<(), Error>>,
        {
            let started = tokio::time::Instant::now();
            let outcome = tokio::time::timeout(Duration::from_secs(5), fut).await;
            let outcome =
                outcome.unwrap_or_else(|_| panic!("{what} did not fail within its bound"));
            let waited = started.elapsed();
            let err = match outcome {
                Ok(()) => panic!("{what} answered Ok from a writer that never acks"),
                Err(err) => err,
            };
            let Error::Io(io) = err else { panic!("{what}: expected an I/O error, got {err:?}") };
            assert_eq!(io.kind(), std::io::ErrorKind::TimedOut, "{what}: {io}");
            (waited, io.to_string())
        }

        // A writer queue nothing drains: the command is accepted, and the ack
        // it would answer with never comes.
        let (writer_tx, queue) = mpsc::unbounded_channel();
        let bound = Duration::from_millis(100);
        let (_reader_tx, reader_rx) = mpsc::unbounded_channel();
        let mut subprocess = Subprocess {
            writer_tx: writer_tx.clone(),
            write_bound: bound,
            reader_rx,
            writer_task: None,
            stderr_task: None,
            child: None,
            pgid: None,
            session: Arc::new(parking_lot::RwLock::new(String::new())),
            closed: false,
        };
        let shared = SharedWriter { writer_tx, bound };

        // **The caller waits twice the writer's bound**, measured: at the
        // writer's own bound it would have given up first, losing the task's
        // verdict - and the record would name a wait it did not take.
        let (waited, why) =
            fails_within("Subprocess::write_line", subprocess.write_line("line\n")).await;
        assert!(waited >= bound * 2, "a caller waits twice the writer's bound: {waited:?}");
        assert!(
            why.contains(&format!("{:?}", bound * 2)),
            "the record names the wait it fired at: {why}"
        );

        fails_within("Subprocess::end_input", subprocess.end_input()).await;
        fails_within("SharedWriter::write_line", shared.write_line("line\n")).await;
        fails_within("SharedWriter::end_input", shared.end_input()).await;
        drop(queue);
    }

    /// **Each close reason names itself**, and none reads as a deliberate
    /// close: a later write must never blame a call that did not run. A typo
    /// in either string would otherwise ship with every other test green.
    #[test]
    fn a_closed_stream_names_why_it_closed() {
        for (reason, needle) in [
            (Closed::EndInput, "end_input"),
            (Closed::Stalled, "outlived its bound"),
            (Closed::Failed, "a write failed"),
        ] {
            let Error::Io(io) = closed_stdin(Some(reason)) else {
                panic!("expected an I/O error for {reason:?}")
            };
            assert_eq!(io.kind(), std::io::ErrorKind::BrokenPipe, "{reason:?}: {io}");
            assert!(io.to_string().contains(needle), "{reason:?}: {io}");
        }
        let Error::Io(io) = closed_stdin(None) else { panic!("expected an I/O error") };
        assert!(io.to_string().contains("already closed"), "{io}");
    }

    /// **A stall names the session the process carries now**, not the one it
    /// was spawned with: `/new` swaps the occupant under a long-lived child,
    /// and that is exactly the session a freeze happens in. The record also
    /// carries the bound it fired at.
    ///
    /// Synchronous like the spawn-record test: it builds its own runtime, so
    /// the capture subscriber is the thread's default for the whole run.
    #[test]
    fn a_stdin_stall_records_the_current_session_and_its_bound() {
        use tracing_subscriber::layer::SubscriberExt;

        let capture = FieldCapture::default();
        let subscriber = tracing_subscriber::registry()
            .with(capture.clone())
            .with(tracing_subscriber::filter::LevelFilter::TRACE);
        tracing::subscriber::with_default(subscriber, || {
            tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .expect("runtime")
                .block_on(async {
                    let sub = spawn_sleep_subprocess_with_bound(30, Duration::from_millis(100));
                    // The session this child carries *after* a `/new`.
                    *sub.session_id().write() = "session-after-new".to_owned();
                    let writer = sub.clone_writer();
                    let line = format!("{}\n", "x".repeat(1 << 20));
                    let _ = tokio::time::timeout(Duration::from_secs(5), writer.write_line(&line))
                        .await;
                });
        });

        let logged = capture.text();
        assert!(
            logged.contains("stdin_write_stalled"),
            "the stall's record never fired, so nothing was proved: {logged:?}",
        );
        assert!(
            logged.contains("session-after-new"),
            "the record must name the session the process carries now: {logged:?}",
        );
        assert!(
            logged.contains("bound_ms=100"),
            "the record must carry the bound it fired at: {logged:?}"
        );
    }

    /// An empty group answers `ESRCH`, and that is not forge's problem:
    /// mapping it to a report would warn on every clean exit.
    #[cfg(unix)]
    #[test]
    fn kill_group_treats_an_empty_group_as_nothing_to_report() {
        // Above any kernel's pid ceiling but inside `i32`, so the call
        // reaches the kernel and fails there.
        assert!(kill_group(2_000_000_000).is_none(), "an empty group is not a failure");
    }

    /// A signal that lands is no failure either, and it reaches the whole
    /// group the child leads.
    #[cfg(unix)]
    #[tokio::test]
    async fn kill_group_signals_a_live_group_without_reporting() {
        let mut command = Command::new("/bin/sleep");
        command.arg("30").process_group(0);
        let mut child = command.spawn().expect("a child in its own group");
        let pid = child.id().expect("a live child has a pid");

        assert!(kill_group(pid).is_none(), "a landed group kill is not a failure");

        let status = child.wait().await.expect("the child is reaped");
        assert!(!status.success(), "the group signal killed the child: {status}");
    }

    #[test]
    fn stderr_tail_keeps_the_last_lines_within_cap() {
        let mut tail = StderrTail::default();
        for i in 0..20_000 {
            tail.push(&format!("line {i}"));
        }
        let out = tail.into_string();
        assert!(out.len() <= STDERR_TAIL_CAP, "tail grew to {} bytes", out.len());
        assert!(out.contains("line 19999"), "tail dropped the newest line");
        assert!(!out.contains("line 0\n"), "tail kept the oldest line");

        let mut tail = StderrTail::default();
        tail.push(&format!("{}REASON_AT_END", "x".repeat(STDERR_TAIL_CAP * 2)));
        let out = tail.into_string();
        assert!(out.len() <= STDERR_TAIL_CAP, "tail grew to {} bytes", out.len());
        assert!(!out.is_empty(), "an oversized sole line was evicted to nothing");
        assert!(out.ends_with("REASON_AT_END"), "kept the head of an over-cap line, not the end");
    }

    #[tokio::test]
    async fn close_reports_drained_stderr_on_nonzero_exit() {
        let mut cmd = Command::new("/bin/sh");
        cmd.arg("-c").arg("echo 'boom: unknown flag --nope' >&2; exit 3");

        match spawn_test_subprocess(cmd).close().await {
            Err(Error::Process { exit_code, stderr }) => {
                assert_eq!(exit_code, Some(3));
                assert!(
                    stderr.contains("boom: unknown flag --nope"),
                    "expected drained stderr in the error, got: {stderr:?}"
                );
                assert!(
                    stderr.starts_with("last stderr from the CLI:"),
                    "tail must be labelled, not presented as the cause: {stderr:?}"
                );
            }
            other => panic!("expected Process error, got {other:?}"),
        }
    }

    /// stderr is inherited by everything `claude` spawns, so a stray
    /// non-UTF-8 byte is realistic. It must not cost the drain the rest
    /// of the stream - the reason arrives last, after the bad byte.
    #[tokio::test]
    async fn close_reports_stderr_past_invalid_utf8() {
        let mut cmd = Command::new("/bin/sh");
        cmd.arg("-c").arg(
            "printf 'early-line\\n\\377\\376bad\\n' >&2; \
             printf 'after-bad\\n' >&2; \
             printf 'THE REAL REASON' >&2; \
             exit 9",
        );

        match spawn_test_subprocess(cmd).close().await {
            Err(Error::Process { exit_code, stderr }) => {
                assert_eq!(exit_code, Some(9));
                assert!(
                    stderr.contains("after-bad"),
                    "drain died at the invalid byte, losing the rest: {stderr:?}"
                );
                assert!(
                    stderr.contains("THE REAL REASON"),
                    "final line without a trailing newline was dropped: {stderr:?}"
                );
            }
            other => panic!("expected Process error, got {other:?}"),
        }
    }
}