processkit 2.3.1

//! Windows implementation: a [Job Object] with kill-on-close.
//!
//! [Job Object]: https://learn.microsoft.com/windows/win32/procthread/job-objects

use std::io;
use std::time::Duration;

use tokio::process::{Child, Command};
// The process-creation `FILETIME` is both the `stats` identity anchor and the
// `process-control` member-snapshot start time, so it (and `GetProcessTimes`
// below) is needed under either feature.
#[cfg(any(feature = "stats", feature = "process-control"))]
use windows_sys::Win32::Foundation::FILETIME;
use windows_sys::Win32::Foundation::{CloseHandle, HANDLE, INVALID_HANDLE_VALUE};
#[cfg(feature = "process-control")]
use windows_sys::Win32::Foundation::{ERROR_INVALID_PARAMETER, ERROR_MORE_DATA};
use windows_sys::Win32::System::Console::{CTRL_BREAK_EVENT, GenerateConsoleCtrlEvent};
use windows_sys::Win32::System::Diagnostics::ToolHelp::{
    CreateToolhelp32Snapshot, TH32CS_SNAPTHREAD, THREADENTRY32, Thread32First, Thread32Next,
};
// System-wide process snapshot for `members_info`'s per-member ppid + image name
// (no per-pid Win32 API yields the parent pid without ntdll).
#[cfg(feature = "process-control")]
use windows_sys::Win32::System::Diagnostics::ToolHelp::{
    PROCESSENTRY32W, Process32FirstW, Process32NextW, TH32CS_SNAPPROCESS,
};
// Ungated: the opt-in console-CTRL graceful teardown is compiled on every
// Windows feature config, and its drain check (`QueryInformationJobObject` on
// the accounting info) and per-leader recycle guard (`IsProcessInJob`) can't be
// gated behind `process-control`/`stats`.
use windows_sys::Win32::System::JobObjects::IsProcessInJob;
use windows_sys::Win32::System::JobObjects::QueryInformationJobObject;
use windows_sys::Win32::System::JobObjects::{
    AssignProcessToJobObject, CreateJobObjectW, JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE,
    JOBOBJECT_EXTENDED_LIMIT_INFORMATION, JobObjectExtendedLimitInformation,
    SetInformationJobObject, TerminateJobObject,
};
#[cfg(feature = "limits")]
use windows_sys::Win32::System::JobObjects::{
    JOB_OBJECT_CPU_RATE_CONTROL_ENABLE, JOB_OBJECT_CPU_RATE_CONTROL_HARD_CAP,
    JOB_OBJECT_LIMIT_ACTIVE_PROCESS, JOB_OBJECT_LIMIT_JOB_MEMORY,
    JOBOBJECT_CPU_RATE_CONTROL_INFORMATION, JobObjectCpuRateControlInformation,
};
use windows_sys::Win32::System::JobObjects::{
    JOBOBJECT_BASIC_ACCOUNTING_INFORMATION, JobObjectBasicAccountingInformation,
};
#[cfg(feature = "process-control")]
use windows_sys::Win32::System::JobObjects::{
    JOBOBJECT_BASIC_PROCESS_ID_LIST, JobObjectBasicProcessIdList,
};
#[cfg(feature = "stats")]
use windows_sys::Win32::System::ProcessStatus::{K32GetProcessMemoryInfo, PROCESS_MEMORY_COUNTERS};
#[cfg(any(feature = "stats", feature = "process-control"))]
use windows_sys::Win32::System::Threading::GetProcessTimes;
use windows_sys::Win32::System::Threading::{
    CREATE_NEW_PROCESS_GROUP, CREATE_SUSPENDED, OpenThread, ResumeThread, THREAD_SUSPEND_RESUME,
};
#[cfg(feature = "process-control")]
use windows_sys::Win32::System::Threading::{
    GetExitCodeProcess, GetProcessIdOfThread, SuspendThread, THREAD_QUERY_LIMITED_INFORMATION,
};
use windows_sys::Win32::System::Threading::{OpenProcess, PROCESS_QUERY_LIMITED_INFORMATION};

use crate::Mechanism;
#[cfg(feature = "process-control")]
use crate::Signal;
#[cfg(feature = "limits")]
use crate::limits::ResourceLimits;
#[cfg(feature = "process-control")]
use crate::member::MemberInfo;
#[cfg(feature = "stats")]
use crate::stats::ProcessGroupStats;
#[cfg(feature = "stats")]
use crate::sys::{ProcIdentity, ProcMetrics};

pub(crate) struct Job {
    /// The job handle — deliberately non-inheritable and never duplicated:
    /// when this process dies (however abruptly), the kernel closes the last
    /// handle and `KILL_ON_JOB_CLOSE` takes the whole tree. That free
    /// kill-on-parent-death guarantee (documented on
    /// `Command::kill_on_parent_death`) breaks if a refactor ever duplicates
    /// or inherits this handle.
    ///
    /// One inherent gap (C10): a child is spawned `CREATE_SUSPENDED` and only then
    /// assigned to the job. If the **parent dies abruptly in that spawn→assign
    /// window** — after `CreateProcess` returns but before `AssignProcessToJobObject`
    /// — the child is not yet a job member, so kill-on-close can't reach it and it
    /// leaks as a permanently-*suspended* orphan (it never ran). The window is a
    /// few instructions wide and the orphan is inert (suspended), but it is not
    /// covered by the "kernel kills the tree even on abrupt parent death" headline.
    handle: HANDLE,
    /// Serializes `spawn`'s create-suspended → assign → resume sequence against
    /// the [`suspend`](Self::suspend)/[`resume`](Self::resume) member-thread
    /// walks. Without it, a walk landing between assign and `spawn`'s resume
    /// double-suspends the new child's primary thread (per-thread suspend
    /// *counts*), and `spawn`'s single resume leaves it suspended forever.
    suspend_lock: std::sync::Mutex<()>,
    /// Set by `graceful_shutdown(escalate=false)` so `Drop` clears
    /// `KILL_ON_JOB_CLOSE` before closing the handle, leaving survivors alive.
    skip_drop_kill: super::SkipDropKill,
    /// Pids of direct children spawned `CREATE_NEW_PROCESS_GROUP` for the opt-in
    /// console-CTRL graceful path (via
    /// [`Command::windows_graceful_ctrl_break`](crate::Command::windows_graceful_ctrl_break)).
    /// Empty unless a child opted in: `graceful_shutdown` takes the CTRL_BREAK →
    /// grace → `TerminateJobObject` path **iff** this is non-empty, so the default
    /// atomic-kill behavior is untouched for every other job. Each pid is a console
    /// **process-group id** (equal to the leader's pid) addressable by
    /// `GenerateConsoleCtrlEvent`; a per-leader `IsProcessInJob` re-check at signal
    /// time keeps a recycled pid from diverting the event onto a stranger's group.
    ctrl_break_leaders: std::sync::Mutex<Vec<u32>>,
}

// The handle is owned solely by this struct and every Win32 job API used here is
// thread-safe, so the raw pointer is sound to send/share across threads.
unsafe impl Send for Job {}
unsafe impl Sync for Job {}

impl Job {
    pub(crate) fn new(#[cfg(feature = "limits")] limits: &ResourceLimits) -> io::Result<Self> {
        // SAFETY: null name/attributes request an unnamed job with defaults.
        let handle = unsafe { CreateJobObjectW(std::ptr::null(), std::ptr::null()) };
        if handle.is_null() {
            return Err(io::Error::last_os_error());
        }
        let job = Job {
            handle,
            suspend_lock: std::sync::Mutex::new(()),
            skip_drop_kill: super::SkipDropKill::new(),
            ctrl_break_leaders: std::sync::Mutex::new(Vec::new()),
        };

        // Kill every process in the job once the last handle closes — i.e. when
        // this struct drops or the owning process dies. This is the Windows
        // analogue of `cgroup.kill` / `killpg`. The memory and process-count caps
        // ride along on the same extended-limit struct.
        let mut info: JOBOBJECT_EXTENDED_LIMIT_INFORMATION = unsafe { std::mem::zeroed() };
        info.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE;
        #[cfg(feature = "limits")]
        {
            if let Some(bytes) = limits.max_memory {
                info.BasicLimitInformation.LimitFlags |= JOB_OBJECT_LIMIT_JOB_MEMORY;
                // `JobMemoryLimit` is SIZE_T; saturate rather than wrap on a 32-bit host.
                info.JobMemoryLimit = usize::try_from(bytes).unwrap_or(usize::MAX);
            }
            if let Some(n) = limits.max_processes {
                info.BasicLimitInformation.LimitFlags |= JOB_OBJECT_LIMIT_ACTIVE_PROCESS;
                info.BasicLimitInformation.ActiveProcessLimit = n;
            }
        }
        // SAFETY: `info` is a fully-initialised struct matching the info class and
        // its size is passed explicitly.
        let ok = unsafe {
            SetInformationJobObject(
                job.handle,
                JobObjectExtendedLimitInformation,
                std::ptr::from_ref(&info).cast(),
                std::mem::size_of::<JOBOBJECT_EXTENDED_LIMIT_INFORMATION>() as u32,
            )
        };
        if ok == 0 {
            // `job` drops here, closing the handle — no leak.
            return Err(io::Error::last_os_error());
        }

        // CPU quota is a separate info class. The hard cap is expressed in 1/100 of
        // a percent of *total* system CPU (1..=10000), so convert our per-core
        // fraction using the host's processor count.
        #[cfg(feature = "limits")]
        if let Some(cores) = limits.cpu_quota {
            let cpus = std::thread::available_parallelism().map_or(1.0, |n| n.get() as f64);
            let rate = cpu_hard_cap_rate(cores, cpus);
            let mut cpu: JOBOBJECT_CPU_RATE_CONTROL_INFORMATION = unsafe { std::mem::zeroed() };
            cpu.ControlFlags =
                JOB_OBJECT_CPU_RATE_CONTROL_ENABLE | JOB_OBJECT_CPU_RATE_CONTROL_HARD_CAP;
            cpu.Anonymous.CpuRate = rate;
            // SAFETY: fully-initialised struct matching the CPU-rate info class; size
            // passed explicitly. `job` drops (closing the handle) on the error path.
            let ok = unsafe {
                SetInformationJobObject(
                    job.handle,
                    JobObjectCpuRateControlInformation,
                    std::ptr::from_ref(&cpu).cast(),
                    std::mem::size_of::<JOBOBJECT_CPU_RATE_CONTROL_INFORMATION>() as u32,
                )
            };
            if ok == 0 {
                return Err(io::Error::last_os_error());
            }
        }

        Ok(job)
    }

