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::ChunkSink;
use crate::io::DrainState;
use crate::io::SinkResult;
use crate::reactor::{Reactor, Token};
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);

/// Orphaned sessions: pty or isolated-pipe sessions whose sweep the caller
/// gave up on (the D-state give-up) but that may still have live members. The
/// reaper thread keeps SIGKILLing them until `/proc` shows no live members, so
/// the give-up — which exists because the *leader* is unreapable — can never
/// double as "the kill failed, abandon it" and leak contained survivors
/// (finding H2). Each entry pairs the session id with the leader's
/// `starttime` (procfs field 22), captured at registration: the numeric sid is
/// only trustworthy while it still names the *same* process incarnation, so a
/// reaped-and-recycled leader pid can never be swept by a stale bare sid
/// (finding F8 / rev6-F1 — the reaper's own liveness gate).
static ORPHANED_SESSIONS: OnceLock<Mutex<HashSet<(pid_t, u64)>>> = OnceLock::new();

/// Register a session leader's pid so the reaper thread keeps sweeping the
/// session (SIGKILL every remaining group) until it is empty. The leader's
/// `starttime` is captured here and verified by the reaper before every sweep:
/// if the pid is already gone (no `/proc/<pid>/stat`) or the numeric sid has
/// been recycled into an unrelated process, the registration is **refused** —
/// a bare sid kill would otherwise hit an unrelated session (finding F8).
pub(super) fn orphan_session(sid: pid_t) {
    let Some(starttime) = crate::proc::starttime(sid) else {
        // Leader already gone or unreadable: sweeping by this numeric sid is
        // unsafe (recycled-pid class) and pointless (nothing to sweep if the
        // whole session exited) — refuse.
        return;
    };
    ORPHANED_SESSIONS
        .get_or_init(|| Mutex::new(HashSet::new()))
        .lock()
        .unwrap()
        .insert((sid, starttime));
    start_reaper();
}

/// 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). Also
/// keeps sweeping orphaned sessions until they are empty (finding H2).
fn reap_orphaned() {
    loop {
        // ── pid arm: prune-by-reaped-only ────────────────────────────────
        // Snapshot, then remove *exactly* the pids this pass confirmed reaped
        // (waitpid == pid or ECHILD). Never intersect a stale snapshot: a pid
        // registered concurrently (between snapshot and prune) survives — the
        // F3/F4 race the old `retain` reintroduced.
        let pids: Vec<pid_t> = ORPHANED
            .get_or_init(|| Mutex::new(HashSet::new()))
            .lock()
            .unwrap()
            .iter()
            .copied()
            .collect();
        let mut reaped = 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))
            {
                reaped.push(pid); // confirmed reaped or gone — prune exactly this pid
            }
        }
        if !reaped.is_empty()
            && let Some(set) = ORPHANED.get()
            && let Ok(mut guard) = set.lock()
        {
            prune_reaped_by_reaped_only(&mut guard, &reaped);
        }
        // ── session arm: leader-liveness-gated sweep ─────────────────────
        // H2 + rev6-F1: keep SIGKILLing every remaining group until /proc
        // shows no live members. The sweep excludes zombies, so a
        // killed-but-unreaped member does not hold the registration open; a
        // genuinely D-state member keeps SIGKILL pending until it wakes and
        // the sweep converges then. Each iteration first verifies the numeric
        // sid still names the *registered leader incarnation* (starttime): if
        // the leader is gone or the pid was recycled, the sid is no longer a
        // safe handle — drop it WITHOUT killing (a bare-sid sweep would hit an
        // unrelated recycled session, finding F8).
        if let Some(sessions) = ORPHANED_SESSIONS.get() {
            let sids: Vec<(pid_t, u64)> = sessions.lock().unwrap().iter().copied().collect();
            let mut converged = Vec::new();
            for (sid, starttime) in sids {
                if crate::proc::starttime(sid) != Some(starttime) {
                    // Leader gone or recycled — never sweep by this bare sid.
                    converged.push((sid, starttime));
                    continue;
                }
                if session_sweep(sid).unwrap_or(false) {
                    converged.push((sid, starttime));
                }
            }
            if !converged.is_empty()
                && let Ok(mut guard) = sessions.lock()
            {
                for entry in converged {
                    guard.remove(&entry);
                }
            }
        }
        std::thread::sleep(Duration::from_millis(250));
    }
}

/// Remove from `set` exactly the pids confirmed reaped this pass. Pure seam
/// (F3/F4 regression): the reaper's pid arm snapshots the orphan set, then
/// prunes by this list — a pid registered *after* the snapshot but before the
/// prune is left untouched.
pub(super) fn prune_reaped_by_reaped_only(
    set: &mut std::collections::HashSet<pid_t>,
    reaped: &[pid_t],
) {
    for pid in reaped {
        set.remove(pid);
    }
}

/// Test-only: number of currently registered orphaned sessions.
#[cfg(test)]
pub(super) fn orphaned_sessions_len() -> usize {
    ORPHANED_SESSIONS
        .get()
        .map(|s| s.lock().unwrap().len())
        .unwrap_or(0)
}

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

/// Policy for what *natural* completion (the leader exiting on its own) does
/// with a still-live isolated/pty session.
///
/// A background member that keeps the pty slave (or a contained descendant
/// holding a captured pipe) open prevents the master EOF that the historical
/// natural gate waited on — without a sweep the job would hang forever (the
/// A4-3 seam). This policy tells Core whether the sweep is the completion
/// trigger (default, kill-totality) or whether members may survive the leader.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum SessionExitPolicy {
    /// Sweep the contained session once the leader is reaped: SIGKILL live
    /// session members and hold completion until the session is empty
    /// (kill-totality, the historic containment guarantee). The sweep is the
    /// *trigger*, not a side-effect of EOF — a slave-holding member is killed
    /// and then the master can EOF. Cannot be combined with
    /// [`CancelPolicy::None`] (which opts out of all signaling).
    #[default]
    Sweep,
    /// Report completion on leader-reap without signaling the session: a
    /// background member (e.g. `nohup sleep &`) survives the leader, matching
    /// plain POSIX shell semantics. The leader-reap remains the completion
    /// gate so the slave-holding-member hang is still impossible.
    LetMembersSurvive,
}

