capsule-runtime 0.6.0

//! Host-process resource bounds for every native subprocess this runtime spawns.
//!
//! Sibling of [`crate::limits`], and deliberately not a replacement for it: that module bounds
//! a WASM *guest* inside its wasmtime store (linear memory, tables, epoch deadline), this one
//! bounds the *operating-system processes* the runtime forks — `shell::execute_shell` and
//! `runtime::dispatch_native_tool`. A capsule that cannot escape its containment can still
//! wedge the host it runs on: fork bombs, unbounded open files, unbounded writes, unbounded
//! per-process CPU and address space, and unbounded workdir growth are all *denial of service*,
//! not boundary defeats — nothing outside the granted scope is read, written, or reached — but
//! a wedged host is still a broken host.
//!
//! Three mechanisms, in descending order of portability:
//!
//! * **`setrlimit(2)` ceilings** ([`apply_hard_rlimits`]), applied inside the forked child's
//!   `pre_exec` window on every Unix platform. Per-process, POSIX-portable, no configuration
//!   required of the host.
//! * **a cgroup v2 scope** ([`crate::cgroup`]) around the whole subprocess *tree*, Linux only.
//!   This is what rlimits structurally cannot do: `RLIMIT_NPROC` is a **per-uid** ceiling,
//!   checked against every thread the uid owns on the host, so it bounds neither this tree nor
//!   anything else in particular. `pids.max` on a cgroup counts only the scope's own tasks, and
//!   a spawn that joins a scope gets no `RLIMIT_NPROC` from this module at all ([`NprocBound`]).
//! * **a periodic workdir-size check** ([`WorkdirGuard`]), on every platform. A plain filesystem
//!   walk on a fixed cadence, so a breach is caught within one poll interval — not instantly.
//!
//! ## A silent manifest means defaults, never "unlimited"
//!
//! [`HostResourceLimits::resolve`] follows [`crate::limits::ExecutionLimits::resolve`] exactly:
//! the manifest's `capabilities.resources` block with every omitted field replaced by its
//! `DEFAULT_*` constant, and the whole block's absence treated as "every field omitted". There
//! is no "unset" state downstream — a capsule that declares nothing is bounded identically to
//! one that declares every default by hand.
//!
//! ## Hard limits, not soft ones
//!
//! `setrlimit(2)` takes both a soft (`rlim_cur`) and a hard (`rlim_max`) value, and an
//! unprivileged process may raise its own soft limit up to the hard one at any time. Against a
//! hostile agent a soft-only cap is therefore advisory: one `setrlimit` call from inside the
//! capsule undoes it. Every limit this module sets writes `rlim_cur == rlim_max`, leaving no
//! ceiling to raise into, and lowering `rlim_max` is irreversible for an unprivileged process.

use std::{
    path::{Path, PathBuf},
    sync::{
        atomic::{AtomicBool, Ordering},
        Arc, Mutex,
    },
    thread,
    time::Duration,
};

/// Default `RLIMIT_NPROC` headroom for a spawn with no cgroup scope — how much past the uid's
/// count at that spawn a subprocess tree may add, in whichever unit this platform's
/// `RLIMIT_NPROC` is checked against. Unused for a spawn that joins a scope. See [`NprocBound`]
/// for which spawns get it and [`uid_task_count`] for the per-platform unit.
pub const DEFAULT_MAX_PROCESSES: u64 = 128;

/// Default `RLIMIT_NOFILE` hard ceiling — open descriptors per spawned subprocess.
pub const DEFAULT_MAX_OPEN_FILES: u64 = 1024;

/// Default `RLIMIT_FSIZE` hard ceiling — 4 GiB, the largest single file a subprocess may write.
pub const DEFAULT_MAX_FILE_SIZE_BYTES: u64 = 4 * 1024 * 1024 * 1024;

/// Default `RLIMIT_CPU` hard ceiling — one hour of CPU time per spawned subprocess.
pub const DEFAULT_CPU_SECONDS: u64 = 3600;

/// Default `RLIMIT_AS` (Linux) / `RLIMIT_DATA` (macOS) hard ceiling — 2 GiB per subprocess.
pub const DEFAULT_MEMORY_BYTES: u64 = 2 * 1024 * 1024 * 1024;

/// Default cgroup `memory.max` — 4 GiB aggregate across the whole subprocess tree. Above the
/// per-process ceiling on purpose: the tree bound exists to catch what *many* processes do
/// together, so setting it at or below the single-process bound would make it redundant.
pub const DEFAULT_CGROUP_MEMORY_BYTES: u64 = 4 * 1024 * 1024 * 1024;

/// Default cgroup `pids.max` — 256 tasks aggregate across the whole subprocess tree.
pub const DEFAULT_CGROUP_PIDS_MAX: u64 = 256;

/// Default cgroup `cpu.max` quota as a percentage of one core: 200 = two cores' worth.
pub const DEFAULT_CGROUP_CPU_PERCENT: u32 = 200;

/// Default cgroup `io.max` read+write throughput on the workdir's backing device — 100 MiB/s.
pub const DEFAULT_CGROUP_IO_BYTES_PER_SEC: u64 = 100 * 1024 * 1024;

/// Default ceiling on total workdir size — 10 GiB, checked periodically.
pub const DEFAULT_WORKDIR_MAX_BYTES: u64 = 10 * 1024 * 1024 * 1024;

