rch_common/

capability_probe.rs

//! Exact user/path capability probes for runtimes, toolchains, and targets
//! (bd-session-history-remediation-ocv9i.12.2).
//!
//! A capability probe runs **as the configured remote user, using the exact
//! executable paths RCH will actually invoke** (not whatever a login shell's
//! PATH happens to resolve), so the facts reflect what an offloaded build will
//! really see. The probe is a single shell script that prints `RCH_FACT k=v`
//! lines; [`parse_capability_probe`] turns that output into [`ProbedFacts`],
//! and [`assess_admissibility`] decides whether the worker is admissible for a
//! given [`CapabilityRequirement`].
//!
//! Crucially this distinguishes **SSH reachability** (did the script run at
//! all?) from **capability admissibility** (it ran, but lacks a needed target /
//! the binary at the exact path is broken / the protocol is stale) — the two
//! failure classes operators kept conflating.

use crate::incident::IncidentReasonCode;
use crate::worker_facts::{
    DiskRootFacts, RuntimeFacts, RustFacts, WorkerBinaryFacts, derive_target_triple,
};

/// Sentinel prefix every probe fact line carries, so probe output is easy to
/// separate from incidental stdout.
pub const FACT_PREFIX: &str = "RCH_FACT ";

/// Why an operator-declared tool probe was refused before it ever ran.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum ToolProbeError {
    /// The name cannot survive the fact wire format or is empty.
    #[error("tool name {name:?} is invalid: {reason}")]
    InvalidName {
        /// The rejected name.
        name: String,
        /// What is wrong with it.
        reason: &'static str,
    },
    /// A probe with no argv can never prove anything.
    #[error("tool {name:?} declares an empty probe command")]
    EmptyCommand {
        /// The tool whose command was empty.
        name: String,
    },
}

/// An operator-declared named tool probe: a FIXED argv, written in worker
/// configuration, run as the configured remote user.
///
/// The argv is never caller-supplied. RCH deliberately exposes no general
/// remote-shell probe API: a build that could ask a worker to run an arbitrary
/// command "to check for a tool" would be a remote execution primitive wearing
/// a capability-probe hat. Declaring the argv in operator config keeps the set
/// of commands a worker will run for a probe finite and reviewable.
///
/// Verification is by EXIT STATUS: zero means present, anything else — missing
/// binary, error exit, signal — means absent. Output is discarded so a chatty
/// tool cannot corrupt the fact stream.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NamedToolProbe {
    /// The name selection and `--require-tool` refer to.
    name: String,
    /// The exact argv to run.
    command: Vec<String>,
}

impl NamedToolProbe {
    /// Declare a probe, rejecting names that cannot round-trip the fact wire
    /// format (`RCH_FACT tool=<name>`): a name containing `=`, whitespace or a
    /// control character would be parsed back as a different name, which is a
    /// silent capability lie rather than a loud config error.
    ///
    /// # Errors
    /// [`ToolProbeError`] when the name is unusable or the command is empty.
    pub fn new(
        name: impl Into<String>,
        command: impl IntoIterator<Item = impl Into<String>>,
    ) -> Result<Self, ToolProbeError> {
        let name = name.into();
        let command: Vec<String> = command.into_iter().map(Into::into).collect();
        let invalid = |reason| ToolProbeError::InvalidName {
            name: name.clone(),
            reason,
        };
        if name.is_empty() {
            return Err(invalid("it is empty"));
        }
        if !name
            .bytes()
            .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_' | b'.' | b'+'))
        {
            return Err(invalid(
                "only ASCII letters, digits and `-` `_` `.` `+` are allowed",
            ));
        }
        if command.is_empty() {
            return Err(ToolProbeError::EmptyCommand { name });
        }
        Ok(Self { name, command })
    }

    /// The declared name.
    #[must_use]
    pub fn name(&self) -> &str {
        &self.name
    }

    /// The declared argv.
    #[must_use]
    pub fn command(&self) -> &[String] {
        &self.command
    }
}

impl TryFrom<&crate::types::WorkerToolProbe> for NamedToolProbe {
    type Error = ToolProbeError;

    /// Validate a `workers.toml` entry. This is the single point where a
    /// malformed declaration is refused, so a bad name fails config load
    /// loudly rather than becoming a worker that quietly advertises nothing.
    fn try_from(entry: &crate::types::WorkerToolProbe) -> Result<Self, Self::Error> {
        Self::new(entry.name.clone(), entry.command.iter().cloned())
    }
}

/// Exact paths/identity the probe must use (never PATH-resolved).
#[derive(Debug, Clone)]
pub struct ProbeSpec {
    /// SSH login user the probe runs as.
    pub remote_user: String,
    /// Exact path to the `rch-wkr` binary RCH will invoke.
    pub rch_wkr_path: String,
    /// Exact path to `cargo`, if known (else the probe tries `command -v`).
    pub cargo_path: Option<String>,
    /// Exact path to `rustup`, if known.
    pub rustup_path: Option<String>,
    /// Disk roots whose capacity to report (temp root, build roots, cargo home).
    pub disk_roots: Vec<String>,
    /// Operator-declared named tool probes for this worker.
    pub tools: Vec<NamedToolProbe>,
}

impl ProbeSpec {
    #[must_use]
    pub fn new(remote_user: impl Into<String>, rch_wkr_path: impl Into<String>) -> Self {
        Self {
            remote_user: remote_user.into(),
            rch_wkr_path: rch_wkr_path.into(),
            cargo_path: None,
            rustup_path: None,
            disk_roots: Vec::new(),
            tools: Vec::new(),
        }
    }

    /// Declare the operator-configured tool probes (builder style).
    #[must_use]
    pub fn with_tools(mut self, tools: impl IntoIterator<Item = NamedToolProbe>) -> Self {
        self.tools = tools.into_iter().collect();
        self
    }
}

/// Shell-quote a value for safe single-argument embedding.
fn shq(s: &str) -> String {
    format!("'{}'", s.replace('\'', "'\\''"))
}

/// The canonical ABSOLUTE path to the deployed `rch-wkr` binary for a remote
/// user and host OS. Unix workers use `/root` or `/home/<user>`; Windows workers
/// run under Git Bash and use `/c/Users/<user>` plus the native `.exe` suffix.
/// The probe script shell-quotes this path, so it MUST be absolute: a literal
/// `~/.local/bin/...` would be single-quoted and never expand, making `[ -x ... ]`
/// fail and every worker look like it has a missing binary. Single source of
/// truth for both the daemon bypass-recovery prober and the
/// `rch self-test --smoke` capability scenario.
#[must_use]
pub fn remote_worker_binary_path(user: &str, declared_os: Option<&str>) -> String {
    if declared_os.is_some_and(|os| os.eq_ignore_ascii_case("windows")) {
        return format!("/c/Users/{user}/.local/bin/rch-wkr.exe");
    }

    let home = if user == "root" {
        "/root".to_string()
    } else {
        format!("/home/{user}")
    };
    format!("{home}/.local/bin/rch-wkr")
}

