coreshift_core/spawn/

mod.rs

// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at https://mozilla.org/MPL/2.0/

//! Process spawning and lifecycle management.
//!
//! This module exposes explicit Linux/Android process primitives. Callers must
//! provide the exact argument vector and choose the spawn backend. Core does not
//! infer shell/root behavior, select backends from platform properties, or
//! silently switch between backends.

use std::os::unix::io::RawFd;
use std::time::{Duration, Instant};

use crate::CoreError;
use crate::error::syscall_ret;
use crate::fd::Fd;
use crate::io::DrainState;
use crate::reactor::Reactor;
use libc::{O_CLOEXEC, WEXITSTATUS, WIFEXITED, WIFSIGNALED, WTERMSIG, pid_t, pipe2, waitpid};
use std::collections::HashSet;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Mutex, OnceLock};

mod exec;
mod fork;
mod posix;

use exec::ExecContext;
use fork::spawn_fork_internal;
use posix::spawn_posix_internal;

unsafe extern "C" {
    pub(crate) static mut environ: *mut *mut libc::c_char;
}

/// Upper bound on how long to keep polling for a reap after SIGKILL has been
/// sent. A child stuck in uninterruptible sleep (D-state) cannot be reaped at
/// all — SIGKILL stays pending until it leaves D-state — so after this window
/// the wait loop gives up and returns the partial output instead of spinning
/// forever. Mirrors the bounded reap wait in [`ManagedProcess`]'s `Drop`.
const D_STATE_REAP_BOUND: Duration = Duration::from_millis(500);

/// Orphaned children: processes this library spawned whose caller will never
/// call `wait` (the `wait = false` path) or that the wait loop gave up on
/// reaping (D-state / cancel-timeout give-up). A reaper thread `waitpid`s each
/// registered pid so they do not accumulate as zombies — a long-lived daemon
/// that detaches children would otherwise exhaust the pid space (finding 15).
///
/// Only *registered* pids are reaped. A global `waitpid(-1)` loop would race
/// with callers explicitly waiting on other children of this process; targeting
/// registered pids is safe because they are our own direct children — the pid
/// cannot be recycled until we reap it.
static ORPHANED: OnceLock<Mutex<HashSet<pid_t>>> = OnceLock::new();
static REAPER_STARTED: AtomicBool = AtomicBool::new(false);

/// Register `pid` as orphaned (nobody will `wait` on it) and ensure the
/// background reaper is running. No-op if the pid is already registered.
fn orphan_child(pid: pid_t) {
    ORPHANED
        .get_or_init(|| Mutex::new(HashSet::new()))
        .lock()
        .unwrap()
        .insert(pid);
    start_reaper();
}

/// Spawn (once) the background reaper thread that reaps [`ORPHANED`] pids.
fn start_reaper() {
    if REAPER_STARTED.load(Ordering::SeqCst) {
        return;
    }
    let r = REAPER_STARTED.compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst);
    if r.is_err() {
        return;
    }
    std::thread::Builder::new()
        .name("spawn-orphan-reaper".into())
        .spawn(reap_orphaned)
        .ok();
}

/// Reaper body: periodically `waitpid` (non-blocking) every orphaned pid and
/// drop it from the set once it has been reaped (or is already gone, which can
/// only mean it was reaped elsewhere — the pid was still registered).
fn reap_orphaned() {
    loop {
        let pids: Vec<pid_t> = ORPHANED
            .get_or_init(|| Mutex::new(HashSet::new()))
            .lock()
            .unwrap()
            .iter()
            .copied()
            .collect();
        let mut still_orphaned = Vec::new();
        for pid in pids {
            let mut status: libc::c_int = 0;
            let r = unsafe { waitpid(pid, &mut status, libc::WNOHANG) };
            if r == pid
                || (r < 0 && std::io::Error::last_os_error().raw_os_error() == Some(libc::ECHILD))
            {
                continue; // reaped or gone — drop from the set
            }
            still_orphaned.push(pid);
        }
        if !still_orphaned.is_empty() {
            if let Some(set) = ORPHANED.get() {
                if let Ok(mut guard) = set.lock() {
                    for pid in still_orphaned {
                        guard.insert(pid);
                    }
                }
            }
        }
        std::thread::sleep(Duration::from_millis(250));
    }
}

/// Detach a pid from the orphan set (used when a previously-orphaned process
/// turns out to be waitable again; currently unused by callers but keeps the
/// registry honest).
#[allow(dead_code)]
fn deorphan_child(pid: pid_t) {
    if let Some(set) = ORPHANED.get() {
        if let Ok(mut guard) = set.lock() {
            guard.remove(&pid);
        }
    }
}

/// Policy for handling process cancellation or timeouts.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum CancelPolicy {
    /// Do nothing on cancellation; let the process run to completion.
    #[default]
    None,
    /// Send SIGTERM, then SIGKILL after a grace period.
    Graceful,
    /// Send SIGKILL immediately.
    Kill,
}

/// Process group and session configuration.
#[derive(Debug, Clone, Copy, Default)]
pub struct ProcessGroup {
    /// Join an existing process group leader.
    pub leader: Option<pid_t>,
    /// Create a new session (`setsid`).
    pub isolated: bool,
}

impl ProcessGroup {
    /// Create a new process group configuration.
    pub fn new(leader: Option<pid_t>, isolated: bool) -> Self {
        Self { leader, isolated }
    }
}

#[inline(always)]
fn errno() -> i32 {
    std::io::Error::last_os_error().raw_os_error().unwrap_or(0)
}

/// Relocate `fd` to the lowest available descriptor `>= 3`, closing the
/// original. Guards against `pipe2` handing back fds 0/1/2 when the daemon
/// runs with stdio closed: a pipe on 0/1/2 would collide with the child's
/// `dup2(…, 0/1/2)` setup (clobbering a still-needed end) and with the
/// stdio-tracking in `close_child_fds_for_policy`.
fn relocate_above_stdio(fd: RawFd, op: &'static str) -> Result<RawFd, CoreError> {
    if fd >= 3 {
        return Ok(fd);
    }
    let new = unsafe { libc::fcntl(fd, libc::F_DUPFD_CLOEXEC, 3) };
    syscall_ret(new, op)?;
    unsafe {
        libc::close(fd);
    }
    Ok(new)
}