    pub(crate) fn spawn(
        &self,
        cmd: &mut Command,
        opts: &crate::sys::SpawnOptions,
    ) -> io::Result<Child> {
        // Race-free containment: start the child's primary thread SUSPENDED so no
        // user code runs (and nothing can fork) before the process is in the job;
        // assign it, then resume. This closes the old spawn→assign window in
        // which a fast-forking child could have escaped the job. Win32 exposes
        // no flag getter, so this overwrite is also where the Command-carried
        // extras (e.g. CREATE_NO_WINDOW) are OR'd back in.
        //
        // Opt-in graceful CTRL path: OR in CREATE_NEW_PROCESS_GROUP so the direct
        // child becomes its own console process group, addressable by its pid via
        // `GenerateConsoleCtrlEvent(CTRL_BREAK_EVENT, pid)` at graceful teardown.
        // A side effect is that CTRL_C is disabled for the new group by default —
        // CTRL_BREAK, which the teardown sends, is unaffected.
        use std::os::windows::process::CommandExt;
        let mut flags = CREATE_SUSPENDED | opts.creation_flags;
        if opts.windows_new_process_group {
            flags |= CREATE_NEW_PROCESS_GROUP;
        }
        cmd.as_std_mut().creation_flags(flags);

        // Arm a reaper for the window between spawn and containment: the child is
        // suspended and not yet in the job, so until `AssignProcessToJobObject`
        // succeeds nothing would reap it — an early return or panic here would
        // leak a suspended orphan. Disarmed once contained, restoring the normal
        // "the job owns teardown" semantics. (A permanent `kill_on_drop` would
        // instead fight `graceful_shutdown(escalate=false)` survivor-sparing, and
        // tokio can't toggle it off after spawn.) Arm it *before* reading the
        // fallible `id()`/`raw_handle()` so even their `?` early-returns reap.
        let guard = UncontainedChildGuard::arm(cmd.spawn()?);
        let pid = guard.child().id().ok_or_else(|| {
            io::Error::other("child exited before it could be assigned to the job")
        })?;
        let handle = guard.child().raw_handle().ok_or_else(|| {
            io::Error::other("child exited before it could be assigned to the job")
        })?;
        // Hold the suspend lock across assign → resume: once assigned, the pid
        // is visible to a concurrent suspend()/resume() member walk, which
        // would otherwise skew the still-suspended primary thread's count
        // (suspend counts nest) and strand or prematurely release the child.
        // Poisoning is impossible to act on here — recover the guard.
        let _suspend_guard = self
            .suspend_lock
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        // SAFETY: the raw handle is valid until the child is dropped; `guard`
        // owns the child for the rest of this scope, well past this call.
        //
        // Nested jobs: if THIS process is itself inside a Job Object that forbids
        // breakaway, the assign can fail with `ERROR_ACCESS_DENIED`. On Windows 8+
        // jobs nest (the child joins our job *and* the outer one), so the common
        // case works; we do not set a breakaway flag (that would let children
        // escape our containment). On failure the suspended child is reaped (the
        // guard) and the error surfaced — we never leak an uncontained child.
        let ok = unsafe { AssignProcessToJobObject(self.handle, handle as HANDLE) };
        if ok == 0 {
            // The reaper kills the still-suspended child as `guard` drops.
            return Err(io::Error::last_os_error());
        }
        // Contained — release the primary thread. A failure here would strand a
        // suspended-but-contained process; the reaper kills it as `guard` drops.
        resume_process_threads(pid)?;
        // Re-arm the kill-on-drop backstop now the child is contained: a prior
        // graceful_shutdown(escalate=false) latched skip_drop_kill to spare
        // survivors; a fresh member must not be spared by that stale latch on
        // Drop. Done after successful containment so a failed spawn leaves the
        // spared survivors alone.
        self.skip_drop_kill.clear();
        // Opt-in: record this direct child as a console-CTRL leader so a later
        // graceful_shutdown addresses it with GenerateConsoleCtrlEvent. Recorded
        // only after successful containment (so a failed spawn tracks nothing) and
        // only when the child was actually spawned into its own process group.
        if opts.windows_new_process_group {
            self.ctrl_break_leaders
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner())
                .push(pid);
        }
        Ok(guard.disarm())
    }

    #[cfg(feature = "process-control")]
    pub(crate) fn adopt(&self, child: &Child) -> io::Result<()> {
        let handle = child
            .raw_handle()
            .ok_or_else(|| io::Error::other("child has no handle (already exited?)"))?;
        // SAFETY: the raw handle is valid while `child` is alive (borrowed here).
        let ok = unsafe { AssignProcessToJobObject(self.handle, handle as HANDLE) };
        if ok == 0 {
            let err = io::Error::last_os_error();
            // The assign fails for an already-terminated process. If the child has
            // in fact exited there is nothing to contain — return Ok (matching the
            // pgroup/cgroup backends); a genuine failure on a still-LIVE process
            // still propagates.
            if process_has_exited(handle as HANDLE) {
                return Ok(());
            }
            return Err(err);
        }
        // A new killable member joined the job — re-arm the kill-on-drop backstop
        // so a prior graceful_shutdown(escalate=false) latch doesn't spare it.
        self.skip_drop_kill.clear();
        Ok(())
    }

    pub(crate) fn kill_all(&self) -> io::Result<()> {
        // SAFETY: `self.handle` is a valid job handle for the lifetime of self.
        let ok = unsafe { TerminateJobObject(self.handle, 1) };
        if ok == 0 {
            return Err(io::Error::last_os_error());
        }
        Ok(())
    }

    /// A Job Object has no POSIX signals: only `Kill` is deliverable (it maps
    /// to the job terminate); everything else is reported as unsupported so the
    /// caller never believes a reload/interrupt was delivered.
    #[cfg(feature = "process-control")]
    pub(crate) fn signal(&self, sig: Signal) -> io::Result<()> {
        match sig {
            Signal::Kill => self.kill_all(),
            other => Err(io::Error::new(
                io::ErrorKind::Unsupported,
                format!("signal({other:?})"),
            )),
        }
    }

    #[cfg(feature = "process-control")]
    pub(crate) fn suspend(&self) -> io::Result<()> {
        self.for_each_member_thread(true)
    }

    #[cfg(feature = "process-control")]
    pub(crate) fn resume(&self) -> io::Result<()> {
        self.for_each_member_thread(false)
    }

    /// The pids currently assigned to the job (whole tree).
    #[cfg(feature = "process-control")]
    pub(crate) fn members(&self) -> io::Result<Vec<u32>> {
        job_member_pids(self.handle)
    }

    /// The whole tree's members enriched with ppid / image name / start time.
    ///
    /// The member set is the same whole-tree pid list as [`members`](Self::members).
    /// Parent pid and image name come from a single system-wide `Toolhelp32`
    /// process snapshot; the start time (creation `FILETIME`) from a per-pid
    /// handle. A member the snapshot doesn't list exited between the job
    /// enumeration and the snapshot and is skipped (a vanished member, never a
    /// fabricated record). `Err` only if the job membership can't be read *or* the
    /// metadata snapshot can't be created (a total inability to read metadata,
    /// distinct from one pid vanishing).
    #[cfg(feature = "process-control")]
    pub(crate) fn members_info(&self) -> io::Result<Vec<MemberInfo>> {
        let pids = job_member_pids(self.handle)?;
        let meta = snapshot_process_metadata()?;
        let mut out = Vec::with_capacity(pids.len());
        for pid in pids {
            // A member absent from the snapshot exited between the job
            // enumeration and the snapshot — skip it (the documented race), never
            // fabricating ppid/exe for a pid we couldn't observe alive.
            let Some((ppid, exe)) = meta.get(&pid) else {
                continue;
            };
            // Start time via a per-pid handle. `None` if the process vanished in
            // this even-later window: the ppid/exe captured while it was alive
            // still stand, so the record is kept (only the finer start-time anchor
            // is missing), matching the honest-`Option` contract.
            let start_time = process_start_time(pid);
            out.push(MemberInfo::new(
                pid,
                Some(*ppid),
                Some(exe.clone()),
                start_time,
            ));
        }
        Ok(out)
    }

    /// Suspend or resume every thread of every process currently in the job.
    ///
    /// Best-effort, not atomic: the member list and the thread snapshot are
    /// taken once, so threads or processes created mid-walk are missed, and
    /// `SuspendThread`/`ResumeThread` maintain per-thread suspend *counts*
    /// (nested suspends need matching resumes). A thread that exits mid-walk is
    /// vacuously handled (not a failure); a genuine `SuspendThread`/
    /// `ResumeThread` failure on a still-open thread does not abort the walk and
    /// is reported after every member has been attempted.
    ///
    /// Recycle-safe (C13): the member list is captured before the thread
    /// snapshot, so a member could exit and its pid be reused by a foreign
    /// process in that gap. `suspend_or_resume_thread` re-verifies, per thread,
    /// that the live owner is *still a member of this job* (`IsProcessInJob`)
    /// before touching it, so a recycled pid can never divert a suspend/resume
    /// onto an unrelated process.
    #[cfg(feature = "process-control")]
    fn for_each_member_thread(&self, suspend: bool) -> io::Result<()> {
        // Mutually exclusive with `spawn`'s assign → resume window (see the
        // `suspend_lock` field doc); held across the pid query AND the walk so
        // the member set can't include a mid-spawn, still-suspended child.
        let _guard = self
            .suspend_lock
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let members: std::collections::HashSet<u32> =
            job_member_pids(self.handle)?.into_iter().collect();
        if members.is_empty() {
            // An empty job is trivially suspended/resumed.
            return Ok(());
        }

        // SAFETY: TH32CS_SNAPTHREAD always snapshots all threads system-wide;
        // returns INVALID_HANDLE_VALUE on failure.
        let snapshot = unsafe { CreateToolhelp32Snapshot(TH32CS_SNAPTHREAD, 0) };
        if snapshot == INVALID_HANDLE_VALUE {
            return Err(io::Error::last_os_error());
        }

        let mut entry: THREADENTRY32 = unsafe { std::mem::zeroed() };
        entry.dwSize = std::mem::size_of::<THREADENTRY32>() as u32;

        let mut last_err = None;
        // SAFETY: valid snapshot; `entry` is sized via its `dwSize` field.
        let mut ok = unsafe { Thread32First(snapshot, &mut entry) };
        while ok != 0 {
            if members.contains(&entry.th32OwnerProcessID)
                && let Err(err) = suspend_or_resume_thread(
                    entry.th32ThreadID,
                    entry.th32OwnerProcessID,
                    self.handle,
                    suspend,
                )
            {
                last_err = Some(err);
            }
            // SAFETY: same valid snapshot and entry.
            ok = unsafe { Thread32Next(snapshot, &mut entry) };
        }
        // SAFETY: handle came from CreateToolhelp32Snapshot; closed exactly once.
        unsafe { CloseHandle(snapshot) };

        match last_err {
            Some(err) => Err(err),
            None => Ok(()),
        }
    }

    pub(crate) async fn graceful_shutdown(
        &self,
        signal: i32,
        timeout: Duration,
        escalate: bool,
    ) -> io::Result<()> {
        // Opt-in console-CTRL path: a direct child was spawned into its own
        // process group (`windows_graceful_ctrl_break`), so there IS a way to
        // *trigger* a soft exit — `GenerateConsoleCtrlEvent(CTRL_BREAK_EVENT)`.
        // Drive the SAME shared escalation loop the unix backends use: signal the
        // leaders, poll the job's active-process count up to `timeout`, then
        // `TerminateJobObject` survivors (escalate) or spare them (!escalate). The
        // driver owns the `begin_shutdown`/`request` epoch handshake, so the
        // re-arm race is handled there, exactly as on unix.
        let leaders = self
            .ctrl_break_leaders
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .clone();
        if !leaders.is_empty() {
            let target = CtrlBreakTarget { job: self, leaders };
            return super::graceful::run(&target, &self.skip_drop_kill, signal, timeout, escalate)
                .await;
        }

        // Default path — no opt-in leaders. A Job Object has no graceful tier: no
        // Windows equivalent of SIGTERM, and the kill is atomic. When
        // `escalate=true`, kill the tree immediately. When `escalate=false`, skip
        // the kill and let survivors run; `Drop` will clear `KILL_ON_JOB_CLOSE`
        // before closing the handle so the tree is not implicitly killed then
        // either.
        //
        // The `timeout` is deliberately NOT used as a drain window (C6): without a
        // soft signal there is nothing to *trigger* a graceful exit, so polling for
        // a natural exit up to `timeout` would, for the common case of a child that
        // ignores the (absent) signal, only delay the inevitable kill by the whole
        // grace — a data-losing 30 s stall, not a graceful drain. Prompt hard-kill
        // at the deadline is the honest behavior; the grace/soft-signal tiers are
        // Unix-only (or the opt-in CTRL path above).
        //
        // Snapshot the re-arm generation up front — before the branch — so a
        // `spawn`/`adopt` that re-arms the backstop concurrently with this shutdown
        // wins over the (stale) `request` below. This body does not poll, but the
        // caller's task can migrate across its `.await` and a spawn/adopt on another
        // thread can still interleave between this snapshot and the request; keying
        // the spare to the epoch makes that concurrent re-arm win (the fresh child
        // keeps its kill-on-close backstop), matching the unix backends.
        let epoch = self.skip_drop_kill.begin_shutdown();
        if escalate {
            self.kill_all()
        } else {
            // Mark Drop to preserve survivors; the latch makes the flag visible
            // whichever thread drops the `Job` (it may differ from the one that
            // ran graceful shutdown, e.g. after a task migrates across `.await`).
            // Keyed to `epoch`, so a concurrent spawn/adopt re-arm wins and this
            // spare no-ops — the fresh child is still killed on job-close.
            self.skip_drop_kill.request(epoch);
            Ok(())
        }
    }

    #[cfg(feature = "stats")]
    pub(crate) fn stats(&self) -> io::Result<ProcessGroupStats> {
        let mut acct: JOBOBJECT_BASIC_ACCOUNTING_INFORMATION = unsafe { std::mem::zeroed() };
        // SAFETY: out param matches the accounting info class and its size.
        let ok = unsafe {
            QueryInformationJobObject(
                self.handle,
                JobObjectBasicAccountingInformation,
                std::ptr::from_mut(&mut acct).cast(),
                std::mem::size_of::<JOBOBJECT_BASIC_ACCOUNTING_INFORMATION>() as u32,
                std::ptr::null_mut(),
            )
        };
        if ok == 0 {
            return Err(io::Error::last_os_error());
        }

        let mut ext: JOBOBJECT_EXTENDED_LIMIT_INFORMATION = unsafe { std::mem::zeroed() };
        // SAFETY: out param matches the extended-limit info class and its size.
        let ok = unsafe {
            QueryInformationJobObject(
                self.handle,
                JobObjectExtendedLimitInformation,
                std::ptr::from_mut(&mut ext).cast(),
                std::mem::size_of::<JOBOBJECT_EXTENDED_LIMIT_INFORMATION>() as u32,
                std::ptr::null_mut(),
            )
        };
        if ok == 0 {
            return Err(io::Error::last_os_error());
        }

        // Job accounting times are in 100-ns units.
        let cpu_100ns = (acct.TotalUserTime as u64).saturating_add(acct.TotalKernelTime as u64);
        Ok(ProcessGroupStats {
            active_process_count: acct.ActiveProcesses as usize,
            total_cpu_time: Some(Duration::from_nanos(cpu_100ns.saturating_mul(100))),
            peak_memory_bytes: Some(ext.PeakJobMemoryUsed as u64),
        })
    }

    pub(crate) fn mechanism(&self) -> Mechanism {
        Mechanism::JobObject
    }
}