/// 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`).
///
/// The pair is returned to the caller so a spawn can be configured *before*
/// the child execs: apply the initial window with [`pty_window`], derive the
/// child's `LINES`/`COLUMNS` env from that read-back, then hand ownership to
/// [`SpawnOptionsBuilder::pty_with`]. Core re-takes ownership inside `Pipes`:
/// from the moment the pair enters the spawn options, Core owns both
/// descriptors and is responsible for their cleanup on every success and
/// failure path — there is no ambiguous "does the caller still own this fd?"
/// state.
pub 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)
        }
    }
}

/// Apply an initial window size to a pty master *before* the child execs and
/// read back the actual `winsize` the kernel holds.
///
/// This is the single source of truth for the pty's starting geometry: callers
/// that need `LINES`/`COLUMNS` in the child environment must derive them from
/// the **returned** `(rows, cols)` — the `TIOCGWINSZ` read-back — not from the
/// values they passed in. Two writers of the same fact (the wire dims *and* a
/// separate `TIOCSWINSZ`) can drift the moment one call site changes; the
/// read-back keeps the env provably consistent with what the kernel/pty layer
/// believes.
///
/// ### Errors
/// - `EINVAL`: `rows` or `cols` is zero.
/// - `ENOTTY`: `master` is not a terminal.
pub fn pty_window(master: &Fd, rows: u16, cols: u16) -> Result<(u16, u16), CoreError> {
    if rows == 0 || cols == 0 {
        return Err(CoreError::sys(
            libc::EINVAL,
            "pty_window: rows and cols must be non-zero",
        ));
    }
    let ws = libc::winsize {
        ws_row: rows,
        ws_col: cols,
        ws_xpixel: 0,
        ws_ypixel: 0,
    };
    let r = unsafe { libc::ioctl(master.raw(), libc::TIOCSWINSZ as libc::Ioctl, &ws) };
    syscall_ret(r, "TIOCSWINSZ")?;
    let mut got: libc::winsize = unsafe { std::mem::zeroed() };
    let r = unsafe { libc::ioctl(master.raw(), libc::TIOCGWINSZ as libc::Ioctl, &mut got) };
    syscall_ret(r, "TIOCGWINSZ")?;
    Ok((got.ws_row, got.ws_col))
}

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,
        pty_fds: Option<(Fd, Fd)>,
    ) -> Result<Self, CoreError> {
        if pty {
            let (master, slave) = match pty_fds {
                // Caller-supplied pair (see `SpawnOptionsBuilder::pty_with`):
                // Core takes ownership here and closes both on every
                // success/failure path.
                Some(pair) => pair,
                None => 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.
///
/// Move-only: when `pty_fds` is `Some` the struct owns a pty pair (raw OS
/// descriptors), so it is not `Clone` — duplicating the builder would duplicate
/// ownership of fds that cannot be duplicated.
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`.
    ///
    /// Termios is **not** configured: the slave keeps the kernel-default
    /// cooked line discipline (`ISIG|ICANON|ECHO|IXON` on, `IUTF8` off). The
    /// caller owns termios (tcsetattr on the slave) — Core is no-policy.
    /// Interactive callers that keep cooked mode must not locally echo
    /// (the kernel already does); a raw-mode caller is responsible for its
    /// own echo and signal mapping.
    pty: bool,
    /// Preexisting pty pair supplied by the caller (via
    /// [`SpawnOptionsBuilder::pty_with`]): Core takes ownership of both
    /// descriptors and is responsible for their cleanup on every success and
    /// failure path. `Some` implies `pty == true`; the pair is used instead of
    /// calling [`make_pty`] internally, so the caller can apply an initial
    /// window (`TIOCSWINSZ`) and read it back (`TIOCGWINSZ`) before the child
    /// execs.
    pty_fds: Option<(Fd, Fd)>,
    /// Opt-in `PR_SET_PDEATHSIG`: the signal the child receives when the
    /// **parent thread that created it** exits (not the process — see
    /// `docs/ARCHITECTURE.md`). Leader-only: it reaches the spawned leader's
    /// whole process, not session members in other process groups. The child
    /// arms it before any other setup and verifies `getppid()` still equals the
    /// expected parent, closing the fork→prctl race. None (default): no
    /// parent-death signal.
    pdeath_signal: Option<i32>,
    /// Natural-exit policy for a contained session (see [`SessionExitPolicy`]).
    /// Defaults to [`SessionExitPolicy::Sweep`].
    session_exit: SessionExitPolicy,
}

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`].
///
/// Move-only when [`SpawnOptionsBuilder::pty_with`] has been called: the
/// builder then owns a pty pair, so it is not `Clone` (see [`SpawnOptions`]).
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,
    pty_fds: Option<(Fd, Fd)>,
    pdeath_signal: Option<i32>,
    session_exit: SessionExitPolicy,
}

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,
            pty_fds: None,
            pdeath_signal: None,
            session_exit: SessionExitPolicy::Sweep,
        }
    }

    /// 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).
    ///
    /// Core creates the pty pair internally. To pre-configure the pty window
    /// before the child execs (and derive the child's terminal env from the
    /// read-back), use [`SpawnOptionsBuilder::pty_with`] instead — it takes a
    /// caller-created pair and is move-only.
    pub fn pty(mut self) -> Self {
        self.pty = true;
        self
    }

    /// Spawn the child on a pseudo-terminal using a **caller-created** pty
    /// pair, whose initial window the caller already configured.
    ///
    /// The typical flow:
    /// 1. [`make_pty`] returns `(master, slave)`;
    /// 2. [`pty_window`] applies the initial size to the master and reads back
    ///    the actual `winsize`;
    /// 3. the caller derives `LINES`/`COLUMNS` from that read-back;
    /// 4. this method hands ownership of the pair to Core.
    ///
    /// Core takes ownership of both descriptors and is responsible for their
    /// cleanup on every spawn success/failure path. The builder (and the
    /// resulting [`SpawnOptions`]) is move-only from this point — a pty pair
    /// is not `Clone`able, so neither is the builder that owns it.
    pub fn pty_with(mut self, master: Fd, slave: Fd) -> Self {
        self.pty = true;
        self.pty_fds = Some((master, slave));
        self
    }

    /// Arm `PR_SET_PDEATHSIG` on the spawned child (opt-in).
    ///
    /// When set, the child receives `sig` when the **parent thread that
    /// created it** exits (see [`SpawnOptions::pdeath_signal`] for the exact
    /// semantics and scope). The child arms the signal before any other setup
    /// and aborts if `getppid()` no longer matches its expected parent —
    /// closing the fork→prctl race that would otherwise leave the signal
    /// silently undelivered.
    pub fn pdeath_signal(mut self, sig: i32) -> Self {
        self.pdeath_signal = Some(sig);
        self
    }

    /// Set the natural-exit policy for a contained session (see
    /// [`SessionExitPolicy`]). Defaults to [`SessionExitPolicy::Sweep`].
    pub fn session_exit(mut self, policy: SessionExitPolicy) -> Self {
        self.session_exit = policy;
        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,
            pty_fds: self.pty_fds,
            pdeath_signal: self.pdeath_signal,
            session_exit: self.session_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,
    /// 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>>,
    /// The session sweep that allowed completion SIGKILLed at least one live
    /// session member (a background/contained process that outlived the
    /// leader). `true` means the job's own exit did not leave the session
    /// empty — the caller (daemon) may want to surface this to the user.
    /// Always `false` for non-session spawns and under
    /// [`SessionExitPolicy::LetMembersSurvive`].
    pub swept_members: bool,
}

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.pdeath_signal.is_some() {
                return Err(CoreError::sys(
                    libc::EINVAL,
                    "posix_spawn pdeath_signal unsupported (no child setup step)",
                ));
            }
            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>,
    /// When the natural-path session sweep first started (see
    /// [`SessionExitPolicy::Sweep`]). Bounds the sweep so a D-state member
    /// cannot keep `/proc` re-enumeration alive forever (the F6 give-up).
    sweep_started_at: Option<Instant>,
    /// Set when a natural-exit session sweep found and SIGKILLed at least one
    /// live session member. Surfaced on [`Output::swept_members`] (F14) so
    /// the caller can distinguish "clean exit" from "exit that killed a
    /// contained background member".
    swept_members: bool,
    /// Natural-exit policy for the contained session (see [`SessionExitPolicy`]).
    session_exit: SessionExitPolicy,
    /// True when the spawn is a pty session. Routes the kill paths through
    /// the session-total machinery ([`signal_session_pgids`]) instead of the
    /// single-group kill, and requires the pty master's EOF (drain
    /// `io_done`) as the authoritative completion condition rather than
    /// leader-reaped — the leader may be reaped while background pgrps still
    /// hold the slave (pty job-control dilemma doc §5b).
    pty: bool,
}

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

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

    /// Write bytes to a pty-spawned child's stdin (the master end).
    ///
    /// Only valid for a [`SpawnOptionsBuilder::pty`] spawn with an active
    /// stdout stream; the write is accepted by the tty line discipline and
    /// delivered to the child as its stdin.
    ///
    /// ### Errors
    /// - `EINVAL`: The spawn was not a pty spawn, or the stream is already
    ///   closed.
    /// - `EIO`: All slave holders have closed (master-side write failure).
    /// - `ETIMEDOUT`: The child did not drain its input within the bound.
    pub fn write_input(&self, bytes: &[u8]) -> Result<usize, CoreError> {
        self.drain.write_input(bytes)
    }

    /// Write bytes to a pty-spawned child's stdin without blocking.
    ///
    /// Returns `Ok(Some(n))` for the bytes written (may be a partial write
    /// when the tty input buffer fills), or `Ok(None)` on `EAGAIN` (buffer
    /// full). The caller owns the input queue: register `POLLOUT` interest on
    /// the master on `EAGAIN` and retry on writability.
    ///
    /// ### Errors
    /// - `EINVAL`: The spawn was not a pty spawn, or the stream is already
    ///   closed.
    /// - `EIO`: All slave holders have closed (master-side write failure).
    pub fn write_input_nonblock(&self, bytes: &[u8]) -> Result<Option<usize>, CoreError> {
        self.drain.write_input_nonblock(bytes)
    }

    /// Arm or disarm the pty master's WRITABLE interest (the input route).
    ///
    /// Direction-preserving: the readable (output) interest is never touched.
    /// The caller arms this when its input queue fills and flushes on each
    /// writable event, disarming when the queue drains. See
    /// [`DrainState::set_pty_writable`].
    ///
    /// ### Errors
    /// - `EINVAL`: The spawn was not a pty spawn, or the stream is already
    ///   closed.
    pub fn set_pty_writable(
        &mut self,
        reactor: &mut Reactor,
        writable: bool,
    ) -> Result<(), CoreError> {
        self.drain.set_pty_writable(reactor, writable)
    }

    /// The pty master's input-route reactor token, when this is a pty spawn.
    /// `None` for pipe mode (no input route). See
    /// [`DrainState::pty_input_token`].
    pub fn pty_input_token(&self) -> Option<Token> {
        self.drain.pty_input_token()
    }
}

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;
            }
        }
        // F6 sweep bound: the natural-path session sweep that started at
        // `sweep_started_at` must also wake the caller past the D-state bound,
        // or a D-state member would keep /proc re-enumeration alive forever.
        if let Some(started) = self.sweep_started_at {
            let bail_at = started + 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() {
            let finished = if self.pty {
                if self.cancel_at.is_some() {
                    // Cancellation: the pty master EOF is the authoritative
                    // completion (the leader may be reaped while background
                    // pgrps still hold the slave, §5b), bounded by the same
                    // D-state give-up as the pipe path. A paused stream must
                    // not short-circuit the session kill loop.
                    let bounded = io_done
                        || self
                            .kill_sent_at
                            .is_some_and(|t| now.duration_since(t) >= D_STATE_REAP_BOUND)
                        || (self.cancel == CancelPolicy::None
                            && self.deadline_passed_at.is_some_and(|passed| {
                                now.duration_since(passed) >= D_STATE_REAP_BOUND
                            }));
                    // H2: the D-state bound must not report the job ended while
                    // the master is open and session members remain — the bound
                    // only proves SIGKILL was sent 500 ms ago, and a member that
                    // survived it (D-state, or forked into the final window) would
                    // escape. Hold completion until the session is empty; the
                    // per-tick sweep keeps the SIGKILLs coming. `CancelPolicy::None`
                    // opted out of all signaling and keeps the legacy bound.
                    if self.cancel == CancelPolicy::None {
                        bounded
                    } else {
                        bounded && self.session_sweep_if_leader()?
                    }
                } else {
                    // H1/F5: natural completion for a pty session. `Sweep`
                    // (default) treats leader-reap as the sweep trigger, not
                    // master EOF: a slave-holding background member keeps the
                    // master open, so EOF alone would hang completion forever
                    // and the sweep (gated behind EOF) would never run
                    // (finding A4-3). The sweep runs on every tick and SIGKILLs
                    // the slave-holder; EOF fires once it dies, and completion
                    // stays gated on an empty session + drain. `LetMembersSurvive`
                    // reports completion on leader-reap without signaling the
                    // session (nohup-style background jobs keep running) —
                    // leader-reap remains the gate, so the hang stays
                    // impossible. `CancelPolicy::None` opted out of all
                    // signaling and keeps the legacy EOF-based completion.
                    if self.cancel == CancelPolicy::None {
                        io_done || paused
                    } else {
                        match self.session_exit {
                            SessionExitPolicy::LetMembersSurvive => true,
                            SessionExitPolicy::Sweep => {
                                let swept = self.session_sweep_if_leader()?;
                                if !swept {
                                    // The sweep found live members (and
                                    // SIGKILLed them) — surface that on the
                                    // completion output (F14).
                                    self.swept_members = true;
                                    if self.sweep_started_at.is_none() {
                                        self.sweep_started_at = Some(now);
                                    }
                                }
                                (io_done || paused) && swept
                            }
                        }
                    }
                }
            } else if self.cancel_at.is_some() {
                // H4: the group kill stops once the leader is reaped (its pid
                // may be recycled), but contained descendants — TERM-immune,
                // stopped, or D-state — would then escape unmanaged. Keep
                // sweeping the isolated session until /proc shows no live
                // members before reporting completion; `session_sweep` fires
                // the SIGKILLs and `advance_cancel` short-circuits on the
                // reaped leader (finding H4). `CancelPolicy::None` opted out
                // of all signaling and keeps the legacy leader-reap
                // completion.
                self.cancel == CancelPolicy::None || self.session_sweep_if_leader()?
            } else {
                // H6/F5: natural pipe completion. A non-session spawn has no
                // sweep surface (`session_sweep_if_leader` returns true), so
                // the legacy EOF gate holds. An isolated pipe session with
                // `Sweep` gets the same leader-reap sweep semantics as a pty
                // (an fd-detached descendant would otherwise survive a job
                // reported exit-0 — finding H6); `LetMembersSurvive` reports
                // on leader-reap without signaling.
                if self.cancel == CancelPolicy::None {
                    io_done || paused
                } else {
                    match self.session_exit {
                        SessionExitPolicy::LetMembersSurvive => {
                            if self.pgroup.isolated {
                                true
                            } else {
                                io_done || paused
                            }
                        }
                        SessionExitPolicy::Sweep => {
                            let swept = self.session_sweep_if_leader()?;
                            if !swept {
                                self.swept_members = true;
                                if self.sweep_started_at.is_none() {
                                    self.sweep_started_at = Some(now);
                                }
                            }
                            (io_done || paused) && swept
                        }
                    }
                }
            };
            if finished {
                return self.finish(reactor, !io_done).map(Some);
            }
        }
        // D-state / sweep give-up (F6): the SIGKILL for an unreapable leader has
        // been pending past the bound, OR the natural-path session sweep has
        // been running past the bound without converging (a D-state member
        // keeps SIGKILL pending until it wakes). The sweep arm has no
        // `status.is_none()` requirement: a reaped leader with a live member
        // in D-state must also give up, or /proc re-enumeration runs forever.
        let kill_gave_up = self.status.is_none()
            && self
                .kill_sent_at
                .is_some_and(|t| now.duration_since(t) >= D_STATE_REAP_BOUND);
        let sweep_gave_up = self
            .sweep_started_at
            .is_some_and(|t| now.duration_since(t) >= D_STATE_REAP_BOUND);
        if kill_gave_up || sweep_gave_up {
            // H2: the give-up exists because the *leader* is unreapable or a
            // *member* cannot be killed, not because the kill has converged —
            // stopping the sweep here would let them leak. Hand the session to
            // the detached reaper (safe: the live, unreapable leader still
            // pins the sid; a reaped leader makes the starttime-gated
            // `orphan_session` a safe no-op — the pending SIGKILL from the
            // last sweep tick dies when the member wakes). The reaper keeps
            // SIGKILLing until /proc empties.
            if self.pty || self.pgroup.isolated {
                orphan_session(self.pid);
            }
            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)
    }

    /// Track B (H1/H6): run the session-emptiness sweep iff the child is an
    /// isolated session leader. A pty spawn always is (`setsid`, validated);
    /// an isolated pipe spawn setsid's too, so `sid == self.pid` and a `/proc`
    /// scan by `self.pid` reaches exactly the contained session. A non-isolated
    /// pipe spawn shares the caller's session — scanning by `self.pid` would
    /// hit unrelated processes, so it stays on the legacy EOF-based completion.
    /// Returns `Ok(true)` when there is nothing to sweep (non-session spawn) or
    /// the session is empty; `Err` when the scan failed (F7 fail-closed).
    fn session_sweep_if_leader(&mut self) -> Result<bool, CoreError> {
        if !(self.pty || self.pgroup.isolated) {
            return Ok(true);
        }
        session_sweep(self.pid)
    }

    fn advance_cancel(&mut self, now: Instant) -> Result<(), CoreError> {
        let Some(cancel_at) = self.cancel_at else {
            return Ok(());
        };
        // A reaped child must not be signaled — its pid may be recycled.
        // Exception: a pty session, where the leader may be reaped while
        // background pgrps still hold the master; those are the session kill
        // loop's responsibility. For a non-pty session the reaped leader ends
        // the group kill, but contained descendants are swept by the H4
        // completion gate in `poll_completion` (kill-totality), so they do
        // not escape unmanaged either.
        if self.status.is_some() && !self.pty {
            return Ok(());
        }
        let running = self
            .running
            .as_ref()
            .ok_or_else(|| CoreError::sys(libc::EINVAL, "managed process completed"))?;
        if self.pty {
            return self.advance_pty_cancel(now, cancel_at, running.io_done());
        }
        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(())
    }

    /// The pty-session cancellation state machine (§5b of the pty job-control
    /// dilemma doc): enumerate the session's PGIDs once, SIGCONT+SIGTERM each,
    /// then after the grace period re-enumerate and SIGKILL every group that
    /// is still present, repeating the (bounded) re-enumeration while the pty
    /// master is still open. The master's EOF — not "all PGIDs disappeared" —
    /// is the authoritative completion condition: it is the kernel's own
    /// observation that every slave holder has exited. The D-state give-up
    /// bound in [`ManagedProcess::poll_completion`] caps the total `/proc`
    /// cost of a pathological spawner that keeps creating descendants.
    fn advance_pty_cancel(
        &mut self,
        now: Instant,
        cancel_at: Instant,
        master_eof: bool,
    ) -> Result<(), CoreError> {
        // Once the master has EOF'd the session is empty by the kernel's own
        // account — there is nothing left to signal.
        if master_eof {
            return Ok(());
        }
        // A pty spawn is a session leader (`setsid`): sid == leader pid.
        let sid = self.pid;
        match self.kill_state {
            KillState::None => match self.cancel {
                CancelPolicy::None => {}
                CancelPolicy::Graceful => {
                    signal_session_pgids(sid, libc::SIGTERM, true)?;
                    self.kill_state = KillState::TermSent;
                }
                CancelPolicy::Kill => {
                    signal_session_pgids(sid, libc::SIGKILL, false)?;
                    self.kill_state = KillState::KillSent;
                    self.kill_sent_at = Some(now);
                }
            },
            KillState::TermSent if now >= cancel_at + self.kill_grace => {
                signal_session_pgids(sid, libc::SIGKILL, false)?;
                self.kill_state = KillState::KillSent;
                self.kill_sent_at = Some(now);
            }
            // Bounded re-enumeration: a group created after the first snapshot
            // is still a slave holder and keeps the master open; SIGKILL it.
            // `kill_sent_at` drives the give-up bound in `poll_completion`.
            KillState::KillSent => {
                signal_session_pgids(sid, libc::SIGKILL, false)?;
            }
            _ => {}
        }
        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,
            swept_members: self.swept_members,
        })
    }

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

    /// Write bytes to a pty-spawned child's stdin (the master end).
    ///
    /// Only valid for a [`SpawnOptionsBuilder::pty`] spawn with an active
    /// stdout stream. See [`RunningProcess::write_input`].
    pub fn write_input(&self, bytes: &[u8]) -> Result<usize, CoreError> {
        self.running
            .as_ref()
            .ok_or_else(|| CoreError::sys(libc::EINVAL, "managed process completed"))?
            .write_input(bytes)
    }

    /// Write bytes to a pty-spawned child's stdin without blocking.
    ///
    /// Only valid for a [`SpawnOptionsBuilder::pty`] spawn with an active
    /// stdout stream. See [`RunningProcess::write_input_nonblock`].
    pub fn write_input_nonblock(&self, bytes: &[u8]) -> Result<Option<usize>, CoreError> {
        self.running
            .as_ref()
            .ok_or_else(|| CoreError::sys(libc::EINVAL, "managed process completed"))?
            .write_input_nonblock(bytes)
    }

    /// Arm or disarm the pty master's WRITABLE interest (the input route).
    ///
    /// Only valid for a [`SpawnOptionsBuilder::pty`] spawn with an active
    /// stdout stream. See [`RunningProcess::set_pty_writable`].
    pub fn set_pty_writable(
        &mut self,
        reactor: &mut Reactor,
        writable: bool,
    ) -> Result<(), CoreError> {
        self.running
            .as_mut()
            .ok_or_else(|| CoreError::sys(libc::EINVAL, "managed process completed"))?
            .set_pty_writable(reactor, writable)
    }

    /// The pty master's input-route reactor token, when this is a pty spawn.
    /// `None` for pipe mode (no input route). See
    /// [`RunningProcess::pty_input_token`].
    pub fn pty_input_token(&self) -> Option<Token> {
        self.running
            .as_ref()
            .ok_or_else(|| CoreError::sys(libc::EINVAL, "managed process completed"))
            .ok()
            .and_then(|p| p.pty_input_token())
    }
}

