capsule_runtime/

cgroup.rs

//! Linux cgroup v2 scoping for a capsule's whole native-subprocess tree.
//!
//! `resources::apply_hard_rlimits` bounds each spawned process individually. That is necessary
//! and not sufficient: `RLIMIT_NPROC` is a **per-uid** ceiling, checked against every thread the
//! uid owns on the host, so it cannot bound one tree, and no rlimit bounds *aggregate* memory or
//! CPU across a tree at all. A cgroup v2 scope does exactly what rlimits structurally cannot:
//! `pids.max`, `memory.max` and `cpu.max` apply to every task in the scope together, and the
//! kernel enforces them at fork/allocate time rather than after the fact. `memory.swap.max` is
//! written as [`SCOPE_SWAP_MAX`] beside `memory.max`, so the memory ceiling bounds the tree's
//! whole footprint rather than only its resident pages. A spawn that joins a
//! scope therefore gets no `RLIMIT_NPROC` at all, and `pids.max` is its only process bound — see
//! `resources::NprocBound`.
//!
//! ## Install requirement: systemd user cgroup delegation
//!
//! Creating a cgroup requires write access to a cgroup directory, and this runtime has none by
//! default. The usual alternative — start as root, create the cgroup, then drop privileges —
//! is not available here: nothing in this codebase ever runs as root or drops privileges (there
//! is no setuid or privilege-drop path anywhere in `capsule-runtime` or `murmur-cli`), so there
//! is no privileged phase to create a cgroup from. The mechanism that fits how this runtime
//! actually runs is **systemd user delegation**: the user unit or scope `mur` runs under carries
//! `Delegate=yes` for `memory pids cpu io`, which hands the unprivileged user write access to
//! its own cgroup subtree. This is the same mechanism rootless Docker and Podman use for the
//! identical problem. The exact unit configuration is in
//! `docs/content/reference/resource-limits-manual-verification.md`.
//!
//! ## Self-relocation before probing an inherited cgroup
//!
//! cgroup v2 refuses to enable controllers for a cgroup's children while that cgroup still holds
//! tasks, so a bounded child can only be built under a base this process is the *sole* occupant
//! of. An interactive shell's cgroup holds the shell, its pane scope, and everything else the
//! user is running; under `cargo test`, `cargo` stays resident beside every test binary. Stepping
//! aside into [`SUPERVISOR_LEAF`] moves only this process, so co-residency alone makes the base
//! unusable regardless of delegation.
//!
//! [`move_self_into_own_systemd_scope`] asks the systemd *user* manager over the session bus for
//! a freshly created transient scope carrying `Delegate=yes`, moving this process into a
//! brand-new cgroup — a sibling of whatever it left, not a child — that it is the sole occupant
//! of by construction. The step-aside logic below then always succeeds, since nothing else is
//! ever in that scope to contend with.
//!
//! The request is best-effort and silent: with no systemd, no session bus, or a policy denial,
//! the runtime probes whatever cgroup it inherited, exactly as it does when this step is skipped.
//!
//! ## Fail closed, but only where the threat exists
//!
//! On Linux, a capsule that can spawn *any* native subprocess and cannot be given a cgroup
//! refuses to launch (`RuntimeError::CgroupDelegationUnavailable`) rather than running that
//! subprocess tree unbounded. A capsule that declares no subprocess capability at all needs no
//! scope and is never blocked — there is no process tree to bound. On macOS, cgroups cannot
//! exist at all, so this module is inert and the runtime falls back to rlimits-only enforcement
//! with a `W-SEC-010` warning; refusing every launch on the primary development platform for a
//! kernel feature that platform will never have would be a policy, not a protection.
//!
//! ## A scope outlives no runtime
//!
//! Each scope [`CgroupScope::create`] makes on Linux has a watcher: a process forked from the
//! runtime that blocks reading a pipe whose write end only the runtime holds. When the read
//! returns end-of-file — the scope's `Drop` closed it, or the kernel closed it because the runtime
//! died by any means, `SIGKILL` and the OOM killer included — the watcher kills the scope and
//! exits. A parent-death signal would reach only a direct child, and a backgrounded server is a
//! grandchild; the pipe reaches every death at once. The watcher stays in the runtime's own
//! cgroup, never in the scope it kills, and holds none of the runtime's other descriptors, so it
//! keeps no pipe from reaching end-of-file and no port bound.
//!
//! ## Cross-platform shape
//!
//! Following `sandbox.rs`: the types here are defined unconditionally and stay type-checked and
//! unit-testable on macOS, with the *behavior* gated internally by `cfg(target_os = "linux")`
//! and cross-platform-dead surface marked `#[cfg_attr(not(target_os = "linux"), allow(dead_code))]`.
//! Nothing is hidden behind a module-level `cfg`.

use std::path::{Path, PathBuf};
use std::sync::{
    atomic::{AtomicU64, Ordering},
    Arc,
};
#[cfg(target_os = "linux")]
use std::time::Duration;

use crate::resources::HostResourceLimits;

/// Where the unified cgroup v2 hierarchy is mounted on a conventional host. Only used as the
/// fallback when `/proc/mounts` does not name a `cgroup2` mount explicitly.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
const DEFAULT_CGROUP2_MOUNT: &str = "/sys/fs/cgroup";

/// The `cpu.max` period, in microseconds. The kernel's own default; the quota this module
/// writes is expressed against it (`cgroup_cpu_percent * 1000` µs of every 100 000 µs).
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
const CPU_MAX_PERIOD_US: u64 = 100_000;

/// `memory.swap.max` for every scope: no swap at all.
///
/// `memory.max` alone bounds only resident pages; with `memory.swap.max` at the kernel default
/// `max`, a tree over `cgroup_memory_bytes` is pushed out to swap instead of being OOM-killed and
/// keeps growing there. Zero makes `cgroup_memory_bytes` the tree's whole memory footprint, which
/// is what the manifest field promises.
pub const SCOPE_SWAP_MAX: &str = "0";

/// One cgroup interface file a scope's limits are written to, and the value written.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct ScopeLimit {
    pub(crate) file: &'static str,
    pub(crate) value: String,
    /// Whether a kernel that exposes no such file still yields a bounded scope. Only
    /// `memory.swap.max` is: it is absent when swap support is compiled out or swap accounting
    /// is off at boot, and on such a kernel there is no per-cgroup swap control to write. A file
    /// that exists and refuses the write is fatal either way.
    pub(crate) absent_ok: bool,
}

/// The limit writes a scope is created with, in write order. `io.max` is not among them: it is
/// best-effort and reported separately (see [`CgroupScope::apply_io_max`]).
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
pub(crate) fn scope_limits(limits: &HostResourceLimits) -> Vec<ScopeLimit> {
    let required = |file, value| ScopeLimit {
        file,
        value,
        absent_ok: false,
    };
    vec![
        required("memory.max", limits.cgroup_memory_bytes.to_string()),
        ScopeLimit {
            file: "memory.swap.max",
            value: SCOPE_SWAP_MAX.to_string(),
            absent_ok: true,
        },
        required("pids.max", limits.cgroup_pids_max.to_string()),
        required(
            "cpu.max",
            format!(
                "{} {CPU_MAX_PERIOD_US}",
                u64::from(limits.cgroup_cpu_percent) * (CPU_MAX_PERIOD_US / 100)
            ),
        ),
    ]
}

/// Whether this kernel gives a cgroup a `memory.swap.max` to write, as `mur doctor` reports it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SwapControl {
    /// A cgroup on this host carries `memory.swap.max`, so a scope's swap is bounded at
    /// [`SCOPE_SWAP_MAX`].
    Exposed,
    /// A cgroup on this host carries `memory.max` and no `memory.swap.max`: the kernel has no
    /// per-cgroup swap control, and a scope's swap is unbounded.
    Absent,
    /// No cgroup this process can see has the memory controller enabled, so the question cannot
    /// be answered from here. Also the answer off Linux.
    Unknown,
}

/// Probe whether this kernel exposes `memory.swap.max`, without creating or moving anything.
///
/// Walks from this process's own cgroup up to the hierarchy root and answers from the first
/// cgroup that has the memory controller (a `memory.max` file): the file set a cgroup gets is the
/// kernel's, so one cgroup with `memory.max` and no `memory.swap.max` speaks for every scope this
/// host would create.
#[must_use]
pub fn probe_swap_control() -> SwapControl {
    #[cfg(not(target_os = "linux"))]
    {
        SwapControl::Unknown
    }

    #[cfg(target_os = "linux")]
    {
        let mount = std::fs::read_to_string("/proc/mounts")
            .map(|contents| parse_cgroup2_mount(&contents))
            .unwrap_or_else(|_| DEFAULT_CGROUP2_MOUNT.to_string());
        let Some(own) = std::fs::read_to_string("/proc/self/cgroup")
            .ok()
            .and_then(|contents| parse_unified_cgroup_path(&contents))
        else {
            return SwapControl::Unknown;
        };
        let root = PathBuf::from(mount);
        let mut dir = root.join(own.trim_start_matches('/'));
        loop {
            if dir.join("memory.max").exists() {
                return if dir.join("memory.swap.max").exists() {
                    SwapControl::Exposed
                } else {
                    SwapControl::Absent
                };
            }
            if dir == root || !dir.pop() || !dir.starts_with(&root) {
                return SwapControl::Unknown;
            }
        }
    }
}

/// How many swap areas `/proc/swaps` lists as active, or `None` where it cannot be read.
#[must_use]
pub fn active_swap_areas() -> Option<usize> {
    let contents = std::fs::read_to_string("/proc/swaps").ok()?;
    Some(count_swap_areas(&contents))
}

/// Count the device lines in `/proc/swaps` contents, below its one header line.
fn count_swap_areas(contents: &str) -> usize {
    contents
        .lines()
        .skip(1)
        .filter(|line| !line.trim().is_empty())
        .count()
}

/// Controllers that must be delegated and settable. `io` is deliberately absent: it is
/// best-effort (see [`CgroupScope::apply_io_max`]).
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
const REQUIRED_CONTROLLERS: &[&str] = &["memory", "pids", "cpu"];

/// Name of the leaf cgroup the runtime moves *itself* into when its delegated base still holds
/// processes. cgroup v2's "no internal processes" rule forbids enabling controllers in a
/// cgroup's `subtree_control` while that cgroup directly contains tasks, so the manager has to
/// step down into its own leaf first — the same move rootless Podman/Docker make.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
const SUPERVISOR_LEAF: &str = "murmur-supervisor";

/// A cgroup v2 scope created for one capsule session, holding that session's whole native
/// subprocess tree.
///
/// Constructed once per launch (never per subprocess) so that `pids.max` and `memory.max` bound
/// the tree *in aggregate* — a per-call scope would let N sequential shell calls each spend the
/// full budget, which is precisely the aggregate case rlimits already fail to cover.
#[derive(Debug)]
pub(crate) struct CgroupScope {
    /// Absolute path of the scope directory, e.g.
    /// `/sys/fs/cgroup/user.slice/user-1000.slice/user@1000.service/app.slice/mur.scope/murmur-ses_abc123`.
    path: PathBuf,
    /// `cgroup.procs`, opened write-only in the **parent** before any fork. The forked child's
    /// `pre_exec` closure only ever writes its own pid to this already-open descriptor — it
    /// never performs a path lookup, matching how `sandbox.rs` pre-opens its Landlock fds rather
    /// than opening paths inside `pre_exec`.
    #[cfg_attr(not(target_os = "linux"), allow(dead_code))]
    procs_fd: Option<std::os::fd::OwnedFd>,
    /// Who is inside the scope right now, which decides whether a counter delta can be pinned on
    /// one call. Behind its own `Arc` so a [`ScopeOccupancy`] can outlive the borrow it was
    /// taken from and move to a demoted command's thread.
    occupancy: Arc<Occupancy>,
    /// The process that kills the scope once the runtime is gone. `None` off Linux and for a
    /// scope with no directory.
    #[cfg(target_os = "linux")]
    watcher: Option<ScopeWatcher>,
}

/// The scope's occupancy counters.
///
/// The scope's kill/denial counters are shared by every process in it, so a delta read across one
/// call's window is evidence about that call only when nothing else ran in the scope during the
/// window. These two counters are what answer that.
#[derive(Debug, Default)]
struct Occupancy {
    /// Shell commands and native tools inside the scope now.
    occupants: AtomicU64,
    /// Entries since the scope was created. Never decremented: a later entry is visible as a
    /// higher number even after it has left.
    entries: AtomicU64,
}

impl Occupancy {
    /// Mark one spawn as inside the scope until the returned guard drops.
    fn enter(self: &Arc<Self>) -> ScopeOccupancy {
        // Every access is `SeqCst`, and `entries` is read on both sides of the `occupants`
        // increment. An entry that raced in between the two increments — counted in one, not yet
        // in the other — moves `entries` past what was read first, so neither call can think it
        // was alone while the other was already on its way in.
        let entries_before = self.entries.load(Ordering::SeqCst);
        let others_inside = self.occupants.fetch_add(1, Ordering::SeqCst) > 0;
        let sequence = self.entries.fetch_add(1, Ordering::SeqCst);
        ScopeOccupancy {
            occupancy: Arc::clone(self),
            alone_at_entry: !others_inside && sequence == entries_before,
            sequence,
        }
    }
}

/// One spawn's stay inside a [`CgroupScope`]. Leaves the scope on drop.
///
/// Held for as long as the process it was taken for runs — a demoted command carries it to its
/// own thread — so a sibling's window overlaps it exactly when the two processes did.
#[derive(Debug)]
pub(crate) struct ScopeOccupancy {
    occupancy: Arc<Occupancy>,
    alone_at_entry: bool,
    sequence: u64,
}

impl ScopeOccupancy {
    /// Whether this spawn had the scope to itself from entry until now: nobody was inside when it
    /// entered, and nobody has entered since.
    pub(crate) fn exclusive(&self) -> bool {
        self.alone_at_entry && self.occupancy.entries.load(Ordering::SeqCst) == self.sequence + 1
    }
}

impl Drop for ScopeOccupancy {
    fn drop(&mut self) {
        self.occupancy.occupants.fetch_sub(1, Ordering::SeqCst);
    }
}

/// Snapshot of the kernel-maintained kill/denial counters on a scope.
///
/// Read before and after a subprocess runs, so attribution rests on the *delta* for that call
/// rather than on a session-cumulative total that any earlier call could have moved.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub(crate) struct CgroupEventCounters {
    /// `memory.events`' `oom_kill` — tasks the kernel killed for exceeding `memory.max`.
    pub(crate) oom_kill: u64,
    /// `pids.events`' `max` — forks refused for exceeding `pids.max`.
    pub(crate) pids_max: u64,
}

impl CgroupEventCounters {
    /// The named limit this delta unambiguously attributes to, or `None`.
    ///
    /// Both counters are maintained by the kernel and incremented only by the condition they
    /// name, so a nonzero delta is evidence rather than inference — the property that makes
    /// these two safe to report where a bare `SIGKILL` is not (see
    /// [`crate::resources::limit_from_signal`]).
    ///
    /// The evidence is about the scope, not about any one process in it. A delta read across one
    /// call's window is about that call only when the call was alone in the scope for the whole
    /// window ([`ScopeOccupancy::exclusive`]); `shell::attribute_resource_limit` is where that
    /// rule is applied.
    pub(crate) fn attribution_since(&self, before: Self) -> Option<&'static str> {
        if self.oom_kill > before.oom_kill {
            return Some("cgroup_memory_bytes");
        }
        if self.pids_max > before.pids_max {
            return Some("cgroup_pids_max");
        }
        None
    }
}