/// The Job-backed [`GracefulTarget`](crate::sys::graceful::GracefulTarget) for the
/// opt-in console-CTRL graceful teardown. It plugs the Job Object into the *same*
/// signal → poll → escalate loop the unix backends drive
/// ([`graceful::run`](crate::sys::graceful::run)):
///
/// - `signal_all` sends `GenerateConsoleCtrlEvent(CTRL_BREAK_EVENT, pid)` to each
///   recorded process-group leader (best-effort);
/// - `is_drained` reads the job's active-process count;
/// - `hard_kill` is `TerminateJobObject` (the escalation fallback).
///
/// It borrows the `Job` (already `Sync`) and a snapshot of the leader pids, so it
/// is automatically `Send + Sync` — no raw handle is carried across the driver's
/// `.await`s.
struct CtrlBreakTarget<'a> {
    job: &'a Job,
    /// Snapshot of the console process-group leader pids at shutdown time.
    leaders: Vec<u32>,
}

impl super::graceful::GracefulTarget for CtrlBreakTarget<'_> {
    fn signal_all(&self, _signal: i32) {
        // Windows delivers a console CTRL_BREAK, not a POSIX signal — the raw
        // `signal` number (SIGTERM/`timeout_signal`) is meaningless here and
        // ignored. CTRL_BREAK is chosen over CTRL_C deliberately: a process can
        // disable CTRL_C for its group (and CREATE_NEW_PROCESS_GROUP does so by
        // default), but CTRL_BREAK is always deliverable.
        for &pid in &self.leaders {
            // Never pass 0: `GenerateConsoleCtrlEvent(_, 0)` targets EVERY process
            // sharing this console — including us. A recorded leader is always a
            // real, nonzero pid, but guard defensively.
            if pid == 0 {
                continue;
            }
            // Recycle guard (mirrors the suspend/resume C13 discipline): only
            // signal a leader that is STILL a live member of THIS job. A leader
            // that already exited may have had its pid reused as an unrelated
            // process's group id sharing our console; `IsProcessInJob` fails safe
            // (a gone/denied pid reads as "not a member"), so the event is never
            // diverted onto a stranger's group.
            if !process_is_in_job(pid, self.job.handle) {
                continue;
            }
            // SAFETY: a console control event to a process group id; a delivery
            // failure (no shared console, the leader just exited) is swallowed —
            // the poll below observes the drain, and the `hard_kill` fallback
            // covers a child that never received it.
            unsafe {
                GenerateConsoleCtrlEvent(CTRL_BREAK_EVENT, pid);
            }
        }
    }

    fn is_drained(&self) -> bool {
        job_is_drained(self.job.handle)
    }

    fn hard_kill(&self) -> io::Result<()> {
        self.job.kill_all()
    }
}

