processkit 1.0.1

//! [`RunningProcess`] — a live handle to a spawned child.
//!
//! Split by concern: this file owns the handle's state and the consuming
//! capture paths (exit driving, kill/teardown, the post-exit checkpoint);
//! [`probes`] holds the non-consuming readiness probes; [`stream`] holds the
//! incremental stdout streaming surface.

mod probes;
mod stream;

pub use stream::{Finished, OutputEvent, OutputEvents, OutputLine, StdoutLines};

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
use std::time::{Duration, Instant, SystemTime};

use encoding_rs::Encoding;
use tokio::io::AsyncReadExt;
use tokio::process::{Child, ChildStderr, ChildStdin, ChildStdout};
use tokio::sync::Notify;
use tokio::task::{AbortHandle, JoinHandle};

use crate::buffer::OutputBufferPolicy;
use crate::error::Error;
use crate::error::Result;
use crate::group::ProcessGroup;
use crate::pump::{LineHandler, SharedLines, pump_lines_core};
use crate::result::{Outcome, ProcessResult};
use crate::stdin::ProcessStdin;

/// How long teardown waits for output pumps to finish before aborting them, so a
/// surviving grandchild holding a pipe can't hang the run.
const PUMP_TEARDOWN: Duration = Duration::from_secs(5);

// Timeout-arbitration states for `RunningProcess::timeout_state`. Whichever of
// the natural reap (claims `EXITED`) or a fired deadline (claims `TIMED_OUT`)
// first `compare_exchange`s from `PENDING` wins — a single CAS arbiter that
// keeps "timed out vs exited" race-free even when the child exits within a
// scheduler quantum of the deadline.
const TS_PENDING: u8 = 0;
const TS_EXITED: u8 = 1;
const TS_TIMED_OUT: u8 = 2;

/// Why a reap-via-wait ended — the race result, not a post-hoc token read.
enum ExitCause {
    /// Child exited on its own (or deadline fired). Cancellation did not win.
    Exited(Outcome),
    /// Cancel arm won: token fired, tree killed. Becomes `Err(Cancelled)`.
    Cancelled,
}

/// Internal result of `finish_lines` — distinct from the public `Finished`.
struct FinishedLines {
    outcome: Outcome,
    stdout_lines: Vec<String>,
    stderr_lines: Vec<String>,
}

/// How [`RunningProcess::finish_lines`] treats the pumped lines.
#[derive(Clone, Copy)]
enum CaptureMode {
    /// Retain both streams' lines (`output_string`).
    Lines,
    /// Pump (so the child never blocks on a full pipe) but drop the lines.
    Discard,
}

/// The fields produced by a spawn, handed to [`RunningProcess::from_spawned`].
pub(crate) struct Spawned {
    pub program: String,
    pub child: Child,
    pub own_group: Option<ProcessGroup>,
    pub stdout: Option<ChildStdout>,
    pub stderr: Option<ChildStderr>,
    pub stdin: Option<ChildStdin>,
    pub stdin_task: Option<JoinHandle<std::io::Result<()>>>,
    pub timeout: Option<Duration>,
    /// Grace window for a graceful timeout (`None` = hard kill at the deadline).
    pub timeout_grace: Option<Duration>,
    /// Raw signal for the graceful-timeout phase (default `SIGTERM`).
    pub timeout_signal: i32,
    pub pid: Option<u32>,
    pub stdout_encoding: &'static Encoding,
    pub stderr_encoding: &'static Encoding,
    pub stdout_handler: Option<LineHandler>,
    pub stderr_handler: Option<LineHandler>,
    pub stdout_tee: Option<crate::pump::TeeSink>,
    pub stderr_tee: Option<crate::pump::TeeSink>,
    pub buffer: OutputBufferPolicy,
    /// Exit codes treated as success (default `[0]`), carried onto the result.
    pub ok_codes: Vec<i32>,
    /// Whether stdout is `Piped` (capturable) vs `Inherit`/`Null`.
    pub stdout_piped: bool,
    pub cancel_token: Option<tokio_util::sync::CancellationToken>,
}

/// A handle to a process spawned by a runner.
pub struct RunningProcess {
    // The Option fields below encode the handle's de-facto states (fresh /
    // streaming / consumed) implicitly. No runtime state enum on purpose:
    // consuming verbs take `self` by value (double consumption is a compile
    // error), and the two &mut entry points handle a repeat call without
    // panicking — `stdout_lines`/`output_events` return a loud `Err`, and
    // `take_stdin` returns `None`. A state enum would only add panic paths to
    // guard doors the borrow checker already locks.
    program: String,
    /// The I/O-bearing half: a real OS child, or a scripted double feeding the
    /// same pump machinery (see [`Backend`]).
    backend: Backend,
    timeout: Option<Duration>,
    timeout_grace: Option<Duration>,
    timeout_signal: i32,
    pid: Option<u32>,
    stdout_encoding: &'static Encoding,
    stderr_encoding: &'static Encoding,
    stdout_handler: Option<LineHandler>,
    stderr_handler: Option<LineHandler>,
    stdout_tee: Option<crate::pump::TeeSink>,
    stderr_tee: Option<crate::pump::TeeSink>,
    buffer: OutputBufferPolicy,
    ok_codes: Vec<i32>,
    stdout_sink: Option<Arc<SharedLines>>,
    stderr_sink: Option<Arc<SharedLines>>,
    // Joined before the overflow check so the last lines are visible.
    stdout_pump: Option<JoinHandle<()>>,
    stderr_pump: Option<JoinHandle<()>>,
    // Non-broken-pipe stdin failure stashed by `observe_stdin_task`; surfaced as
    // `Error::Stdin` by `checked_outcome` only when the run otherwise succeeded.
    stdin_error: Option<std::io::Error>,
    // Bulk capture verbs fail loudly on non-piped stdout rather than returning empty.
    stdout_piped: bool,
    // Streaming deadline watchdog; aborted on drop.
    deadline_task: Option<JoinHandle<()>>,
    // Shared (`Arc`) because the watchdog is detached. See `TS_*` constants.
    timeout_state: Arc<AtomicU8>,
    cancel_token: Option<tokio_util::sync::CancellationToken>,
    // Armed at spawn time so every consuming path kills the tree when the token
    // fires, not just `drive_to_exit`.
    cancel_task: Option<JoinHandle<()>>,
    // Cancel disposition snapshotted at first reap (first-observation wins);
    // `None` = not yet snapshotted.
    cancel_at_exit: Option<bool>,
    started: Instant,
    start_time: SystemTime,
}

/// A boxed output reader: real `ChildStdout`/`ChildStderr` or scripted bytes.
/// Both flow through the same pump machinery via `AsyncRead`.
type OutputReader = Box<dyn tokio::io::AsyncRead + Send + Unpin>;

/// The I/O-bearing half of a [`RunningProcess`]: a real OS child or a scripted
/// double that feeds canned bytes through the same pumps/sinks. Platform code
/// only ever constructs `Real`.
enum Backend {
    // Boxed: both variants are large and the enum lives in every handle.
    Real(Box<RealProc>),
    Scripted(Box<ScriptedProc>),
}

/// The real-child fields — exactly the ones that touch the OS.
struct RealProc {
    child: Child,
    // `Arc` so a streaming deadline timer can hold a `Weak` to kill the tree
    // without keeping the group alive (kill-on-close on drop stays prompt).
    own_group: Option<Arc<ProcessGroup>>,
    stdout_pipe: Option<ChildStdout>,
    stderr_pipe: Option<ChildStderr>,
    stdin_pipe: Option<ChildStdin>,
    stdin_task: Option<JoinHandle<std::io::Result<()>>>,
}

/// Shared kill state for a scripted child, clonable so a detached watchdog can
/// end the run. `fire` hangs up feeders, flags the child dead, and wakes a
/// parked `backend_wait`.
#[derive(Clone)]
struct ScriptedKill {
    killed: Arc<AtomicBool>,
    /// `notify_one` stores a permit so a kill before the wait parks is not missed.
    signal: Arc<Notify>,
    /// Aborting a writer drops its end, EOF-ing the reader — as a real tree's
    /// death closes its pipes. `abort` is idempotent.
    feeders: Arc<Vec<AbortHandle>>,
}

impl ScriptedKill {
    fn fire(&self) {
        self.killed.store(true, Ordering::Release);
        for feeder in self.feeders.iter() {
            feeder.abort();
        }
        self.signal.notify_one();
    }
}

