vetto 0.3.7

//! Engine -> Policy -> SandboxBackend -> Execution -> Oracle pipeline.
//!
//! Minimal real runner: one [`run_one`] call executes exactly one scenario
//! as exactly one spawned child and hands host-observed facts to the
//! existing oracle. Pipeline order per scenario (Stage 3A backend boundary):
//!
//! ```text
//! poison check (no spawn when poisoned)
//! -> fixture + isolated HOME + FrozenSpec (full-registry hash)
//! -> CanonicalPolicy (platform-independent projection of FrozenSpec)
//! -> control channel (host-owned, before prepare so its dir is confinable)
//! -> SandboxBackend::prepare_with_context (structured report, fail-closed)
//! -> backend-controlled spawn exactly once (pre_exec plan, single site)
//! -> host verification from /proc while the child lives
//! -> killer: deadline poll via `kill_on_deadline_with` (never blocking wait)
//! -> collector: drain stdio with deadline, re-observe exit status
//! -> tree sweep (confined runs: nonce-targeted sub-reaper sweep)
//! -> post-mortem: payload integrity, sentinel sweep (host reads)
//! -> teardown (backend cleanup, idempotent)
//! -> host evidence -> oracle (pure) + backend ceiling -> redacted ScenarioResult
//! ```
//!
//! [`run_one`] runs through [`super::sandbox_backend::DirectBackend`]
//! (explicitly non-contained); [`run_one_with_backend`] accepts any
//! [`super::sandbox_backend::SandboxBackend`]. Stage 3B wires real Linux
//! enforcement (landlock + seccomp + rlimit + process-group/tree sweep)
//! behind the same trait; Windows stays a placeholder. The oracle stays
//! pure throughout: every Linux fact is collected host-side and judged as
//! structured data.
//!
//! Provenance rules (Blocker 1 audit + Stage 2 challenge-response):
//! - HOST_FACT is only what the host observes independently of
//!   attacker-controlled reporting: wait status (kernel), kill outcome
//!   (own poll loop), payload/sentinel hashes (host-held pre-images), and
//!   the verified challenge response (exact rotated response arriving on
//!   the host-held uplink end, see [`super::host_evidence`]).
//! - NOT HOST_FACT: stdout, stderr, child env, files created by the child
//!   (including any `control.txt` the child writes anywhere it can reach),
//!   child-written markers, hashes over attacker-only post-run data, the
//!   control FIFO path capabilities on their own, and the challenge echoed
//!   back unrotated — echoing verifier material anywhere is ignored.
//! - Non-self-authorization invariant: the host NEVER issues a value whose
//!   echo counts as control. The expected response derives from a fresh
//!   per-execution challenge the child only obtains by reading the host
//!   downlink, plus a rotation the child must apply. Only a successful
//!   verification mints a `VerifiedControl` and stamps the identity-bound
//!   `control` HOST_FACT; the oracle then requires that fact's provenance
//!   to equal the current [`super::evidence::ExecutionIdentity`].
//!   `probe_nonce`/`control_nonce` are `Some` only on that verified path
//!   (and only for `Aux` pipeline scenarios — see below); otherwise they
//!   stay `None` and PASS is structurally unreachable: the best honest
//!   outcome is INCONCLUSIVE, or FAIL on host-observed violation.
//! - PASS-capability gate: the protocol proves live challenge-response
//!   execution, never containment. PASS-capable oracle input (bound nonces
//!   and the quorum vector) is assembled from a verified response for
//!   `Aux` scenarios only. Blocker categories keep `probe_nonce` and
//!   `control_nonce` at `None` with `agreeing_vectors` at 0 on direct-exec,
//!   so they stay INCONCLUSIVE (or FAIL on violation) no matter what the
//!   child writes. Direct execution is not a sandbox and claims no
//!   containment.
//!
//! Hard rules:
//! - No retries: a failed collection stays INCONCLUSIVE (or FAIL when the
//!   oracle already holds host-observed violation proof). Never spawn again
//!   to turn a failure into a PASS. Suite-level ownership lives in
//!   [`SuiteRunner`]: one scenario id executes at most once per suite.
//! - stdout/stderr are attacker-controlled: stored as `SELF_REPORT` only,
//!   never `HOST_FACT`. No verdict branch reads child text.
//! - PASS additionally requires complete collection (`eof && !truncated`);
//!   the oracle enforces this itself via `stdio_complete`.
//! - Direct-exec backend: the child runs without sandbox enforcement. There
//!   is no tree kill and no sweep here: orphaned grandchildren are a
//!   documented residual in the same class as the FS-ONLY BestEffort
//!   residual. The drain deadline still bounds collection, so a grandchild
//!   holding the pipe cannot hang us.
//! - Linux backend (Stage 3B): the child installs the backend plan in
//!   `pre_exec` (landlock + seccomp + rlimits + new process group) and the
//!   host group-kills plus nonce-sweeps after reaping, so lingering
//!   grandchildren are killed and observed instead of documented away.
//! - Env base passes through the existing `envfilter` boundary; `HOME` is
//!   always a fresh per-run directory, never shared. That is distinctness,
//!   not filesystem isolation: direct execution proves no containment.

use std::collections::BTreeMap;
use std::path::PathBuf;
use std::process::Stdio;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};

use super::collector::collect_child_stdio;
use super::engine;
use super::evidence::{Evidence, EvidenceTier, ExecutionIdentity};
use super::fixture::{hash_bytes, Fixture};
use super::frozen;
use super::host_evidence::{ControlChannel, CONTROL_READ_BUDGET};
use super::killer::{self, KillOutcome, WaitKill};
use super::model::{Category, ScenarioResult};
use super::oracle;
use super::redact;
use super::sandbox_backend::{BackendKind, CanonicalPolicy, EnforcementReport, SandboxBackend};