impl Drop for ManagedProcess {
    fn drop(&mut self) {
        let Some(running) = self.running.take() else {
            return;
        };
        let process = &running.process;
        let pid = process.pid();
        let session_like = self.pty || self.pgroup.isolated;
        // If the child was already reaped by `poll_completion`, the pid may
        // have been recycled — never signal it blindly. For a NON-session
        // spawn this is the end: the pipes are dropped with `running` and
        // there is nothing left to clean up. For a SESSION spawn (pty /
        // isolated) the leader can be reaped while contained members still
        // hold the pty slave — the session kill must still run (F1/F2). The
        // fresh sweep is itself the freshness check: it only SIGKILLs groups
        // that are live in the session *right now*, so a recycled sid whose
        // leader is gone cannot be hit by a stale numeric kill. If the sweep
        // does not converge, hand the session to the detached reaper
        // (starttime-gated, so it will not sweep a recycled sid either).
        if self.status.is_some() {
            if !session_like || self.cancel == CancelPolicy::None {
                return;
            }
            // F7 fail-closed: an Err sweep must NOT be treated as "empty" —
            // hand the session to the reaper (which keeps retrying) instead
            // of silently reporting a clean Drop with live members.
            if !session_sweep(pid).unwrap_or(false) {
                orphan_session(pid);
            }
            return;
        }
        // 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 {
            if self.pty {
                // Session-total kill: the pty session may be fragmented into
                // several pgrps; enumerate once and SIGKILL every group.
                if signal_session_pgids(self.pid, libc::SIGKILL, false).is_err() {
                    // Scan failure: the kill cannot be verified total. Hand the
                    // session to the reaper, which keeps retrying each tick
                    // (F7 fail-closed — never drop without a confirmed kill).
                    orphan_session(self.pid);
                }
            } else {
                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)
    }
}