/// Build the capability-probe shell script. It is intentionally fail-soft:
/// every probe that errors simply omits its fact line, so a missing rustup or a
/// broken binary shows up as *absent facts* (capability), never as a script
/// crash (reachability).
#[must_use]
pub fn build_capability_probe_script(spec: &ProbeSpec) -> String {
    let wkr = shq(&spec.rch_wkr_path);
    let cargo = spec
        .cargo_path
        .as_deref()
        .map_or_else(|| "cargo".to_string(), shq);
    let rustup = spec
        .rustup_path
        .as_deref()
        .map_or_else(|| "rustup".to_string(), shq);

    let mut s = String::new();
    s.push_str("set -u; P='RCH_FACT '; ");
    // Host facts.
    s.push_str("printf '%sos=%s\\n' \"$P\" \"$(uname -s | tr 'A-Z' 'a-z')\"; ");
    s.push_str("printf '%sarch=%s\\n' \"$P\" \"$(uname -m)\"; ");
    s.push_str("printf '%suser=%s\\n' \"$P\" \"$(id -un 2>/dev/null)\"; ");
    // Worker binary at the EXACT path (version + protocol). Absent => broken.
    s.push_str(&format!("printf '%srch_wkr_path=%s\\n' \"$P\" {wkr}; "));
    s.push_str(&format!(
        "if [ -x {wkr} ]; then v=$({wkr} --version 2>/dev/null) && printf '%sworker_version=%s\\n' \"$P\" \"$v\"; \
         pr=$({wkr} --protocol-version 2>/dev/null) && printf '%sworker_protocol=%s\\n' \"$P\" \"$pr\"; fi; "
    ));
    // Cargo / rust.
    s.push_str(&format!(
        "cv=$({cargo} --version 2>/dev/null) && printf '%scargo_version=%s\\n' \"$P\" \"$cv\"; "
    ));
    // Toolchains + installed targets via rustup (each on its own fact line).
    s.push_str(&format!(
        "{rustup} toolchain list 2>/dev/null | awk -v p=\"$P\" '{{print p\"toolchain=\"$1}}'; "
    ));
    s.push_str(&format!(
        "{rustup} target list --installed 2>/dev/null | awk -v p=\"$P\" '{{print p\"target=\"$1}}'; "
    ));
    // Installed components, reported PER TOOLCHAIN as `component=<toolchain>:<name>`.
    //
    // Per-toolchain is the whole point, not extra precision: components are
    // installed against a specific toolchain, and consuming projects pin their
    // own via `rust-toolchain.toml`. A worker can hold `clippy` for `stable` and
    // still be unable to run `cargo clippy` for a pinned nightly, which is
    // exactly the state that let 4 of 12 workers report healthy while silently
    // voiding a mandated lint gate (bd-vc61a). Probing only the default
    // toolchain would have reported OK for every one of them.
    //
    // `rustup toolchain list` prints the active toolchain with a ` (default)` /
    // ` (override)` suffix, so `$1` is taken as the name. Each inner listing is
    // best-effort: a toolchain that cannot be queried simply contributes no
    // component facts, which reads downstream as "not known to have it" rather
    // than as "known to lack it".
    s.push_str(&format!(
        "for tc in $({rustup} toolchain list 2>/dev/null | awk '{{print $1}}'); do \
           rh=$({rustup} run \"$tc\" rustc -vV 2>/dev/null | awk '$1 == \"host:\" {{print $2; exit}}'); \
           {rustup} component list --installed --toolchain \"$tc\" 2>/dev/null \
             | awk -v p=\"$P\" -v t=\"$tc\" -v h=\"$rh\" \
                 '{{n=$1; if (h != \"\") sub(\"-\" h \"$\", \"\", n); print p\"component=\"t\":\"n}}'; \
         done; "
    ));
    // JS runtimes (PATH-resolved is acceptable for these advisory facts).
    s.push_str("bv=$(bun --version 2>/dev/null) && printf '%sbun_version=%s\\n' \"$P\" \"$bv\"; ");
    s.push_str(
        "nv=$(node --version 2>/dev/null) && printf '%snode_version=%s\\n' \"$P\" \"$nv\"; ",
    );
    s.push_str(
        "npmv=$(npm --version 2>/dev/null) && printf '%snpm_version=%s\\n' \"$P\" \"$npmv\"; ",
    );
    // Go toolchain (PATH-resolved; advisory fact gating go build/test/vet routing).
    s.push_str("gov=$(go version 2>/dev/null) && printf '%sgo_version=%s\\n' \"$P\" \"$gov\"; ");
    // Zig + cargo-zigbuild (PATH-resolved; gate `cargo zigbuild` cross-compile
    // routing). BOTH are required: cargo-zigbuild drives the build, zig is the C
    // compiler/cross-linker it shells out to. Reported as separate facts.
    s.push_str("zv=$(zig version 2>/dev/null) && printf '%szig_version=%s\\n' \"$P\" \"$zv\"; ");
    s.push_str(
        "czv=$(cargo-zigbuild --version 2>/dev/null) && printf '%scargo_zigbuild_version=%s\\n' \"$P\" \"$czv\"; ",
    );
    // Nix: only report a version when a populated `/nix/store` also exists, since
    // a `nix` binary without a store cannot build derivations (gates nix routing).
    s.push_str(
        "if [ -d /nix/store ] && [ -n \"$(ls -A /nix/store 2>/dev/null | head -n1)\" ]; then \
           nixv=$(nix --version 2>/dev/null) && printf '%snix_version=%s\\n' \"$P\" \"$nixv\"; \
         fi; ",
    );
    // Operator-declared named tools. Each is the exact argv from worker config,
    // run as this (already configured) user with output discarded — verification
    // is the EXIT STATUS, and a chatty tool must not be able to inject fact
    // lines. Both outcomes are reported: `tool=` for present, `tool_absent=` for
    // a declared probe that failed. The difference matters operationally — "the
    // fleet never declared clang" and "this worker declares clang but the probe
    // fails" have different fixes, and a facts stream that only carried
    // successes could not tell them apart.
    for tool in &spec.tools {
        let argv = tool
            .command()
            .iter()
            .map(|part| shq(part))
            .collect::<Vec<_>>()
            .join(" ");
        let name = shq(tool.name());
        s.push_str(&format!(
            "if {argv} >/dev/null 2>&1; then printf '%stool=%s\\n' \"$P\" {name}; \
             else printf '%stool_absent=%s\\n' \"$P\" {name}; fi; "
        ));
    }
    // Disk roots: path;total_kb;avail_kb;avail_inodes (df -Pk and df -Pi).
    // A configured future build directory may not exist yet. Measure its
    // closest existing ancestor, preserving the requested root in the fact,
    // so disk recovery can require evidence for every configured location.
    for root in &spec.disk_roots {
        let q = shq(root);
        s.push_str(&format!(
            "rp={q}; \
             while [ ! -e \"$rp\" ] && [ ! -L \"$rp\" ]; do \
               parent=$(dirname -- \"$rp\") || break; \
               [ \"$parent\" != \"$rp\" ] || break; rp=$parent; \
             done; \
             if [ -d \"$rp\" ]; then \
               b=$(df -Pk \"$rp\" 2>/dev/null | awk 'NR==2{{print $2\";\"$4}}'); \
               i=$(df -Pi \"$rp\" 2>/dev/null | awk 'NR==2{{print $4}}'); \
               printf '%sdisk=%s;%s;%s\\n' \"$P\" {q} \"$b\" \"$i\"; \
             fi; "
        ));
    }
    s
}

/// Structured facts parsed from probe output. Sub-facts are `None`/empty when
/// the corresponding probe produced no line (i.e. the capability is absent).
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ProbedFacts {
    /// Probed host OS, normalized by [`normalize_probed_os`].
    pub os: Option<String>,
    pub arch: Option<String>,
    pub probed_user: Option<String>,
    pub rch_wkr_path: Option<String>,
    pub worker: Option<WorkerBinaryFacts>,
    pub rust: RustFacts,
    pub runtimes: RuntimeFacts,
    pub disk_roots: Vec<DiskRootFacts>,
    /// Operator-declared tools whose probe exited zero on this worker.
    pub tools_present: Vec<String>,
    /// Operator-declared tools whose probe ran and did NOT exit zero.
    pub tools_absent: Vec<String>,
    /// Raw worker_version line (kept even if protocol was missing).
    worker_version: Option<String>,
    worker_protocol: Option<u32>,
}

impl ProbedFacts {
    /// The worker's derived target triple from probed os/arch (libc unknown here
    /// — defaults to gnu on linux; collectors that know musl override on the
    /// assembled [`crate::worker_facts::HostFacts`]).
    #[must_use]
    pub fn target_triple(&self) -> Option<String> {
        match (&self.os, &self.arch) {
            (Some(os), Some(arch)) => Some(derive_target_triple(os, arch, None)),
            _ => None,
        }
    }