/// Whether the declared `capabilities.resources.cgroup_io_bytes_per_sec` ceiling is actually on
/// this session's cgroup scope.
///
/// Four states rather than a `bool` because "the write failed" and "there was never a scope to
/// write to" are different findings with different remediations, and neither is the same as "this
/// value was never established". [`Self::NotProbed`] is the `Default` so a report nobody filled in
/// claims nothing.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, serde::Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum IoMaxStatus {
    /// The `io.max` write succeeded against the device backing the workdir. The ceiling is on the
    /// scope.
    Enforced,
    /// A scope exists and the `io.max` write did not succeed. `memory.max`, `pids.max` and
    /// `cpu.max` are still enforced on it; I/O bandwidth is not bounded. What `W-SEC-021` reports.
    Unavailable,
    /// No scope was asked for: either this capsule declares no way to reach a native subprocess,
    /// or this is not Linux, where cgroups cannot exist. Not a shortfall — there is no process
    /// tree for a ceiling to apply to.
    NotRequired,
    /// Nobody asked. The `Default`, so a [`IoMaxReport`] that was constructed rather than measured
    /// asserts nothing about the host.
    #[default]
    NotProbed,
}

/// The declared I/O ceiling and what became of it, carried into `--explain-scope`, into
/// `session_start.effective_grants` and into `W-SEC-021`.
///
/// `io.max` is non-fatal, so a ceiling that did not apply refuses nothing. This value is what
/// carries that outcome alongside the claim, so the manifest's declared ceiling and what the
/// kernel accepted cannot disagree in silence.
#[derive(Debug, Clone, PartialEq, Eq, Default, serde::Serialize)]
pub struct IoMaxReport {
    /// The effective `capabilities.resources.cgroup_io_bytes_per_sec`, after
    /// [`HostResourceLimits::resolve`] has applied its default. Reported whatever the status, so a
    /// reader can see what was asked for as well as what happened to it.
    pub declared_bytes_per_sec: u64,
    /// What happened to it.
    pub status: IoMaxStatus,
    /// Why, for every status but [`IoMaxStatus::Enforced`]. Absent from the JSON when there is
    /// nothing to say.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub reason: Option<String>,
}

impl IoMaxReport {
    /// The ceiling applied.
    #[cfg_attr(not(target_os = "linux"), allow(dead_code))]
    fn enforced(declared_bytes_per_sec: u64) -> Self {
        Self {
            declared_bytes_per_sec,
            status: IoMaxStatus::Enforced,
            reason: None,
        }
    }

    /// A scope exists and the ceiling is not on it.
    #[cfg_attr(not(target_os = "linux"), allow(dead_code))]
    fn unavailable(declared_bytes_per_sec: u64, reason: impl Into<String>) -> Self {
        Self {
            declared_bytes_per_sec,
            status: IoMaxStatus::Unavailable,
            reason: Some(reason.into()),
        }
    }

    /// No cgroup scope is required because this capsule can reach no native subprocess.
    ///
    /// Public, and takes no reason argument: `mur run --explain-scope` has to produce the
    /// byte-for-byte report a launch produces — `trace.rs`'s `session_start_effective_grants_match_explain_scope_json`
    /// asserts the two objects are equal — so the wording lives here rather than once per caller.
    #[must_use]
    pub fn no_scope_required(declared_bytes_per_sec: u64) -> Self {
        Self::not_required(declared_bytes_per_sec, NO_SCOPE_REASON)
    }

    /// No scope was asked for. Always carries its reason: "not required" is the one status whose
    /// two causes — a capsule that can spawn nothing, and a platform with no cgroups — a reader
    /// cannot tell apart from the status alone.
    fn not_required(declared_bytes_per_sec: u64, reason: impl Into<String>) -> Self {
        Self {
            declared_bytes_per_sec,
            status: IoMaxStatus::NotRequired,
            reason: Some(reason.into()),
        }
    }
}

/// Reason text for the two [`IoMaxStatus::NotRequired`] causes, in one place so `prepare_scope`
/// and [`probe_io_max`] cannot word the same finding differently.
const NO_SCOPE_REASON: &str = "this capsule declares no way to run a native subprocess, so it is \
     given no cgroup scope and there is no process tree for an I/O ceiling to apply to";
#[cfg_attr(target_os = "linux", allow(dead_code))]
const NOT_LINUX_REASON: &str =
    "cgroups are a Linux mechanism and this host has none, so no io.max ceiling can exist here";

/// The message `W-SEC-021` carries, minus the reason the host gave.
const IO_MAX_UNAVAILABLE_WARNING: &str = "the declared \
capabilities.resources.cgroup_io_bytes_per_sec ceiling did not apply to this session's cgroup \
scope, so this capsule's native subprocess tree has no I/O bandwidth bound; memory.max, pids.max \
and cpu.max are still enforced on the scope";

/// Pure decision for the unenforced-`io.max` warning, split out of
/// [`warn_for_unenforced_io_max`] the same way `sandbox::aggregate_bounding_warning` is split out
/// of its emitter, so a test can assert it without capturing stderr.
///
/// Fires on [`IoMaxStatus::Unavailable`] alone. `NotRequired` is not a shortfall and `NotProbed`
/// is not a finding, so neither warns.
pub(crate) fn io_max_warning(report: &IoMaxReport) -> Option<(&'static str, String)> {
    if report.status != IoMaxStatus::Unavailable {
        return None;
    }
    let declared = report.declared_bytes_per_sec;
    let reason = report.reason.as_deref().unwrap_or("no reason reported");
    Some((
        murmur_artifact::W_SEC_021,
        format!("{IO_MAX_UNAVAILABLE_WARNING} (declared {declared} bytes/s): {reason}"),
    ))
}

/// Fires at every launch, not just once.
pub(crate) fn warn_for_unenforced_io_max(
    workdir: &crate::workdir_fs::WorkdirRoot,
    report: &IoMaxReport,
) {
    if let Some((code, message)) = io_max_warning(report) {
        let link = murmur_artifact::security_warning_link(code);
        crate::runtime_err!("[capsule-runtime] warning[{code}]: {message} ({link})");
        crate::agent::append_bootstrap_log(
            workdir,
            &format!("[capability-policy] warning[{code}]: {message} ({link})"),
        );
    }
}

/// What [`prepare_scope`] establishes: the scope, and what became of its I/O ceiling.
///
/// A struct rather than a tuple so the two values cannot be swapped at a call site, and so a
/// caller cannot take the scope without also being handed the ceiling's outcome.
#[derive(Debug)]
pub(crate) struct PreparedScope {
    /// The scope, or `None` where none was required or none can exist.
    pub(crate) scope: Option<Arc<CgroupScope>>,
    /// Whether the declared `io.max` ceiling applied, established by performing the write rather
    /// than by inferring it.
    pub(crate) io_max: IoMaxReport,
}

/// Decide-and-create entry point, called once per launch before any WASM is instantiated.
///
/// * `required == false` (the capsule declares no way to spawn a native subprocess) → `Ok(None)`
///   on every platform. There is no process tree to bound, so demanding host configuration here
///   would be a regression for WASM-only capsules with nothing to show for it.
/// * non-Linux → `Ok(None)` always. Never an error: cgroups are structurally impossible there,
///   not a host misconfiguration.
/// * Linux and required → probe delegation and create the scope; `Err(reason)` if the host
///   cannot delegate one, which the caller turns into
///   [`crate::errors::RuntimeError::CgroupDelegationUnavailable`].
///
/// The returned [`PreparedScope`] always carries an [`IoMaxReport`], on every one of those paths:
/// the declared I/O ceiling is reported as applying, as not applying, or as not required, and
/// never left silent.
pub(crate) fn prepare_scope(
    required: bool,
    limits: &HostResourceLimits,
    session_id: &str,
    workdir: &Path,
) -> Result<PreparedScope, String> {
    if !required {
        return Ok(PreparedScope {
            scope: None,
            io_max: IoMaxReport::no_scope_required(limits.cgroup_io_bytes_per_sec),
        });
    }
    #[cfg(target_os = "linux")]
    {
        #[cfg(test)]
        if FORCE_DELEGATION_FAILURE.with(std::cell::Cell::get) {
            return Err("forced: delegation refused by the test seam".to_string());
        }
        let (scope, io_max) = CgroupScope::create(limits, session_id, workdir)?;
        Ok(PreparedScope {
            scope: Some(Arc::new(scope)),
            io_max,
        })
    }
    #[cfg(not(target_os = "linux"))]
    {
        let _ = (session_id, workdir);
        Ok(PreparedScope {
            scope: None,
            io_max: IoMaxReport::not_required(limits.cgroup_io_bytes_per_sec, NOT_LINUX_REASON),
        })
    }
}

/// Answers "would the declared `io.max` ceiling apply on this host?" by performing the write,
/// against a throwaway cgroup created and removed inside this call.
///
/// The diagnostic half of [`prepare_scope`], for `mur run --explain-scope`, which runs before any
/// workdir exists and before any session is staged. Inferring the answer from "is the `io`
/// controller delegated" is wrong: the write can fail for a device the block layer will not
/// accept (tmpfs, overlayfs, a device-mapper stack) on a host whose `io` controller is delegated
/// perfectly well. Only performing the write answers it.
///
/// `workdir` need not exist. The device is resolved from its nearest existing ancestor, so the
/// probe names the device a launch would use without creating anything.
///
/// Off Linux it reports [`IoMaxStatus::NotRequired`]. Whatever the outcome, the throwaway
/// directory does not survive the call.
#[must_use]
pub fn probe_io_max(limits: &HostResourceLimits, workdir: &Path) -> IoMaxReport {
    let declared = limits.cgroup_io_bytes_per_sec;
    #[cfg(target_os = "linux")]
    {
        let base = match delegated_base() {
            Ok(base) => base,
            Err(reason) => {
                return IoMaxReport::unavailable(
                    declared,
                    format!("no delegated cgroup base to write io.max into: {reason}"),
                )
            }
        };
        let path = base.join(scope_dir_name("explain-scope"));
        if let Err(error) = std::fs::create_dir(&path) {
            return IoMaxReport::unavailable(
                declared,
                format!(
                    "could not create a throwaway cgroup at {}: {error}",
                    path.display()
                ),
            );
        }
        let report = write_io_max(&path, declared, workdir);
        // Best-effort by necessity and not by preference: the scope holds no tasks, so the only
        // way this fails is a host that took the directory away underneath us.
        let _ = std::fs::remove_dir(&path);
        report
    }
    #[cfg(not(target_os = "linux"))]
    {
        let _ = workdir;
        IoMaxReport::not_required(declared, NOT_LINUX_REASON)
    }
}

/// Does this policy give the capsule any way to run a native subprocess at all?
///
/// The exact condition the Linux fail-closed launch refusal keys on: every route
/// [`crate::runtime::has_subprocess_route_at_staging`] counts (`capabilities.shell.allow`,
/// `capabilities.spawn.allow`, and `capabilities.install.tool`, whose pattern can match a native
/// tool a pull installs), or a declared artifact whose implementation is native (passed in by the
/// caller, which is the only place the installed-artifact list is in hand). The same routes the
/// `E-CAP-005` egress refusal counts, from the same function.
pub fn requires_process_bounding(
    policy: &crate::types::CapabilityPolicy,
    has_native_artifact: bool,
) -> bool {
    crate::runtime::has_subprocess_route_at_staging(policy) || has_native_artifact
}

impl CgroupScope {
    /// Absolute path of this scope's directory.
    pub(crate) fn path(&self) -> &Path {
        &self.path
    }

    /// Move the **calling** process into this scope.
    ///
    /// Called from inside the forked child's `pre_exec` closure: it writes the child's own pid
    /// to the parent-opened `cgroup.procs` descriptor with a single `write(2)`, formatting the
    /// pid into a stack buffer so the whole path stays allocation-free and async-signal-safe.
    /// Every process the child subsequently forks inherits the scope, which is what makes the
    /// bound apply to the *tree* rather than to one process.
    #[cfg_attr(not(target_os = "linux"), allow(dead_code))]
    #[allow(unsafe_code)]
    pub(crate) fn join_current_process(&self) -> std::io::Result<()> {
        use std::os::fd::AsRawFd;

        let Some(fd) = self.procs_fd.as_ref() else {
            return Ok(());
        };

        // SAFETY: `getpid` takes no arguments and cannot fail.
        let pid = unsafe { libc::getpid() };
        let mut buf = [0u8; 24];
        let written = format_pid(pid, &mut buf);

        let mut offset = 0;
        while offset < written {
            // SAFETY: `fd` is an open write-only descriptor owned by `self`, and the slice
            // passed is a live sub-range of the stack buffer above.
            let rc = unsafe {
                libc::write(
                    fd.as_raw_fd(),
                    buf[offset..written].as_ptr().cast::<libc::c_void>(),
                    written - offset,
                )
            };
            if rc < 0 {
                let error = std::io::Error::last_os_error();
                if error.kind() == std::io::ErrorKind::Interrupted {
                    continue;
                }
                return Err(error);
            }
            offset += rc as usize;
        }
        Ok(())
    }

    /// Mark one spawn as inside this scope until the returned guard drops. Taken before the
    /// counter snapshot a call's attribution is read against.
    pub(crate) fn enter(&self) -> ScopeOccupancy {
        self.occupancy.enter()
    }

    /// Whether [`Self::join_current_process`] moves the caller into this scope, which it does
    /// exactly when the `cgroup.procs` descriptor is open; without it a join returns `Ok(())` and
    /// moves nothing. Every scope [`prepare_scope`] hands out has one. The `RLIMIT_NPROC` decision
    /// keys on this, not on the scope's presence, so a join that places nothing never leaves a
    /// spawn without its process bound.
    pub(crate) fn places_subprocesses(&self) -> bool {
        self.procs_fd.is_some()
    }

    /// Current `memory.events`/`pids.events` counters.
    ///
    /// Must be read while the scope directory still exists — the files vanish with it, which is
    /// why cleanup happens in `Drop` at session end and never right after a child is reaped.
    #[cfg_attr(not(target_os = "linux"), allow(dead_code))]
    pub(crate) fn event_counters(&self) -> CgroupEventCounters {
        CgroupEventCounters {
            oom_kill: read_event_counter(&self.path.join("memory.events"), "oom_kill"),
            pids_max: read_event_counter(&self.path.join("pids.events"), "max"),
        }
    }
}