/// How often [`WorkdirGuard`] walks the workdir tree.
///
/// An internal constant rather than a manifest key, matching [`crate::limits::EPOCH_TICK_INTERVAL`]:
/// the cadence changes *when* a breach is noticed, never *whether* it is, so exposing it would
/// add a knob with no security meaning. The consequence: a disk filler is caught within one
/// interval of crossing the ceiling, not at the instant it crosses.
pub(crate) const WORKDIR_CHECK_INTERVAL: Duration = Duration::from_secs(10);

/// Fully-resolved host-process limits for one session: the manifest's `capabilities.resources`
/// block with every omitted field replaced by its `DEFAULT_*`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct HostResourceLimits {
    /// `RLIMIT_NPROC` headroom above the uid's count at each spawn, applied only to a spawn
    /// with no cgroup scope ([`NprocBound::Headroom`]) and counted in the unit that uid's kernel
    /// enforces against ([`uid_task_count`]). A spawn that joins a scope is bounded by
    /// `cgroup_pids_max` instead and ignores this field.
    pub max_processes: u64,
    /// `RLIMIT_NOFILE` hard ceiling.
    pub max_open_files: u64,
    /// `RLIMIT_FSIZE` hard ceiling, in bytes.
    pub max_file_size_bytes: u64,
    /// `RLIMIT_CPU` hard ceiling, in CPU-seconds.
    pub cpu_seconds: u64,
    /// `RLIMIT_AS` on Linux, `RLIMIT_DATA` on macOS (which has no `RLIMIT_AS`), in bytes.
    pub memory_bytes: u64,
    /// cgroup v2 `memory.max`, in bytes. Linux only.
    pub cgroup_memory_bytes: u64,
    /// cgroup v2 `pids.max`. Linux only.
    pub cgroup_pids_max: u64,
    /// cgroup v2 `cpu.max` quota as a percentage of one core. Linux only.
    pub cgroup_cpu_percent: u32,
    /// cgroup v2 `io.max` rbps+wbps on the workdir's backing device. Linux only, best-effort.
    pub cgroup_io_bytes_per_sec: u64,
    /// Ceiling on total workdir size, in bytes. Every platform.
    pub workdir_max_bytes: u64,
}

impl Default for HostResourceLimits {
    fn default() -> Self {
        Self {
            max_processes: DEFAULT_MAX_PROCESSES,
            max_open_files: DEFAULT_MAX_OPEN_FILES,
            max_file_size_bytes: DEFAULT_MAX_FILE_SIZE_BYTES,
            cpu_seconds: DEFAULT_CPU_SECONDS,
            memory_bytes: DEFAULT_MEMORY_BYTES,
            cgroup_memory_bytes: DEFAULT_CGROUP_MEMORY_BYTES,
            cgroup_pids_max: DEFAULT_CGROUP_PIDS_MAX,
            cgroup_cpu_percent: DEFAULT_CGROUP_CPU_PERCENT,
            cgroup_io_bytes_per_sec: DEFAULT_CGROUP_IO_BYTES_PER_SEC,
            workdir_max_bytes: DEFAULT_WORKDIR_MAX_BYTES,
        }
    }
}

impl HostResourceLimits {
    /// Resolve a manifest `capabilities.resources` block, substituting the default for each
    /// field the manifest left out (and for the whole block when it is absent).
    #[must_use]
    pub fn resolve(declared: Option<&murmur_artifact::ResourceCapabilities>) -> Self {
        let defaults = Self::default();
        let Some(declared) = declared else {
            return defaults;
        };
        Self {
            max_processes: declared.max_processes.unwrap_or(defaults.max_processes),
            max_open_files: declared.max_open_files.unwrap_or(defaults.max_open_files),
            max_file_size_bytes: declared
                .max_file_size_bytes
                .unwrap_or(defaults.max_file_size_bytes),
            cpu_seconds: declared.cpu_seconds.unwrap_or(defaults.cpu_seconds),
            memory_bytes: declared.memory_bytes.unwrap_or(defaults.memory_bytes),
            cgroup_memory_bytes: declared
                .cgroup_memory_bytes
                .unwrap_or(defaults.cgroup_memory_bytes),
            cgroup_pids_max: declared.cgroup_pids_max.unwrap_or(defaults.cgroup_pids_max),
            cgroup_cpu_percent: declared
                .cgroup_cpu_percent
                .unwrap_or(defaults.cgroup_cpu_percent),
            cgroup_io_bytes_per_sec: declared
                .cgroup_io_bytes_per_sec
                .unwrap_or(defaults.cgroup_io_bytes_per_sec),
            workdir_max_bytes: declared
                .workdir_max_bytes
                .unwrap_or(defaults.workdir_max_bytes),
        }
    }
}

/// Resolve a manifest `capabilities.resources` block into [`HostResourceLimits`].
///
/// Free function mirroring `ExecutionLimits::resolve`'s call shape at the one call site that
/// matters (`types::capability_policy_from_runtime_manifest`), so the two sibling limit blocks
/// read identically there.
#[must_use]
pub(crate) fn resolve(
    declared: Option<&murmur_artifact::ResourceCapabilities>,
) -> HostResourceLimits {
    HostResourceLimits::resolve(declared)
}