/// Creates a pipe with O_CLOEXEC, relocated above stdio. Both ends stay
/// blocking; the parent-facing ends are flipped to O_NONBLOCK by
/// [`DrainState`] after spawn so the child never inherits a non-blocking
/// stdio (which would silently truncate child output on `EAGAIN`).
/// Invariants: FDs returned are strictly >= 3 and will close automatically on drop.
#[inline(always)]
fn make_pipe() -> Result<(Fd, Fd), CoreError> {
    let mut fds = [0; 2];
    let r = unsafe { pipe2(fds.as_mut_ptr(), O_CLOEXEC) };
    syscall_ret(r, "pipe2")?;
    let r0 = match relocate_above_stdio(fds[0], "pipe2:relocate") {
        Ok(fd) => fd,
        Err(e) => {
            // fds[0] is still open when its relocation fails; close to avoid
            // leaking under fd pressure (EMFILE).
            unsafe {
                libc::close(fds[0]);
            }
            return Err(e);
        }
    };
    let r1 = match relocate_above_stdio(fds[1], "pipe2:relocate") {
        Ok(fd) => fd,
        Err(e) => {
            // fds[1] is still open (relocation failed), and r0 was relocated
            // above — both would leak on this error path.
            unsafe {
                libc::close(r0);
                libc::close(fds[1]);
            }
            return Err(e);
        }
    };
    Ok((Fd::new(r0, "pipe2")?, Fd::new(r1, "pipe2")?))
}

fn make_cloexec_pipe() -> Result<(RawFd, RawFd), CoreError> {
    let mut fds = [0; 2];
    let r = unsafe { pipe2(fds.as_mut_ptr(), O_CLOEXEC) };
    syscall_ret(r, "pipe2")?;
    let r0 = match relocate_above_stdio(fds[0], "pipe2:relocate") {
        Ok(fd) => fd,
        Err(e) => {
            unsafe {
                libc::close(fds[0]);
            }
            return Err(e);
        }
    };
    let r1 = match relocate_above_stdio(fds[1], "pipe2:relocate") {
        Ok(fd) => fd,
        Err(e) => {
            unsafe {
                libc::close(r0);
                libc::close(fds[1]);
            }
            return Err(e);
        }
    };
    Ok((r0, r1))
}

struct Pipes {
    stdin_r: Option<Fd>,
    stdin_w: Option<Fd>,
    stdout_r: Option<Fd>,
    stdout_w: Option<Fd>,
    stderr_r: Option<Fd>,
    stderr_w: Option<Fd>,
}

impl Pipes {
    fn new(in_buf: Option<&[u8]>, out: bool, err: bool) -> Result<Self, CoreError> {
        let (stdin_r, stdin_w) = if in_buf.is_some() {
            let (r, w) = make_pipe()?;
            (Some(r), Some(w))
        } else {
            (None, None)
        };

        let (stdout_r, stdout_w) = if out {
            let (r, w) = make_pipe()?;
            (Some(r), Some(w))
        } else {
            (None, None)
        };

        let (stderr_r, stderr_w) = if err {
            let (r, w) = make_pipe()?;
            (Some(r), Some(w))
        } else {
            (None, None)
        };

        Ok(Self {
            stdin_r,
            stdin_w,
            stdout_r,
            stdout_w,
            stderr_r,
            stderr_w,
        })
    }

    #[inline(always)]
    fn close_all(&mut self) {
        self.stdin_r.take();
        self.stdin_w.take();
        self.stdout_r.take();
        self.stdout_w.take();
        self.stderr_r.take();
        self.stderr_w.take();
    }
}

/// Represents the termination status of a process.
#[derive(Debug, PartialEq, Eq)]
pub enum ExitStatus {
    /// Process exited normally with the specified code.
    Exited(i32),
    /// Process was terminated by a signal.
    Signaled(i32),
}

/// Explicit process spawning backend.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SpawnBackend {
    /// Force the use of `posix_spawn`.
    PosixSpawn,
    /// Force the use of `fork`/`exec`.
    ///
    /// The fork backend supports explicit [`SpawnFdPolicy`] handling before
    /// `execve`.
    Fork,
}

/// Explicit file-descriptor inheritance policy for spawned children.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub enum SpawnFdPolicy {
    /// Inherit descriptors according to their existing `FD_CLOEXEC` flags.
    #[default]
    CloexecOnly,
    /// For the fork backend, close every descriptor >= 3 before `execve`,
    /// except Core-required pipe descriptors.
    CloseFrom3,
    /// For the fork backend, close every descriptor >= 3 before `execve`,
    /// except Core-required pipe descriptors and the listed descriptors.
    ///
    /// Core does not close allowlisted descriptors, but their existing
    /// `FD_CLOEXEC` state still applies. Callers that want an allowlisted
    /// descriptor to survive `execve` must clear `FD_CLOEXEC` before spawning.
    Allowlist(Vec<RawFd>),
}

#[inline(always)]
fn decode_status(status: i32) -> ExitStatus {
    if WIFEXITED(status) {
        ExitStatus::Exited(WEXITSTATUS(status))
    } else if WIFSIGNALED(status) {
        ExitStatus::Signaled(WTERMSIG(status))
    } else {
        ExitStatus::Exited(-1)
    }
}

/// A handle to a spawned process.
///
/// ### Fork Safety
/// The process handle contains a PID. After a `fork`, the child process will
/// have a copy of this PID, but it refers to the same original process.
/// Calling `wait` or `kill` from the child may lead to confusing results
/// if multiple processes are managing the same PID.
pub struct Process {
    pid: pid_t,
}