/// Whether the job has fully drained — no process is still active in it.
///
/// Best-effort: a failed query (a torn-down handle, a transient error) reports
/// "not drained" so the driver keeps waiting and then takes its escalation
/// (`TerminateJobObject`) / spare decision at the deadline, never a premature
/// "drained" that would skip the fallback kill.
fn job_is_drained(handle: HANDLE) -> bool {
    let mut acct: JOBOBJECT_BASIC_ACCOUNTING_INFORMATION = unsafe { std::mem::zeroed() };
    // SAFETY: out param matches the accounting info class and its size.
    let ok = unsafe {
        QueryInformationJobObject(
            handle,
            JobObjectBasicAccountingInformation,
            std::ptr::from_mut(&mut acct).cast(),
            std::mem::size_of::<JOBOBJECT_BASIC_ACCOUNTING_INFORMATION>() as u32,
            std::ptr::null_mut(),
        )
    };
    ok != 0 && acct.ActiveProcesses == 0
}

/// Whether the process behind `handle` has already exited —
/// `GetExitCodeProcess` reports an exit code other than `STILL_ACTIVE` (259).
/// A *live* process always reports `STILL_ACTIVE`, so this never false-positives
/// a live child as exited. The only ambiguity is a child that genuinely exited
/// with code 259: it reads as "still active", so `adopt` surfaces the assign
/// error for it rather than the nothing-to-contain `Ok` — an acceptable rarity.
#[cfg(feature = "process-control")]
fn process_has_exited(handle: HANDLE) -> bool {
    const STILL_ACTIVE: u32 = 259;
    let mut code: u32 = 0;
    // SAFETY: `handle` is a valid process handle borrowed from the live `Child`.
    let ok = unsafe { GetExitCodeProcess(handle, &mut code) };
    ok != 0 && code != STILL_ACTIVE
}

/// Reaps a freshly-spawned, not-yet-contained child if [`Job::spawn`] unwinds
/// (an early `Err` or a panic) before the child is assigned to the job. Until
/// containment succeeds the child — created `CREATE_SUSPENDED` — is reachable by
/// nothing that would reap it, so dropping it un-disarmed would leak a suspended
/// orphan. [`disarm`](Self::disarm) hands the child back once it is contained,
/// after which the job's kill-on-close owns teardown.
struct UncontainedChildGuard {
    // `None` only after `disarm` has taken the child.
    child: Option<Child>,
}

impl UncontainedChildGuard {
    fn arm(child: Child) -> Self {
        Self { child: Some(child) }
    }

    /// Borrow the guarded child (present from `arm` until `disarm`). Used to read
    /// the child's `id()`/`raw_handle()` while the reaper is armed.
    fn child(&self) -> &Child {
        self.child
            .as_ref()
            .expect("the guarded child is present until disarm")
    }

    /// Containment succeeded: stop guarding and return the child unharmed.
    fn disarm(mut self) -> Child {
        self.child
            .take()
            .expect("the guarded child is taken exactly once")
    }
}

impl Drop for UncontainedChildGuard {
    fn drop(&mut self) {
        if let Some(mut child) = self.child.take() {
            // Best-effort: `start_kill` issues `TerminateProcess` on the
            // suspended child; dropping the `Child` then closes its handle. This
            // is the same kill the explicit error paths used to do inline, now
            // also covering an unwind.
            let _ = child.start_kill();
        }
    }
}

/// Resume every thread of `pid`. A child spawned `CREATE_SUSPENDED` has exactly
/// one thread (its primary); we walk a thread snapshot because std/tokio surface
/// only the process handle, not the `PROCESS_INFORMATION` thread handle returned
/// by `CreateProcess`.
fn resume_process_threads(pid: u32) -> io::Result<()> {
    // SAFETY: TH32CS_SNAPTHREAD always snapshots all threads system-wide (the
    // pid argument is ignored for the thread list); returns INVALID_HANDLE_VALUE
    // on failure.
    let snapshot = unsafe { CreateToolhelp32Snapshot(TH32CS_SNAPTHREAD, 0) };
    if snapshot == INVALID_HANDLE_VALUE {
        return Err(io::Error::last_os_error());
    }

    let mut entry: THREADENTRY32 = unsafe { std::mem::zeroed() };
    entry.dwSize = std::mem::size_of::<THREADENTRY32>() as u32;

    let mut resumed = 0u32;
    let mut last_err = None;
    // SAFETY: valid snapshot; `entry` is sized via its `dwSize` field.
    let mut ok = unsafe { Thread32First(snapshot, &mut entry) };
    while ok != 0 {
        if entry.th32OwnerProcessID == pid {
            match resume_thread(entry.th32ThreadID) {
                Ok(()) => resumed += 1,
                Err(err) => last_err = Some(err),
            }
        }
        // SAFETY: same valid snapshot and entry.
        ok = unsafe { Thread32Next(snapshot, &mut entry) };
    }
    // SAFETY: handle came from CreateToolhelp32Snapshot; closed exactly once.
    unsafe { CloseHandle(snapshot) };

    if resumed == 0 {
        return Err(last_err
            .unwrap_or_else(|| io::Error::other("no thread found to resume the contained child")));
    }
    Ok(())
}

/// Resume a single thread by id (decrement its suspend count).
fn resume_thread(tid: u32) -> io::Result<()> {
    // SAFETY: opens the thread by id; returns null on failure.
    let thread = unsafe { OpenThread(THREAD_SUSPEND_RESUME, 0, tid) };
    if thread.is_null() {
        return Err(io::Error::last_os_error());
    }
    // Resume until the suspend count reaches 0. A `CREATE_SUSPENDED` child's
    // primary thread is normally at count 1, but a member-walk suspend racing the
    // spawn (bounded by `suspend_lock`, yet possible) could nest it higher, and a
    // single decrement would leave it stuck suspended forever. `ResumeThread`
    // returns the PREVIOUS count and decrements by one each call, so loop until it
    // reports `<= 1` (now 0); bounded by the suspend depth. The failure is
    // captured BEFORE `CloseHandle`, which can overwrite the last-error.
    let err = loop {
        // SAFETY: valid thread handle; a `u32::MAX` return signals failure.
        let prev = unsafe { ResumeThread(thread) };
        if prev == u32::MAX {
            break Some(io::Error::last_os_error());
        }
        if prev <= 1 {
            break None; // prev == 1 → now 0 (running); prev == 0 → already running
        }
    };
    // SAFETY: handle came from OpenThread; closed exactly once.
    unsafe { CloseHandle(thread) };
    match err {
        Some(err) => Err(err),
        None => Ok(()),
    }
}