/// The named limit a kill signal unambiguously attributes to, or `None`.
///
/// Only two signals qualify, and the list is deliberately not longer:
///
/// * `SIGXCPU` — the kernel raises exactly this for an `RLIMIT_CPU` overrun. Unambiguous.
/// * `SIGXFSZ` — likewise for `RLIMIT_FSIZE`. Unambiguous.
///
/// Everything else stays unattributed **on purpose**. `RLIMIT_AS`/`RLIMIT_DATA` overrun surfaces
/// as `ENOMEM` inside the child's own allocator, not as a parent-visible signal with a unique
/// identity, so mapping `SIGSEGV`/`SIGABRT` to `memory_bytes` would be a guess presented as a
/// fact. `RLIMIT_NPROC` and `RLIMIT_NOFILE` do not kill anything at all: they fail a
/// `fork()`/`open()` inside the child with `EAGAIN`/`EMFILE`, visible in that child's own stderr
/// and exit code. A bare `SIGKILL` with no cgroup evidence has too many causes (host OOM killer,
/// an operator, an unrelated crash) to name one.
#[cfg_attr(not(unix), allow(dead_code))]
pub(crate) fn limit_from_signal(signal: i32) -> Option<&'static str> {
    match signal {
        libc::SIGXCPU => Some("cpu_seconds"),
        libc::SIGXFSZ => Some("max_file_size_bytes"),
        _ => None,
    }
}

/// The line that tells the model which `capabilities.resources` limit stopped a shell command,
/// and what to do about it: `resource_limit: <name> — <what happened>. <what to do>.`
///
/// One line, no trailing newline. `limit` is a name a call was attributed to — one of
/// [`limit_from_signal`]'s or [`crate::cgroup::CgroupEventCounters::attribution_since`]'s. A name
/// with no line of its own gets a generic one that still names its `capabilities.resources` field.
///
/// The `cgroup_*` lines say "the capsule … while this command ran" because the cgroup scope is
/// shared by everything the session runs: a counter delta proves the capsule hit the ceiling
/// during the call, not which process did. The signal lines say "this command" because the kernel
/// delivered the signal to that process.
///
/// Every place the model reads an attributed limit renders it through this function.
pub(crate) fn resource_limit_line(limit: &str) -> String {
    let explanation = match limit {
        "cgroup_pids_max" => "the capsule reached its process limit while this command ran, so \
             starting a new process failed (EAGAIN). Retry with fewer parallel processes"
            .to_string(),
        "cgroup_memory_bytes" => "the capsule reached its memory limit while this command ran, \
             and the kernel killed a process in it to free memory. Retry using less memory at \
             once (smaller inputs, fewer parallel jobs)"
            .to_string(),
        "cpu_seconds" => "this command used up its CPU-time limit and was stopped. Retry with \
             less work per command"
            .to_string(),
        "max_file_size_bytes" => "this command tried to write a file larger than the capsule's \
             file-size limit and was stopped. Retry writing smaller files"
            .to_string(),
        other => format!(
            "this command was stopped by the capsule's capabilities.resources.{other} limit. \
             Retry with less work"
        ),
    };
    format!(
        "resource_limit: {limit} — {explanation}, or ask the operator to raise \
         capabilities.resources.{limit}."
    )
}

/// What one spawn's `RLIMIT_NPROC` is, decided in the parent before its `pre_exec` closure is
/// installed — the closure only receives this `Copy` value.
///
/// `RLIMIT_NPROC` is checked against **everything the uid owns on the host**, not against the
/// tasks in this tree, and on Linux the unit it counts is *threads*: `setrlimit(2)` calls it "the
/// maximum number of processes (or, more precisely on Linux, threads) that can be created for the
/// real user ID". On macOS, whose BSD-derived limit is per-process, it counts processes. Either
/// way an interactive account is already deep into that number before the capsule starts, and the
/// number moves with every thread the desktop, the harness or anything else on the host starts or
/// stops. A limit set from a count taken at launch therefore loses headroom to the rest of the
/// host for the whole session, and a `fork()` in the subprocess fails with `EAGAIN` once enough
/// unrelated threads have started.
///
/// Produced by `sandbox::ShellEnforcement::nproc_bound` once per spawn.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum NprocBound {
    /// Leave `RLIMIT_NPROC` at the value the runtime inherited. The child joins a cgroup scope in
    /// the same `pre_exec` closure, fatally on failure, and the scope's `pids.max`
    /// ([`HostResourceLimits::cgroup_pids_max`]) bounds its tree: it counts only the scope's own
    /// tasks, so nothing else on the host can spend it.
    CgroupPidsMax,
    /// `baseline + max_processes`, where `baseline` is [`uid_task_count`] measured in the parent
    /// for this spawn, so `max_processes` means "how much past the host's usage at this spawn this
    /// tree may add". `0` when the count cannot be taken, which applies the declared value
    /// literally and errs toward the tighter bound.
    Headroom { baseline: u64 },
}

