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::io::ChunkSink;
use crate::io::SinkResult;
use crate::reactor::Reactor;
use libc::{O_CLOEXEC, WEXITSTATUS, WIFEXITED, WIFSIGNALED, WTERMSIG, pid_t, pipe2, waitpid};
use std::collections::HashSet;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Mutex, OnceLock};

mod clone3;
mod exec;
mod fork;
mod posix;

use clone3::spawn_clone3_internal;
use exec::ExecContext;
use fork::{spawn_fork_internal, spawn_vfork_internal};
use posix::spawn_posix_internal;

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

/// Raw syscall numbers the `libc` crate does not expose on every target
/// (notably Android). `clone3` (435) and `pidfd_send_signal` (424) use the same
/// number on every architecture that implements them.
#[cfg(any(
    target_arch = "x86_64",
    target_arch = "aarch64",
    target_arch = "arm",
    target_arch = "riscv64",
    target_arch = "loongarch64",
    target_arch = "powerpc64",
    target_arch = "s390x"
))]
const SYS_CLONE3: libc::c_long = 435;
#[cfg(any(
    target_arch = "x86_64",
    target_arch = "aarch64",
    target_arch = "arm",
    target_arch = "riscv64",
    target_arch = "loongarch64",
    target_arch = "powerpc64",
    target_arch = "s390x"
))]
const SYS_PIDFD_SEND_SIGNAL: libc::c_long = 424;

/// `CLONE_PIDFD` flag for `clone3`: the kernel writes a pidfd for the child
/// into the `pidfd` field of `clone_args`.
const CLONE_PIDFD: u64 = 0x0000_1000;

/// 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.
pub(super) 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)
        .map_err(|_| REAPER_STARTED.store(false, Ordering::SeqCst))
        .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))
}

/// Open a new pseudo-terminal and return `(master, slave)`, both relocated
/// above stdio with `O_CLOEXEC`.
///
/// The master is drained as the child's single merged stdout+stderr stream;
/// the slave is dup2'd to the child's fd 0/1/2 in the child setup. Both are
/// `O_NOCTTY` so neither side accidentally becomes a controlling terminal of
/// the daemon (only the child claims it via `TIOCSCTTY`).
fn make_pty() -> Result<(Fd, Fd), CoreError> {
    let master = unsafe { libc::open(c"/dev/ptmx".as_ptr(), libc::O_RDWR | libc::O_NOCTTY | libc::O_CLOEXEC) };
    syscall_ret(master, "open /dev/ptmx")?;
    let master = match relocate_above_stdio(master, "ptmx:relocate") {
        Ok(fd) => fd,
        Err(e) => {
            unsafe {
                libc::close(master);
            }
            return Err(e);
        }
    };
    let result = (|| -> Result<RawFd, CoreError> {
        // `grantpt` on Linux devpts is a no-op success, but keep it for
        // portability; `unlockpt` is required before the slave can be opened.
        let r = unsafe { libc::grantpt(master) };
        syscall_ret(r, "grantpt")?;
        let r = unsafe { libc::unlockpt(master) };
        syscall_ret(r, "unlockpt")?;
        let mut name = [0 as libc::c_char; 4096];
        let r = unsafe { libc::ptsname_r(master, name.as_mut_ptr(), name.len()) };
        if r != 0 {
            return Err(CoreError::sys(r, "ptsname_r"));
        }
        let slave = unsafe {
            libc::open(
                name.as_ptr(),
                libc::O_RDWR | libc::O_NOCTTY | libc::O_CLOEXEC,
            )
        };
        syscall_ret(slave, "open pty slave")?;
        Ok(slave)
    })();
    match result {
        Ok(slave) => {
            let slave = match relocate_above_stdio(slave, "pty slave:relocate") {
                Ok(fd) => fd,
                Err(e) => {
                    unsafe {
                        libc::close(slave);
                        libc::close(master);
                    }
                    return Err(e);
                }
            };
            Ok((Fd::new(master, "pty master")?, Fd::new(slave, "pty slave")?))
        }
        Err(e) => {
            unsafe {
                libc::close(master);
            }
            Err(e)
        }
    }
}

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>,
    /// Pty mode: the master end, drained as the child's single merged stdout
    /// stream (parent side). `O_CLOEXEC`, relocated above stdio.
    pty_master: Option<Fd>,
    /// Pty mode: the slave end, dup2'd to the child's fd 0/1/2 and made its
    /// controlling terminal. `O_CLOEXEC` so the original (≥3) closes on exec
    /// after the dup2s.
    pty_slave: Option<Fd>,
}