    /// Whether a named tool was VERIFIED present on this worker. Absence of
    /// evidence reads as absence: a tool nobody declared a probe for is not
    /// present, which is what keeps an unknown `--require-tool` name
    /// inadmissible everywhere instead of silently unconstrained.
    #[must_use]
    pub fn has_tool(&self, name: &str) -> bool {
        self.tools_present.iter().any(|t| t == name)
    }

    /// Whether this worker DECLARED a probe for the tool (either outcome).
    /// Distinguishes "the fleet never declared it" from "declared, probe fails".
    #[must_use]
    pub fn declares_tool(&self, name: &str) -> bool {
        self.has_tool(name) || self.tools_absent.iter().any(|t| t == name)
    }
}

/// Collapse the many shapes `uname -s` takes on Windows into plain `windows`.
///
/// The probe is a POSIX script, so on a Windows worker it runs under MSYS2,
/// Git-Bash or Cygwin, each of which reports a versioned kernel name —
/// `mingw64_nt-10.0-26100`, `msys_nt-10.0-26100`, `cygwin_nt-10.0`. Left raw,
/// none of them compare equal to the `windows` an operator writes in config or
/// a `*-pc-windows-msvc` triple implies. Everything else passes through
/// untouched.
#[must_use]
pub fn normalize_probed_os(raw: &str) -> String {
    let lower = raw.trim().to_ascii_lowercase();
    for prefix in ["mingw32_nt", "mingw64_nt", "msys_nt", "cygwin_nt"] {
        if lower.starts_with(prefix) {
            return "windows".to_string();
        }
    }
    lower
}

/// Parse `RCH_FACT k=v` probe output into [`ProbedFacts`]. Lines without the
/// prefix are ignored, so incidental stdout never corrupts the parse.
#[must_use]
pub fn parse_capability_probe(stdout: &str) -> ProbedFacts {
    let mut f = ProbedFacts::default();
    for line in stdout.lines() {
        let Some(kv) = line.trim().strip_prefix(FACT_PREFIX) else {
            continue;
        };
        let Some((key, value)) = kv.split_once('=') else {
            continue;
        };
        let value = value.trim();
        match key {
            "os" => f.os = Some(normalize_probed_os(value)),
            "arch" => f.arch = Some(value.to_string()),
            "user" => f.probed_user = Some(value.to_string()),
            "rch_wkr_path" => f.rch_wkr_path = Some(value.to_string()),
            "worker_version" => f.worker_version = Some(value.to_string()),
            "worker_protocol" => f.worker_protocol = value.parse::<u32>().ok(),
            "cargo_version" => f.rust.rustc_version = Some(value.to_string()),
            "toolchain" => f.rust.toolchains.push(value.to_string()),
            "target" => f.rust.targets.push(value.to_string()),
            // `component=<toolchain>:<name>`. A value without a separator is
            // dropped rather than stored under an empty toolchain: an entry that
            // cannot say WHICH toolchain owns the component would answer
            // `has_component` for every toolchain, which is the fail-open
            // direction this fact exists to close.
            "component" => {
                if let Some((toolchain, component)) = value.split_once(':') {
                    let toolchain = toolchain.trim();
                    let component = component.trim();
                    if !toolchain.is_empty() && !component.is_empty() {
                        f.rust.components.push(format!("{toolchain}:{component}"));
                    }
                }
            }
            "tool" => {
                if !value.is_empty() && !f.tools_present.iter().any(|t| t == value) {
                    f.tools_present.push(value.to_string());
                }
            }
            "tool_absent" => {
                if !value.is_empty() && !f.tools_absent.iter().any(|t| t == value) {
                    f.tools_absent.push(value.to_string());
                }
            }
            "bun_version" => f.runtimes.bun_version = Some(value.to_string()),
            "node_version" => f.runtimes.node_version = Some(value.to_string()),
            "npm_version" => f.runtimes.npm_version = Some(value.to_string()),
            "nix_version" => f.runtimes.nix_version = Some(value.to_string()),
            "go_version" => f.runtimes.go_version = Some(value.to_string()),
            "zig_version" => f.runtimes.zig_version = Some(value.to_string()),
            "cargo_zigbuild_version" => f.runtimes.cargo_zigbuild_version = Some(value.to_string()),
            "disk" => {
                // path;total_kb;avail_kb;avail_inodes
                let parts: Vec<&str> = value.split(';').collect();
                if parts.len() == 4 {
                    let kb = |s: &str| s.parse::<u64>().unwrap_or(0).saturating_mul(1024);
                    f.disk_roots.push(DiskRootFacts {
                        path: parts[0].to_string(),
                        total_bytes: kb(parts[1]),
                        available_bytes: kb(parts[2]),
                        available_inodes: parts[3].parse::<u64>().unwrap_or(0),
                    });
                }
            }
            _ => {}
        }
    }
    // Assemble the worker binary facts only if a version was reported (i.e. the
    // binary at the exact path actually ran).
    if let Some(version) = f.worker_version.clone() {
        f.worker = Some(WorkerBinaryFacts {
            rch_wkr_path: f.rch_wkr_path.clone().unwrap_or_default(),
            version,
            protocol_version: f.worker_protocol.unwrap_or(0),
        });
    }
    f
}

/// What a command needs from a worker before it is admissible.
#[derive(Debug, Clone, Default)]
pub struct CapabilityRequirement {
    /// rustup targets the build needs (e.g. `wasm32-unknown-unknown`).
    pub needs_targets: Vec<String>,
    /// Minimum acceptable worker wire protocol.
    pub min_protocol: u32,
    /// Whether a working cargo is required.
    pub needs_cargo: bool,
    /// Required host OS for the produced artifact (e.g. `linux`, `darwin`).
    /// Matched case-insensitively against the worker's probed `os`.
    pub needs_os: Option<String>,
    /// Required host arch (e.g. `x86_64`, `aarch64`), matched case-insensitively.
    pub needs_arch: Option<String>,
    /// Whether a working `bun` runtime is required.
    pub needs_bun: bool,
    /// Whether a working `node` runtime is required.
    pub needs_node: bool,
    /// Whether a working Go toolchain is required.
    pub needs_go: bool,
    /// Whether `zig` + `cargo-zigbuild` are required (a `cargo zigbuild`
    /// cross-compile). Enforced ON TOP OF `needs_cargo` and `needs_targets`.
    pub needs_zig: bool,
    /// Whether a working `nix` (binary + populated `/nix/store`) is required.
    pub needs_nix: bool,
    /// Specific rustup toolchains the build needs (prefix-matched against the
    /// worker's installed toolchains, e.g. `nightly-2025-11-01` matches
    /// `nightly-2025-11-01-x86_64-unknown-linux-gnu`).
    pub needs_toolchains: Vec<String>,
    /// Rustup components the build needs, each as `"<toolchain>:<component>"`
    /// (e.g. `"nightly:clippy"`). The toolchain half is prefix-matched exactly
    /// like [`Self::needs_toolchains`]; the component half must match exactly.
    ///
    /// Components are toolchain-scoped on purpose: a worker holding `clippy`
    /// for `stable` cannot lint a project that pins a nightly, and without this
    /// the scheduler routed lint work to such a worker and surfaced
    /// `error: 'cargo-clippy' is not installed for the toolchain '...'` as a
    /// plain exit 1 — indistinguishable from a real lint failure (bd-vc61a).
    ///
    /// Empty by default, so this constrains nothing until a caller opts in.
    pub needs_components: Vec<String>,
    /// Operator-declared named tools the work needs verified on the worker
    /// (`rch exec --job --require-tool NAME`, or a project's
    /// `jobs.required_tools`).
    ///
    /// Empty by default: no requirement means no gate. A name no worker
    /// declares a probe for is inadmissible EVERYWHERE rather than ignored —
    /// a typo that silently disabled the gate would be worse than a refusal,
    /// because the caller asked for the gate precisely because the job cannot
    /// run without the tool.
    pub needs_tools: Vec<String>,
}