/// A scripted "child": canned output readers (fed by detached writer tasks so
/// per-line delays work under a paused clock) plus a canned exit.
pub(crate) struct ScriptedProc {
    stdout: Option<tokio::io::DuplexStream>,
    stderr: Option<tokio::io::DuplexStream>,
    kill: ScriptedKill,
    code: Option<i32>,
    timed_out: bool,
    signal: Option<i32>,
    /// When the scripted child "exits": `Some(at)` resolves at that instant
    /// (now = immediately), `None` never exits on its own (`Reply::pending` —
    /// cancel/timeout still end it).
    exit_at: Option<tokio::time::Instant>,
}

impl ScriptedProc {
    /// Assemble a scripted child. Each output's text is fed through a duplex
    /// pipe by a detached writer task — with `line_delay`, the writer sleeps
    /// before each line (virtual-time friendly under a paused clock). The
    /// "process" exits after `lifetime` (`None` = never on its own).
    pub(crate) fn new(
        stdout_text: String,
        stderr_text: String,
        code: Option<i32>,
        timed_out: bool,
        signal: Option<i32>,
        lifetime: Option<Duration>,
        line_delay: Option<Duration>,
    ) -> Self {
        let mut feeders = Vec::new();
        let mut feed = |text: String| {
            let (mut tx, rx) = tokio::io::duplex(64 * 1024);
            if text.is_empty() {
                return rx; // dropped tx → immediate EOF
            }
            // Detached: `AbortHandle` is the only way to hang it up early.
            let task = tokio::spawn(async move {
                use tokio::io::AsyncWriteExt;
                match line_delay {
                    None => {
                        let _ = tx.write_all(text.as_bytes()).await;
                    }
                    Some(delay) => {
                        for line in text.split_inclusive('\n') {
                            tokio::time::sleep(delay).await;
                            if tx.write_all(line.as_bytes()).await.is_err() {
                                break;
                            }
                        }
                    }
                }
                // tx drops → EOF
            });
            feeders.push(task.abort_handle());
            rx
        };
        let stdout = feed(stdout_text);
        let stderr = feed(stderr_text);
        Self {
            stdout: Some(stdout),
            stderr: Some(stderr),
            kill: ScriptedKill {
                killed: Arc::new(AtomicBool::new(false)),
                signal: Arc::new(Notify::new()),
                feeders: Arc::new(feeders),
            },
            code,
            timed_out,
            signal,
            exit_at: lifetime.map(|d| tokio::time::Instant::now() + d),
        }
    }

    fn kill(&self) {
        self.kill.fire();
    }
}

impl Backend {
    fn own_group(&self) -> Option<&Arc<ProcessGroup>> {
        match self {
            Backend::Real(real) => real.own_group.as_ref(),
            Backend::Scripted(_) => None,
        }
    }

    fn scripted_kill(&self) -> Option<ScriptedKill> {
        match self {
            Backend::Real(_) => None,
            Backend::Scripted(s) => Some(s.kill.clone()),
        }
    }

    fn take_stdout_reader(&mut self) -> Option<OutputReader> {
        match self {
            Backend::Real(real) => real.stdout_pipe.take().map(|p| Box::new(p) as OutputReader),
            Backend::Scripted(s) => s.stdout.take().map(|p| Box::new(p) as OutputReader),
        }
    }

    fn take_stderr_reader(&mut self) -> Option<OutputReader> {
        match self {
            Backend::Real(real) => real.stderr_pipe.take().map(|p| Box::new(p) as OutputReader),
            Backend::Scripted(s) => s.stderr.take().map(|p| Box::new(p) as OutputReader),
        }
    }
}

impl RunningProcess {
    pub(crate) fn from_spawned(s: Spawned) -> Self {
        Self {
            program: s.program,
            backend: Backend::Real(Box::new(RealProc {
                child: s.child,
                own_group: s.own_group.map(Arc::new),
                stdout_pipe: s.stdout,
                stderr_pipe: s.stderr,
                stdin_pipe: s.stdin,
                stdin_task: s.stdin_task,
            })),
            timeout: s.timeout,
            timeout_grace: s.timeout_grace,
            timeout_signal: s.timeout_signal,
            pid: s.pid,
            stdout_encoding: s.stdout_encoding,
            stderr_encoding: s.stderr_encoding,
            stdout_handler: s.stdout_handler,
            stderr_handler: s.stderr_handler,
            stdout_tee: s.stdout_tee,
            stderr_tee: s.stderr_tee,
            buffer: s.buffer,
            ok_codes: s.ok_codes,
            stdout_sink: None,
            stderr_sink: None,
            stdout_pump: None,
            stderr_pump: None,
            stdin_error: None,
            stdout_piped: s.stdout_piped,
            deadline_task: None,
            timeout_state: Arc::new(AtomicU8::new(TS_PENDING)),
            cancel_token: s.cancel_token,
            cancel_task: None,
            cancel_at_exit: None,
            started: Instant::now(),
            start_time: SystemTime::now(),
        }
    }

    /// Build a scripted handle: same encodings/handlers/buffer/timeout/token as
    /// a real run so hermetic tests exercise the same pump machinery. `pid()` is
    /// `None`.
    pub(crate) fn from_scripted(command: &crate::command::Command, scripted: ScriptedProc) -> Self {
        Self {
            program: command.program_name(),
            backend: Backend::Scripted(Box::new(scripted)),
            timeout: command.configured_timeout(),
            timeout_grace: command.configured_timeout_grace(),
            timeout_signal: command.timeout_signal_raw(),
            pid: None,
            stdout_encoding: command.out_encoding(),
            stderr_encoding: command.err_encoding(),
            stdout_handler: command.stdout_handler(),
            stderr_handler: command.stderr_handler(),
            stdout_tee: command.stdout_tee_sink(),
            stderr_tee: command.stderr_tee_sink(),
            buffer: command.output_buffer_policy(),
            ok_codes: command.ok_codes_vec(),
            stdout_sink: None,
            stderr_sink: None,
            stdout_pump: None,
            stderr_pump: None,
            stdin_error: None,
            stdout_piped: command.stdout_is_piped(),
            deadline_task: None,
            timeout_state: Arc::new(AtomicU8::new(TS_PENDING)),
            cancel_token: command.cancel_token(),
            cancel_task: None,
            cancel_at_exit: None,
            started: Instant::now(),
            start_time: SystemTime::now(),
        }
    }

    pub(crate) fn attach_group(&mut self, group: ProcessGroup) {
        if let Backend::Real(real) = &mut self.backend {
            real.own_group = Some(Arc::new(group));
        }
        // Re-arm the cancel watchdog now that the group is known: upgrade from
        // the pid-only task armed in `launch` to a full group+pid kill.
        self.arm_cancel_watchdog();
    }

    /// Arm (or re-arm) the cancel kill task. Aborts any existing task first so
    /// `attach_group` upgrades from pid-only to group+pid. No-op without a token.
    pub(crate) fn arm_cancel_watchdog(&mut self) {
        {
            if let Some(old) = self.cancel_task.take() {
                old.abort();
            }
            let Some(token) = self.cancel_token.clone() else {
                return;
            };
            let group_weak = self.backend.own_group().map(Arc::downgrade);
            let pid = self.pid;
            let timeout_state = self.timeout_state.clone();
            self.cancel_task = Some(tokio::spawn(async move {
                token.cancelled().await;
                // The arbiter holds `TS_PENDING` only until a reap claims it;
                // `abort_watchdogs` also aborts this task on reap. This check
                // closes the residual window where a cancel fires between reap
                // and abort — don't signal a potentially-recycled pid.
                if timeout_state.load(Ordering::Acquire) != TS_PENDING {
                    return;
                }
                if let Some(g) = group_weak.and_then(|w| w.upgrade()) {
                    let _ = g.terminate_all();
                }
                stream::kill_direct_child(pid);
            }));
        }
    }