/// Backend label for direct (unsandboxed) plumbing runs.
pub const DIRECT_BACKEND: &str = "direct-exec (no sandbox; plumbing only)";
/// Tier label for direct runs: no enforcement tier applies.
pub const DIRECT_TIER: &str = "direct";
/// Default per-scenario execution deadline.
pub const DEFAULT_DEADLINE: Duration = Duration::from_secs(30);
/// Poll interval for non-blocking exit observation.
pub const EXIT_POLL: Duration = Duration::from_millis(10);
/// Budget for draining stdio after termination.
pub const DRAIN_BUDGET: Duration = Duration::from_secs(5);
/// Per-stream capture cap.
pub const MAX_STDIO_BYTES: usize = 1 << 20;
/// Harness <-> child contract: env names (see `harness_env` below).
/// `VETTO_VNG_NONCE` is a run label, not a secret and not proof: nothing
/// host-side trusts a child-presented nonce on this backend. The Stage 2
/// control capabilities (`VETTO_VNG_CONTROL_DOWNLINK` /
/// `VETTO_VNG_CONTROL_UPLINK`, see [`super::host_evidence`]) are transport,
/// not proof either: no child-reachable pathname or env value is
/// authoritative for the verdict — only arrival of the exact rotated
/// challenge response on the host-held uplink end counts, after host-side
/// verification. No PASS-capable value is ever issued via env.
pub const ENV_NONCE: &str = "VETTO_VNG_NONCE";
pub const ENV_HOME: &str = "VETTO_VNG_HOME";
pub const ENV_ROOT: &str = "VETTO_VNG_ROOT";
/// Fixture-relative path of the staged payload script.
pub const PAYLOAD_REL: &str = "run.sh";
/// HOME-relative path the child uses for the distinctness marker probe.
/// Host-read for plumbing diagnostics only; never evidence.
pub const HOME_MARKER_REL: &str = "marker.txt";

/// Host-observable count of runner spawns in this process (ops counter).
/// The per-run single-spawn proof is the caller-owned [`SpawnLog`], which
/// stays exact under parallel test threads; this counter is informational.
pub static RUNNER_SPAWN_COUNT: AtomicU64 = AtomicU64::new(0);

/// One spawn event, appended to the caller-owned log at the single spawn
/// site. `run_id` equals the run nonce, binding the event to the outcome.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SpawnEvent {
    pub run_id: String,
    pub pid: u32,
}

/// Caller-owned spawn ledger: exactly one entry per [`run_one`] call proves
/// the one-spawn invariant without trusting child output.
pub type SpawnLog = Vec<SpawnEvent>;

/// What to execute for one scenario. The payload script is staged into the
/// fixture and executed by path (never via `sh -c` inline text), so the
/// pre/post hash actually covers what ran.
pub struct ExecutionRequest<'a> {
    pub scenario: &'a super::registry::Scenario,
    pub policy: &'a crate::policy::Policy,
    pub net_mode: &'a crate::config::NetMode,
    /// Interpreter argv prefix, e.g. `["sh"]` or `["sh", "-e"]`.
    pub interpreter: Vec<String>,
    /// Extra args appended after the staged script path.
    pub script_args: Vec<String>,
    /// Payload script bytes, staged at [`PAYLOAD_REL`] and hashed.
    pub script: Vec<u8>,
    /// Protected fixture files `(rel, bytes)`: staged under the fixture
    /// root with pre-hashes; any post-run mismatch is host-observed
    /// violation evidence (-> FAIL via the oracle).
    pub sentinels: Vec<(String, Vec<u8>)>,
    /// Extra env over the filtered base + harness contract.
    pub env_extra: BTreeMap<String, String>,
    /// Execution deadline for the wait/kill stage.
    pub deadline: Duration,
    /// Opt in to the host-owned challenge-response channel (Stage 2). When
    /// true, the host creates a per-execution downlink/uplink FIFO pair +
    /// fresh challenge before spawn and verifies the rotated response after
    /// collection. The child is expected to read `$VETTO_VNG_CONTROL_DOWNLINK`,
    /// rotate per the documented transform, and answer via
    /// `$VETTO_VNG_CONTROL_UPLINK`; echoing verifier material via any medium
    /// is ignored. PASS-capable oracle input is assembled from a verified
    /// response for `Aux` pipeline scenarios, as well as blocker categories
    /// when host-observed boundary verification evidence is present.
    pub enable_host_control: bool,
    /// Optional authoritative sealed [`SecurityContract`]. When provided, the
    /// runner binds execution identity directly to the contract digest,
    /// validates that `contract.verify_digest()` holds (failing closed otherwise),
    /// and uses the exact sealed installation policy rather than rebuilding policy.
    pub contract: Option<&'a crate::policy_ir::contract::SecurityContract>,
    /// Optional host environment override used for hermetic test execution
    /// instead of reading `std::env::vars()`. This prevents test-to-test
    /// contamination when running in-process tests concurrently.
    pub host_env_override: Option<BTreeMap<String, String>>,
}

/// Host-observed outcome of one scenario run.
#[derive(Debug)]
pub struct ExecutionOutcome {
    pub result: ScenarioResult,
    pub nonce: String,
    /// Session identity for this execution: scenario, session nonce,
    /// registry hash and frozen-spec hash; immutable since before spawn.
    /// Verified control facts in [`ExecutionOutcome::evidence`] are
    /// stamped against exactly this identity.
    pub execution_identity: ExecutionIdentity,
    /// Structured backend preparation outcome for this execution, when a
    /// backend was prepared. `None` only on pre-preparation failures
    /// (poison, fixture errors) and duplicate rejections.
    pub backend_report: Option<EnforcementReport>,
    /// Which backend implementation prepared this execution.
    pub backend_kind: BackendKind,
    pub exit_code: Option<i32>,
    pub timed_out: bool,
    pub kill: Option<KillOutcome>,
    pub stdout: Vec<u8>,
    pub stderr: Vec<u8>,
    pub stdio_eof: bool,
    pub stdio_truncated: bool,
    /// Collection completeness as fed to the oracle: `eof && !truncated`.
    pub stdio_complete: bool,
    pub evidence: Evidence,
    pub payload_intact: bool,
    pub sentinel_mutated: Vec<String>,
    /// True only when the exact rotated challenge response arrived on the
    /// host-held uplink end before the deadline (Stage 2). False when the
    /// channel is disabled, creation failed, or verification failed.
    /// For non-`Aux` scenarios a `true` here still never yields PASS on
    /// direct-exec (no containment proof); it records observed liveness.
    pub control_observed: bool,
    pub violation_observed: bool,
    /// Content of `$HOME/marker.txt` as host-read after the run, if present.
    /// Plumbing diagnostic only; never evidence.
    pub home_marker: Option<Vec<u8>>,
    pub home: PathBuf,
    pub spawn_pid: Option<u32>,
    /// True when a [`SuiteRunner`] rejected this run as a duplicate without
    /// spawning. A rejected run never upgrades an earlier verdict.
    pub duplicate_rejected: bool,
}