/// Suspend (increment) or resume (decrement) a single thread's suspend count.
///
/// `job` is the job whose members are being walked; it backs the pid-recycle
/// membership re-check (C13) below.
#[cfg(feature = "process-control")]
fn suspend_or_resume_thread(
    tid: u32,
    expected_pid: u32,
    job: HANDLE,
    suspend: bool,
) -> io::Result<()> {
    // Also request QUERY access so we can confirm the thread's owner below (C11).
    // SAFETY: opens the thread by id; returns null on failure (e.g. exited).
    let thread = unsafe {
        OpenThread(
            THREAD_SUSPEND_RESUME | THREAD_QUERY_LIMITED_INFORMATION,
            0,
            tid,
        )
    };
    if thread.is_null() {
        let err = io::Error::last_os_error();
        // A STALE tid — a thread that exited between the system-wide snapshot and
        // this open — fails `ERROR_INVALID_PARAMETER` and is *vacuously*
        // suspended/resumed, so swallow it: one churning thread must not fail the
        // whole job suspend/resume. ANY OTHER open failure (e.g.
        // `ERROR_ACCESS_DENIED` on a live but protected thread) is genuine and IS
        // reported — a live thread is never silently left suspended.
        if err.raw_os_error() == Some(ERROR_INVALID_PARAMETER as i32) {
            return Ok(());
        }
        return Err(err);
    }
    // C11: the system-wide thread snapshot named this tid as owned by a member
    // process, but a tid can be **recycled** between the snapshot and this open —
    // to a thread of an entirely different process. Verify the live owner before
    // touching it, so a suspend/resume never lands on a foreign process's thread.
    // SAFETY: valid thread handle from OpenThread; returns 0 on failure.
    let owner = unsafe { GetProcessIdOfThread(thread) };
    if owner != expected_pid {
        // Recycled (or unqueryable) — not our member's thread; leave it alone.
        // SAFETY: handle came from OpenThread; closed exactly once.
        unsafe { CloseHandle(thread) };
        return Ok(());
    }
    // C13: the owner check above only proves the thread belongs to `expected_pid`
    // *now* — but `expected_pid` itself may be a **recycled** pid. Between
    // `job_member_pids` (member snapshot) and the thread snapshot, a member
    // (typically a handle-less grandchild) can exit and its pid be reused by a
    // FOREIGN process X; X's threads then surface under a pid still in `members`,
    // and the C11 owner check passes because X genuinely owns them — so a bare
    // owner check would `SuspendThread` all of X's threads, freezing (and later
    // decrementing the suspend count of) an unrelated process. Close that window
    // by confirming the owner is STILL a member of THIS job before touching the
    // thread. Fail-safe: any failure to open the process or query membership is
    // treated as "not our member", so an uncertain result never suspends/resumes
    // a foreign thread.
    if !process_is_in_job(owner, job) {
        // SAFETY: handle came from OpenThread; closed exactly once.
        unsafe { CloseHandle(thread) };
        return Ok(());
    }
    // SAFETY: valid thread handle; both calls signal failure with `u32::MAX`.
    let prev = unsafe {
        if suspend {
            SuspendThread(thread)
        } else {
            ResumeThread(thread)
        }
    };
    // Capture the failure BEFORE `CloseHandle`, which can overwrite the
    // thread-local last-error.
    let err = (prev == u32::MAX).then(io::Error::last_os_error);
    // SAFETY: handle came from OpenThread; closed exactly once.
    unsafe { CloseHandle(thread) };
    match err {
        Some(err) => Err(err),
        None => Ok(()),
    }
}

/// Whether the process named by `pid` is currently a member of `job` — the
/// pid-recycle guard (C13) shared by the process-control suspend/resume walk and
/// the opt-in console-CTRL teardown's per-leader signal check.
///
/// Fail-safe by construction: a failure to open the process (gone, denied) or to
/// query membership yields `false`, i.e. "treat as NOT our member". A false
/// negative merely skips a suspend/resume for one thread, or a CTRL_BREAK for one
/// leader (both best-effort, with a `TerminateJobObject` backstop), whereas a
/// false positive would freeze a foreign process or divert a console event onto a
/// stranger's group — so uncertainty must resolve to "leave it alone".
fn process_is_in_job(pid: u32, job: HANDLE) -> bool {
    // Least-privilege: `IsProcessInJob` only needs query access.
    // SAFETY: opens the process by id; returns null on failure (e.g. exited).
    let handle = unsafe { OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid) };
    if handle.is_null() {
        return false;
    }
    let mut in_job: i32 = 0;
    // SAFETY: `handle` is a valid process handle from OpenProcess just above,
    // `job` is our live job handle, and `in_job` is an owned out-param. Returns 0
    // on failure, leaving `in_job` untouched (still 0 → treated as not-a-member).
    let ok = unsafe { IsProcessInJob(handle, job, &mut in_job) };
    // SAFETY: handle came from OpenProcess; closed exactly once.
    unsafe { CloseHandle(handle) };
    ok != 0 && in_job != 0
}

/// Enumerate the pids currently assigned to the job.
///
/// Best-effort snapshot: a process created or reaped during the query may be
/// briefly missing or present. The pid list is a variable-length struct (a
/// two-`u32` header followed by an inline `usize` array), so query into a
/// `u64`-backed buffer (alignment ≥ the struct's) and grow on `ERROR_MORE_DATA`.
#[cfg(feature = "process-control")]
fn job_member_pids(handle: HANDLE) -> io::Result<Vec<u32>> {
    // Seed generously so the common case is a single query.
    let mut cap: usize = 64;
    loop {
        let bytes = std::mem::size_of::<JOBOBJECT_BASIC_PROCESS_ID_LIST>()
            + cap.saturating_sub(1) * std::mem::size_of::<usize>();
        // u64 alignment (8) ≥ the struct's (usize) on every Windows target, so
        // casting the buffer to the struct pointer below is sound.
        let mut buf = vec![0u64; bytes.div_ceil(std::mem::size_of::<u64>())];
        // SAFETY: `buf` spans at least `bytes` writable bytes, the info class
        // matches the out-struct, and the size is passed explicitly.
        let ok = unsafe {
            QueryInformationJobObject(
                handle,
                JobObjectBasicProcessIdList,
                buf.as_mut_ptr().cast(),
                bytes as u32,
                std::ptr::null_mut(),
            )
        };
        let list = buf.as_ptr().cast::<JOBOBJECT_BASIC_PROCESS_ID_LIST>();
        if ok == 0 {
            let err = io::Error::last_os_error();
            if err.raw_os_error() == Some(ERROR_MORE_DATA as i32) {
                // The header is populated even when the list didn't fit — size
                // the retry from it (with headroom for races), and make sure we
                // always grow so the loop can't spin in place.
                // SAFETY: on ERROR_MORE_DATA the fixed header fields are valid.
                let assigned = unsafe { (*list).NumberOfAssignedProcesses } as usize;
                cap = assigned.max(cap).saturating_mul(2);
                continue;
            }
            return Err(err);
        }
        // SAFETY: a successful query wrote the header and `NumberOfProcessIdsInList`
        // pids contiguously from `ProcessIdList[0]`, all within `bytes`.
        let n = unsafe { (*list).NumberOfProcessIdsInList } as usize;
        // SAFETY: a successful query wrote `n` pids starting at `ProcessIdList[0]`.
        // `addr_of!` avoids creating a reference to the `[ULONG_PTR;1]` field
        // (which would have incorrect provenance for the out-of-bounds elements),
        // taking the raw address of its first element directly instead.
        // `ProcessIdList[0]` is always within the struct definition (the field
        // is declared as a 1-element array), so `addr_of!` is valid even when
        // `n == 0`; `from_raw_parts(ptr, 0)` is a zero-length slice, which is
        // always sound for any non-null aligned pointer.
        let ids =
            unsafe { std::slice::from_raw_parts(std::ptr::addr_of!((*list).ProcessIdList[0]), n) };
        return Ok(ids.iter().map(|&pid| pid as u32).collect());
    }
}

