gunnar-server 2.0.0

//! **"If a tenant's cap rose to N bytes, could you serve it?"** — `Admin.Capacity`.
//!
//! The question a seller must ask before the money moves. monetize grows an
//! "increase my quota" button: a tenant on 10 GB asks for 100 GB and something
//! upstream is about to buy 90 GB from a cloud vendor and charge for it. gunnar
//! is the only component that knows whether those bytes can ever reach a
//! repository, and on the appliance the honest answer is frequently **no**.
//!
//! # Two facts, not one number
//!
//! A single free-bytes answer is the dangerous one: a caller reads a small
//! number as "buy another disk", and on the appliance that is money converted
//! into nothing. So [`answer`] returns a `verdict` — can this request be served
//! out of what exists right now — **and**, independently, a `growth` — can this
//! data set ever hold more than it does, and by whose hand. `growth` is
//! reported on the yes arm as well, because "yes, and this is the last capacity
//! this box will ever have" is a different sale from "yes, and there is more
//! where that came from".
//!
//! ## When the appliance is SEALED — and when it is NOT
//!
//! SEALED was once the only answer, on three facts:
//!
//! * `/data` is an md set laid once by `korp-installer`'s `lay_set`
//!   (`edda/crates/korp-installer/src/install.rs`), which runs **only** on
//!   `InstallMode::Fresh`.
//! * gunnar had no data-set growth of any kind.
//! * gunnar is **PID 1 with no shell**, so nothing on the machine can
//!   `xfs_growfs` a newly attached disk either.
//!
//! The second is no longer true, and [`decide_growth`] must not say it is.
//! `korp-start` carries the boot-time grow engine
//! (`korp_installer::grow`, `DATA-SET-GROWTH-DESIGN.md`) and, under the
//! `grow-runtime` feature, THIS server links the same engine's runtime half
//! (`korp_installer::grow::runtime`, driven by [`crate::grow`]). So the
//! verdict is a ladder over MEASURED facts — the journal on the box
//! ([`GrowthFacts::grow`]), this build's engine ([`GrowthFacts::runtime_engine`])
//! and the kernel's PCI hot-plug ([`GrowthFacts::hotplug`]) — and SEALED is
//! only its bottom rung, reached when **this box keeps no grow journal**.
//! Nothing here may claim "no such path in any build": that is a statement
//! about the binary, and the binary is measured one field away.
//!
//! On that bottom rung a disk bought and attached to the box is **billed and
//! invisible**. That is what [`SetGrowth::GrowthSealed`] says, in those words,
//! in the reason line — so a caller that renders nothing but the sentence
//! still cannot mistake it for "not enough space right now".
//!
//! # Measured, never declared, and measured ONCE (one implementation)
//!
//! The bytes are [`gunnar_ssh::DiskHeadroomGate::headroom`] — the same
//! `statvfs` the push door refuses under, so the capacity a seller is quoted
//! and the floor a push dies at cannot disagree. The set's shape is
//! [`crate::dashboard::probe::disk_kind`] — the same `/proc/mounts` + md sysfs
//! read the console's header band prints `4×75G raid0 xfs` from. This module
//! opens no third spelling of either question; it decides over their output.
//!
//! [`answer`] is pure over [`Question`] precisely so the SEALED arm can be
//! driven red on a developer box, where gunnar is never PID 1 and `/data` is
//! never an md set.
//!
//! # The box: cores and RAM, measured (product scope)
//!
//! Disk is a tenant's: a cap is bytes in one namespace. Cores and RAM are the
//! **box's**: no tenant owns a core, so the ceiling a seller may sell either
//! up to is whatever this machine has — and that must be read off the
//! machine, not declared in a config that was true of the plan somebody
//! ordered. [`MachineFacts`] reads them the way the rest of the server
//! already does (one implementation): `available_parallelism` is what
//! [`crate::budget::Machine`] sizes the thread pools from, `/proc/meminfo`
//! is [`crate::dashboard::probe::meminfo`] behind the header band's `ram`
//! cell, the cgroup clamp is [`crate::budget::cgroup_memory`]'s, and the
//! load is the band's `load1`.
//!
//! Every box field on the wire is `optional`, and **absent means unmeasured**
//! — never `0`. A `0` is a number a caller will divide by, sell against, or
//! read as "this box has no RAM"; an absence with `machine_reason` naming
//! the OS error is the same UNMEASURED discipline the disk half has.

use std::path::{Path, PathBuf};

use gunnar_proto::{CapacityResponse, CapacityVerdict, SetGrowth};
use gunnar_ssh::{headroom::human, Headroom};

use crate::dashboard::probe::{self, DiskKind};

/// **What the machine says about the set under the data root.** Two measured
/// facts and nothing inferred; [`decide_growth`] turns them into a verdict.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct GrowthFacts {
    /// The mount and the array under it, as
    /// [`crate::dashboard::probe::disk_kind`] reads them. `None` when the data
    /// root is not a mount point of its own, or `/proc/mounts` is unreadable —
    /// which is the honest UNKNOWN, not a growable answer.
    pub kind: Option<DiskKind>,
    /// **This process is PID 1.** On the appliance gunnar IS init: there is no
    /// shell beside it, so nothing on the machine can grow a filesystem
    /// out of band. On a developer box or in a container beside an init this
    /// is false and an operator's hands still exist.
    pub is_pid1: bool,
    /// **The boot stack's grow journal** (`/serialdat/grow-journal`,
    /// `DATA-SET-GROWTH-DESIGN.md`), when this box's `korp-start` keeps one —
    /// which is the measured fact that the boot stack CAN grow the set at a
    /// restart. `None` on an image older than the engine, where the
    /// appliance is as sealed as it ever was.
    pub grow: Option<gunnar_grpc::grow_journal::Summary>,
    /// **This gunnar-server links the runtime grow engine** (`crate::grow`,
    /// feature `grow-runtime`): with the journal on the box, the set grows
    /// WITHOUT a restart through `Admin.GrowStart`. A build fact, read once.
    pub runtime_engine: bool,
    /// **The running kernel has PCI hot-plug**
    /// ([`crate::dashboard::probe::pci_hotplug`]). UpCloud resizes only
    /// DETACHED volumes, so without it a member detached and re-attached while
    /// gunnar runs is invisible: the engine is there, the growth is not.
    pub hotplug: bool,
}

impl GrowthFacts {
    /// Read them off the running machine.
    pub fn measure(root: &Path) -> GrowthFacts {
        GrowthFacts {
            kind: crate::dashboard::probe::disk_kind(root),
            is_pid1: std::process::id() == 1,
            grow: gunnar_grpc::grow_journal::Summary::measure(),
            runtime_engine: cfg!(feature = "grow-runtime"),
            hotplug: crate::dashboard::probe::pci_hotplug().present,
        }
    }
}