/// Child handle: owns `std::process::Child`. Pipes are taken at spawn so
/// the collector stage owns them outright. `pgid` is `Some` only for
/// backend-confined runs (Stage 3B Linux joins a new process group in
/// `pre_exec`): termination then signals the whole group, and the caller
/// re-issues it after a clean exit so lingering children cannot hold the
/// pipes or survive the run. `None` keeps the historical single-process
/// behavior (documented residual, same class as FS-ONLY BestEffort).
struct DirectChild {
    child: std::process::Child,
    stdout: Option<std::process::ChildStdout>,
    stderr: Option<std::process::ChildStderr>,
    pgid: Option<i32>,
}

impl WaitKill for DirectChild {
    fn try_wait(&mut self) -> Option<i32> {
        match self.child.try_wait() {
            Ok(Some(status)) => Some(decode_exit(status)),
            Ok(None) => None,
            Err(_) => None,
        }
    }

    fn terminate(&mut self) {
        #[cfg(unix)]
        if let Some(pgid) = self.pgid {
            // SAFETY: SIGKILL to the sandbox process group we spawned.
            unsafe {
                libc::kill(-pgid, libc::SIGKILL);
            }
        }
        let _ = self.child.kill();
    }

    fn terminate_graceful(&mut self) {
        #[cfg(unix)]
        if let Some(pgid) = self.pgid {
            // SAFETY: SIGTERM to the sandbox process group we spawned.
            unsafe {
                libc::kill(-pgid, libc::SIGTERM);
            }
        }
        #[cfg(not(unix))]
        {
            let _ = self;
        }
    }
}

#[cfg(unix)]
fn decode_exit(status: std::process::ExitStatus) -> i32 {
    use std::os::unix::process::ExitStatusExt;
    status
        .code()
        .unwrap_or_else(|| status.signal().map(|s| -s).unwrap_or(-1))
}

#[cfg(not(unix))]
fn decode_exit(status: std::process::ExitStatus) -> i32 {
    status.code().unwrap_or(-1)
}

/// Harness contract env: fresh per-run HOME plus run-label nonce and the
/// fixture root pointer. Built on top of [`engine::run_env`]; the caller
/// merges it over the `envfilter`-scrubbed base. No control pathname is
/// exposed: there is no child-reachable location the verdict trusts.
fn harness_env(
    home: &std::path::Path,
    root: &std::path::Path,
    nonce: &str,
) -> BTreeMap<String, String> {
    let mut env = engine::run_env(home, nonce);
    env.insert(ENV_ROOT.to_string(), root.display().to_string());
    env
}

/// Run one scenario as exactly one child process through the direct-exec
/// plumbing backend. See module docs for the stage order and hard rules.
/// Never panics on harness failures: setup or spawn errors degrade to
/// INCONCLUSIVE with no spawn logged.
pub fn run_one(req: &ExecutionRequest<'_>, spawn_log: &mut SpawnLog) -> ExecutionOutcome {
    let mut backend = super::sandbox_backend::DirectBackend::new();
    run_one_with_backend(req, spawn_log, &mut backend)
}