/// A system-wide `pid -> (ppid, image name)` map from one `Toolhelp32` process
/// snapshot — the source of [`members_info`](Job::members_info)'s parent-pid and
/// executable-name fields (neither is obtainable per-pid without ntdll, so one
/// snapshot is both cheaper and the only way to get the parent pid).
///
/// Best-effort, like the thread walk in [`for_each_member_thread`](Job::for_each_member_thread):
/// a process created or reaped during the walk may be briefly present or missing.
/// `Err` only when the snapshot itself can't be created — a genuine inability to
/// read any metadata, which the caller surfaces rather than reporting a populated
/// job as empty.
#[cfg(feature = "process-control")]
fn snapshot_process_metadata() -> io::Result<std::collections::HashMap<u32, (u32, String)>> {
    // SAFETY: TH32CS_SNAPPROCESS snapshots all processes system-wide; returns
    // INVALID_HANDLE_VALUE on failure.
    let snapshot = unsafe { CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0) };
    if snapshot == INVALID_HANDLE_VALUE {
        return Err(io::Error::last_os_error());
    }

    let mut entry: PROCESSENTRY32W = unsafe { std::mem::zeroed() };
    entry.dwSize = std::mem::size_of::<PROCESSENTRY32W>() as u32;

    let mut map = std::collections::HashMap::new();
    // SAFETY: valid snapshot; `entry` is sized via its `dwSize` field.
    let mut ok = unsafe { Process32FirstW(snapshot, &mut entry) };
    while ok != 0 {
        // `szExeFile` is a NUL-terminated UTF-16 array; decode up to the NUL.
        let len = entry
            .szExeFile
            .iter()
            .position(|&c| c == 0)
            .unwrap_or(entry.szExeFile.len());
        let exe = String::from_utf16_lossy(&entry.szExeFile[..len]);
        map.insert(entry.th32ProcessID, (entry.th32ParentProcessID, exe));
        // SAFETY: same valid snapshot and entry.
        ok = unsafe { Process32NextW(snapshot, &mut entry) };
    }
    // SAFETY: handle came from CreateToolhelp32Snapshot; closed exactly once.
    unsafe { CloseHandle(snapshot) };
    Ok(map)
}

/// The process-creation `FILETIME` of `pid` as its raw
/// [`MemberInfo`](crate::MemberInfo) start-time token (100-ns units since
/// 1601-01-01 UTC), or `None` if the process is gone / unqueryable. Fixed at spawn
/// and never reused within a boot, so it tells a recycled pid apart from the
/// original. (The `stats`-gated [`process_identity`] wraps the same read in a
/// `ProcIdentity`; this returns the bare token for the `process-control` snapshot,
/// which has no `stats` dependency.)
#[cfg(feature = "process-control")]
fn process_start_time(pid: u32) -> Option<u64> {
    // SAFETY: limited-information access; returns null on failure (e.g. gone).
    let handle = unsafe { OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid) };
    if handle.is_null() {
        return None;
    }
    let mut creation = FILETIME {
        dwLowDateTime: 0,
        dwHighDateTime: 0,
    };
    let mut exit = creation;
    let mut kernel = creation;
    let mut user = creation;
    // SAFETY: valid handle; all four out params are owned locals.
    let ok = unsafe { GetProcessTimes(handle, &mut creation, &mut exit, &mut kernel, &mut user) };
    // SAFETY: handle came from OpenProcess and is closed exactly once.
    unsafe { CloseHandle(handle) };
    (ok != 0).then(|| filetime_units(creation))
}

/// A FILETIME as its raw 64-bit 100-ns unit count (high/low halves combined).
/// The process-creation FILETIME serves as the [`ProcIdentity`] anchor (the
/// `stats` metrics gate) and as the [`MemberInfo`](crate::MemberInfo) start-time
/// token (the `process-control` snapshot), compared directly in these units;
/// [`filetime_nanos`] scales the CPU-time FILETIMEs to ns.
#[cfg(any(feature = "stats", feature = "process-control"))]
fn filetime_units(ft: FILETIME) -> u64 {
    ((ft.dwHighDateTime as u64) << 32) | ft.dwLowDateTime as u64
}

/// Combine a FILETIME (100-ns units) into nanoseconds.
#[cfg(feature = "stats")]
fn filetime_nanos(ft: FILETIME) -> u64 {
    filetime_units(ft).saturating_mul(100)
}

/// Convert a per-core CPU quota into a Job Object hard-cap `CpuRate`: 1/100 of a
/// percent of *total* system CPU, in `1..=10000`. `cores` is a fraction of one core
/// (`0.5` = half a core); `cpus` is the host processor count. A quota meeting or
/// exceeding the core count saturates at 100% (`10000`), and the result floors at
/// `1` since the API rejects a zero rate.
#[cfg(feature = "limits")]
fn cpu_hard_cap_rate(cores: f64, cpus: f64) -> u32 {
    let rate = ((cores / cpus) * 10_000.0).round();
    // `f64 as u32` is saturating, but clamp first so the floor-at-1 (zero is invalid)
    // and the 100% ceiling are explicit rather than relying on cast behaviour.
    rate.clamp(1.0, 10_000.0) as u32
}

/// The process-creation `FILETIME` of the process at `pid`, as its raw
/// [`ProcIdentity`] token, or `None` if the process is gone / unqueryable. The
/// creation time is fixed at spawn and never reused within a boot, so it tells a
/// recycled pid apart from the original process (the Windows analogue of Linux's
/// `/proc/<pid>/stat` starttime).
#[cfg(feature = "stats")]
pub(crate) fn process_identity(pid: u32) -> Option<ProcIdentity> {
    // SAFETY: limited-information access; returns null on failure (e.g. gone).
    let handle = unsafe { OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid) };
    if handle.is_null() {
        return None;
    }
    let mut creation = FILETIME {
        dwLowDateTime: 0,
        dwHighDateTime: 0,
    };
    let mut exit = creation;
    let mut kernel = creation;
    let mut user = creation;
    // SAFETY: valid handle; all four out params are owned locals.
    let ok = unsafe { GetProcessTimes(handle, &mut creation, &mut exit, &mut kernel, &mut user) };
    // SAFETY: handle came from OpenProcess and is closed exactly once.
    unsafe { CloseHandle(handle) };
    (ok != 0).then(|| ProcIdentity::from_raw(filetime_units(creation)))
}

#[cfg(feature = "stats")]
pub(crate) fn process_metrics(pid: u32, expected: Option<ProcIdentity>) -> ProcMetrics {
    let mut metrics = ProcMetrics::default();
    // SAFETY: limited-information access; returns null on failure (e.g. gone).
    let handle = unsafe { OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid) };
    if handle.is_null() {
        return metrics;
    }

    let mut creation = FILETIME {
        dwLowDateTime: 0,
        dwHighDateTime: 0,
    };
    let mut exit = creation;
    let mut kernel = creation;
    let mut user = creation;
    // SAFETY: valid handle; all four out params are owned locals.
    let ok = unsafe { GetProcessTimes(handle, &mut creation, &mut exit, &mut kernel, &mut user) };

    // Identity gate. `OpenProcess(pid)` resolves the number to *whatever process
    // holds it now* — possibly one that recycled it after our child was reaped —
    // and the handle then pins that process. Comparing the pinned process's
    // creation time (read via the same handle) against the captured identity proves
    // it is our process before we trust ANY reading from this handle, memory
    // included. If the times read failed we can't verify identity, so when one was
    // demanded that counts as a mismatch: return defaults and touch nothing else.
    if let Some(expected) = expected {
        let confirmed = ok != 0 && filetime_units(creation) == expected.raw();
        if !confirmed {
            // SAFETY: handle came from OpenProcess and is closed exactly once.
            unsafe { CloseHandle(handle) };
            return ProcMetrics::default();
        }
    }

    if ok != 0 {
        metrics.cpu_time = Some(Duration::from_nanos(
            filetime_nanos(kernel).saturating_add(filetime_nanos(user)),
        ));
    }

    let mut counters: PROCESS_MEMORY_COUNTERS = unsafe { std::mem::zeroed() };
    counters.cb = std::mem::size_of::<PROCESS_MEMORY_COUNTERS>() as u32;
    // SAFETY: valid handle; `counters` sized via its `cb` field. Reading through the
    // same identity-confirmed handle keeps the memory figure bound to our process,
    // never a recycled stranger's.
    let ok = unsafe { K32GetProcessMemoryInfo(handle, &mut counters, counters.cb) };
    if ok != 0 {
        metrics.peak_memory_bytes = Some(counters.PeakWorkingSetSize as u64);
    }

    // SAFETY: handle came from OpenProcess and is closed exactly once.
    unsafe { CloseHandle(handle) };
    metrics
}