impl CapabilityRequirement {
    /// A bare Rust build requirement (cargo, given wire protocol).
    #[must_use]
    pub fn rust(min_protocol: u32) -> Self {
        Self {
            min_protocol,
            needs_cargo: true,
            ..Self::default()
        }
    }

    /// Require these rustup targets (builder style).
    #[must_use]
    pub fn with_targets<I, S>(mut self, targets: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.needs_targets = targets.into_iter().map(Into::into).collect();
        self
    }

    /// Require a host OS (e.g. `linux` / `darwin`).
    #[must_use]
    pub fn with_os(mut self, os: impl Into<String>) -> Self {
        self.needs_os = Some(os.into());
        self
    }

    /// Require a host arch (e.g. `x86_64` / `aarch64`).
    #[must_use]
    pub fn with_arch(mut self, arch: impl Into<String>) -> Self {
        self.needs_arch = Some(arch.into());
        self
    }

    /// Require verified named tools (builder style).
    #[must_use]
    pub fn with_tools<I, S>(mut self, tools: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.needs_tools = tools.into_iter().map(Into::into).collect();
        self
    }

    /// Require specific rustup toolchains (builder style).
    #[must_use]
    pub fn with_toolchains<I, S>(mut self, toolchains: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.needs_toolchains = toolchains.into_iter().map(Into::into).collect();
        self
    }
}

/// Outcome of an admissibility assessment. Reachability is the caller's concern
/// (did the SSH probe run?); this answers "it ran — is it usable?".
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CapabilityVerdict {
    /// Worker is admissible for the requirement.
    Admissible,
    /// Worker ran the probe but is not admissible, with the incident reason.
    Rejected {
        reason: IncidentReasonCode,
        detail: String,
    },
}

/// Assess admissibility of probed facts against a requirement. Assumes the
/// probe was reachable (non-empty facts); a fully empty parse should be treated
/// as unreachable by the caller, not passed here.
#[must_use]
pub fn assess_admissibility(facts: &ProbedFacts, req: &CapabilityRequirement) -> CapabilityVerdict {
    // The exact-path worker binary must have produced a version (root-good /
    // user-broken: the binary at the configured path is unusable for this user).
    let Some(worker) = &facts.worker else {
        return CapabilityVerdict::Rejected {
            reason: IncidentReasonCode::WrongUserPathWorkerBinary,
            detail: format!(
                "rch-wkr at {} did not report a version as the configured user",
                facts.rch_wkr_path.as_deref().unwrap_or("<unknown path>")
            ),
        };
    };
    if worker.protocol_version < req.min_protocol {
        return CapabilityVerdict::Rejected {
            reason: IncidentReasonCode::WrongUserPathWorkerBinary,
            detail: format!(
                "worker protocol {} < required {}",
                worker.protocol_version, req.min_protocol
            ),
        };
    }
    // Host OS/arch must match the required output triple — a worker that ran the
    // probe fine can still produce the wrong-platform artifact.
    if let Some(needed_os) = &req.needs_os {
        let matches = facts
            .os
            .as_deref()
            .is_some_and(|os| os.eq_ignore_ascii_case(needed_os));
        if !matches {
            return CapabilityVerdict::Rejected {
                reason: IncidentReasonCode::OsArchMismatch,
                detail: format!(
                    "worker OS {} does not satisfy required {}",
                    facts.os.as_deref().unwrap_or("<unknown>"),
                    needed_os
                ),
            };
        }
    }
    if let Some(needed_arch) = &req.needs_arch {
        let matches = facts
            .arch
            .as_deref()
            .is_some_and(|arch| arch.eq_ignore_ascii_case(needed_arch));
        if !matches {
            return CapabilityVerdict::Rejected {
                reason: IncidentReasonCode::OsArchMismatch,
                detail: format!(
                    "worker arch {} does not satisfy required {}",
                    facts.arch.as_deref().unwrap_or("<unknown>"),
                    needed_arch
                ),
            };
        }
    }
    if req.needs_cargo && facts.rust.rustc_version.is_none() {
        return CapabilityVerdict::Rejected {
            reason: IncidentReasonCode::MissingRuntimeToolchainTarget,
            detail: "cargo/rust toolchain not found at the configured user/path".to_string(),
        };
    }
    for needed in &req.needs_toolchains {
        // Match the exact name OR a `-`-suffixed host-triple form, so
        // `nightly-2025-11-01` satisfies
        // `nightly-2025-11-01-x86_64-unknown-linux-gnu` but NOT
        // `nightly-2025-11-010` (bd-review-toolchain-prefix).
        if !facts
            .rust
            .toolchains
            .iter()
            .any(|t| t == needed || t.starts_with(&format!("{needed}-")))
        {
            return CapabilityVerdict::Rejected {
                reason: IncidentReasonCode::MissingRuntimeToolchainTarget,
                detail: format!("missing rustup toolchain {needed}"),
            };
        }
    }
    for needed in &req.needs_targets {
        if !facts.rust.targets.iter().any(|t| t == needed) {
            return CapabilityVerdict::Rejected {
                reason: IncidentReasonCode::MissingRuntimeToolchainTarget,
                detail: format!("missing rustup target {needed}"),
            };
        }
    }
    // Toolchain-scoped components. Rejecting here turns "this worker cannot run
    // the command" into a routing decision, so the caller sees a capability
    // rejection (and, if every candidate is rejected, the same local fallback a
    // missing target produces) instead of a 2-second exit 1 that reads like a
    // lint failure (bd-vc61a).
    for needed in &req.needs_components {
        let Some((toolchain, component)) = needed.split_once(':') else {
            return CapabilityVerdict::Rejected {
                reason: IncidentReasonCode::MissingRuntimeToolchainTarget,
                detail: format!(
                    "malformed component requirement {needed}; expected <toolchain>:<component>"
                ),
            };
        };
        let prefix = format!("{toolchain}-");
        let present = facts.rust.components.iter().any(|entry| {
            entry.split_once(':').is_some_and(|(owner, name)| {
                name == component && (owner == toolchain || owner.starts_with(&prefix))
            })
        });
        if !present {
            return CapabilityVerdict::Rejected {
                reason: IncidentReasonCode::MissingRuntimeToolchainTarget,
                detail: format!("missing rustup component {component} for toolchain {toolchain}"),
            };
        }
    }
    // Operator-declared named tools. Rejecting here — before any slot is
    // reserved — is the point: a job that needs a tool this worker lacks would
    // otherwise hold a reservation only to fail on the worker, and the failure
    // would be indistinguishable from the job's own nonzero exit (job mode
    // surfaces remote status verbatim).
    for needed in &req.needs_tools {
        // Share exact-version semantics with the daemon's live selection gate.
        // A versioned requirement cannot fall back to an operator-named label.
        if let Some(check) = crate::tool_requirement::check_exact_runtime(
            needed,
            facts.runtimes.node_version.as_deref(),
            facts.runtimes.bun_version.as_deref(),
        ) {
            if let Err(detail) = check {
                return CapabilityVerdict::Rejected {
                    reason: IncidentReasonCode::MissingRuntimeToolchainTarget,
                    detail,
                };
            }
            continue;
        }
        if !facts.has_tool(needed) {
            return CapabilityVerdict::Rejected {
                reason: IncidentReasonCode::MissingRuntimeToolchainTarget,
                detail: if facts.declares_tool(needed) {
                    format!(
                        "required tool `{needed}` is declared for this worker but its probe did not succeed"
                    )
                } else {
                    format!("required tool `{needed}` has no declared probe on this worker")
                },
            };
        }
    }
    if req.needs_bun && facts.runtimes.bun_version.is_none() {
        return CapabilityVerdict::Rejected {
            reason: IncidentReasonCode::MissingRuntimeToolchainTarget,
            detail: "bun runtime not found at the configured user/path".to_string(),
        };
    }
    if req.needs_node && facts.runtimes.node_version.is_none() {
        return CapabilityVerdict::Rejected {
            reason: IncidentReasonCode::MissingRuntimeToolchainTarget,
            detail: "node runtime not found at the configured user/path".to_string(),
        };
    }
    if req.needs_nix && facts.runtimes.nix_version.is_none() {
        return CapabilityVerdict::Rejected {
            reason: IncidentReasonCode::MissingRuntimeToolchainTarget,
            detail: "nix (binary + populated /nix/store) not found on the worker".to_string(),
        };
    }
    if req.needs_go && facts.runtimes.go_version.is_none() {
        return CapabilityVerdict::Rejected {
            reason: IncidentReasonCode::MissingRuntimeToolchainTarget,
            detail: "go toolchain not found at the configured user/path".to_string(),
        };
    }
    // Zig cross-build needs BOTH zig and cargo-zigbuild (the requested rustup
    // target std is enforced by the needs_targets loop above). Missing either →
    // this worker can't run it; if every candidate is rejected the build falls
    // back to local, exactly like a missing rustup target.
    if req.needs_zig
        && (facts.runtimes.zig_version.is_none() || facts.runtimes.cargo_zigbuild_version.is_none())
    {
        return CapabilityVerdict::Rejected {
            reason: IncidentReasonCode::MissingRuntimeToolchainTarget,
            detail: "zig + cargo-zigbuild not found on the worker".to_string(),
        };
    }
    CapabilityVerdict::Admissible
}