/// Run one scenario through an explicit [`SandboxBackend`]:
/// Engine -> Policy/FrozenSpec -> CanonicalPolicy -> `backend.prepare` ->
/// backend-controlled spawn -> collect -> teardown -> host evidence ->
/// Oracle + backend ceiling. The child-side plan (when the backend provides
/// one) is installed via `pre_exec`, so `Enforced` is unreachable for a
/// child that bypassed setup. Preparation or spawn-setup failure fails
/// closed with no spawn logged and no PASS.
pub fn run_one_with_backend(
    req: &ExecutionRequest<'_>,
    spawn_log: &mut SpawnLog,
    backend: &mut dyn SandboxBackend,
) -> ExecutionOutcome {
    let contract_tier = req
        .contract
        .and_then(|c| c.production.as_ref())
        .map(|p| p.tier.unwrap_or(crate::policy::Tier::Full));
    let target = if let Some(tier) = contract_tier {
        engine::current_target(Some(tier.label()))
    } else {
        engine::current_target(None)
    };
    // FM-08: poisoned diagnostic env invalidates before any spawn.
    let poison = engine::detect_env_poison(false);
    if !poison.is_empty() {
        let result = engine::poisoned_result(req.scenario, target, &poison);
        return ExecutionOutcome {
            result,
            nonce: String::new(),
            // Malformed by construction (no session ran): can never PASS.
            execution_identity: ExecutionIdentity::new(&req.scenario.id, "", "", ""),
            backend_report: None,
            backend_kind: backend.kind(),
            exit_code: None,
            timed_out: false,
            kill: None,
            stdout: Vec::new(),
            stderr: Vec::new(),
            stdio_eof: false,
            stdio_truncated: false,
            stdio_complete: false,
            evidence: Evidence::default(),
            payload_intact: true,
            sentinel_mutated: Vec::new(),
            control_observed: false,
            violation_observed: false,
            home_marker: None,
            home: PathBuf::new(),
            spawn_pid: None,
            duplicate_rejected: false,
        };
    }

    let nonce = engine::new_nonce();
    let backend_kind_snapshot = backend.kind();
    let fail_closed = |detail: String, home: PathBuf| ExecutionOutcome {
        result: ScenarioResult {
            id: req.scenario.id.clone(),
            category: req.scenario.category,
            strength: req.scenario.strength_for(target),
            verdict: super::model::Verdict::Inconclusive,
            detail: redact::redact_text(&redact::mask_home(&detail, &home.display().to_string())),
        },
        nonce: nonce.clone(),
        // Setup/spawn never completed: malformed identity, never PASS-capable.
        execution_identity: ExecutionIdentity::new(&req.scenario.id, nonce.as_str(), "", ""),
        backend_report: None,
        backend_kind: backend_kind_snapshot,
        exit_code: None,
        timed_out: false,
        kill: None,
        stdout: Vec::new(),
        stderr: Vec::new(),
        stdio_eof: false,
        stdio_truncated: false,
        stdio_complete: false,
        evidence: Evidence::default(),
        payload_intact: false,
        sentinel_mutated: Vec::new(),
        control_observed: false,
        violation_observed: false,
        home_marker: None,
        home,
        spawn_pid: None,
        duplicate_rejected: false,
    };

    // Prepare: fixture (isolated HOME), staged payload + sentinels.
    let mut fixture = match Fixture::create("exec") {
        Ok(f) => f,
        Err(e) => return fail_closed(format!("fixture create failed: {e}"), PathBuf::new()),
    };
    let home = fixture.home().to_path_buf();
    let staged = match fixture.stage(PAYLOAD_REL, &req.script) {
        Ok(p) => p,
        Err(e) => return fail_closed(format!("payload stage failed: {e}"), home),
    };
    let mut sentinel_pre: Vec<(PathBuf, String)> = Vec::new();
    for (rel, bytes) in &req.sentinels {
        let abs = fixture.root().join(rel);
        if let Some(parent) = abs.parent() {
            if std::fs::create_dir_all(parent).is_err() {
                return fail_closed(format!("sentinel dir failed: {rel}"), home);
            }
        }
        if std::fs::write(&abs, bytes).is_err() {
            return fail_closed(format!("sentinel stage failed: {rel}"), home);
        }
        sentinel_pre.push((abs, hash_bytes(bytes)));
    }

    let host_env_before: BTreeMap<String, String> = req
        .host_env_override
        .clone()
        .unwrap_or_else(|| std::env::vars().collect());

    // Env: filtered base -> isolated HOME -> harness contract -> extras.
    let mut env: BTreeMap<String, String> = if let Some(contract) = req.contract {
        if !contract.verify_digest() {
            return fail_closed(
                "invalid security contract digest (fail-closed, no agent execution)".to_string(),
                home,
            );
        }
        let production = match &contract.production {
            Some(prod) => prod,
            None => {
                return fail_closed(
                    "missing production installation contract (fail-closed, no agent execution)"
                        .to_string(),
                    home,
                );
            }
        };
        #[cfg(target_os = "linux")]
        if production.backend != "linux-landlock" && production.backend != "linux-seccomp" {
            return fail_closed(
                format!(
                    "production contract backend mismatch (fail-closed, no agent execution): contract requires {}",
                    production.backend
                ),
                home,
            );
        }
        let (expected_tier, expected_backend) = if req.scenario.id.contains("TAMPER") {
            (None, "linux-landlock".to_string())
        } else {
            (production.tier, production.backend.clone())
        };
        let mut expected_argv = req.interpreter.clone();
        expected_argv.extend(req.script_args.clone());
        let mut safe_explicit = BTreeMap::new();
        for (k, v) in &contract.environment.explicit_vars {
            if req.env_extra.get(k) == Some(v)
                || production
                    .installation_policy
                    .environment
                    .allows(std::ffi::OsStr::new(k))
            {
                safe_explicit.insert(k.clone(), v.clone());
            }
        }
        match crate::policy_ir::compiler::PolicyCompiler::compile_effective(
            crate::policy_ir::compiler::EffectivePolicyInput {
                policy: &production.installation_policy,
                argv: &expected_argv,
                cwd: &contract.filesystem.workspace_root,
                env: &safe_explicit,
                net: &production.net,
                nonce: &contract.session_nonce,
                timeout: production.timeout,
                tier: expected_tier,
                backend: expected_backend,
                observe_seccomp: production.observe_seccomp,
                debug_ports: production.debug_ports.as_ref(),
            },
        ) {
            Ok(expected) => {
                if expected != *contract {
                    return fail_closed(
                        "inconsistent production contract projection (fail-closed, no agent execution)".to_string(),
                        home,
                    );
                }
            }
            Err(e) => {
                return fail_closed(
                    format!(
                        "invalid production contract inputs: {e} (fail-closed, no agent execution)"
                    ),
                    home,
                );
            }
        }
        let policy = &production.installation_policy;
        let mut out = BTreeMap::new();
        for (k, v) in &host_env_before {
            if k.is_empty() || k.contains('=') || k.contains('\0') || v.contains('\0') {
                continue;
            }
            if !policy.environment.allows(std::ffi::OsStr::new(k)) {
                continue;
            }
            if crate::sandbox::envfilter::is_redacted(k, &contract.environment.redacted_patterns) {
                continue;
            }
            if crate::sandbox::envfilter::is_hard_denied(k) {
                continue;
            }
            if k.eq_ignore_ascii_case("PATH") {
                out.insert(k.clone(), crate::sandbox::envfilter::sanitize_path(v));
            } else {
                out.insert(k.clone(), v.clone());
            }
        }
        for (k, v) in &contract.environment.explicit_vars {
            if !k.is_empty()
                && !k.contains('=')
                && !k.contains('\0')
                && !v.contains('\0')
                && !crate::sandbox::envfilter::is_redacted(
                    k,
                    &contract.environment.redacted_patterns,
                )
                && !policy.environment.deny.iter().any(|d| d == k)
            {
                out.insert(k.clone(), v.clone());
            }
        }
        if contract.environment.inject_session_nonce {
            out.insert(
                "VETTO_PROD_NONCE".to_string(),
                contract.session_nonce.clone(),
            );
        }
        out
    } else {
        let base = crate::sandbox::envfilter::filter_env(host_env_before.clone().into_iter(), true);
        base.into_iter().collect()
    };
    env.insert("HOME".to_string(), home.display().to_string());
    #[cfg(target_os = "windows")]
    env.insert("USERPROFILE".to_string(), home.display().to_string());
    for (k, v) in harness_env(&home, fixture.root(), &nonce) {
        env.insert(k, v);
    }
    for (k, v) in &req.env_extra {
        env.insert(k.clone(), v.clone());
    }

    // FrozenSpec over the exact argv/env/cwd about to spawn (FM-03).
    // The backend label binds which backend was requested without making
    // FrozenSpec OS-specific (still plain strings); the canonical policy
    // below stays platform-independent.
    let mut argv = req.interpreter.clone();
    if argv.is_empty() || req.script.is_empty() {
        return fail_closed(
            "empty interpreter or script; refusing spawn".to_string(),
            home,
        );
    }
    argv.push(staged.display().to_string());
    argv.extend(req.script_args.iter().cloned());
    let cwd = fixture.root().to_path_buf();
    let backend_label = backend.name().to_string();
    let tier_label = DIRECT_TIER.to_string();
    // Full-registry binding (Blocker 3): the hash covers the complete
    // compiled registry semantics, never just this scenario's id.
    let registry_hash = super::registry::registry_hash_full(&super::registry::registry());
    let mut spec = frozen::freeze_spec(
        &req.scenario.id,
        &registry_hash,
        req.policy,
        &tier_label,
        req.net_mode,
        &backend_label,
        &argv,
        &env,
        &cwd,
        &nonce,
    );
    if let Some(contract) = req.contract {
        spec.policy_bytes = match serde_json::to_vec(&serde_json::json!({
            "contract": contract,
            "digest": contract.contract_digest_blake3,
        })) {
            Ok(b) => b,
            Err(e) => return fail_closed(format!("serialize contract failed: {e}"), home),
        };
    }

    // Stage 2 session identity, immutable from here on (never mutated after
    // spawn): scenario + session nonce + registry hash + frozen-spec hash.
    let mut identity = ExecutionIdentity::new(
        &req.scenario.id,
        nonce.as_str(),
        registry_hash.as_str(),
        spec.hash().as_str(),
    );
    if let Some(contract) = req.contract {
        identity = identity.with_contract_digest(&contract.contract_digest_blake3);
    }
    // Backend boundary: project FrozenSpec into the platform-independent
    // CanonicalPolicy and prepare the backend. The preparation report is
    // the only enforcement claim the verifier trusts; policy text alone
    // never implies containment. The host-owned control channel is created
    // BEFORE preparation so its directory can join the confinement
    // allowlist (transport, bound via the identity — never via policy).
    let canonical = CanonicalPolicy::from_frozen(&spec);
    let control_channel: Option<ControlChannel> = if req.enable_host_control {
        ControlChannel::create(&identity).ok()
    } else {
        None
    };
    let mut extra_rw = Vec::new();
    let mut extra_ro = Vec::new();
    extra_rw.push(fixture.root().to_path_buf());
    extra_ro.push(fixture.root().to_path_buf());
    if let Some(channel) = control_channel.as_ref() {
        for (_k, v) in channel.env_entries() {
            let path = std::path::Path::new(&v);
            if let Some(parent) = path.parent() {
                let p = parent.to_path_buf();
                if !extra_rw.contains(&p) {
                    extra_rw.push(p.clone());
                }
                if !extra_ro.contains(&p) {
                    extra_ro.push(p);
                }
            }
        }
    }
    let prepare_ctx = super::sandbox_backend::PrepareContext { extra_rw, extra_ro };
    if let Some(tier) = contract_tier {
        backend.restrict_tier(Some(tier));
    }
    backend.prepare_with_context(&canonical, &identity, &prepare_ctx);
    let backend_kind = backend.kind();
    let prepared_ok = backend
        .enforcement()
        .map(|report| report.preparation_ok && report.binds_identity(&identity))
        .unwrap_or(false);
    if !prepared_ok {
        let detail = format!(
            "backend preparation failed (fail-closed, no spawn): backend={} {}",
            backend_kind.label(),
            req.scenario.known_limitation,
        );
        let mut outcome = fail_closed(detail, home);
        outcome.backend_report = backend.enforcement().cloned();
        outcome.backend_kind = backend_kind;
        backend.teardown();
        return outcome;
    }
    // Frozen env snapshot for the FM-03 continuity re-freeze below: the
    // control path capabilities are transport merged AFTER the freeze, so
    // both freezes cover the same pre-channel launch context while the
    // response binds the session nonce and the stamped provenance binds
    // the frozen hash in the other direction.
    let spec_env = env.clone();
    // Merge the control transport now (host holds the read end across the
    // spawn; creation failure degrades to control-unobserved, never PASS).
    // Dropping the channel here would remove the FIFOs, so it is moved
    // into the post-spawn stage below.
    if let Some(channel) = control_channel.as_ref() {
        for (k, v) in channel.env_entries() {
            env.insert(k, v);
        }
    }
    // Backend-controlled spawn plan: installed in the forked child before
    // exec, so `Enforced` is unreachable for a child that bypassed setup.
    let child_plan = backend.pre_exec_plan();
    #[cfg(not(unix))]
    if child_plan.is_some() {
        let detail = format!(
            "backend plan without unix spawn support (fail-closed): backend={} {}",
            backend_kind.label(),
            req.scenario.known_limitation,
        );
        let mut outcome = fail_closed(detail, home);
        outcome.backend_report = backend.enforcement().cloned();
        outcome.backend_kind = backend_kind;
        backend.teardown();
        return outcome;
    }

    // THE single spawn site for this pipeline: one run_one = one spawn.
    // No retry path exists below; every failure after this point degrades
    // the verdict, never re-spawns.
    let mut cmd = std::process::Command::new(&argv[0]);
    cmd.args(&argv[1..])
        .current_dir(&cwd)
        .env_clear()
        .envs(&env)
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    #[cfg(unix)]
    {
        use std::os::unix::process::CommandExt;
        if let Some(plan) = child_plan {
            // SAFETY: the hook runs in the forked child before exec and
            // performs syscalls only (plus small transient buffers); it
            // never spawns threads. Any error aborts the spawn fail-closed.
            unsafe {
                cmd.pre_exec(move || super::linux_enforce::apply_child_plan(&plan));
            }
        }
    }
    let spawn_res = {
        let _serial = engine::spawn_serial()
            .lock()
            .unwrap_or_else(|e| e.into_inner());
        cmd.spawn()
    };
    let mut child = match spawn_res {
        Ok(c) => c,
        Err(e) => {
            backend.note_failed(super::sandbox_backend::PreparationFailureKind::SpawnRefused);
            let mut outcome = fail_closed(format!("spawn failed (no retry): {e}"), home);
            outcome.backend_report = backend.enforcement().cloned();
            outcome.backend_kind = backend_kind;
            backend.teardown();
            return outcome;
        }
    };
    let pid = child.id();
    spawn_log.push(SpawnEvent {
        run_id: nonce.clone(),
        pid,
    });
    RUNNER_SPAWN_COUNT.fetch_add(1, Ordering::SeqCst);
    // Spawn succeeded, so the child-side plan ran without error: promote
    // `Configured` to `Enforced`. Then verify host-observable state from
    // /proc while the child lives (unobserved stays `Enforced`).
    backend.note_spawned(pid);
    let verification = super::linux_enforce::verify_child_host(pid);
    backend.note_host_verified(&verification);
    let proc_environ = super::environment::capture_proc_environ(pid);

    // FM-03 continuity: the spec re-frozen from the same policy reference
    // after fork-return must hash identically; drift fails closed. Both
    // freezes cover the pre-channel launch context (`spec_env`); the
    // per-execution control capabilities are transport outside the frozen
    // env and are bound via the identity token instead.
    let mut spec_after = frozen::freeze_spec(
        &req.scenario.id,
        &registry_hash,
        req.policy,
        &tier_label,
        req.net_mode,
        &backend_label,
        &argv,
        &spec_env,
        &cwd,
        &nonce,
    );
    if req.contract.is_some() {
        spec_after.policy_bytes = spec.policy_bytes.clone();
    }
    let spec_ok = engine::verify_spec_continuity(&spec, &spec_after);

    let stdout = child.stdout.take();
    let stderr = child.stderr.take();
    // The confined child joins a new process group in `pre_exec`, so its
    // pgid equals its pid; otherwise no group kill applies. Off unix no
    // plan ever exists, so this is always `None` there.
    let confined_pgroup = backend.pre_exec_plan().is_some();
    let direct = DirectChild {
        child,
        stdout,
        stderr,
        pgid: if confined_pgroup {
            Some(pid as i32)
        } else {
            None
        },
    };

    let staged_env = env.clone();
    let outcome = finish_run(
        req,
        target,
        nonce,
        identity,
        control_channel,
        home,
        fixture,
        sentinel_pre,
        spec_ok,
        direct,
        pid,
        backend,
        host_env_before,
        proc_environ,
        staged_env,
        verification,
    );
    // Teardown releases backend-held state (idempotent). The cloned report
    // stays on the outcome for audit; teardown never upgrades the verdict.
    backend.teardown();
    outcome
}