impl CgroupScope {
    /// Kill every process in this scope, whatever session or process group it put itself in.
    ///
    /// Writes `1` to `cgroup.kill` (Linux 5.14+), which `SIGKILL`s the whole scope in one kernel
    /// operation that races no fork. Where that file does not exist, falls back to freezing the
    /// scope, `SIGKILL`ing every pid `cgroup.procs` lists and thawing it, repeated until the scope
    /// is empty or [`KILL_FALLBACK_DEADLINE`] has passed. Returns as soon as the signals are sent;
    /// [`Self::wait_until_empty`] waits for the processes to be gone.
    ///
    /// The scope is a child of the delegated base, never the runtime's own cgroup, so the runtime
    /// is never in what this kills.
    pub(crate) fn kill_all(&self) -> std::io::Result<()> {
        if self.path.as_os_str().is_empty() {
            return Ok(());
        }
        #[cfg(target_os = "linux")]
        {
            #[cfg(test)]
            let forced = FORCE_KILL_FALLBACK.with(std::cell::Cell::get);
            #[cfg(not(test))]
            let forced = false;
            if !forced {
                match write_existing(&self.path.join("cgroup.kill"), "1") {
                    Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
                    written => return written,
                }
            }
            self.kill_by_freezing()
        }
        #[cfg(not(target_os = "linux"))]
        Ok(())
    }

    /// Whether `cgroup.procs` lists no process, polled until it does or `timeout` passes.
    ///
    /// An exited process leaves `cgroup.procs` when it exits, not when its parent reaps it, so
    /// a killed command the runtime has not yet waited for does not hold this up.
    pub(crate) fn wait_until_empty(&self, timeout: std::time::Duration) -> bool {
        let deadline = std::time::Instant::now() + timeout;
        loop {
            if self.member_pids().is_empty() {
                return true;
            }
            if std::time::Instant::now() >= deadline {
                return false;
            }
            std::thread::sleep(KILL_POLL_INTERVAL);
        }
    }

    /// Every pid `cgroup.procs` lists now. Empty when the file cannot be read, which is a scope
    /// whose directory is already gone.
    pub(crate) fn member_pids(&self) -> Vec<i32> {
        if self.path.as_os_str().is_empty() {
            return Vec::new();
        }
        std::fs::read_to_string(self.path.join("cgroup.procs"))
            .map(|procs| {
                procs
                    .lines()
                    .filter_map(|line| line.trim().parse().ok())
                    .collect()
            })
            .unwrap_or_default()
    }

    /// `cgroup.kill` for a kernel that has none: freeze, `SIGKILL` each listed pid, thaw.
    ///
    /// Freezing first stops a listed process from forking a child the loop has not seen, and
    /// keeps a pid from being reused between reading it and signalling it. A frozen task still
    /// takes `SIGKILL`; it dies once thawed.
    #[cfg(target_os = "linux")]
    #[allow(unsafe_code)]
    fn kill_by_freezing(&self) -> std::io::Result<()> {
        let freeze = self.path.join("cgroup.freeze");
        let deadline = std::time::Instant::now() + KILL_FALLBACK_DEADLINE;
        loop {
            let pids = self.member_pids();
            if pids.is_empty() {
                return Ok(());
            }
            // Best-effort: a kernel without the freezer still gets every listed pid killed.
            let _ = write_existing(&freeze, "1");
            for pid in pids {
                // SAFETY: `kill` takes two integers and dereferences nothing.
                unsafe { libc::kill(pid, libc::SIGKILL) };
            }
            let _ = write_existing(&freeze, "0");
            if std::time::Instant::now() >= deadline {
                return Err(std::io::Error::other(format!(
                    "processes were still in {} after {} ms of SIGKILL",
                    self.path.display(),
                    KILL_FALLBACK_DEADLINE.as_millis()
                )));
            }
            std::thread::sleep(KILL_POLL_INTERVAL);
        }
    }
}

/// How long [`CgroupScope::kill_all`]'s fallback keeps signalling a scope that is not yet empty.
const KILL_FALLBACK_DEADLINE: std::time::Duration = std::time::Duration::from_secs(2);

/// How often a kill or an emptiness wait re-reads `cgroup.procs`.
const KILL_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_millis(10);

/// Writes `value` to a control file that must already exist, so an absent file reads as
/// `NotFound` rather than as a refused create.
#[cfg(target_os = "linux")]
fn write_existing(path: &Path, value: &str) -> std::io::Result<()> {
    use std::io::Write as _;
    let file = std::fs::OpenOptions::new().write(true).open(path)?;
    (&file).write_all(value.as_bytes())
}

// Forces `CgroupScope::kill_all` onto its freeze-and-signal fallback, as on a kernel with no
// `cgroup.kill`. Read on the calling thread, so it reaches a `kill_all` the test calls itself.
#[cfg(all(test, target_os = "linux"))]
thread_local! {
    pub(crate) static FORCE_KILL_FALLBACK: std::cell::Cell<bool> =
        const { std::cell::Cell::new(false) };
}

// Makes a required `prepare_scope` fail as on a host that cannot delegate a scope. Read on the
// calling thread, so it reaches a `prepare_scope` the test calls itself.
#[cfg(all(test, target_os = "linux"))]
thread_local! {
    pub(crate) static FORCE_DELEGATION_FAILURE: std::cell::Cell<bool> =
        const { std::cell::Cell::new(false) };
}

impl Drop for CgroupScope {
    /// Kills whatever is still in the scope, then removes the scope directory, best-effort.
    ///
    /// The backstop for a session that never reached `ShellEnforcement::end_session`, which has
    /// already killed the scope on every ordinary path. A cgroup directory can only be removed
    /// once empty, so the removal waits, bounded, for the kill to land; a scope still occupied
    /// after that is logged and moved past rather than retried, because blocking teardown on a
    /// process that will not die would turn a leaked directory into a hung exit.
    fn drop(&mut self) {
        if self.path.as_os_str().is_empty() {
            return;
        }
        if !self.member_pids().is_empty() {
            let _ = self.kill_all();
            self.wait_until_empty(KILL_FALLBACK_DEADLINE);
        }
        // Drop the descriptor before the rmdir so nothing holds the directory open, and reap the
        // watcher, which holds its own descriptors on the scope's files until it exits.
        self.procs_fd = None;
        #[cfg(target_os = "linux")]
        {
            self.watcher = None;
        }
        if let Err(error) = std::fs::remove_dir(&self.path) {
            crate::runtime_err!(
                "[capsule-runtime] note: could not remove cgroup scope {} ({error}); \
                 it will be reclaimed once its last task exits",
                self.path.display()
            );
        }
    }
}

/// The process that kills one scope once nothing holds its lifeline's write end open.
///
/// [`Self::spawn`] forks it from the runtime; dropping this closes the write end and reaps it.
/// See the module documentation, "A scope outlives no runtime".
#[cfg(target_os = "linux")]
#[derive(Debug)]
struct ScopeWatcher {
    /// The lifeline's write end. Opened `O_CLOEXEC`, so no program the runtime execs holds a copy,
    /// and the watcher closes the copy it inherited: its read sees end-of-file exactly when this
    /// is closed or the runtime is gone.
    lifeline: std::sync::Mutex<Option<std::os::fd::OwnedFd>>,
    pid: libc::pid_t,
}

/// The descriptors a watcher kills its scope through, opened by the runtime before the fork:
/// `cgroup.kill` where the kernel has it, else `cgroup.freeze` (when present) and `cgroup.procs`
/// for the freeze-and-signal loop [`CgroupScope::kill_all`] falls back to.
#[cfg(target_os = "linux")]
#[derive(Clone, Copy)]
struct WatcherKill {
    kill: libc::c_int,
    freeze: libc::c_int,
    procs: libc::c_int,
}

#[cfg(target_os = "linux")]
impl ScopeWatcher {
    /// Fork the watcher of the scope at `scope`. Everything it needs is opened here, in the
    /// runtime, so the forked child — one thread of a multithreaded process — runs only
    /// async-signal-safe syscalls on descriptors and stack buffers that already exist.
    #[allow(unsafe_code)]
    fn spawn(scope: &Path) -> std::io::Result<Self> {
        use std::os::fd::{AsRawFd, FromRawFd, OwnedFd};

        #[cfg(test)]
        let forced = FORCE_KILL_FALLBACK.with(std::cell::Cell::get);
        #[cfg(not(test))]
        let forced = false;
        let kill = if forced {
            None
        } else {
            open_if_present(&scope.join("cgroup.kill"), libc::O_WRONLY)?
        };
        let (freeze, procs) = match kill {
            Some(_) => (None, None),
            None => (
                open_if_present(&scope.join("cgroup.freeze"), libc::O_WRONLY)?,
                Some(open_with(&scope.join("cgroup.procs"), libc::O_RDONLY)?),
            ),
        };
        let devnull = open_with(Path::new("/dev/null"), libc::O_RDWR)?;

        let mut ends = [0 as libc::c_int; 2];
        // SAFETY: `ends` is a two-element array, the size `pipe2` writes into.
        if unsafe { libc::pipe2(ends.as_mut_ptr(), libc::O_CLOEXEC) } != 0 {
            return Err(std::io::Error::last_os_error());
        }
        // SAFETY: both descriptors were just returned by the successful `pipe2` above and nothing
        // else owns them.
        let (read_end, write_end) =
            unsafe { (OwnedFd::from_raw_fd(ends[0]), OwnedFd::from_raw_fd(ends[1])) };

        let raw = |fd: &Option<OwnedFd>| fd.as_ref().map_or(-1, AsRawFd::as_raw_fd);
        let plan = WatcherKill {
            kill: raw(&kill),
            freeze: raw(&freeze),
            procs: raw(&procs),
        };
        // Sorted here, before the fork: the child keeps exactly these. The entries are distinct:
        // Rust's startup opens 0-2 when they are closed, so every descriptor opened above is 3 or
        // higher, and the child's `dup2` onto 0-2 replaces none of them.
        let mut keep = [-1 as libc::c_int; 7];
        let mut kept = 0;
        for fd in [
            0,
            1,
            2,
            read_end.as_raw_fd(),
            plan.kill,
            plan.freeze,
            plan.procs,
        ] {
            if fd >= 0 {
                keep[kept] = fd;
                kept += 1;
            }
        }
        keep[..kept].sort_unstable();
        let devnull_fd = devnull.as_raw_fd();
        let read_fd = read_end.as_raw_fd();
        let deadline_ns = KILL_FALLBACK_DEADLINE.as_nanos() as i64;

        // SAFETY: `fork()` from a possibly-multithreaded process is sound as long as the child
        // touches nothing but async-signal-safe primitives. The child runs only `dup2`,
        // `close_range`/`close`, `setpgid`, `sigaction`, `read`, `lseek`, `write`, `kill`,
        // `clock_gettime` and `nanosleep` on descriptors and stack values prepared above, then
        // `_exit`: no allocation, no locks, no stdio, no destructor.
        let pid = unsafe { libc::fork() };
        if pid < 0 {
            return Err(std::io::Error::last_os_error());
        }
        if pid == 0 {
            // SAFETY: forked-child context; `watch` ends in `_exit` and never returns.
            unsafe { watch(devnull_fd, &keep[..kept], read_fd, plan, deadline_ns) }
        }
        Ok(Self {
            lifeline: std::sync::Mutex::new(Some(write_end)),
            pid,
        })
    }

    /// Close the lifeline's write end, as the runtime's death would, without reaping.
    #[cfg(test)]
    fn cut_lifeline(&self) {
        self.lifeline
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .take();
    }
}

#[cfg(target_os = "linux")]
impl Drop for ScopeWatcher {
    /// Closes the lifeline, so the watcher kills the scope (by now empty, on every path through
    /// [`CgroupScope`]'s `Drop`) and exits, then reaps it.
    #[allow(unsafe_code)]
    fn drop(&mut self) {
        self.lifeline
            .get_mut()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .take();
        let mut status = 0;
        loop {
            // SAFETY: `self.pid` is the watcher this value forked; `status` is a live local.
            let waited = unsafe { libc::waitpid(self.pid, &mut status, 0) };
            if waited >= 0 || std::io::Error::last_os_error().raw_os_error() != Some(libc::EINTR) {
                return;
            }
        }
    }
}

/// `path` opened with `flags | O_CLOEXEC`.
#[cfg(target_os = "linux")]
fn open_with(path: &Path, flags: libc::c_int) -> std::io::Result<std::os::fd::OwnedFd> {
    use std::os::unix::fs::OpenOptionsExt;

    std::fs::OpenOptions::new()
        .read(flags & libc::O_ACCMODE != libc::O_WRONLY)
        .write(flags & libc::O_ACCMODE != libc::O_RDONLY)
        .custom_flags(libc::O_CLOEXEC)
        .open(path)
        .map(std::os::fd::OwnedFd::from)
}

/// [`open_with`], or `None` for a control file this kernel does not expose.
#[cfg(target_os = "linux")]
fn open_if_present(
    path: &Path,
    flags: libc::c_int,
) -> std::io::Result<Option<std::os::fd::OwnedFd>> {
    match open_with(path, flags) {
        Ok(fd) => Ok(Some(fd)),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
        Err(error) => Err(error),
    }
}

/// The watcher's whole life, in the forked child: hold no descriptor but `keep`, leave the
/// runtime's process group, ignore the signals a terminal or launcher sends a group, block until
/// the lifeline reads end-of-file, kill the scope, exit.
///
/// # Safety
///
/// Called only in a child just forked by [`ScopeWatcher::spawn`], with `keep` sorted and holding
/// 0, 1, 2, `read_fd` and every descriptor `plan` names. Never returns.
#[cfg(target_os = "linux")]
#[allow(unsafe_code)]
unsafe fn watch(
    devnull: libc::c_int,
    keep: &[libc::c_int],
    read_fd: libc::c_int,
    plan: WatcherKill,
    deadline_ns: i64,
) -> ! {
    // SAFETY (whole body): every call is an async-signal-safe syscall on descriptors the parent
    // opened before the fork or on stack values; nothing allocates or takes a lock.
    unsafe {
        // Standard streams onto /dev/null, so no pipe of the runtime's stays open through them.
        for target in 0..3 {
            libc::dup2(devnull, target);
        }
        let mut low: libc::c_uint = 0;
        for &fd in keep {
            let fd = fd as libc::c_uint;
            if fd > low {
                close_range(low, fd - 1);
            }
            low = fd + 1;
        }
        close_range(low, libc::c_uint::MAX);

        libc::setpgid(0, 0);
        let mut ignore: libc::sigaction = std::mem::zeroed();
        ignore.sa_sigaction = libc::SIG_IGN;
        for signal in [libc::SIGINT, libc::SIGTERM, libc::SIGHUP, libc::SIGPIPE] {
            libc::sigaction(signal, &ignore, std::ptr::null_mut());
        }

        let mut byte = [0u8; 1];
        loop {
            let read = libc::read(read_fd, byte.as_mut_ptr().cast(), 1);
            if read == 0 {
                break;
            }
            if read < 0 && *libc::__errno_location() != libc::EINTR {
                break;
            }
        }

        kill_scope(plan, deadline_ns);
        libc::_exit(0)
    }
}

/// Close every descriptor in `low..=high`: `close_range(2)`, else one `close` per descriptor up to
/// the open-files limit, for a kernel older than 5.9.
///
/// # Safety
///
/// Async-signal-safe; called only from [`watch`].
#[cfg(target_os = "linux")]
#[allow(unsafe_code)]
unsafe fn close_range(low: libc::c_uint, high: libc::c_uint) {
    // SAFETY: integer arguments only.
    unsafe {
        if libc::syscall(libc::SYS_close_range, low, high, 0 as libc::c_uint) == 0 {
            return;
        }
        let mut limit: libc::rlimit = std::mem::zeroed();
        let ceiling = if libc::getrlimit(libc::RLIMIT_NOFILE, &mut limit) == 0 {
            limit.rlim_cur.min(libc::c_uint::MAX as libc::rlim_t) as libc::c_uint
        } else {
            1 << 20
        };
        let mut fd = low;
        while fd <= high && fd < ceiling {
            libc::close(fd as libc::c_int);
            fd += 1;
        }
    }
}

