capsule-runtime 0.3.0

//! 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, so a tree of distinct,
//! short-lived processes that fork and exit faster than the count is observed slips past it,
//! 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.
//!
//! ## 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.
//!
//! ## 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::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;

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

/// 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`]).
    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: &Path, report: &IoMaxReport) {
    if let Some((code, message)) = io_max_warning(report) {
        let link = murmur_artifact::security_warning_link(code);
        eprintln!("[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")]
    {
        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: `capabilities.shell.allow`,
/// `capabilities.spawn.allow`, 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).
pub fn requires_process_bounding(
    policy: &crate::types::CapabilityPolicy,
    has_native_artifact: bool,
) -> bool {
    !policy.shell_allow.is_empty() || !policy.spawn_allow.is_empty() || has_native_artifact
}

impl CgroupScope {
    /// Absolute path of this scope's directory.
    // The scope's own path is read only by this module's tests, which assert the limit files
    // systemd wrote under it; the launch path carries the descriptors it needs instead.
    #[allow(dead_code)]
    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(())
    }

    /// 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 Drop for CgroupScope {
    /// Best-effort `rmdir` of the scope directory.
    ///
    /// A cgroup directory can only be removed once empty, so a lingering grandchild the runtime
    /// never reaped keeps it alive. That is logged and moved past rather than retried in a loop:
    /// blocking session teardown on a process the runtime does not control would turn a leaked
    /// directory into a hung exit.
    fn drop(&mut self) {
        // Drop the descriptor before the rmdir so nothing holds the directory open.
        self.procs_fd = None;
        if self.path.as_os_str().is_empty() {
            return;
        }
        if let Err(error) = std::fs::remove_dir(&self.path) {
            eprintln!(
                "[capsule-runtime] note: could not remove cgroup scope {} ({error}); \
                 it will be reclaimed once its last task exits",
                self.path.display()
            );
        }
    }
}

/// 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. Any failure to establish `memory.max`, `pids.max` or `cpu.max` is fatal — those
    /// three 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.
    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,
        };

        scope.write_limit("memory.max", &limits.cgroup_memory_bytes.to_string())?;
        scope.write_limit("pids.max", &limits.cgroup_pids_max.to_string())?;
        scope.write_limit(
            "cpu.max",
            &format!(
                "{} {CPU_MAX_PERIOD_US}",
                u64::from(limits.cgroup_cpu_percent) * (CPU_MAX_PERIOD_US / 100)
            ),
        )?;
        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()
            )
        })?;

        Ok((
            Self {
                path: scope.take_path(),
                procs_fd: Some(procs_fd),
            },
            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
    }

    fn write_limit(&self, file: &str, value: &str) -> Result<(), String> {
        let path = self.path.join(file);
        std::fs::write(&path, format!("{value}\n"))
            .map_err(|error| format!("could not write {} to {}: {error}", value, 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's blocking API drives its own executor; run it off-thread so it never blocks a
    // caller already inside another async runtime (e.g. tokio) on a nested reactor.
    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::zvariant::Value;

    let connection = zbus::blocking::Connection::session()
        .map_err(|error| format!("no systemd session bus: {error}"))?;

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

    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),
        )
        .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 in 20ms steps 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.
#[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(20));
    }
}

/// 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, `detect_egress_namespace_blocker` 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() {
        eprintln!(
            "[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 \
             docs/content/reference/resource-limits-manual-verification.md"
        );
        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 bare = CapabilityPolicy::default();

        assert!(requires_process_bounding(&shell, false));
        assert!(requires_process_bounding(&spawn, false));
        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]
    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");
        }

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

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

    #[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));
    }
}