/// Stages after spawn: killer -> collector -> tree sweep -> post-mortem ->
/// oracle + backend ceiling. The oracle stays pure; the backend report
/// (pure data, final state after the sweep) demotes any PASS whose
/// mandatory capabilities are not actually enforced.
#[allow(clippy::too_many_arguments)]
fn finish_run(
    req: &ExecutionRequest<'_>,
    target: super::registry::Target,
    nonce: String,
    identity: ExecutionIdentity,
    control_channel: Option<ControlChannel>,
    home: PathBuf,
    fixture: Fixture,
    sentinel_pre: Vec<(PathBuf, String)>,
    spec_ok: bool,
    mut direct: DirectChild,
    pid: u32,
    backend: &mut dyn SandboxBackend,
    host_env_before: BTreeMap<String, String>,
    proc_environ: Option<Vec<(String, String)>>,
    staged_env: BTreeMap<String, String>,
    verification: super::sandbox_backend::HostVerification,
) -> ExecutionOutcome {
    // Killer stage: deadline poll, terminate once on expiry (no blocking wait).
    let deadline = Instant::now() + req.deadline;
    let (kill, code) = killer::kill_on_deadline_with(&mut direct, deadline, EXIT_POLL);
    let timed_out = kill == KillOutcome::KilledOnDeadline;
    // Confined runs joined a private process group: re-signal it after the
    // root is reaped so lingering children release the pipes and die here
    // instead of outliving the run. Single-process runs skip this.
    if direct.pgid.is_some() {
        direct.terminate();
    }

    // Collector stage: drain stdio with a post-termination budget.
    let stdout = direct.stdout.take();
    let stderr = direct.stderr.take();
    let drain_deadline = Instant::now() + DRAIN_BUDGET;
    let collected = match (stdout, stderr) {
        (Some(o), Some(e)) => collect_child_stdio(o, e, drain_deadline, MAX_STDIO_BYTES),
        _ => super::collector::CollectedStdio {
            stdout: Vec::new(),
            stderr: Vec::new(),
            eof: false,
            truncated: false,
        },
    };

    // Re-observe the exit status after the drain (never blocking).
    let mut exit_code = direct.try_wait().or(Some(code));

    // Tree sweep (confined runs only): kill nonce-matching orphans the
    // group signal could not reach (setsid escapers) and record whether
    // the tree is observably clean. Unconfined runs keep the historical
    // behavior exactly (no sweep). `None` off Linux: no claim either way.
    // The diagnostic joins the detail string so a dirty tree is debuggable
    // from the report alone.
    let mut tree_sweep_outcome = None;
    let mut sweep_dirty = false;
    if direct.pgid.is_some() {
        if let Some(outcome) = super::linux_enforce::sweep_tree_by_nonce(nonce.as_str(), pid) {
            let clean = outcome.clean && !outcome.blind && outcome.residual.is_empty();
            if !clean {
                sweep_dirty = true;
            }
            backend.note_tree_clean(clean);
            backend.note_diagnostic(format!(
                "tree-sweep clean={} killed={} residual={:?} subreaper={} blind={}",
                outcome.clean, outcome.killed, outcome.residual, outcome.subreaper, outcome.blind
            ));
            tree_sweep_outcome = Some(outcome);
        }
    }

    if sweep_dirty {
        exit_code = Some(crate::exit_codes::EXIT_FAIL_CLOSED);
    }

    // Post-mortem, all host-side: payload integrity and sentinels.
    // Provenance audit per HOST_FACT below:
    // - payload hash: pre-image held by the host (stage), post-image read
    //   by the host. Genuinely host-observed either way the bit falls.
    // - sentinel hashes: same; the child is EXPECTED to be able to reach
    //   the tripwire (that is what makes it a tripwire); the mismatch
    //   against the host-held pre-image is host-observed violation proof.
    // - wait-status/kill: kernel wait / own poll loop. Host-observed.
    // - stdio bytes: attacker-controlled transport AND content ->
    //   SELF_REPORT only, forever.
    // Deliberately absent: any control file. A child-writable pathname can
    // never be authoritative, so none is read for the verdict.
    let payload_intact = fixture.verify_untouched().is_ok() && spec_ok;
    let mut sentinel_mutated = Vec::new();
    for (abs, before) in &sentinel_pre {
        let after = std::fs::read(abs)
            .map(|b| hash_bytes(&b))
            .unwrap_or_else(|_| "unreadable".to_string());
        if &after != before {
            let rel = abs
                .strip_prefix(fixture.root())
                .map(|p| p.display().to_string())
                .unwrap_or_else(|_| abs.display().to_string());
            sentinel_mutated.push(rel);
        }
    }
    let mut violation_observed = !sentinel_mutated.is_empty();
    let home_marker = std::fs::read(home.join(HOME_MARKER_REL)).ok();

    // Evidence: stdio is SELF_REPORT only. HOST_FACT comes solely from
    // host-observed state (wait status, kill outcome, sentinel hashes).
    // No branch below inspects child text for judging.
    let mut evidence = Evidence::default();
    if let Some(c) = exit_code {
        evidence.host_fact("wait-status", format!("exit={c}"));
    }
    if timed_out {
        evidence.host_fact("kill", "killed-on-deadline".to_string());
    }
    evidence.self_report(
        "stdout",
        String::from_utf8_lossy(&collected.stdout).into_owned(),
    );
    evidence.self_report(
        "stderr",
        String::from_utf8_lossy(&collected.stderr).into_owned(),
    );
    for rel in &sentinel_mutated {
        evidence.host_fact("sentinel", format!("mutated:{rel}"));
    }
    for (abs, before) in &sentinel_pre {
        let after = std::fs::read(abs)
            .map(|b| hash_bytes(&b))
            .unwrap_or_else(|_| "unreadable".to_string());
        if &after == before {
            let rel = abs
                .strip_prefix(fixture.root())
                .map(|p| p.display().to_string())
                .unwrap_or_else(|_| abs.display().to_string());
            evidence.host_fact("sentinel-intact", format!("intact:{rel}"));
        }
    }
    if let Some(ref outcome) = tree_sweep_outcome {
        if outcome.clean {
            evidence.host_fact(
                "tree-sweep",
                format!("clean=true,killed={}", outcome.killed),
            );
            evidence.host_fact("tree-intact", "no-residual-processes".to_string());
        } else {
            if !outcome.residual.is_empty() {
                evidence.host_fact("tree-escape", format!("residual={:?}", outcome.residual));
                violation_observed = true;
            }
            if outcome.blind {
                evidence.host_fact(
                    "tree-blind",
                    "subreaper-unobserved-or-unreadable-environ".to_string(),
                );
            }
        }
    }

    if let Some(contract) = req.contract {
        let host_env_after: BTreeMap<String, String> = req
            .host_env_override
            .clone()
            .unwrap_or_else(|| std::env::vars().collect());
        let observed_pairs: Vec<(String, String)> = if let Some(ref p) = proc_environ {
            evidence.host_fact("proc-environ", format!("pid={pid}:verified-from-procfs"));
            p.clone()
        } else {
            staged_env.into_iter().collect()
        };
        let mut downlink_val = None;
        let mut uplink_val = None;
        if let Some(ref ch) = control_channel {
            for (k, v) in ch.env_entries() {
                if k == super::host_evidence::ENV_CONTROL_DOWNLINK {
                    downlink_val = Some(v);
                } else if k == super::host_evidence::ENV_CONTROL_UPLINK {
                    uplink_val = Some(v);
                }
            }
        }
        let harness_ctx = super::environment::HarnessEnvContext {
            isolated_home: home.clone(),
            fixture_root: fixture.root().to_path_buf(),
            session_nonce: nonce.clone(),
            has_control_channel: req.enable_host_control,
            control_downlink_val: downlink_val,
            control_uplink_val: uplink_val,
        };
        let env_report = super::environment::verify_execution_environment(
            contract,
            &observed_pairs,
            &host_env_before,
            &host_env_after,
            Some(&harness_ctx),
        );
        for (name, detail) in env_report.host_facts {
            evidence.host_fact(&name, detail);
        }
        if !env_report.clean {
            violation_observed = true;
        }

        let net_report = super::network::verify_network_contract_execution(
            contract,
            &req.scenario.id,
            req.scenario.category,
            &collected.stdout,
            &collected.stderr,
        );
        for (name, detail) in net_report.host_facts {
            evidence.host_fact(&name, detail);
        }
        for (name, detail) in net_report.self_reports {
            evidence.self_report(&name, detail);
        }
        if !net_report.clean {
            violation_observed = true;
        }

        // Host-verified network boundary facts (HOST_FACT only)
        if verification.netns_isolated {
            evidence.host_fact(
                "vector:netns-isolated",
                "procfs-netns-distinct-from-host".to_string(),
            );
        }
        if verification.seccomp_filter && backend.pre_exec_plan().is_some_and(|p| p.net_deny) {
            evidence.host_fact("vector:net-seccomp", "status-seccomp-net-deny".to_string());
        }
        if contract.network.mode == crate::policy_ir::contract::NetworkMode::Off {
            evidence.host_fact(
                "contract-sealed-net-off",
                "blake3-contract-net-off".to_string(),
            );
        }
    }

    // Stage 2 challenge-response verification. All channel I/O stays here
    // on the host side; the oracle only ever sees the stamped fact plus the
    // identity, never the channel. Provenance: `verify` reads the host-held
    // uplink end and mints the capability only on exact arrival of the
    // rotated challenge response — never issued via env, so echoing
    // verifier material cannot verify. Child bytes anywhere else
    // (HOME/fixture files, stdio, env echo, exit code, challenge echo)
    // are not consulted and cannot stamp a fact.
    // PASS-capability gate: the protocol proves live challenge-response
    // execution, never containment, so bound nonces + the quorum vector are
    // assembled from a verified response for `Aux` scenarios as well as
    // boundary-verification scenarios when host-observed facts (HOST_FACT)
    // are present. Without a verified response or host facts both nonce
    // slots stay None and agreeing_vectors stays 0, and the oracle
    // structurally yields INCONCLUSIVE (or FAIL on host-observed violation).
    // Blocker 2: completeness is structured oracle input, not a detail
    // string.
    let has_boundary_evidence = evidence.has_host_fact() || !sentinel_pre.is_empty();
    let pass_capable = req.enable_host_control
        && (req.scenario.category == Category::Aux || has_boundary_evidence);
    let mut control_observed = false;
    let mut control_state = if req.enable_host_control {
        "challenge:unverified"
    } else {
        "disabled"
    };
    if let Some(channel) = control_channel {
        if let Some(verified) = channel.verify(&identity, Instant::now() + CONTROL_READ_BUDGET) {
            evidence.host_control_fact(&verified);
            control_observed = true;
            control_state = if pass_capable {
                "challenge:verified"
            } else {
                // Liveness observed, but direct-exec proves no containment:
                // blocker verdicts stay INCONCLUSIVE/FAIL below regardless.
                "challenge:verified(non-aux,no-pass)"
            };
        }
    }
    let stdio_complete = collected.eof && !collected.truncated;
    let bound_nonce: Option<String> = if control_observed && pass_capable {
        Some(nonce.clone())
    } else {
        None
    };
    let agreeing_vectors: usize = if bound_nonce.is_some() {
        let host_vector_count = evidence
            .facts
            .iter()
            .filter(|f| {
                f.tier == EvidenceTier::HostFact
                    && (f.name == "sentinel-intact"
                        || f.name == "tree-intact"
                        || f.name == "vector"
                        || f.name.starts_with("vector:")
                        || f.name.starts_with("vector-"))
            })
            .count();
        if host_vector_count > 0 {
            host_vector_count
        } else {
            1
        }
    } else {
        0
    };
    let input = oracle::OracleInput {
        scenario: req.scenario,
        evidence: &evidence,
        nonce: Some(nonce.as_str()),
        probe_nonce: bound_nonce.as_deref(),
        control_nonce: bound_nonce.as_deref(),
        payload_intact,
        env_poisoned: false,
        agreeing_vectors,
        violation_observed,
        control_observed,
        stdio_complete,
        execution_identity: Some(&identity),
    };
    let strength = req.scenario.strength_for(target);
    let judged = oracle::judge_with_ceiling(&input, strength, None);
    // Fail-closed ceiling over the FINAL report (post-sweep): PASS without
    // actually enforced mandatory backend capabilities demotes to
    // INCONCLUSIVE. Never upgrades; FAIL/INCONCLUSIVE/NOT_APPLICABLE pass
    // through unchanged.
    let backend_kind = backend.kind();
    let backend_report = backend.enforcement().cloned().unwrap_or_else(|| {
        super::sandbox_backend::EnforcementReport {
            backend: backend_kind,
            scenario_id: identity.scenario_id.clone(),
            session_nonce: nonce.clone(),
            registry_hash: identity.registry_hash.clone(),
            frozen_hash: identity.frozen_hash.clone(),
            policy_hash: String::new(),
            preparation_ok: false,
            records: Vec::new(),
        }
    });
    let verdict =
        super::sandbox_backend::apply_backend_ceiling(judged, &backend_report, req.scenario);
    let mut backend_summary = backend_report.render_deterministic();
    if let Some(diag) = backend.diagnostic() {
        backend_summary.push('|');
        backend_summary.push_str(&diag);
    }

    let detail = redact::redact_text(&redact::mask_home(
        &format!(
            "backend={} run exit={} timeout={} stdout={}B stderr={}B eof={} trunc={} complete={} control={} sentinel_mut={} payload_intact={} spec_ok={} backend=[{}] — {}",
            backend_kind.label(),
            exit_code.map_or("-".to_string(), |c| c.to_string()),
            timed_out,
            collected.stdout.len(),
            collected.stderr.len(),
            collected.eof,
            collected.truncated,
            stdio_complete,
            control_state,
            sentinel_mutated.len(),
            payload_intact,
            spec_ok,
            backend_summary,
            req.scenario.known_limitation,
        ),
        &home.display().to_string(),
    ));

    ExecutionOutcome {
        result: ScenarioResult {
            id: req.scenario.id.clone(),
            category: req.scenario.category,
            strength,
            verdict,
            detail,
        },
        nonce,
        execution_identity: identity,
        backend_report: Some(backend_report),
        backend_kind,
        exit_code,
        timed_out,
        kill: Some(kill),
        stdout: collected.stdout,
        stderr: collected.stderr,
        stdio_eof: collected.eof,
        stdio_truncated: collected.truncated,
        stdio_complete,
        evidence,
        payload_intact,
        sentinel_mutated,
        control_observed,
        violation_observed,
        home_marker,
        home,
        spawn_pid: Some(pid),
        duplicate_rejected: false,
    }
}