impl Process {
    /// Create a handle for an existing PID.
    pub fn new(pid: pid_t) -> Self {
        Self { pid }
    }

    /// Return the process ID.
    pub fn pid(&self) -> pid_t {
        self.pid
    }

    /// Perform a non-blocking wait for process termination.
    ///
    /// ### Errors
    /// - `ECHILD`: The process does not exist or is not a child of the caller.
    /// - `EINTR`: The call was interrupted by a signal (handled internally).
    pub fn wait_step(&self) -> Result<Option<ExitStatus>, CoreError> {
        loop {
            let mut status = 0;
            let r = unsafe { waitpid(self.pid, &mut status, libc::WNOHANG) };
            if r == 0 {
                return Ok(None);
            }
            if r < 0 {
                let e = errno();
                if e == libc::EINTR {
                    continue;
                }
                return Err(CoreError::sys(e, "waitpid_step"));
            }
            return Ok(Some(decode_status(status)));
        }
    }

    /// Block until the process terminates.
    ///
    /// ### Errors
    /// - `ECHILD`: The process does not exist or is not a child of the caller.
    pub fn wait_blocking(&self) -> Result<ExitStatus, CoreError> {
        loop {
            let mut status = 0;
            let r = unsafe { waitpid(self.pid, &mut status, 0) };
            if r < 0 {
                let e = errno();
                if e == libc::EINTR {
                    continue;
                }
                return Err(CoreError::sys(e, "waitpid_blocking"));
            }
            return Ok(decode_status(status));
        }
    }

    /// Send a signal to the process.
    ///
    /// ### Errors
    /// - `EINVAL`: Invalid signal number, or a non-positive pid (pid `0`
    ///   would signal the caller's own process group).
    /// - `EPERM`: The caller does not have permission to send the signal.
    /// - `ESRCH`: The process does not exist.
    pub fn kill(&self, sig: i32) -> Result<(), CoreError> {
        if self.pid <= 0 {
            return Err(CoreError::sys(libc::EINVAL, "kill: invalid pid"));
        }
        let r = unsafe { libc::kill(self.pid, sig) };
        if r < 0 {
            let e = errno();
            if e == libc::ESRCH {
                return Ok(());
            }
            syscall_ret(-1, "kill")?;
        }
        Ok(())
    }

    /// Signal the process group whose id equals [`Self::pid`] — valid only
    /// when the process is its own group/session leader. For a child placed
    /// into a custom leader's group use [`Self::kill_group`].
    ///
    /// ### Errors
    /// Same as [`Self::kill`].
    pub fn kill_pgroup(&self, sig: i32) -> Result<(), CoreError> {
        self.kill_group(self.pid, sig)
    }

    /// Send a signal to an explicit process group.
    ///
    /// The pgid must be the child's actual group (its own pid after `setsid`,
    /// or the configured leader's id after `setpgid`), never guessed from the
    /// pid, and never `0` or negative — `kill(-0)` would signal the caller's
    /// own process group.
    ///
    /// ### Errors
    /// Same as [`Self::kill`], plus `EINVAL` for a non-positive pgid.
    pub fn kill_group(&self, pgid: pid_t, sig: i32) -> Result<(), CoreError> {
        if pgid <= 0 {
            return Err(CoreError::sys(libc::EINVAL, "kill_group: invalid pgid"));
        }
        let r = unsafe { libc::kill(-pgid, sig) };
        if r < 0 {
            let e = errno();
            if e == libc::ESRCH {
                return Ok(());
            }
            syscall_ret(-1, "kill_group")?;
        }
        Ok(())
    }
}

/// Configuration options for spawning a new process.
#[derive(Clone)]
pub struct SpawnOptions {
    ctx: ExecContext,
    stdin: Option<Box<[u8]>>,
    capture_stdout: bool,
    capture_stderr: bool,
    wait: bool,
    pgroup: ProcessGroup,
    max_output: usize,
    timeout_ms: Option<u32>,
    kill_grace_ms: u32,
    cancel: CancelPolicy,
    backend: SpawnBackend,
    fd_policy: SpawnFdPolicy,
    early_exit: Option<fn(&[u8]) -> bool>,
}

impl SpawnOptions {
    /// Create a new builder for process spawning.
    pub fn builder(argv: Vec<String>, backend: SpawnBackend) -> SpawnOptionsBuilder {
        SpawnOptionsBuilder::new(argv, backend)
    }

    /// Execute the process according to the options and block until completion.
    pub fn run(self) -> Result<Output, CoreError> {
        spawn(self)
    }
}

/// Builder for [`SpawnOptions`].
#[derive(Clone)]
pub struct SpawnOptionsBuilder {
    argv: Vec<String>,
    env: Option<Vec<String>>,
    cwd: Option<String>,
    stdin: Option<Box<[u8]>>,
    capture_stdout: bool,
    capture_stderr: bool,
    wait: bool,
    pgroup: ProcessGroup,
    max_output: usize,
    timeout_ms: Option<u32>,
    kill_grace_ms: u32,
    cancel: CancelPolicy,
    backend: SpawnBackend,
    fd_policy: SpawnFdPolicy,
    early_exit: Option<fn(&[u8]) -> bool>,
}

impl SpawnOptionsBuilder {
    /// Create a new builder with the specified argument vector.
    pub fn new(argv: Vec<String>, backend: SpawnBackend) -> Self {
        Self {
            argv,
            env: None,
            cwd: None,
            stdin: None,
            capture_stdout: false,
            capture_stderr: false,
            wait: true,
            pgroup: ProcessGroup::default(),
            max_output: 1024 * 1024,
            timeout_ms: None,
            kill_grace_ms: 2000,
            cancel: CancelPolicy::Kill,
            backend,
            fd_policy: SpawnFdPolicy::default(),
            early_exit: None,
        }
    }

    /// Set environment variables.
    pub fn env(mut self, env: Vec<String>) -> Self {
        self.env = Some(env);
        self
    }