/// Enumerate the distinct process group ids that share session `sid`, from a
/// single `/proc` traversal.
///
/// Used only at termination time (see the pty job-control dilemma doc §5b): a
/// session does thousands of reads/writes/resizes but is cancelled at most
/// once, so one scan per cancellation is bounded and unobjectionable — never
/// in the streaming path. Natural completion (the H1/H6 emptiness gate) also
/// scans: EOF alone is not authoritative, so termination is where the session
/// is verified empty (or swept).
///
/// The test-only enumeration counter ([`pty_session_enumerated`]) proves that
/// invariant: a live, streaming session performs zero `/proc` walks.
fn session_pgids(sid: pid_t) -> Result<HashSet<pid_t>, CoreError> {
    session_pgids_at(std::path::Path::new("/proc"), sid)
}

/// [`session_pgids`] against an explicit proc root — the testable seam (F7):
/// a scan against a root that cannot be read must return `Err`, never a
/// fabricated empty set.
pub(super) fn session_pgids_at(
    proc_root: &std::path::Path,
    sid: pid_t,
) -> Result<HashSet<pid_t>, CoreError> {
    #[cfg(test)]
    PTY_SESSION_ENUMERATIONS.lock().unwrap().insert(sid);
    let mut pgids = HashSet::new();
    // Fail-closed: a session that cannot be scanned must not silently look
    // empty (F7). A false "empty" would let the caller report the job
    // COMPLETED with contained members still live. The caller decides how to
    // react (fail the job, keep the sid registered for the reaper's next
    // 250 ms tick); Core never fabricates an empty scan.
    let entries = std::fs::read_dir(proc_root).map_err(|e| {
        CoreError::sys(
            e.raw_os_error().unwrap_or(libc::EIO),
            "session_pgids:read_dir",
        )
    })?;
    for entry in entries.flatten() {
        let name = entry.file_name();
        let Some(name) = name.to_str() else { continue };
        let Ok(_pid) = name.parse::<pid_t>() else {
            continue;
        };
        // /proc/[pid]/stat: "pid (comm) state ppid pgrp session tty_nr tpgid …".
        // `comm` may contain spaces and ')' — split on the LAST ')'.
        let Ok(stat) = std::fs::read_to_string(proc_root.join(name).join("stat")) else {
            continue;
        };
        let Some(rest) = stat.rsplit_once(')') else {
            continue;
        };
        let Some(rest) = rest.1.strip_prefix(' ') else {
            continue;
        };
        let mut fields = rest.split(' ');
        let _state = fields.next();
        let _ppid = fields.next();
        let Some(pgrp) = fields.next().and_then(|f| f.parse::<pid_t>().ok()) else {
            continue;
        };
        let Some(sess) = fields.next().and_then(|f| f.parse::<pid_t>().ok()) else {
            continue;
        };
        if sess == sid && pgrp > 0 {
            pgids.insert(pgrp);
        }
    }
    Ok(pgids)
}