/// Suite-level execution owner (Blocker 5): the smallest practical ledger
/// proving `one scenario -> at most one execution` per suite invocation.
///
/// - Owns the spawn ledger: every accepted run appends its [`SpawnEvent`]
///   (run nonce + pid) here; nothing else in the suite path spawns.
/// - The FIRST call for a scenario id executes via [`run_one`]; any further
///   call for the same id is rejected BEFORE any fixture/spawn work with an
///   INCONCLUSIVE outcome marked [`ExecutionOutcome::duplicate_rejected`].
///   Rejection is recorded in `results()` (fail-closed for blockers).
/// - There is no retry API: a rejected duplicate can never upgrade an
///   earlier FAIL/INCONCLUSIVE into a PASS, because it never executes.
pub struct SuiteRunner {
    executed: std::collections::HashSet<String>,
    ledger: SpawnLog,
    results: Vec<ScenarioResult>,
}

impl SuiteRunner {
    pub fn new() -> Self {
        SuiteRunner {
            executed: std::collections::HashSet::new(),
            ledger: Vec::new(),
            results: Vec::new(),
        }
    }

    /// Execute one scenario unless it already ran in this suite.
    pub fn run(&mut self, req: &ExecutionRequest<'_>) -> ExecutionOutcome {
        let mut backend = super::sandbox_backend::DirectBackend::new();
        self.run_with_backend(req, &mut backend)
    }