/// The growth verdict and the clause the reason line carries for it.
pub fn decide_growth(facts: &GrowthFacts, root: &Path) -> (SetGrowth, String) {
    let root = root.display();
    match (&facts.kind, facts.is_pid1) {
        (None, _) => (
            SetGrowth::GrowthUnknown,
            format!(
                "gunnar cannot tell what {root} sits on — it is not a mount point of its own, \
                 so whether this set can grow is UNKNOWN, which is not a yes"
            ),
        ),
        // The boot stack carries the grow engine (its journal is on this
        // box) AND this server links it: the set grows WITHOUT a restart
        // (`Admin.GrowStart`: drain, unmount, wait for larger members, grow,
        // mount, reopen — DATA-SET-GROWTH-FLOW.md §3). The restart path
        // below still holds as the fallback for a crash mid-way.
        (Some(kind), true) if facts.grow.is_some() && facts.runtime_engine && facts.hotplug => {
            let journal = facts.grow.as_ref().expect("guarded");
            let n = kind.set.as_ref().map(|s| s.members).unwrap_or(1);
            (
                SetGrowth::GrowthAtRuntime,
                format!(
                    "AND THE SET GROWS WITHOUT A RESTART: {root} is a {kind} set, this gunnar-server \
                     links the grow engine, the kernel has PCI hot-plug and the journal is on this box — `Admin.GrowStart` drains \
                     the box, unmounts, waits for ALL {n} members to come back larger at one equal \
                     size, grows in place and serves again in minutes; a crash mid-way is finished \
                     at the next boot. Grow journal: {}",
                    journal.detail()
                ),
            )
        }
        // The boot stack carries the grow engine (its journal is on this
        // box): the set grows at the NEXT RESTART into enlarged member
        // volumes. Still no hot-plug, still no shell — an attached disk is
        // as wasted as under SEALED, and that is said in the same breath.
        (Some(kind), true) if facts.grow.is_some() => {
            let journal = facts.grow.as_ref().expect("guarded");
            let n = kind.set.as_ref().map(|s| s.members).unwrap_or(1);
            (
                SetGrowth::GrowthAtRestart,
                format!(
                    "AND THE SET GROWS AT RESTART: {root} is a {kind} set and this box's boot stack \
                     (korp-start) grows it into enlarged member volumes at the next boot — enlarge \
                     ALL {n} members to one equal size, restart, and the next measurement reads the \
                     larger set. It does NOT grow while gunnar runs{}, and a disk ATTACHED to this \
                     machine is still invisible to it. Grow journal: {}",
                    if facts.runtime_engine && !facts.hotplug {
                        " (the grow engine is linked, but the running kernel has NO PCI hot-plug: a member detached and re-attached while gunnar runs is invisible, so a runtime growth is refused by name)"
                    } else {
                        ""
                    },
                    journal.detail()
                ),
            )
        }
        (Some(kind), true) => {
            // The `lay_set` clause is claimed only where an md set was
            // actually measured. A single-disk PID-1 box is just as sealed —
            // no growth path, no shell — but it was not laid by
            // `korp-installer`'s array pass, and saying so would be inventing
            // a fact about a machine nobody looked at.
            let laid = if kind.set.is_some() {
                " laid once at install (korp-installer `lay_set`, InstallMode::Fresh only)"
            } else {
                ""
            };
            // The MEASURED reason this box cannot grow is that it keeps no
            // grow journal — its boot stack predates the engine. Whether
            // THIS binary links the runtime half is a separate fact, and it
            // is measured (`runtime_engine`), so the line names it instead
            // of asserting "no such path in any build" over a binary that
            // may well carry one.
            //
            // "so it refuses by name" below was a CLAIM until 2026-09-18 and
            // is a fact since: an audit found `GrowMachine::start` had no
            // journal precondition, so a `GrowStart` on exactly the box this
            // arm describes was accepted and drained the gate before failing
            // at `note(Step::Drain)`. The refusal now sits in
            // `grow::GrowMachine::start`, on the SAME file `facts.grow`
            // measures here, so this sentence and that door cannot disagree.
            let engine = if facts.runtime_engine {
                " (this server links the runtime grow engine, but there is no grow journal on this box for it to continue, so it refuses by name)"
            } else {
                ""
            };
            (
                SetGrowth::GrowthSealed,
                format!(
                    "AND THE SET CANNOT GROW WHILE GUNNAR RUNS: {root} is a {kind} set{laid}, \
                     this box's boot stack keeps no grow journal{engine}, and gunnar is PID 1 \
                     with no shell — so a disk attached to this machine would be UNUSED AND \
                     INVISIBLE. Serving more needs a fresh install onto a larger set, not \
                     another disk"
                ),
            )
        }
        (Some(kind), false) => (
            SetGrowth::GrowthOperatorOnly,
            format!(
                "gunnar will not grow the set itself here — the grow engine runs only where \
                 gunnar is PID 1 over a set its own boot stack laid, and this process is not \
                 PID 1 — but that is also why an operator with a shell can extend {root} \
                 ({kind}) out of band, and the next measurement will see it"
            ),
        ),
    }
}

/// **What the box has: cores, RAM, and how busy it is.** Product-scope facts,
/// each either measured or carrying the reason it was not; [`machine_fields`]
/// turns them into the wire's `optional` fields and the `machine_reason`.
#[derive(Debug, Clone, PartialEq)]
pub struct MachineFacts {
    /// CPUs this process may **run on**: the affinity mask, further limited by
    /// a cgroup v2 `cpu.max` quota when one is set
    /// (`std::thread::available_parallelism`, the spelling
    /// [`crate::budget::Machine`] sizes the pools from). `Err` is the OS's
    /// reason the kernel would not say.
    pub cpus: Result<usize, String>,
    /// `(MemTotal, MemAvailable)` in bytes as `/proc/meminfo` reports them,
    /// then clamped by `cgroup_memory_max` when a limit is set. `Err` names why
    /// `/proc/meminfo` could not be read or parsed.
    pub memory: Result<(u64, u64), String>,
    /// The cgroup v2 `memory.max` that clamped `memory`, when one did and was
    /// smaller than the box — reported so the reason line can say the quoted
    /// RAM is the container's, not the host's.
    pub cgroup_memory_max: Option<u64>,
    /// The 1-minute load average, or why `/proc/loadavg` could not be read.
    pub load1: Result<f64, String>,
    /// **What the data root's disk did at start-up**: the last outcome of
    /// [`crate::disktest`], or `None` when the test has not finished one (the
    /// gate was closed, or the thread is still running). A LOOKUP — reading
    /// this never starts a disk test, because the answer wanted is what the
    /// disk did at start-up on an idle box, not what it does under whatever
    /// load is on it now.
    pub disk: Option<crate::disktest::Outcome>,
}

impl MachineFacts {
    /// Read them off the running machine. Every reader is one the server
    /// already has (one implementation); this only keeps their failures instead of
    /// flattening them to a dash or a zero.
    pub fn measure() -> MachineFacts {
        let cpus = std::thread::available_parallelism()
            .map(|n| n.get())
            .map_err(|err| format!("the affinity mask could not be read ({err})"));
        let mut memory = probe::meminfo();
        let mut cgroup_memory_max = None;
        // The clamp `budget::Machine::read` applies, for the reason it gives:
        // inside a cgroup `/proc/meminfo` is the HOST's memory, and the limit
        // is the smaller truth. The appliance has no limit; there this is a
        // no-op.
        if let (Ok((total, available)), Some((cap, used))) = (&memory, crate::budget::cgroup_memory())
        {
            if cap < *total {
                cgroup_memory_max = Some(cap);
                memory = Ok((cap, (*available).min(cap.saturating_sub(used))));
            }
        }
        MachineFacts {
            cpus,
            memory,
            cgroup_memory_max,
            load1: probe::load1_value(),
            disk: crate::disktest::last(),
        }
    }
}