impl Pipes {
    fn new(in_buf: Option<&[u8]>, out: bool, err: bool, pty: bool) -> Result<Self, CoreError> {
        if pty {
            let (master, slave) = make_pty()?;
            return Ok(Self {
                stdin_r: None,
                stdin_w: None,
                stdout_r: None,
                stdout_w: None,
                stderr_r: None,
                stderr_w: None,
                pty_master: Some(master),
                pty_slave: Some(slave),
            });
        }
        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,
            pty_master: None,
            pty_slave: None,
        })
    }

    #[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();
        self.pty_master.take();
        self.pty_slave.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,
    /// Force the use of `vfork`/`exec`.
    ///
    /// `vfork` shares the parent's address space with the child until it
    /// `execve`s (or `_exit`s), so it avoids the page-table work of `fork`.
    /// The child runs only async-signal-safe setup before `execve`, and the
    /// calling thread is blocked until the child execs. Safe for the child
    /// because the Linux `vfork` child inherits a *copy* of the descriptor
    /// table, so [`SpawnFdPolicy`] handling works as with [`SpawnBackend::Fork`].
    ///
    /// Use only when the shared-address-space semantics are understood:
    /// the child must never return from the spawn entry point, and a bug in the
    /// child setup can corrupt the parent's memory.
    Vfork,
    /// Force the use of `clone3(2)`/`exec` (kernel 5.3+).
    ///
    /// `clone3` with process flags creates a child with copy-on-write memory
    /// and a copied descriptor table, like [`SpawnBackend::Fork`], but lets the
    /// caller control clone flags directly. Supported by the same child setup
    /// as the fork backend. Returns `ENOSYS` on kernels without `clone3`.
    Clone3,
    /// Force the use of `clone3(2)` with `CLONE_PIDFD` + `exec` (kernel 5.3+).
    ///
    /// Identical to [`SpawnBackend::Clone3`], but the kernel additionally hands
    /// the parent a pidfd for the child. The resulting [`Process`] carries that
    /// pidfd: signaling uses `pidfd_send_signal` (immune to pid reuse), and
    /// exit detection `poll`s the pidfd instead of polling `waitpid`. Returns
    /// `ENOSYS` on kernels without `clone3`.
    Clone3Pidfd,
}