/// [`CgroupScope::kill_all`] for the watcher: one write to `cgroup.kill`, else freeze, `SIGKILL`
/// every pid `cgroup.procs` lists, thaw, until none is listed or `deadline_ns` has passed.
///
/// # Safety
///
/// Async-signal-safe; called only from [`watch`].
#[cfg(target_os = "linux")]
#[allow(unsafe_code)]
unsafe fn kill_scope(plan: WatcherKill, deadline_ns: i64) {
    // SAFETY: syscalls on the descriptors `plan` names and on stack values.
    unsafe {
        if plan.kill >= 0 {
            write_control(plan.kill, b"1");
            return;
        }
        let start = monotonic_ns();
        loop {
            if plan.freeze >= 0 {
                write_control(plan.freeze, b"1");
            }
            let signalled = signal_listed(plan.procs);
            if plan.freeze >= 0 {
                write_control(plan.freeze, b"0");
            }
            if signalled == 0 || monotonic_ns() - start >= deadline_ns {
                return;
            }
            let pause = libc::timespec {
                tv_sec: 0,
                tv_nsec: KILL_POLL_INTERVAL.as_nanos() as libc::c_long,
            };
            libc::nanosleep(&pause, std::ptr::null_mut());
        }
    }
}

/// Write `value` at the start of the control file `fd`.
///
/// # Safety
///
/// Async-signal-safe.
#[cfg(target_os = "linux")]
#[allow(unsafe_code)]
unsafe fn write_control(fd: libc::c_int, value: &[u8]) {
    // SAFETY: `value` is a live slice for the duration of the call.
    unsafe {
        libc::lseek(fd, 0, libc::SEEK_SET);
        libc::write(fd, value.as_ptr().cast(), value.len());
    }
}

/// `SIGKILL` every pid the `cgroup.procs` descriptor `procs` lists now, read from its start into a
/// stack buffer and parsed as it arrives. Returns how many were signalled.
///
/// # Safety
///
/// Async-signal-safe.
#[cfg(target_os = "linux")]
#[allow(unsafe_code)]
unsafe fn signal_listed(procs: libc::c_int) -> usize {
    // SAFETY: reads into a stack buffer of the length passed; `kill` takes integers.
    unsafe {
        if procs < 0 || libc::lseek(procs, 0, libc::SEEK_SET) < 0 {
            return 0;
        }
        let mut buf = [0u8; 512];
        let mut pid: libc::pid_t = 0;
        let mut digits = false;
        let mut signalled = 0;
        loop {
            let read = libc::read(procs, buf.as_mut_ptr().cast(), buf.len());
            if read < 0 && *libc::__errno_location() == libc::EINTR {
                continue;
            }
            if read <= 0 {
                break;
            }
            for &byte in &buf[..read as usize] {
                if byte.is_ascii_digit() {
                    pid = pid
                        .saturating_mul(10)
                        .saturating_add(libc::pid_t::from(byte - b'0'));
                    digits = true;
                } else {
                    if digits && pid > 0 {
                        libc::kill(pid, libc::SIGKILL);
                        signalled += 1;
                    }
                    pid = 0;
                    digits = false;
                }
            }
        }
        if digits && pid > 0 {
            libc::kill(pid, libc::SIGKILL);
            signalled += 1;
        }
        signalled
    }
}

/// `CLOCK_MONOTONIC` in nanoseconds.
///
/// # Safety
///
/// Async-signal-safe.
#[cfg(target_os = "linux")]
#[allow(unsafe_code)]
unsafe fn monotonic_ns() -> i64 {
    // SAFETY: `now` is a live local `clock_gettime` writes into.
    unsafe {
        let mut now: libc::timespec = std::mem::zeroed();
        libc::clock_gettime(libc::CLOCK_MONOTONIC, &mut now);
        now.tv_sec as i64 * 1_000_000_000 + now.tv_nsec as i64
    }
}

/// Format `pid` into `buf` as decimal ASCII followed by a newline, returning the byte count.
///
/// Hand-rolled because [`CgroupScope::join_current_process`] runs between `fork` and `execve`,
/// where `format!` (and the allocator lock behind it) is not async-signal-safe.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
fn format_pid(pid: i32, buf: &mut [u8; 24]) -> usize {
    let mut digits = [0u8; 20];
    let mut count = 0;
    let mut value = pid.unsigned_abs() as u64;
    if value == 0 {
        digits[0] = b'0';
        count = 1;
    }
    while value > 0 {
        digits[count] = b'0' + (value % 10) as u8;
        value /= 10;
        count += 1;
    }

    let mut written = 0;
    for index in (0..count).rev() {
        buf[written] = digits[index];
        written += 1;
    }
    buf[written] = b'\n';
    written + 1
}

/// Read one `<key> <value>` counter out of a cgroup `*.events` file, or `0` if the file or key
/// is absent. Absence is not an error: `memory.events` only exists once the memory controller is
/// enabled, and treating a missing counter as zero means attribution simply declines to name a
/// limit rather than inventing one.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
fn read_event_counter(path: &Path, key: &str) -> u64 {
    let Ok(contents) = std::fs::read_to_string(path) else {
        return 0;
    };
    parse_event_counter(&contents, key)
}

/// Pure parse half of [`read_event_counter`] — no syscalls, so it is unit-testable on macOS.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
fn parse_event_counter(contents: &str, key: &str) -> u64 {
    for line in contents.lines() {
        let mut parts = line.split_whitespace();
        if parts.next() == Some(key) {
            return parts
                .next()
                .and_then(|value| value.parse().ok())
                .unwrap_or(0);
        }
    }
    0
}

/// Turn a session id into a directory name safe to create under the delegated base: everything
/// outside `[A-Za-z0-9._-]` becomes `_`.
///
/// The name is also made unique per scope, with the pid and a per-process serial. The id alone
/// cannot carry that: `plan::execute` passes the plan's *authored* id, so two plans running
/// concurrently in one process would otherwise land on the same directory — and because
/// [`CgroupScope`]'s `Drop` removes the scope by path, the first to finish would delete a cgroup
/// the other is still bounded by, leaving that subprocess tree unbounded. Sanitizing stays a
/// belt-and-braces guard against a hostile id rather than a live escape vector.
///
/// Operators still find scopes by the documented `murmur-*` glob.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
fn scope_dir_name(session_id: &str) -> String {
    static SERIAL: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
    let serial = SERIAL.fetch_add(1, std::sync::atomic::Ordering::Relaxed);

    let sanitized: String = session_id
        .chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() || c == '.' || c == '_' || c == '-' {
                c
            } else {
                '_'
            }
        })
        .collect();
    let pid = std::process::id();
    if sanitized.is_empty() {
        format!("murmur-{pid}-{serial}")
    } else {
        format!("murmur-{sanitized}-{pid}-{serial}")
    }
}

/// Extract the unified-hierarchy path from `/proc/self/cgroup` contents.
///
/// A cgroup v2 host writes exactly one line whose hierarchy id is `0` and whose controller list
/// is empty: `0::/user.slice/...`. Returning `None` for anything else is what makes a v1-only
/// (or hybrid, v2-less) host a clean "cannot delegate" rather than a wrong path.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
fn parse_unified_cgroup_path(contents: &str) -> Option<String> {
    for line in contents.lines() {
        let mut parts = line.splitn(3, ':');
        let (Some("0"), Some(""), Some(path)) = (parts.next(), parts.next(), parts.next()) else {
            continue;
        };
        if path.starts_with('/') {
            return Some(path.to_string());
        }
    }
    None
}

/// Find the `cgroup2` mount point in `/proc/mounts` contents, falling back to the conventional
/// `/sys/fs/cgroup`.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
fn parse_cgroup2_mount(contents: &str) -> String {
    for line in contents.lines() {
        let mut parts = line.split_whitespace();
        let (Some(_source), Some(target), Some(fstype)) =
            (parts.next(), parts.next(), parts.next())
        else {
            continue;
        };
        if fstype == "cgroup2" {
            return target.to_string();
        }
    }
    DEFAULT_CGROUP2_MOUNT.to_string()
}

/// Which of `wanted` are missing from a `cgroup.subtree_control` file's contents.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
fn missing_controllers(subtree_control: &str, wanted: &[&str]) -> Vec<String> {
    let enabled: Vec<&str> = subtree_control.split_whitespace().collect();
    wanted
        .iter()
        .filter(|controller| !enabled.contains(*controller))
        .map(|controller| (*controller).to_string())
        .collect()
}

/// Encode a `st_dev` value as the `MAJ:MIN` string `io.max` expects, using glibc's device-number
/// layout (12+20 bit split, not the legacy 8+8).
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
fn device_major_minor(dev: u64) -> (u64, u64) {
    let major = ((dev >> 8) & 0xfff) | ((dev >> 32) & !0xfff);
    let minor = (dev & 0xff) | ((dev >> 12) & !0xff);
    (major, minor)
}

#[cfg(target_os = "linux")]
impl CgroupScope {
    /// Probe delegation, enable the controllers, create the scope directory, and write its
    /// limits from [`scope_limits`]. Any failure to establish `memory.max`, `memory.swap.max`,
    /// `pids.max` or `cpu.max` is fatal — they are settable on any cgroup v2 host once the
    /// controllers are delegated, so a failure there means the bound genuinely does not exist and
    /// the launch must not proceed. The one exception is a kernel that exposes no
    /// `memory.swap.max` at all (see [`ScopeLimit::absent_ok`]).
    fn create(
        limits: &HostResourceLimits,
        session_id: &str,
        workdir: &Path,
    ) -> Result<(Self, IoMaxReport), String> {
        let base = delegated_base()?;

        let path = base.join(scope_dir_name(session_id));
        if let Err(error) = std::fs::create_dir(&path) {
            if error.kind() != std::io::ErrorKind::AlreadyExists {
                return Err(format!(
                    "could not create cgroup scope at {}: {error}",
                    path.display()
                ));
            }
        }

        let scope = Self {
            path,
            procs_fd: None,
            occupancy: Arc::default(),
            watcher: None,
        };

        for limit in scope_limits(limits) {
            scope.write_limit(&limit)?;
        }
        let io_max = scope.apply_io_max(limits.cgroup_io_bytes_per_sec, workdir);

        // Opened last, so a scope that failed any of the fatal writes above is never handed out
        // with a usable join descriptor.
        let procs_fd = open_write_only(&scope.path.join("cgroup.procs")).map_err(|error| {
            format!(
                "could not open {}/cgroup.procs for subprocess placement: {error}",
                scope.path.display()
            )
        })?;

        let watcher = ScopeWatcher::spawn(&scope.path).map_err(|error| {
            format!(
                "could not start the watcher that kills {} if the runtime dies: {error}",
                scope.path.display()
            )
        })?;

        Ok((
            Self {
                path: scope.take_path(),
                procs_fd: Some(procs_fd),
                occupancy: Arc::default(),
                watcher: Some(watcher),
            },
            io_max,
        ))
    }

    /// Move the path out of a scope that must not run its `Drop` (which would `rmdir` the
    /// directory the returned value is about to take ownership of).
    fn take_path(mut self) -> PathBuf {
        let path = std::mem::take(&mut self.path);
        // `self.path` is now empty, which `Drop` treats as "nothing to remove".
        path
    }

    /// Opened without `O_CREAT`, so a file this kernel does not expose reads as `NotFound`
    /// rather than as a refused create.
    fn write_limit(&self, limit: &ScopeLimit) -> Result<(), String> {
        let path = self.path.join(limit.file);
        let value = &limit.value;
        match write_existing(&path, &format!("{value}\n")) {
            Err(error) if limit.absent_ok && error.kind() == std::io::ErrorKind::NotFound => Ok(()),
            written => written
                .map_err(|error| format!("could not write {value} to {}: {error}", path.display())),
        }
    }

    /// Best-effort `io.max` on the workdir's backing block device, reported rather than printed.
    ///
    /// Non-fatal by design, and the only one of the four controllers treated that way: the
    /// backing device of a path cannot always be resolved to a real `MAJ:MIN` the block layer
    /// accepts (overlayfs, tmpfs, btrfs subvolumes and device-mapper stacks all break the
    /// assumption), and I/O throughput is the least safety-critical of the four — a capsule that
    /// saturates disk bandwidth is slow, where one that exhausts memory or pids is fatal.
    ///
    /// Non-fatal is not the same as unreported: the returned [`IoMaxReport`] travels into
    /// `--explain-scope`, into `session_start.effective_grants.io_max` and into `W-SEC-021`, so
    /// the manifest's declared ceiling and what the kernel accepted cannot disagree in silence.
    fn apply_io_max(&self, bytes_per_sec: u64, workdir: &Path) -> IoMaxReport {
        write_io_max(&self.path, bytes_per_sec, workdir)
    }
}

/// Writes `io.max` into `scope_path` for the device behind `workdir`, and says what happened.
///
/// The one place the `io.max` line is composed, shared by [`CgroupScope::apply_io_max`] at launch
/// and by [`probe_io_max`] under `--explain-scope`, so the diagnostic exercises the identical
/// write the launch performs rather than a paraphrase of it.
///
/// The device comes from [`io_max_device`], which resolves it from the nearest *existing*
/// ancestor of `workdir`: `--explain-scope` runs before the workdir is created, and a launch's
/// workdir sits on the same filesystem as the project directory it is created under.
#[cfg(target_os = "linux")]
fn write_io_max(scope_path: &Path, bytes_per_sec: u64, workdir: &Path) -> IoMaxReport {
    let (major, minor) = match io_max_device(workdir) {
        Ok(device) => device,
        Err(reason) => return IoMaxReport::unavailable(bytes_per_sec, reason),
    };
    let value = format!("{major}:{minor} rbps={bytes_per_sec} wbps={bytes_per_sec}");
    match std::fs::write(scope_path.join("io.max"), format!("{value}\n")) {
        Ok(()) => IoMaxReport::enforced(bytes_per_sec),
        // The scope directory is deliberately not named: `--explain-scope` writes into a throwaway
        // cgroup and a launch writes into the session's, so including it would make the two
        // reports differ for no reason a reader could act on — and
        // `session_start.effective_grants` has to be the byte-for-byte object `--explain-scope`
        // prints. The `MAJ:MIN` and the errno are what identify the failure.
        Err(error) => IoMaxReport::unavailable(
            bytes_per_sec,
            format!("writing `{value}` to the scope's io.max: {error}"),
        ),
    }
}