    /// Set the working directory.
    pub fn cwd(mut self, cwd: String) -> Self {
        self.cwd = Some(cwd);
        self
    }

    /// Provide data to be written to the child's stdin.
    pub fn stdin(mut self, data: impl Into<Box<[u8]>>) -> Self {
        self.stdin = Some(data.into());
        self
    }

    /// Enable stdout capture.
    pub fn capture_stdout(mut self) -> Self {
        self.capture_stdout = true;
        self
    }

    /// Enable stderr capture.
    pub fn capture_stderr(mut self) -> Self {
        self.capture_stderr = true;
        self
    }

    /// Set whether to wait for the process to terminate (default: true).
    pub fn wait(mut self, wait: bool) -> Self {
        self.wait = wait;
        self
    }

    /// Set process group and isolation policy.
    pub fn pgroup(mut self, pgroup: ProcessGroup) -> Self {
        self.pgroup = pgroup;
        self
    }

    /// Set the combined stdout+stderr output buffer size (default: 1MB).
    ///
    /// If captured output exceeds this limit, spawn drains the child pipes to
    /// completion and returns `EOVERFLOW`.
    pub fn max_output(mut self, max: usize) -> Self {
        self.max_output = max;
        self
    }

    /// Set the execution timeout in milliseconds.
    pub fn timeout_ms(mut self, ms: u32) -> Self {
        self.timeout_ms = Some(ms);
        self
    }

    /// Set the grace period before SIGKILL (default: 2s).
    pub fn kill_grace_ms(mut self, ms: u32) -> Self {
        self.kill_grace_ms = ms;
        self
    }

    /// Set the cancellation policy (default: Kill).
    pub fn cancel(mut self, policy: CancelPolicy) -> Self {
        self.cancel = policy;
        self
    }

    /// Set the child file-descriptor inheritance policy.
    pub fn fd_policy(mut self, policy: SpawnFdPolicy) -> Self {
        self.fd_policy = policy;
        self
    }

    /// Set an early exit callback.
    pub fn early_exit(mut self, callback: fn(&[u8]) -> bool) -> Self {
        self.early_exit = Some(callback);
        self
    }

    /// Build the spawn options.
    pub fn build(self) -> Result<SpawnOptions, CoreError> {
        let ctx = ExecContext::new(self.argv, self.env, self.cwd)?;
        Ok(SpawnOptions {
            ctx,
            stdin: self.stdin,
            capture_stdout: self.capture_stdout,
            capture_stderr: self.capture_stderr,
            wait: self.wait,
            pgroup: self.pgroup,
            max_output: self.max_output,
            timeout_ms: self.timeout_ms,
            kill_grace_ms: self.kill_grace_ms,
            cancel: self.cancel,
            backend: self.backend,
            fd_policy: self.fd_policy,
            early_exit: self.early_exit,
        })
    }
}

/// The result of a process execution.
#[derive(Debug)]
pub struct Output {
    /// The PID of the finished process.
    pub pid: pid_t,
    /// Final exit status (None if `wait=false`).
    pub status: Option<ExitStatus>,
    /// Captured stdout buffer.
    pub stdout: Vec<u8>,
    /// Captured stderr buffer.
    pub stderr: Vec<u8>,
    /// Whether the process timed out.
    pub timed_out: bool,
    /// Whether stdout drain stopped because the early-exit callback matched.
    pub stdout_early_exited: bool,
}

fn validate_backend(opts: &SpawnOptions) -> Result<(), CoreError> {
    validate_fd_policy(&opts.fd_policy)?;
    match opts.backend {
        SpawnBackend::PosixSpawn => {
            if opts.ctx.cwd.is_some() {
                return Err(CoreError::sys(libc::EINVAL, "posix_spawn cwd unsupported"));
            }
            if opts.pgroup.isolated {
                return Err(CoreError::sys(
                    libc::EINVAL,
                    "posix_spawn setsid unsupported",
                ));
            }
            if opts.fd_policy != SpawnFdPolicy::CloexecOnly {
                return Err(CoreError::sys(
                    libc::EINVAL,
                    "posix_spawn fd policy unsupported",
                ));
            }
            Ok(())
        }
        SpawnBackend::Fork => {
            // After `setsid` the child is a session leader in a brand-new
            // session; `setpgid(0, leader)` for a leader outside that session
            // always fails with EPERM. A zero leader means "own pid" (the
            // child's own group after setsid), which is valid.
            if opts.pgroup.isolated && opts.pgroup.leader.is_some_and(|l| l != 0) {
                return Err(CoreError::sys(
                    libc::EINVAL,
                    "fork isolated + custom setpgid leader unsupported",
                ));
            }
            Ok(())
        }
    }
}

fn validate_fd_policy(policy: &SpawnFdPolicy) -> Result<(), CoreError> {
    if let SpawnFdPolicy::Allowlist(fds) = policy {
        let mut seen = Vec::with_capacity(fds.len());
        for &fd in fds {
            if fd < 0 {
                return Err(CoreError::sys(libc::EINVAL, "spawn fd allowlist invalid"));
            }
            let flags = unsafe { libc::fcntl(fd, libc::F_GETFD) };
            if flags < 0 {
                return Err(CoreError::sys(errno(), "spawn fd allowlist fcntl(F_GETFD)"));
            }
            if seen.contains(&fd) {
                return Err(CoreError::sys(libc::EINVAL, "spawn fd allowlist duplicate"));
            }
            seen.push(fd);
        }
    }
    Ok(())
}

/// Specialized drain state for process spawning.
pub type SpawnDrain = DrainState<fn(&[u8]) -> bool>;

/// A process that is currently running and being monitored.
///
/// ### Fork Safety
/// This handle contains both a PID and owned file descriptors for process I/O.
/// Upon `fork`, the descriptors are inherited. Standard `O_CLOEXEC` behavior
/// applies after `exec`.
pub struct RunningProcess {
    /// Handle to the process.
    pub process: Process,
    drain: SpawnDrain,
}