/// Sets every [`HostResourceLimits`] rlimit field as a **hard** limit (`rlim_cur == rlim_max`),
/// plus `RLIMIT_CORE = 0`. `RLIMIT_NPROC` is set only for [`NprocBound::Headroom`]; for
/// [`NprocBound::CgroupPidsMax`] no `RLIMIT_NPROC` call is made at all.
///
/// Called from inside a forked child's `pre_exec` closure, so everything here is restricted to
/// async-signal-safe operations: two syscalls per limit and no allocation, no locking, and no
/// error formatting (errors are returned as bare `io::Error::from_raw_os_error`).
///
/// `RLIMIT_CORE` is fixed at zero with no manifest surface at all — a core dump of a capsule
/// subprocess writes a potentially multi-gigabyte file, holding whatever was in that process's
/// memory, into the workdir or the host's core pattern. It belongs to the same category as the
/// fixed capsule device set in `sandbox.rs`: an invariant, not a setting.
///
/// A requested value above the inherited hard limit is **clamped down to it** rather than
/// treated as an error: an unprivileged process cannot raise `rlim_max`, so the inherited value
/// is the real ceiling either way, and failing the spawn would replace an unwidenable bound with
/// no subprocess at all.
///
/// ## `RLIMIT_NPROC`: headroom without a scope, untouched with one
///
/// A per-uid count cannot be turned into a bound on one tree, only into headroom above whatever
/// the uid owns. Without a cgroup scope that headroom is the only process bound there is, so it
/// is set from a baseline measured in the unit the kernel checks: a *process* count fed to a
/// Linux kernel counting threads yields a ceiling below the uid's live usage, which is the same
/// `EAGAIN` the headroom exists to prevent.
///
/// With a scope, `pids.max` bounds the tree exactly and headroom adds nothing but a way for an
/// unrelated thread elsewhere on the host to fail a `fork()` here, so the inherited value is left
/// in place.
#[cfg(unix)]
#[allow(unsafe_code)]
// The `as u32` on each `RLIMIT_*` below is redundant on Linux, where the constants are already
// `__rlimit_resource_t` (`c_uint`), and load-bearing on macOS, where they are `c_int`. Clippy only
// ever sees one of the two platforms at a time, so on Linux it reads the portability cast as a
// no-op.
#[allow(clippy::unnecessary_cast)]
pub(crate) fn apply_hard_rlimits(
    limits: &HostResourceLimits,
    nproc: NprocBound,
) -> std::io::Result<()> {
    if let NprocBound::Headroom { baseline } = nproc {
        set_hard_rlimit(
            libc::RLIMIT_NPROC as u32,
            baseline.saturating_add(limits.max_processes),
        )?;
    }
    set_hard_rlimit(libc::RLIMIT_NOFILE as u32, limits.max_open_files)?;
    set_hard_rlimit(libc::RLIMIT_FSIZE as u32, limits.max_file_size_bytes)?;
    set_hard_rlimit(libc::RLIMIT_CPU as u32, limits.cpu_seconds)?;
    set_hard_rlimit(libc::RLIMIT_CORE as u32, 0)?;

    // `memory_bytes` maps to `RLIMIT_AS` (total address space) on Linux, where it is enforced
    // and a failure to set it is fatal.
    #[cfg(target_os = "linux")]
    set_hard_rlimit(libc::RLIMIT_AS as u32, limits.memory_bytes)?;

    // macOS has no `RLIMIT_AS`, and its `RLIMIT_DATA` is present in the headers but not
    // enforceable: the kernel rejects any finite value with `EINVAL`. The call is still made (a
    // BSD that does honour it gets the bound) but its failure is tolerated rather than turned
    // into a failed spawn — refusing to run any subprocess at all on the primary development
    // platform, over a knob that platform's kernel will never implement, would trade a missing
    // bound for a broken runtime. The residual gap is exactly what `W-SEC-010` documents: on a
    // host with no cgroups, memory is bounded by neither an aggregate nor a per-process ceiling.
    #[cfg(not(target_os = "linux"))]
    let _ = set_hard_rlimit(libc::RLIMIT_DATA as u32, limits.memory_bytes);

    Ok(())
}

/// One `getrlimit`/`setrlimit` pair: read the inherited hard limit, clamp `value` to it, then
/// write the result to **both** the soft and hard slot.
#[cfg(unix)]
#[allow(unsafe_code)]
fn set_hard_rlimit(resource: u32, value: u64) -> std::io::Result<()> {
    let mut current = libc::rlimit {
        rlim_cur: 0,
        rlim_max: 0,
    };
    // SAFETY: `resource` is one of the `libc::RLIMIT_*` constants (cast to the width this
    // platform's `setrlimit` takes) and `current` is a live, correctly-sized stack `rlimit`.
    if unsafe { libc::getrlimit(resource as _, &mut current) } != 0 {
        return Err(std::io::Error::last_os_error());
    }

    // `rlim_t` is 64-bit on every platform this runtime targets, but saturate rather than cast
    // so a 32-bit `rlim_t` would clamp to its own maximum instead of silently wrapping.
    let requested = libc::rlim_t::try_from(value).unwrap_or(libc::rlim_t::MAX);
    let target = if current.rlim_max == libc::RLIM_INFINITY {
        requested
    } else {
        requested.min(current.rlim_max)
    };

    let limit = libc::rlimit {
        rlim_cur: target,
        rlim_max: target,
    };
    // SAFETY: same as above; `limit` is a live stack `rlimit` and is only read by the call.
    if unsafe { libc::setrlimit(resource as _, &limit) } != 0 {
        return Err(std::io::Error::last_os_error());
    }
    Ok(())
}