/// The `MAJ:MIN` an `io.max` line has to name for the filesystem behind `path`, or why that
/// filesystem has no device a ceiling can name.
///
/// `st_dev` is not the answer on its own. Two things stand between a path and a device the block
/// layer accepts:
///
/// * A filesystem the kernel gives an anonymous `dev_t` — btrfs, overlayfs, tmpfs, every FUSE
///   mount — has major `0`, which names no device. `/proc/self/mountinfo` records what the
///   covering mount was mounted from, and where that source is a block device node its `st_rdev`
///   is the device the filesystem's I/O actually reaches.
/// * `blkcg` binds a limit to a request queue and only a whole disk carries one, so the block
///   layer answers `ENODEV` for a partition — which is what `st_dev` reports for the common case
///   of a filesystem on `/dev/sda1` or `/dev/nvme0n1p7`. [`whole_disk_device`] takes the last hop.
///
/// A filesystem with no block device behind it at all carries no I/O ceiling, and the `Err` says
/// which mount and what it is backed by. That is a measurement of the resolution, not a guess
/// about the write: there is no device for a write to name.
#[cfg(target_os = "linux")]
fn io_max_device(path: &Path) -> Result<(u64, u64), String> {
    use std::os::linux::fs::MetadataExt;

    let Some(existing) = nearest_existing_ancestor(path) else {
        return Err(format!(
            "could not stat {} or any ancestor of it to find its backing device",
            path.display()
        ));
    };
    let metadata = existing.metadata().map_err(|error| {
        format!(
            "could not stat {} to find its backing device: {error}",
            existing.display()
        )
    })?;
    let (major, minor) = device_major_minor(metadata.st_dev());
    let (major, minor) = if major == 0 {
        mount_source_device(existing)?
    } else {
        (major, minor)
    };
    Ok(whole_disk_device(major, minor))
}

/// The block device the mount covering `path` was mounted from.
///
/// Reached only for a filesystem whose `st_dev` is anonymous. `path` must exist, so that
/// canonicalizing it yields the path the kernel's own mount lookup would take — a mount point
/// reached through a symlink matches nothing in `mountinfo`.
#[cfg(target_os = "linux")]
fn mount_source_device(path: &Path) -> Result<(u64, u64), String> {
    use std::os::unix::fs::{FileTypeExt, MetadataExt};

    let resolved = path.canonicalize().unwrap_or_else(|_| path.to_path_buf());
    let mountinfo = std::fs::read_to_string("/proc/self/mountinfo").map_err(|error| {
        format!(
            "reading /proc/self/mountinfo to find the device behind {}: {error}",
            resolved.display()
        )
    })?;
    let Some(mount) = mount_covering(&mountinfo, &resolved) else {
        return Err(format!(
            "no mount in /proc/self/mountinfo covers {}, so its backing device is unknown",
            resolved.display()
        ));
    };
    match std::fs::metadata(&mount.source) {
        Ok(metadata) if metadata.file_type().is_block_device() => {
            Ok(device_major_minor(metadata.rdev()))
        }
        _ => Err(format!(
            "the filesystem mounted at {} is backed by `{}`, which is not a block device, so no \
             io.max ceiling can name one",
            mount.mount_point.display(),
            mount.source.display()
        )),
    }
}

/// `major:minor` of the whole disk carrying `major:minor`, which is itself where it is not a
/// partition.
///
/// A partition has a `partition` attribute in sysfs and sits inside its disk's directory, so the
/// parent's `dev` is the disk. Every failure to read sysfs answers with the device as given: the
/// write that follows is what decides, and reporting a device the block layer then refuses is a
/// better outcome than reporting a device nothing was tried against.
#[cfg(target_os = "linux")]
fn whole_disk_device(major: u64, minor: u64) -> (u64, u64) {
    let device = PathBuf::from(format!("/sys/dev/block/{major}:{minor}"));
    if !device.join("partition").exists() {
        return (major, minor);
    }
    std::fs::read_to_string(device.join("../dev"))
        .ok()
        .and_then(|contents| parse_dev_line(&contents))
        .unwrap_or((major, minor))
}

/// The `MAJ:MIN` on a sysfs `dev` attribute.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
fn parse_dev_line(contents: &str) -> Option<(u64, u64)> {
    let (major, minor) = contents.trim().split_once(':')?;
    Some((major.parse().ok()?, minor.parse().ok()?))
}

/// One `/proc/self/mountinfo` line, reduced to the two fields that identify a filesystem's
/// backing device.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
#[derive(Debug, PartialEq, Eq)]
struct MountEntry {
    /// Where the filesystem is mounted, as the kernel spells it.
    mount_point: PathBuf,
    /// What it was mounted from: a block device node for an ordinary disk filesystem, and a bare
    /// name such as `tmpfs` for one with nothing behind it.
    source: PathBuf,
}

/// The mount `path` lands in: the entry whose mount point is the longest prefix of it.
///
/// Later lines win a tie. `mountinfo` is in mount order, so a filesystem mounted over an earlier
/// one at the same point is the one a lookup reaches.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
fn mount_covering(mountinfo: &str, path: &Path) -> Option<MountEntry> {
    let mut best: Option<(usize, MountEntry)> = None;
    for line in mountinfo.lines() {
        let Some(entry) = parse_mount_entry(line) else {
            continue;
        };
        if !path.starts_with(&entry.mount_point) {
            continue;
        }
        let depth = entry.mount_point.components().count();
        if best.as_ref().is_none_or(|(best, _)| depth >= *best) {
            best = Some((depth, entry));
        }
    }
    best.map(|(_, entry)| entry)
}

/// Mount point and source out of one `mountinfo` line.
///
/// The optional-fields run is terminated by a lone `-`, which no field can contain: `mountinfo`
/// escapes space, tab, newline and backslash as octal, so splitting on `" - "` is unambiguous.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
fn parse_mount_entry(line: &str) -> Option<MountEntry> {
    let (mounted, backing) = line.split_once(" - ")?;
    let mount_point = mounted.split_whitespace().nth(4)?;
    let source = backing.split_whitespace().nth(1)?;
    Some(MountEntry {
        mount_point: unescape_octal(mount_point),
        source: unescape_octal(source),
    })
}

/// Decode `mountinfo`'s `\NNN` octal escapes, byte-wise, so a mount point containing a space or a
/// non-ASCII byte compares equal to the path the kernel reports for it.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
fn unescape_octal(field: &str) -> PathBuf {
    use std::os::unix::ffi::OsStringExt;

    let bytes = field.as_bytes();
    let mut out: Vec<u8> = Vec::with_capacity(bytes.len());
    let mut index = 0;
    while index < bytes.len() {
        let escape = bytes[index] == b'\\'
            && index + 3 < bytes.len()
            && bytes[index + 1..index + 4]
                .iter()
                .all(|byte| (b'0'..=b'7').contains(byte));
        if escape {
            let digits = &field[index + 1..index + 4];
            out.push(u8::from_str_radix(digits, 8).expect("three octal digits"));
            index += 4;
        } else {
            out.push(bytes[index]);
            index += 1;
        }
    }
    PathBuf::from(std::ffi::OsString::from_vec(out))
}

/// The closest ancestor of `path` that exists, which is `path` itself where it does.
///
/// The backing device is a property of the filesystem, not of the leaf, so an ancestor's answer
/// is the answer for a path that has not been created yet — the state `--explain-scope` finds the
/// would-be workdir in.
#[cfg(target_os = "linux")]
fn nearest_existing_ancestor(path: &Path) -> Option<&Path> {
    let mut candidate = Some(path);
    while let Some(current) = candidate {
        if current.exists() {
            return Some(current);
        }
        candidate = current.parent();
    }
    None
}

/// The delegated cgroup root this process creates its scopes under, resolved once.
///
/// [`probe_delegation`] derives the root from `/proc/self/cgroup`, but establishing the first
/// scope may move this process into `<root>/murmur-supervisor` to vacate the root — cgroup v2
/// forbids a cgroup from both holding processes and enabling controllers for its children. Once
/// that move has happened the process's own cgroup is no longer the delegated root, so
/// re-deriving it per call mistakes the leaf for the root and nests one level deeper every time,
/// until the nested leaf has no delegated controllers at all and every further session is
/// refused. Any process that opens a second session — concurrently or in sequence — hits this.
///
/// Resolving once, under a lock, keeps every scope a sibling directly under the real root, and
/// makes concurrent first-callers agree on that root instead of racing to relocate each other.
/// Only success is cached: a host that is not delegated must keep reporting so on every attempt.
#[cfg(target_os = "linux")]
fn delegated_base() -> Result<PathBuf, String> {
    static BASE: std::sync::Mutex<Option<PathBuf>> = std::sync::Mutex::new(None);

    // A poisoned lock guards no invariant worth honouring here: the cache is either populated or
    // it is not, and a caller that panicked cannot have left it half-written.
    let mut cached = BASE.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
    if let Some(base) = cached.as_ref() {
        return Ok(base.clone());
    }

    // Best-effort: relocate into our own scope before probing. On failure, the probe below runs
    // against the inherited cgroup — the correct fallback on a host with no systemd to ask.
    move_self_into_own_systemd_scope();

    let base = probe_delegation()?;
    enable_controllers(&base)?;
    *cached = Some(base.clone());
    Ok(base)
}

/// Ask the systemd user manager for a freshly created, delegated transient scope holding this
/// process, so the cgroup this runtime builds under is one it is the only occupant of.
///
/// Best-effort and non-fatal: every failure here (no systemd, no session bus, policy denial, a
/// migration that never lands) leaves the process exactly where it was — the state
/// [`probe_delegation`] handles. Never let a failed optimisation become a launch refusal.
///
/// Runs at most once per process — a scope is a property of the process, not the session, and
/// [`delegated_base`] retries on failure but caches only success, so the guard has to live here.
#[cfg(target_os = "linux")]
fn move_self_into_own_systemd_scope() {
    static ATTEMPTED: std::sync::Once = std::sync::Once::new();
    ATTEMPTED.call_once(|| {
        let _ = request_transient_scope(&transient_scope_unit_name());
    });
}

/// Name of the transient scope unit to ask systemd for.
///
/// `murmur-` keeps the operator-facing convention the scope *directories* already use
/// ([`scope_dir_name`]), so `systemctl --user list-units 'murmur-*.scope'` finds the runtime's
/// units the same way a `murmur-*` glob finds its cgroup directories. The pid and a per-process
/// serial keep it unique against a recycled pid whose earlier unit systemd has not yet collected.
#[cfg(target_os = "linux")]
fn transient_scope_unit_name() -> String {
    static SERIAL: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
    let serial = SERIAL.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    format!("murmur-{}-{serial}.scope", std::process::id())
}

/// The D-Bus half of [`move_self_into_own_systemd_scope`], plus confirmation that the kernel
/// actually moved us.
#[cfg(target_os = "linux")]
fn request_transient_scope(unit: &str) -> Result<(), String> {
    // `zbus::block_on` drives the connection on the calling thread and panics or deadlocks inside
    // another runtime's task; run it on a thread of its own so a caller inside tokio is safe.
    let owned_unit = unit.to_string();
    std::thread::Builder::new()
        .name("murmur-cgroup-scope".to_string())
        .spawn(move || start_transient_scope(&owned_unit))
        .map_err(|error| format!("could not spawn the D-Bus thread: {error}"))?
        .join()
        .map_err(|_| "the D-Bus thread panicked".to_string())??;

    await_scope_membership(unit)
}