/// Full process lifecycle driven by a caller-owned reactor.
///
/// `ManagedProcess` preserves the blocking [`spawn`] semantics while allowing
/// an application reactor to stay responsive: Core owns timeout/cancellation
/// escalation, process-group signaling, pipe draining, overflow reporting, and
/// `waitpid` reaping; the caller only routes readiness events and polls on
/// [`Self::next_deadline`].
pub struct ManagedProcess {
    running: Option<RunningProcess>,
    pid: pid_t,
    timeout_at: Option<Instant>,
    kill_grace: Duration,
    cancel: CancelPolicy,
    pgroup: ProcessGroup,
    cancel_at: Option<Instant>,
    kill_state: KillState,
    status: Option<ExitStatus>,
    timed_out: bool,
    kill_sent_at: Option<Instant>,
}

impl RunningProcess {
    /// Register active stdio pipe descriptors with a reactor.
    ///
    /// Call this once after [`spawn_start`] when the process was started with
    /// captured output or stdin data. The assigned tokens are kept internally
    /// and later matched by [`Self::handle_reactor_event`].
    pub fn register_with_reactor(&mut self, reactor: &mut Reactor) -> Result<(), CoreError> {
        self.drain.register_with_reactor(reactor)
    }

    /// Apply one reactor readiness event to this process' stdio drain state.
    ///
    /// Events for unrelated tokens are ignored. Callers remain responsible for
    /// waiting on [`Self::process`] and driving the reactor until [`Self::io_done`]
    /// returns true.
    pub fn handle_reactor_event(
        &mut self,
        reactor: &mut Reactor,
        event: &crate::fd::Event,
    ) -> Result<(), CoreError> {
        if self.drain.stdout_matches(event.token) {
            if event.readable || event.hangup {
                self.drain.handle_stdout_ready(reactor)?;
            } else if event.error {
                self.drain.drop_stdout(reactor)?;
            }
        } else if self.drain.stderr_matches(event.token) {
            if event.readable || event.hangup {
                self.drain.handle_stderr_ready(reactor)?;
            } else if event.error {
                self.drain.drop_stderr(reactor)?;
            }
        } else if self.drain.stdin_matches(event.token) {
            if event.writable {
                self.drain.handle_stdin_writable(reactor)?;
            } else if event.error || event.hangup {
                self.drain.drop_stdin(reactor)?;
            }
        }
        Ok(())
    }

    /// Return whether all managed stdio pipes have been drained or closed.
    pub fn io_done(&self) -> bool {
        self.drain.is_done()
    }

    /// Consume the running process handle and return captured stdout/stderr buffers.
    pub fn into_output_parts(self) -> (Vec<u8>, Vec<u8>) {
        self.drain.into_parts()
    }
}

impl ManagedProcess {
    /// Return the child PID.
    ///
    /// The PID is captured at spawn time, so this remains available after the
    /// process has completed (unlike the running handle, which is consumed).
    pub fn pid(&self) -> pid_t {
        self.pid
    }

    /// Register active child I/O descriptors with the caller's reactor.
    pub fn register_with_reactor(&mut self, reactor: &mut Reactor) -> Result<(), CoreError> {
        self.running
            .as_mut()
            .ok_or_else(|| CoreError::sys(libc::EINVAL, "managed process completed"))?
            .register_with_reactor(reactor)
    }

    /// Route one reactor event to the child's I/O drain state.
    pub fn handle_reactor_event(
        &mut self,
        reactor: &mut Reactor,
        event: &crate::fd::Event,
    ) -> Result<(), CoreError> {
        self.running
            .as_mut()
            .ok_or_else(|| CoreError::sys(libc::EINVAL, "managed process completed"))?
            .handle_reactor_event(reactor, event)
    }

    /// Request cancellation using the daemon-owned policy from
    /// [`SpawnOptionsBuilder::cancel`]. Repeated requests are idempotent.
    pub fn request_cancel(&mut self) {
        self.cancel_at.get_or_insert_with(Instant::now);
    }

    /// Earliest time at which [`Self::poll_completion`] should run again.
    ///
    /// A bounded reap tick is returned while the child is live, and exact
    /// timeout / TERM-to-KILL deadlines take precedence. `None` means the
    /// completion was already consumed.
    pub fn next_deadline(&self) -> Option<Instant> {
        self.running.as_ref()?;
        let now = Instant::now();
        let mut next = now + Duration::from_millis(100);
        if !self.timed_out
            && let Some(timeout_at) = self.timeout_at
            && timeout_at < next
        {
            next = timeout_at;
        }
        if self.kill_state == KillState::TermSent
            && let Some(cancel_at) = self.cancel_at
        {
            let kill_at = cancel_at + self.kill_grace;
            if kill_at < next {
                next = kill_at;
            }
        }
        // D-state bound: wake the caller once the post-SIGKILL reap window has
        // elapsed so `poll_completion` can give up on an unreapable child.
        if let Some(sent_at) = self.kill_sent_at {
            let bail_at = sent_at + D_STATE_REAP_BOUND;
            if bail_at < next {
                next = bail_at;
            }
        }
        Some(next)
    }