/// 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.
///
/// When the process was spawned by [`SpawnBackend::Clone3Pidfd`], the handle
/// additionally owns the child's pidfd. Signaling then uses
/// `pidfd_send_signal`, which cannot race with pid reuse, and exit detection
/// `poll`s the pidfd. The pidfd is closed when the handle is dropped.
pub struct Process {
    pid: pid_t,
    pidfd: Option<RawFd>,
}

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

    /// Create a handle for an existing PID that also owns its pidfd.
    pub(crate) fn with_pidfd(pid: pid_t, pidfd: RawFd) -> Self {
        Self {
            pid,
            pidfd: Some(pidfd),
        }
    }

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

    /// Return the pidfd owned by this handle, if any.
    pub fn pidfd(&self) -> Option<RawFd> {
        self.pidfd
    }

    /// Perform a non-blocking wait for process termination.
    ///
    /// When the handle owns a pidfd, the wait first `poll`s the pidfd (which
    /// becomes readable exactly when the child exits) and then reaps with
    /// `waitpid`, avoiding the `ECHILD`-race of polling `waitpid` directly.
    ///
    /// ### 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> {
        if let Some(pidfd) = self.pidfd {
            return wait_step_pidfd(pidfd, self.pid);
        }
        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.
    ///
    /// When the handle owns a pidfd, the signal is delivered with
    /// `pidfd_send_signal`, which cannot target a recycled pid; on kernels
    /// without it (`ENOSYS`, kernel < 5.1) it falls back to `kill`.
    ///
    /// ### Errors
    /// - `EINVAL`: Invalid signal number, or a non-positive pid (pid `0`
    ///   would signal the caller's own process group). With a pidfd,
    ///   `pidfd_send_signal` returns `EINVAL` for an invalid signal and this
    ///   is reported, not downgraded to a `kill` fallback.
    /// - `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 let Some(pidfd) = self.pidfd {
            let r = unsafe {
                libc::syscall(
                    SYS_PIDFD_SEND_SIGNAL,
                    pidfd,
                    sig,
                    std::ptr::null_mut::<libc::siginfo_t>(),
                    0,
                )
            };
            if r < 0 {
                let e = errno();
                if e == libc::ESRCH {
                    return Ok(());
                }
                // `pidfd_send_signal` returns EINVAL for an invalid signal
                // number or an unsupported flag — falling back to `kill` on
                // EINVAL would change semantics (e.g. signal 0 becomes an
                // existence check). Only a kernel that lacks the syscall
                // entirely (ENOSYS, pre-5.1) warrants the `kill` fallback.
                if e != libc::ENOSYS {
                    return Err(CoreError::sys(e, "pidfd_send_signal"));
                }
                // Kernel lacks pidfd_send_signal; fall through to kill.
            } else {
                return Ok(());
            }
        }
        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(())
    }
}

impl Drop for Process {
    fn drop(&mut self) {
        if let Some(pidfd) = self.pidfd.take() {
            unsafe {
                libc::close(pidfd);
            }
        }
    }
}

/// Non-blocking exit wait using a pidfd: `poll(2)` on the pidfd becomes
/// readable exactly when the child exits, and reaping still uses `waitpid`
/// (our own child cannot be pid-recycled while it is unreaped). Returns
/// `Ok(None)` while the child is running or was already reaped.
fn wait_step_pidfd(pidfd: RawFd, pid: pid_t) -> Result<Option<ExitStatus>, CoreError> {
    let mut pfd = libc::pollfd {
        fd: pidfd,
        events: libc::POLLIN,
        revents: 0,
    };
    loop {
        let r = unsafe { libc::poll(&mut pfd, 1, 0) };
        if r < 0 {
            let e = errno();
            if e == libc::EINTR {
                continue;
            }
            return Err(CoreError::sys(e, "poll(pidfd)"));
        }
        break;
    }
    if pfd.revents & libc::POLLIN == 0 {
        return Ok(None);
    }
    loop {
        let mut status = 0;
        let r = unsafe { waitpid(pid, &mut status, libc::WNOHANG) };
        if r == pid {
            return Ok(Some(decode_status(status)));
        }
        if r < 0 {
            let e = errno();
            if e == libc::EINTR {
                continue;
            }
            if e == libc::ECHILD {
                // Reaped elsewhere; the pidfd stays readable.
                return Ok(None);
            }
            return Err(CoreError::sys(e, "waitpid(pidfd step)"));
        }
        // r == 0: readiness raced with a concurrent reap; not running now.
        return Ok(None);
    }
}