    /// Arm a deadline watchdog for a scripted streamed run. A scripted handle has
    /// no process group; this task hangs up the feeders at the deadline instead,
    /// claiming `TIMED_OUT` via the arbiter. No-op for a real backend or when a
    /// watchdog is already armed.
    fn arm_scripted_deadline(&mut self) {
        if self.deadline_task.is_some() {
            return;
        }
        let (Some(limit), Some(kill)) = (self.timeout, self.backend.scripted_kill()) else {
            return;
        };
        // Anchor to spawn time so a late stream call can't re-grant the full limit.
        let started = self.started;
        let timeout_state = self.timeout_state.clone();
        self.deadline_task = Some(tokio::spawn(async move {
            let remaining = limit
                .checked_sub(started.elapsed())
                .unwrap_or(Duration::ZERO);
            tokio::time::sleep(remaining).await;
            if timeout_state
                .compare_exchange(
                    TS_PENDING,
                    TS_TIMED_OUT,
                    Ordering::AcqRel,
                    Ordering::Relaxed,
                )
                .is_err()
            {
                return; // the script already exited on its own — no kill
            }
            kill.fire();
        }));
    }

    /// Take the raw stdout pipe for `Pipeline` plumbing. `None` for a scripted
    /// backend (scripted doubles don't compose into real pipelines).
    pub(crate) fn take_stdout_pipe(&mut self) -> Option<ChildStdout> {
        match &mut self.backend {
            Backend::Real(real) => real.stdout_pipe.take(),
            Backend::Scripted(_) => None,
        }
    }

    /// The program this handle is running (for error/outcome attribution).
    pub(crate) fn program_name(&self) -> &str {
        &self.program
    }
}

// Manual impl: pipes, pump tasks, and line handlers are opaque.
impl std::fmt::Debug for RunningProcess {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RunningProcess")
            .field("program", &self.program)
            .field("pid", &self.pid)
            .field("timeout", &self.timeout)
            .finish_non_exhaustive()
    }
}

impl RunningProcess {
    /// The OS process id, or `None` if the child has already been reaped.
    pub fn pid(&self) -> Option<u32> {
        self.pid
    }

    /// Wall-clock instant the process was started.
    pub fn start_time(&self) -> SystemTime {
        self.start_time
    }

    /// Time elapsed since the process started (sampled now).
    pub fn elapsed(&self) -> Duration {
        self.started.elapsed()
    }

    /// CPU time (user + kernel) consumed so far, if the platform can report it.
    #[cfg(feature = "stats")]
    pub fn cpu_time(&self) -> Option<Duration> {
        self.pid
            .and_then(|pid| crate::sys::process_metrics(pid).cpu_time)
    }

    /// Peak resident memory in bytes, if the platform can report it.
    #[cfg(feature = "stats")]
    pub fn peak_memory_bytes(&self) -> Option<u64> {
        self.pid
            .and_then(|pid| crate::sys::process_metrics(pid).peak_memory_bytes)
    }

    /// Lines read from stdout so far (counts every line, even ones dropped by an
    /// [`OutputBufferPolicy`]). Live only once stdout is being pumped.
    pub fn stdout_line_count(&self) -> usize {
        self.stdout_sink.as_ref().map_or(0, |s| s.count())
    }

    /// Lines read from stderr so far (see [`stdout_line_count`](Self::stdout_line_count)).
    pub fn stderr_line_count(&self) -> usize {
        self.stderr_sink.as_ref().map_or(0, |s| s.count())
    }

    /// Take the interactive stdin writer, if the command was built with
    /// [`keep_stdin_open`](crate::Command::keep_stdin_open). Returns `None` after
    /// the first call (or when stdin was not kept open).
    pub fn take_stdin(&mut self) -> Option<ProcessStdin> {
        match &mut self.backend {
            Backend::Real(real) => real.stdin_pipe.take().map(ProcessStdin::new),
            // Scripted doubles don't model interactive stdin yet; `None` matches
            // the "stdin wasn't kept open" contract.
            Backend::Scripted(_) => None,
        }
    }

    /// Whether **dropping** this handle will tear down (hard-kill) the process
    /// tree.
    ///
    /// `true` — owns a **private** process group; drop hard-kills the whole tree.
    /// `false` — runs inside a **shared** [`ProcessGroup`](crate::ProcessGroup)
    /// whose lifetime the group owns (drop does *not* kill the tree), or a
    /// scripted test double (no OS tree).
    pub fn kills_tree_on_drop(&self) -> bool {
        self.backend.own_group().is_some()
    }

    /// A bulk capture verb on a stdout that wasn't piped (`Inherit`/`Null`) would
    /// return silently-empty output — surface it as a clear error instead.
    /// `stdout_piped` reflects the command's `stdout` mode for *both* real and
    /// scripted handles, so a scripted run with `stdout(Null)` errors here too.
    fn ensure_stdout_capturable(&self) -> Result<()> {
        if self.stdout_piped {
            return Ok(());
        }
        Err(Error::Io(std::io::Error::new(
            std::io::ErrorKind::InvalidInput,
            format!(
                "`{}`: stdout is not piped (Command::stdout was set to Inherit/Null), so the \
                 capture verbs have nothing to read — use StdioMode::Piped to capture it",
                self.program
            ),
        )))
    }