    /// Advance timeout/cancellation, reap state, and completion.
    ///
    /// Returns `Ok(None)` while work remains, the normal [`Output`] once the
    /// child is reaped and its pipes are drained, or `EOVERFLOW` when the
    /// configured combined output limit was exceeded on the fully-drained
    /// path. A forced-close (timeout/cancel with a wedged pipe) returns the
    /// partial output and the `timed_out` flag instead, matching blocking
    /// [`spawn`].
    pub fn poll_completion(&mut self, reactor: &mut Reactor) -> Result<Option<Output>, CoreError> {
        let now = Instant::now();
        if !self.timed_out
            && let Some(timeout_at) = self.timeout_at
            && now >= timeout_at
        {
            self.timed_out = true;
            self.cancel_at.get_or_insert(timeout_at);
        }

        self.advance_cancel(now)?;

        let running = self
            .running
            .as_ref()
            .ok_or_else(|| CoreError::sys(libc::EINVAL, "managed process completed"))?;
        if self.status.is_none() {
            self.status = running.process.wait_step()?;
        }

        let io_done = running.io_done();
        if self.status.is_some() && (io_done || self.cancel_at.is_some()) {
            return self.finish(reactor, !io_done).map(Some);
        }
        // D-state: SIGKILL sent but the child still cannot be reaped. A child
        // stuck in uninterruptible sleep keeps the signal pending until it
        // leaves D-state; return the partial output instead of polling forever.
        if self.status.is_none()
            && self
                .kill_sent_at
                .is_some_and(|t| now.duration_since(t) >= D_STATE_REAP_BOUND)
        {
            return self.finish(reactor, true).map(Some);
        }
        Ok(None)
    }

    fn advance_cancel(&mut self, now: Instant) -> Result<(), CoreError> {
        let Some(cancel_at) = self.cancel_at else {
            return Ok(());
        };
        // The child is already reaped — its pid may be recycled. Never signal.
        if self.status.is_some() {
            return Ok(());
        }
        let running = self
            .running
            .as_ref()
            .ok_or_else(|| CoreError::sys(libc::EINVAL, "managed process completed"))?;
        let process = &running.process;
        let pid = process.pid();
        let pgid = effective_pgid(pid, self.pgroup);
        let target_is_group = self.pgroup.isolated || self.pgroup.leader.is_some();
        match self.kill_state {
            KillState::None => match self.cancel {
                CancelPolicy::None => {}
                CancelPolicy::Graceful => {
                    let result = signal_process(process, target_is_group, pgid, libc::SIGTERM);
                    self.kill_state = if result.is_ok() {
                        KillState::TermSent
                    } else {
                        KillState::KillSent
                    };
                    if self.kill_state == KillState::KillSent {
                        self.kill_sent_at = Some(now);
                    }
                }
                CancelPolicy::Kill => {
                    let _ = signal_process(process, target_is_group, pgid, libc::SIGKILL);
                    self.kill_state = KillState::KillSent;
                    self.kill_sent_at = Some(now);
                }
            },
            KillState::TermSent if now >= cancel_at + self.kill_grace => {
                let _ = signal_process(process, target_is_group, pgid, libc::SIGKILL);
                self.kill_state = KillState::KillSent;
                self.kill_sent_at = Some(now);
            }
            _ => {}
        }
        Ok(())
    }

    fn finish(&mut self, reactor: &mut Reactor, force_close: bool) -> Result<Output, CoreError> {
        let mut running = self
            .running
            .take()
            .ok_or_else(|| CoreError::sys(libc::EINVAL, "managed process completed"))?;
        for slot in running.drain.take_all_slots() {
            if force_close {
                let _ = reactor.del(&slot.fd);
            } else {
                reactor.del(&slot.fd)?;
            }
        }
        let pid = running.process.pid();
        let (stdout, stderr, output_limit_exceeded, stdout_early_exited) =
            running.drain.into_parts_with_state();
        // If the child was never reaped (D-state give-up / forced close with an
        // unreapable child), it will eventually exit and become a zombie — hand
        // it to the reaper so it does not accumulate in a long-lived daemon
        // (finding 15).
        if self.status.is_none() {
            orphan_child(pid);
        }
        // Mirror blocking `spawn`: overflow is reported only when the drain
        // completed naturally. On the forced-close path (timeout/cancel with a
        // wedged pipe) the caller gets the partial output and the timed-out
        // flag instead, matching the blocking N4 behavior.
        if output_limit_exceeded && !force_close {
            return Err(CoreError::sys(libc::EOVERFLOW, "spawn output limit"));
        }
        Ok(Output {
            pid,
            status: self.status.take(),
            stdout,
            stderr,
            timed_out: self.timed_out,
            stdout_early_exited,
        })
    }
}

impl Drop for ManagedProcess {
    fn drop(&mut self) {
        let Some(running) = self.running.take() else {
            return;
        };
        // If the child was already reaped by `poll_completion`, the pid may
        // have been recycled — never signal it. The pipes are dropped with
        // `running`, so there is nothing left to clean up.
        if self.status.is_some() {
            return;
        }
        let process = &running.process;
        let pid = process.pid();
        let pgid = effective_pgid(pid, self.pgroup);
        let target_is_group = self.pgroup.isolated || self.pgroup.leader.is_some();
        let _ = signal_process(process, target_is_group, pgid, libc::SIGKILL);
        // Bound the reap wait: SIGKILL terminates a runnable child
        // immediately, but a child stuck in uninterruptible sleep (D-state)
        // never dies. Poll with WNOHANG so `Drop` cannot wedge the caller's
        // reactor thread forever on a stuck child.
        let deadline = Instant::now() + Duration::from_millis(100);
        while Instant::now() < deadline {
            match process.wait_step() {
                Ok(Some(_)) => return,
                Ok(None) => std::thread::sleep(Duration::from_millis(5)),
                Err(_) => return,
            }
        }
    }
}

fn effective_pgid(pid: pid_t, pgroup: ProcessGroup) -> pid_t {
    match pgroup.leader {
        Some(0) | None => pid,
        Some(leader) => leader,
    }
}

fn signal_process(
    process: &Process,
    target_is_group: bool,
    pgid: pid_t,
    signal: i32,
) -> Result<(), CoreError> {
    if target_is_group {
        process.kill_group(pgid, signal)
    } else {
        process.kill(signal)
    }
}