impl Drop for Job {
    fn drop(&mut self) {
        if self.skip_drop_kill.is_set() {
            // Clear KILL_ON_JOB_CLOSE so closing the handle does not kill the tree.
            // `SetInformationJobObject` with `JobObjectExtendedLimitInformation`
            // *replaces* the entire extended-limit structure, so a zeroed struct
            // sets `LimitFlags = 0`, dropping `KILL_ON_JOB_CLOSE` and the
            // memory/active-process caps. Intentional — this path only runs under
            // `escalate=false`, so orphaning survivors uncapped is the desired
            // outcome and the caps are no longer meaningful.
            let info: JOBOBJECT_EXTENDED_LIMIT_INFORMATION = unsafe { std::mem::zeroed() };
            // Best-effort: if clearing fails the handle close will still kill —
            // a safe fallback (unexpected kill is better than orphaning ambiguity).
            let _ = unsafe {
                SetInformationJobObject(
                    self.handle,
                    JobObjectExtendedLimitInformation,
                    std::ptr::from_ref(&info).cast(),
                    std::mem::size_of::<JOBOBJECT_EXTENDED_LIMIT_INFORMATION>() as u32,
                )
            };
            // The CPU hard cap lives in a SEPARATE info class, so zeroing the
            // extended-limit struct above does NOT lift it. Clear it too (zeroed
            // `ControlFlags` = disabled) or orphaned survivors stay CPU-throttled
            // forever, inconsistent with the memory/process caps just dropped.
            // Harmless when no CPU cap was set.
            #[cfg(feature = "limits")]
            {
                let cpu: JOBOBJECT_CPU_RATE_CONTROL_INFORMATION = unsafe { std::mem::zeroed() };
                let _ = unsafe {
                    SetInformationJobObject(
                        self.handle,
                        JobObjectCpuRateControlInformation,
                        std::ptr::from_ref(&cpu).cast(),
                        std::mem::size_of::<JOBOBJECT_CPU_RATE_CONTROL_INFORMATION>() as u32,
                    )
                };
            }
        }
        // Closing the last handle triggers KILL_ON_JOB_CLOSE → the tree is reaped
        // (unless cleared above). SAFETY: handle came from CreateJobObjectW and is
        // closed exactly once.
        unsafe { CloseHandle(self.handle) };
    }
}

#[cfg(all(test, feature = "process-control"))]
mod thread_tests {
    use super::{process_is_in_job, suspend_or_resume_thread};
    use windows_sys::Win32::Foundation::CloseHandle;
    use windows_sys::Win32::System::JobObjects::CreateJobObjectW;
    use windows_sys::Win32::System::Threading::GetCurrentProcessId;

    /// A stale/invalid tid — a thread that exited between the system-wide
    /// snapshot and the `OpenThread` — fails `ERROR_INVALID_PARAMETER` and is
    /// *vacuously* suspended/resumed, not a failure (a single churning thread must
    /// not fail the whole job suspend). `tid = 1` is not a valid thread id (the
    /// kernel allocates thread/process ids as multiples of 4, and 0 is reserved),
    /// so `OpenThread` deterministically fails with `ERROR_INVALID_PARAMETER` and
    /// the fix returns `Ok` — and it can never open or suspend a real thread.
    #[test]
    fn suspend_or_resume_a_stale_tid_is_ok() {
        // `expected_pid`/`job` are irrelevant here: `tid = 1` fails `OpenThread`
        // before the C11 owner check or the C13 membership check ever runs, so a
        // null job handle is never dereferenced.
        let job = std::ptr::null_mut();
        assert!(suspend_or_resume_thread(1, 0, job, true).is_ok());
        assert!(suspend_or_resume_thread(1, 0, job, false).is_ok());
    }

    /// The C13 pid-recycle guard: a process that is NOT a member of *the* job in
    /// question reads as a non-member, so a suspend/resume is skipped. A freshly
    /// created job has no members, so the current process (never assigned to it)
    /// must fail the check — the exact outcome that spares a foreign process whose
    /// pid recycled into a stale member set. Also covers the fail-safe leg: a pid
    /// that cannot be opened yields `false` rather than a spurious "member".
    #[test]
    fn non_member_and_unopenable_pids_are_rejected() {
        // SAFETY: null name/attributes request an unnamed job with defaults;
        // returns null only on failure.
        let job = unsafe { CreateJobObjectW(std::ptr::null(), std::ptr::null()) };
        assert!(!job.is_null(), "failed to create a test job object");

        // The current process is openable but was never assigned to `job`, so
        // `IsProcessInJob` reports it is not a member of THIS job.
        // SAFETY: a plain read of our own pid.
        let me = unsafe { GetCurrentProcessId() };
        assert!(
            !process_is_in_job(me, job),
            "a process not assigned to this job must not read as a member"
        );

        // Fail-safe: `1` is not a valid pid (ids are multiples of 4), so
        // `OpenProcess` fails and the guard returns false — never a stray suspend.
        assert!(
            !process_is_in_job(1, job),
            "an unopenable pid must be treated as a non-member"
        );

        // SAFETY: handle came from CreateJobObjectW; closed exactly once.
        unsafe { CloseHandle(job) };
    }
}

// Un-gated (the guard is a core, non-feature-gated type) so a default
// `cargo test -- --ignored` exercises it, not only `--features limits`.
#[cfg(test)]
mod guard_tests {
    /// Whether `pid` is a live process. Self-contained FFI (independent of the
    /// crate's feature-gated helpers) so the test compiles in any config.
    fn pid_alive(pid: u32) -> bool {
        use windows_sys::Win32::Foundation::CloseHandle;
        use windows_sys::Win32::System::Threading::{
            GetExitCodeProcess, OpenProcess, PROCESS_QUERY_LIMITED_INFORMATION,
        };
        const STILL_ACTIVE: u32 = 259;
        // SAFETY: a plain query open; the handle is closed exactly once below.
        let handle = unsafe { OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid) };
        if handle.is_null() {
            return false; // gone (or no longer openable) — dead for our purposes
        }
        let mut code: u32 = 0;
        // SAFETY: `handle` is a valid process handle from OpenProcess.
        let ok = unsafe { GetExitCodeProcess(handle, &mut code) };
        // SAFETY: closing the handle obtained above, exactly once.
        unsafe { CloseHandle(handle) };
        ok != 0 && code == STILL_ACTIVE
    }

    /// A child created `CREATE_SUSPENDED` and never resumed — the exact state
    /// [`Job::spawn`](super::Job) guards: spawned but not yet assigned to the
    /// job. It runs no user code (a suspended process is still "alive" — its
    /// exit code reads `STILL_ACTIVE`); the guard's reap, or the test cleanup,
    /// terminates it via `TerminateProcess`.
    fn spawn_suspended() -> tokio::process::Child {
        use windows_sys::Win32::System::Threading::CREATE_SUSPENDED;
        tokio::process::Command::new("cmd")
            .args(["/C", "ping -n 30 127.0.0.1 > NUL"])
            .creation_flags(CREATE_SUSPENDED)
            .spawn()
            .expect("spawn the suspended child")
    }

    #[tokio::test]
    #[ignore = "spawns a real subprocess"]
    async fn uncontained_guard_reaps_the_child_on_an_armed_drop() {
        // An armed guard dropped without disarm must terminate the suspended,
        // not-yet-contained child (the spawn→assign unwind path).
        let child = spawn_suspended();
        let pid = child.id().expect("the child has a pid");
        assert!(
            pid_alive(pid),
            "the suspended child is alive right after spawn"
        );
        drop(super::UncontainedChildGuard::arm(child)); // armed → reaps on drop
        let mut dead = false;
        for _ in 0..200 {
            if !pid_alive(pid) {
                dead = true;
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        }
        assert!(dead, "an armed guard drop must terminate the child");
    }

    #[tokio::test]
    #[ignore = "spawns a real subprocess"]
    async fn uncontained_guard_disarm_hands_back_a_live_child() {
        // The success path: disarm returns the same child, still running, for the
        // job to own — the guard must not kill a contained child.
        let child = spawn_suspended();
        let pid = child.id().expect("the child has a pid");
        let mut kept = super::UncontainedChildGuard::arm(child).disarm();
        assert!(pid_alive(pid), "disarm must leave the child running");
        // Clean up the suspended child.
        let _ = kept.start_kill();
        let _ = kept.wait().await;
    }
}

#[cfg(all(test, feature = "limits"))]
mod tests {
    use super::cpu_hard_cap_rate;

    #[test]
    fn cpu_rate_maps_per_core_fraction_to_total_system_percent() {
        // Half a core out of eight = 6.25% of the whole machine.
        assert_eq!(cpu_hard_cap_rate(0.5, 8.0), 625);
        // A whole single core on a 1-CPU host = 100%.
        assert_eq!(cpu_hard_cap_rate(1.0, 1.0), 10_000);
        // Asking for every core = 100%.
        assert_eq!(cpu_hard_cap_rate(4.0, 4.0), 10_000);
        // Over-subscribing (more cores than exist) saturates at 100%, never above.
        assert_eq!(cpu_hard_cap_rate(8.0, 4.0), 10_000);
        // A vanishingly small quota floors at 1 — the API rejects a zero rate.
        assert_eq!(cpu_hard_cap_rate(0.0001, 64.0), 1);
    }
}

// Un-gated (`graceful_shutdown` and the latch are core, not feature-gated) so the
// default `cargo test` exercises the Windows re-arm race — no subprocess needed.
#[cfg(test)]
mod rearm_race_tests {
    use std::time::Duration;