/// How many **tasks** — in the unit this platform's `RLIMIT_NPROC` is actually checked against —
/// the runtime's own uid currently owns, or `None` if the host cannot be asked.
///
/// The unit differs per platform, and matching it is the entire point of this function:
///
///   - **Linux**: threads. `setrlimit(2)` states the limit is "the maximum number of processes
///     (or, more precisely on Linux, threads) that can be created for the real user ID", and the
///     kernel's `fork()` path compares against the uid's live task count. So this sums each owned
///     process's `/proc/<pid>/task/` entries (one per thread, the leader included) rather than
///     counting `/proc/<pid>` directories, which would count one per *process* and land an order
///     of magnitude low on any account running threaded software.
///   - **macOS**: processes. The BSD-derived `RLIMIT_NPROC` there genuinely counts one entry per
///     process, so `proc_listpids` is already the matching unit.
///
/// Measured **in the parent** once per spawn that has no cgroup scope — it walks `/proc` (Linux)
/// or calls into libproc (macOS), neither of which is async-signal-safe, so it must never be
/// called from `pre_exec`. Its one consumer is [`NprocBound::Headroom`]; see there for why a
/// per-uid limit is meaningless without it.
///
/// The Linux walk is inherently racy — a process can exit between the `read_dir("/proc")` and the
/// read of its `task/` directory — and every such race is absorbed by skipping that process. A
/// baseline slightly stale in either direction is harmless: it is added to `max_processes`
/// headroom, not compared for equality against anything.
#[cfg_attr(not(unix), allow(dead_code))]
// Each `cfg` arm returns explicitly so that exactly one of them is live per target without the
// arms having to be arranged so the live one happens to be last — on the target where an arm is
// the final live statement, clippy sees its `return` as redundant, but removing it breaks every
// other target.
#[allow(clippy::needless_return)]
pub(crate) fn uid_task_count() -> Option<u64> {
    #[cfg(target_os = "linux")]
    {
        // SAFETY: `geteuid` takes no arguments and cannot fail.
        let uid = {
            #[allow(unsafe_code)]
            unsafe {
                libc::geteuid()
            }
        };
        return uid_task_count_in(Path::new("/proc"), uid);
    }

    #[cfg(target_os = "macos")]
    {
        /// `PROC_UID_ONLY` from `<libproc.h>`.
        const PROC_UID_ONLY: u32 = 4;

        // Declared here rather than taken from the `libc` crate, which does not expose libproc on
        // this target. The symbol lives in libSystem, which every macOS binary already links, so
        // this needs no build-script or link attribute. Called with a null buffer, it returns the
        // *byte size* a full listing would need — the count without the listing.
        extern "C" {
            fn proc_listpids(
                r#type: u32,
                typeinfo: u32,
                buffer: *mut libc::c_void,
                buffersize: libc::c_int,
            ) -> libc::c_int;
        }

        // SAFETY: the null-buffer/zero-size form is the documented size query; it writes nothing.
        #[allow(unsafe_code)]
        let bytes = unsafe {
            let uid = libc::geteuid();
            proc_listpids(PROC_UID_ONLY, uid, std::ptr::null_mut(), 0)
        };
        if bytes <= 0 {
            return None;
        }
        return Some(bytes as u64 / std::mem::size_of::<u32>() as u64);
    }

    #[cfg(not(any(target_os = "linux", target_os = "macos")))]
    None
}

/// The Linux walk behind [`uid_task_count`], over a `/proc`-shaped tree at `proc_root`: the
/// `task/` entries of every all-digit directory owned by `uid`, summed.
///
/// `None` only when `proc_root` itself cannot be read. A process directory whose `task/` cannot be
/// read — one that exited mid-walk — contributes 0.
#[cfg(target_os = "linux")]
pub(crate) fn uid_task_count_in(proc_root: &Path, uid: u32) -> Option<u64> {
    use std::os::unix::fs::MetadataExt;

    // A `/proc/<pid>` directory is owned by the uid the process runs as, so a `stat` per entry
    // finds this uid's processes without opening or parsing a single `status` file.
    let entries = std::fs::read_dir(proc_root).ok()?;
    let count = entries
        .flatten()
        .filter(|entry| {
            entry
                .file_name()
                .to_str()
                .is_some_and(|name| name.bytes().all(|b| b.is_ascii_digit()))
                && entry.metadata().is_ok_and(|metadata| metadata.uid() == uid)
        })
        .map(|entry| {
            // One `task/` entry per thread of that process, the group leader included.
            std::fs::read_dir(entry.path().join("task"))
                .map(|threads| threads.flatten().count() as u64)
                .unwrap_or(0)
        })
        .sum();
    Some(count)
}

/// A workdir-size ceiling that was crossed: what was allowed, and what was actually observed.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct WorkdirBreach {
    pub(crate) max_bytes: u64,
    pub(crate) observed_bytes: u64,
}

/// Periodic workdir-size check for one session.
///
/// Walks the workdir tree every [`WORKDIR_CHECK_INTERVAL`] on a plain OS thread (not a tokio
/// task — the session runs guests on several runtimes and the guard must outlive all of them),
/// and latches the first breach it sees. Three consumers read that latch:
///
///   * `shell::execute_shell` and `runtime::dispatch_native_tool` refuse to spawn once it is
///     set, which stops a disk filler from writing another byte;
///   * the agent turn loop turns it into [`crate::errors::RuntimeError::WorkdirSizeExceeded`],
///     which terminates the session.
///
/// **Why a periodic check and not a tmpfs-backed workdir with a size mount option.** The
/// structural version needs a mount namespace the runtime does not have: nothing in this
/// codebase's process model ever runs as root, and the `sealed-containment-runtime` work that
/// would create one is separate and unbuilt. A poll is what is available without it, at the cost
/// of a one-interval detection lag.
#[derive(Debug)]
pub(crate) struct WorkdirGuard {
    max_bytes: u64,
    breach: Mutex<Option<WorkdirBreach>>,
    stop: AtomicBool,
}

impl WorkdirGuard {
    /// Spawn the checker thread for `workdir`. The returned handle owns the latch; dropping the
    /// last clone stops the thread within one interval.
    pub(crate) fn spawn(workdir: &Path, max_bytes: u64) -> Arc<Self> {
        let guard = Arc::new(Self {
            max_bytes,
            breach: Mutex::new(None),
            stop: AtomicBool::new(false),
        });

        // The thread holds a Weak, so it can never keep the session's guard alive; the stop flag
        // covers the reverse (guard dropped while the thread is mid-sleep). Same both-ends shape
        // as `limits::EpochTicker`.
        let weak = Arc::downgrade(&guard);
        let workdir = workdir.to_path_buf();
        thread::spawn(move || loop {
            thread::sleep(WORKDIR_CHECK_INTERVAL);
            let Some(guard) = weak.upgrade() else {
                break;
            };
            if guard.stop.load(Ordering::Relaxed) || guard.breach().is_some() {
                break;
            }
            let observed = directory_size_bytes(&workdir);
            if observed > guard.max_bytes {
                guard.record_breach(&workdir, observed);
                break;
            }
        });

        guard
    }