/// A worker's live admission state, separate from its (structural) capability.
/// Capability says *can this worker ever run the command*; liveness says *is it
/// usable right now*. Keeping them distinct is what lets selection report a
/// missing-capability reason instead of conflating it with an unhealthy or busy
/// worker.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct WorkerLiveness {
    /// The worker's capability facts are too stale to trust.
    pub telemetry_stale: bool,
    /// The worker is healthy (circuit closed, reachable).
    pub healthy: bool,
    /// The worker has at least one free build slot.
    pub has_free_slot: bool,
}

impl WorkerLiveness {
    /// A fresh, healthy, idle worker.
    #[must_use]
    pub fn ready() -> Self {
        Self {
            telemetry_stale: false,
            healthy: true,
            has_free_slot: true,
        }
    }
}

/// The distinct eligibility outcomes selection must tell apart, each with its
/// own incident reason — so an agent learns *why* a worker was not chosen
/// (missing capability vs unhealthy vs busy vs stale facts) rather than a single
/// opaque "no admissible workers".
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum EligibilityVerdict {
    /// Capable, fresh, healthy, and has a free slot.
    Eligible,
    /// The worker structurally cannot run this command.
    MissingCapability {
        reason: IncidentReasonCode,
        detail: String,
    },
    /// Capability is unknowable because the facts are stale.
    StaleTelemetry,
    /// Capable but unhealthy (circuit open / unreachable).
    Unhealthy,
    /// Capable and healthy but at capacity right now.
    Busy,
}

impl EligibilityVerdict {
    /// The incident reason for this verdict (`None` when eligible).
    #[must_use]
    pub fn reason(&self) -> Option<IncidentReasonCode> {
        match self {
            EligibilityVerdict::Eligible => None,
            EligibilityVerdict::MissingCapability { reason, .. } => Some(*reason),
            EligibilityVerdict::StaleTelemetry => Some(IncidentReasonCode::TelemetryStale),
            EligibilityVerdict::Unhealthy => Some(IncidentReasonCode::CircuitOpen),
            EligibilityVerdict::Busy => Some(IncidentReasonCode::InsufficientSlots),
        }
    }

    /// Whether the worker is eligible to run the command.
    #[must_use]
    pub fn is_eligible(&self) -> bool {
        matches!(self, EligibilityVerdict::Eligible)
    }
}