/// Test-only record of which session ids have been enumerated via
/// [`session_pgids`]. Compiled out of production builds — the enumeration
/// itself is a rare control-plane operation (once per termination), and this
/// record exists solely to prove it never enters the terminal data path.
///
/// Per-session (not a global counter) so tests can run in parallel: each pty
/// test asserts its own session was never enumerated during streaming and was
/// enumerated at termination, without interference from other tests' sessions.
#[cfg(test)]
static PTY_SESSION_ENUMERATIONS: std::sync::LazyLock<Mutex<HashSet<pid_t>>> =
    std::sync::LazyLock::new(|| Mutex::new(HashSet::new()));

/// Clear the test-only enumeration record ([`PTY_SESSION_ENUMERATIONS`]).
#[cfg(test)]
pub(crate) fn reset_pty_enumeration() {
    PTY_SESSION_ENUMERATIONS.lock().unwrap().clear();
}

/// Whether the given session has been enumerated via [`session_pgids`] since
/// the last reset. A live, streaming session must report `false` — only
/// cancellation/drop enumerates.
#[cfg(test)]
pub(crate) fn pty_session_enumerated(sid: pid_t) -> bool {
    PTY_SESSION_ENUMERATIONS.lock().unwrap().contains(&sid)
}

/// Signal every process group in the session identified by `sid`.
///
/// A pty spawn is a session leader (`setsid` ⇒ sid == leader pid), and job
/// control can fragment the session into many pgrps (`setpgid` is allowed
/// under the pty containment variant). `kill(-sid)` would only reach the
/// leader's own group, and the pty master's EOF is an observation, not a
/// delivery mechanism (§5b E1/E2) — so the kill mechanism is: enumerate the
/// session's PGIDs at termination and signal each one (§5b E3). A group that
/// has already exited yields `ESRCH` and is ignored, matching the existing
/// `signal_process` behavior.
///
/// F7 fail-closed: a session whose scan fails returns `Err` instead of
/// silently signaling nobody — a "clean" termination that never killed the
/// contained members must not look successful. The caller decides whether to
/// retry, fail the job, or hand the session to the reaper.
fn signal_session_pgids(sid: pid_t, sig: i32, cont_before: bool) -> Result<(), CoreError> {
    for pgid in session_pgids(sid)? {
        if cont_before {
            // A stopped group must be continued before it can take the signal;
            // otherwise SIGTERM is queued and the process stays immune.
            unsafe {
                libc::kill(-pgid, libc::SIGCONT);
            }
        }
        unsafe {
            libc::kill(-pgid, sig);
        }
    }
    Ok(())
}