    /// Build a bare `Job`, papering over the `limits`-feature gate on `Job::new`.
    fn new_job() -> super::Job {
        #[cfg(feature = "limits")]
        {
            super::Job::new(&crate::limits::ResourceLimits::default()).expect("create a test job")
        }
        #[cfg(not(feature = "limits"))]
        {
            super::Job::new().expect("create a test job")
        }
    }

    /// The documented reuse semantics, through the real `graceful_shutdown`
    /// (`escalate = false`) path: with nothing racing it, the shutdown spares
    /// survivors (Drop clears `KILL_ON_JOB_CLOSE`), and a subsequent spawn/adopt
    /// (which calls `clear()`) re-arms the kill-on-close backstop for the newcomer.
    #[tokio::test]
    async fn non_escalating_shutdown_spares_then_a_rearm_re_arms() {
        let job = new_job();
        job.graceful_shutdown(0, Duration::ZERO, false)
            .await
            .expect("graceful shutdown");
        assert!(
            job.skip_drop_kill.is_set(),
            "escalate=false spares survivors: Drop clears KILL_ON_JOB_CLOSE"
        );
        job.skip_drop_kill.clear();
        assert!(
            !job.skip_drop_kill.is_set(),
            "a member that joined after the spare re-arms Drop's kill-on-close"
        );
    }

    /// T-079 (Windows job re-arm race): a spawn/adopt that re-arms the backstop
    /// between a non-escalating shutdown's epoch snapshot and its final `request`
    /// must win — the stale request must not re-spare the fresh child. The Windows
    /// `graceful_shutdown(escalate = false)` body is exactly `begin_shutdown()` …
    /// `request(epoch)` with the concurrent spawn/adopt's `clear()` interleaving
    /// between; this reproduces that ordering deterministically on the real `Job`'s
    /// latch, no subprocess required.
    #[tokio::test]
    async fn a_concurrent_rearm_wins_over_a_stale_non_escalating_request() {
        let job = new_job();
        job.skip_drop_kill.clear(); // a live reused job — backstop already armed
        // The shutdown snapshots its generation…
        let epoch = job.skip_drop_kill.begin_shutdown();
        // …a concurrent spawn/adopt re-arms the backstop for a fresh child…
        job.skip_drop_kill.clear();
        // …and only now does the shutdown's stale request land.
        job.skip_drop_kill.request(epoch);
        assert!(
            !job.skip_drop_kill.is_set(),
            "a child assigned to the job mid-shutdown must keep its kill-on-close \
             backstop — the stale request must not re-spare it"
        );
    }
}

// T-139: the opt-in console-CTRL graceful path. Un-gated (the leader tracking,
// routing and drain check are core, not feature-gated) so the default
// `cargo test` exercises them — no subprocess needed, driven against an empty job
// and our own (non-member) pid as a stand-in leader.
#[cfg(test)]
mod ctrl_break_tests {
    use std::time::Duration;

    use windows_sys::Win32::System::Threading::GetCurrentProcessId;

    /// Build a bare `Job`, papering over the `limits`-feature gate on `Job::new`.
    fn new_job() -> super::Job {
        #[cfg(feature = "limits")]
        {
            super::Job::new(&crate::limits::ResourceLimits::default()).expect("create a test job")
        }
        #[cfg(not(feature = "limits"))]
        {
            super::Job::new().expect("create a test job")
        }
    }

    /// The drain check reports a fresh, empty job as drained (no active member),
    /// so the CTRL driver's poll ends immediately rather than riding the grace.
    #[test]
    fn an_empty_job_reports_drained() {
        let job = new_job();
        assert!(
            super::job_is_drained(job.handle),
            "an empty job has zero active processes, so it is drained"
        );
    }

    /// With a recorded leader, `graceful_shutdown` takes the CTRL_BREAK path
    /// (`graceful::run`), not the atomic branch. The recorded leader here is our
    /// OWN pid, which is NOT a member of this job — so the per-leader
    /// `IsProcessInJob` recycle guard must skip it, never delivering a CTRL_BREAK
    /// that would terminate the (handler-less) test runner. The job being empty,
    /// the driver's first drain check returns "drained" and it returns Ok promptly
    /// without a hard kill. The test process surviving to its assertions is the
    /// proof the membership gate held.
    #[tokio::test]
    async fn a_recorded_non_member_leader_is_never_signalled() {
        let job = new_job();
        // SAFETY: a plain read of our own pid.
        let me = unsafe { GetCurrentProcessId() };
        job.ctrl_break_leaders
            .lock()
            .expect("lock leaders")
            .push(me);

        let start = std::time::Instant::now();
        job.graceful_shutdown(crate::sys::SIGTERM_RAW, Duration::from_secs(30), true)
            .await
            .expect("ctrl-break graceful shutdown of an empty job");
        // The empty job is drained on the first poll, so the driver never waits
        // out the 30s grace; a bug that signalled/killed us would not get here.
        assert!(
            start.elapsed() < Duration::from_secs(5),
            "a drained job ends the CTRL grace immediately (took {:?})",
            start.elapsed()
        );
        assert!(
            job.ctrl_break_leaders.lock().expect("lock leaders").len() == 1,
            "the recorded leader is retained across a shutdown"
        );
    }

    /// The CTRL path honors `escalate = false` sparing exactly as the atomic path
    /// and the unix backends do: with a leader recorded but the job already
    /// drained, a non-escalating shutdown latches `skip_drop_kill` so `Drop`
    /// clears `KILL_ON_JOB_CLOSE`. Reuses the shared `graceful::run` epoch
    /// handshake, so the documented spawn/adopt re-arm race stays covered by the
    /// `SkipDropKill` unit tests.
    #[tokio::test]
    async fn the_ctrl_path_spares_survivors_when_not_escalating() {
        let job = new_job();
        // SAFETY: a plain read of our own pid — a non-member leader (skipped by
        // the recycle guard), so no CTRL_BREAK reaches the test runner.
        let me = unsafe { GetCurrentProcessId() };
        job.ctrl_break_leaders
            .lock()
            .expect("lock leaders")
            .push(me);

        job.graceful_shutdown(crate::sys::SIGTERM_RAW, Duration::ZERO, false)
            .await
            .expect("non-escalating ctrl-break shutdown");
        assert!(
            job.skip_drop_kill.is_set(),
            "a non-escalating CTRL shutdown spares survivors: Drop clears KILL_ON_JOB_CLOSE"
        );
        // A subsequent spawn/adopt re-arms it, as on every backend.
        job.skip_drop_kill.clear();
        assert!(
            !job.skip_drop_kill.is_set(),
            "a reused job re-arms the backstop"
        );
    }
}

// The per-process identity gate (T-090): a metrics read that names a pid whose
// current OS start identity does not match the captured one must yield defaults,
// never the stranger's counters — the fail-safe that stops a recycled pid from
// corrupting a `profile`/`cpu_time`/`peak_memory_bytes` sample. Driven against our
// OWN live process (`GetCurrentProcessId`), which is guaranteed present and has a
// stable creation time, so a wrong identity deterministically stands in for a
// reused pid — no second process or reuse simulation needed.
#[cfg(all(test, feature = "stats"))]
mod metrics_identity_tests {
    use super::{ProcIdentity, process_identity, process_metrics};
    use windows_sys::Win32::System::Threading::GetCurrentProcessId;

    #[test]
    fn identity_is_captured_and_matches_a_same_process_read() {
        // SAFETY: a plain read of our own pid.
        let me = unsafe { GetCurrentProcessId() };
        let id = process_identity(me).expect("our own live process has a creation time");

        // The captured identity matches on a re-read (a live process keeps its
        // creation time), so the gated read returns real counters.
        let gated = process_metrics(me, Some(id));
        assert!(
            gated.cpu_time.is_some(),
            "an identity-matched read of our own process reports CPU time"
        );
        assert!(
            gated.peak_memory_bytes.is_some(),
            "an identity-matched read of our own process reports peak memory"
        );
    }

    #[test]
    fn a_mismatched_identity_yields_defaults_not_the_live_process_counters() {
        // SAFETY: a plain read of our own pid.
        let me = unsafe { GetCurrentProcessId() };
        let real = process_identity(me).expect("our own live process has a creation time");

        // A wrong identity models a pid recycled by a different process: even though
        // the pid is very much alive (it is us), the gate must refuse to fold ANY of
        // this process's counters, returning the all-`None` default.
        let bogus = ProcIdentity::from_raw(real.raw().wrapping_add(1));
        let gated = process_metrics(me, Some(bogus));
        assert!(
            gated.cpu_time.is_none() && gated.peak_memory_bytes.is_none(),
            "a mismatched identity must yield defaults, never the live process's \
             CPU/memory — the recycled-pid fail-safe"
        );

        // Without a demanded identity the number-only behavior is preserved.
        let ungated = process_metrics(me, None);
        assert!(
            ungated.cpu_time.is_some(),
            "an unchecked read (identity None) still reports metrics"
        );
    }
}