/// 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,
    session_containment: bool,
    max_output: usize,
    timeout_ms: Option<u32>,
    kill_grace_ms: u32,
    cancel: CancelPolicy,
    backend: SpawnBackend,
    fd_policy: SpawnFdPolicy,
    early_exit: Option<fn(&[u8]) -> bool>,
    /// Optional streaming chunk observer: every retained output chunk is
    /// forwarded here as it is read (`is_stdout`, bytes) instead of being
    /// accumulated for the completion [`Output`]. Return [`SinkResult::Pause`]
    /// to stop draining (the chunk is retained and re-delivered on resume);
    /// bytes are never dropped on this path and the read loop never blocks.
    /// Ignored when the stream is not captured.
    chunk_sink: Option<ChunkSink>,
    /// Spawn the child on a pseudo-terminal instead of captured pipes: the
    /// slave becomes the child's controlling terminal (setsid + `TIOCSCTTY`,
    /// dup2'd to fd 0/1/2) and the master is drained as a single merged
    /// stdout+stderr stream. Requires an isolated process group (a session is
    /// needed before `TIOCSCTTY`) and is unsupported on the posix_spawn
    /// backend (no child setup step). The master is exposed to the caller's
    /// drain for reads and to [`RunningProcess::resize_pty`] for `TIOCSWINSZ`.
    pty: 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,
    session_containment: bool,
    max_output: usize,
    timeout_ms: Option<u32>,
    kill_grace_ms: u32,
    cancel: CancelPolicy,
    backend: SpawnBackend,
    fd_policy: SpawnFdPolicy,
    early_exit: Option<fn(&[u8]) -> bool>,
    chunk_sink: Option<ChunkSink>,
    pty: 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(),
            session_containment: false,
            max_output: 1024 * 1024,
            timeout_ms: None,
            kill_grace_ms: 2000,
            cancel: CancelPolicy::Kill,
            backend,
            fd_policy: SpawnFdPolicy::default(),
            early_exit: None,
            chunk_sink: None,
            pty: false,
        }
    }

    /// 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
    }

    /// Contain the child inside the process group/session it is placed into.
    ///
    /// A seccomp filter installed in the child (after the daemon's own
    /// `setsid`/`setpgid`, before `execve`) denies `setsid`, `setpgid`,
    /// `setpgrp`, `unshare`, and `setns`. Because filters are inherited
    /// across `fork` and `execve` and can only be tightened, never loosened,
    /// the child and every descendant are locked into the group/session —
    /// making `kill_group` (timeout/cancel deactivation) total even against a
    /// hostile root child that tries to escape by daemonizing or changing its
    /// process group. Requires an isolated process group
    /// ([`ProcessGroup::new(None, true)`](ProcessGroup::new)); rejected on
    /// [`SpawnBackend::PosixSpawn`](SpawnBackend::PosixSpawn), which has no
    /// child setup step.
    pub fn session_containment(mut self) -> Self {
        self.session_containment = true;
        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
    }

    /// Enable streaming drain: forward every retained output chunk to `sink`
    /// as it is read instead of accumulating it for the completion [`Output`].
    ///
    /// The sink returns [`SinkResult::Pause`] when its bounded queue is full;
    /// the drain then stops reading the child (kernel backpressure applies)
    /// without dropping the held chunk and without blocking the reactor.
    /// Resume via the managed-process or drain resume methods once the queue
    /// drains. When a sink is set, `max_output` no longer truncates: bytes
    /// are never dropped on the streaming path.
    pub fn chunk_sink<F>(mut self, sink: F) -> Self
    where
        F: Fn(bool, &[u8]) -> SinkResult + Send + Sync + 'static,
    {
        self.chunk_sink = Some(Arc::new(sink));
        self
    }

    /// Spawn the child on a pseudo-terminal (see [`SpawnOptions::pty`]).
    ///
    /// Mutually exclusive with pipe capture: the slave replaces
    /// `capture_stdout`/`capture_stderr`/`stdin` as the child's stdio, and the
    /// master replaces the stdout pipe on the drain (single merged stream).
    pub fn pty(mut self) -> Self {
        self.pty = true;
        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,
            session_containment: self.session_containment,
            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,
            chunk_sink: self.chunk_sink,
            pty: self.pty,
        })
    }
}

/// 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,
    /// Streaming mode: the stdout chunk held while the sink queue was full at
    /// completion (empty/none when no sink was attached). The caller must
    /// flush it before delivering the terminal frame.
    pub stdout_pending: Option<Vec<u8>>,
    /// Streaming mode: the stderr chunk held while the sink queue was full at
    /// completion.
    pub stderr_pending: Option<Vec<u8>>,
}