/// Start spawning a process and return a monitor handle.
///
/// This initializes the pipes and starts the process, but does not block. Use
/// [`RunningProcess::register_with_reactor`],
/// [`RunningProcess::handle_reactor_event`], [`RunningProcess::io_done`], and
/// [`RunningProcess::into_output_parts`] to drive captured stdio without
/// exposing internal drain state.
///
/// ### Errors
/// - `EACCES`: Permission denied for the executable.
/// - `EINVAL`: Invalid spawn options (e.g. background capture without wait).
/// - `EMFILE`: Process limit on open file descriptors hit.
/// - `ENOENT`: The executable was not found.
/// - `ENOMEM`: Insufficient memory to spawn the process.
pub fn spawn_start(opts: SpawnOptions) -> Result<RunningProcess, CoreError> {
    if !opts.wait && (opts.stdin.is_some() || opts.capture_stdout || opts.capture_stderr) {
        return Err(CoreError::sys(
            libc::EINVAL,
            "background I/O capture not supported (wait must be true)",
        ));
    }

    validate_backend(&opts)?;

    let (pid, drain) = match opts.backend {
        SpawnBackend::PosixSpawn => spawn_posix_internal(opts)?,
        SpawnBackend::Fork => spawn_fork_internal(opts)?,
    };

    Ok(RunningProcess {
        process: Process::new(pid),
        drain,
    })
}

/// Start a process whose complete lifecycle is driven by a caller-owned
/// reactor.
pub fn spawn_managed(opts: SpawnOptions) -> Result<ManagedProcess, CoreError> {
    if !opts.wait {
        return Err(CoreError::sys(
            libc::EINVAL,
            "managed process requires wait=true",
        ));
    }
    let timeout_at = opts
        .timeout_ms
        .map(|ms| Instant::now() + Duration::from_millis(ms as u64));
    let kill_grace = Duration::from_millis(opts.kill_grace_ms as u64);
    let cancel = opts.cancel;
    let pgroup = opts.pgroup;
    let running = spawn_start(opts)?;
    let pid = running.process.pid();
    Ok(ManagedProcess {
        running: Some(running),
        pid,
        timeout_at,
        kill_grace,
        cancel,
        pgroup,
        cancel_at: None,
        kill_state: KillState::None,
        status: None,
        timed_out: false,
        kill_sent_at: None,
    })
}