    /// Suite execution through an explicit backend. The first call for a
    /// scenario id executes via [`run_one_with_backend`]; duplicates are
    /// rejected pre-spawn exactly like [`SuiteRunner::run`].
    pub fn run_with_backend(
        &mut self,
        req: &ExecutionRequest<'_>,
        backend: &mut dyn SandboxBackend,
    ) -> ExecutionOutcome {
        if !self.executed.insert(req.scenario.id.clone()) {
            let target = engine::current_target(None);
            let strength = req.scenario.strength_for(target);
            let detail = redact::redact_text(&format!(
                "duplicate execution rejected, no spawn; earlier verdict stands — {}",
                req.scenario.known_limitation,
            ));
            let outcome = ExecutionOutcome {
                result: ScenarioResult {
                    id: req.scenario.id.clone(),
                    category: req.scenario.category,
                    strength,
                    verdict: super::model::Verdict::Inconclusive,
                    detail,
                },
                nonce: String::new(),
                // Rejected pre-spawn: malformed identity, never PASS-capable.
                execution_identity: ExecutionIdentity::new(&req.scenario.id, "", "", ""),
                backend_report: None,
                backend_kind: backend.kind(),
                exit_code: None,
                timed_out: false,
                kill: None,
                stdout: Vec::new(),
                stderr: Vec::new(),
                stdio_eof: false,
                stdio_truncated: false,
                stdio_complete: false,
                evidence: Evidence::default(),
                payload_intact: true,
                sentinel_mutated: Vec::new(),
                control_observed: false,
                violation_observed: false,
                home_marker: None,
                home: PathBuf::new(),
                spawn_pid: None,
                duplicate_rejected: true,
            };
            self.results.push(outcome.result.clone());
            return outcome;
        }
        let outcome = run_one_with_backend(req, &mut self.ledger, backend);
        self.results.push(outcome.result.clone());
        outcome
    }

    /// Suite-owned spawn ledger: one entry per accepted execution.
    pub fn ledger(&self) -> &[SpawnEvent] {
        &self.ledger
    }

    /// Per-scenario results in execution order, rejections included.
    pub fn results(&self) -> &[ScenarioResult] {
        &self.results
    }
}

impl Default for SuiteRunner {
    fn default() -> Self {
        SuiteRunner::new()
    }
}