fn validate_backend(opts: &SpawnOptions) -> Result<(), CoreError> {
    validate_fd_policy(&opts.fd_policy)?;
    if opts.pty {
        // Pty mode has no stdin path yet: the child's stdin is the slave, and
        // writing to it would go through the master, which the drain does not
        // expose until TX_EXEC_WRITE-style write support lands. A stdin buffer
        // with pty mode would silently target a pipe that does not exist.
        if opts.stdin.is_some() {
            return Err(CoreError::sys(
                libc::EINVAL,
                "pty stdin unsupported (write support pending)",
            ));
        }
    }
    match opts.backend {
        SpawnBackend::PosixSpawn => {
            if opts.pty {
                return Err(CoreError::sys(
                    libc::EINVAL,
                    "posix_spawn pty unsupported (no child setup step)",
                ));
            }
            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.session_containment {
                return Err(CoreError::sys(
                    libc::EINVAL,
                    "posix_spawn session containment unsupported",
                ));
            }
            if opts.fd_policy != SpawnFdPolicy::CloexecOnly {
                return Err(CoreError::sys(
                    libc::EINVAL,
                    "posix_spawn fd policy unsupported",
                ));
            }
            Ok(())
        }
        SpawnBackend::Fork
        | SpawnBackend::Vfork
        | SpawnBackend::Clone3
        | SpawnBackend::Clone3Pidfd => {
            // 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. Applies to
            // every exec-style backend: they all run the same child setup.
            if opts.pgroup.isolated && opts.pgroup.leader.is_some_and(|l| l != 0) {
                return Err(CoreError::sys(
                    libc::EINVAL,
                    "exec isolated + custom setpgid leader unsupported",
                ));
            }
            // Session containment pins the child to the group/session the
            // daemon placed it in; without isolation there is no such
            // boundary to pin to.
            if opts.session_containment && !opts.pgroup.isolated {
                return Err(CoreError::sys(
                    libc::EINVAL,
                    "session containment requires an isolated process group",
                ));
            }
            // A controlling terminal requires the child to be a session
            // leader first (TIOCSCTTY fails with EPERM otherwise).
            if opts.pty && !opts.pgroup.isolated {
                return Err(CoreError::sys(
                    libc::EINVAL,
                    "pty requires an isolated process group",
                ));
            }
            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>,
    deadline_passed_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()
    }

    /// Return whether the stdout stream is paused on a full sink queue.
    pub fn stdout_paused(&self) -> bool {
        self.drain.stdout_paused()
    }

    /// Return whether the stderr stream is paused on a full sink queue.
    pub fn stderr_paused(&self) -> bool {
        self.drain.stderr_paused()
    }

    /// Re-deliver the held stdout chunk (if any) and re-register the fd when
    /// the sink has room again. Returns `true` when the stream is resumed.
    pub fn resume_stdout(&mut self, reactor: &mut Reactor) -> Result<bool, CoreError> {
        self.drain.resume_stdout(reactor)
    }

    /// Re-deliver the held stderr chunk (if any) and re-register the fd when
    /// the sink has room again. Returns `true` when the stream is resumed.
    pub fn resume_stderr(&mut self, reactor: &mut Reactor) -> Result<bool, CoreError> {
        self.drain.resume_stderr(reactor)
    }

    /// Take the un-delivered stdout chunk (streaming mode), if any.
    pub(crate) fn take_stdout_pending(&mut self) -> Option<Vec<u8>> {
        self.drain.take_stdout_pending()
    }

    /// Take the un-delivered stderr chunk (streaming mode), if any.
    pub(crate) fn take_stderr_pending(&mut self) -> Option<Vec<u8>> {
        self.drain.take_stderr_pending()
    }

    /// 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()
    }

    /// Apply a new terminal window size to a pty-spawned child.
    ///
    /// Only valid for a [`SpawnOptionsBuilder::pty`] spawn with an active
    /// stdout stream; callers typically follow this with a `SIGWINCH` to the
    /// child (or its foreground group) so the program can re-read the size.
    ///
    /// ### Errors
    /// - `EINVAL`: The spawn was not a pty spawn, or `rows`/`cols` is zero.
    /// - `ENOTTY`: The pty master is unexpectedly not a terminal.
    pub fn resize_pty(&self, rows: u16, cols: u16) -> Result<(), CoreError> {
        self.drain.resize_pty(rows, cols)
    }
}

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)
    }

    /// Return whether the stdout stream is paused on a full sink queue.
    pub fn stdout_paused(&self) -> bool {
        self.running
            .as_ref()
            .is_some_and(|running| running.stdout_paused())
    }

    /// Return whether the stderr stream is paused on a full sink queue.
    pub fn stderr_paused(&self) -> bool {
        self.running
            .as_ref()
            .is_some_and(|running| running.stderr_paused())
    }

    /// Re-deliver the held stdout chunk (if any) and re-register the fd when
    /// the sink has room again. Returns `true` when the stream is resumed.
    pub fn resume_stdout(&mut self, reactor: &mut Reactor) -> Result<bool, CoreError> {
        self.running
            .as_mut()
            .ok_or_else(|| CoreError::sys(libc::EINVAL, "managed process completed"))?
            .resume_stdout(reactor)
    }

    /// Re-deliver the held stderr chunk (if any) and re-register the fd when
    /// the sink has room again. Returns `true` when the stream is resumed.
    pub fn resume_stderr(&mut self, reactor: &mut Reactor) -> Result<bool, CoreError> {
        self.running
            .as_mut()
            .ok_or_else(|| CoreError::sys(libc::EINVAL, "managed process completed"))?
            .resume_stderr(reactor)
    }

    /// 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);
            if self.cancel == CancelPolicy::None {
                // `CancelPolicy::None` never signals, so the D-state bound
                // below never fires; record when the deadline passed so the
                // give-up bound mirrors blocking `spawn` (finding 14).
                self.deadline_passed_at = Some(self.deadline_passed_at.unwrap_or(now));
            }
        }

        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();
        let paused = running.stdout_paused() || running.stderr_paused();
        // A paused stream (full sink queue, fd removed from the reactor) can
        // never make progress on its own: once the child is reaped, finish with
        // the partial output and the held pending chunk instead of waiting for
        // a readiness event that will never arrive.
        if self.status.is_some() && (io_done || self.cancel_at.is_some() || paused) {
            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);
        }
        // `CancelPolicy::None`: the deadline elapsed but nothing was ever
        // signaled, so a wedged child would poll forever. Give up with the
        // partial output after the same bound as the D-state path (finding 14).
        if self.status.is_none()
            && self.cancel == CancelPolicy::None
            && self
                .deadline_passed_at
                .is_some_and(|passed| now.duration_since(passed) >= 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 slot.token.is_none() {
                continue;
            }
            if force_close {
                let _ = reactor.del(&slot.fd);
            } else {
                reactor.del(&slot.fd)?;
            }
        }
        let pid = running.process.pid();
        let stdout_pending = running.drain.take_stdout_pending();
        let stderr_pending = running.drain.take_stderr_pending();
        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,
            stdout_pending,
            stderr_pending,
        })
    }

    /// Apply a new terminal window size to a pty-spawned child.
    ///
    /// Only valid for a [`SpawnOptionsBuilder::pty`] spawn with an active
    /// stdout stream; callers typically follow this with a `SIGWINCH` to the
    /// child (or its foreground group) so the program can re-read the size.
    ///
    /// ### Errors
    /// - `EINVAL`: The spawn was not a pty spawn, the stream is already
    ///   closed, or `rows`/`cols` is zero.
    /// - `ENOTTY`: The pty master is unexpectedly not a terminal.
    pub fn resize_pty(&self, rows: u16, cols: u16) -> Result<(), CoreError> {
        self.running
            .as_ref()
            .ok_or_else(|| CoreError::sys(libc::EINVAL, "managed process completed"))?
            .resize_pty(rows, cols)
    }
}

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();
        // Respect CancelPolicy::None: "do nothing on cancellation" must not
        // kill the child on Drop either — the caller asked that cancellation
        // leave the child alone.
        if self.cancel != CancelPolicy::None {
            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,
            }
        }
        // Give-up: the child is unreapable right now (D-state) or still
        // running under `CancelPolicy::None`. Nobody will `waitpid` it now;
        // hand it to the reaper so it does not become a zombie on exit.
        orphan_child(pid);
    }
}

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 (process, drain) = match opts.backend {
        SpawnBackend::PosixSpawn => spawn_posix_internal(opts)?,
        SpawnBackend::Fork => spawn_fork_internal(opts)?,
        SpawnBackend::Vfork => spawn_vfork_internal(opts)?,
        SpawnBackend::Clone3 => spawn_clone3_internal(opts, false)?,
        SpawnBackend::Clone3Pidfd => spawn_clone3_internal(opts, true)?,
    };

    Ok(RunningProcess { process, 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,
        deadline_passed_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;

    if let Err(e) = drain.register_with_reactor(&mut reactor) {
        // The child is live but stdio registration failed; `running` is
        // dropped here so nobody will `waitpid` it. Hand it to the reaper.
        orphan_child(pid);
        return Err(e);
    }

    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,
            stdout_pending: None,
            stderr_pending: None,
        });
    }

    wait_loop(
        running.process,
        drain,
        reactor,
        timeout_ms,
        kill_grace_ms,
        cancel,
        pgroup,
    )
}

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