/// The wire's `disk_*` fields out of one [`crate::disktest::Outcome`], and
/// nothing else set — the same shape and the same contract as
/// [`machine_fields`], so both arms of [`answer`] fill them the same way.
///
/// **Absent, never zero, and never a fast disk.** Every arm below either
/// fills a number that was really measured or leaves the field `None` and
/// says why in `disk_reason`. The one subtle case is `cached`: the read leg
/// came back over [`crate::disktest::CACHED_BPS`], which means the page cache
/// answered and the device was never touched, so `disk_seq_read_bps` is
/// ABSENT while the write and sync legs of the same run stand. A caller
/// comparing two boxes must be able to tell "this disk is slow" from "this
/// disk was not measured", and that is the whole reason this function exists
/// instead of a `unwrap_or(0)`.
pub fn disk_fields(disk: &Option<crate::disktest::Outcome>) -> CapacityResponse {
    use crate::disktest::{Outcome, CACHED_BPS};
    use gunnar_proto::DiskTest;

    let bps = |v: f64| -> Option<u64> {
        // A negative or non-finite rate is not a measurement.
        (v.is_finite() && v > 0.0).then(|| v.round() as u64)
    };
    match disk {
        None => CapacityResponse {
            disk_test: DiskTest::DisktestNotRun as i32,
            disk_reason: "The start-up disk test has not finished a run in this process: its \
                          gate is closed (not the appliance's PID 1 and GUNNAR_DISKTEST unset) \
                          or it is still running. Every disk_* speed field is absent — this is \
                          NOT a report of a slow disk."
                .to_owned(),
            ..Default::default()
        },
        Some(Outcome::Skipped { free, need }) => CapacityResponse {
            disk_test: DiskTest::DisktestSkipped as i32,
            disk_reason: format!(
                "The start-up disk test SKIPPED: {} free on the data root, below the {} it \
                 needs, and the test must never be the thing that fills the disk. Every disk_* \
                 speed field is absent — nothing was measured.",
                human(*free),
                human(*need)
            ),
            ..Default::default()
        },
        Some(Outcome::Failed(why)) => CapacityResponse {
            disk_test: DiskTest::DisktestFailed as i32,
            disk_reason: format!(
                "The start-up disk test FAILED ({why}). Every disk_* speed field is absent — \
                 the run did not produce numbers."
            ),
            ..Default::default()
        },
        Some(Outcome::Done(r)) => {
            let cached = r.seq_read_bps > CACHED_BPS;
            let mut clauses = Vec::with_capacity(3);
            let disk_seq_write_bps = bps(r.seq_write_bps);
            match disk_seq_write_bps {
                Some(_) => clauses.push(format!(
                    "sequential write {}",
                    crate::disktest::fmt_rate(r.seq_write_bps)
                )),
                None => clauses
                    .push("sequential write UNMEASURED (the leg reported no rate)".to_owned()),
            }
            let disk_seq_read_bps = if cached {
                clauses.push(format!(
                    "sequential read CACHED and therefore ABSENT: the leg came back at {} \
                     ({}), over the {} bar, so the page cache answered and the device was never \
                     touched — this is not a fast disk, it is no reading of the disk at all",
                    crate::disktest::fmt_rate(r.seq_read_bps),
                    r.read_how.as_str(),
                    crate::disktest::fmt_rate(CACHED_BPS)
                ));
                None
            } else {
                let v = bps(r.seq_read_bps);
                match v {
                    Some(_) => clauses.push(format!(
                        "sequential read {} ({})",
                        crate::disktest::fmt_rate(r.seq_read_bps),
                        r.read_how.as_str()
                    )),
                    None => clauses
                        .push("sequential read UNMEASURED (the leg reported no rate)".to_owned()),
                }
                v
            };
            let disk_sync_milliops_per_s = (r.sync_ops_per_s.is_finite()
                && r.sync_ops_per_s > 0.0)
                .then(|| (r.sync_ops_per_s * 1000.0).round() as u64);
            match disk_sync_milliops_per_s {
                Some(_) => clauses.push(format!(
                    "4 KiB overwrite + fsync {:.0} ops/s",
                    r.sync_ops_per_s
                )),
                None => clauses
                    .push("4 KiB fsync UNMEASURED (the leg reported no rate)".to_owned()),
            }
            let us = |d: std::time::Duration| Some(u64::try_from(d.as_micros()).unwrap_or(u64::MAX));
            CapacityResponse {
                disk_test: DiskTest::DisktestMeasured as i32,
                disk_seq_write_bps,
                disk_seq_read_bps,
                disk_sync_milliops_per_s,
                disk_read_how: r.read_how.as_str().to_owned(),
                // What the three rates are made of, raw: the same run, its
                // bytes and each leg's own clock. Present even when the read
                // came back cached — the TIME is still what happened; it is
                // the rate derived from it that is not a disk's.
                disk_test_bytes: Some(r.file_bytes),
                disk_write_us: us(r.write_elapsed),
                disk_read_us: us(r.read_elapsed),
                disk_sync_ops: Some(r.sync_ops),
                disk_sync_us: us(r.sync_elapsed),
                disk_test_us: us(r.elapsed),
                disk_reason: format!(
                    "This disk, at start-up, on an idle box: {}. Taken in {:.1} s.",
                    clauses.join("; "),
                    r.elapsed.as_secs_f64()
                ),
                ..Default::default()
            }
        }
    }
}

/// The wire's box fields out of [`MachineFacts`] — and nothing else set, so a
/// caller builds a response as `CapacityResponse { ..., ..machine_fields(&m) }`
/// and both arms of [`answer`] fill these the same way (one implementation).
///
/// **Absent, never zero.** An `Err` fact leaves its fields `None` and puts the
/// reason in `machine_reason`; `machine_reason` is never empty, because a
/// caller that renders only the strings must still learn what was measured.
pub fn machine_fields(m: &MachineFacts) -> CapacityResponse {
    let mut clauses = Vec::with_capacity(3);
    let cpu_millicores_total = match &m.cpus {
        Ok(n) => {
            clauses.push(format!(
                "{n} CPUs = {} millicores gunnar may run on (affinity mask, cgroup cpu.max limit \
                 applied)",
                n * 1000
            ));
            Some((*n as u64) * 1000)
        }
        Err(why) => {
            clauses.push(format!("cores UNMEASURED ({why}), so cpu_millicores_total is absent"));
            None
        }
    };
    let (ram_bytes_total, ram_bytes_available) = match &m.memory {
        Ok((total, available)) => {
            let clamp = match m.cgroup_memory_max {
                Some(cap) => format!(
                    " — clamped to the cgroup memory.max of {}, which is what this process \
                     may use; /proc/meminfo reports the host",
                    human(cap)
                ),
                None => String::new(),
            };
            clauses.push(format!(
                "RAM {} total, {} available (/proc/meminfo MemTotal/MemAvailable{clamp})",
                human(*total),
                human(*available)
            ));
            (Some(*total), Some(*available))
        }
        Err(why) => {
            clauses.push(format!(
                "RAM UNMEASURED ({why}), so ram_bytes_total and ram_bytes_available are absent"
            ));
            (None, None)
        }
    };
    let cpu_millicores_busy = match &m.load1 {
        Ok(load) => {
            let busy = (load * 1000.0).round().max(0.0) as u64;
            clauses.push(format!(
                "1-min load {load:.2} ≈ {busy} millicores busy (/proc/loadavg × 1000: an \
                 APPROXIMATION of the whole box's runnable tasks, not gunnar's own CPU time)"
            ));
            Some(busy)
        }
        Err(why) => {
            clauses.push(format!("load UNMEASURED ({why}), so cpu_millicores_busy is absent"));
            None
        }
    };
    // The disk's own fields ride along, because every caller builds its
    // answer as `CapacityResponse { ..., ..machine_fields(&m) }` and two
    // struct updates cannot compose — one seam, one implementation, and the
    // disk can never be filled on one arm of `answer` and not the other.
    let d = disk_fields(&m.disk);
    CapacityResponse {
        cpu_millicores_total,
        ram_bytes_total,
        ram_bytes_available,
        cpu_millicores_busy,
        machine_reason: format!("This box: {}.", clauses.join("; ")),
        ..d
    }
}

/// **Every disk on the box, for `Admin.Capacity { disks: true }`** (2026-09-28).
///
/// edda's product-neutral `disk-snapshot` does the reading — the mount table,
/// one statvfs per storage mount, `/sys/block` — and this names gunnar's own
/// paths on it: `system` is `/`, `data` the store root (the explode read cache
/// lives inside each repository there, so it is ON this mount and not a disk of
/// its own), `slow` the off-stripe root. The floor is the `--disk-floor` in
/// force, applied per mount, so the `data` row's floor is the number the push
/// door refuses at.
pub fn measure_disks(
    data_root: &Path,
    slow_root: Option<&Path>,
    floor: gunnar_ssh::DiskFloor,
) -> gunnar_proto::DiskSnapshot {
    let mut volumes = vec![
        disk_snapshot::Volume::new("system", "/"),
        disk_snapshot::Volume::new("data", data_root),
    ];
    if let Some(slow) = slow_root {
        volumes.push(disk_snapshot::Volume::new("serialdat", slow));
    }
    let snap = disk_snapshot::take(
        &volumes,
        disk_snapshot::Floor {
            percent: floor.percent,
            min_bytes: floor.bytes,
        },
    );
    disks_to_wire(snap, gunnar_core::now_ms())
}

/// **What is on /serialdat, thing by thing** — the known paths gunnar keeps
/// there, each measured as the blocks it holds (a directory summed), plus
/// `other`: the mount's used bytes no known path accounts for.
///
/// `known` is `(label, path)`; a path that does not exist holds 0 bytes and is
/// measured as such (nothing is there). A path whose walk failed is
/// `measured: false` with the error, and is left out of the sum, so `other`
/// is never made smaller by a number nobody read. `used` is the mount's own
/// statvfs used bytes; `None` (unmeasured mount) → no `other` row.
pub fn serialdat_breakdown(known: &[(&str, PathBuf)], used: Option<u64>) -> Vec<gunnar_proto::PathUse> {
    let mut rows: Vec<gunnar_proto::PathUse> = known
        .iter()
        .map(|(label, path)| {
            let mut row = gunnar_proto::PathUse {
                label: (*label).to_owned(),
                path: path.display().to_string(),
                ..Default::default()
            };
            match allocated_bytes(path) {
                Ok(bytes) => {
                    row.measured = true;
                    row.bytes = bytes;
                }
                Err(why) => row.error = why,
            }
            row
        })
        .collect();
    // A known path INSIDE another (the audit trail inside the log directory)
    // is counted once: the outer row keeps only what the inner ones do not.
    let inner: Vec<(PathBuf, u64)> = known
        .iter()
        .zip(rows.iter())
        .filter(|(_, r)| r.measured)
        .map(|((_, p), r)| (p.clone(), r.bytes))
        .collect();
    for ((_, outer), row) in known.iter().zip(rows.iter_mut()) {
        if !row.measured {
            continue;
        }
        let nested: u64 = inner
            .iter()
            .filter(|(p, _)| p != outer && p.starts_with(outer))
            .map(|(_, b)| *b)
            .sum();
        row.bytes = row.bytes.saturating_sub(nested);
    }
    if let Some(used) = used {
        let known_sum: u64 = rows.iter().filter(|r| r.measured).map(|r| r.bytes).sum();
        rows.push(gunnar_proto::PathUse {
            label: "other".to_owned(),
            path: String::new(),
            measured: true,
            bytes: used.saturating_sub(known_sum),
            error: String::new(),
        });
    }
    rows
}