/// Track B (H1/H6): report whether the session `sid` is empty, and when it is
/// not, SIGKILL every contained live group.
///
/// Gates natural completion: master/pipe EOF proves the stdio fds closed, not
/// that the session is empty — a slave-/fd-detached background member would
/// otherwise outlive a job reported COMPLETED (pty job-control dilemma doc
/// §5b, findings H1/H6). `setsid` is denied under the containment filter, so
/// every member stays in this session and the sweep is total; the caller
/// re-checks until this returns `true`.
///
/// Only LIVE members count toward emptiness: a zombie has already died and is
/// merely awaiting reap by its parent/init — SIGKILL on it is a no-op and it
/// cannot outlive the job, so gating on it would delay completion by init's
/// reap timing. A member stuck in D-state keeps the signal pending until it
/// leaves D-state (H2 seam; tracked in REDTEAM-NATIVE-MIGRATION-REVIEW.md).
fn session_sweep(sid: pid_t) -> Result<bool, CoreError> {
    session_sweep_at(std::path::Path::new("/proc"), sid)
}

/// [`session_sweep`] against an explicit proc root — the testable seam (F7):
/// a scan against a root that cannot be read must return `Err`, never a
/// fabricated "empty" (which would report the job COMPLETED with live
/// members).
pub(super) fn session_sweep_at(proc_root: &std::path::Path, sid: pid_t) -> Result<bool, CoreError> {
    #[cfg(test)]
    PTY_SESSION_ENUMERATIONS.lock().unwrap().insert(sid);
    let mut live_pgrps = HashSet::new();
    // Fail-closed (F7): a session that cannot be scanned must not silently
    // look empty. `Ok(true)` means "empty" and is the ONLY thing that lets
    // the caller report COMPLETED; returning `true` on a failed scan would
    // report the job done with contained members still live. The caller
    // decides how to react; Core never fabricates an empty scan.
    let entries = std::fs::read_dir(proc_root).map_err(|e| {
        CoreError::sys(
            e.raw_os_error().unwrap_or(libc::EIO),
            "session_sweep:read_dir",
        )
    })?;
    for entry in entries.flatten() {
        let name = entry.file_name();
        let Some(name) = name.to_str() else { continue };
        let Ok(_pid) = name.parse::<pid_t>() else {
            continue;
        };
        // /proc/[pid]/stat: "pid (comm) state ppid pgrp session tty_nr tpgid …".
        // `comm` may contain spaces and ')' — split on the LAST ')'.
        let Ok(stat) = std::fs::read_to_string(proc_root.join(name).join("stat")) else {
            continue;
        };
        let Some(rest) = stat.rsplit_once(')') else {
            continue;
        };
        let Some(rest) = rest.1.strip_prefix(' ') else {
            continue;
        };
        let mut fields = rest.split(' ');
        let state = fields.next();
        let _ppid = fields.next();
        let Some(pgrp) = fields.next().and_then(|f| f.parse::<pid_t>().ok()) else {
            continue;
        };
        let Some(sess) = fields.next().and_then(|f| f.parse::<pid_t>().ok()) else {
            continue;
        };
        if sess == sid && pgrp > 0 && state != Some("Z") {
            live_pgrps.insert(pgrp);
        }
    }
    if live_pgrps.is_empty() {
        return Ok(true);
    }
    for pgid in live_pgrps {
        // A stopped group must be continued before it can take the signal;
        // otherwise the queued SIGKILL stays pending and the group survives.
        unsafe {
            libc::kill(-pgid, libc::SIGCONT);
        }
        unsafe {
            libc::kill(-pgid, libc::SIGKILL);
        }
    }
    Ok(false)
}