    /// Fail loud if streaming is not possible: (a) stdout not piped, or
    /// (b) a prior streaming verb already consumed stdout on this handle.
    fn ensure_stdout_streamable(&self) -> Result<()> {
        self.ensure_stdout_capturable()?; // (a) non-piped stdout
        if self.stdout_sink.is_some() {
            return Err(Error::Io(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                format!(
                    "`{}`: stdout was already consumed by an earlier stdout_lines/output_events \
                     call — stream it once (a second call would yield an empty stream)",
                    self.program
                ),
            )));
        }
        Ok(())
    }

    /// Drain both streams, wait for exit, and return the captured text output
    /// (line-normalized to `\n`).
    ///
    /// If you previously called [`stdout_lines`](Self::stdout_lines) and
    /// consumed some lines from the stream, those already-consumed lines are
    /// gone from the buffer; `output_string` returns only the unconsumed tail.
    /// To capture the full output, avoid mixing streaming and `output_string`.
    pub async fn output_string(mut self) -> Result<ProcessResult<String>> {
        let finished = self
            .finish_lines(CaptureMode::Lines, /* expose_counts */ true, || {})
            .await?;
        // `dropped()` = lines the buffer policy discarded, NOT lines a prior
        // stream consumed — so partial streaming under the unbounded policy is
        // never mis-reported as truncated.
        let truncated = self.stdout_sink.as_ref().is_some_and(|s| s.dropped() > 0)
            || self.stderr_sink.as_ref().is_some_and(|s| s.dropped() > 0);
        let total_lines = self.stdout_sink.as_ref().map_or(0, |s| s.count())
            + self.stderr_sink.as_ref().map_or(0, |s| s.count());
        let total_bytes = self.stdout_sink.as_ref().map_or(0, |s| s.seen_bytes())
            + self.stderr_sink.as_ref().map_or(0, |s| s.seen_bytes());
        let duration = self.started.elapsed();
        Ok(ProcessResult::new(
            self.program.clone(),
            finished.stdout_lines.join("\n"),
            finished.stderr_lines.join("\n"),
            finished.outcome,
            self.timeout,
        )
        .with_duration(duration)
        .with_truncated(truncated)
        .with_overflow_totals(total_lines, total_bytes)
        .with_ok_codes(self.ok_codes.clone()))
    }

    /// Drain both streams, wait for exit, and return the exact raw stdout bytes
    /// (stderr captured as text). On timeout/cancellation returns bytes read so
    /// far — a killed run's bytes are a best-effort prefix.
    ///
    /// # Errors
    ///
    /// Returns [`Error::Io(InvalidInput)`](std::io::ErrorKind::InvalidInput) if
    /// stdout is not piped, or if a prior streaming call already consumed stdout
    /// as decoded lines (the raw bytes cannot be reconstructed).
    pub async fn output_bytes(mut self) -> Result<ProcessResult<Vec<u8>>> {
        self.ensure_stdout_capturable()?;
        if self.stdout_sink.is_some() || self.stderr_sink.is_some() {
            return Err(Error::Io(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                format!(
                    "`{}`: output_bytes cannot follow a streaming call (stdout was already \
                     consumed as lines) — use output_string to collect the streamed lines, or \
                     call output_bytes without streaming first",
                    self.program
                ),
            )));
        }
        let stderr_sink = SharedLines::new(&self.buffer);
        self.stderr_pump = self.backend.take_stderr_reader().map(|pipe| {
            tokio::spawn(pump_lines_core(
                pipe,
                self.stderr_encoding,
                self.stderr_handler.clone(),
                self.stderr_tee.clone(),
                stderr_sink.clone(),
            ))
        });
        self.stderr_sink = Some(stderr_sink.clone());

        // Read stdout raw, concurrently, so it never blocks the child. Bytes
        // accumulate in a shared buffer (not the task's return value) so the
        // bounded teardown below can salvage a partial read. Stored on `self` (not
        // a frame-local) so a `drive_to_exit` error aborts it via `Drop` instead
        // of leaving it to grow `out_buf` unboundedly on a shared-group handle.
        let mut stdout_pipe = self.backend.take_stdout_reader();
        let out_buf = Arc::new(std::sync::Mutex::new(Vec::new()));
        self.stdout_pump = Some({
            let out_buf = out_buf.clone();
            tokio::spawn(async move {
                if let Some(pipe) = &mut stdout_pipe {
                    let mut chunk = [0u8; 8 * 1024];
                    loop {
                        match pipe.read(&mut chunk).await {
                            Ok(0) => break,
                            Ok(n) => out_buf
                                .lock()
                                .expect("stdout buffer poisoned")
                                .extend_from_slice(&chunk[..n]),
                            Err(_e) => {
                                // Read error: keep partial capture, surface via tracing.
                                #[cfg(feature = "tracing")]
                                tracing::warn!(target: "processkit", error = %_e, "stdout read error; ending byte capture early");
                                break;
                            }
                        }
                    }
                }
            })
        });

        let outcome = self.drive_to_exit().await?;
        self.observe_stdin_task().await;
        // Bound the stdout drain: a surviving grandchild can hold stdout open past
        // the child's death; an unbounded read would park forever.
        if let Some(out_task) = self.stdout_pump.take() {
            let abort = out_task.abort_handle();
            if tokio::time::timeout(PUMP_TEARDOWN, out_task).await.is_err() {
                abort.abort();
            }
        }
        let stdout = std::mem::take(&mut *out_buf.lock().expect("stdout buffer poisoned"));
        join_pumps(self.stderr_pump.take().into_iter().collect()).await;
        let outcome = self.checked_outcome(outcome)?;

        if stderr_sink.overflowed() {
            return Err(crate::Error::OutputTooLarge {
                program: self.program.clone(),
                line_limit: self.buffer.max_lines,
                byte_limit: self.buffer.max_bytes,
                total_lines: stderr_sink.count(),
                total_bytes: stderr_sink.seen_bytes(),
            });
        }

        let stderr_lines = stderr_sink.drain();
        let truncated = stderr_sink.dropped() > 0;
        let duration = self.started.elapsed();
        Ok(ProcessResult::new(
            self.program.clone(),
            stdout,
            stderr_lines.join("\n"),
            outcome,
            self.timeout,
        )
        .with_duration(duration)
        .with_truncated(truncated)
        .with_overflow_totals(stderr_sink.count(), stderr_sink.seen_bytes())
        .with_ok_codes(self.ok_codes.clone()))
    }

    /// Wait for exit, returning how the run ended as an [`Outcome`] (output is
    /// drained and discarded so the child never blocks on a full pipe).
    ///
    /// Reports the raw outcome — timeout and signals are not raised as errors
    /// here. Exception: cancellation via `Command::cancel_on` always errors with
    /// `Error::Cancelled`.
    pub async fn wait(mut self) -> Result<Outcome> {
        Ok(self
            .finish_lines(CaptureMode::Discard, /* expose_counts */ false, || {})
            .await?
            .outcome)
    }

    /// Gracefully stop the process tree: `SIGTERM`, wait up to `grace`, then
    /// `SIGKILL` any survivor. On Windows the kill is atomic and `grace` is not
    /// awaited.
    ///
    /// Only an **own-group** handle can be shut down here — a **shared-group**
    /// handle returns [`Error::Unsupported`](crate::Error::Unsupported) because
    /// shutting it down would tear down the caller's other children too.
    ///
    /// If the configured timeout deadline already elapsed when `shutdown` is
    /// called the run is classified as `Outcome::TimedOut`.
    pub async fn shutdown(mut self, grace: std::time::Duration) -> Result<Outcome> {
        let Some(group) = self.backend.own_group().cloned() else {
            return Err(Error::Unsupported {
                operation: "shutdown (a shared-group handle does not own its group — \
                            use ProcessGroup::shutdown, or start_kill for just this child)"
                    .into(),
            });
        };
        // Disable the concurrent `wait()`'s deadline arm to avoid two overlapping
        // graceful teardowns. A timeout that already elapsed still classifies
        // as `TimedOut` — claim the arbiter before nulling `self.timeout`.
        if let Some(limit) = self.timeout
            && self.started.elapsed() >= limit
        {
            let _ = self.timeout_state.compare_exchange(
                TS_PENDING,
                TS_TIMED_OUT,
                Ordering::AcqRel,
                Ordering::Relaxed,
            );
        }
        self.timeout = None;
        if let Some(task) = self.deadline_task.take() {
            task.abort();
        }
        // Reap concurrently: an unreaped zombie still answers `kill(pgid, 0)`
        // probes, so without a concurrent reap a SIGTERM-handling child would
        // look alive for the whole grace and eat a pointless SIGKILL.
        let (term_result, outcome) = tokio::join!(
            group.graceful_terminate(grace, crate::sys::SIGTERM_RAW),
            self.wait(),
        );
        term_result?;
        outcome
    }

    /// Minimal non-consuming exit wait — the [`wait_any`](crate::wait_any) race
    /// participant. Spawns no pumps, applies no timeout. Cancel-safe and
    /// re-awaitable (tokio caches exit status). A cancelled run returns
    /// `Err(Cancelled)`; a non-broken-pipe stdin failure on an otherwise-
    /// successful run returns `Err(Stdin)`.
    pub(crate) async fn wait_exit(&mut self) -> Result<Outcome> {
        // Must NOT close an untaken `keep_stdin_open` pipe: `wait_any`/`wait_all`
        // borrow contenders and promise losers "remain fully usable".
        // Short-circuit when a prior reap already snapshotted `cancel_at_exit`
        // so a late cancel can't flip a cached natural exit to `Err(Cancelled)`.
        let cause = if self.cancel_at_exit.is_some() {
            ExitCause::Exited(self.backend_wait().await?)
        } else {
            // No deadline arm: a streamed run's deadline is owned by its watchdog.
            let token = self.cancel_token.clone();
            let cancelled = async {
                match &token {
                    Some(token) => token.cancelled().await,
                    None => std::future::pending::<()>().await,
                }
            };
            tokio::select! {
                biased; // cancel arm first: a cancel that fires mid-wait wins
                () = cancelled => {
                    self.kill_tree().await;
                    ExitCause::Cancelled
                }
                outcome = self.backend_wait() => ExitCause::Exited(outcome?),
            }
        };
        let outcome = self.on_reaped(cause);
        self.observe_stdin_task().await;
        self.checked_outcome(outcome)
    }

    /// Run the process to completion while sampling CPU and memory every `every`,
    /// returning a [`RunProfile`](crate::stats::RunProfile). Behaves like
    /// [`wait`](Self::wait) — output is discarded, timeout applies. A zero
    /// `every` is clamped to 1 ms.
    #[cfg(feature = "stats")]
    pub async fn profile(mut self, every: Duration) -> Result<crate::stats::RunProfile> {
        use std::sync::{Arc, Mutex};

        #[derive(Default)]
        struct Acc {
            cpu_time: Option<Duration>,
            peak_memory_bytes: Option<u64>,
            samples: usize,
        }

        // tokio panics on a zero interval period; clamp rather than panic a
        // detached sampling task on a legal-looking input.
        let every = every.max(Duration::from_millis(1));
        let started = self.started;
        let acc = Arc::new(Mutex::new(Acc::default()));
        // Set by `on_exit` once the child is reaped, so the sampler stops folding
        // readings taken against a pid the OS may have freed for reuse (which would
        // corrupt the numbers). Checked before the read and before the fold. This
        // narrows but does not fully close the recycled-pid window — the real reap
        // happens a few frames earlier in `backend_wait` — bounding it to at most
        // one stale sample, the same scheduler-quantum residual the watchdogs take.
        let reaped = Arc::new(AtomicBool::new(false));
        let sampler = self.pid.map(|pid| {
            let acc = Arc::clone(&acc);
            let reaped = Arc::clone(&reaped);
            tokio::spawn(async move {
                let mut ticker = tokio::time::interval(every);
                ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
                loop {
                    ticker.tick().await;
                    // Check before and after the read: avoids folding a reading
                    // taken against a recycled pid (reap can land mid-flight).
                    if reaped.load(Ordering::Acquire) {
                        break;
                    }
                    let metrics = crate::sys::process_metrics(pid);
                    if reaped.load(Ordering::Acquire) {
                        break;
                    }
                    if let Ok(mut acc) = acc.lock() {
                        acc.samples += 1;
                        if let Some(cpu) = metrics.cpu_time {
                            acc.cpu_time = Some(cpu);
                        }
                        if let Some(peak) = metrics.peak_memory_bytes {
                            acc.peak_memory_bytes =
                                Some(acc.peak_memory_bytes.map_or(peak, |prev| prev.max(peak)));
                        }
                    }
                }
            })
        });

        // Abort the sampler if the `profile()` future is dropped before it returns
        // (e.g. `tokio::time::timeout(d, p.profile(e))`).`on_exit` below is the
        // primary path; this is the fallback.
        struct AbortOnDrop(tokio::task::AbortHandle);
        impl Drop for AbortOnDrop {
            fn drop(&mut self) {
                self.0.abort();
            }
        }
        let _sampler_guard = sampler.as_ref().map(|h| AbortOnDrop(h.abort_handle()));

        // `reaped` flag does the real work: the sampler checks it before folding.
        // `abort` is async and the pump drain can run for PUMP_TEARDOWN on a
        // leaked pipe — long enough for a recycled pid to be read without the flag.
        let outcome = self
            .finish_lines(CaptureMode::Discard, /* expose_counts */ false, || {
                reaped.store(true, Ordering::Release);
                if let Some(task) = &sampler {
                    task.abort();
                }
            })
            .await?
            .outcome;
        let exit_code = outcome.code();
        let duration = started.elapsed();
        let (cpu_time, peak_memory_bytes, samples) = match acc.lock() {
            Ok(acc) => (acc.cpu_time, acc.peak_memory_bytes, acc.samples),
            Err(_) => (None, None, 0),
        };
        Ok(crate::stats::RunProfile {
            exit_code,
            duration,
            cpu_time,
            peak_memory_bytes,
            samples,
        })
    }

    /// Shared consuming core behind `output_string`, `wait`, and `profile`:
    /// spawn pumps, drive to exit, call `on_exit` between the await and `?`
    /// (fires even on error — `profile` uses it to abort the sampler before
    /// reap), join pumps, check cancellation, drain per `capture`.
    ///
    /// `expose_counts` stores the sinks on `self` for the live
    /// `stdout_line_count`/`stderr_line_count` accessors.
    ///
    /// `output_bytes` and `finish` deliberately do not route here — their
    /// teardown spines differ by nature.
    async fn finish_lines(
        &mut self,
        capture: CaptureMode,
        expose_counts: bool,
        on_exit: impl FnOnce(),
    ) -> Result<FinishedLines> {
        // The capturing path needs a piped stdout; fail loudly rather than return
        // empty. The discard path (wait/profile) reads nothing, so it is exempt.
        if matches!(capture, CaptureMode::Lines) {
            self.ensure_stdout_capturable()?;
        }
        // Reuse a sink already populated by a prior streaming call so that
        // output_string after stdout_lines/output_events sees those lines rather
        // than returning empty. For the discard path use a retain-nothing sink
        // (not the user's policy) so a chatty child never accumulates O(total)
        // heap in wait/profile.
        let discard_policy = OutputBufferPolicy::bounded(0);
        let sink_policy: &OutputBufferPolicy = match capture {
            CaptureMode::Discard => &discard_policy,
            CaptureMode::Lines => &self.buffer,
        };
        let stdout_sink = self
            .stdout_sink
            .clone()
            .unwrap_or_else(|| SharedLines::new(sink_policy));
        let stderr_sink = self
            .stderr_sink
            .clone()
            .unwrap_or_else(|| SharedLines::new(sink_policy));
        self.spawn_line_pumps(&stdout_sink, &stderr_sink);
        if expose_counts {
            if self.stdout_sink.is_none() {
                self.stdout_sink = Some(stdout_sink.clone());
            }
            if self.stderr_sink.is_none() {
                self.stderr_sink = Some(stderr_sink.clone());
            }
        }

        let outcome = self.drive_to_exit().await;
        on_exit();
        let outcome = outcome?;
        self.observe_stdin_task().await;
        let pumps: Vec<_> = [self.stdout_pump.take(), self.stderr_pump.take()]
            .into_iter()
            .flatten()
            .collect();
        join_pumps(pumps).await;
        let outcome = self.checked_outcome(outcome)?;

        if matches!(capture, CaptureMode::Lines) {
            for sink in [&stdout_sink, &stderr_sink] {
                if sink.overflowed() {
                    return Err(crate::Error::OutputTooLarge {
                        program: self.program.clone(),
                        line_limit: self.buffer.max_lines,
                        byte_limit: self.buffer.max_bytes,
                        total_lines: sink.count(),
                        total_bytes: sink.seen_bytes(),
                    });
                }
            }
        }

        let (stdout_lines, stderr_lines) = match capture {
            CaptureMode::Lines => (stdout_sink.drain(), stderr_sink.drain()),
            CaptureMode::Discard => (Vec::new(), Vec::new()),
        };
        Ok(FinishedLines {
            outcome,
            stdout_lines,
            stderr_lines,
        })
    }

    /// Spawn line pumps for still-untaken pipes into the given sinks.
    /// Handles stored on `self` so `Drop` aborts them on error propagation.
    fn spawn_line_pumps(&mut self, stdout_sink: &Arc<SharedLines>, stderr_sink: &Arc<SharedLines>) {
        if let Some(pipe) = self.backend.take_stdout_reader() {
            self.stdout_pump = Some(tokio::spawn(pump_lines_core(
                pipe,
                self.stdout_encoding,
                self.stdout_handler.clone(),
                self.stdout_tee.clone(),
                stdout_sink.clone(),
            )));
        }
        if let Some(pipe) = self.backend.take_stderr_reader() {
            self.stderr_pump = Some(tokio::spawn(pump_lines_core(
                pipe,
                self.stderr_encoding,
                self.stderr_handler.clone(),
                self.stderr_tee.clone(),
                stderr_sink.clone(),
            )));
        }
    }

    /// Post-exit checkpoint every consuming path passes after pumps settle:
    /// cancellation always wins (returns `Err(Cancelled)`), then a non-broken-
    /// pipe stdin failure surfaces as `Err(Stdin)` only on an otherwise-
    /// successful run.
    fn checked_outcome(&mut self, outcome: Outcome) -> Result<Outcome> {
        // Pre-pump snapshot: prevents a cancel firing during `join_pumps` from
        // discarding real output. `unwrap_or(false)` — `None` is not yet
        // snapshotted; treat conservatively as "not cancelled".
        if self.cancel_at_exit.unwrap_or(false) {
            return Err(Error::Cancelled {
                program: self.program.clone(),
            });
        }
        let succeeded = matches!(outcome, Outcome::Exited(code) if self.ok_codes.contains(&code));
        if succeeded && let Some(source) = self.stdin_error.take() {
            return Err(Error::Stdin {
                program: self.program.clone(),
                source,
            });
        }
        Ok(outcome)
    }

    /// Stash a non-broken-pipe stdin writer failure in `self.stdin_error` for
    /// `checked_outcome`. Only a writer that already finished is observed; a
    /// still-parked task is left for `Drop`'s abort.
    async fn observe_stdin_task(&mut self) {
        let task = match &mut self.backend {
            Backend::Real(real) => real.stdin_task.take(),
            Backend::Scripted(_) => None,
        };
        let Some(task) = task else {
            return;
        };
        if !task.is_finished() {
            if let Backend::Real(real) = &mut self.backend {
                real.stdin_task = Some(task);
            }
            return;
        }
        let observed = match task.await {
            Ok(Ok(())) => None,
            // Routine EPIPE (child exited before reading all stdin) — not a failure.
            Ok(Err(e)) if is_broken_pipe(&e) => None,
            Ok(Err(e)) => Some(e),
            // In practice a panic; the only abort site (`Drop`) takes the handle first.
            Err(join_err) => Some(std::io::Error::other(if join_err.is_panic() {
                format!("stdin writer task panicked: {join_err}")
            } else {
                format!("stdin writer task did not complete: {join_err}")
            })),
        };
        if let Some(e) = observed {
            #[cfg(feature = "tracing")]
            tracing::warn!(
                target: "processkit",
                program = %self.program,
                error = %e,
                "stdin writer failed"
            );
            self.stdin_error = Some(e);
        }
    }

    /// Abort all watchdog tasks and clear the recorded pid after reap.
    /// Aborting before the pid is freed limits the recycled-pid window to a
    /// scheduler quantum (an already-executing kill cannot be recalled).
    fn abort_watchdogs(&mut self) {
        self.pid = None;
        if let Some(task) = self.deadline_task.take() {
            task.abort();
        }
        if let Some(task) = self.cancel_task.take() {
            task.abort();
        }
    }

    /// Post-reap bookkeeping run in one fixed order: (1) snapshot the cancel
    /// disposition from `cause` (first-observation wins — not a post-hoc token
    /// read), (2) abort watchdogs, (3) classify a fired deadline as `TimedOut`.
    fn on_reaped(&mut self, cause: ExitCause) -> Outcome {
        if self.cancel_at_exit.is_none() {
            self.cancel_at_exit = Some(matches!(cause, ExitCause::Cancelled));
        }
        self.abort_watchdogs();
        let outcome = match cause {
            ExitCause::Exited(outcome) => outcome,
            // Moot — `checked_outcome` maps the cancel snapshot to `Err(Cancelled)`.
            ExitCause::Cancelled => Outcome::Signalled(None),
        };
        self.classify_timed_out(outcome)
    }

    /// Wait for the child to exit, applying the timeout (killing the tree on
    /// elapse). Returns the [`Outcome`] of the run.
    async fn drive_to_exit(&mut self) -> Result<Outcome> {
        // Close an untaken `keep_stdin_open` pipe so a stdin-reading child sees
        // EOF instead of blocking to its timeout.
        if let Backend::Real(real) = &mut self.backend {
            drop(real.stdin_pipe.take());
        }
        // Short-circuit when already reaped: re-running the select would fire the
        // cancel arm immediately for an already-cancelled token and overwrite the
        // snapshot. `backend_wait` returns the cached status; `on_reaped` preserves
        // the first-observation snapshot.
        let cause = if self.cancel_at_exit.is_some() {
            ExitCause::Exited(self.backend_wait().await?)
        } else {
            self.drive_to_exit_inner().await?
        };
        let outcome = self.on_reaped(cause);
        #[cfg(feature = "tracing")]
        tracing::debug!(
            target: "processkit",
            program = %self.program,
            outcome = ?outcome,
            elapsed_ms = self.started.elapsed().as_millis() as u64,
            "process exited"
        );
        Ok(outcome)
    }

    /// A fired deadline overrides whatever `backend_wait` observed — a child that
    /// exits cleanly within the grace still timed out. Cancellation is classified
    /// later in `checked_outcome` and always wins over `TimedOut`.
    fn classify_timed_out(&self, outcome: Outcome) -> Outcome {
        if self.timeout_state.load(Ordering::Acquire) == TS_TIMED_OUT {
            Outcome::TimedOut
        } else {
            outcome
        }
    }

    /// Raw exit wait — no timeout/cancel. Real: maps exit status to `Outcome`
    /// (captures Unix signal number when available). Scripted: resolves at the
    /// canned `exit_at`, or immediately as `Signalled` if killed.
    async fn backend_wait(&mut self) -> Result<Outcome> {
        let outcome = match &mut self.backend {
            Backend::Real(real) => {
                let status = real.child.wait().await.map_err(Error::Io)?;
                match status.code() {
                    Some(code) => Outcome::Exited(code),
                    None => {
                        #[cfg(unix)]
                        {
                            use std::os::unix::process::ExitStatusExt;
                            Outcome::Signalled(status.signal())
                        }
                        #[cfg(not(unix))]
                        Outcome::Signalled(None)
                    }
                }
            }
            Backend::Scripted(s) => {
                // A kill AFTER the scripted child already exited naturally must
                // still report the cached natural outcome — a real child's exit
                // status survives a post-exit kill. Only an un-exited (or
                // never-exiting `pending`) script that is killed is `Signalled`.
                let already_exited =
                    matches!(s.exit_at, Some(at) if at <= tokio::time::Instant::now());
                let classify = |s: &ScriptedProc| match (s.code, s.timed_out) {
                    (_, true) => Outcome::TimedOut,
                    (Some(code), false) => Outcome::Exited(code),
                    (None, false) => Outcome::Signalled(s.signal),
                };
                if s.kill.killed.load(Ordering::Acquire) && !already_exited {
                    Outcome::Signalled(None)
                } else if already_exited {
                    // Cached natural outcome even if a kill landed afterwards.
                    classify(s)
                } else {
                    match s.exit_at {
                        // Race so a streaming `deadline_task` can still end this wait.
                        Some(at) => {
                            tokio::select! {
                                biased;
                                () = s.kill.signal.notified() => Outcome::Signalled(None),
                                () = tokio::time::sleep_until(at) => classify(s),
                            }
                        }
                        None => {
                            s.kill.signal.notified().await;
                            Outcome::Signalled(None)
                        }
                    }
                }
            }
        };
        // Claim natural reap. If a deadline already won (`TS_TIMED_OUT`), this
        // CAS fails and the run stays `TimedOut`.
        let _ = self.timeout_state.compare_exchange(
            TS_PENDING,
            TS_EXITED,
            Ordering::AcqRel,
            Ordering::Relaxed,
        );
        Ok(outcome)
    }

    /// Race the cancel token against the deadline-bounded wait. Unset knobs
    /// become never-resolving arms. `biased` with cancel first so a simultaneous
    /// cancel+deadline always hard-kills rather than routing through the graceful
    /// teardown tier.
    async fn drive_to_exit_inner(&mut self) -> Result<ExitCause> {
        // Own the knobs so the helper futures borrow nothing from `self` —
        // only `self.backend_wait()` does, keeping the select! borrows disjoint.
        let limit = self.timeout;
        let token = self.cancel_token.clone();
        let started = self.started;
        // Disable this select's deadline arm when a streaming watchdog already
        // owns it — otherwise the graceful signal would be delivered twice.
        let watchdog_owns_deadline = self.deadline_task.is_some();
        let cancelled = async {
            match &token {
                Some(token) => token.cancelled().await,
                None => std::future::pending::<()>().await,
            }
        };
        // Anchor to spawn time so a late consuming call can't re-grant the full limit.
        let deadline = async move {
            match limit {
                Some(limit) if !watchdog_owns_deadline => {
                    let remaining = limit
                        .checked_sub(started.elapsed())
                        .unwrap_or(Duration::ZERO);
                    tokio::time::sleep(remaining).await
                }
                _ => std::future::pending::<()>().await,
            }
        };
        tokio::select! {
            biased; // cancel arm checked first: always beats a simultaneous deadline
            () = cancelled => {
                #[cfg(feature = "tracing")]
                tracing::debug!(
                    target: "processkit",
                    program = %self.program,
                    "cancellation fired; killing the tree"
                );
                self.kill_tree().await;
                Ok(ExitCause::Cancelled)
            }
            outcome = self.backend_wait() => outcome.map(ExitCause::Exited),
            () = deadline => {
                #[cfg(feature = "tracing")]
                tracing::warn!(
                    target: "processkit",
                    program = %self.program,
                    timeout_ms = limit.map(|l| l.as_millis() as u64).unwrap_or(0),
                    "timeout elapsed; killing the tree"
                );
                let _ = self.timeout_state.compare_exchange(
                    TS_PENDING,
                    TS_TIMED_OUT,
                    Ordering::AcqRel,
                    Ordering::Relaxed,
                );
                self.teardown_on_timeout().await;
                Ok(ExitCause::Exited(Outcome::TimedOut))
            }
        }
    }

    /// Hard-kill the child and its tree (for a private group), then reap.
    async fn kill_tree(&mut self) {
        match &mut self.backend {
            Backend::Real(real) => {
                let _ = real.child.start_kill();
                if let Some(group) = &real.own_group {
                    let _ = group.terminate_all();
                }
                // Bound the reap: a D-state child can ignore SIGKILL until I/O
                // unblocks, and an unbounded wait hangs shared-group handles.
                let _ = tokio::time::timeout(PUMP_TEARDOWN, real.child.wait()).await;
            }
            Backend::Scripted(s) => s.kill(),
        }
    }

    /// Teardown when the deadline elapses. With `timeout_grace`: signal → wait up
    /// to grace → SIGKILL, concurrent with the reap so a signal-handling child
    /// ends the grace early. Without grace: hard `kill_tree`. Windows has no signal
    /// tier; graceful degrades to the atomic kill.
    async fn teardown_on_timeout(&mut self) {
        let Some(grace) = self.timeout_grace else {
            self.kill_tree().await;
            return;
        };
        let signal = self.timeout_signal;
        match &mut self.backend {
            Backend::Real(real) => {
                let pid = real.child.id();
                let own = real.own_group.clone();
                let teardown = async {
                    match &own {
                        Some(group) => {
                            let _ = group.graceful_terminate(grace, signal).await;
                        }
                        // Shared group: terminate only our direct child.
                        None => {
                            crate::running::stream::graceful_kill_pid(pid, grace, signal).await;
                        }
                    }
                };
                // Bound the reap: a D-state child can ignore the final SIGKILL.
                let reap =
                    tokio::time::timeout(grace.saturating_add(PUMP_TEARDOWN), real.child.wait());
                let _ = tokio::join!(teardown, reap);
            }
            Backend::Scripted(s) => s.kill(),
        }
    }

    /// Whether the child has already exited, polled without blocking.
    fn has_exited_now(&mut self) -> bool {
        let exited = match &mut self.backend {
            Backend::Real(real) => matches!(real.child.try_wait(), Ok(Some(_))),
            Backend::Scripted(s) => {
                s.kill.killed.load(Ordering::Acquire)
                    || s.exit_at
                        .is_some_and(|at| tokio::time::Instant::now() >= at)
            }
        };
        if exited {
            // Claim the arbiter: a deadline watchdog racing on another thread could
            // win `PENDING -> TIMED_OUT` before `abort_watchdogs` stops it,
            // misclassifying a clean exit. Claiming `EXITED` closes that window.
            let _ = self.timeout_state.compare_exchange(
                TS_PENDING,
                TS_EXITED,
                Ordering::AcqRel,
                Ordering::Relaxed,
            );
            self.abort_watchdogs();
            // Snapshot cancel disposition on FIRST reap only — a repeat probe
            // after a late cancel must not overwrite the cached result.
            if self.cancel_at_exit.is_none() {
                self.cancel_at_exit =
                    Some(self.cancel_token.as_ref().is_some_and(|t| t.is_cancelled()));
            }
        }
        exited
    }

    /// Send a kill to the process without waiting for it to exit. The owning
    /// group still governs the rest of the tree.
    ///
    /// The [`Outcome`] afterwards is platform-dependent: `Signalled` on Unix,
    /// `Exited` with a platform code on Windows. A scripted handle reports
    /// `Signalled(None)`.
    ///
    /// **Idempotent:** killing an already-reaped child is a successful no-op.
    pub fn start_kill(&mut self) -> Result<()> {
        match &mut self.backend {
            Backend::Real(real) => match real.child.start_kill() {
                Ok(()) => {}
                // tokio/std currently return `Ok` for a reaped child; treat
                // `InvalidInput` as the same no-op in case that ever changes.
                Err(e) if e.kind() == std::io::ErrorKind::InvalidInput => {}
                Err(e) => return Err(Error::Io(e)),
            },
            Backend::Scripted(s) => s.kill(),
        }
        Ok(())
    }
}