/// Assess a worker's full eligibility for a command, distinguishing missing
/// capability from an unhealthy or busy worker. Pure and total.
///
/// Precedence: stale facts (can't trust capability) first; then a structural
/// capability rejection (no point waiting for a slot it can never use); then
/// health; then capacity. Capacity/health are transient, so they only matter
/// once the worker is known capable on fresh facts.
#[must_use]
pub fn assess_worker_eligibility(
    facts: &ProbedFacts,
    req: &CapabilityRequirement,
    liveness: &WorkerLiveness,
) -> EligibilityVerdict {
    if liveness.telemetry_stale {
        return EligibilityVerdict::StaleTelemetry;
    }
    if let CapabilityVerdict::Rejected { reason, detail } = assess_admissibility(facts, req) {
        return EligibilityVerdict::MissingCapability { reason, detail };
    }
    if !liveness.healthy {
        return EligibilityVerdict::Unhealthy;
    }
    if !liveness.has_free_slot {
        return EligibilityVerdict::Busy;
    }
    EligibilityVerdict::Eligible
}

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

    #[test]
    fn test_normalize_probed_os_collapses_windows_shells() {
        // The probe is a POSIX script, so on Windows it runs under MSYS2 /
        // Git-Bash / Cygwin and `uname -s` names the emulation layer, not the OS.
        for raw in [
            "MINGW64_NT-10.0-26100",
            "mingw32_nt-10.0-22631",
            "MSYS_NT-10.0-26100",
            "CYGWIN_NT-10.0",
        ] {
            assert_eq!(normalize_probed_os(raw), "windows", "raw={raw}");
        }
    }

    #[test]
    fn test_normalize_probed_os_passes_through_unix() {
        assert_eq!(normalize_probed_os("Linux"), "linux");
        assert_eq!(normalize_probed_os("Darwin"), "darwin");
        assert_eq!(normalize_probed_os(" linux "), "linux");
    }

    #[test]
    fn test_parse_normalizes_windows_os_fact() {
        let facts =
            parse_capability_probe("RCH_FACT os=MINGW64_NT-10.0-26100\nRCH_FACT arch=x86_64\n");
        assert_eq!(facts.os.as_deref(), Some("windows"));
    }

    fn spec() -> ProbeSpec {
        let mut s = ProbeSpec::new("rch", "/home/rch/.local/bin/rch-wkr");
        s.cargo_path = Some("/home/rch/.cargo/bin/cargo".to_string());
        s.disk_roots = vec!["/data/tmp".to_string()];
        s
    }

    fn tool(name: &str, argv: &[&str]) -> NamedToolProbe {
        NamedToolProbe::new(name, argv.iter().copied()).expect("valid probe")
    }

    #[test]
    fn named_tool_probe_refuses_names_that_cannot_survive_the_wire_format() {
        // A fact line is `RCH_FACT tool=<name>`, parsed on the FIRST `=`. A name
        // carrying `=`, whitespace or a control character would parse back as a
        // different name — a silent capability lie. Refuse at declaration.
        for bad in [
            "",
            "clang=1",
            "two words",
            "new\nline",
            "tab\there",
            "sh;rm",
        ] {
            assert!(
                NamedToolProbe::new(bad, ["true"]).is_err(),
                "{bad:?} must be refused"
            );
        }
        for good in [
            "clang",
            "ld.lld",
            "cargo-nextest",
            "gcc-14",
            "llvm_cov",
            "g++",
        ] {
            assert!(
                NamedToolProbe::new(good, ["true"]).is_ok(),
                "{good:?} must be accepted"
            );
        }
        // A probe with no argv can never prove anything.
        assert!(matches!(
            NamedToolProbe::new("clang", Vec::<String>::new()),
            Err(ToolProbeError::EmptyCommand { .. })
        ));
    }

    #[test]
    fn script_probes_declared_tools_by_exit_status_and_reports_both_outcomes() {
        let spec = spec().with_tools([
            tool("clang", &["clang", "--version"]),
            tool("ld.lld", &["/usr/bin/ld.lld", "--version"]),
        ]);
        let script = build_capability_probe_script(&spec);
        // The operator's exact argv, shell-quoted, with output discarded: the
        // verification is the exit status, and a chatty tool must not be able to
        // inject its own RCH_FACT lines into the stream.
        assert!(script.contains("if 'clang' '--version' >/dev/null 2>&1;"));
        assert!(script.contains("if '/usr/bin/ld.lld' '--version' >/dev/null 2>&1;"));
        assert!(script.contains("tool=%s"));
        assert!(script.contains("tool_absent=%s"));
        // No tools declared => no tool probes at all (no requirement, no gate).
        assert!(!build_capability_probe_script(&self::spec()).contains("tool="));
    }

    #[test]
    fn parsed_tool_facts_separate_verified_from_declared_but_failing() {
        let facts = parse_capability_probe(
            "RCH_FACT os=linux\nRCH_FACT tool=clang\nRCH_FACT tool_absent=ld.lld\nRCH_FACT tool=clang\n",
        );
        assert_eq!(facts.tools_present, vec!["clang".to_string()], "deduped");
        assert_eq!(facts.tools_absent, vec!["ld.lld".to_string()]);
        assert!(facts.has_tool("clang"));
        assert!(!facts.has_tool("ld.lld"));
        // Declared-but-failing is NOT the same as never declared.
        assert!(facts.declares_tool("ld.lld"));
        assert!(!facts.declares_tool("wild"));
    }

    #[test]
    fn tool_requirements_gate_admissibility_and_unknown_names_are_inadmissible() {
        let facts = parse_capability_probe(
            "RCH_FACT os=linux\nRCH_FACT arch=x86_64\nRCH_FACT rch_wkr_path=/w\n\
             RCH_FACT worker_version=1\nRCH_FACT worker_protocol=9\n\
             RCH_FACT cargo_version=cargo 1.99\nRCH_FACT tool=clang\nRCH_FACT tool_absent=ld.lld\n",
        );
        let base = CapabilityRequirement::rust(1);
        // No requirement => no gate.
        assert_eq!(
            assess_admissibility(&facts, &base),
            CapabilityVerdict::Admissible
        );
        // Verified tool => admissible.
        assert_eq!(
            assess_admissibility(&facts, &base.clone().with_tools(["clang"])),
            CapabilityVerdict::Admissible
        );
        // Declared but failing probe => rejected, and the detail says so.
        let CapabilityVerdict::Rejected { reason, detail } =
            assess_admissibility(&facts, &base.clone().with_tools(["ld.lld"]))
        else {
            panic!("a failing probe must not be admissible");
        };
        assert_eq!(reason, IncidentReasonCode::MissingRuntimeToolchainTarget);
        assert!(detail.contains("did not succeed"), "{detail}");
        // A name NOBODY declared (typo, or a tool the fleet never probed) is
        // inadmissible rather than silently unconstrained — the caller asked
        // for the gate because the job cannot run without the tool.
        let CapabilityVerdict::Rejected { detail, .. } =
            assess_admissibility(&facts, &base.clone().with_tools(["clanggg"]))
        else {
            panic!("an undeclared tool must not be admissible");
        };
        assert!(detail.contains("no declared probe"), "{detail}");
        // Every required tool must hold, not just one.
        assert!(matches!(
            assess_admissibility(&facts, &base.with_tools(["clang", "ld.lld"])),
            CapabilityVerdict::Rejected { .. }
        ));
    }

    #[test]
    fn script_uses_exact_paths_and_probes_all_component_facts() {
        let script = build_capability_probe_script(&spec());
        // Exact rch-wkr path is embedded (not PATH-resolved).
        assert!(script.contains("/home/rch/.local/bin/rch-wkr"));
        assert!(script.contains("/home/rch/.cargo/bin/cargo"));
        // Probes every required capability dimension.
        assert!(script.contains("--version"));
        assert!(script.contains("--protocol-version"));
        assert!(script.contains("toolchain list"));
        assert!(script.contains("target list --installed"));
        // Components must be enumerated PER toolchain, not for the default one:
        // a worker holding clippy for `stable` still cannot lint a pinned
        // nightly, and probing only the default would report it healthy
        // (bd-vc61a).
        assert!(script.contains("component list --installed --toolchain"));
        assert!(script.contains("rustc -vV"));
        assert!(script.contains("sub(\"-\" h \"$\""));
        assert!(
            script.contains("for tc in"),
            "component probe must loop over every installed toolchain"
        );
        assert!(script.contains("bun --version"));
        assert!(script.contains("df -Pk"));
        assert!(script.contains("df -Pi"));
        // Quoted disk root.
        assert!(script.contains("'/data/tmp'"));
    }

    #[cfg(unix)]
    #[test]
    fn live_disk_probe_measures_future_roots_without_creating_them() {
        let root = tempfile::tempdir().unwrap();
        let parent = root.path().join("worker's build volume");
        std::fs::create_dir(&parent).unwrap();
        let future = parent.join("not-created/cargo-target");
        let not_directory = root.path().join("file");
        std::fs::write(&not_directory, b"not a build root").unwrap();
        let mut spec = ProbeSpec::new("probe", "/not-installed/rch-wkr");
        spec.disk_roots = vec![
            parent.to_str().unwrap().to_owned(),
            future.to_str().unwrap().to_owned(),
            not_directory.to_str().unwrap().to_owned(),
        ];
        let output = std::process::Command::new("sh")
            .arg("-c")
            .arg(build_capability_probe_script(&spec))
            .output()
            .unwrap();
        assert!(
            output.status.success(),
            "{}",
            String::from_utf8_lossy(&output.stderr)
        );
        let facts = parse_capability_probe(std::str::from_utf8(&output.stdout).unwrap());
        assert_eq!(facts.disk_roots.len(), 2, "{:?}", facts.disk_roots);
        assert_eq!(facts.disk_roots[0].path, spec.disk_roots[0]);
        assert_eq!(facts.disk_roots[1].path, spec.disk_roots[1]);
        assert!(facts.disk_roots[0].total_bytes > 0);
        assert_eq!(
            facts.disk_roots[0].total_bytes,
            facts.disk_roots[1].total_bytes
        );
        assert!(
            !future.exists(),
            "a read-only disk probe must not create build directories"
        );
        assert_eq!(std::fs::read(&not_directory).unwrap(), b"not a build root");
    }

    fn good_output() -> &'static str {
        "RCH_FACT os=linux\n\
         RCH_FACT arch=x86_64\n\
         RCH_FACT user=rch\n\
         RCH_FACT rch_wkr_path=/home/rch/.local/bin/rch-wkr\n\
         RCH_FACT worker_version=1.0.41\n\
         RCH_FACT worker_protocol=3\n\
         RCH_FACT cargo_version=cargo 1.98.0-nightly\n\
         RCH_FACT toolchain=stable\n\
         RCH_FACT toolchain=nightly-2026-05-22\n\
         RCH_FACT target=x86_64-unknown-linux-gnu\n\
         RCH_FACT target=wasm32-unknown-unknown\n\
         RCH_FACT component=stable:clippy\n\
         RCH_FACT component=stable:rustfmt\n\
         RCH_FACT component=nightly-2026-05-22:rustfmt\n\
         RCH_FACT component=malformed-without-separator\n\
         RCH_FACT bun_version=1.1.0\n\
         RCH_FACT disk=/data/tmp;1048576;524288;900000\n\
         incidental noise line that must be ignored\n"
    }

    #[test]
    fn parses_full_probe_output() {
        let f = parse_capability_probe(good_output());
        assert_eq!(f.os.as_deref(), Some("linux"));
        assert_eq!(f.arch.as_deref(), Some("x86_64"));
        assert_eq!(f.probed_user.as_deref(), Some("rch"));
        let w = f.worker.as_ref().unwrap();
        assert_eq!(w.version, "1.0.41");
        assert_eq!(w.protocol_version, 3);
        assert_eq!(w.rch_wkr_path, "/home/rch/.local/bin/rch-wkr");
        assert_eq!(f.rust.toolchains, vec!["stable", "nightly-2026-05-22"]);
        assert!(f.rust.targets.iter().any(|t| t == "wasm32-unknown-unknown"));
        // Components stay toolchain-qualified, and the malformed line carrying
        // no `<toolchain>:<name>` separator is dropped rather than stored under
        // an empty owner -- an unqualified entry would answer `has_component`
        // for EVERY toolchain, which is the fail-open direction.
        assert_eq!(
            f.rust.components,
            vec![
                "stable:clippy",
                "stable:rustfmt",
                "nightly-2026-05-22:rustfmt"
            ]
        );
        assert_eq!(f.runtimes.bun_version.as_deref(), Some("1.1.0"));
        assert_eq!(f.disk_roots.len(), 1);
        assert_eq!(f.disk_roots[0].path, "/data/tmp");
        assert_eq!(f.disk_roots[0].available_bytes, 524288 * 1024);
        assert_eq!(f.disk_roots[0].available_inodes, 900000);
        assert_eq!(
            f.target_triple().as_deref(),
            Some("x86_64-unknown-linux-gnu")
        );
    }

    fn req_wasm() -> CapabilityRequirement {
        CapabilityRequirement {
            needs_targets: vec!["wasm32-unknown-unknown".to_string()],
            min_protocol: 3,
            needs_cargo: true,
            ..CapabilityRequirement::default()
        }
    }

    #[test]
    fn admissible_when_all_capabilities_present() {
        let f = parse_capability_probe(good_output());
        assert_eq!(
            assess_admissibility(&f, &req_wasm()),
            CapabilityVerdict::Admissible
        );
    }

    #[test]
    fn component_requirement_is_toolchain_scoped() {
        let f = parse_capability_probe(good_output());
        let with_components = |needed: &[&str]| CapabilityRequirement {
            needs_components: needed.iter().map(|s| (*s).to_string()).collect(),
            ..req_wasm()
        };

        // Present for the toolchain that actually owns it, including the
        // `-`-suffixed pinned form.
        assert_eq!(
            assess_admissibility(&f, &with_components(&["stable:clippy"])),
            CapabilityVerdict::Admissible
        );
        assert_eq!(
            assess_admissibility(&f, &with_components(&["nightly:rustfmt"])),
            CapabilityVerdict::Admissible
        );

        // The regression this exists to catch: the worker HAS clippy, but for
        // `stable` only. A pinned-nightly lint must not be routed here.
        match assess_admissibility(&f, &with_components(&["nightly:clippy"])) {
            CapabilityVerdict::Rejected { reason, detail } => {
                assert_eq!(reason, IncidentReasonCode::MissingRuntimeToolchainTarget);
                assert!(
                    detail.contains("clippy") && detail.contains("nightly"),
                    "rejection must name the component and toolchain: {detail}"
                );
            }
            other => panic!("expected rejection, got {other:?}"),
        }

        // A requirement that cannot say which toolchain it means is rejected
        // rather than matched loosely.
        match assess_admissibility(&f, &with_components(&["clippy"])) {
            CapabilityVerdict::Rejected { detail, .. } => {
                assert!(detail.contains("malformed"), "got {detail}");
            }
            other => panic!("expected rejection, got {other:?}"),
        }
    }

    #[test]
    fn no_component_requirement_constrains_nothing() {
        // Default is empty, so every existing caller keeps its current routing.
        // A worker reporting no components at all stays admissible.
        let out = "RCH_FACT os=linux\nRCH_FACT arch=x86_64\nRCH_FACT user=rch\n\
                   RCH_FACT rch_wkr_path=/home/rch/.local/bin/rch-wkr\n\
                   RCH_FACT worker_version=1.0.41\nRCH_FACT worker_protocol=3\n\
                   RCH_FACT cargo_version=cargo 1.98.0-nightly\n\
                   RCH_FACT target=wasm32-unknown-unknown\n";
        let f = parse_capability_probe(out);
        assert!(f.rust.components.is_empty());
        assert_eq!(
            assess_admissibility(&f, &req_wasm()),
            CapabilityVerdict::Admissible
        );
    }

    #[test]
    fn user_broken_binary_rejected_distinct_from_unreachable() {
        // The probe RAN (host facts present) but the exact-path binary produced
        // no version — root-good / user-broken. This is capability rejection,
        // not unreachability.
        let out = "RCH_FACT os=linux\nRCH_FACT arch=x86_64\n\
                   RCH_FACT rch_wkr_path=/home/rch/.local/bin/rch-wkr\n";
        let f = parse_capability_probe(out);
        assert!(f.worker.is_none());
        match assess_admissibility(&f, &req_wasm()) {
            CapabilityVerdict::Rejected { reason, .. } => {
                assert_eq!(reason, IncidentReasonCode::WrongUserPathWorkerBinary);
            }
            other => panic!("expected rejection, got {other:?}"),
        }
    }

    #[test]
    fn missing_rustup_rejected_for_cargo_requirement() {
        // Worker fine, but no cargo/toolchains were probed.
        let out = "RCH_FACT os=linux\nRCH_FACT arch=x86_64\n\
                   RCH_FACT worker_version=1.0.41\nRCH_FACT worker_protocol=3\n";
        let f = parse_capability_probe(out);
        match assess_admissibility(&f, &req_wasm()) {
            CapabilityVerdict::Rejected { reason, detail } => {
                assert_eq!(reason, IncidentReasonCode::MissingRuntimeToolchainTarget);
                assert!(detail.contains("cargo"));
            }
            other => panic!("expected rejection, got {other:?}"),
        }
    }

    #[test]
    fn missing_wasm_target_rejected() {
        // Has cargo + protocol, but lacks the wasm target.
        let out = "RCH_FACT worker_version=1.0.41\nRCH_FACT worker_protocol=3\n\
                   RCH_FACT cargo_version=cargo 1.98\nRCH_FACT target=x86_64-unknown-linux-gnu\n";
        let f = parse_capability_probe(out);
        match assess_admissibility(&f, &req_wasm()) {
            CapabilityVerdict::Rejected { reason, detail } => {
                assert_eq!(reason, IncidentReasonCode::MissingRuntimeToolchainTarget);
                assert!(detail.contains("wasm32-unknown-unknown"));
            }
            other => panic!("expected rejection, got {other:?}"),
        }
    }

    #[test]
    fn stale_worker_protocol_rejected() {
        let out = "RCH_FACT worker_version=0.9.0\nRCH_FACT worker_protocol=1\n\
                   RCH_FACT cargo_version=cargo 1.98\nRCH_FACT target=wasm32-unknown-unknown\n";
        let f = parse_capability_probe(out);
        match assess_admissibility(&f, &req_wasm()) {
            CapabilityVerdict::Rejected { reason, detail } => {
                assert_eq!(reason, IncidentReasonCode::WrongUserPathWorkerBinary);
                assert!(detail.contains("protocol"));
            }
            other => panic!("expected rejection, got {other:?}"),
        }
    }

    #[test]
    fn empty_output_parses_empty() {
        // A fully empty parse is the caller's signal of unreachability.
        let f = parse_capability_probe("");
        assert_eq!(f, ProbedFacts::default());
        assert!(f.worker.is_none());
        assert!(f.os.is_none());
    }

    // --- 12.3: OS/arch, runtime, toolchain capability dimensions -----------

    #[test]
    fn os_mismatch_rejected_with_osarch_reason() {
        // A linux worker cannot produce a darwin artifact.
        let f = parse_capability_probe(good_output());
        let req = CapabilityRequirement::rust(3).with_os("darwin");
        match assess_admissibility(&f, &req) {
            CapabilityVerdict::Rejected { reason, detail } => {
                assert_eq!(reason, IncidentReasonCode::OsArchMismatch);
                assert!(detail.contains("darwin"));
            }
            other => panic!("expected OS-mismatch rejection, got {other:?}"),
        }
        // The matching OS is admissible.
        assert_eq!(
            assess_admissibility(&f, &CapabilityRequirement::rust(3).with_os("linux")),
            CapabilityVerdict::Admissible
        );
    }

    #[test]
    fn arch_mismatch_rejected_with_osarch_reason() {
        let f = parse_capability_probe(good_output());
        let req = CapabilityRequirement::rust(3).with_arch("aarch64");
        match assess_admissibility(&f, &req) {
            CapabilityVerdict::Rejected { reason, .. } => {
                assert_eq!(reason, IncidentReasonCode::OsArchMismatch);
            }
            other => panic!("expected arch-mismatch rejection, got {other:?}"),
        }
    }

    #[test]
    fn missing_node_runtime_rejected() {
        // good_output has bun but no node.
        let f = parse_capability_probe(good_output());
        let mut req = CapabilityRequirement::rust(3);
        req.needs_node = true;
        match assess_admissibility(&f, &req) {
            CapabilityVerdict::Rejected { reason, detail } => {
                assert_eq!(reason, IncidentReasonCode::MissingRuntimeToolchainTarget);
                assert!(detail.contains("node"));
            }
            other => panic!("expected node rejection, got {other:?}"),
        }
        // bun IS present, so a bun requirement is admissible.
        let mut bun_req = CapabilityRequirement::rust(3);
        bun_req.needs_bun = true;
        assert_eq!(
            assess_admissibility(&f, &bun_req),
            CapabilityVerdict::Admissible
        );
    }

    #[test]
    fn specific_toolchain_prefix_matched() {
        let f = parse_capability_probe(good_output());
        // Installed nightly-2026-05-22 satisfies the bare prefix.
        let ok = CapabilityRequirement::rust(3).with_toolchains(["nightly-2026-05-22"]);
        assert_eq!(assess_admissibility(&f, &ok), CapabilityVerdict::Admissible);
        // A different pinned nightly is missing.
        let missing = CapabilityRequirement::rust(3).with_toolchains(["nightly-2025-01-01"]);
        match assess_admissibility(&f, &missing) {
            CapabilityVerdict::Rejected { reason, detail } => {
                assert_eq!(reason, IncidentReasonCode::MissingRuntimeToolchainTarget);
                assert!(detail.contains("nightly-2025-01-01"));
            }
            other => panic!("expected toolchain rejection, got {other:?}"),
        }
    }

    // --- 12.3: eligibility distinguishes missing-capability / unhealthy /
    //           busy / stale ------------------------------------------------

    #[test]
    fn eligible_when_capable_fresh_healthy_idle() {
        let f = parse_capability_probe(good_output());
        let v = assess_worker_eligibility(&f, &req_wasm(), &WorkerLiveness::ready());
        assert_eq!(v, EligibilityVerdict::Eligible);
        assert!(v.is_eligible());
        assert_eq!(v.reason(), None);
    }

    #[test]
    fn no_worker_with_runtime_is_missing_capability_not_busy() {
        // A worker without cargo, even if idle+healthy, is missing-capability —
        // distinct from a busy or unhealthy worker.
        let out = "RCH_FACT os=linux\nRCH_FACT arch=x86_64\n\
                   RCH_FACT worker_version=1.0.41\nRCH_FACT worker_protocol=3\n";
        let f = parse_capability_probe(out);
        let v = assess_worker_eligibility(&f, &req_wasm(), &WorkerLiveness::ready());
        assert_eq!(
            v.reason(),
            Some(IncidentReasonCode::MissingRuntimeToolchainTarget)
        );
        assert!(matches!(v, EligibilityVerdict::MissingCapability { .. }));
    }

    #[test]
    fn os_arch_mismatch_surfaces_as_missing_capability() {
        let f = parse_capability_probe(good_output());
        let req = CapabilityRequirement::rust(3).with_os("darwin");
        let v = assess_worker_eligibility(&f, &req, &WorkerLiveness::ready());
        assert_eq!(v.reason(), Some(IncidentReasonCode::OsArchMismatch));
    }

    #[test]
    fn unhealthy_and_busy_are_distinct_from_missing_capability() {
        let f = parse_capability_probe(good_output());
        // Capable but circuit-open => Unhealthy.
        let unhealthy = WorkerLiveness {
            telemetry_stale: false,
            healthy: false,
            has_free_slot: true,
        };
        assert_eq!(
            assess_worker_eligibility(&f, &req_wasm(), &unhealthy).reason(),
            Some(IncidentReasonCode::CircuitOpen)
        );
        // Capable + healthy but no slot => Busy.
        let busy = WorkerLiveness {
            telemetry_stale: false,
            healthy: true,
            has_free_slot: false,
        };
        assert_eq!(
            assess_worker_eligibility(&f, &req_wasm(), &busy).reason(),
            Some(IncidentReasonCode::InsufficientSlots)
        );
    }

    #[test]
    fn stale_capability_cache_is_distinct_reason() {
        let f = parse_capability_probe(good_output());
        let stale = WorkerLiveness {
            telemetry_stale: true,
            healthy: true,
            has_free_slot: true,
        };
        let v = assess_worker_eligibility(&f, &req_wasm(), &stale);
        assert_eq!(v, EligibilityVerdict::StaleTelemetry);
        assert_eq!(v.reason(), Some(IncidentReasonCode::TelemetryStale));
    }

    #[test]
    fn capability_refresh_changes_eligibility() {
        // Before refresh: worker lacks the wasm target => missing capability.
        let before = parse_capability_probe(
            "RCH_FACT os=linux\nRCH_FACT arch=x86_64\n\
             RCH_FACT worker_version=1.0.41\nRCH_FACT worker_protocol=3\n\
             RCH_FACT cargo_version=cargo 1.98\nRCH_FACT target=x86_64-unknown-linux-gnu\n",
        );
        let v_before = assess_worker_eligibility(&before, &req_wasm(), &WorkerLiveness::ready());
        assert!(matches!(
            v_before,
            EligibilityVerdict::MissingCapability { .. }
        ));

        // After refresh: the wasm target was installed => now eligible.
        let after = parse_capability_probe(good_output());
        let v_after = assess_worker_eligibility(&after, &req_wasm(), &WorkerLiveness::ready());
        assert_eq!(v_after, EligibilityVerdict::Eligible);
    }

    #[test]
    fn remote_worker_binary_path_is_absolute_per_user() {
        assert_eq!(
            remote_worker_binary_path("root", None),
            "/root/.local/bin/rch-wkr"
        );
        assert_eq!(
            remote_worker_binary_path("ubuntu", Some("linux")),
            "/home/ubuntu/.local/bin/rch-wkr"
        );
        // The path must be absolute: the probe script single-quotes it, so a
        // literal `~` would never expand and `[ -x ... ]` would always fail.
        for user in ["root", "rch", "ubuntu"] {
            let path = remote_worker_binary_path(user, None);
            assert!(!path.contains('~'), "{path} must not contain a tilde");
            assert!(path.starts_with('/'), "{path} must be absolute");
            let spec = ProbeSpec::new(user, path);
            let script = build_capability_probe_script(&spec);
            // The shell-quoted binary path embedded in the script is absolute,
            // so `[ -x ... ]` resolves on the real worker.
            assert!(script.contains("/.local/bin/rch-wkr'"));
            assert!(!script.contains("'~/.local/bin/rch-wkr'"));
        }
    }

    #[test]
    fn remote_worker_binary_path_uses_git_bash_home_for_windows() {
        assert_eq!(
            remote_worker_binary_path("jeffr", Some("windows")),
            "/c/Users/jeffr/.local/bin/rch-wkr.exe"
        );
        assert_eq!(
            remote_worker_binary_path("jeffr", Some("WINDOWS")),
            "/c/Users/jeffr/.local/bin/rch-wkr.exe"
        );
    }
}