/// The blocks a path holds on disk (`st_blocks × 512`), a directory summed
/// without following symlinks. A missing path is 0: nothing is there.
fn allocated_bytes(path: &Path) -> Result<u64, String> {
    use std::os::unix::fs::MetadataExt as _;
    let meta = match std::fs::symlink_metadata(path) {
        Ok(m) => m,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(0),
        Err(e) => return Err(format!("{}: {e}", path.display())),
    };
    let mut total = meta.blocks() * 512;
    if meta.is_dir() {
        let entries = std::fs::read_dir(path).map_err(|e| format!("{}: {e}", path.display()))?;
        for entry in entries {
            let entry = entry.map_err(|e| format!("{}: {e}", path.display()))?;
            total += allocated_bytes(&entry.path())?;
        }
    }
    Ok(total)
}

/// The snapshot, as the wire carries it. Pure, so a recorded snapshot can be
/// checked field by field: an unmeasured mount keeps `measured = false` and its
/// reason, and its numbers stay 0 rather than being invented.
pub fn disks_to_wire(snap: disk_snapshot::Snapshot, at_unix_ms: i64) -> gunnar_proto::DiskSnapshot {
    let mounts = snap
        .disks
        .into_iter()
        .map(|d| {
            let mut row = gunnar_proto::DiskMount {
                source: d.mount.source,
                mount_point: d.mount.mount_point.display().to_string(),
                fs_type: d.mount.fstype,
                read_only: d.mount.read_only,
                labels: d.labels,
                floor_bytes: d.floor_bytes,
                servable_bytes: d.servable_bytes,
                ..Default::default()
            };
            match d.figures {
                Ok(f) => {
                    row.measured = true;
                    row.total_bytes = f.total_bytes;
                    row.free_bytes = f.free_bytes;
                    row.available_bytes = f.available_bytes;
                    row.used_bytes = f.used_bytes;
                    row.reserved_bytes = f.reserved_bytes;
                    row.inodes_total = f.inodes_total;
                    row.inodes_free = f.inodes_free;
                }
                Err(why) => row.error = why,
            }
            row
        })
        .collect();
    let devices = snap
        .block_devices
        .into_iter()
        .map(|b| gunnar_proto::BlockDevice {
            name: b.name,
            size_bytes: b.size_bytes,
            rotational: b.rotational,
            removable: b.removable,
            read_only: b.read_only,
            partitions: b
                .partitions
                .into_iter()
                .map(|(name, size_bytes)| gunnar_proto::BlockPartition { name, size_bytes })
                .collect(),
            holders: b.holders,
            members: b.members,
            raid_level: b.raid_level,
        })
        .collect();
    let unresolved = snap
        .unresolved
        .into_iter()
        .map(|u| gunnar_proto::UnresolvedDisk {
            label: u.label,
            path: u.path.display().to_string(),
            why: u.why,
        })
        .collect();
    gunnar_proto::DiskSnapshot {
        measured_at_unix_ms: at_unix_ms,
        mounts,
        devices,
        unresolved,
        problems: snap.problems,
        // Filled by the caller, which knows gunnar's own paths on /serialdat.
        serialdat: Vec::new(),
    }
}

/// Everything [`answer`] decides from. Built by
/// [`crate::control_plane::CoreControlPlane`] out of the disk gate and the
/// namespace's meter; built by hand in the tests, which is the point.
#[derive(Debug, Clone)]
pub struct Question {
    /// The store root the numbers are of.
    pub root: PathBuf,
    /// The one `statvfs`, or the OS's reason it failed. `Err` is
    /// [`CapacityVerdict::CapacityUnmeasured`] — never a yes.
    pub room: Result<Headroom, String>,
    /// `--disk-floor` as configured, for the reason line.
    pub floor_spec: String,
    /// The cap being contemplated.
    pub requested_bytes: u64,
    /// What the namespace already holds (`Usage.Read`'s `pack_bytes`).
    pub held_bytes: u64,
    /// Whose cap is rising; empty asks about bytes that are not here yet.
    pub namespace: String,
    pub growth: GrowthFacts,
    /// The box, for the product-scope fields. Independent of the question:
    /// the same cores and RAM are reported whatever cap was asked about.
    pub machine: MachineFacts,
}