fn wait_loop(
    process: Process,
    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 pid = process.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() {
                    if slot.token.is_some() {
                        reactor.del(&slot.fd)?;
                    }
                }
                let stdout_pending = drain.take_stdout_pending();
                let stderr_pending = drain.take_stderr_pending();
                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,
                    stdout_pending,
                    stderr_pending,
                });
            }
        }

        // Streaming mode: a paused stream (full sink queue) cannot progress
        // even after the child is reaped — the fd is not registered, so no
        // readiness event will ever arrive. Return the partial output and the
        // held pending chunk for the caller to flush (the blocking-path mirror
        // of `poll_completion`'s paused-finish branch).
        if status_raw.is_some() && (drain.stdout_paused() || drain.stderr_paused()) {
            for slot in drain.take_all_slots() {
                if slot.token.is_some() {
                    let _ = reactor.del(&slot.fd);
                }
            }
            let stdout_pending = drain.take_stdout_pending();
            let stderr_pending = drain.take_stderr_pending();
            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,
                stdout_early_exited,
                stdout_pending,
                stderr_pending,
            });
        }

        // 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() {
                if slot.token.is_some() {
                    let _ = reactor.del(&slot.fd);
                }
            }
            let stdout_pending = drain.take_stdout_pending();
            let stderr_pending = drain.take_stderr_pending();
            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,
                stdout_pending,
                stderr_pending,
            });
        }

        // 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() {
                if slot.token.is_some() {
                    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_pending = drain.take_stdout_pending();
            let stderr_pending = drain.take_stderr_pending();
            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,
                stdout_pending,
                stderr_pending,
            });
        }

        // `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() {
                if slot.token.is_some() {
                    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_pending = drain.take_stdout_pending();
            let stderr_pending = drain.take_stderr_pending();
            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,
                stdout_pending,
                stderr_pending,
            });
        }

        // Streaming backpressure: while a stream is paused its fd is not
        // registered (no readiness events). Keep trying to resume so a
        // concurrent queue consumer's drained capacity re-registers the fd,
        // and bound the poll so the loop cannot block forever on a paused
        // stream.
        if drain.stdout_paused() || drain.stderr_paused() {
            if drain.stdout_paused() {
                let _ = drain.resume_stdout(&mut reactor);
            }
            if drain.stderr_paused() {
                let _ = drain.resume_stderr(&mut reactor);
            }
            if !(0..=10).contains(&poll_timeout) {
                poll_timeout = 10;
            }
        }

        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)?;
                }
            }
        }
    }
}