    /// The latched breach, if the workdir has crossed its ceiling.
    pub(crate) fn breach(&self) -> Option<WorkdirBreach> {
        self.breach.lock().ok().and_then(|breach| *breach)
    }

    fn record_breach(&self, workdir: &Path, observed_bytes: u64) {
        let breach = WorkdirBreach {
            max_bytes: self.max_bytes,
            observed_bytes,
        };
        if let Ok(mut slot) = self.breach.lock() {
            if slot.is_none() {
                *slot = Some(breach);
            }
        }
        let message = workdir_breach_message(breach);
        crate::runtime_err!("[capsule-runtime] {message}");
        crate::agent::append_bootstrap_log(workdir, &format!("[resources] {message}"));
    }
}

impl Drop for WorkdirGuard {
    fn drop(&mut self) {
        self.stop.store(true, Ordering::Relaxed);
    }
}

/// The one wording for a workdir breach, shared by the guard's log line and by the error the
/// subprocess spawn paths return, so the two cannot drift.
pub(crate) fn workdir_breach_message(breach: WorkdirBreach) -> String {
    format!(
        "workdir grew to {} bytes, past the {} byte ceiling \
         (capabilities.resources.workdir_max_bytes)",
        breach.observed_bytes, breach.max_bytes
    )
}

/// Total size of every regular file beneath `root`, in bytes.
///
/// Iterative rather than recursive, and reads `symlink_metadata` so a symlink counts as its own
/// (tiny) entry and is never followed — a symlinked directory would otherwise let a capsule both
/// hide growth outside the workdir and spin this walk on a cycle. Unreadable entries are skipped:
/// this is a best-effort accounting of growth, and a partial read that under-counts is strictly
/// better than a walk that gives up entirely.
pub(crate) fn directory_size_bytes(root: &Path) -> u64 {
    let mut total: u64 = 0;
    let mut stack: Vec<PathBuf> = vec![root.to_path_buf()];

    while let Some(dir) = stack.pop() {
        let Ok(entries) = std::fs::read_dir(&dir) else {
            continue;
        };
        for entry in entries.flatten() {
            let Ok(metadata) = entry.path().symlink_metadata() else {
                continue;
            };
            if metadata.is_dir() {
                stack.push(entry.path());
            } else {
                total = total.saturating_add(metadata.len());
            }
        }
    }

    total
}

#[cfg(all(test, target_os = "linux"))]
pub(crate) mod test_support {
    use std::fs::{File, OpenOptions};
    use std::sync::{Arc, Barrier};

    /// Serializes the tests that start [`HeldThreads`] or assert on how the uid's task count
    /// moves, so one test's holders never land inside another's before/after window. Held until
    /// the returned file is dropped. Other processes the uid owns still move the count; the
    /// margins in those tests absorb that.
    ///
    /// An exclusive lock on a file in the temp directory, not a static mutex: cargo-nextest runs
    /// every test in its own process, where a static serializes nothing. Each call opens its own
    /// file description, so the lock also excludes other threads of the same `cargo test` binary.
    pub(crate) fn host_threads_lock() -> File {
        let path = std::env::temp_dir().join("murmur-capsule-runtime-host-threads.lock");
        let file = OpenOptions::new()
            .create(true)
            .truncate(false)
            .write(true)
            .open(&path)
            .unwrap_or_else(|error| panic!("opening {}: {error}", path.display()));
        file.lock()
            .unwrap_or_else(|error| panic!("locking {}: {error}", path.display()));
        file
    }

    /// `count` threads in the test process, each parked on a barrier, standing in for threads the
    /// rest of the host starts while a session runs. [`Self::start`] returns only once every
    /// one of them is running, so the uid's task count already includes them; dropping the value
    /// releases and joins them.
    pub(crate) struct HeldThreads {
        release: Arc<Barrier>,
        handles: Vec<std::thread::JoinHandle<()>>,
    }

    impl HeldThreads {
        pub(crate) fn start(count: usize) -> Self {
            let started = Arc::new(Barrier::new(count + 1));
            let release = Arc::new(Barrier::new(count + 1));
            let handles = (0..count)
                .map(|_| {
                    let started = Arc::clone(&started);
                    let release = Arc::clone(&release);
                    std::thread::Builder::new()
                        .stack_size(64 * 1024)
                        .spawn(move || {
                            started.wait();
                            release.wait();
                        })
                        .expect("a holder thread must start")
                })
                .collect();
            started.wait();
            Self { release, handles }
        }
    }

    impl Drop for HeldThreads {
        fn drop(&mut self) {
            self.release.wait();
            for handle in self.handles.drain(..) {
                let _ = handle.join();
            }
        }
    }

    /// This process's own `(soft, hard)` `RLIMIT_NPROC`, the values a child inherits when
    /// `apply_hard_rlimits` leaves the limit alone.
    #[allow(unsafe_code)]
    pub(crate) fn own_rlimit_nproc() -> (libc::rlim_t, libc::rlim_t) {
        let mut current = libc::rlimit {
            rlim_cur: 0,
            rlim_max: 0,
        };
        // SAFETY: a plain `getrlimit` read into a live stack `rlimit`.
        assert_eq!(
            unsafe { libc::getrlimit(libc::RLIMIT_NPROC, &mut current) },
            0
        );
        (current.rlim_cur, current.rlim_max)
    }