/// **The answer.** Pure: every field below is arithmetic over [`Question`].
pub fn answer(q: &Question) -> CapacityResponse {
    let (growth, growth_clause) = decide_growth(&q.growth, &q.root);
    // The box fields, once for both arms: what the machine has does not
    // depend on whether the disk could be measured.
    let mut machine = machine_fields(&q.machine);
    // The journal rides both arms: what the boot stack did about the set is
    // a fact whether or not the filesystem could be measured now.
    machine.grow_state = q.growth.grow.as_ref().map(|s| s.to_wire());
    machine.hotplug = q.growth.hotplug;
    let data_set = q
        .growth
        .kind
        .as_ref()
        .map(|k| k.to_string())
        .unwrap_or_default();
    let root = q.root.display();

    // Bytes the raise actually needs: what a tenant already holds is already
    // spent out of free space, so a 10 GB tenant asking for 100 GB needs 90,
    // not 100. An unnamed namespace has held 0 and needs the whole cap.
    let additional = q.requested_bytes.saturating_sub(q.held_bytes);
    let asked = if q.namespace.is_empty() {
        format!("a cap of {}", human(q.requested_bytes))
    } else {
        format!(
            "raising {} to {} (it holds {})",
            q.namespace,
            human(q.requested_bytes),
            human(q.held_bytes)
        )
    };

    let room = match &q.room {
        Err(err) => {
            return CapacityResponse {
                verdict: CapacityVerdict::CapacityUnmeasured as i32,
                growth: growth as i32,
                root: q.root.display().to_string(),
                requested_bytes: q.requested_bytes,
                held_bytes: q.held_bytes,
                additional_bytes: additional,
                data_set,
                reason: format!(
                    "{root} could not be measured ({err}), so gunnar will not answer that {asked} \
                     can be served — an unmeasured box has not said yes. {growth_clause}"
                ),
                // Nothing was measured, so nothing is reported. A plausible
                // number here would be a blind guard, asserting nothing.
                total_bytes: 0,
                free_bytes: 0,
                floor_bytes: 0,
                servable_bytes: 0,
                shortfall_bytes: 0,
                ..machine
            };
        }
        Ok(room) => *room,
    };

    // The floor is space that EXISTS and may never be sold: it is what lets the
    // box compact a pack and rewrite its control documents after it has stopped
    // accepting pushes. Quoting against `free` rather than against this is how
    // a seller sells the reserve.
    let servable = room.free.saturating_sub(room.floor);
    let measured = format!(
        "{root} has {} free of {}, {} of it reserved by the disk floor (--disk-floor {}), so {} \
         may be served",
        human(room.free),
        human(room.total),
        human(room.floor),
        q.floor_spec,
        human(servable),
    );

    // One constructor for both arms (one implementation): the verdict, the shortfall and
    // the clause differ; the twelve measured fields do not, and writing them
    // twice is how the two arms drift into reporting different things.
    let (verdict, shortfall, outcome) = if additional <= servable {
        (
            CapacityVerdict::CapacityAvailable,
            0,
            format!("needs {} more: YES, it can be served.", human(additional)),
        )
    } else {
        let shortfall = additional - servable;
        (
            CapacityVerdict::CapacityExhausted,
            shortfall,
            format!(
                "needs {} more, which is {} beyond that: NO.",
                human(additional),
                human(shortfall)
            ),
        )
    };
    CapacityResponse {
        verdict: verdict as i32,
        growth: growth as i32,
        root: q.root.display().to_string(),
        total_bytes: room.total,
        free_bytes: room.free,
        floor_bytes: room.floor,
        servable_bytes: servable,
        requested_bytes: q.requested_bytes,
        held_bytes: q.held_bytes,
        additional_bytes: additional,
        shortfall_bytes: shortfall,
        data_set,
        reason: format!("{measured}. {asked} {outcome} {growth_clause}"),
        ..machine
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::dashboard::probe::SetFacts;

    const GIB: u64 = 1024 * 1024 * 1024;

    /// The live gunnar.rs appliance, as `/proc/mounts` and the md sysfs read
    /// on it: `/data` is a 4×75 GiB raid0 xfs set and gunnar is PID 1.
    fn appliance() -> GrowthFacts {
        GrowthFacts {
            kind: Some(DiskKind {
                fs: "xfs".into(),
                set: Some(SetFacts {
                    level: "raid0".into(),
                    members: 4,
                    member_bytes: 75 * GIB,
                    chunk_bytes: 512 * 1024,
                }),
            }),
            is_pid1: true,
            grow: None,
            runtime_engine: false,
            hotplug: false,
        }
    }

    /// The same disk under an init, with a shell beside gunnar.
    fn developer_box() -> GrowthFacts {
        GrowthFacts {
            is_pid1: false,
            ..appliance()
        }
    }

    /// The appliance re-imaged with the grow engine in its boot stack: the
    /// journal exists, with `text` in it.
    fn growing_appliance(text: &str) -> GrowthFacts {
        GrowthFacts {
            grow: Some(gunnar_grpc::grow_journal::Summary::of(text, "/serialdat/grow-journal")),
            ..appliance()
        }
    }

    /// **The runtime arm** (`DATA-SET-GROWTH-FLOW.md` ruling 1): the journal
    /// on the box AND this server linking the engine is `GROWTH_AT_RUNTIME`;
    /// either fact alone is not — the journal without the engine is the
    /// restart arm, the engine without the journal is SEALED (measured, not
    /// declared). The clause names the verb and "ALL 4 members".
    #[test]
    fn the_engine_in_the_server_with_the_journal_on_the_box_grows_at_runtime() {
        let both = GrowthFacts { runtime_engine: true, hotplug: true, ..growing_appliance("") };
        let (v, line) = decide_growth(&both, Path::new("/data"));
        assert_eq!(v, SetGrowth::GrowthAtRuntime);
        assert!(line.contains("WITHOUT A RESTART") && line.contains("Admin.GrowStart") && line.contains("ALL 4 members"), "{line}");
        let engine_only = GrowthFacts { runtime_engine: true, hotplug: true, ..appliance() };
        assert_eq!(decide_growth(&engine_only, Path::new("/data")).0, SetGrowth::GrowthSealed, "no journal: the engine alone proves nothing");
        assert_eq!(decide_growth(&growing_appliance(""), Path::new("/data")).0, SetGrowth::GrowthAtRestart, "journal alone: the restart arm");
    }

    /// **No hot-plug, no runtime growth** (owner's rule 2026-09-15): the engine
    /// linked and the journal on the box, but a kernel without PCI hot-plug, is
    /// `GROWTH_AT_RESTART` with the reason named — and `hotplug` rides the wire
    /// on every arm so monetize stops guessing.
    #[test]
    fn the_engine_without_a_hotplug_kernel_grows_only_at_restart_and_says_why() {
        let no_hp = GrowthFacts { runtime_engine: true, hotplug: false, ..growing_appliance("") };
        let (v, line) = decide_growth(&no_hp, Path::new("/data"));
        assert_eq!(v, SetGrowth::GrowthAtRestart);
        assert!(line.contains("NO PCI hot-plug") && line.contains("refused by name"), "{line}");
        let mut q = question(gunnar_rs_today(), no_hp.clone());
        q.requested_bytes = 1;
        assert!(!answer(&q).hotplug);
        q.growth.hotplug = true;
        let a = answer(&q);
        assert!(a.hotplug && a.growth == SetGrowth::GrowthAtRuntime as i32, "{a:?}");
        q.room = Err("EIO".into());
        assert!(answer(&q).hotplug, "the unmeasured arm carries it too");
    }

    /// **The verdict the grow engine adds** (`DATA-SET-GROWTH-DESIGN.md`
    /// §3, §5): a PID-1 md-set box whose boot stack keeps the journal is
    /// `GROWTH_AT_RESTART`, never SEALED — and the clause still says an
    /// attached disk is wasted, names "ALL 4 members" and carries the
    /// journal's own sentence (idle / in progress / done / PARTIAL). The
    /// journal rides the wire on the yes arm, the no arm and the unmeasured
    /// arm alike; a box with no journal has no `grow_state` and stays SEALED.
    #[test]
    fn a_boot_stack_with_the_grow_engine_grows_at_restart_and_reports_the_journal() {
        let (v, line) = decide_growth(&growing_appliance(""), Path::new("/data"));
        assert_eq!(v, SetGrowth::GrowthAtRestart);
        assert!(line.contains("GROWS AT RESTART") && line.contains("ALL 4 members") && line.contains("invisible to it"), "{line}");
        assert!(line.contains("no grow has run"), "{line}");
        assert_eq!(decide_growth(&appliance(), Path::new("/data")).0, SetGrowth::GrowthSealed, "no journal: sealed as before");

        // partial resize: the last boot refused, and the clause says which members grew
        let partial = "epoch=0 target_sectors=134215680 member=0 step=PARTIAL at=1\nepoch=0 target_sectors=134215680 member=1 step=PARTIAL at=1\n";
        let (v, line) = decide_growth(&growing_appliance(partial), Path::new("/data"));
        assert_eq!(v, SetGrowth::GrowthAtRestart);
        assert!(line.contains("PARTIAL RESIZE: 2 member(s) [0, 1]"), "{line}");

        // on the wire, on every arm
        let done = "epoch=1 target_sectors=134215680 member=- step=G1 at=1\nepoch=1 target_sectors=134215680 member=- step=G6 at=2\n";
        let mut q = question(gunnar_rs_today(), growing_appliance(done));
        q.requested_bytes = 1;
        let a = answer(&q);
        assert_eq!(a.growth, SetGrowth::GrowthAtRestart as i32);
        let g = a.grow_state.expect("the journal is on the wire");
        assert!(g.engine_present && g.phase == "done" && g.epoch == 1 && g.target_sectors == 134_215_680 && g.journal_path == "/serialdat/grow-journal", "{g:?}");
        assert!(a.reason.contains("epoch 1 done"), "{}", a.reason);
        q.requested_bytes = 1000 * GIB;
        assert!(answer(&q).grow_state.is_some(), "the no arm carries it too");
        q.room = Err("EIO".into());
        assert!(answer(&q).grow_state.is_some(), "and the unmeasured arm");
        let q = question(gunnar_rs_today(), appliance());
        assert!(answer(&q).grow_state.is_none(), "no journal, no grow_state — an absence, never an empty one");
    }

    /// The live gunnar.rs appliance's box: the UpCloud STARTER plan, 4 vCPU
    /// and 16 GiB, no cgroup (gunnar is PID 1), a quiet minute.
    fn appliance_box() -> MachineFacts {
        MachineFacts {
            cpus: Ok(4),
            memory: Ok((16 * GIB, 12 * GIB)),
            cgroup_memory_max: None,
            load1: Ok(0.42),
            disk: Some(crate::disktest::Outcome::Done(crate::disktest::Report {
                seq_write_bps: 812e6,
                seq_read_bps: 1.9e9,
                read_how: crate::disktest::ReadHow::Direct,
                sync_ops_per_s: 3100.0,
                elapsed: std::time::Duration::from_millis(1600),
                file_bytes: 256 << 20,
                write_elapsed: std::time::Duration::from_millis(330),
                read_elapsed: std::time::Duration::from_millis(141),
                sync_ops: 1550,
                sync_elapsed: std::time::Duration::from_millis(500),
            })),
        }
    }

    fn question(room: Headroom, growth: GrowthFacts) -> Question {
        Question {
            root: PathBuf::from("/data"),
            room: Ok(room),
            floor_spec: "5%,2.0 GiB".into(),
            requested_bytes: 0,
            held_bytes: 0,
            namespace: String::new(),
            growth,
            machine: appliance_box(),
        }
    }

    /// gunnar.rs today: 300 GB of set, ~230 GB used, the 5 % floor in force.
    fn gunnar_rs_today() -> Headroom {
        let total = 300 * GIB;
        Headroom {
            total,
            free: 70 * GIB,
            floor: (total / 100) * 5,
        }
    }

    #[test]
    fn the_growth_verdict_separates_a_sealed_appliance_from_a_box_with_a_shell() {
        let (sealed, sealed_line) = decide_growth(&appliance(), Path::new("/data"));
        let (operator, operator_line) = decide_growth(&developer_box(), Path::new("/data"));
        let (unknown, unknown_line) = decide_growth(
            &GrowthFacts {
                kind: None,
                is_pid1: false,
                grow: None,
                runtime_engine: false,
                hotplug: false,
            },
            Path::new("/tmp/x"),
        );

        // Three answers, never two: an unmeasured set is not a growable one,
        // and a sealed set is not merely "gunnar will not do it".
        assert_eq!(sealed, SetGrowth::GrowthSealed);
        assert_eq!(operator, SetGrowth::GrowthOperatorOnly);
        assert_eq!(unknown, SetGrowth::GrowthUnknown);
        assert!(
            sealed != operator && operator != unknown && sealed != unknown,
            "the same facts must not collapse to one verdict"
        );

        // The sealed clause says the thing that stops a purchase, in words.
        assert!(
            sealed_line.contains("UNUSED AND INVISIBLE"),
            "the sealed clause must say an attached disk is wasted: {sealed_line}"
        );
        assert!(
            sealed_line.contains("fresh install") && sealed_line.contains("not another disk"),
            "and must name what WOULD work: {sealed_line}"
        );
        assert!(
            sealed_line.contains("4\u{d7}75G raid0 xfs"),
            "and must name the measured set, not a generic one: {sealed_line}"
        );
        // The difference is real, not a wording variant: the operator arm
        // promises the opposite thing.
        assert!(
            !operator_line.contains("UNUSED AND INVISIBLE")
                && operator_line.contains("an operator with a shell"),
            "{operator_line}"
        );
        assert!(unknown_line.contains("UNKNOWN, which is not a yes"), "{unknown_line}");
    }

    /// **The row this whole RPC exists for.** gunnar.rs, 300 GiB of raid0,
    /// 70 GiB free, a tenant holding 10 GiB asking to be raised to 100 GiB —
    /// 90 GiB of new bytes, against 55 GiB servable.
    #[test]
    fn ninety_more_gigabytes_on_the_sealed_appliance_is_a_no_that_names_the_reinstall() {
        let mut q = question(gunnar_rs_today(), appliance());
        q.namespace = "tenant".into();
        q.held_bytes = 10 * GIB;
        q.requested_bytes = 100 * GIB;
        let a = answer(&q);

        assert_eq!(a.verdict, CapacityVerdict::CapacityExhausted as i32, "{}", a.reason);
        assert_eq!(a.growth, SetGrowth::GrowthSealed as i32);
        // The arithmetic, every term of it.
        assert_eq!(a.free_bytes, 70 * GIB);
        assert_eq!(a.floor_bytes, 15 * GIB, "5 % of 300 GiB");
        assert_eq!(a.servable_bytes, 55 * GIB, "free minus the floor, not free");
        assert_eq!(a.additional_bytes, 90 * GIB, "the raise needs the DELTA, not the cap");
        assert_eq!(a.shortfall_bytes, 35 * GIB);
        assert_eq!(a.data_set, "4\u{d7}75G raid0 xfs");
        // And the sentence carries both halves.
        assert!(a.reason.contains("NO."), "{}", a.reason);
        assert!(a.reason.contains("UNUSED AND INVISIBLE"), "{}", a.reason);
    }

    /// The SAME numbers on a box with a shell answer NO too — but for a
    /// different reason, and that difference is the whole point of the second
    /// field. A caller reading only `verdict` would buy a disk here and be
    /// right; on the appliance it would buy one and be robbed.
    #[test]
    fn the_same_shortfall_on_a_box_with_a_shell_does_not_say_a_disk_is_wasted() {
        let mut sealed = question(gunnar_rs_today(), appliance());
        sealed.requested_bytes = 90 * GIB;
        let mut shelled = sealed.clone();
        shelled.growth = developer_box();

        let a = answer(&sealed);
        let b = answer(&shelled);
        assert_eq!(a.verdict, b.verdict, "the shortfall is identical");
        assert_eq!(a.shortfall_bytes, b.shortfall_bytes);
        assert_ne!(a.growth, b.growth, "the growth answer is NOT the verdict");
        assert_ne!(a.reason, b.reason, "and it changes what the operator is told");
        assert!(b.reason.contains("NO.") && !b.reason.contains("UNUSED AND INVISIBLE"));
    }

    /// A yes still reports the growth fact — "yes, and this is the last
    /// capacity this box will ever have" is a different sale.
    #[test]
    fn a_request_that_fits_is_a_yes_that_still_says_the_set_is_sealed() {
        let mut q = question(gunnar_rs_today(), appliance());
        q.requested_bytes = 20 * GIB;
        let a = answer(&q);
        assert_eq!(a.verdict, CapacityVerdict::CapacityAvailable as i32);
        assert_eq!(a.shortfall_bytes, 0);
        assert!(a.reason.contains("YES"), "{}", a.reason);
        assert_eq!(
            a.growth,
            SetGrowth::GrowthSealed as i32,
            "the growth fact is reported on the YES arm too"
        );
        assert!(a.reason.contains("UNUSED AND INVISIBLE"), "{}", a.reason);

        // The boundary is `servable`, not `free`: 60 GiB is under the 70 GiB
        // free and over the 55 GiB servable, and must be refused.
        q.requested_bytes = 60 * GIB;
        assert_eq!(
            answer(&q).verdict,
            CapacityVerdict::CapacityExhausted as i32,
            "a request under free but over free-minus-floor must be refused: the floor is not \
             for sale"
        );
        q.requested_bytes = 55 * GIB;
        assert_eq!(answer(&q).verdict, CapacityVerdict::CapacityAvailable as i32);
    }

    #[test]
    fn a_filesystem_that_cannot_be_measured_is_not_a_yes_and_reports_no_numbers() {
        let mut q = question(gunnar_rs_today(), appliance());
        q.room = Err("No such file or directory (os error 2)".into());
        q.requested_bytes = 1;
        let a = answer(&q);
        assert_eq!(a.verdict, CapacityVerdict::CapacityUnmeasured as i32);
        assert_ne!(
            a.verdict,
            CapacityVerdict::CapacityAvailable as i32,
            "one byte would fit on any real disk; an unmeasured one must still not say yes"
        );
        assert_eq!(
            (a.total_bytes, a.free_bytes, a.floor_bytes, a.servable_bytes),
            (0, 0, 0, 0),
            "nothing was measured, so nothing may be reported"
        );
        assert!(a.reason.contains("os error 2"), "{}", a.reason);
        assert!(a.reason.contains("has not said yes"), "{}", a.reason);
    }

    /// **The box, as the live appliance would answer it**: 4 vCPU = 4000
    /// millicores, 16 GiB / 12 GiB, load 0.42 ≈ 420 millicores — and the
    /// reason line names each source, calls the load an approximation, and
    /// is carried on the UNMEASURED-disk arm too.
    #[test]
    fn the_appliance_box_is_reported_in_millicores_and_bytes_with_its_sources_named() {
        let q = question(gunnar_rs_today(), appliance());
        let a = answer(&q);
        assert_eq!(a.cpu_millicores_total, Some(4000));
        assert_eq!(a.ram_bytes_total, Some(16 * GIB));
        assert_eq!(a.ram_bytes_available, Some(12 * GIB));
        assert_eq!(a.cpu_millicores_busy, Some(420), "0.42 × 1000, rounded");
        for word in ["affinity mask", "/proc/meminfo", "/proc/loadavg", "APPROXIMATION"] {
            assert!(a.machine_reason.contains(word), "{word}: {}", a.machine_reason);
        }
        assert!(
            !a.machine_reason.contains("clamped"),
            "no cgroup clamped the appliance: {}",
            a.machine_reason
        );

        // The same box on the disk-unmeasured arm: the fields do not depend
        // on whether statvfs worked, so one constructor serves both.
        let mut unmeasured = q.clone();
        unmeasured.room = Err("No such file or directory (os error 2)".into());
        let u = answer(&unmeasured);
        assert_eq!(
            (u.cpu_millicores_total, u.ram_bytes_total, u.ram_bytes_available, u.cpu_millicores_busy),
            (a.cpu_millicores_total, a.ram_bytes_total, a.ram_bytes_available, a.cpu_millicores_busy)
        );
        assert_eq!(u.machine_reason, a.machine_reason);
    }

    /// **Absent, never zero.** A box whose `/proc/meminfo` cannot be read
    /// reports NO RAM fields — `None` on the wire, not `Some(0)` — and the
    /// reason names the OS error; the cores it DID measure are still there.
    #[test]
    fn an_unreadable_meminfo_leaves_the_ram_fields_absent_and_names_why() {
        let mut q = question(gunnar_rs_today(), appliance());
        q.machine.memory = Err("/proc/meminfo could not be read (No such file or directory (os error 2))".into());
        q.machine.load1 = Err("/proc/loadavg could not be read (Permission denied (os error 13))".into());
        let a = answer(&q);
        assert_eq!(a.ram_bytes_total, None, "absent, not 0: {a:?}");
        assert_eq!(a.ram_bytes_available, None);
        assert_eq!(a.cpu_millicores_busy, None);
        assert_ne!(a.ram_bytes_total, Some(0), "a zero here would be a box with no RAM");
        assert_eq!(a.cpu_millicores_total, Some(4000), "the cores were measured and stay");
        assert!(a.machine_reason.contains("RAM UNMEASURED"), "{}", a.machine_reason);
        assert!(a.machine_reason.contains("os error 2"), "{}", a.machine_reason);
        assert!(a.machine_reason.contains("load UNMEASURED"), "{}", a.machine_reason);
        assert!(a.machine_reason.contains("os error 13"), "{}", a.machine_reason);

        // And the whole box unmeasured is a reason line, not an empty one.
        q.machine.cpus = Err("the affinity mask could not be read (os error 22)".into());
        let none = answer(&q);
        assert_eq!(
            (none.cpu_millicores_total, none.ram_bytes_total, none.ram_bytes_available, none.cpu_millicores_busy),
            (None, None, None, None)
        );
        assert!(none.machine_reason.contains("cores UNMEASURED"), "{}", none.machine_reason);
        // The disk answer is untouched by any of it.
        assert_eq!(none.verdict, a.verdict);
        assert_eq!(none.reason, a.reason);
    }

    /// A cgroup limit smaller than the host is the RAM gunnar may use, and
    /// the reason says the number is the container's, not the host's.
    #[test]
    fn a_cgroup_memory_clamp_is_reported_as_the_ram_and_named_in_the_reason() {
        let mut q = question(gunnar_rs_today(), appliance());
        q.machine.memory = Ok((8 * GIB, 6 * GIB));
        q.machine.cgroup_memory_max = Some(8 * GIB);
        let a = answer(&q);
        assert_eq!(a.ram_bytes_total, Some(8 * GIB));
        assert!(a.machine_reason.contains("clamped to the cgroup memory.max of 8.0 GiB"), "{}", a.machine_reason);
    }

    /// **The thirteen disk fields are independent of the box.** The same
    /// question with a measured box and with an unmeasured one yields
    /// byte-identical disk fields — and the wire proves it in
    /// `tests/capacity.rs`, where the bytes are decoded by an OLDER client.
    #[test]
    fn the_disk_answer_does_not_change_with_the_box() {
        let q = question(gunnar_rs_today(), appliance());
        let mut blind = q.clone();
        blind.machine = MachineFacts {
            cpus: Err("x".into()),
            memory: Err("y".into()),
            cgroup_memory_max: None,
            load1: Err("z".into()),
            disk: None,
        };
        let (a, b) = (answer(&q), answer(&blind));
        assert_ne!(a.machine_reason, b.machine_reason, "the fixtures differ");
        assert_ne!(a.ram_bytes_total, b.ram_bytes_total);
        // Mask the BOX's own fields and the start-up disk TEST's — both are
        // facts about the machine rather than about the store's free space,
        // and this test is about the `statvfs` answer alone. The disk test
        // joined the mask 2026-09-21 when it reached the wire: a box that has
        // measured its disk and one that has not must still give the same
        // capacity verdict, which is exactly what the mask asserts.
        let disk = |r: &CapacityResponse| {
            CapacityResponse {
                cpu_millicores_total: None,
                ram_bytes_total: None,
                ram_bytes_available: None,
                cpu_millicores_busy: None,
                machine_reason: String::new(),
                disk_test: 0,
                disk_seq_write_bps: None,
                disk_seq_read_bps: None,
                disk_sync_milliops_per_s: None,
                disk_read_how: String::new(),
                disk_reason: String::new(),
                // The run's raw bytes and clocks (2026-09-28) are the same
                // disk test's, so they are masked with it.
                disk_test_bytes: None,
                disk_write_us: None,
                disk_read_us: None,
                disk_sync_ops: None,
                disk_sync_us: None,
                disk_test_us: None,
                ..r.clone()
            }
        };
        assert_eq!(disk(&a), disk(&b));
        // …and the two fixtures really did differ on the disk test, or the
        // mask above would be asserting nothing.
        assert_ne!(a.disk_test, b.disk_test);
        assert_ne!(a.disk_reason, b.disk_reason);
    }
}

/// **The disk on the wire** — added 2026-09-21 because the twin election has
/// to compare two boxes' disks and `Admin.Capacity` did not carry them.
///
/// The property under test is one sentence: *a box that could not measure
/// must be distinguishable from a box that measured badly, and neither may
/// look like a fast disk.* Every arm below drives one way of not measuring
/// and asserts that it is named, and that no speed field carries a zero.
#[cfg(test)]
mod disk_on_the_wire {
    use super::*;
    use crate::disktest::{Outcome, ReadHow, Report};
    use gunnar_proto::DiskTest;
    use std::time::Duration;

    fn done(seq_read_bps: f64) -> Outcome {
        Outcome::Done(Report {
            seq_write_bps: 812e6,
            seq_read_bps,
            read_how: ReadHow::Direct,
            sync_ops_per_s: 3100.4,
            elapsed: Duration::from_millis(1600),
            file_bytes: 256 << 20,
            write_elapsed: Duration::from_millis(330),
            read_elapsed: Duration::from_millis(141),
            sync_ops: 1550,
            sync_elapsed: Duration::from_millis(500),
        })
    }

    /// A real run puts three numbers on the wire, in the wire's units.
    #[test]
    fn a_measured_disk_carries_its_three_numbers_and_says_how_it_read() {
        let a = disk_fields(&Some(done(1.9e9)));
        assert_eq!(a.disk_test, DiskTest::DisktestMeasured as i32);
        assert_eq!(a.disk_seq_write_bps, Some(812_000_000));
        assert_eq!(a.disk_seq_read_bps, Some(1_900_000_000));
        // ops/s × 1000, so the wire stays integral the way millicores do.
        assert_eq!(a.disk_sync_milliops_per_s, Some(3_100_400));
        assert_eq!(a.disk_read_how, "direct");
        assert!(a.disk_reason.contains("812 MB/s"), "{}", a.disk_reason);
        assert!(a.disk_reason.contains("1.9 GB/s"), "{}", a.disk_reason);
        // …and what they are made of: the bytes and each leg's clock.
        assert_eq!(a.disk_test_bytes, Some(256 << 20));
        assert_eq!((a.disk_write_us, a.disk_read_us), (Some(330_000), Some(141_000)));
        assert_eq!((a.disk_sync_ops, a.disk_sync_us), (Some(1550), Some(500_000)));
        assert_eq!(a.disk_test_us, Some(1_600_000));
    }

    /// **The cached read is the subtle one.** Over 5 GB/s the page cache
    /// answered and the device was never touched — so the read is ABSENT,
    /// not fast; and the write and fsync legs of the same run still stand,
    /// because they were really measured.
    #[test]
    fn a_cached_read_is_absent_from_the_wire_and_is_never_reported_as_a_fast_disk() {
        let a = disk_fields(&Some(done(7.2e9)));
        assert_eq!(a.disk_test, DiskTest::DisktestMeasured as i32, "the run DID measure");
        assert_eq!(a.disk_seq_read_bps, None, "not 7.2 GB/s and not 0 — absent");
        assert_eq!(a.disk_seq_write_bps, Some(812_000_000), "the other legs stand");
        assert_eq!(a.disk_sync_milliops_per_s, Some(3_100_400));
        assert!(a.disk_reason.contains("CACHED"), "{}", a.disk_reason);
        assert!(
            a.disk_reason.contains("not a fast disk"),
            "the reason says what the absence MEANS: {}",
            a.disk_reason
        );
    }

    /// The four silences are four different facts, each named, and none of
    /// them puts a number anywhere.
    #[test]
    fn a_disk_that_was_not_measured_is_named_and_carries_no_number_at_all() {
        let cases = [
            (None, DiskTest::DisktestNotRun, "gate"),
            (
                Some(Outcome::Skipped { free: 800 << 20, need: 1 << 30 }),
                DiskTest::DisktestSkipped,
                "SKIPPED",
            ),
            (
                Some(Outcome::Failed("write: No space left on device".into())),
                DiskTest::DisktestFailed,
                "FAILED",
            ),
        ];
        for (outcome, want, word) in cases {
            let a = disk_fields(&outcome);
            assert_eq!(a.disk_test, want as i32, "{outcome:?}");
            assert_eq!(a.disk_seq_write_bps, None, "{outcome:?}");
            assert_eq!(a.disk_seq_read_bps, None, "{outcome:?}");
            assert_eq!(a.disk_sync_milliops_per_s, None, "{outcome:?}");
            assert!(a.disk_read_how.is_empty(), "{outcome:?}");
            assert_eq!(
                (a.disk_test_bytes, a.disk_write_us, a.disk_read_us, a.disk_sync_ops, a.disk_sync_us, a.disk_test_us),
                (None, None, None, None, None, None),
                "no run, no bytes and no times: {outcome:?}"
            );
            assert!(!a.disk_reason.is_empty(), "never silent: {outcome:?}");
            assert!(a.disk_reason.contains(word), "{}", a.disk_reason);
        }
        // Skipped names BOTH figures, so an operator can act on it.
        let s = disk_fields(&Some(Outcome::Skipped { free: 800 << 20, need: 1 << 30 }));
        assert!(s.disk_reason.contains("800") && s.disk_reason.contains("1.0"), "{}", s.disk_reason);
        // And the three silences do not say the same thing as each other.
        let reasons: std::collections::BTreeSet<String> = [
            disk_fields(&None).disk_reason,
            disk_fields(&Some(Outcome::Skipped { free: 0, need: 1 })).disk_reason,
            disk_fields(&Some(Outcome::Failed("x".into()))).disk_reason,
        ]
        .into_iter()
        .collect();
        assert_eq!(reasons.len(), 3);
    }

    /// The disk rides on the SAME seam as the box's cores and RAM, so it can
    /// never be filled on one arm of `answer` and forgotten on the other.
    #[test]
    fn the_box_fields_and_the_disk_fields_come_out_of_one_call() {
        let m = MachineFacts {
            cpus: Ok(4),
            memory: Ok((16 << 30, 12 << 30)),
            cgroup_memory_max: None,
            load1: Ok(0.42),
            disk: Some(done(1.9e9)),
        };
        let a = machine_fields(&m);
        assert_eq!(a.cpu_millicores_total, Some(4000));
        assert_eq!(a.disk_seq_write_bps, Some(812_000_000));
        assert!(!a.machine_reason.is_empty() && !a.disk_reason.is_empty());
        // A box with no disk reading still reports its cores.
        let blind = machine_fields(&MachineFacts { disk: None, ..m });
        assert_eq!(blind.cpu_millicores_total, Some(4000));
        assert_eq!(blind.disk_test, DiskTest::DisktestNotRun as i32);
    }

    /// The latch is a LOOKUP: reading it never starts a disk test, and an
    /// outcome recorded is the outcome read back.
    #[test]
    fn the_latch_hands_back_what_was_recorded_and_runs_nothing() {
        let before = crate::disktest::last();
        crate::disktest::record(&done(1.9e9));
        assert_eq!(crate::disktest::last(), Some(done(1.9e9)));
        crate::disktest::record(&Outcome::Failed("boom".into()));
        assert_eq!(crate::disktest::last(), Some(Outcome::Failed("boom".into())));
        // Put back whatever another test in this binary may have left.
        if let Some(b) = before {
            crate::disktest::record(&b);
        }
    }
}

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

    /// A recorded snapshot reaches the wire field by field, and an unmeasured
    /// mount stays unmeasured: `measured = false`, the reason, and zeroes that
    /// the flag says are not readings.
    #[test]
    fn a_snapshot_reaches_the_wire_and_an_unmeasured_mount_is_not_a_number() {
        let mount = |src: &str, at: &str| disk_snapshot::Mount {
            source: src.to_owned(),
            mount_point: PathBuf::from(at),
            fstype: "xfs".to_owned(),
            read_only: false,
        };
        let snap = disk_snapshot::Snapshot {
            disks: vec![
                disk_snapshot::DiskEntry {
                    mount: mount("/dev/md0", "/data"),
                    labels: vec!["data".to_owned()],
                    paths: vec![PathBuf::from("/data")],
                    figures: disk_snapshot::Figures::from_counts(1000, 400, 350, 4096, 10, 5),
                    floor_bytes: 4096 * 50,
                    servable_bytes: 4096 * 300,
                },
                disk_snapshot::DiskEntry {
                    mount: mount("/dev/vda2", "/"),
                    labels: vec!["system".to_owned()],
                    paths: vec![PathBuf::from("/")],
                    figures: Err("statvfs(/): EIO".to_owned()),
                    floor_bytes: 0,
                    servable_bytes: 0,
                },
            ],
            unresolved: vec![],
            block_devices: vec![disk_snapshot::BlockDevice {
                name: "md0".to_owned(),
                size_bytes: 4096 * 1000,
                members: vec!["vdb1".to_owned(), "vdc1".to_owned()],
                raid_level: "raid0".to_owned(),
                ..Default::default()
            }],
            problems: vec![],
        };
        let wire = disks_to_wire(snap, 42);
        assert_eq!(wire.measured_at_unix_ms, 42);
        let data = &wire.mounts[0];
        assert!(data.measured && data.labels == ["data"]);
        assert_eq!(
            (data.total_bytes, data.free_bytes, data.available_bytes, data.used_bytes, data.reserved_bytes),
            (4096 * 1000, 4096 * 400, 4096 * 350, 4096 * 600, 4096 * 50)
        );
        assert_eq!((data.floor_bytes, data.servable_bytes), (4096 * 50, 4096 * 300));
        let system = &wire.mounts[1];
        assert!(!system.measured && system.error.contains("EIO") && system.total_bytes == 0);
        assert_eq!(wire.devices[0].members, ["vdb1", "vdc1"]);
        assert_eq!(wire.devices[0].raid_level, "raid0");
    }

    /// **The /serialdat breakdown sums**: each known path measured by its
    /// blocks, the audit trail counted once though it sits inside the log
    /// directory, a missing path 0, and `other` = the mount's used bytes minus
    /// what the known paths hold — so the rows add up to the mount's use.
    #[test]
    fn the_serialdat_breakdown_sums_to_the_mounts_use() {
        use std::io::Write as _;
        let dir = tempfile::tempdir().expect("a temp mount");
        let root = dir.path();
        let write = |rel: &str, n: usize| {
            let p = root.join(rel);
            std::fs::create_dir_all(p.parent().unwrap()).unwrap();
            let mut f = std::fs::File::create(&p).unwrap();
            f.write_all(&vec![7u8; n]).unwrap();
            f.sync_all().unwrap();
        };
        write("meta/principals.json", 10_000);
        write("log/audit-chain.jsonl", 70_000);
        write("log/console.log", 30_000);
        write("crash/last.txt", 5_000);
        let known = vec![
            ("metadata copy", root.join("meta")),
            ("crash record", root.join("crash")),
            ("audit trail", root.join("log/audit-chain.jsonl")),
            ("logs", root.join("log")),
            ("grow journal", root.join("grow-journal")),
        ];
        let used = 10 << 20;
        let rows = serialdat_breakdown(&known, Some(used));
        let get = |label: &str| rows.iter().find(|r| r.label == label).unwrap_or_else(|| panic!("{label}: {rows:?}"));
        assert!(rows.iter().all(|r| r.measured), "{rows:?}");
        assert!(get("audit trail").bytes >= 70_000, "{rows:?}");
        let logs = get("logs").bytes;
        assert!((30_000..70_000).contains(&logs), "the audit trail was counted inside the logs too: {rows:?}");
        assert_eq!(get("grow journal").bytes, 0, "nothing there is 0, not an error");
        let sum: u64 = rows.iter().map(|r| r.bytes).sum();
        assert_eq!(sum, used, "the rows must add up to the mount's used bytes: {rows:?}");
        // An unmeasured mount: no `other`, nothing invented.
        assert!(serialdat_breakdown(&known, None).iter().all(|r| r.label != "other"));
    }

    /// The live road on this machine: a data root in a temp dir is placed on
    /// the mount it sits on and measured, and `/` is labelled `system`.
    #[test]
    fn this_machines_disks_carry_gunnars_labels() {
        let dir = tempfile::tempdir().expect("a temp dir");
        let wire = measure_disks(dir.path(), None, gunnar_ssh::DiskFloor::default());
        assert!(wire.problems.is_empty(), "{:?}", wire.problems);
        let data = wire
            .mounts
            .iter()
            .find(|m| m.labels.iter().any(|l| l == "data"))
            .expect("the data root is on some mount");
        assert!(data.measured && data.total_bytes > 0, "{data:?}");
        assert!(
            wire.mounts.iter().any(|m| m.labels.iter().any(|l| l == "system")),
            "{:?}",
            wire.mounts
        );
    }
}