impl Drop for RunningProcess {
    fn drop(&mut self) {
        match &mut self.backend {
            Backend::Real(real) => {
                if let Some(task) = real.stdin_task.take() {
                    task.abort();
                }
            }
            Backend::Scripted(s) => s.kill(),
        }
        if let Some(task) = self.deadline_task.take() {
            task.abort();
        }
        if let Some(task) = self.cancel_task.take() {
            task.abort();
        }
        // A surviving grandchild holding the pipe could keep a pump alive
        // indefinitely on a shared-group handle without this abort.
        if let Some(task) = self.stdout_pump.take() {
            task.abort();
        }
        if let Some(task) = self.stderr_pump.take() {
            task.abort();
        }
    }
}

/// Whether `e` is the routine pipe-closed write error — `BrokenPipe`, plus the
/// raw Windows encodings (`ERROR_BROKEN_PIPE` = 109, `ERROR_NO_DATA` = 232)
/// that don't always map to the kind.
fn is_broken_pipe(e: &std::io::Error) -> bool {
    e.kind() == std::io::ErrorKind::BrokenPipe || matches!(e.raw_os_error(), Some(109 | 232))
}

/// Await the output pumps, bounded by [`PUMP_TEARDOWN`]; abort stragglers.
async fn join_pumps(tasks: Vec<JoinHandle<()>>) {
    if tasks.is_empty() {
        return;
    }
    let aborts: Vec<_> = tasks.iter().map(|t| t.abort_handle()).collect();
    let join = async {
        for task in tasks {
            // A panicking pump closes its sink via close-on-drop: partial output
            // is intact. Surface it for diagnostics, never as a run error.
            #[cfg(feature = "tracing")]
            if let Err(e) = task.await {
                tracing::warn!(target: "processkit", error = %e, "output pump task ended abnormally");
            }
            #[cfg(not(feature = "tracing"))]
            let _ = task.await;
        }
    };
    if tokio::time::timeout(PUMP_TEARDOWN, join).await.is_err() {
        // A pipe is still held open past the child's death (the surviving-
        // grandchild case PUMP_TEARDOWN exists for) — abort and keep what
        // arrived.
        #[cfg(feature = "tracing")]
        tracing::warn!(
            target: "processkit",
            timeout_ms = PUMP_TEARDOWN.as_millis() as u64,
            aborted = aborts.len(),
            "output pumps overran teardown grace; aborting stragglers"
        );
        for abort in aborts {
            abort.abort();
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::command::Command;
    use crate::doubles::{Reply, ScriptedRunner};
    use crate::runner::ProcessRunner;

    /// A scripted (hermetic) handle for `tool`, with the given `ok_codes`.
    async fn scripted_handle(ok_codes: &[i32]) -> RunningProcess {
        let cmd = Command::new("tool").ok_codes(ok_codes.iter().copied());
        ScriptedRunner::new()
            .fallback(Reply::ok(""))
            .start(&cmd)
            .await
            .expect("scripted start")
    }

    /// A stashed non-broken-pipe stdin failure surfaces as `Error::Stdin` only on
    /// an otherwise-successful outcome; a non-zero exit or a signal is the "realer"
    /// failure and wins (outcome passed through).
    #[tokio::test]
    async fn stdin_error_surfaces_only_on_a_successful_outcome() {
        let mut run = scripted_handle(&[0]).await;
        run.stdin_error = Some(std::io::Error::other("boom"));
        match run.checked_outcome(Outcome::Exited(0)) {
            Err(Error::Stdin { program, source }) => {
                assert_eq!(program, "tool");
                assert_eq!(source.to_string(), "boom");
            }
            other => panic!("expected Error::Stdin, got {other:?}"),
        }

        // Non-zero exit wins: outcome returned for the caller's classifier.
        let mut run = scripted_handle(&[0]).await;
        run.stdin_error = Some(std::io::Error::other("boom"));
        assert!(matches!(
            run.checked_outcome(Outcome::Exited(7)),
            Ok(Outcome::Exited(7))
        ));

        // A signal wins too (not a success).
        let mut run = scripted_handle(&[0]).await;
        run.stdin_error = Some(std::io::Error::other("boom"));
        assert!(matches!(
            run.checked_outcome(Outcome::Signalled(Some(9))),
            Ok(Outcome::Signalled(Some(9)))
        ));
    }

    /// The success gate honors `ok_codes`: a code widened to "accepted" is a
    /// success, so the stdin failure surfaces there too.
    #[tokio::test]
    async fn stdin_error_respects_ok_codes_widened_success() {
        let mut run = scripted_handle(&[0, 3]).await;
        run.stdin_error = Some(std::io::Error::other("boom"));
        assert!(matches!(
            run.checked_outcome(Outcome::Exited(3)),
            Err(Error::Stdin { .. })
        ));
    }

    #[tokio::test]
    async fn no_stdin_error_is_a_clean_passthrough() {
        let mut run = scripted_handle(&[0]).await;
        assert!(matches!(
            run.checked_outcome(Outcome::Exited(0)),
            Ok(Outcome::Exited(0))
        ));
    }

    /// `output_string` after a partial `stdout_lines` stream must NOT report
    /// truncation under the default unbounded policy — the consumed lines were
    /// popped by the stream, not discarded by the buffer.
    #[tokio::test]
    async fn output_string_after_partial_stream_is_not_truncated() {
        use tokio_stream::StreamExt;

        let mut run = ScriptedRunner::new()
            .fallback(Reply::lines(["a", "b", "c", "d"]))
            .start(&Command::new("tool"))
            .await
            .expect("scripted start");

        {
            let mut lines = run.stdout_lines().unwrap();
            assert_eq!(lines.next().await.as_deref(), Some("a"));
            assert_eq!(lines.next().await.as_deref(), Some("b"));
        }

        let result = run.output_string().await.expect("output_string");
        assert!(
            !result.truncated(),
            "consumed lines are not truncation under unbounded policy: {result:?}"
        );
        assert_eq!(
            result.stdout(),
            "c\nd",
            "output_string returns the unconsumed tail"
        );
    }

    #[tokio::test]
    async fn output_bytes_after_streaming_errors_instead_of_empty() {
        let mut run = ScriptedRunner::new()
            .fallback(Reply::lines(["a", "b"]))
            .start(&Command::new("tool"))
            .await
            .expect("scripted start");

        drop(run.stdout_lines().unwrap());

        let err = run
            .output_bytes()
            .await
            .expect_err("output_bytes after streaming must error, not return empty");
        match err {
            Error::Io(e) => assert_eq!(e.kind(), std::io::ErrorKind::InvalidInput),
            other => panic!("expected Io(InvalidInput), got {other:?}"),
        }
    }

    /// Other direction: a bounded buffer that genuinely discards lines during
    /// streaming must STILL report `truncated=true` — narrowing only the false
    /// positive (consumed-by-stream), never masking real truncation. Filling
    /// `bounded(2)` with four un-consumed lines drops two deterministically.
    #[tokio::test]
    async fn output_string_after_stream_still_reports_real_truncation() {
        let cmd = Command::new("tool").output_buffer(OutputBufferPolicy::bounded(2));
        let mut run = ScriptedRunner::new()
            .fallback(Reply::lines(["a", "b", "c", "d"]))
            .start(&cmd)
            .await
            .expect("scripted start");

        drop(run.stdout_lines().unwrap());

        let result = run.output_string().await.expect("output_string");
        assert!(
            result.truncated(),
            "a bounded buffer that dropped lines during streaming must report truncation: {result:?}"
        );
    }

    #[tokio::test]
    async fn output_bytes_returns_exact_raw_stdout() {
        let result = ScriptedRunner::new()
            .fallback(Reply::ok("raw\u{0}bytes\nno trailing newline"))
            .start(&Command::new("tool"))
            .await
            .expect("scripted start")
            .output_bytes()
            .await
            .expect("output_bytes");
        assert_eq!(result.stdout(), b"raw\x00bytes\nno trailing newline");
        assert!(!result.truncated(), "no policy drop: {result:?}");
    }

    /// A `TS_TIMED_OUT` arbiter state overrides `backend_wait`'s clean exit 0.
    #[tokio::test]
    async fn timed_out_flag_classifies_a_clean_exit_as_timed_out() {
        let run = scripted_handle(&[0]).await; // Reply::ok -> Exited(0)
        run.timeout_state.store(TS_TIMED_OUT, Ordering::Release); // simulate the watchdog firing
        let outcome = run.wait().await.expect("wait");
        assert_eq!(
            outcome,
            Outcome::TimedOut,
            "a run whose deadline fired must report TimedOut, not the in-grace exit"
        );
    }

    /// Cancellation is checked after `classify_timed_out` and always wins.
    #[tokio::test]
    async fn cancellation_beats_the_timed_out_flag() {
        let token = crate::CancellationToken::new();
        let run = ScriptedRunner::new()
            .fallback(Reply::ok(""))
            .start(&Command::new("tool").cancel_on(token.clone()))
            .await
            .expect("scripted start");
        run.timeout_state.store(TS_TIMED_OUT, Ordering::Release);
        token.cancel();
        match run.wait().await {
            Err(Error::Cancelled { .. }) => {}
            other => panic!("expected Err(Cancelled), got {other:?}"),
        }
    }

    /// Cancel disposition is the race result (`ExitCause`), not a post-hoc read.
    /// A token cancelled after a natural `wait_any` reap cannot flip the cached exit.
    #[tokio::test]
    async fn a_natural_wait_any_exit_is_not_flipped_by_a_late_cancel() {
        let token = crate::CancellationToken::new();
        let mut run = ScriptedRunner::new()
            .fallback(Reply::ok("done\n"))
            .start(&Command::new("tool").cancel_on(token.clone()))
            .await
            .expect("scripted start");
        let (idx, outcome) = crate::wait_any(&mut [&mut run]).await.expect("wait_any");
        assert_eq!((idx, outcome), (0, Outcome::Exited(0)));
        token.cancel();
        let outcome = run
            .wait()
            .await
            .expect("a late cancel must not flip a natural exit");
        assert_eq!(outcome, Outcome::Exited(0));
    }

    /// `wait_exit` applies `classify_timed_out` so the `stdout_lines` → `wait_any`
    /// composition is consistent with `finish`.
    #[tokio::test]
    async fn wait_any_classifies_a_timed_out_run() {
        let mut run = scripted_handle(&[0]).await; // Reply::ok -> Exited(0)
        run.timeout_state.store(TS_TIMED_OUT, Ordering::Release); // simulate the watchdog firing
        let (idx, outcome) = crate::wait_any(&mut [&mut run]).await.expect("wait_any");
        assert_eq!(idx, 0);
        assert_eq!(
            outcome,
            Outcome::TimedOut,
            "a timed-out run must report TimedOut through wait_any, not the raw exit"
        );
    }

    /// The timeout arbiter is race-free. Once the natural reap claims
    /// `EXITED`, a watchdog whose timer fires late cannot flip the run to
    /// `TimedOut` (its CAS from `PENDING` fails), so a child that exits on its own
    /// within a scheduler quantum of the deadline keeps its real outcome. (The
    /// reverse — the deadline claiming `TIMED_OUT` first — is covered by
    /// `timed_out_flag_classifies_a_clean_exit_as_timed_out`.)
    #[tokio::test]
    async fn natural_reap_claim_beats_a_late_timeout_cas() {
        let run = scripted_handle(&[0]).await;
        assert!(
            run.timeout_state
                .compare_exchange(TS_PENDING, TS_EXITED, Ordering::AcqRel, Ordering::Relaxed)
                .is_ok()
        );
        assert!(
            run.timeout_state
                .compare_exchange(
                    TS_PENDING,
                    TS_TIMED_OUT,
                    Ordering::AcqRel,
                    Ordering::Relaxed
                )
                .is_err()
        );
        assert_eq!(
            run.classify_timed_out(Outcome::Exited(0)),
            Outcome::Exited(0)
        );
    }

    #[tokio::test]
    async fn scripted_handle_does_not_kill_a_tree_on_drop() {
        let run = scripted_handle(&[0]).await;
        assert!(
            !run.kills_tree_on_drop(),
            "a scripted double has no OS tree to tear down"
        );
    }

    #[tokio::test]
    async fn capture_verbs_error_on_a_non_piped_stdout() {
        let runner = ScriptedRunner::new().fallback(Reply::ok("ignored"));

        let run = runner
            .start(&Command::new("tool").stdout(crate::StdioMode::Null))
            .await
            .unwrap();
        match run.output_string().await {
            Err(Error::Io(e)) => assert_eq!(e.kind(), std::io::ErrorKind::InvalidInput),
            other => panic!("expected Io(InvalidInput), got {other:?}"),
        }

        // output_bytes on an Inherit stdout → also errors.
        let run = runner
            .start(&Command::new("tool").stdout(crate::StdioMode::Inherit))
            .await
            .unwrap();
        assert!(matches!(run.output_bytes().await, Err(Error::Io(_))));

        let run = ScriptedRunner::new()
            .fallback(Reply::ok("hi"))
            .start(&Command::new("tool"))
            .await
            .unwrap();
        assert_eq!(run.output_string().await.unwrap().stdout(), "hi");

        let run = runner
            .start(&Command::new("tool").stdout(crate::StdioMode::Null))
            .await
            .unwrap();
        assert!(
            run.wait().await.is_ok(),
            "discard verbs do not require a piped stdout"
        );
    }
}