/// 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 pty = opts.pty;
    let session_exit = opts.session_exit;
    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,
        sweep_started_at: None,
        swept_members: false,
        session_exit,
        pty,
    })
}

/// 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 pty = opts.pty;
    let session_exit = opts.session_exit;

    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,
            swept_members: false,
        });
    }

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

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

#[allow(clippy::too_many_arguments)] // internal completion state machine; grouped params obscure the flow
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,
    pty: bool,
    session_exit: SessionExitPolicy,
) -> 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;
    // When the natural-path session sweep first started (F6 give-up bound for
    // a D-state member under `SessionExitPolicy::Sweep`).
    let mut sweep_started_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;
    // F14: set when a natural-exit session sweep found and SIGKILLed at least
    // one live session member; surfaced on `Output::swept_members`.
    let mut swept_members = false;

    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. Exception: a pty session,
                // where the leader may be reaped while background pgrps still
                // hold the master; those are the session kill loop's
                // responsibility (§5b). The wedged-pipe path below returns the
                // partial output without sending any signal.
                if !(status_raw.is_some() && !pty) {
                    match state {
                        KillState::None => {
                            if pty {
                                match cancel {
                                    CancelPolicy::Graceful => {
                                        signal_session_pgids(pid, libc::SIGTERM, true)?;
                                        state = KillState::TermSent;
                                    }
                                    CancelPolicy::Kill => {
                                        signal_session_pgids(pid, libc::SIGKILL, false)?;
                                        state = KillState::KillSent;
                                        kill_sent_at = Some(Instant::now());
                                    }
                                    CancelPolicy::None => {}
                                }
                            } else 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 => {
                            if pty {
                                signal_session_pgids(pid, libc::SIGKILL, false)?;
                            } else {
                                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());
                        }
                        // Bounded re-enumeration for a pty session: a group
                        // created after the first snapshot is still a slave
                        // holder and keeps the master open; SIGKILL it. The
                        // D-state give-up bound below caps the /proc cost.
                        KillState::KillSent if pty && !drain.is_done() => {
                            signal_session_pgids(pid, libc::SIGKILL, false)?;
                        }
                        _ => {}
                    }
                }
                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);
        }

        // F5: natural completion for a contained session under `Sweep`
        // triggers on leader-reap, not master EOF. A slave-holding background
        // member keeps the master open, so EOF alone would hang completion
        // forever and the sweep (gated behind `drain.is_done`) would never run
        // (finding A4-3). Sweep every tick once the leader is reaped; the
        // SIGKILL releases the slave so EOF can fire, and the completion path
        // below stays gated on an empty session + drain.
        if status_raw.is_some()
            && (pty || pgroup.isolated)
            && cancel != CancelPolicy::None
            && session_exit == SessionExitPolicy::Sweep
            && !drain.is_done()
        {
            // F7 fail-closed: a failed scan must not be treated as "empty" —
            // propagate the Err so the job fails rather than reporting COMPLETED
            // with live members.
            if !session_sweep(pid)? {
                swept_members = true;
                if sweep_started_at.is_none() {
                    sweep_started_at = Some(Instant::now());
                }
            }
        }

        // F5: `LetMembersSurvive` reports natural completion on leader-reap
        // without signaling the session (nohup-style background jobs keep
        // running). The master may still be open — force-close the drain and
        // return the partial output with the reaped status.
        if status_raw.is_some()
            && (pty || pgroup.isolated)
            && cancel != CancelPolicy::None
            && session_exit == SessionExitPolicy::LetMembersSurvive
            && !timed_out
        {
            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.take(),
                stdout,
                stderr,
                timed_out,
                stdout_early_exited,
                stdout_pending,
                stderr_pending,
                swept_members,
            });
        }

        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 {
                // H6/F5: EOF + leader reap must not be reported while contained
                // session members survive under `Sweep` — a detached descendant
                // would outlive a job reported exit-0 (finding H6). Sweep the
                // isolated session until /proc shows no live members before
                // completing. `LetMembersSurvive` completes on leader-reap
                // without signaling (handled by the earlier branch; this is the
                // EOF-first path where the leader may still be alive).
                // `CancelPolicy::None` opted out of all signaling and keeps the
                // legacy EOF-based completion.
                let sweep_needed = (pty || pgroup.isolated)
                    && cancel != CancelPolicy::None
                    && session_exit == SessionExitPolicy::Sweep
                    && !session_sweep(pid)?;
                if sweep_needed {
                    swept_members = true;
                    status_raw = Some(s);
                    // The pipes are already EOF'd, so there are no further
                    // readiness events; bound the reactor wait so the loop
                    // re-scans the session at the existing 10 ms cadence
                    // without a raw thread sleep (a D-state member keeps the
                    // SIGKILL pending until it wakes — H2 seam).
                    if sweep_started_at.is_none() {
                        sweep_started_at = Some(Instant::now());
                    }
                    poll_timeout = 10;
                } else {
                    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,
                        swept_members,
                    });
                }
            }
        }

        // 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()) {
            // H1/H6/F5: a paused stream + reaped leader must also wait for the
            // session to empty before reporting completion under `Sweep` (the
            // wait_loop mirror of `poll_completion`'s `io_done || paused`
            // gate); `LetMembersSurvive` never sweeps, so it completes here.
            if (pty || pgroup.isolated)
                && cancel != CancelPolicy::None
                && session_exit == SessionExitPolicy::Sweep
                && !session_sweep(pid)?
            {
                swept_members = true;
                if sweep_started_at.is_none() {
                    sweep_started_at = Some(Instant::now());
                }
                // Fall through to the bounded reactor wait below: the paused
                // stream produces no readiness events, and the backpressure
                // block bounds the poll at 10 ms, re-scanning the session on
                // the existing cadence without a raw thread sleep.
                poll_timeout = 10;
            } else {
                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,
                    swept_members,
                });
            }
        }

        // 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. For a pty session the master is
        // the drain, and "wedged" means a background pgrp still holds the
        // slave: the session kill loop must get its bounded chance first, so
        // only finish on master EOF or after the D-state give-up bound.
        if timed_out && status_raw.is_some() {
            // F7 fail-closed: a failed sweep scan must fail the job (Err)
            // rather than let a false "empty" return the timed-out result.
            let pty_sweep = if pty && cancel != CancelPolicy::None {
                Some(session_sweep(pid)?)
            } else {
                None
            };
            let isolated_sweep = if pgroup.isolated && cancel != CancelPolicy::None && !pty {
                Some(session_sweep(pid)?)
            } else {
                None
            };
            let can_finish = if pty {
                if cancel == CancelPolicy::None {
                    drain.is_done()
                        || kill_sent_at.is_some_and(|t| t.elapsed() >= D_STATE_REAP_BOUND)
                } else {
                    // H2: the D-state bound must not return the timed-out
                    // result while the master is open and session members
                    // remain — the bound only proves SIGKILL was sent 500 ms
                    // ago. Hold until the sweep empties the session (a
                    // D-state member keeps SIGKILL pending until it wakes).
                    (drain.is_done()
                        || kill_sent_at.is_some_and(|t| t.elapsed() >= D_STATE_REAP_BOUND))
                        && pty_sweep.unwrap_or(false)
                }
            } else if pgroup.isolated && cancel != CancelPolicy::None {
                // H4: the group kill stopped when the leader was reaped, but
                // contained descendants (e.g. the member holding the pipe
                // write end) survive — sweep the session until /proc shows no
                // live members before returning the timed-out result
                // (kill-totality, finding H4). A D-state member keeps the
                // SIGKILL pending until it wakes (H2 seam).
                isolated_sweep.unwrap_or(false)
            } else {
                true
            };
            if can_finish {
                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,
                    swept_members,
                });
            }
        }

        // D-state / sweep give-up (F6): SIGKILL has been sent but the child is
        // still unreaped after the bound, OR the natural-path session sweep has
        // been running past the bound without converging (a D-state member
        // keeps SIGKILL pending until it wakes). 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).
        let kill_gave_up = kill_sent_at
            .is_some_and(|sent_at| sent_at.elapsed() >= D_STATE_REAP_BOUND)
            && status_raw.is_none();
        let sweep_gave_up = sweep_started_at.is_some_and(|t| t.elapsed() >= D_STATE_REAP_BOUND);
        if kill_gave_up || sweep_gave_up {
            for slot in drain.take_all_slots() {
                if slot.token.is_some() {
                    let _ = reactor.del(&slot.fd);
                }
            }
            // H2: the give-up is about the unreapable *leader* or an
            // unkillable *member*; others may still be alive, and stopping the
            // sweep here would leak them. Hand the session to the detached
            // reaper (safe: the live leader pins the sid; a reaped leader
            // makes the starttime-gated `orphan_session` a safe no-op), which
            // keeps SIGKILLing until /proc empties (finding H2). The pending
            // SIGKILL dies when the member wakes.
            if pty || pgroup.isolated {
                orphan_session(pid);
            }
            // 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). A reaped leader is already
            // gone — skip the re-registration.
            if status_raw.is_none() {
                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: status_raw,
                stdout,
                stderr,
                timed_out,
                stdout_early_exited,
                stdout_pending,
                stderr_pending,
                swept_members,
            });
        }

        // `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,
                swept_members,
            });
        }

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