/// `StartTransientUnit` on the systemd user manager: create `unit` as a scope containing this
/// process, with its cgroup subtree delegated to us.
///
/// The call is
/// `org.freedesktop.systemd1.Manager.StartTransientUnit(in s name, in s mode, in a(sv) properties,
/// in a(sa(sv)) aux)` on `/org/freedesktop/systemd1` at the well-known name
/// `org.freedesktop.systemd1`, over the **session** bus (`$DBUS_SESSION_BUS_ADDRESS`, falling back
/// to `$XDG_RUNTIME_DIR/bus`) so the unit lands in this user's manager and needs no privilege.
/// The properties are the same set `systemd-run --user --scope -p Delegate=yes` sends:
///
/// * `PIDs` (`au`) — this process, which is what makes systemd *move* it rather than fork a new one.
/// * `Delegate` (`b`) — hand the scope's subtree to us, controllers and all, and stop managing it.
/// * `CollectMode` (`s`) — `inactive-or-failed`, so the unit is garbage-collected when this
///   process exits rather than lingering as a failed unit an operator has to reset.
/// * `Description` (`s`) — what `systemctl --user list-units` shows beside the unit.
#[cfg(target_os = "linux")]
fn start_transient_scope(unit: &str) -> Result<(), String> {
    use zbus::Variant;

    let properties: Vec<(&str, Variant<'_>)> = vec![
        (
            "Description",
            Variant::from("Murmur capsule subprocess scope"),
        ),
        ("PIDs", Variant::from(vec![std::process::id()])),
        ("Delegate", Variant::from(true)),
        ("CollectMode", Variant::from("inactive-or-failed")),
    ];
    // No auxiliary units: a scope has none.
    let aux: Vec<(&str, Vec<(&str, Variant<'_>)>)> = Vec::new();

    zbus::block_on(async {
        let connection = zbus::Connection::session()
            .await
            .map_err(|error| format!("no systemd session bus: {error}"))?;
        connection
            .call_method(
                Some("org.freedesktop.systemd1"),
                "/org/freedesktop/systemd1",
                Some("org.freedesktop.systemd1.Manager"),
                "StartTransientUnit",
                // "fail" rather than "replace": a name collision means something is wrong with
                // our assumption of uniqueness, and displacing an unrelated unit is never the fix.
                &(unit, "fail", properties, aux),
            )
            .await
            .map(|_| ())
            .map_err(|error| format!("systemd refused to start {unit}: {error}"))
    })
}

/// Wait until `/proc/self/cgroup` actually reports us inside `unit`'s cgroup.
///
/// `StartTransientUnit` returns once the job is *enqueued*, not once the move has happened, and
/// every later step (the delegation probe, the controller enable, the scope directory) depends on
/// where this process actually lives — so migration is confirmed from `/proc/self/cgroup`, the
/// same source the rest of this module reads, rather than inferred from a job object path.
///
/// The 2s deadline mirrors [`enable_controllers`]: both wait on systemd/kernel bookkeeping that
/// normally completes in a millisecond or two, and a wait that long means something structural is
/// wrong. A timeout here is cheap — the caller ignores it and falls back to probing the inherited
/// cgroup. The step is 1ms because this wait sits on the launch path of every capsule that needs a
/// scope, and the move lands a millisecond or two after `StartTransientUnit` returns, so every
/// step the loop oversleeps is added to the launch.
#[cfg(target_os = "linux")]
fn await_scope_membership(unit: &str) -> Result<(), String> {
    let deadline = std::time::Instant::now() + Duration::from_secs(2);
    loop {
        let current = std::fs::read_to_string("/proc/self/cgroup").unwrap_or_default();
        if let Some(path) = parse_unified_cgroup_path(&current) {
            if path.rsplit('/').next() == Some(unit) {
                return Ok(());
            }
        }
        if std::time::Instant::now() >= deadline {
            return Err(format!(
                "systemd accepted the request for {unit} but this process is not in it"
            ));
        }
        std::thread::sleep(Duration::from_millis(1));
    }
}

/// Locate the delegated cgroup v2 directory this process may create children under, confirming
/// the controllers this runtime needs are actually available there.
///
/// Returns the base directory on success, or a message naming what is missing — that message is
/// what reaches the operator inside
/// [`crate::errors::RuntimeError::CgroupDelegationUnavailable`], so it names the delegation
/// requirement rather than only the failed syscall.
#[cfg(target_os = "linux")]
fn probe_delegation() -> Result<PathBuf, String> {
    let self_cgroup = std::fs::read_to_string("/proc/self/cgroup")
        .map_err(|error| format!("could not read /proc/self/cgroup: {error}"))?;
    let relative = parse_unified_cgroup_path(&self_cgroup).ok_or_else(|| {
        "this host has no cgroup v2 unified hierarchy (no `0::` line in /proc/self/cgroup); \
         cgroup v1-only and hybrid hosts cannot be delegated a v2 scope"
            .to_string()
    })?;

    let mount = std::fs::read_to_string("/proc/mounts")
        .map(|contents| parse_cgroup2_mount(&contents))
        .unwrap_or_else(|_| DEFAULT_CGROUP2_MOUNT.to_string());

    let base = ascend_out_of_supervisor_leaves(
        PathBuf::from(mount).join(relative.trim_start_matches('/')),
    );
    if !base.is_dir() {
        return Err(format!(
            "delegated cgroup directory {} does not exist",
            base.display()
        ));
    }

    let controllers =
        std::fs::read_to_string(base.join("cgroup.controllers")).map_err(|error| {
            format!(
                "could not read {}/cgroup.controllers: {error}",
                base.display()
            )
        })?;
    let unavailable = missing_controllers(&controllers, REQUIRED_CONTROLLERS);
    if !unavailable.is_empty() {
        return Err(format!(
            "cgroup controllers [{}] are not delegated to {}; the unit `mur` runs under needs \
             `Delegate=yes` (or `Delegate=memory pids cpu io`)",
            unavailable.join(", "),
            base.display()
        ));
    }

    // The probe must confirm we can actually *create* a child here, not just read the directory:
    // a readable cgroup with no write permission is the exact shape of an undelegated host.
    // Named per-pid so two `mur` processes probing concurrently never race on the same path.
    let probe_dir = base.join(format!("murmur-delegation-probe-{}", std::process::id()));
    let _ = std::fs::remove_dir(&probe_dir);
    std::fs::create_dir(&probe_dir).map_err(|error| {
        format!(
            "cannot create a child cgroup under {} ({error}); the unit `mur` runs under needs \
             `Delegate=yes` for memory, pids, cpu and io",
            base.display()
        )
    })?;
    let _ = std::fs::remove_dir(&probe_dir);

    Ok(base)
}

/// Enable `memory`, `pids`, `cpu` (and `io`, best-effort) in the delegated base's
/// `cgroup.subtree_control` so a child cgroup can carry their limit files.
///
/// If the base still holds this process directly, the write is refused (`EBUSY`) by cgroup v2's
/// "no internal processes" rule; the recovery is to move this process into its own leaf and try
/// once more. That relocation stays entirely inside the delegated subtree.
#[cfg(target_os = "linux")]
fn enable_controllers(base: &Path) -> Result<(), String> {
    let subtree_control = base.join("cgroup.subtree_control");
    let current = std::fs::read_to_string(&subtree_control).unwrap_or_default();

    // `io` is requested opportunistically: a host that delegates only the three required
    // controllers still gets a fully-enforced scope, just without `io.max`.
    let available = std::fs::read_to_string(base.join("cgroup.controllers")).unwrap_or_default();
    let mut wanted: Vec<&str> = REQUIRED_CONTROLLERS.to_vec();
    if available.split_whitespace().any(|c| c == "io") {
        wanted.push("io");
    }

    let missing = missing_controllers(&current, &wanted);
    if missing.is_empty() {
        return Ok(());
    }
    let directive = missing
        .iter()
        .map(|controller| format!("+{controller}"))
        .collect::<Vec<_>>()
        .join(" ");

    if std::fs::write(&subtree_control, format!("{directive}\n")).is_ok() {
        return Ok(());
    }

    move_self_to_supervisor_leaf(base)?;

    // Stepping aside vacates *this* process, and cgroup v2 refuses the write while the base holds
    // any task at all. When several murmur processes share a base — a test binary and the `mur`
    // children it spawns, or two capsules launched together — each is independently discovering
    // it must step aside, so the first write after our own move can still hit `EBUSY` on peers
    // that have not moved yet. Retrying briefly lets the group finish draining; a peer that wins
    // the race and enables the controllers first is just as good, which is why the loop re-reads
    // `subtree_control` rather than only retrying the write.
    //
    // Bounded, because a task that never leaves is a genuine misconfiguration and has to surface
    // as one rather than hang the launch.
    let deadline = std::time::Instant::now() + Duration::from_secs(2);
    // Left uninitialised: every path that reaches the deadline check has just gone through the
    // write's `Err` arm, so the fallback below only ever describes a real failure.
    let mut last_error;
    loop {
        let current = std::fs::read_to_string(&subtree_control).unwrap_or_default();
        if missing_controllers(&current, &wanted).is_empty() {
            return Ok(());
        }
        match std::fs::write(&subtree_control, format!("{directive}\n")) {
            Ok(()) => return Ok(()),
            Err(error) => last_error = Some(error),
        }
        if std::time::Instant::now() >= deadline {
            break;
        }
        std::thread::sleep(Duration::from_millis(20));
    }

    Err(format!(
        "could not enable controllers [{}] in {} ({}); the unit `mur` runs under needs \
         `Delegate=yes` for memory, pids, cpu and io",
        missing.join(", "),
        subtree_control.display(),
        last_error.map_or_else(|| "still occupied".to_string(), |error| error.to_string()),
    ))
}

/// Walk up out of any `murmur-supervisor` leaves the path ends in.
///
/// The leaf is where a murmur process steps aside so controllers can be enabled on the delegated
/// base. Every process it then spawns *inherits that leaf as its own cgroup* — so a `mur`
/// subprocess reading `/proc/self/cgroup` would take the leaf for its delegated base and step
/// aside again inside it, one level deeper per generation, until the innermost leaf has no
/// controllers delegated to it at all and every launch below that point is refused.
///
/// Ascending here makes every generation agree on the same real base, which is also what lets
/// them share one scope parent rather than fragmenting the delegation.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
fn ascend_out_of_supervisor_leaves(mut base: PathBuf) -> PathBuf {
    while base.file_name().and_then(|name| name.to_str()) == Some(SUPERVISOR_LEAF) {
        base.pop();
    }
    base
}

/// Move this process into `<base>/murmur-supervisor`, creating it if needed.
#[cfg(target_os = "linux")]
fn move_self_to_supervisor_leaf(base: &Path) -> Result<(), String> {
    let leaf = base.join(SUPERVISOR_LEAF);
    if let Err(error) = std::fs::create_dir(&leaf) {
        if error.kind() != std::io::ErrorKind::AlreadyExists {
            return Err(format!(
                "could not create {} to vacate the delegated cgroup: {error}",
                leaf.display()
            ));
        }
    }
    std::fs::write(
        leaf.join("cgroup.procs"),
        format!("{}\n", std::process::id()),
    )
    .map_err(|error| {
        format!(
            "could not move this process into {} to vacate the delegated cgroup: {error}",
            leaf.display()
        )
    })
}

/// `open(path, O_WRONLY | O_CLOEXEC)`.
///
/// `CLOEXEC` matters twice over: the descriptor is written from `pre_exec` (still open at that
/// point) and must not survive into the exec'd binary, where it would hand a capsule the ability
/// to move arbitrary pids into its own cgroup.
#[cfg(target_os = "linux")]
#[allow(unsafe_code)]
fn open_write_only(path: &Path) -> std::io::Result<std::os::fd::OwnedFd> {
    use std::os::unix::fs::OpenOptionsExt;

    std::fs::OpenOptions::new()
        .write(true)
        .custom_flags(libc::O_CLOEXEC)
        .open(path)
        .map(std::os::fd::OwnedFd::from)
}

/// Whether this host can hand the runtime a delegated cgroup v2 scope — the precondition
/// [`prepare_scope`] fails closed on.
///
/// Test support, not a runtime code path: `mur` never asks this question, it asks systemd for a
/// scope and reports [`crate::errors::RuntimeError::CgroupDelegationUnavailable`] when it cannot
/// get one. Tests that launch a capsule tripping [`requires_process_bounding`] call this to skip
/// visibly instead, because a containerised runner has no delegable subtree to give and the
/// refusal there says nothing about the code under test. What CI therefore cannot cover is
/// covered by hand: `docs/content/reference/resource-limits-manual-verification.md`.
///
/// Probed by actually creating a transient `--user` scope with `Delegate=yes`, since a
/// `systemd-run` binary on `PATH` says nothing about a reachable user manager — and the user
/// manager is the half a runner is missing. Off Linux there are no cgroups, this module is inert
/// and nothing is ever refused, so the answer is `true`.
pub fn cgroup_delegation_available() -> bool {
    #[cfg(not(target_os = "linux"))]
    {
        true
    }

    #[cfg(target_os = "linux")]
    {
        // Answered through `delegated_base`, the same call `prepare_scope` makes, so the answer
        // cannot disagree with what a launch does on this host. A separate `systemd-run` probe
        // answers a different question and is wrong in both directions: a user manager hands out
        // a scope on a host whose controllers `enable_controllers` cannot delegate, and the
        // inherited cgroup carries delegated controllers on a host with no reachable session bus.
        delegated_base().is_ok()
    }
}

/// Whether the filesystem behind `path` has a block device an `io.max` ceiling can name.
///
/// Test support beside [`cgroup_delegation_available`], for the half of the `io.max` scenario
/// that needs a ceiling to actually apply: a checkout on tmpfs, on overlayfs or on a FUSE mount
/// has no device for the block layer to bind a limit to, and `enforced` is unreachable there
/// however well the host delegates cgroups. Answered through [`io_max_device`], the same
/// resolution a launch performs, so a test cannot stand down on a host the launch would have
/// satisfied. Off Linux there is no `io.max` at all.
#[must_use]
pub fn io_max_device_available(path: &Path) -> bool {
    #[cfg(not(target_os = "linux"))]
    {
        let _ = path;
        false
    }

    #[cfg(target_os = "linux")]
    {
        io_max_device(path).is_ok()
    }
}

/// Whether a test that launches a subprocess-capable capsule must stand down on this host,
/// printing the blocker when it must.
///
/// Both blockers are asked through the same calls a launch makes -- `delegated_base` for the
/// cgroup scope, a fresh `HostProbe`'s network attempt for the network namespace -- so the gate and
/// the launch cannot reach different conclusions about the host. A containerised CI runner
/// typically clears the first and fails the second.
pub fn skip_without_host_support(test_name: &str) -> bool {
    if crate::network_namespace::skip_without_egress_namespace(test_name) {
        return true;
    }
    if !cgroup_delegation_available() {
        crate::runtime_err!(
            "[SKIP-HOST] {test_name}: this host cannot delegate a cgroup v2 scope, so a capsule \
             that can spawn native subprocesses refuses to launch with E-RUN-012 before anything \
             this test observes happens -- see {}",
            murmur_artifact::docs_reference_url!("resource-limits/#platform-behavior")
        );
        return true;
    }
    false
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::types::CapabilityPolicy;

    /// The wire spelling every consumer reads: `--explain-scope --json`, `mur doctor` and
    /// `session_start.effective_grants.io_max` in `trace.jsonl`. Kebab-case, not the `Debug`
    /// rendering, which is free to change.
    #[test]
    fn io_max_status_wire_names_are_kebab_case() {
        let spellings: Vec<String> = [
            IoMaxStatus::Enforced,
            IoMaxStatus::Unavailable,
            IoMaxStatus::NotRequired,
            IoMaxStatus::NotProbed,
        ]
        .into_iter()
        .map(|status| {
            serde_json::to_value(status)
                .unwrap()
                .as_str()
                .unwrap()
                .to_string()
        })
        .collect();
        assert_eq!(
            spellings,
            vec!["enforced", "unavailable", "not-required", "not-probed"]
        );
    }

    /// A report nobody filled in must claim nothing, which is what makes it safe to construct one
    /// at stage time and overwrite it once the scope exists.
    #[test]
    fn an_unfilled_report_claims_nothing() {
        let report = IoMaxReport::default();
        assert_eq!(report.status, IoMaxStatus::NotProbed);
        assert_eq!(report.declared_bytes_per_sec, 0);
        assert_eq!(report.reason, None);
        assert_eq!(io_max_warning(&report), None);
    }

    /// `reason` is the only optional key, and it is present for every status but `Enforced`.
    #[test]
    fn only_a_failed_write_omits_nothing_and_a_successful_one_explains_nothing() {
        let enforced = serde_json::to_value(IoMaxReport::enforced(1_000)).unwrap();
        assert_eq!(enforced["status"], "enforced");
        assert_eq!(enforced["declared_bytes_per_sec"], 1_000);
        assert!(
            enforced.get("reason").is_none(),
            "an applied ceiling has nothing to explain: {enforced}"
        );

        let unavailable =
            serde_json::to_value(IoMaxReport::unavailable(1_000, "no io controller")).unwrap();
        assert_eq!(unavailable["status"], "unavailable");
        assert_eq!(unavailable["reason"], "no io controller");
    }

    /// `W-SEC-021` fires on the one status that is a shortfall, and its message names the
    /// declared ceiling, the host's reason, and the three limits that *are* still enforced —
    /// without which "io.max did not apply" reads as "this session is unbounded".
    #[test]
    fn the_warning_fires_only_where_a_declared_ceiling_did_not_apply() {
        assert_eq!(io_max_warning(&IoMaxReport::enforced(4_096)), None);
        assert_eq!(
            io_max_warning(&IoMaxReport::not_required(4_096, NO_SCOPE_REASON)),
            None
        );

        let (code, message) = io_max_warning(&IoMaxReport::unavailable(
            4_096,
            "device 0:24 is not a block device",
        ))
        .expect("an unapplied ceiling must be reported");
        assert_eq!(code, murmur_artifact::W_SEC_021);
        assert!(message.contains("4096 bytes/s"), "message was: {message}");
        assert!(
            message.contains("device 0:24 is not a block device"),
            "message was: {message}"
        );
        assert!(message.contains("memory.max"), "message was: {message}");
    }

    /// `--explain-scope` runs before the workdir exists, so the probe has to name the device a
    /// launch would use without creating anything.
    #[cfg(target_os = "linux")]
    #[test]
    fn the_device_is_resolved_from_the_nearest_existing_ancestor() {
        let temp = tempfile::tempdir().unwrap();
        let missing = temp
            .path()
            .join("workdir")
            .join("does")
            .join("not")
            .join("exist");
        assert_eq!(
            nearest_existing_ancestor(&missing),
            Some(temp.path()),
            "an ancestor of a tempdir always exists"
        );
        assert_eq!(
            io_max_device(&missing).ok(),
            io_max_device(temp.path()).ok(),
            "a path that does not exist yet must resolve to its filesystem's device"
        );
        assert!(!missing.exists(), "resolving a device must create nothing");
    }

    /// The mount a path lands in is the deepest one covering it, not the first line that matches
    /// — `/` covers every path, and a nested mount is what a lookup actually reaches.
    #[test]
    fn the_covering_mount_is_the_deepest_one() {
        let mountinfo = "\
25 30 0:29 / / rw,relatime shared:1 - btrfs /dev/nvme0n1p5 rw,subvol=/
26 25 0:31 / /dev/shm rw,nosuid shared:2 - tmpfs tmpfs rw,inode64
27 25 0:41 / /space rw,relatime shared:3 - btrfs /dev/nvme0n1p7 rw,subvol=/
";
        let covering = |path: &str| mount_covering(mountinfo, Path::new(path)).unwrap();

        assert_eq!(
            covering("/space/project/workdir").source,
            PathBuf::from("/dev/nvme0n1p7")
        );
        assert_eq!(covering("/dev/shm/project").source, PathBuf::from("tmpfs"));
        assert_eq!(
            covering("/home/someone").source,
            PathBuf::from("/dev/nvme0n1p5"),
            "a path under no nested mount belongs to the root filesystem"
        );
        assert_eq!(
            covering("/spaceship").source,
            PathBuf::from("/dev/nvme0n1p5"),
            "mount points match whole path components, not string prefixes"
        );
    }

    /// A filesystem mounted over another at the same point is the one a lookup reaches, so the
    /// later line wins.
    #[test]
    fn a_mount_over_another_at_the_same_point_wins() {
        let mountinfo = "\
25 30 0:29 / / rw shared:1 - ext4 /dev/sda1 rw
26 25 0:31 / /mnt rw shared:2 - ext4 /dev/sdb1 rw
27 25 0:32 / /mnt rw shared:3 - ext4 /dev/sdc1 rw
";
        assert_eq!(
            mount_covering(mountinfo, Path::new("/mnt/data"))
                .unwrap()
                .source,
            PathBuf::from("/dev/sdc1")
        );
    }

    /// `mountinfo` escapes the four characters that would break its own field separation, and a
    /// mount point that does not decode compares equal to nothing.
    #[test]
    fn mount_point_escapes_are_decoded() {
        let mountinfo = "\
25 30 0:29 / /media/a\\040disk rw shared:1 - ext4 /dev/sdb1 rw
";
        let entry = mount_covering(mountinfo, Path::new("/media/a disk/project")).unwrap();
        assert_eq!(entry.mount_point, PathBuf::from("/media/a disk"));
        assert_eq!(entry.source, PathBuf::from("/dev/sdb1"));
        assert_eq!(unescape_octal("no\\134escape"), PathBuf::from("no\\escape"));
        assert_eq!(
            unescape_octal("\\04"),
            PathBuf::from("\\04"),
            "a truncated escape is not an escape"
        );
    }

    /// A sysfs `dev` attribute is the only place the parent disk's number comes from.
    #[test]
    fn sysfs_dev_attributes_are_read_as_major_minor() {
        assert_eq!(parse_dev_line("259:0\n"), Some((259, 0)));
        assert_eq!(parse_dev_line(""), None);
        assert_eq!(parse_dev_line("259"), None);
    }

    /// The block layer binds an `io.max` entry to a request queue, and only a whole disk carries
    /// one — so whatever device this checkout sits on, the resolved one must not be a partition.
    /// Naming a partition is the failure that makes `enforced` unreachable on an ordinary host.
    #[cfg(target_os = "linux")]
    #[test]
    fn the_resolved_device_is_never_a_partition() {
        let Ok((major, minor)) = io_max_device(Path::new(".")) else {
            // A checkout on a filesystem with no block device behind it has no device to check,
            // and `write_io_max` reports that rather than naming one.
            return;
        };
        let sysfs = PathBuf::from(format!("/sys/dev/block/{major}:{minor}"));
        assert!(
            !sysfs.join("partition").exists(),
            "io.max was given the partition {major}:{minor}, which the block layer refuses"
        );
    }

    /// The diagnostic answers the question by exercising it, and leaves nothing behind whichever
    /// way it answers.
    #[cfg(target_os = "linux")]
    #[test]
    fn probing_io_max_creates_nothing_that_survives_the_call() {
        if skip_without_host_support("probing_io_max_creates_nothing_that_survives_the_call") {
            return;
        }
        let temp = tempfile::tempdir().unwrap();
        let report = probe_io_max(&HostResourceLimits::default(), &temp.path().join("workdir"));
        assert!(
            matches!(
                report.status,
                IoMaxStatus::Enforced | IoMaxStatus::Unavailable
            ),
            "a performed probe must report what it performed, got {report:?}"
        );

        // Counted by name rather than by directory count: this binary's other tests create and
        // remove scopes under the same base concurrently, and only `probe_io_max` names one
        // `explain-scope`.
        let leftovers: Vec<String> = std::fs::read_dir(delegated_base().expect("a delegated base"))
            .unwrap()
            .filter_map(|entry| Some(entry.ok()?.file_name().to_string_lossy().into_owned()))
            .filter(|name| name.contains("explain-scope"))
            .collect();
        assert!(
            leftovers.is_empty(),
            "the throwaway cgroup outlived the probe: {leftovers:?}"
        );
    }

    #[test]
    fn scope_is_required_by_any_route_to_a_native_subprocess() {
        let shell = CapabilityPolicy {
            shell_allow: vec!["bash".to_string()],
            ..CapabilityPolicy::default()
        };
        let spawn = CapabilityPolicy {
            spawn_allow: vec!["child".to_string()],
            ..CapabilityPolicy::default()
        };
        let install_tool = CapabilityPolicy {
            install_tool: vec!["local/tool-*".to_string()],
            ..CapabilityPolicy::default()
        };
        let install_skill = CapabilityPolicy {
            install_skill: vec!["local/skill-*".to_string()],
            ..CapabilityPolicy::default()
        };
        let bare = CapabilityPolicy::default();

        assert!(requires_process_bounding(&shell, false));
        assert!(requires_process_bounding(&spawn, false));
        assert!(
            requires_process_bounding(&install_tool, false),
            "a tool the capsule may pull can be native"
        );
        assert!(
            !requires_process_bounding(&install_skill, false),
            "a pulled skill gains no grant and runs no subprocess"
        );
        for policy in [&shell, &spawn, &install_tool, &install_skill, &bare] {
            assert_eq!(
                requires_process_bounding(policy, false),
                crate::runtime::has_subprocess_route_at_staging(policy),
                "{policy:?}"
            );
        }
        assert!(
            requires_process_bounding(&bare, true),
            "a native artifact alone is enough"
        );
        assert!(
            !requires_process_bounding(&bare, false),
            "a WASM-only capsule has no process tree to bound and must not be blocked"
        );
    }

    /// A capsule with no subprocess capability never needs delegation — on any platform, and
    /// notably including Linux, where a blanket requirement would be a disproportionate
    /// regression for WASM-only capsules.
    #[test]
    fn prepare_scope_returns_none_when_bounding_is_not_required() {
        let temp = tempfile::tempdir().unwrap();
        let scope = prepare_scope(
            false,
            &HostResourceLimits::default(),
            "ses_test",
            temp.path(),
        )
        .expect("a capsule with no subprocess capability must never be refused");
        assert!(scope.scope.is_none());
        // Not a shortfall and never a warning: there is no process tree for a ceiling to apply to.
        assert_eq!(scope.io_max.status, IoMaxStatus::NotRequired);
        assert!(
            scope.io_max.reason.is_some(),
            "not-required must say which cause"
        );
        assert_eq!(io_max_warning(&scope.io_max), None);
    }

    /// macOS can never have a cgroup, so a required scope there must be a silent `None` rather
    /// than a launch-blocking error.
    #[cfg(not(target_os = "linux"))]
    #[test]
    fn prepare_scope_is_never_an_error_off_linux() {
        let temp = tempfile::tempdir().unwrap();
        let scope = prepare_scope(
            true,
            &HostResourceLimits::default(),
            "ses_test",
            temp.path(),
        )
        .expect("a non-Linux host must not refuse to launch for a missing cgroup");
        assert!(scope.scope.is_none());
        assert_eq!(scope.io_max.status, IoMaxStatus::NotRequired);
        assert!(scope
            .io_max
            .reason
            .as_deref()
            .expect("not-required off Linux must say cgroups cannot exist here")
            .contains("Linux"));
    }

    /// The point of asking systemd for a scope: a Linux host with a working user session hands
    /// out a bounded scope no matter what this process's *inherited* cgroup looked like — an
    /// interactive shell's cgroup full of co-resident processes, or the one `cargo test` keeps
    /// itself resident in while running this very binary.
    #[cfg(target_os = "linux")]
    #[test]
    #[allow(clippy::print_stdout)]
    fn a_required_scope_is_created_whatever_cgroup_this_process_inherited() {
        if skip_without_host_support(
            "a_required_scope_is_created_whatever_cgroup_this_process_inherited",
        ) {
            return;
        }
        let temp = tempfile::tempdir().unwrap();
        let prepared = prepare_scope(
            true,
            &HostResourceLimits::default(),
            "ses_selftest",
            temp.path(),
        )
        .expect("a Linux host with a systemd user session must be able to bound a subprocess tree");
        let io_max = prepared.io_max.clone();
        let scope = prepared
            .scope
            .expect("a required scope on Linux is never `None`");

        for file in ["memory.max", "pids.max", "cpu.max"] {
            let contents = std::fs::read_to_string(scope.path().join(file))
                .unwrap_or_else(|error| panic!("{file} unreadable in the scope: {error}"));
            assert!(!contents.trim().is_empty(), "{file} was left unset");
            assert_ne!(contents.trim(), "max", "{file} carries no ceiling");
        }
        match std::fs::read_to_string(scope.path().join("memory.swap.max")) {
            Ok(contents) => {
                assert_eq!(
                    contents.trim(),
                    SCOPE_SWAP_MAX,
                    "memory.swap.max was not bounded"
                );
                assert_eq!(contents.trim(), "0");
                println!(
                    "memory.swap.max in the created scope reads {}",
                    contents.trim()
                );
            }
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
                assert_eq!(probe_swap_control(), SwapControl::Absent);
                println!("this kernel exposes no memory.swap.max; swap is not bounded here");
            }
            Err(error) => panic!("memory.swap.max unreadable in the scope: {error}"),
        }

        // Whatever the host did with `io.max`, the scope says which of the two it was, and never
        // `NotProbed`: a scope that exists has had the write attempted against it.
        assert!(
            matches!(
                io_max.status,
                IoMaxStatus::Enforced | IoMaxStatus::Unavailable
            ),
            "a created scope must report a performed io.max write, got {io_max:?}"
        );
        assert_eq!(
            io_max.declared_bytes_per_sec,
            HostResourceLimits::default().cgroup_io_bytes_per_sec
        );
        if io_max.status == IoMaxStatus::Unavailable {
            assert!(io_max.reason.is_some(), "unavailable must carry a reason");
            assert!(io_max_warning(&io_max).is_some());
        } else {
            assert_eq!(io_max.reason, None);
            assert_eq!(io_max_warning(&io_max), None);
        }
    }

    /// On a kernel with no `cgroup.kill`, killing a scope freezes it, signals every member and
    /// thaws it until it is empty — reaching a `setsid` process and a grandchild as surely as the
    /// one-write kill does.
    #[cfg(target_os = "linux")]
    #[test]
    #[allow(clippy::print_stdout)]
    fn killing_a_scope_without_cgroup_kill_empties_it() {
        if skip_without_host_support("killing_a_scope_without_cgroup_kill_empties_it") {
            return;
        }
        let temp = tempfile::tempdir().unwrap();
        let scope = prepare_scope(
            true,
            &HostResourceLimits::default(),
            "ses_kill_fallback",
            temp.path(),
        )
        .expect("a delegated scope")
        .scope
        .expect("a required scope on Linux is never None");
        let enforcement = crate::sandbox::ShellEnforcement {
            cgroup_scope: Some(Arc::clone(&scope)),
            ..crate::sandbox::ShellEnforcement::environment_only().with_test_session_dirs()
        };
        let mut command = std::process::Command::new("bash");
        command
            .args([
                "-c",
                "setsid sleep 300 >/dev/null 2>&1 & sleep 300 >/dev/null 2>&1 & wait",
            ])
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null());
        crate::sandbox::attach_process_limits(&mut command, &enforcement);
        let mut child = command.spawn().expect("bash spawns into the scope");

        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
        while scope.member_pids().len() < 3 {
            assert!(
                std::time::Instant::now() < deadline,
                "the scope never held bash and both sleepers: {:?}",
                scope.member_pids()
            );
            std::thread::sleep(std::time::Duration::from_millis(20));
        }
        let members = scope.member_pids();

        FORCE_KILL_FALLBACK.with(|flag| flag.set(true));
        let killed = scope.kill_all();
        FORCE_KILL_FALLBACK.with(|flag| flag.set(false));
        let emptied = scope.wait_until_empty(std::time::Duration::from_secs(2));
        let status = child.wait().expect("bash is reaped");
        println!(
            "members before: {members:?}; fallback kill: {killed:?}; empty: {emptied}; \
             cgroup.procs after: {:?}; bash: {status}",
            scope.member_pids()
        );
        killed.expect("the fallback kill succeeds");
        assert!(emptied, "the scope still holds {:?}", scope.member_pids());
        assert_eq!(
            std::os::unix::process::ExitStatusExt::signal(&status),
            Some(libc::SIGKILL)
        );
    }

    /// A required scope holding a `bash` that started a `setsid` sleeper and a plain one, once all
    /// three are in it. `fallback` makes the scope's watcher use the freeze-and-signal loop.
    #[cfg(target_os = "linux")]
    fn scope_with_sleepers(
        session: &str,
        fallback: bool,
    ) -> (Arc<CgroupScope>, std::process::Child) {
        let temp = tempfile::tempdir().unwrap();
        FORCE_KILL_FALLBACK.with(|flag| flag.set(fallback));
        let scope = prepare_scope(true, &HostResourceLimits::default(), session, temp.path());
        FORCE_KILL_FALLBACK.with(|flag| flag.set(false));
        let scope = scope
            .expect("a delegated scope")
            .scope
            .expect("a required scope on Linux is never None");
        let enforcement = crate::sandbox::ShellEnforcement {
            cgroup_scope: Some(Arc::clone(&scope)),
            ..crate::sandbox::ShellEnforcement::environment_only().with_test_session_dirs()
        };
        let mut command = std::process::Command::new("bash");
        command
            .args([
                "-c",
                "setsid sleep 300 >/dev/null 2>&1 & sleep 300 >/dev/null 2>&1 & wait",
            ])
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null());
        crate::sandbox::attach_process_limits(&mut command, &enforcement);
        let child = command.spawn().expect("bash spawns into the scope");
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
        while scope.member_pids().len() < 3 {
            assert!(
                std::time::Instant::now() < deadline,
                "the scope never held bash and both sleepers: {:?}",
                scope.member_pids()
            );
            std::thread::sleep(std::time::Duration::from_millis(20));
        }
        (scope, child)
    }

    #[cfg(target_os = "linux")]
    fn watcher_of(scope: &CgroupScope) -> &ScopeWatcher {
        scope
            .watcher
            .as_ref()
            .expect("a scope on Linux has a watcher")
    }

    /// Closing the lifeline is what the kernel does when the runtime dies: the watcher empties the
    /// scope with no `Drop`, no `kill_all` and no call from the runtime.
    #[cfg(target_os = "linux")]
    fn assert_a_cut_lifeline_empties_the_scope(test_name: &str, fallback: bool) {
        if skip_without_host_support(test_name) {
            return;
        }
        let (scope, mut child) = scope_with_sleepers(test_name, fallback);
        let members = scope.member_pids();
        watcher_of(&scope).cut_lifeline();
        assert!(
            scope.wait_until_empty(std::time::Duration::from_secs(5)),
            "the watcher left {:?} of {members:?} in the scope",
            scope.member_pids()
        );
        for pid in members {
            assert!(
                !Path::new(&format!("/proc/{pid}")).exists()
                    || std::fs::read_to_string(format!("/proc/{pid}/stat"))
                        .is_ok_and(|stat| stat.contains(") Z ")),
                "{pid} is still running"
            );
        }
        let status = child.wait().expect("bash is reaped");
        assert_eq!(
            std::os::unix::process::ExitStatusExt::signal(&status),
            Some(libc::SIGKILL)
        );
    }

    #[cfg(target_os = "linux")]
    #[test]
    fn scope_watcher_kills_the_scope_through_cgroup_kill_when_the_lifeline_closes() {
        assert_a_cut_lifeline_empties_the_scope(
            "scope_watcher_kills_the_scope_through_cgroup_kill_when_the_lifeline_closes",
            false,
        );
    }

    #[cfg(target_os = "linux")]
    #[test]
    fn scope_watcher_kills_the_scope_by_freezing_when_the_lifeline_closes() {
        assert_a_cut_lifeline_empties_the_scope(
            "scope_watcher_kills_the_scope_by_freezing_when_the_lifeline_closes",
            true,
        );
    }

    /// The watcher sits outside the scope it kills, in a process group of its own, holding
    /// `/dev/null` on 0–2, its lifeline's read end and the scope's control files — none of this
    /// process's descriptors.
    #[cfg(target_os = "linux")]
    #[test]
    #[allow(unsafe_code)]
    fn scope_watcher_holds_only_its_own_descriptors_outside_the_scope() {
        if skip_without_host_support(
            "scope_watcher_holds_only_its_own_descriptors_outside_the_scope",
        ) {
            return;
        }
        // A descriptor of this process's own that the watcher must not hold.
        let own = tempfile::tempfile().unwrap();
        let own_target = std::fs::read_link(format!(
            "/proc/self/fd/{}",
            std::os::fd::AsRawFd::as_raw_fd(&own)
        ))
        .unwrap();
        let (scope, mut child) = scope_with_sleepers("ses_watcher_fds", false);
        let pid = watcher_of(&scope).pid;

        let fd_dir = format!("/proc/{pid}/fd");
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
        let targets = loop {
            let targets: Vec<(String, PathBuf)> = std::fs::read_dir(&fd_dir)
                .unwrap()
                .filter_map(Result::ok)
                .filter_map(|entry| {
                    let target = std::fs::read_link(entry.path()).ok()?;
                    Some((entry.file_name().to_string_lossy().into_owned(), target))
                })
                .collect();
            let settled = targets.iter().all(|(_, target)| {
                target == Path::new("/dev/null")
                    || target.to_string_lossy().starts_with("pipe:")
                    || target.starts_with(scope.path())
            });
            if settled || std::time::Instant::now() >= deadline {
                break targets;
            }
            std::thread::sleep(std::time::Duration::from_millis(20));
        };
        let mut numbers: Vec<u32> = targets
            .iter()
            .map(|(name, _)| name.parse().unwrap())
            .collect();
        numbers.sort_unstable();
        assert!(numbers.starts_with(&[0, 1, 2]), "{targets:?}");
        for (name, target) in &targets {
            assert_ne!(target, &own_target, "fd {name}");
            if ["0", "1", "2"].contains(&name.as_str()) {
                assert_eq!(target, Path::new("/dev/null"), "fd {name}");
            } else {
                assert!(
                    target.to_string_lossy().starts_with("pipe:")
                        || target.starts_with(scope.path()),
                    "fd {name} -> {}",
                    target.display()
                );
            }
        }
        let pipes = targets
            .iter()
            .filter(|(_, target)| target.to_string_lossy().starts_with("pipe:"))
            .count();
        assert_eq!(pipes, 1, "the lifeline is its only pipe: {targets:?}");

        assert!(!scope.member_pids().contains(&pid));
        // SAFETY: `getpgid` takes an integer and dereferences nothing.
        assert_eq!(
            unsafe { libc::getpgid(pid) },
            pid,
            "a process group of its own"
        );

        scope.kill_all().unwrap();
        child.wait().unwrap();
    }

    /// Dropping the scope reaps its watcher: a second wait finds no such child.
    #[cfg(target_os = "linux")]
    #[test]
    #[allow(unsafe_code)]
    fn scope_watcher_is_reaped_when_the_scope_drops() {
        if skip_without_host_support("scope_watcher_is_reaped_when_the_scope_drops") {
            return;
        }
        let (scope, mut child) = scope_with_sleepers("ses_watcher_reaped", false);
        let pid = watcher_of(&scope).pid;
        let path = scope.path().to_path_buf();
        drop(Arc::try_unwrap(scope).expect("the test holds the only reference"));
        child.wait().unwrap();
        let mut status = 0;
        // SAFETY: `pid` is an integer and `status` a live local.
        let waited = unsafe { libc::waitpid(pid, &mut status, libc::WNOHANG) };
        assert_eq!(waited, -1);
        assert_eq!(
            std::io::Error::last_os_error().raw_os_error(),
            Some(libc::ECHILD)
        );
        assert!(!path.exists(), "the scope directory is removed");
    }

    /// Swap is bounded at zero whatever the manifest declares, beside a `memory.max` that is the
    /// declared ceiling itself.
    #[test]
    fn the_scope_limits_bound_swap_at_zero() {
        let declared = HostResourceLimits {
            cgroup_memory_bytes: 268_435_456,
            ..HostResourceLimits::default()
        };
        for limits in [HostResourceLimits::default(), declared] {
            let writes: Vec<(&str, String)> = scope_limits(&limits)
                .into_iter()
                .map(|limit| (limit.file, limit.value))
                .collect();
            assert!(
                writes.contains(&("memory.swap.max", "0".to_string())),
                "writes were {writes:?}"
            );
            assert!(writes.contains(&("memory.max", limits.cgroup_memory_bytes.to_string())));
            assert_eq!(
                writes.iter().map(|(file, _)| *file).collect::<Vec<_>>(),
                ["memory.max", "memory.swap.max", "pids.max", "cpu.max"]
            );
        }
        let absent_ok: Vec<&str> = scope_limits(&HostResourceLimits::default())
            .into_iter()
            .filter(|limit| limit.absent_ok)
            .map(|limit| limit.file)
            .collect();
        assert_eq!(absent_ok, ["memory.swap.max"]);
    }

    #[test]
    fn active_swap_areas_are_the_lines_below_the_header() {
        let header = "Filename\t\t\t\tType\t\tSize\t\tUsed\t\tPriority\n";
        assert_eq!(count_swap_areas(header), 0);
        assert_eq!(count_swap_areas(""), 0);
        assert_eq!(
            count_swap_areas(&format!(
                "{header}/swapfile                               file\t\t33554428\t\t0\t\t-2\n"
            )),
            1
        );
    }

    /// The unit name systemd is asked for stays inside the `murmur-*` convention operators
    /// already use to find this runtime's cgroup directories, and stays unique per request.
    #[cfg(target_os = "linux")]
    #[test]
    fn transient_scope_unit_names_are_murmur_prefixed_and_unique() {
        let first = transient_scope_unit_name();
        let second = transient_scope_unit_name();
        assert!(first.starts_with(&format!("murmur-{}-", std::process::id())));
        assert!(first.ends_with(".scope"));
        assert_ne!(first, second);
    }

    #[test]
    fn unified_cgroup_path_is_read_from_the_zero_hierarchy_line() {
        let v2_only = "0::/user.slice/user-1000.slice/user@1000.service/app.slice/mur.scope\n";
        assert_eq!(
            parse_unified_cgroup_path(v2_only).as_deref(),
            Some("/user.slice/user-1000.slice/user@1000.service/app.slice/mur.scope")
        );

        let hybrid = "12:pids:/user.slice\n11:memory:/user.slice\n0::/user.slice/session.scope\n";
        assert_eq!(
            parse_unified_cgroup_path(hybrid).as_deref(),
            Some("/user.slice/session.scope")
        );

        let v1_only = "12:pids:/user.slice\n11:memory:/user.slice\n";
        assert_eq!(parse_unified_cgroup_path(v1_only), None);
    }

    #[test]
    fn cgroup2_mount_falls_back_to_the_conventional_path() {
        let mounts = "cgroup2 /custom/cgroup cgroup2 rw,nsdelegate 0 0\n";
        assert_eq!(parse_cgroup2_mount(mounts), "/custom/cgroup");
        assert_eq!(
            parse_cgroup2_mount("proc /proc proc rw 0 0\n"),
            "/sys/fs/cgroup"
        );
    }

    #[test]
    fn missing_controllers_reports_only_what_is_absent() {
        assert_eq!(
            missing_controllers("cpu memory", REQUIRED_CONTROLLERS),
            vec!["pids".to_string()]
        );
        assert!(missing_controllers("cpu io memory pids", REQUIRED_CONTROLLERS).is_empty());
        assert_eq!(
            missing_controllers("", REQUIRED_CONTROLLERS),
            vec!["memory".to_string(), "pids".to_string(), "cpu".to_string()]
        );
    }

    #[test]
    fn event_counters_parse_the_named_key_only() {
        let memory_events = "low 0\nhigh 0\nmax 12\noom 3\noom_kill 2\n";
        assert_eq!(parse_event_counter(memory_events, "oom_kill"), 2);
        assert_eq!(parse_event_counter(memory_events, "max"), 12);
        assert_eq!(parse_event_counter("max 7\n", "max"), 7);
        assert_eq!(parse_event_counter("", "oom_kill"), 0);
        assert_eq!(parse_event_counter("oom_kill\n", "oom_kill"), 0);
    }

    /// Attribution keys on the delta, not the absolute counter: a session-cumulative total would
    /// mis-attribute every later call once any earlier one had been OOM-killed.
    #[test]
    fn attribution_uses_the_delta_across_one_call() {
        let before = CgroupEventCounters {
            oom_kill: 1,
            pids_max: 4,
        };
        assert_eq!(before.attribution_since(before), None);
        assert_eq!(
            CgroupEventCounters {
                oom_kill: 2,
                pids_max: 4
            }
            .attribution_since(before),
            Some("cgroup_memory_bytes")
        );
        assert_eq!(
            CgroupEventCounters {
                oom_kill: 1,
                pids_max: 5
            }
            .attribution_since(before),
            Some("cgroup_pids_max")
        );
    }

    /// A scope with no directory behind it: occupancy needs no kernel, and `Drop` skips the
    /// `rmdir` for an empty path.
    fn scope_without_kernel() -> CgroupScope {
        CgroupScope {
            path: PathBuf::new(),
            procs_fd: None,
            occupancy: Arc::default(),
            #[cfg(target_os = "linux")]
            watcher: None,
        }
    }

    #[test]
    fn occupancy_alone_is_exclusive() {
        let scope = scope_without_kernel();
        let call = scope.enter();
        assert!(call.exclusive());
    }

    #[test]
    fn occupancy_entering_beside_another_is_not_exclusive() {
        let scope = scope_without_kernel();
        let first = scope.enter();
        let second = scope.enter();
        assert!(!second.exclusive(), "someone was already inside");
        drop(first);
        assert!(
            !second.exclusive(),
            "the sibling leaving does not clean the window it shared"
        );
    }

    #[test]
    fn occupancy_someone_entering_during_the_window_makes_it_shared() {
        let scope = scope_without_kernel();
        let first = scope.enter();
        assert!(first.exclusive());
        drop(scope.enter());
        assert!(
            !first.exclusive(),
            "a call that entered and left inside this window could have moved the counters"
        );
    }

    #[test]
    fn occupancy_drop_releases_the_scope() {
        let scope = scope_without_kernel();
        drop(scope.enter());
        let next = scope.enter();
        assert!(
            next.exclusive(),
            "an earlier call that has left does not share this window"
        );
    }

    /// The guard is owned: it moves to a demoted command's thread and keeps the scope occupied
    /// there after the scope's borrow has ended.
    #[test]
    fn occupancy_guard_outlives_its_borrow_on_another_thread() {
        let scope = scope_without_kernel();
        let demoted = scope.enter();
        let held = std::thread::spawn(move || demoted);
        let demoted = held.join().unwrap();
        let foreground = scope.enter();
        assert!(!foreground.exclusive());
        assert!(!demoted.exclusive());
    }

    #[test]
    fn scope_dir_name_is_sanitized_and_never_empty() {
        assert!(scope_dir_name("ses_abc123").starts_with("murmur-ses_abc123-"));
        assert!(scope_dir_name("../escape").starts_with("murmur-.._escape-"));
        assert!(!scope_dir_name("../escape").contains('/'));
        assert!(scope_dir_name("").starts_with("murmur-"));
    }

    /// A subprocess inherits its parent's cgroup, so a `mur` spawned by a runtime that already
    /// stepped aside starts life *inside* the supervisor leaf. Reading that as its delegated base
    /// would nest another leaf inside it, one per generation, until nothing is delegated.
    #[test]
    fn delegated_base_ascends_out_of_inherited_supervisor_leaves() {
        let base = PathBuf::from("/sys/fs/cgroup/user.slice/app.slice/mur.scope");

        assert_eq!(ascend_out_of_supervisor_leaves(base.clone()), base);
        assert_eq!(
            ascend_out_of_supervisor_leaves(base.join(SUPERVISOR_LEAF)),
            base
        );
        assert_eq!(
            ascend_out_of_supervisor_leaves(base.join(SUPERVISOR_LEAF).join(SUPERVISOR_LEAF)),
            base,
            "a base already nested by an earlier build must still resolve to the real root"
        );
    }

    /// Two scopes must never share a directory even when handed the same id: `Drop` removes the
    /// scope by path, so a collision would have one session delete the cgroup still bounding
    /// another's subprocess tree.
    #[test]
    fn scope_dir_name_is_unique_per_call_for_one_id() {
        let first = scope_dir_name("p");
        let second = scope_dir_name("p");
        assert_ne!(
            first, second,
            "same plan id must not reuse a scope directory"
        );
    }

    #[test]
    fn pid_is_formatted_without_allocating() {
        let mut buf = [0u8; 24];
        let len = format_pid(4321, &mut buf);
        assert_eq!(&buf[..len], b"4321\n");

        let len = format_pid(7, &mut buf);
        assert_eq!(&buf[..len], b"7\n");
    }

    #[test]
    fn device_numbers_use_the_wide_glibc_encoding() {
        // 8:0 (`/dev/sda`) encodes as 0x800 under the 12+20 bit layout.
        assert_eq!(device_major_minor(0x800), (8, 0));
        // 259:3 (`nvme0n1p3`) needs the wide major field, which the legacy 8+8 layout truncates.
        assert_eq!(device_major_minor((259 << 8) | 3), (259, 3));
    }
}