    /// An rlimit as `ulimit` prints it: `unlimited` for `RLIM_INFINITY`, the number otherwise.
    pub(crate) fn rendered_rlimit(value: libc::rlim_t) -> String {
        if value == libc::RLIM_INFINITY {
            "unlimited".to_string()
        } else {
            value.to_string()
        }
    }
}

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

    #[test]
    fn resolve_uses_defaults_when_manifest_is_silent() {
        assert_eq!(
            HostResourceLimits::resolve(None),
            HostResourceLimits::default()
        );
    }

    /// The rule inherited from `capabilities.limits`: a manifest that declares the section but
    /// omits a field gets that field's default, never "unlimited".
    #[test]
    fn resolve_fills_each_omitted_field_independently() {
        let declared = murmur_artifact::ResourceCapabilities {
            max_open_files: Some(16),
            cgroup_pids_max: Some(32),
            ..murmur_artifact::ResourceCapabilities::default()
        };
        let limits = HostResourceLimits::resolve(Some(&declared));

        assert_eq!(limits.max_open_files, 16);
        assert_eq!(limits.cgroup_pids_max, 32);
        assert_eq!(limits.max_processes, DEFAULT_MAX_PROCESSES);
        assert_eq!(limits.max_file_size_bytes, DEFAULT_MAX_FILE_SIZE_BYTES);
        assert_eq!(limits.cpu_seconds, DEFAULT_CPU_SECONDS);
        assert_eq!(limits.memory_bytes, DEFAULT_MEMORY_BYTES);
        assert_eq!(limits.cgroup_memory_bytes, DEFAULT_CGROUP_MEMORY_BYTES);
        assert_eq!(limits.cgroup_cpu_percent, DEFAULT_CGROUP_CPU_PERCENT);
        assert_eq!(
            limits.cgroup_io_bytes_per_sec,
            DEFAULT_CGROUP_IO_BYTES_PER_SEC
        );
        assert_eq!(limits.workdir_max_bytes, DEFAULT_WORKDIR_MAX_BYTES);
    }

    #[test]
    fn free_resolve_matches_the_inherent_one() {
        let declared = murmur_artifact::ResourceCapabilities {
            cpu_seconds: Some(5),
            ..murmur_artifact::ResourceCapabilities::default()
        };
        assert_eq!(
            resolve(Some(&declared)),
            HostResourceLimits::resolve(Some(&declared))
        );
    }

    /// Attribution must name a limit only where the kernel's own signal identifies exactly one.
    /// The negative half matters too: over-attributing a `SIGKILL` to `memory_bytes` would put a
    /// guess into the trace as if it were evidence.
    #[test]
    fn only_the_two_unambiguous_signals_name_a_limit() {
        assert_eq!(limit_from_signal(libc::SIGXCPU), Some("cpu_seconds"));
        assert_eq!(
            limit_from_signal(libc::SIGXFSZ),
            Some("max_file_size_bytes")
        );
        for ambiguous in [libc::SIGKILL, libc::SIGSEGV, libc::SIGABRT, libc::SIGTERM] {
            assert_eq!(
                limit_from_signal(ambiguous),
                None,
                "signal {ambiguous} has more than one possible cause and must stay unattributed"
            );
        }
    }

    /// Each name is taken from the producer that attributes it, so renaming an attribution
    /// without giving the new name its line fails here rather than falling through to the generic
    /// wording.
    #[test]
    fn every_attributable_limit_has_its_own_line() {
        use crate::cgroup::CgroupEventCounters;

        let cpu = limit_from_signal(libc::SIGXCPU).unwrap();
        let fsize = limit_from_signal(libc::SIGXFSZ).unwrap();
        let memory = CgroupEventCounters {
            oom_kill: 1,
            pids_max: 0,
        }
        .attribution_since(CgroupEventCounters::default())
        .unwrap();
        let pids = CgroupEventCounters {
            oom_kill: 0,
            pids_max: 1,
        }
        .attribution_since(CgroupEventCounters::default())
        .unwrap();

        let expected = [
            (
                pids,
                "resource_limit: cgroup_pids_max — the capsule reached its process limit while \
                 this command ran, so starting a new process failed (EAGAIN). Retry with fewer \
                 parallel processes, or ask the operator to raise \
                 capabilities.resources.cgroup_pids_max.",
            ),
            (
                memory,
                "resource_limit: cgroup_memory_bytes — the capsule reached its memory limit while \
                 this command ran, and the kernel killed a process in it to free memory. Retry \
                 using less memory at once (smaller inputs, fewer parallel jobs), or ask the \
                 operator to raise capabilities.resources.cgroup_memory_bytes.",
            ),
            (
                cpu,
                "resource_limit: cpu_seconds — this command used up its CPU-time limit and was \
                 stopped. Retry with less work per command, or ask the operator to raise \
                 capabilities.resources.cpu_seconds.",
            ),
            (
                fsize,
                "resource_limit: max_file_size_bytes — this command tried to write a file larger \
                 than the capsule's file-size limit and was stopped. Retry writing smaller files, \
                 or ask the operator to raise capabilities.resources.max_file_size_bytes.",
            ),
        ];
        for (name, text) in expected {
            let line = resource_limit_line(name);
            assert_eq!(line, text);
            assert!(line.starts_with(&format!("resource_limit: {name} — ")));
            assert!(line.contains(&format!("capabilities.resources.{name}")));
            assert!(line.contains("ask the operator to raise"));
            assert!(!line.contains('\n'), "{line:?} must be a single line");
        }

        for cgroup in [pids, memory] {
            let line = resource_limit_line(cgroup);
            assert!(line.contains("the capsule"), "{line}");
            assert!(line.contains("while this command ran"), "{line}");
        }
        for signal in [cpu, fsize] {
            assert!(resource_limit_line(signal).contains("this command"));
        }

        assert_eq!(
            resource_limit_line("some_future_limit"),
            "resource_limit: some_future_limit — this command was stopped by the capsule's \
             capabilities.resources.some_future_limit limit. Retry with less work, or ask the \
             operator to raise capabilities.resources.some_future_limit."
        );
    }

    #[test]
    fn directory_size_sums_nested_files() {
        let temp = tempfile::tempdir().unwrap();
        std::fs::write(temp.path().join("a"), vec![0u8; 100]).unwrap();
        std::fs::create_dir(temp.path().join("nested")).unwrap();
        std::fs::write(temp.path().join("nested").join("b"), vec![0u8; 250]).unwrap();

        assert_eq!(directory_size_bytes(temp.path()), 350);
    }

    #[test]
    fn directory_size_of_a_missing_path_is_zero() {
        assert_eq!(
            directory_size_bytes(Path::new("/nonexistent-murmur-workdir-probe")),
            0
        );
    }

    /// The guard starts un-breached, and stays that way for a workdir well under its ceiling —
    /// the "no false positive" half. The breach half is a real-host manual check (see
    /// `docs/content/reference/resource-limits-manual-verification.md`), not a timing-dependent
    /// unit test.
    #[test]
    fn guard_reports_no_breach_for_a_workdir_under_the_ceiling() {
        let temp = tempfile::tempdir().unwrap();
        std::fs::write(temp.path().join("small"), vec![0u8; 64]).unwrap();

        let guard = WorkdirGuard::spawn(temp.path(), DEFAULT_WORKDIR_MAX_BYTES);
        assert_eq!(guard.breach(), None);
    }

    /// The baseline must be a real, plausible count — a `0` here would silently turn
    /// `max_processes` back into the absolute per-uid ceiling that makes every `fork()` in a
    /// subprocess fail on any account already past the limit.
    #[cfg(any(target_os = "linux", target_os = "macos"))]
    #[test]
    fn uid_task_count_reports_a_plausible_baseline() {
        let count = uid_task_count().expect("this platform can be asked for its task count");
        assert!(
            count > 0,
            "the test process itself is owned by this uid, so the count cannot be zero"
        );
    }

    /// On Linux `RLIMIT_NPROC` is enforced against threads, so the baseline sums `task/` entries:
    /// the three owned processes below hold 7 threads between them, where a count of processes
    /// would come out as 3. Directories that are not pids (`self`), a process with no readable
    /// `task/` (`400`), and processes owned by another uid add nothing.
    ///
    /// Run against a synthetic tree because the live count covers the whole uid and moves with
    /// every other thread on the machine, so no assertion that it rises can be deterministic.
    #[cfg(target_os = "linux")]
    #[test]
    fn uid_task_count_sums_threads_not_processes() {
        let temp = tempfile::tempdir().unwrap();
        let root = temp.path();
        for (pid, threads) in [("100", 1), ("200", 4), ("300", 2), ("self", 5)] {
            let task = root.join(pid).join("task");
            std::fs::create_dir_all(&task).unwrap();
            for thread in 0..threads {
                std::fs::write(task.join(thread.to_string()), b"").unwrap();
            }
        }
        std::fs::create_dir(root.join("400")).unwrap();

        // The owner the walk reads off each pid directory, taken from one of them.
        let own_uid = {
            use std::os::unix::fs::MetadataExt;
            std::fs::metadata(root.join("100")).unwrap().uid()
        };

        assert_eq!(uid_task_count_in(root, own_uid), Some(7));
        assert_eq!(uid_task_count_in(root, own_uid.wrapping_add(1)), Some(0));
        assert_eq!(uid_task_count_in(&root.join("missing"), own_uid), None);
    }

    /// `rlim_cur == rlim_max` is the whole point of this module's rlimit path: a soft-only cap
    /// can be raised back by the capsule itself. Asserted against the live process by lowering
    /// a safely lowerable limit (`RLIMIT_CORE`, which this runtime pins at zero anyway)
    /// and reading it back.
    #[cfg(unix)]
    #[test]
    #[allow(unsafe_code)]
    // Same portability cast as `apply_hard_rlimits`: `RLIMIT_CORE` is `c_uint` on Linux and
    // `c_int` on macOS, so the cast only looks redundant from whichever platform clippy runs on.
    #[allow(clippy::unnecessary_cast)]
    fn set_hard_rlimit_writes_both_the_soft_and_hard_slot() {
        set_hard_rlimit(libc::RLIMIT_CORE as u32, 0).unwrap();

        let mut current = libc::rlimit {
            rlim_cur: 1,
            rlim_max: 1,
        };
        // SAFETY: a plain `getrlimit` read into a live stack `rlimit`.
        assert_eq!(
            unsafe { libc::getrlimit(libc::RLIMIT_CORE as _, &mut current) },
            0
        );
        assert_eq!(current.rlim_cur, 0);
        assert_eq!(
            current.rlim_max, 0,
            "the hard slot must be set, not only the soft one"
        );
    }
}