/// Spawn a process and block until completion or timeout.
///
/// This is the primary high-level interface for process execution. It handles
/// the full lifecycle, including I/O multiplexing and signal management.
///
/// ### Errors
/// Returns the same errors as [`spawn_start`], plus any I/O or reactor errors
/// encountered during the wait loop.
pub fn spawn(opts: SpawnOptions) -> Result<Output, CoreError> {
    let wait = opts.wait;
    let timeout_ms = opts.timeout_ms;
    let kill_grace_ms = opts.kill_grace_ms;
    let cancel = opts.cancel;
    let pgroup = opts.pgroup;

    let mut reactor = Reactor::new()?;
    let running = spawn_start(opts)?;

    let pid = running.process.pid();
    let mut drain = running.drain;

    drain.register_with_reactor(&mut reactor)?;

    if !wait {
        let (stdout, stderr) = drain.into_parts();
        // The caller will never `wait` on this pid — hand it to the reaper so
        // it does not become a zombie when it exits (finding 15).
        orphan_child(pid);
        return Ok(Output {
            pid,
            status: None,
            stdout,
            stderr,
            timed_out: false,
            stdout_early_exited: false,
        });
    }

    wait_loop(
        pid,
        drain,
        reactor,
        timeout_ms,
        kill_grace_ms,
        cancel,
        pgroup,
    )
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum KillState {
    None,
    TermSent,
    KillSent,
}

fn wait_loop(
    pid: pid_t,
    mut drain: crate::io::DrainState<fn(&[u8]) -> bool>,
    mut reactor: Reactor,
    timeout_ms: Option<u32>,
    kill_grace_ms: u32,
    cancel: CancelPolicy,
    pgroup: ProcessGroup,
) -> Result<Output, CoreError> {
    let process = Process::new(pid);
    // M8: the child's effective pgid is the configured leader when one is set
    // (Setpgid is applied after Setsid in the child), else its own pid. A
    // timeout must signal `-pgid`; `kill(-pid)` would target a different
    // group for a custom leader and the child would never die.
    let pgid = effective_pgid(pid, pgroup);
    let mut status_raw = process.wait_step()?;
    let mut state = KillState::None;
    let mut timed_out = false;
    // D-state bound: recorded once SIGKILL has been sent. If the child still
    // refuses to die (or be reaped) after `D_STATE_REAP_BOUND`, give up and
    // return the partial output instead of spinning on a stuck child.
    let mut kill_sent_at: Option<Instant> = None;
    // Deadline give-up bound for `CancelPolicy::None`: no signal is ever sent,
    // so `kill_sent_at` stays unset and the D-state bound never fires. A wedged
    // child (pipe held open by a descendant, child unreaped) would otherwise
    // poll at 100 ms forever. Once the deadline has passed we give up after the
    // same bound, returning the partial output with `timed_out` set.
    let mut deadline_passed_at: Option<Instant> = None;

    let start_time = std::time::Instant::now();
    let deadline = timeout_ms.map(|t| std::time::Duration::from_millis(t as u64));

    loop {
        let mut poll_timeout = -1;

        if let Some(dl) = deadline {
            let elapsed = start_time.elapsed();
            if elapsed >= dl {
                timed_out = true;
                deadline_passed_at = Some(deadline_passed_at.unwrap_or_else(Instant::now));
                let elapsed_over = (elapsed - dl).as_millis();

                let target_is_group = pgroup.isolated || pgroup.leader.is_some();

                // Only signal while the child is unreaped. Once waitpid has
                // reaped it the pid may already be recycled by the OS — killing
                // it would hit an unrelated process. The wedged-pipe path below
                // returns the partial output without sending any signal.
                if status_raw.is_none() {
                    match state {
                        KillState::None => {
                            if cancel == CancelPolicy::Graceful {
                                let r = if target_is_group {
                                    process.kill_group(pgid, libc::SIGTERM)
                                } else {
                                    process.kill(libc::SIGTERM)
                                };
                                if r.is_err() {
                                    state = KillState::KillSent; // Process already gone
                                    kill_sent_at = Some(Instant::now());
                                } else {
                                    state = KillState::TermSent;
                                }
                            } else if cancel == CancelPolicy::Kill {
                                let _ = if target_is_group {
                                    process.kill_group(pgid, libc::SIGKILL)
                                } else {
                                    process.kill(libc::SIGKILL)
                                };
                                state = KillState::KillSent;
                                kill_sent_at = Some(Instant::now());
                            } else {
                                // CancelPolicy::None just times out without killing
                            }
                        }
                        KillState::TermSent if elapsed_over > kill_grace_ms as u128 => {
                            let _ = if target_is_group {
                                process.kill_group(pgid, libc::SIGKILL)
                            } else {
                                process.kill(libc::SIGKILL)
                            };
                            state = KillState::KillSent;
                            kill_sent_at = Some(Instant::now());
                        }
                        _ => {}
                    }
                }
                poll_timeout = 100; // Poll frequently while waiting for kill to take effect
            } else {
                let remaining = dl - elapsed;
                poll_timeout = remaining.as_millis().min(i32::MAX as u128) as i32;
            }
        }

        if status_raw.is_none()
            && let Some(s) = process.wait_step()?
        {
            status_raw = Some(s);
        }

        if drain.is_done() {
            let s = if status_raw.is_some() {
                status_raw.take()
            } else if deadline.is_none() {
                // C1: all pipes drained but the child is still alive, and no
                // deadline is set → block until it exits (intended semantics).
                Some(process.wait_blocking()?)
            } else {
                // C1: pipes drained with a deadline set → never block here; fall
                // through to the bounded `reactor.wait` below so the deadline
                // logic at the top of the loop kills and reaps. A later
                // `wait_step` reaps the child and we return from this branch.
                None
            };

            if let Some(s) = s {
                for slot in drain.take_all_slots() {
                    reactor.del(&slot.fd)?;
                }
                let (stdout, stderr, output_limit_exceeded, stdout_early_exited) =
                    drain.into_parts_with_state();
                if output_limit_exceeded {
                    return Err(CoreError::sys(libc::EOVERFLOW, "spawn output limit"));
                }
                return Ok(Output {
                    pid,
                    status: Some(s),
                    stdout,
                    stderr,
                    timed_out,
                    stdout_early_exited,
                });
            }
        }

        // N4: the deadline has elapsed and the child is reaped, but a wedged
        // pipe (a descendant inheriting the write end) keeps the drain from
        // closing. The absolute deadline is authoritative — return the partial
        // output instead of spinning forever.
        if timed_out && status_raw.is_some() {
            for slot in drain.take_all_slots() {
                let _ = reactor.del(&slot.fd);
            }
            let (stdout, stderr, _output_limit_exceeded, stdout_early_exited) =
                drain.into_parts_with_state();
            return Ok(Output {
                pid,
                status: status_raw,
                stdout,
                stderr,
                timed_out: true,
                stdout_early_exited,
            });
        }

        // D-state: SIGKILL has been sent but the child is still unreaped after
        // the bound. A child stuck in uninterruptible sleep keeps the signal
        // pending until it leaves D-state, so no further wait can succeed —
        // return the partial output rather than polling forever. The pid is
        // not signaled again (it may be recycled once it finally exits).
        if let Some(sent_at) = kill_sent_at
            && sent_at.elapsed() >= D_STATE_REAP_BOUND
            && status_raw.is_none()
        {
            for slot in drain.take_all_slots() {
                let _ = reactor.del(&slot.fd);
            }
            // The child is unreapable right now but will eventually leave
            // D-state and exit; nobody will wait on it after this give-up, so
            // hand it to the reaper (finding 15).
            orphan_child(pid);
            let (stdout, stderr, _output_limit_exceeded, stdout_early_exited) =
                drain.into_parts_with_state();
            return Ok(Output {
                pid,
                status: None,
                stdout,
                stderr,
                timed_out: true,
                stdout_early_exited,
            });
        }

        // `CancelPolicy::None`: the deadline elapsed but nothing was ever
        // signaled, so the child may stay wedged (pipe held by a descendant,
        // child unreaped) indefinitely. Give up with the partial output after
        // the same bound as the D-state path — otherwise this polls at 100 ms
        // forever (finding 14).
        if cancel == CancelPolicy::None
            && timed_out
            && status_raw.is_none()
            && deadline_passed_at.is_some_and(|passed| passed.elapsed() >= D_STATE_REAP_BOUND)
        {
            for slot in drain.take_all_slots() {
                let _ = reactor.del(&slot.fd);
            }
            // The child was never signaled and may still be running; nobody
            // will wait on it now — hand it to the reaper (finding 15).
            orphan_child(pid);
            let (stdout, stderr, _output_limit_exceeded, stdout_early_exited) =
                drain.into_parts_with_state();
            return Ok(Output {
                pid,
                status: None,
                stdout,
                stderr,
                timed_out: true,
                stdout_early_exited,
            });
        }

        let timeout = poll_timeout;

        let mut events = Vec::new();
        let nevents = reactor.wait(&mut events, 64, timeout)?;

        for ev in events.iter().take(nevents) {
            if drain.stdout_matches(ev.token) {
                if ev.readable || ev.hangup {
                    drain.handle_stdout_ready(&mut reactor)?;
                } else if ev.error {
                    drain.drop_stdout(&mut reactor)?;
                }
            } else if drain.stderr_matches(ev.token) {
                if ev.readable || ev.hangup {
                    drain.handle_stderr_ready(&mut reactor)?;
                } else if ev.error {
                    drain.drop_stderr(&mut reactor)?;
                }
            } else if drain.stdin_matches(ev.token) {
                if ev.writable {
                    drain.handle_stdin_writable(&mut reactor)?;
                } else if ev.error || ev.hangup {
                    drain.drop_stdin(&mut reactor)?;
                }
            }
        }
    }
}