cellos-supervisor 0.5.1

//! Environment-driven runtime composition for `cellos-supervisor`.
//!
//! Keeps `main.rs` thin by centralizing adapter selection and env-based wiring.

use std::collections::HashMap;
use std::path::Path;
use std::sync::Arc;

use serde_json::json;

type ExportSinkBundle = (Arc<dyn ExportSink>, HashMap<String, Arc<dyn ExportSink>>);

/// U1-04: a single structured "this env var was ignored / fell back to the
/// default" record. Accumulated across composition by [`StartupConfigWarnings`]
/// so strict mode (`CELLOS_STRICT_CONFIG=1`) can fail loud at the end of
/// startup with a list of every offence rather than only the first one.
#[derive(Debug)]
struct ConfigWarning {
    /// Env var name that was misconfigured (e.g. `CELLOS_CELL_BACKEND`).
    var: &'static str,
    /// The value the operator supplied (best-effort; empty when the var was
    /// unset but a companion var demanded it).
    value: String,
    /// Operator-facing explanation: "value didn't match enum", "missing
    /// companion var", "parse failure", etc.
    reason: String,
}

/// U1-04: accumulator for [`ConfigWarning`]s discovered during composition.
///
/// Each silent-fallback site that previously degraded to a default sink /
/// adapter / value calls [`StartupConfigWarnings::record`], which both emits a
/// structured `tracing::warn!` and stashes the warning. After composition
/// finishes, [`StartupConfigWarnings::finalize`] is called once: if
/// `CELLOS_STRICT_CONFIG=1` (or `true`/`yes`/`on`) is set AND the collector is
/// non-empty, startup fails with a single error that lists every recorded
/// warning. Otherwise the warnings are already in the log and startup
/// proceeds — preserving "WARN-then-continue" as the default so existing
/// operators don't break.
#[derive(Debug, Default)]
struct StartupConfigWarnings {
    items: Vec<ConfigWarning>,
}

impl StartupConfigWarnings {
    fn new() -> Self {
        Self::default()
    }

    /// Record one fallback decision. Emits a structured `tracing::warn!`
    /// against `target = "cellos.supervisor.config"` and keeps the record so
    /// `finalize()` can escalate under strict mode.
    fn record(&mut self, var: &'static str, value: impl Into<String>, reason: impl Into<String>) {
        let value = value.into();
        let reason = reason.into();
        tracing::warn!(
            target: "cellos.supervisor.config",
            var = %var,
            value = %value,
            reason = %reason,
            "env-var fallback: ignored value, using default (set CELLOS_STRICT_CONFIG=1 to fail loud)"
        );
        self.items.push(ConfigWarning { var, value, reason });
    }

    /// If strict mode is enabled and any warnings were recorded, emit a single
    /// summary `tracing::error!` and return `Err`. Otherwise return `Ok(())`.
    fn finalize(self) -> anyhow::Result<()> {
        if self.items.is_empty() {
            return Ok(());
        }
        let strict = strict_config_enabled();
        if !strict {
            return Ok(());
        }
        let summary: Vec<String> = self
            .items
            .iter()
            .map(|w| format!("{}={:?} ({})", w.var, w.value, w.reason))
            .collect();
        tracing::error!(
            target: "cellos.supervisor.config",
            strict = true,
            count = self.items.len(),
            ignored = ?summary,
            "CELLOS_STRICT_CONFIG=1: refusing to start — one or more env vars were ignored or misconfigured during composition"
        );
        Err(anyhow::anyhow!(
            "CELLOS_STRICT_CONFIG is set: refusing to start with {count} ignored/misconfigured env var(s): [{joined}]",
            count = self.items.len(),
            joined = summary.join("; ")
        ))
    }
}

/// U1-04: returns `true` when `CELLOS_STRICT_CONFIG` is set to a truthy value
/// (`1` / `true` / `yes` / `on`). Mirrors the on/off-style env helpers used
/// elsewhere (`trust_verify_keys_required`, etc.).
fn strict_config_enabled() -> bool {
    std::env::var("CELLOS_STRICT_CONFIG")
        .map(|v| {
            let t = v.trim().to_ascii_lowercase();
            matches!(t.as_str(), "1" | "true" | "yes" | "on")
        })
        .unwrap_or(false)
}

use cellos_broker_env::EnvSecretBroker;
use cellos_broker_file::FileSecretBroker;
use cellos_broker_oidc::GithubActionsOidcBroker;
use cellos_broker_vault::VaultAppRoleBroker;
use cellos_core::ports::{
    CellBackend, EventSink, ExportSink, NoopEventSink, NoopExportSink, SecretBroker,
};
use cellos_core::{
    redact_url_credentials_for_logs, validate_authorization_policy, validate_policy_pack_document,
    AuthorizationPolicy, AuthorizationPolicyDocument, ExecutionCellDocument, ExportTarget,
    PolicyPackDocument,
};
use cellos_export_http::HttpExportSink;
use cellos_export_local::LocalExportSink;
use cellos_export_s3::PresignedS3ExportSink;
use cellos_host_cellos::{MemorySecretBroker, ProprietaryCellBackend};
use cellos_host_firecracker::FirecrackerCellBackend;
use cellos_host_stub::StubCellBackend;
use cellos_sink_dlq::DlqSink;
use cellos_sink_jetstream::JetStreamEventSink;
use cellos_sink_jsonl::JsonlEventSink;
use cellos_sink_redact::RedactingEventSink;

use crate::event_signing::SigningEventSink;
use crate::spec_input::resolve_event_subject;
use crate::supervisor::Supervisor;

/// A2-02 / ADR-0007: read `CELLOS_CALLER_IDENTITY` once at the supervisor
/// boundary and return it (defaults to `"default"` when unset or empty).
///
/// **D11 boundary**: this is the *only* surface that reads the env var.
/// `cellos-core` is forbidden from touching `std::env`, so the resolved value
/// is threaded into
/// [`cellos_core::validate_secret_refs_against_allowlist`] as a parameter
/// (and into any future RBAC gate consumer the same way). Whitespace is
/// trimmed; an empty / whitespace-only value is treated as unset and falls
/// back to `"default"` so a misconfigured systemd unit still admits under the
/// permissive (no-allowlist) policy pack — strict mode lives in a separate
/// env (`CELLOS_REQUIRE_CALLER_IDENTITY`) per ADR-0007.
pub(crate) fn resolve_caller_identity() -> String {
    std::env::var("CELLOS_CALLER_IDENTITY")
        .ok()
        .map(|v| v.trim().to_string())
        .filter(|v| !v.is_empty())
        .unwrap_or_else(|| "default".to_string())
}

#[allow(dead_code)]
fn selected_host_backend_kind() -> &'static str {
    match std::env::var("CELLOS_CELL_BACKEND")
        .unwrap_or_default()
        .as_str()
    {
        "firecracker" => "firecracker",
        "gvisor" => "gvisor",
        "stub" => "stub",
        _ => "proprietary",
    }
}

#[allow(dead_code)]
fn selected_secret_broker_kind() -> &'static str {
    match std::env::var("CELLOS_BROKER").unwrap_or_default().as_str() {
        "env" => "env",
        "file" => "file",
        "github-oidc" => "github-oidc",
        "vault-approle" => "vault-approle",
        _ => "memory",
    }
}

#[allow(dead_code)]
fn selected_default_export_sink_kind() -> &'static str {
    if scoped_env("CELLOS_EXPORT_HTTP_BASE_URL", None).is_some() {
        "http"
    } else if std::env::var("CELLOS_EXPORT_DIR").is_ok() {
        "local"
    } else {
        "noop"
    }
}

/// Build a no-side-effect summary for `cellos-supervisor --validate` output.
///
/// This function reports the **resolved configuration intent** (selected
/// adapters and key env-driven toggles). `main.rs` still calls
/// [`build_supervisor`] before printing this summary, so adapter init and
/// connectivity checks happen on the real code path.
#[allow(dead_code)]
pub(crate) fn build_validation_summary(
    doc: &ExecutionCellDocument,
    run_id: &str,
    raw_spec_hash: &str,
    spec_path: &Path,
) -> serde_json::Value {
    let named_targets = doc
        .spec
        .export
        .as_ref()
        .and_then(|e| e.targets.as_ref())
        .map(|targets| {
            targets
                .iter()
                .map(|target| match target {
                    ExportTarget::Http(http) => json!({
                        "name": http.name,
                        "kind": "http",
                    }),
                    ExportTarget::S3(s3) => json!({
                        "name": s3.name,
                        "kind": "s3",
                    }),
                })
                .collect::<Vec<_>>()
        })
        .unwrap_or_default();

    json!({
        "mode": "validate",
        "specPath": spec_path.display().to_string(),
        "specId": doc.spec.id,
        "runId": run_id,
        "specSha256": raw_spec_hash,
        "deploymentProfile": std::env::var("CELLOS_DEPLOYMENT_PROFILE").unwrap_or_else(|_| "hardened".to_string()),
        "hostBackend": selected_host_backend_kind(),
        "secretBroker": selected_secret_broker_kind(),
        "eventSink": {
            "primary": if std::env::var("CELL_OS_USE_NOOP_SINK").is_ok() { "noop" } else { "jetstream-or-noop-fallback" },
            "subject": resolve_event_subject(
                &doc.spec.id,
                run_id,
                doc.spec.correlation.as_ref().and_then(|c| c.tenant_id.as_deref()),
            ),
            "jsonlMirrorEnabled": std::env::var("CELL_OS_JSONL_EVENTS").is_ok(),
        },
        "export": {
            "defaultSink": selected_default_export_sink_kind(),
            "namedTargets": named_targets,
        },
        "policyPackConfigured": std::env::var("CELLOS_POLICY_PACK_PATH").ok().filter(|v| !v.trim().is_empty()).is_some(),
        "authorityKeysConfigured": std::env::var("CELLOS_AUTHORITY_KEYS_PATH").ok().filter(|v| !v.trim().is_empty()).is_some(),
    })
}

/// Enforce the named deployment profile selected via
/// `CELLOS_DEPLOYMENT_PROFILE`.
///
/// Currently the only recognised profile is `hardened`, which collapses
/// four independent opt-ins into a single deployment decision:
///
/// 1. `CELLOS_AUTHORITY_KEYS_PATH` — must be set so that authority
///    derivation tokens are actually verified (see [`load_authority_keys`]).
/// 2. `CELL_OS_REQUIRE_JETSTREAM` — auto-set to `1` so a JetStream connect
///    failure is fatal instead of silently falling back to the noop sink
///    (see [`build_primary_event_sink`]).
/// 3. `CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS` — auto-set to `1` so
///    universal (parentRunId: null) derivation tokens are rejected.
/// 4. `CELLOS_REQUIRE_AUTHORITY_DERIVATION` — auto-set to `1` so any spec
///    that declares a non-empty authority bundle (egress rules or secret
///    refs) without an `authorityDerivation` token is rejected. Closes
///    SEAM-3 from adversarial review Session 2: in earlier hardened
///    deployments the keys file would load and then do nothing for specs
///    that simply omitted the token. See
///    [`enforce_authority_derivation_requirement`].
///
/// Auto-setting (2), (3) and (4) is safe because this runs during
/// single-threaded startup before any worker thread observes the
/// environment. Returns an error only if a precondition cannot be
/// satisfied automatically — today that means `CELLOS_AUTHORITY_KEYS_PATH`
/// is missing.
pub(crate) fn enforce_deployment_profile() -> anyhow::Result<()> {
    let profile = std::env::var("CELLOS_DEPLOYMENT_PROFILE").unwrap_or_default();

    if profile.eq_ignore_ascii_case("hardened") {
        // 1. Authority keys must be configured — operator-provided file,
        //    cannot be auto-generated.
        if std::env::var_os("CELLOS_AUTHORITY_KEYS_PATH").is_none() {
            return Err(anyhow::anyhow!(
                "CELLOS_DEPLOYMENT_PROFILE=hardened requires CELLOS_AUTHORITY_KEYS_PATH to be set"
            ));
        }

        // 2. JetStream must be required (no silent noop fallback).
        if std::env::var("CELL_OS_REQUIRE_JETSTREAM").is_err() {
            // SAFETY: single-threaded startup, no race with worker threads.
            std::env::set_var("CELL_OS_REQUIRE_JETSTREAM", "1");
            tracing::info!(
                target: "cellos.supervisor.profile",
                "hardened profile: CELL_OS_REQUIRE_JETSTREAM auto-enabled"
            );
        }

        // 3. Universal derivation tokens must be rejected. Hardened mode
        //    OVERRIDES any explicit permissive opt-out: setting the var to
        //    `0`/`false`/`no`/`off` while also asking for hardened mode is a
        //    contradiction — fix the contradiction in favour of hardened.
        let scoped_value = std::env::var("CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS")
            .ok()
            .map(|v| v.trim().to_ascii_lowercase());
        let is_explicit_permissive = matches!(
            scoped_value.as_deref(),
            Some("0") | Some("false") | Some("no") | Some("off")
        );
        if scoped_value.is_none() || is_explicit_permissive {
            // SAFETY: single-threaded startup, no race with worker threads.
            std::env::set_var("CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS", "1");
            if is_explicit_permissive {
                tracing::warn!(
                    target: "cellos.supervisor.profile",
                    "hardened profile: CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS permissive opt-out OVERRIDDEN — strict mode required under hardened"
                );
            } else {
                tracing::info!(
                    target: "cellos.supervisor.profile",
                    "hardened profile: CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS auto-enabled"
                );
            }
        }

        // 4. Specs declaring authority must carry a derivation token (SEAM-3).
        if std::env::var("CELLOS_REQUIRE_AUTHORITY_DERIVATION").is_err() {
            // SAFETY: single-threaded startup, no race with worker threads.
            std::env::set_var("CELLOS_REQUIRE_AUTHORITY_DERIVATION", "1");
            tracing::info!(
                target: "cellos.supervisor.profile",
                "hardened profile: CELLOS_REQUIRE_AUTHORITY_DERIVATION auto-enabled"
            );
        }

        // 5. F4a — specs MUST declare a `spec.telemetry` block. Hardened
        //    deployments emit structured telemetry over vsock-cbor; an
        //    operator who picks `hardened` and accepts a spec without a
        //    telemetry block is blind to in-cell observability. Auto-set so
        //    the gate fires uniformly across all admitted specs.
        if std::env::var("CELLOS_REQUIRE_TELEMETRY_DECLARED").is_err() {
            // SAFETY: single-threaded startup, no race with worker threads.
            std::env::set_var("CELLOS_REQUIRE_TELEMETRY_DECLARED", "1");
            tracing::info!(
                target: "cellos.supervisor.profile",
                "hardened profile: CELLOS_REQUIRE_TELEMETRY_DECLARED auto-enabled"
            );
        }

        // 6. Wave 2 red-team (MED-W2D-5): hardened mode previously left
        //    `CELLOS_STRICT_CONFIG` untouched, so an operator who typo'd e.g.
        //    `CELLOS_BROKER=valt-approle` would silently fall through to the
        //    in-memory toy broker even with all the other hardened guarantees
        //    active. The asymmetry — strict on authority/jetstream/telemetry
        //    but lenient on env-var spelling — is a real foot-gun under
        //    hardened. Auto-enable strict mode so the StartupConfigWarnings
        //    collector's `finalize()` escalates any env-var typo to a startup
        //    error. Explicit operator value (including `0`) is preserved.
        if std::env::var("CELLOS_STRICT_CONFIG").is_err() {
            // SAFETY: single-threaded startup, no race with worker threads.
            std::env::set_var("CELLOS_STRICT_CONFIG", "1");
            tracing::info!(
                target: "cellos.supervisor.profile",
                "hardened profile: CELLOS_STRICT_CONFIG auto-enabled (env-var typos fail-loud)"
            );
        }

        tracing::info!(
            target: "cellos.supervisor.profile",
            profile = "hardened",
            "deployment profile enforced: authority keys required, JetStream fail-closed, scoped derivation tokens required, authority derivation required for specs declaring egress or secrets, telemetry block required, strict config fail-closed"
        );
    }

    Ok(())
}

/// I4: hardened-vs-portable deployment mode resolved at startup.
///
/// `CELLOS_REQUIRE_ISOLATION=1` is the **1.0 default** (per
/// Plans/cellos-code-complete-roadmap.md §7 Phase I and ADR-0006).
/// Operators that explicitly need to run on a host without
/// unshare/seccomp/cgroup v2 must set `CELLOS_DEPLOYMENT_PROFILE=portable`;
/// the supervisor then degrades the isolation requirement to a soft warning
/// rather than a fail-closed startup gate.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum DeploymentMode {
    Hardened,
    Portable,
}

impl DeploymentMode {
    fn as_str(self) -> &'static str {
        match self {
            DeploymentMode::Hardened => "Hardened",
            DeploymentMode::Portable => "Portable",
        }
    }
}

/// Resolve the effective [`DeploymentMode`] from `CELLOS_DEPLOYMENT_PROFILE`.
///
/// `hardened` (or unset) → `Hardened`. `portable` → `Portable`. Any other
/// value is treated as a typo and rejected at startup.
pub(crate) fn resolve_deployment_mode() -> anyhow::Result<DeploymentMode> {
    let raw = std::env::var("CELLOS_DEPLOYMENT_PROFILE").unwrap_or_default();
    let normalised = raw.trim().to_ascii_lowercase();
    match normalised.as_str() {
        "" | "hardened" => Ok(DeploymentMode::Hardened),
        "portable" => Ok(DeploymentMode::Portable),
        other => Err(anyhow::anyhow!(
            "CELLOS_DEPLOYMENT_PROFILE: unknown profile {other:?} — expected 'hardened' or 'portable'"
        )),
    }
}

/// Probe the host for the isolation primitives the supervisor relies on
/// (unshare/seccomp/cgroup v2). Returns the names of any primitives that
/// are MISSING — empty vec means the host is fully isolation-capable.
///
/// Test-injection hook: when
/// `CELLOS_TEST_FORCE_MISSING_ISOLATION_PRIMITIVES` is set, its
/// comma-separated value forces the named primitives to appear missing
/// regardless of the real host capabilities. Recognised tokens: `unshare`,
/// `seccomp`, `cgroup-v2`, plus the wildcards `all` / `*`. Unknown tokens
/// are ignored (with a single trace) so a typo cannot silently disable the
/// gate. This lets CI exercise the "host without unshare" code path on
/// fully-capable runners without elevated privileges.
pub(crate) fn probe_missing_isolation_primitives() -> Vec<&'static str> {
    if let Some(forced) = forced_missing_primitives_from_env() {
        return forced;
    }

    let mut missing: Vec<&'static str> = Vec::new();
    #[cfg(target_os = "linux")]
    {
        if !std::path::Path::new("/proc/self/ns/user").exists() {
            missing.push("unshare");
        }
        if !std::path::Path::new("/sys/fs/cgroup/cgroup.controllers").exists() {
            missing.push("cgroup-v2");
        }
        let seccomp_supported = std::fs::read_to_string("/proc/self/status")
            .map(|s| s.lines().any(|l| l.starts_with("Seccomp:")))
            .unwrap_or(false);
        if !seccomp_supported {
            missing.push("seccomp");
        }
    }
    #[cfg(not(target_os = "linux"))]
    {
        missing.push("unshare");
        missing.push("seccomp");
        missing.push("cgroup-v2");
    }
    missing
}

/// Parse `CELLOS_TEST_FORCE_MISSING_ISOLATION_PRIMITIVES` into the canonical
/// `&'static str` slice form returned by [`probe_missing_isolation_primitives`].
///
/// Returns `None` when the env var is unset or empty, so the real host probe
/// runs unmodified. Returns `Some(vec)` (possibly empty if every token is
/// unknown) when the var is set, which short-circuits the host probe.
fn forced_missing_primitives_from_env() -> Option<Vec<&'static str>> {
    let raw = std::env::var("CELLOS_TEST_FORCE_MISSING_ISOLATION_PRIMITIVES").ok()?;
    let trimmed = raw.trim();
    if trimmed.is_empty() {
        return None;
    }
    let mut out: Vec<&'static str> = Vec::new();
    for token in trimmed.split(',') {
        let normalised = token.trim().to_ascii_lowercase();
        match normalised.as_str() {
            "" => continue,
            "all" | "*" => {
                out.clear();
                out.push("unshare");
                out.push("seccomp");
                out.push("cgroup-v2");
                return Some(out);
            }
            "unshare" => {
                if !out.contains(&"unshare") {
                    out.push("unshare");
                }
            }
            "seccomp" => {
                if !out.contains(&"seccomp") {
                    out.push("seccomp");
                }
            }
            "cgroup-v2" | "cgroup_v2" | "cgroupv2" => {
                if !out.contains(&"cgroup-v2") {
                    out.push("cgroup-v2");
                }
            }
            other => {
                tracing::trace!(
                    target: "cellos.supervisor.profile",
                    token = %other,
                    "CELLOS_TEST_FORCE_MISSING_ISOLATION_PRIMITIVES: ignoring unknown token"
                );
            }
        }
    }
    Some(out)
}

/// I4: in `Hardened` mode the supervisor refuses to start without
/// unshare/seccomp/cgroup v2 support unless the operator explicitly named
/// `CELLOS_DEPLOYMENT_PROFILE=portable`. Auto-sets
/// `CELLOS_REQUIRE_ISOLATION=1` whenever it isn't already in the
/// environment. In `Portable` mode the missing primitives are surfaced as a
/// single warning event but the supervisor proceeds.
pub(crate) fn enforce_isolation_default(mode: DeploymentMode) -> anyhow::Result<()> {
    let missing = probe_missing_isolation_primitives();
    match mode {
        DeploymentMode::Hardened => {
            // SAFETY: single-threaded startup, no race with worker threads.
            if std::env::var("CELLOS_REQUIRE_ISOLATION").is_err() {
                std::env::set_var("CELLOS_REQUIRE_ISOLATION", "1");
            }
            if !missing.is_empty() {
                tracing::error!(
                    target: "cellos.supervisor.profile",
                    event = "host_capability_check_failed",
                    profile = "hardened",
                    missing = ?missing,
                    "hardened profile: host is missing isolation primitives — refusing to start"
                );
                return Err(anyhow::anyhow!(
                    "CELLOS_DEPLOYMENT_PROFILE=hardened (1.0 default): host is missing isolation \
                     primitives: {missing:?}. Set CELLOS_DEPLOYMENT_PROFILE=portable to run \
                     without isolation enforcement (audit-trail-only mode)."
                ));
            }
        }
        DeploymentMode::Portable => {
            if !missing.is_empty() {
                tracing::warn!(
                    target: "cellos.supervisor.profile",
                    profile = "portable",
                    missing = ?missing,
                    "portable profile: isolation primitives missing — degraded posture, \
                     CELLOS_REQUIRE_ISOLATION not enforced"
                );
            } else {
                tracing::info!(
                    target: "cellos.supervisor.profile",
                    profile = "portable",
                    "portable profile: isolation primitives available but enforcement disabled \
                     by operator opt-out"
                );
            }
        }
    }
    Ok(())
}

/// I4: emit a single, loud startup banner naming the resolved deployment
/// profile so operators can see at a glance whether isolation is enforced.
pub(crate) fn emit_startup_banner(mode: DeploymentMode) {
    let (label, summary) = match mode {
        DeploymentMode::Hardened => (
            "Hardened",
            "isolation enforced (unshare/seccomp/cgroup v2 required) — 1.0 default",
        ),
        DeploymentMode::Portable => (
            "Portable",
            "isolation NOT enforced — explicit operator opt-out via CELLOS_DEPLOYMENT_PROFILE=portable",
        ),
    };
    tracing::info!(
        target: "cellos.supervisor.profile",
        event = "startup_banner",
        profile = mode.as_str(),
        "[cellos-supervisor] startup-profile: {label} ({summary})"
    );
}

/// Reject specs that declare a non-empty authority bundle without an
/// `authorityDerivation` token when `CELLOS_REQUIRE_AUTHORITY_DERIVATION` is
/// set (closes SEAM-3 from adversarial review Session 2).
///
/// Without this check, a hardened deployment could load
/// `CELLOS_AUTHORITY_KEYS_PATH` and still admit specs that declare egress
/// rules or `secretRefs` without any cryptographic proof of delegation —
/// the keys file would load and do nothing. This function closes that gap
/// by making the token mandatory whenever there is something to delegate.
///
/// "Non-empty authority bundle" means at least one of:
/// - `spec.authority.egressRules` present and non-empty
/// - `spec.authority.secretRefs` present and non-empty
///
/// The opaque `filesystem` / `network` blobs are intentionally not part of
/// this check — they are not yet promoted into the typed
/// `AuthorityCapability` lattice that derivation tokens scope over.
///
/// Specs with an empty authority bundle pass unchanged regardless of the
/// flag — there is nothing to delegate, so the token is irrelevant.
///
/// This runs after `validate_execution_cell_document` (which has no env
/// access) and before `verify_authority_derivation` so that operators see a
/// clear "missing token" error before the supervisor would otherwise hand
/// an authority-bearing spec straight to `build_supervisor`.
pub(crate) fn enforce_authority_derivation_requirement(
    doc: &ExecutionCellDocument,
) -> anyhow::Result<()> {
    if std::env::var("CELLOS_REQUIRE_AUTHORITY_DERIVATION").is_err() {
        return Ok(());
    }
    let authority = &doc.spec.authority;
    let has_egress = authority
        .egress_rules
        .as_ref()
        .is_some_and(|r| !r.is_empty());
    let has_secrets = authority
        .secret_refs
        .as_ref()
        .is_some_and(|r| !r.is_empty());
    let has_authority = has_egress || has_secrets;
    let has_token = authority.authority_derivation.is_some();
    if has_authority && !has_token {
        return Err(anyhow::anyhow!(
            "CELLOS_REQUIRE_AUTHORITY_DERIVATION is set: spec declares authority \
             (egressRules or secretRefs) without an authorityDerivation token — \
             cryptographic authority proof is required"
        ));
    }
    Ok(())
}

/// F4a — reject specs that omit `spec.telemetry` when
/// `CELLOS_REQUIRE_TELEMETRY_DECLARED` is set.
///
/// Mirrors [`enforce_authority_derivation_requirement`]: pure env-gated
/// admission check that runs after `validate_execution_cell_document` (which
/// has no env access) and before `build_supervisor`. Auto-enabled by
/// `CELLOS_DEPLOYMENT_PROFILE=hardened`.
///
/// When the flag is unset (and the operator did not opt into hardened mode),
/// specs without a telemetry block pass unchanged — the cell simply emits no
/// agent events. When the flag IS set, every admitted spec must declare its
/// telemetry intent so SIEM consumers and the in-guest agent see a consistent
/// contract.
///
/// The block's structural validity is enforced separately by
/// [`cellos_core::validate_execution_cell_document`] (cross-references
/// `telemetry.events` `net.*` entries against `authority.egressRules`).
pub(crate) fn enforce_telemetry_declared(doc: &ExecutionCellDocument) -> anyhow::Result<()> {
    if std::env::var("CELLOS_REQUIRE_TELEMETRY_DECLARED").is_err() {
        return Ok(());
    }
    if doc.spec.telemetry.is_none() {
        return Err(anyhow::anyhow!(
            "CELLOS_REQUIRE_TELEMETRY_DECLARED is set: spec.telemetry block is required — \
             every admitted cell must declare its telemetry intent (channel, events, \
             agent version) under hardened deployments"
        ));
    }
    Ok(())
}

/// T11-5: enforce Kubernetes-namespace placement at the supervisor.
///
/// When `CELLOS_K8S_NAMESPACE` is unset, every spec passes — the supervisor
/// is namespace-agnostic and the gate is effectively a no-op.
///
/// When `CELLOS_K8S_NAMESPACE` is set, every spec that **declares**
/// `spec.placement.kubernetesNamespace` MUST match the supervisor's
/// configured namespace exactly. Specs that omit the field still pass
/// (they're portable — a missing constraint is not a contradiction).
///
/// This mirrors the fleet's pool-id gate at the supervisor seam: the same
/// "explicit constraint must match the runner's identity" rule, just at a
/// different layer of the placement stack.
pub(crate) fn enforce_kubernetes_namespace_placement(
    doc: &ExecutionCellDocument,
) -> anyhow::Result<()> {
    let Ok(runner_ns) = std::env::var("CELLOS_K8S_NAMESPACE") else {
        return Ok(());
    };
    let runner_ns = runner_ns.trim();
    if runner_ns.is_empty() {
        return Ok(());
    }
    let Some(placement) = &doc.spec.placement else {
        return Ok(());
    };
    let Some(spec_ns) = placement.kubernetes_namespace.as_deref() else {
        return Ok(());
    };
    if spec_ns != runner_ns {
        return Err(anyhow::anyhow!(
            "CELLOS_K8S_NAMESPACE={runner_ns}: spec.placement.kubernetesNamespace={spec_ns} \
             does not match — refusing to admit cell {spec_id}",
            spec_id = doc.spec.id,
        ));
    }
    Ok(())
}

/// Load the operator's role → ED25519 verifying key map from
/// `CELLOS_AUTHORITY_KEYS_PATH`.
///
/// File format: a JSON object whose keys are role IDs and whose values are
/// base64-STANDARD-encoded 32-byte ED25519 verifying keys.
///
/// ```json
/// { "role-ci-runner": "<base64-32-bytes>", "role-build-bot": "<base64-32-bytes>" }
/// ```
///
/// If the env var is not set, returns an empty map — specs without a derivation
/// token are still accepted; specs with a token will fail verification with
/// `unknown role` (intentional — operators must opt in to derivation by
/// providing the keys file).
///
/// If the env var IS set but the file cannot be read or parsed, returns an
/// error so that startup fails loudly rather than silently dropping
/// authority enforcement.
pub(crate) fn load_authority_keys() -> anyhow::Result<HashMap<String, String>> {
    let Some(path) = std::env::var_os("CELLOS_AUTHORITY_KEYS_PATH") else {
        return Ok(HashMap::new());
    };
    // SEC-15b: O_NOFOLLOW so a swapped-in symlink at the final path component
    // cannot redirect authority-key loading to an attacker-controlled file.
    // Matches the same protection applied to CELLOS_POLICY_PACK_PATH.
    #[cfg(unix)]
    let raw = {
        use std::io::Read;
        use std::os::unix::fs::OpenOptionsExt;
        let mut opts = std::fs::OpenOptions::new();
        opts.read(true);
        opts.custom_flags(libc::O_RDONLY | libc::O_NOFOLLOW);
        let mut file = opts.open(&path).map_err(|e| {
            anyhow::anyhow!(
                "CELLOS_AUTHORITY_KEYS_PATH: cannot read '{}': {e}",
                path.to_string_lossy()
            )
        })?;
        let mut buf = String::new();
        file.read_to_string(&mut buf).map_err(|e| {
            anyhow::anyhow!(
                "CELLOS_AUTHORITY_KEYS_PATH: cannot read '{}': {e}",
                path.to_string_lossy()
            )
        })?;
        buf
    };
    #[cfg(not(unix))]
    let raw = std::fs::read_to_string(&path).map_err(|e| {
        anyhow::anyhow!(
            "CELLOS_AUTHORITY_KEYS_PATH: cannot read '{}': {e}",
            path.to_string_lossy()
        )
    })?;
    let keys: HashMap<String, String> = serde_json::from_str(&raw).map_err(|e| {
        anyhow::anyhow!(
            "CELLOS_AUTHORITY_KEYS_PATH: JSON parse error in '{}': {e}",
            path.to_string_lossy()
        )
    })?;
    tracing::info!(
        path = %path.to_string_lossy(),
        role_count = keys.len(),
        "authority verifying keys loaded"
    );
    Ok(keys)
}

/// Build the supervisor, wiring in adapters from the environment.
///
/// Order of side-effecting startup steps that matter for this docstring:
///
/// 1. Host backend, secret broker, primary event sink (JetStream/noop),
///    optional JSONL sink, and export sinks are constructed.
/// 2. The `Supervisor` struct is initialized with the operator authority key
///    map.
/// 3. **SEC-25 Phase 2 trust-keyset establishment runs here**, AFTER the
///    primary event sink is built so the keyset-verified / keyset-verification-
///    failed CloudEvent can be emitted synchronously through the same
///    `EventSink` pipeline used by the rest of the lifecycle. The order is
///    deliberate — see `crate::trust_keyset_load`. The verify-and-emit step
///    is once-per-supervisor-startup, NOT per-cell, so the cost is bounded
///    even for high-cell-rate operator deployments.
/// 4. Optional policy pack load.
pub(crate) async fn build_supervisor(
    doc: &ExecutionCellDocument,
    run_id: &str,
    authority_keys: Arc<HashMap<String, String>>,
) -> anyhow::Result<Supervisor> {
    // A2-02 / ADR-0007: resolve the caller identity once at supervisor
    // startup. This is the single seam where `CELLOS_CALLER_IDENTITY` is read
    // (D11: cellos-core is env-free; the resolved value is passed as a
    // parameter into `validate_secret_refs_against_allowlist`). Logged at
    // debug level only — the identity itself can be sensitive (an internal
    // service-account name; a CI runner ID) and is leaked into CloudEvents
    // only on rejection per ADR-0007 §"What 1.0 ships".
    let caller_identity = resolve_caller_identity();
    tracing::debug!(
        target: "cellos.supervisor.rbac",
        caller_identity = %caller_identity,
        "A2-02: resolved caller identity for secretRef admission"
    );

    // U1-04: accumulate config warnings across composition; on the strict
    // path (CELLOS_STRICT_CONFIG=1), `warnings.finalize()?` below escalates
    // them into a fatal startup error.
    let mut warnings = StartupConfigWarnings::new();

    // Build the primary event sink first so the firecracker backend can
    // receive a clone for its best-effort `pool_checkout` CloudEvent
    // emission. The sink build is independent of the host backend (it
    // consults only env + spec correlation), so the reorder is safe.
    let event_sink = build_primary_event_sink(doc, run_id, &mut warnings).await?;
    let host = build_host_backend(&mut warnings, Some(Arc::clone(&event_sink)))?;
    let broker = build_secret_broker(&mut warnings)?;
    let jsonl_sink = build_jsonl_sink();
    let (default_export_sink, named_export_sinks) = build_export_sinks(doc, &mut warnings)?;

    let mut supervisor = Supervisor::new(
        host,
        broker,
        event_sink,
        jsonl_sink,
        default_export_sink,
        named_export_sinks,
        authority_keys,
    );

    // SEC-25 Phase 2: load the operator's trust verifying-keys keyring (if
    // configured) and, when CELLOS_TRUST_KEYSET_PATH points at a signed
    // envelope, verify it once at startup. Emits a single
    // `dev.cellos.events.cell.trust.v1.keyset_verified` or
    // `.keyset_verification_failed` CloudEvent through the just-built primary
    // sink. Under CELLOS_REQUIRE_TRUST_VERIFY_KEYS=1 the keyring file MUST
    // load and the envelope MUST verify, otherwise this returns Err and the
    // supervisor refuses to run. Multi-signer threshold remains Phase 3.
    let require_trust_verify_keys = trust_verify_keys_required();
    let trust_keys =
        crate::trust_keyset_load::load_trust_verify_keys_from_env(require_trust_verify_keys)?;
    supervisor.trust_verify_keys = Arc::clone(&trust_keys);
    let outcome = crate::trust_keyset_load::load_and_verify_trust_keyset_from_env(
        trust_keys.as_ref(),
        require_trust_verify_keys,
        std::time::SystemTime::now(),
    )?;
    crate::trust_keyset_load::emit_keyset_outcome(
        &outcome,
        &supervisor.event_sink,
        supervisor.jsonl_sink.as_ref(),
        chrono::Utc::now(),
    )
    .await?;

    // Load policy pack from CELLOS_POLICY_PACK_PATH when set.
    // SEC-15b: open with O_NOFOLLOW on Unix so a swapped-in symlink at the final
    // path component cannot redirect the supervisor to an attacker-controlled
    // policy pack (matches the SEC-08 protection on the spec path — see
    // `spec_input.rs`). Windows keeps `read_to_string` (no O_NOFOLLOW in std).
    if let Some(path) = std::env::var_os("CELLOS_POLICY_PACK_PATH") {
        #[cfg(unix)]
        let raw = {
            use std::io::Read;
            use std::os::unix::fs::OpenOptionsExt;
            let mut opts = std::fs::OpenOptions::new();
            opts.read(true);
            opts.custom_flags(libc::O_RDONLY | libc::O_NOFOLLOW);
            let mut file = opts.open(&path).map_err(|e| {
                anyhow::anyhow!(
                    "CELLOS_POLICY_PACK_PATH: cannot read '{}': {e}",
                    path.to_string_lossy()
                )
            })?;
            let mut buf = String::new();
            file.read_to_string(&mut buf).map_err(|e| {
                anyhow::anyhow!(
                    "CELLOS_POLICY_PACK_PATH: cannot read '{}': {e}",
                    path.to_string_lossy()
                )
            })?;
            buf
        };
        #[cfg(not(unix))]
        let raw = std::fs::read_to_string(&path).map_err(|e| {
            anyhow::anyhow!(
                "CELLOS_POLICY_PACK_PATH: cannot read '{}': {e}",
                path.to_string_lossy()
            )
        })?;
        let pack_doc: PolicyPackDocument = serde_json::from_str(&raw).map_err(|e| {
            anyhow::anyhow!(
                "CELLOS_POLICY_PACK_PATH: JSON parse error in '{}': {e}",
                path.to_string_lossy()
            )
        })?;
        validate_policy_pack_document(&pack_doc)
            .map_err(|e| {
                // P4-04: structural validation is strict (no env override). The
                // operator-driven downgrade allow lives below, after the
                // structural check passes for everything BUT the version floor.
                anyhow::anyhow!(
                    "CELLOS_POLICY_PACK_PATH: invalid policy pack in '{}': {e}",
                    path.to_string_lossy()
                )
            })
            .or_else(|err| {
                // Re-run the version gate with the env-driven override, but
                // only when the structural error came from the version-floor
                // check (cellos-core's strict-by-default call). Other
                // structural errors are unrecoverable regardless of the env.
                let msg = err.to_string();
                let is_version_floor_err = msg.contains("is older than runtime-supported floor");
                if !is_version_floor_err {
                    return Err(err);
                }
                let allow_downgrade = std::env::var(cellos_core::POLICY_ALLOW_DOWNGRADE_ENV)
                    .map(|v| {
                        let t = v.trim().to_ascii_lowercase();
                        matches!(t.as_str(), "1" | "true" | "yes" | "on")
                    })
                    .unwrap_or(false);
                cellos_core::check_policy_pack_version_compatibility(
                    pack_doc.spec.version.as_deref(),
                    allow_downgrade,
                )
                .map_err(|e| {
                    anyhow::anyhow!(
                        "CELLOS_POLICY_PACK_PATH: invalid policy pack in '{}': {e}",
                        path.to_string_lossy()
                    )
                })
            })?;
        tracing::info!(
            pack_id = %pack_doc.spec.id,
            path = %path.to_string_lossy(),
            "policy pack loaded"
        );
        supervisor.policy_pack = Some(pack_doc.spec);
    }

    // T12 RBAC — load the operator's authorization policy from
    // CELLOS_AUTHZ_POLICY_PATH when set. When unset, the authz gate is
    // skipped entirely (legacy/permissive default). When set, the file MUST
    // parse and validate, otherwise startup fails — same fail-loud posture
    // as CELLOS_POLICY_PACK_PATH.
    supervisor.authz_policy = load_authz_policy()?;

    // U1-04: with composition complete, escalate accumulated warnings to a
    // hard error if (and only if) `CELLOS_STRICT_CONFIG=1`. Default behaviour
    // is "WARN-then-continue" — every offence is already in the log via
    // `StartupConfigWarnings::record`.
    warnings.finalize()?;

    Ok(supervisor)
}

/// T12 RBAC — load the operator's authorization policy from
/// `CELLOS_AUTHZ_POLICY_PATH` (JSON file, schema
/// `contracts/schemas/authorization-policy-v1.schema.json`).
///
/// Returns `Ok(None)` when the env var is unset — the authz gate is skipped
/// in that case (legacy/permissive default). Returns `Ok(Some(policy))` after
/// successful read + parse + structural validation. Returns `Err` when the
/// var is set but the file cannot be read, parsed, or validated.
///
/// Symlink hardening: on Unix the final path component is opened with
/// `O_NOFOLLOW` so a swapped-in symlink cannot redirect the supervisor to
/// an attacker-controlled policy (mirrors the SEC-15b protection on
/// `CELLOS_POLICY_PACK_PATH` / `CELLOS_AUTHORITY_KEYS_PATH`).
pub(crate) fn load_authz_policy() -> anyhow::Result<Option<AuthorizationPolicy>> {
    let Some(path) = std::env::var_os("CELLOS_AUTHZ_POLICY_PATH") else {
        return Ok(None);
    };
    #[cfg(unix)]
    let raw = {
        use std::io::Read;
        use std::os::unix::fs::OpenOptionsExt;
        let mut opts = std::fs::OpenOptions::new();
        opts.read(true);
        opts.custom_flags(libc::O_RDONLY | libc::O_NOFOLLOW);
        let mut file = opts.open(&path).map_err(|e| {
            anyhow::anyhow!(
                "CELLOS_AUTHZ_POLICY_PATH: cannot read '{}': {e}",
                path.to_string_lossy()
            )
        })?;
        let mut buf = String::new();
        file.read_to_string(&mut buf).map_err(|e| {
            anyhow::anyhow!(
                "CELLOS_AUTHZ_POLICY_PATH: cannot read '{}': {e}",
                path.to_string_lossy()
            )
        })?;
        buf
    };
    #[cfg(not(unix))]
    let raw = std::fs::read_to_string(&path).map_err(|e| {
        anyhow::anyhow!(
            "CELLOS_AUTHZ_POLICY_PATH: cannot read '{}': {e}",
            path.to_string_lossy()
        )
    })?;
    let doc: AuthorizationPolicyDocument = serde_json::from_str(&raw).map_err(|e| {
        anyhow::anyhow!(
            "CELLOS_AUTHZ_POLICY_PATH: JSON parse error in '{}': {e}",
            path.to_string_lossy()
        )
    })?;
    validate_authorization_policy(&doc).map_err(|e| {
        anyhow::anyhow!(
            "CELLOS_AUTHZ_POLICY_PATH: invalid authorization policy in '{}': {e}",
            path.to_string_lossy()
        )
    })?;
    tracing::info!(
        target: "cellos.supervisor.rbac",
        path = %path.to_string_lossy(),
        subjects = doc.spec.subjects.len(),
        allowed_pools = doc.spec.allowed_pools.len(),
        allowed_packs = doc.spec.allowed_policy_packs.len(),
        max_cells_per_hour = ?doc.spec.max_cells_per_hour,
        "T12: authorization policy loaded"
    );
    Ok(Some(doc.spec))
}

/// Returns `true` when `CELLOS_REQUIRE_TRUST_VERIFY_KEYS` is set to a truthy
/// value (`1` / `true` / `yes` / `on`). Mirrors the on/off-style env helpers
/// used elsewhere in the supervisor (`CELLOS_DNS_AUTHORITY_REFRESH`,
/// `CELLOS_TRUST_PLANE_EVENTS`, etc.).
fn trust_verify_keys_required() -> bool {
    std::env::var("CELLOS_REQUIRE_TRUST_VERIFY_KEYS")
        .map(|v| {
            let t = v.trim().to_ascii_lowercase();
            matches!(t.as_str(), "1" | "true" | "yes" | "on")
        })
        .unwrap_or(false)
}

fn build_host_backend(
    warnings: &mut StartupConfigWarnings,
    event_sink: Option<Arc<dyn EventSink>>,
) -> anyhow::Result<Arc<dyn CellBackend>> {
    let raw = std::env::var("CELLOS_CELL_BACKEND").unwrap_or_default();
    let backend: Arc<dyn CellBackend> = match raw.as_str() {
        "firecracker" => {
            let mut backend = FirecrackerCellBackend::from_env()
                .map_err(|e| anyhow::anyhow!("firecracker backend init: {e}"))?;
            if let Some(sink) = event_sink.clone() {
                backend = backend.with_event_sink(sink);
            }
            tracing::info!(
                target: "cellos.supervisor.host",
                backend_kind = "firecracker",
                jailer_configured = backend.config().jailer_binary_path.is_some(),
                chroot_base = %backend.config().chroot_base_dir.display(),
                "host backend selected"
            );
            // L2-06-2 wiring: when the operator has opted into the warm pool
            // (`CELLOS_FIRECRACKER_POOL_SIZE > 0`), spawn a best-effort
            // background task that drives one `fill()` cycle. Per the doc on
            // `FirecrackerPool::fill`, individual slot failures are logged and
            // leave the slot `Empty` — they never fail supervisor startup.
            // Linux-gated because Firecracker is Linux-only; on other targets
            // the pool is a zero-slot no-op and there is nothing to fill.
            #[cfg(target_os = "linux")]
            {
                let backend_for_fill: Arc<FirecrackerCellBackend> = Arc::new(backend);
                let pool_size_env = std::env::var(cellos_host_firecracker::pool::POOL_SIZE_ENV)
                    .ok()
                    .and_then(|v| v.trim().parse::<usize>().ok())
                    .unwrap_or(0);
                if pool_size_env > 0 {
                    let bg = Arc::clone(&backend_for_fill);
                    tokio::spawn(async move {
                        tracing::info!(
                            target: "cellos.supervisor.host",
                            pool_size = pool_size_env,
                            "warm pool fill: starting background fill task"
                        );
                        bg.fill_pool().await;
                        tracing::info!(
                            target: "cellos.supervisor.host",
                            available = bg.pool_available().await,
                            "warm pool fill: background fill task complete"
                        );
                    });
                }
                backend_for_fill as Arc<dyn CellBackend>
            }
            #[cfg(not(target_os = "linux"))]
            {
                Arc::new(backend) as Arc<dyn CellBackend>
            }
        }
        "stub" => {
            tracing::info!(
                target: "cellos.supervisor.host",
                backend_kind = "stub",
                "host backend selected"
            );
            Arc::new(StubCellBackend)
        }
        "gvisor" => {
            // L2-06-5 — gVisor (`runsc`) OCI-runtime backend. Linux-only
            // because `runsc` relies on `ptrace`/KVM/systrap; on macOS or
            // any other host we fail closed at startup rather than silently
            // falling back to the proprietary backend (mirrors firecracker's
            // posture: opt-in selection means opt-in failure).
            #[cfg(target_os = "linux")]
            {
                use cellos_host_gvisor::GVisorCellBackend;
                tracing::info!(
                    target: "cellos.supervisor.host",
                    backend_kind = "gvisor",
                    "host backend selected"
                );
                Arc::new(GVisorCellBackend::new()) as Arc<dyn CellBackend>
            }
            #[cfg(not(target_os = "linux"))]
            {
                anyhow::bail!(
                    "CELLOS_CELL_BACKEND=gvisor requires Linux (runsc is Linux-only); \
                     unset the variable or choose `stub` on non-Linux hosts"
                );
            }
        }
        "" => {
            // Unset — operator did not opt in to anything; fall through to the
            // default proprietary backend without recording a warning.
            tracing::info!(
                target: "cellos.supervisor.host",
                backend_kind = "proprietary",
                "host backend selected"
            );
            Arc::new(ProprietaryCellBackend::new())
        }
        _ => {
            // U1-04: unknown enum value. Previously this silently fell through
            // to the proprietary backend, which is dangerous if the operator
            // typo'd (e.g. `CELLOS_CELL_BACKEND=firecraker`) and expected a VM.
            warnings.record(
                "CELLOS_CELL_BACKEND",
                raw.clone(),
                "value didn't match enum {firecracker, gvisor, stub}; falling back to proprietary backend",
            );
            tracing::info!(
                target: "cellos.supervisor.host",
                backend_kind = "proprietary",
                "host backend selected"
            );
            Arc::new(ProprietaryCellBackend::new())
        }
    };
    Ok(backend)
}

fn build_secret_broker(
    warnings: &mut StartupConfigWarnings,
) -> anyhow::Result<Arc<dyn SecretBroker>> {
    let raw = std::env::var("CELLOS_BROKER").unwrap_or_default();
    let broker: Arc<dyn SecretBroker> = match raw.as_str() {
        "env" => Arc::new(EnvSecretBroker::new()),
        "file" => Arc::new(FileSecretBroker::new()),
        "github-oidc" => Arc::new(
            GithubActionsOidcBroker::new()
                .map_err(|e| anyhow::anyhow!("github-oidc broker init: {e}"))?,
        ),
        "vault-approle" => Arc::new(
            VaultAppRoleBroker::from_env()
                .map_err(|e| anyhow::anyhow!("vault-approle broker init: {e}"))?,
        ),
        "" => Arc::new(MemorySecretBroker::new()),
        _ => {
            // U1-04: unknown enum value. Previously this silently fell through
            // to the in-memory broker, which means an operator typo
            // (`CELLOS_BROKER=valt-approle`) would route real-world secret
            // requests to the toy broker without complaint.
            warnings.record(
                "CELLOS_BROKER",
                raw.clone(),
                "value didn't match enum {env, file, github-oidc, vault-approle}; falling back to in-memory broker",
            );
            Arc::new(MemorySecretBroker::new())
        }
    };
    Ok(broker)
}

async fn build_primary_event_sink(
    doc: &ExecutionCellDocument,
    run_id: &str,
    warnings: &mut StartupConfigWarnings,
) -> anyhow::Result<Arc<dyn EventSink>> {
    let event_sink: Arc<dyn EventSink> = if std::env::var("CELL_OS_USE_NOOP_SINK").is_ok() {
        // Explicit operator opt-in to noop — not a silent fallback, no warning.
        tracing::info!(
            target: "cellos.supervisor.observability",
            sink_kind = "noop",
            sink_reason = "CELL_OS_USE_NOOP_SINK",
            "primary event sink is noop (explicit env)"
        );
        Arc::new(NoopEventSink)
    } else {
        let nats_url = std::env::var("NATS_URL").unwrap_or_else(|_| "nats://127.0.0.1:4222".into());
        let nats_url_log = redact_url_credentials_for_logs(&nats_url);
        let subject = resolve_event_subject(
            &doc.spec.id,
            run_id,
            doc.spec
                .correlation
                .as_ref()
                .and_then(|c| c.tenant_id.as_deref()),
        );
        let nats_ca_file = std::env::var("NATS_CA_FILE")
            .ok()
            .filter(|s| !s.trim().is_empty())
            .map(std::path::PathBuf::from);
        let nats_ca_path = nats_ca_file.as_deref();

        match JetStreamEventSink::connect_with_root_ca(&nats_url, subject, nats_ca_path).await {
            Ok(sink) => {
                tracing::info!(
                    target: "cellos.supervisor.observability",
                    sink_kind = "jetstream",
                    nats_url = %nats_url_log,
                    "JetStream event sink connected"
                );
                Arc::new(sink)
            }
            Err(e) => {
                if std::env::var("CELL_OS_REQUIRE_JETSTREAM").is_ok() {
                    return Err(anyhow::anyhow!(
                        "CELL_OS_REQUIRE_JETSTREAM is set but JetStream connect failed (broker {nats_url_log}): {e}"
                    ));
                }
                // U1-04: noop fallback because JetStream connect failed and
                // the operator did NOT pin `CELL_OS_REQUIRE_JETSTREAM`. Record
                // through the warnings collector so strict mode escalates,
                // and keep the legacy targeted warn for log-based alerting.
                tracing::warn!(
                    target: "cellos.supervisor.observability",
                    sink_kind = "noop",
                    sink_reason = "jetstream_connect_failed",
                    error = %e,
                    nats_url = %nats_url_log,
                    "JetStream unavailable; using noop event sink"
                );
                warnings.record(
                    "NATS_URL",
                    nats_url_log.to_string(),
                    format!(
                        "JetStream connect failed and CELL_OS_REQUIRE_JETSTREAM is unset; \
                         primary event sink degraded to noop ({e})"
                    ),
                );
                Arc::new(NoopEventSink)
            }
        }
    };

    // Composition order (innermost → outermost):
    //   1. event_sink (primary transport)
    //   2. RedactingEventSink::from_env — strip secrets from event payloads
    //   3. SigningEventSink::from_env (I5) — sign-after-redact so the signed
    //      payload reflects the post-redaction text; a consumer re-running
    //      redaction would otherwise diverge from what the producer signed.
    //   4. DlqSink::from_env (P3-03) — OUTERMOST wrap. Captures emit failures
    //      surfaced by the inner pipeline (transport, redaction, OR signing).
    //      Disabled by default — identity unless CELLOS_DLQ_DIR is set to a
    //      non-empty, existing-and-writable directory.
    // U1-04: route SigningEventSink misconfig through StartupConfigWarnings
    // so a malformed CELLOS_EVENT_SIGNING_* triple fails loud under
    // CELLOS_STRICT_CONFIG=1 instead of silently degrading to passthrough
    // (reviewer wave 2, bebc77b).
    let (signed_sink, signing_warnings) =
        SigningEventSink::from_env_with_warnings(RedactingEventSink::from_env(event_sink));
    for w in signing_warnings {
        warnings.record(w.var, w.value, w.reason);
    }
    Ok(DlqSink::from_env(signed_sink))
}

fn build_jsonl_sink() -> Option<Arc<dyn EventSink>> {
    std::env::var("CELL_OS_JSONL_EVENTS")
        .ok()
        .map(|path| -> Arc<dyn EventSink> {
            // Same composition order as build_primary_event_sink:
            // DLQ(Sign(Redact(JsonlEventSink))).
            DlqSink::from_env(SigningEventSink::from_env(RedactingEventSink::from_env(
                Arc::new(JsonlEventSink::new(path)),
            )))
        })
}

fn build_export_sinks(
    doc: &ExecutionCellDocument,
    warnings: &mut StartupConfigWarnings,
) -> anyhow::Result<ExportSinkBundle> {
    let cell_id = doc.spec.id.clone();
    let named_export_sinks = build_named_export_sinks(doc, &cell_id, warnings)?;

    let default_export_sink: Arc<dyn ExportSink> =
        if scoped_env("CELLOS_EXPORT_HTTP_BASE_URL", None).is_some() {
            build_http_transport_sink(&cell_id, None, warnings)?
        } else if let Ok(root) = std::env::var("CELLOS_EXPORT_DIR") {
            Arc::new(
                LocalExportSink::new(root, &cell_id)
                    .map_err(|e| anyhow::anyhow!("CELLOS_EXPORT_DIR local sink: {e}"))?,
            )
        } else {
            // No default export configured. This is the documented "operator
            // didn't ask for export" path — we don't record a warning because
            // there is no companion var demanding export.
            Arc::new(NoopExportSink)
        };

    Ok((default_export_sink, named_export_sinks))
}

fn export_target_env_suffix(target_name: &str) -> String {
    target_name
        .chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() {
                c.to_ascii_uppercase()
            } else {
                '_'
            }
        })
        .collect()
}

fn scoped_env(base: &str, target_name: Option<&str>) -> Option<String> {
    if let Some(name) = target_name {
        let scoped = format!("{base}__{}", export_target_env_suffix(name));
        if let Ok(value) = std::env::var(&scoped) {
            if !value.trim().is_empty() {
                return Some(value);
            }
        }
    }
    std::env::var(base).ok().filter(|s| !s.trim().is_empty())
}

fn scoped_retry_attempts(base: &str, target_name: Option<&str>) -> anyhow::Result<usize> {
    let Some(raw) = scoped_env(base, target_name) else {
        return Ok(1);
    };
    let attempts = raw
        .parse::<usize>()
        .map_err(|e| anyhow::anyhow!("{base} must be an integer >= 1: {e}"))?;
    if attempts == 0 {
        return Err(anyhow::anyhow!("{base} must be >= 1"));
    }
    Ok(attempts)
}

fn scoped_retry_backoff_ms(base: &str, target_name: Option<&str>) -> anyhow::Result<u64> {
    let Some(raw) = scoped_env(base, target_name) else {
        return Ok(0);
    };
    raw.parse::<u64>()
        .map_err(|e| anyhow::anyhow!("{base} must be an integer >= 0: {e}"))
}

fn build_http_transport_sink(
    cell_id: &str,
    target_name: Option<&str>,
    warnings: &mut StartupConfigWarnings,
) -> anyhow::Result<Arc<dyn ExportSink>> {
    let Some(base_url) = scoped_env("CELLOS_EXPORT_HTTP_BASE_URL", target_name) else {
        if std::env::var("CELL_OS_REQUIRE_HTTP_EXPORT").is_ok() {
            let scope = target_name
                .map(|name| format!("target {name:?}"))
                .unwrap_or_else(|| "default export sink".into());
            return Err(anyhow::anyhow!(
                "CELL_OS_REQUIRE_HTTP_EXPORT is set but no HTTP export base URL is configured for {scope}"
            ));
        }
        // U1-04: HTTP export wiring requested by spec but the companion env
        // var is missing — record so strict mode can escalate. Keep the
        // legacy targeted warn for log-based alerting.
        if let Some(name) = target_name {
            tracing::warn!(target = %name, "HTTP export URL not configured; using noop sink");
            warnings.record(
                "CELLOS_EXPORT_HTTP_BASE_URL",
                String::new(),
                format!(
                    "missing companion var for HTTP export target {name:?}; degraded to noop sink"
                ),
            );
        } else {
            tracing::warn!("HTTP export URL not configured; using noop sink");
            warnings.record(
                "CELLOS_EXPORT_HTTP_BASE_URL",
                String::new(),
                "missing companion var for default HTTP export sink; degraded to noop sink",
            );
        }
        return Ok(Arc::new(NoopExportSink));
    };

    let bearer_token = scoped_env("CELLOS_EXPORT_HTTP_BEARER_TOKEN", target_name);
    let max_attempts = scoped_retry_attempts("CELLOS_EXPORT_HTTP_MAX_ATTEMPTS", target_name)?;
    let retry_backoff_ms =
        scoped_retry_backoff_ms("CELLOS_EXPORT_HTTP_RETRY_BACKOFF_MS", target_name)?;
    let sink = HttpExportSink::new(
        base_url,
        cell_id.to_string(),
        bearer_token,
        max_attempts,
        retry_backoff_ms,
    )
    .map_err(|e| anyhow::anyhow!("HTTP export sink init: {e}"))?;
    Ok(Arc::new(sink))
}

fn build_s3_transport_sink(
    cell_id: &str,
    target_name: &str,
    bucket: &str,
    key_prefix: Option<String>,
    region: Option<String>,
    warnings: &mut StartupConfigWarnings,
) -> anyhow::Result<Arc<dyn ExportSink>> {
    let presigned_url = if let Some(url) =
        scoped_env("CELLOS_EXPORT_S3_PRESIGNED_URL", Some(target_name))
    {
        Some(url)
    } else if let Some(url) = scoped_env("CELLOS_EXPORT_HTTP_BASE_URL", Some(target_name)) {
        // U1-04: legacy fallback path. The operator configured the wrong env
        // var name for this target — record so strict mode can refuse the
        // legacy alias.
        tracing::warn!(
            target = %target_name,
            "using legacy CELLOS_EXPORT_HTTP_BASE_URL fallback for S3 target; prefer CELLOS_EXPORT_S3_PRESIGNED_URL"
        );
        warnings.record(
            "CELLOS_EXPORT_HTTP_BASE_URL",
            url.clone(),
            format!(
                "legacy alias used for S3 target {target_name:?}; prefer CELLOS_EXPORT_S3_PRESIGNED_URL"
            ),
        );
        Some(url)
    } else {
        None
    };

    let Some(presigned_url) = presigned_url else {
        if std::env::var("CELL_OS_REQUIRE_S3_EXPORT").is_ok()
            || std::env::var("CELL_OS_REQUIRE_HTTP_EXPORT").is_ok()
        {
            return Err(anyhow::anyhow!(
                "S3 export is required but no presigned URL is configured for target {target_name:?}"
            ));
        }

        // U1-04: S3 export wiring requested by spec but the companion env var
        // is missing — record so strict mode can escalate. Keep the legacy
        // targeted warn for log-based alerting.
        tracing::warn!(
            target = %target_name,
            "S3 export URL not configured; using noop sink"
        );
        warnings.record(
            "CELLOS_EXPORT_S3_PRESIGNED_URL",
            String::new(),
            format!(
                "missing companion var for S3 export target {target_name:?}; degraded to noop sink"
            ),
        );
        return Ok(Arc::new(NoopExportSink));
    };

    // Region: prefer per-target env override, fall back to spec-declared region.
    let effective_region = scoped_env("CELLOS_EXPORT_S3_REGION", Some(target_name)).or(region);
    if let Some(ref r) = effective_region {
        tracing::info!(target = %target_name, region = %r, "S3 export target region");
    }

    let sink = PresignedS3ExportSink::new(
        presigned_url,
        cell_id.to_string(),
        bucket.to_string(),
        key_prefix,
        Some(target_name.to_string()),
        effective_region,
        scoped_retry_attempts("CELLOS_EXPORT_S3_MAX_ATTEMPTS", Some(target_name))?,
        scoped_retry_backoff_ms("CELLOS_EXPORT_S3_RETRY_BACKOFF_MS", Some(target_name))?,
    )
    .map_err(|e| anyhow::anyhow!("S3 export sink init: {e}"))?;
    Ok(Arc::new(sink))
}

fn build_named_export_sinks(
    doc: &ExecutionCellDocument,
    cell_id: &str,
    warnings: &mut StartupConfigWarnings,
) -> anyhow::Result<HashMap<String, Arc<dyn ExportSink>>> {
    let mut named = HashMap::new();
    let Some(export) = &doc.spec.export else {
        return Ok(named);
    };
    let Some(targets) = &export.targets else {
        return Ok(named);
    };

    for target in targets {
        let sink: Arc<dyn ExportSink> = match target {
            ExportTarget::Http(http) => {
                tracing::info!(target = %http.name, "configuring HTTP export target");
                build_http_transport_sink(cell_id, Some(&http.name), warnings)?
            }
            ExportTarget::S3(s3) => {
                tracing::info!(
                    target = %s3.name,
                    bucket = %s3.bucket,
                    region = s3.region.as_deref().unwrap_or("(spec unset)"),
                    "configuring S3 export target"
                );
                build_s3_transport_sink(
                    cell_id,
                    &s3.name,
                    &s3.bucket,
                    s3.key_prefix.clone(),
                    s3.region.clone(),
                    warnings,
                )?
            }
        };
        named.insert(target.name().to_string(), sink);
    }

    Ok(named)
}

#[cfg(test)]
mod tests {
    use std::sync::Mutex;

    use super::{
        enforce_authority_derivation_requirement, enforce_deployment_profile,
        enforce_isolation_default, export_target_env_suffix, forced_missing_primitives_from_env,
        probe_missing_isolation_primitives, scoped_env, scoped_retry_attempts,
        scoped_retry_backoff_ms, strict_config_enabled, DeploymentMode, StartupConfigWarnings,
    };
    use cellos_core::{
        AuthorityBundle, AuthorityCapability, AuthorityDerivationToken, AuthoritySignature,
        EgressRule, ExecutionCellDocument, ExecutionCellSpec, Lifetime, RoleId,
    };

    static ENV_MUTEX: Mutex<()> = Mutex::new(());

    #[test]
    fn scoped_env_prefers_target_specific_value() {
        let _guard = ENV_MUTEX.lock().unwrap();
        std::env::set_var(
            "CELLOS_EXPORT_HTTP_BASE_URL",
            "https://default.example/upload",
        );
        std::env::set_var(
            "CELLOS_EXPORT_HTTP_BASE_URL__ARTIFACT_BUCKET",
            "https://scoped.example/upload",
        );

        let value = scoped_env("CELLOS_EXPORT_HTTP_BASE_URL", Some("artifact-bucket"));

        std::env::remove_var("CELLOS_EXPORT_HTTP_BASE_URL");
        std::env::remove_var("CELLOS_EXPORT_HTTP_BASE_URL__ARTIFACT_BUCKET");
        assert_eq!(value.as_deref(), Some("https://scoped.example/upload"));
    }

    #[test]
    fn scoped_env_falls_back_to_unsuffixed_value() {
        let _guard = ENV_MUTEX.lock().unwrap();
        std::env::set_var(
            "CELLOS_EXPORT_S3_PRESIGNED_URL",
            "https://default.example/upload",
        );
        std::env::remove_var("CELLOS_EXPORT_S3_PRESIGNED_URL__ARTIFACT_BUCKET");

        let value = scoped_env("CELLOS_EXPORT_S3_PRESIGNED_URL", Some("artifact-bucket"));

        std::env::remove_var("CELLOS_EXPORT_S3_PRESIGNED_URL");
        assert_eq!(value.as_deref(), Some("https://default.example/upload"));
    }

    #[test]
    fn export_target_env_suffix_normalizes_name() {
        assert_eq!(
            export_target_env_suffix("artifact-bucket.eu-west-1"),
            "ARTIFACT_BUCKET_EU_WEST_1"
        );
    }

    #[test]
    fn scoped_retry_attempts_prefers_target_specific_value() {
        let _guard = ENV_MUTEX.lock().unwrap();
        std::env::set_var("CELLOS_EXPORT_HTTP_MAX_ATTEMPTS", "2");
        std::env::set_var("CELLOS_EXPORT_HTTP_MAX_ATTEMPTS__ARTIFACT_API", "4");

        let attempts =
            scoped_retry_attempts("CELLOS_EXPORT_HTTP_MAX_ATTEMPTS", Some("artifact-api"))
                .expect("parse attempts");

        std::env::remove_var("CELLOS_EXPORT_HTTP_MAX_ATTEMPTS");
        std::env::remove_var("CELLOS_EXPORT_HTTP_MAX_ATTEMPTS__ARTIFACT_API");
        assert_eq!(attempts, 4);
    }

    #[test]
    fn scoped_retry_attempts_rejects_zero() {
        let _guard = ENV_MUTEX.lock().unwrap();
        std::env::set_var("CELLOS_EXPORT_S3_MAX_ATTEMPTS", "0");

        let err = scoped_retry_attempts("CELLOS_EXPORT_S3_MAX_ATTEMPTS", Some("artifact-bucket"))
            .expect_err("zero attempts should fail");

        std::env::remove_var("CELLOS_EXPORT_S3_MAX_ATTEMPTS");
        assert!(err.to_string().contains("must be >= 1"));
    }

    #[test]
    fn scoped_retry_backoff_defaults_to_zero() {
        let _guard = ENV_MUTEX.lock().unwrap();
        std::env::remove_var("CELLOS_EXPORT_HTTP_RETRY_BACKOFF_MS");

        let backoff =
            scoped_retry_backoff_ms("CELLOS_EXPORT_HTTP_RETRY_BACKOFF_MS", Some("artifact-api"))
                .expect("default backoff");

        assert_eq!(backoff, 0);
    }

    #[test]
    fn hardened_profile_requires_authority_keys_path() {
        let _guard = ENV_MUTEX.lock().unwrap();

        // Capture pre-test state so we can restore exactly. enforce_deployment_profile()
        // may auto-set CELL_OS_REQUIRE_JETSTREAM / CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS /
        // CELLOS_REQUIRE_AUTHORITY_DERIVATION, but in this failure case the
        // missing-keys-path check returns Err before any env mutation happens.
        // We still defensively snapshot+restore so future test ordering changes
        // can't leak state.
        let prev_profile = std::env::var_os("CELLOS_DEPLOYMENT_PROFILE");
        let prev_keys_path = std::env::var_os("CELLOS_AUTHORITY_KEYS_PATH");
        let prev_require_js = std::env::var_os("CELL_OS_REQUIRE_JETSTREAM");
        let prev_require_scoped = std::env::var_os("CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS");
        let prev_require_deriv = std::env::var_os("CELLOS_REQUIRE_AUTHORITY_DERIVATION");
        let prev_require_telemetry = std::env::var_os("CELLOS_REQUIRE_TELEMETRY_DECLARED");

        std::env::set_var("CELLOS_DEPLOYMENT_PROFILE", "hardened");
        std::env::remove_var("CELLOS_AUTHORITY_KEYS_PATH");

        let err = enforce_deployment_profile()
            .expect_err("hardened profile without authority keys path must error");
        let msg = err.to_string();

        // Restore before assert so a panic doesn't leave global env mutated.
        match prev_profile {
            Some(v) => std::env::set_var("CELLOS_DEPLOYMENT_PROFILE", v),
            None => std::env::remove_var("CELLOS_DEPLOYMENT_PROFILE"),
        }
        match prev_keys_path {
            Some(v) => std::env::set_var("CELLOS_AUTHORITY_KEYS_PATH", v),
            None => std::env::remove_var("CELLOS_AUTHORITY_KEYS_PATH"),
        }
        match prev_require_js {
            Some(v) => std::env::set_var("CELL_OS_REQUIRE_JETSTREAM", v),
            None => std::env::remove_var("CELL_OS_REQUIRE_JETSTREAM"),
        }
        match prev_require_scoped {
            Some(v) => std::env::set_var("CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS", v),
            None => std::env::remove_var("CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS"),
        }
        match prev_require_deriv {
            Some(v) => std::env::set_var("CELLOS_REQUIRE_AUTHORITY_DERIVATION", v),
            None => std::env::remove_var("CELLOS_REQUIRE_AUTHORITY_DERIVATION"),
        }
        match prev_require_telemetry {
            Some(v) => std::env::set_var("CELLOS_REQUIRE_TELEMETRY_DECLARED", v),
            None => std::env::remove_var("CELLOS_REQUIRE_TELEMETRY_DECLARED"),
        }

        assert!(
            msg.contains("CELLOS_AUTHORITY_KEYS_PATH"),
            "error message should mention CELLOS_AUTHORITY_KEYS_PATH; got: {msg}"
        );
    }

    #[test]
    fn hardened_profile_auto_enables_require_authority_derivation() {
        let _guard = ENV_MUTEX.lock().unwrap();

        // Snapshot every env var enforce_deployment_profile() may touch.
        let prev_profile = std::env::var_os("CELLOS_DEPLOYMENT_PROFILE");
        let prev_keys_path = std::env::var_os("CELLOS_AUTHORITY_KEYS_PATH");
        let prev_require_js = std::env::var_os("CELL_OS_REQUIRE_JETSTREAM");
        let prev_require_scoped = std::env::var_os("CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS");
        let prev_require_deriv = std::env::var_os("CELLOS_REQUIRE_AUTHORITY_DERIVATION");
        let prev_require_telemetry = std::env::var_os("CELLOS_REQUIRE_TELEMETRY_DECLARED");

        std::env::set_var("CELLOS_DEPLOYMENT_PROFILE", "hardened");
        std::env::set_var("CELLOS_AUTHORITY_KEYS_PATH", "/dev/null");
        std::env::remove_var("CELLOS_REQUIRE_AUTHORITY_DERIVATION");
        std::env::remove_var("CELLOS_REQUIRE_TELEMETRY_DECLARED");

        let result = enforce_deployment_profile();
        let after = std::env::var("CELLOS_REQUIRE_AUTHORITY_DERIVATION").ok();
        let after_telemetry = std::env::var("CELLOS_REQUIRE_TELEMETRY_DECLARED").ok();

        // Restore.
        match prev_profile {
            Some(v) => std::env::set_var("CELLOS_DEPLOYMENT_PROFILE", v),
            None => std::env::remove_var("CELLOS_DEPLOYMENT_PROFILE"),
        }
        match prev_keys_path {
            Some(v) => std::env::set_var("CELLOS_AUTHORITY_KEYS_PATH", v),
            None => std::env::remove_var("CELLOS_AUTHORITY_KEYS_PATH"),
        }
        match prev_require_js {
            Some(v) => std::env::set_var("CELL_OS_REQUIRE_JETSTREAM", v),
            None => std::env::remove_var("CELL_OS_REQUIRE_JETSTREAM"),
        }
        match prev_require_scoped {
            Some(v) => std::env::set_var("CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS", v),
            None => std::env::remove_var("CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS"),
        }
        match prev_require_deriv {
            Some(v) => std::env::set_var("CELLOS_REQUIRE_AUTHORITY_DERIVATION", v),
            None => std::env::remove_var("CELLOS_REQUIRE_AUTHORITY_DERIVATION"),
        }
        match prev_require_telemetry {
            Some(v) => std::env::set_var("CELLOS_REQUIRE_TELEMETRY_DECLARED", v),
            None => std::env::remove_var("CELLOS_REQUIRE_TELEMETRY_DECLARED"),
        }

        result.expect("hardened profile with keys path must succeed");
        assert_eq!(
            after.as_deref(),
            Some("1"),
            "hardened profile should auto-enable CELLOS_REQUIRE_AUTHORITY_DERIVATION"
        );
        assert_eq!(
            after_telemetry.as_deref(),
            Some("1"),
            "hardened profile should auto-enable CELLOS_REQUIRE_TELEMETRY_DECLARED"
        );
    }

    /// Wave 2 red-team (MED-W2D-5): hardened mode previously left
    /// `CELLOS_STRICT_CONFIG` untouched, so an env-var typo like
    /// `CELLOS_BROKER=valt-approle` would silently fall through to the
    /// in-memory toy broker even under hardened authority/jetstream/
    /// telemetry guarantees. This test pins the new auto-enable
    /// behaviour so a future drive-by edit can't regress the hardening
    /// asymmetry.
    #[test]
    fn hardened_profile_auto_enables_strict_config() {
        let _guard = ENV_MUTEX.lock().unwrap();

        let prev_profile = std::env::var_os("CELLOS_DEPLOYMENT_PROFILE");
        let prev_keys_path = std::env::var_os("CELLOS_AUTHORITY_KEYS_PATH");
        let prev_require_js = std::env::var_os("CELL_OS_REQUIRE_JETSTREAM");
        let prev_require_scoped = std::env::var_os("CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS");
        let prev_require_deriv = std::env::var_os("CELLOS_REQUIRE_AUTHORITY_DERIVATION");
        let prev_require_telemetry = std::env::var_os("CELLOS_REQUIRE_TELEMETRY_DECLARED");
        let prev_strict = std::env::var_os("CELLOS_STRICT_CONFIG");

        std::env::set_var("CELLOS_DEPLOYMENT_PROFILE", "hardened");
        std::env::set_var("CELLOS_AUTHORITY_KEYS_PATH", "/dev/null");
        std::env::remove_var("CELLOS_STRICT_CONFIG");

        let result = enforce_deployment_profile();
        let after_strict = std::env::var("CELLOS_STRICT_CONFIG").ok();

        // Restore.
        match prev_profile {
            Some(v) => std::env::set_var("CELLOS_DEPLOYMENT_PROFILE", v),
            None => std::env::remove_var("CELLOS_DEPLOYMENT_PROFILE"),
        }
        match prev_keys_path {
            Some(v) => std::env::set_var("CELLOS_AUTHORITY_KEYS_PATH", v),
            None => std::env::remove_var("CELLOS_AUTHORITY_KEYS_PATH"),
        }
        match prev_require_js {
            Some(v) => std::env::set_var("CELL_OS_REQUIRE_JETSTREAM", v),
            None => std::env::remove_var("CELL_OS_REQUIRE_JETSTREAM"),
        }
        match prev_require_scoped {
            Some(v) => std::env::set_var("CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS", v),
            None => std::env::remove_var("CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS"),
        }
        match prev_require_deriv {
            Some(v) => std::env::set_var("CELLOS_REQUIRE_AUTHORITY_DERIVATION", v),
            None => std::env::remove_var("CELLOS_REQUIRE_AUTHORITY_DERIVATION"),
        }
        match prev_require_telemetry {
            Some(v) => std::env::set_var("CELLOS_REQUIRE_TELEMETRY_DECLARED", v),
            None => std::env::remove_var("CELLOS_REQUIRE_TELEMETRY_DECLARED"),
        }
        match prev_strict {
            Some(v) => std::env::set_var("CELLOS_STRICT_CONFIG", v),
            None => std::env::remove_var("CELLOS_STRICT_CONFIG"),
        }

        result.expect("hardened profile with keys path must succeed");
        assert_eq!(
            after_strict.as_deref(),
            Some("1"),
            "hardened profile must auto-enable CELLOS_STRICT_CONFIG so env-var typos fail loud"
        );
    }

    /// Companion: when the operator explicitly sets `CELLOS_STRICT_CONFIG`
    /// (to any value), hardened mode must respect that choice rather than
    /// silently overwriting it. Otherwise the auto-enable would surprise an
    /// operator who deliberately set `CELLOS_STRICT_CONFIG=0` for a soak
    /// run.
    #[test]
    fn hardened_profile_respects_explicit_strict_config_value() {
        let _guard = ENV_MUTEX.lock().unwrap();

        let prev_profile = std::env::var_os("CELLOS_DEPLOYMENT_PROFILE");
        let prev_keys_path = std::env::var_os("CELLOS_AUTHORITY_KEYS_PATH");
        let prev_require_js = std::env::var_os("CELL_OS_REQUIRE_JETSTREAM");
        let prev_require_scoped = std::env::var_os("CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS");
        let prev_require_deriv = std::env::var_os("CELLOS_REQUIRE_AUTHORITY_DERIVATION");
        let prev_require_telemetry = std::env::var_os("CELLOS_REQUIRE_TELEMETRY_DECLARED");
        let prev_strict = std::env::var_os("CELLOS_STRICT_CONFIG");

        std::env::set_var("CELLOS_DEPLOYMENT_PROFILE", "hardened");
        std::env::set_var("CELLOS_AUTHORITY_KEYS_PATH", "/dev/null");
        std::env::set_var("CELLOS_STRICT_CONFIG", "0");

        let result = enforce_deployment_profile();
        let after_strict = std::env::var("CELLOS_STRICT_CONFIG").ok();

        // Restore.
        match prev_profile {
            Some(v) => std::env::set_var("CELLOS_DEPLOYMENT_PROFILE", v),
            None => std::env::remove_var("CELLOS_DEPLOYMENT_PROFILE"),
        }
        match prev_keys_path {
            Some(v) => std::env::set_var("CELLOS_AUTHORITY_KEYS_PATH", v),
            None => std::env::remove_var("CELLOS_AUTHORITY_KEYS_PATH"),
        }
        match prev_require_js {
            Some(v) => std::env::set_var("CELL_OS_REQUIRE_JETSTREAM", v),
            None => std::env::remove_var("CELL_OS_REQUIRE_JETSTREAM"),
        }
        match prev_require_scoped {
            Some(v) => std::env::set_var("CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS", v),
            None => std::env::remove_var("CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS"),
        }
        match prev_require_deriv {
            Some(v) => std::env::set_var("CELLOS_REQUIRE_AUTHORITY_DERIVATION", v),
            None => std::env::remove_var("CELLOS_REQUIRE_AUTHORITY_DERIVATION"),
        }
        match prev_require_telemetry {
            Some(v) => std::env::set_var("CELLOS_REQUIRE_TELEMETRY_DECLARED", v),
            None => std::env::remove_var("CELLOS_REQUIRE_TELEMETRY_DECLARED"),
        }
        match prev_strict {
            Some(v) => std::env::set_var("CELLOS_STRICT_CONFIG", v),
            None => std::env::remove_var("CELLOS_STRICT_CONFIG"),
        }

        result.expect("hardened profile with keys path must succeed");
        assert_eq!(
            after_strict.as_deref(),
            Some("0"),
            "explicit operator CELLOS_STRICT_CONFIG=0 must be preserved (auto-enable is opt-out)"
        );
    }

    // ── enforce_authority_derivation_requirement (SEAM-3) ─────────────────
    //
    // These run with ENV_MUTEX held because they mutate a shared process-wide
    // env var that other tests in the crate can read.

    fn doc_with_authority(bundle: AuthorityBundle) -> ExecutionCellDocument {
        ExecutionCellDocument {
            api_version: "cellos.io/v1".into(),
            kind: "ExecutionCell".into(),
            spec: ExecutionCellSpec {
                id: "seam3-test".into(),
                correlation: None,
                ingress: None,
                environment: None,
                placement: None,
                policy: None,
                identity: None,
                run: None,
                authority: bundle,
                lifetime: Lifetime { ttl_seconds: 60 },
                export: None,
                telemetry: None,
            },
        }
    }

    fn stub_token() -> AuthorityDerivationToken {
        // Structurally valid; no signature verification happens in this code path.
        AuthorityDerivationToken {
            role_root: RoleId("role-test".into()),
            parent_run_id: Some("run-seam3".into()),
            derivation_steps: vec![],
            leaf_capability: AuthorityCapability {
                egress_rules: vec![],
                secret_refs: vec![],
            },
            grantor_signature: AuthoritySignature {
                algorithm: "ed25519".into(),
                bytes: "AA==".into(),
            },
        }
    }

    /// Snapshot+restore wrapper for CELLOS_REQUIRE_AUTHORITY_DERIVATION.
    fn with_require_authority_derivation<F: FnOnce() -> R, R>(set: bool, f: F) -> R {
        let prev = std::env::var_os("CELLOS_REQUIRE_AUTHORITY_DERIVATION");
        if set {
            std::env::set_var("CELLOS_REQUIRE_AUTHORITY_DERIVATION", "1");
        } else {
            std::env::remove_var("CELLOS_REQUIRE_AUTHORITY_DERIVATION");
        }
        let out = f();
        match prev {
            Some(v) => std::env::set_var("CELLOS_REQUIRE_AUTHORITY_DERIVATION", v),
            None => std::env::remove_var("CELLOS_REQUIRE_AUTHORITY_DERIVATION"),
        }
        out
    }

    #[test]
    fn require_authority_derivation_passes_when_flag_unset() {
        let _guard = ENV_MUTEX.lock().unwrap();
        let doc = doc_with_authority(AuthorityBundle {
            secret_refs: Some(vec!["MY_SECRET".into()]),
            ..AuthorityBundle::default()
        });
        with_require_authority_derivation(false, || {
            enforce_authority_derivation_requirement(&doc)
                .expect("flag unset → spec without token must pass");
        });
    }

    #[test]
    fn require_authority_derivation_rejects_secret_refs_without_token() {
        let _guard = ENV_MUTEX.lock().unwrap();
        let doc = doc_with_authority(AuthorityBundle {
            secret_refs: Some(vec!["MY_SECRET".into()]),
            ..AuthorityBundle::default()
        });
        let err = with_require_authority_derivation(true, || {
            enforce_authority_derivation_requirement(&doc)
                .expect_err("flag set + authority + no token must fail")
        });
        let msg = err.to_string();
        assert!(
            msg.contains("authorityDerivation"),
            "error must mention authorityDerivation; got: {msg}"
        );
    }

    #[test]
    fn require_authority_derivation_rejects_egress_rules_without_token() {
        let _guard = ENV_MUTEX.lock().unwrap();
        let doc = doc_with_authority(AuthorityBundle {
            egress_rules: Some(vec![EgressRule {
                host: "api.example.com".into(),
                port: 443,
                protocol: Some("https".into()),
                dns_egress_justification: None,
            }]),
            ..AuthorityBundle::default()
        });
        let err = with_require_authority_derivation(true, || {
            enforce_authority_derivation_requirement(&doc)
                .expect_err("flag set + egress + no token must fail")
        });
        let msg = err.to_string();
        assert!(
            msg.contains("authorityDerivation"),
            "error must mention authorityDerivation; got: {msg}"
        );
    }

    #[test]
    fn require_authority_derivation_passes_when_token_present() {
        let _guard = ENV_MUTEX.lock().unwrap();
        let doc = doc_with_authority(AuthorityBundle {
            secret_refs: Some(vec!["MY_SECRET".into()]),
            authority_derivation: Some(stub_token()),
            ..AuthorityBundle::default()
        });
        with_require_authority_derivation(true, || {
            enforce_authority_derivation_requirement(&doc)
                .expect("flag set + authority + token must pass");
        });
    }

    #[test]
    fn require_authority_derivation_passes_for_empty_authority() {
        let _guard = ENV_MUTEX.lock().unwrap();
        let doc = doc_with_authority(AuthorityBundle::default());
        with_require_authority_derivation(true, || {
            enforce_authority_derivation_requirement(&doc)
                .expect("flag set + empty authority must pass");
        });
    }

    #[test]
    fn require_authority_derivation_passes_for_empty_secret_refs_vec() {
        let _guard = ENV_MUTEX.lock().unwrap();
        // `Some(vec![])` should count as empty — there is nothing to delegate.
        let doc = doc_with_authority(AuthorityBundle {
            egress_rules: Some(vec![]),
            secret_refs: Some(vec![]),
            ..AuthorityBundle::default()
        });
        with_require_authority_derivation(true, || {
            enforce_authority_derivation_requirement(&doc)
                .expect("flag set + empty (Some(vec![])) authority must pass");
        });
    }

    // ── I4: CELLOS_TEST_FORCE_MISSING_ISOLATION_PRIMITIVES probe hook ─────
    //
    // These tests mutate a process-wide env var read by
    // `forced_missing_primitives_from_env` / `probe_missing_isolation_primitives`,
    // so they share the crate-wide ENV_MUTEX with the other env-driven tests.

    /// Snapshot+restore wrapper for CELLOS_TEST_FORCE_MISSING_ISOLATION_PRIMITIVES.
    fn with_forced_missing<F: FnOnce() -> R, R>(value: Option<&str>, f: F) -> R {
        let prev = std::env::var_os("CELLOS_TEST_FORCE_MISSING_ISOLATION_PRIMITIVES");
        match value {
            Some(v) => std::env::set_var("CELLOS_TEST_FORCE_MISSING_ISOLATION_PRIMITIVES", v),
            None => std::env::remove_var("CELLOS_TEST_FORCE_MISSING_ISOLATION_PRIMITIVES"),
        }
        let out = f();
        match prev {
            Some(v) => std::env::set_var("CELLOS_TEST_FORCE_MISSING_ISOLATION_PRIMITIVES", v),
            None => std::env::remove_var("CELLOS_TEST_FORCE_MISSING_ISOLATION_PRIMITIVES"),
        }
        out
    }

    // ── enforce_telemetry_declared (F4a) ───────────────────────────────────

    fn with_require_telemetry_declared<F: FnOnce() -> R, R>(set: bool, f: F) -> R {
        let prev = std::env::var_os("CELLOS_REQUIRE_TELEMETRY_DECLARED");
        if set {
            std::env::set_var("CELLOS_REQUIRE_TELEMETRY_DECLARED", "1");
        } else {
            std::env::remove_var("CELLOS_REQUIRE_TELEMETRY_DECLARED");
        }
        let out = f();
        match prev {
            Some(v) => std::env::set_var("CELLOS_REQUIRE_TELEMETRY_DECLARED", v),
            None => std::env::remove_var("CELLOS_REQUIRE_TELEMETRY_DECLARED"),
        }
        out
    }

    // ── U1-04: CELLOS_STRICT_CONFIG + StartupConfigWarnings ───────────────

    /// Snapshot+restore wrapper for CELLOS_STRICT_CONFIG.
    fn with_strict_config<F: FnOnce() -> R, R>(value: Option<&str>, f: F) -> R {
        let prev = std::env::var_os("CELLOS_STRICT_CONFIG");
        match value {
            Some(v) => std::env::set_var("CELLOS_STRICT_CONFIG", v),
            None => std::env::remove_var("CELLOS_STRICT_CONFIG"),
        }
        let out = f();
        match prev {
            Some(v) => std::env::set_var("CELLOS_STRICT_CONFIG", v),
            None => std::env::remove_var("CELLOS_STRICT_CONFIG"),
        }
        out
    }

    #[test]
    fn forced_missing_single_primitive() {
        let _guard = ENV_MUTEX.lock().unwrap();
        let parsed = with_forced_missing(Some("unshare"), forced_missing_primitives_from_env);
        assert_eq!(parsed, Some(vec!["unshare"]));
    }

    #[test]
    fn forced_missing_multiple_primitives() {
        let _guard = ENV_MUTEX.lock().unwrap();
        let parsed =
            with_forced_missing(Some("unshare,seccomp"), forced_missing_primitives_from_env);
        assert_eq!(parsed, Some(vec!["unshare", "seccomp"]));
    }

    #[test]
    fn forced_missing_all_wildcard_returns_full_set() {
        let _guard = ENV_MUTEX.lock().unwrap();
        let via_all = with_forced_missing(Some("all"), forced_missing_primitives_from_env);
        let via_star = with_forced_missing(Some("*"), forced_missing_primitives_from_env);
        let expected = Some(vec!["unshare", "seccomp", "cgroup-v2"]);
        assert_eq!(via_all, expected);
        assert_eq!(via_star, expected);
    }

    #[test]
    fn forced_missing_unknown_token_ignored_known_kept() {
        let _guard = ENV_MUTEX.lock().unwrap();
        let parsed = with_forced_missing(
            Some("unshare,bogus-primitive,cgroup-v2"),
            forced_missing_primitives_from_env,
        );
        assert_eq!(parsed, Some(vec!["unshare", "cgroup-v2"]));
    }

    #[test]
    fn forced_missing_unset_returns_none() {
        let _guard = ENV_MUTEX.lock().unwrap();
        let parsed = with_forced_missing(None, forced_missing_primitives_from_env);
        assert!(
            parsed.is_none(),
            "unset env must return None, got {parsed:?}"
        );
    }

    #[test]
    fn forced_missing_drives_probe_output() {
        let _guard = ENV_MUTEX.lock().unwrap();
        let probed = with_forced_missing(Some("seccomp"), probe_missing_isolation_primitives);
        assert_eq!(probed, vec!["seccomp"]);
    }

    #[test]
    fn enforce_isolation_default_hardened_errors_when_primitives_missing() {
        let _guard = ENV_MUTEX.lock().unwrap();

        let prev_require_iso = std::env::var_os("CELLOS_REQUIRE_ISOLATION");
        std::env::remove_var("CELLOS_REQUIRE_ISOLATION");

        let err = with_forced_missing(Some("unshare"), || {
            enforce_isolation_default(DeploymentMode::Hardened)
                .expect_err("hardened + forced-missing must error")
        });
        let msg = err.to_string();

        match prev_require_iso {
            Some(v) => std::env::set_var("CELLOS_REQUIRE_ISOLATION", v),
            None => std::env::remove_var("CELLOS_REQUIRE_ISOLATION"),
        }

        assert!(
            msg.contains("host is missing isolation primitives"),
            "error must surface the structured-error phrase; got: {msg}"
        );
    }

    #[test]
    fn enforce_isolation_default_portable_warns_but_succeeds_when_primitives_missing() {
        let _guard = ENV_MUTEX.lock().unwrap();

        with_forced_missing(Some("all"), || {
            enforce_isolation_default(DeploymentMode::Portable)
                .expect("portable mode must succeed even with missing primitives");
        });
    }

    #[test]
    fn require_telemetry_declared_passes_when_flag_unset() {
        let _guard = ENV_MUTEX.lock().unwrap();
        let doc = doc_with_authority(AuthorityBundle::default());
        with_require_telemetry_declared(false, || {
            super::enforce_telemetry_declared(&doc)
                .expect("flag unset → spec without telemetry must pass");
        });
    }

    #[test]
    fn require_telemetry_declared_rejects_spec_without_telemetry() {
        let _guard = ENV_MUTEX.lock().unwrap();
        let doc = doc_with_authority(AuthorityBundle::default());
        let err = with_require_telemetry_declared(true, || {
            super::enforce_telemetry_declared(&doc)
                .expect_err("flag set + missing telemetry must fail")
        });
        let msg = err.to_string();
        assert!(
            msg.contains("spec.telemetry"),
            "error must mention spec.telemetry; got: {msg}"
        );
    }

    #[test]
    fn require_telemetry_declared_passes_when_telemetry_present() {
        use cellos_core::{TelemetryChannel, TelemetrySpec};
        let _guard = ENV_MUTEX.lock().unwrap();
        let mut doc = doc_with_authority(AuthorityBundle::default());
        doc.spec.telemetry = Some(TelemetrySpec {
            channel: TelemetryChannel::VsockCbor,
            events: vec!["process.spawn".into()],
            rate_limits: None,
            host_vs_guest_fields: None,
            agent_version: "1.0.0".into(),
        });
        with_require_telemetry_declared(true, || {
            super::enforce_telemetry_declared(&doc)
                .expect("flag set + telemetry present must pass");
        });
    }

    // ── Red-team D — admission negative tests ─────────────────────────────
    //
    // These tests harden the admission contract against the failure modes
    // documented in `red-team-findings-D.md` (NOOP-D9): the egress / secrets
    // gate must fire FIRST when both that gate AND the telemetry gate would
    // otherwise reject a spec, and the `Some(vec![])` empty-bundle case must
    // not be confused with the non-empty case. ADR-0006 Doctrine #11 reads:
    // "egress validated at admission BEFORE telemetry contract is read" —
    // so when an operator runs the equivalent of `CELLOS_DEPLOYMENT_PROFILE=
    // hardened`, the error message must mention `authorityDerivation`, not
    // `spec.telemetry`, when both gates would reject.
    //
    // These run with ENV_MUTEX held because they set BOTH
    // CELLOS_REQUIRE_AUTHORITY_DERIVATION and CELLOS_REQUIRE_TELEMETRY_DECLARED
    // in the process and other tests in the crate share the env.

    /// Red-team D — when BOTH the egress gate and the telemetry gate would
    /// reject the spec, the EGRESS error must be the first to surface (this
    /// is what `main.rs` line ordering guarantees, and what ADR-0006 #11
    /// promises operators). A regression that swapped the gate order would
    /// have telemetry fail first and the operator would never see the
    /// (more important) egress-authority error.
    #[test]
    fn redteam_d_egress_gate_fires_before_telemetry_gate() {
        let _guard = ENV_MUTEX.lock().unwrap();
        // Spec has egress + no token AND no telemetry — both gates would reject.
        let doc = doc_with_authority(AuthorityBundle {
            egress_rules: Some(vec![EgressRule {
                host: "api.example.com".into(),
                port: 443,
                protocol: Some("https".into()),
                dns_egress_justification: None,
            }]),
            ..AuthorityBundle::default()
        });
        // Set both env flags at once to mirror hardened profile.
        with_require_authority_derivation(true, || {
            with_require_telemetry_declared(true, || {
                // Mirror main.rs ordering: egress first, then telemetry.
                let egress_result = enforce_authority_derivation_requirement(&doc);
                let err = egress_result
                    .expect_err("egress gate must reject before telemetry gate is consulted");
                let msg = err.to_string();
                assert!(
                    msg.contains("authorityDerivation"),
                    "first gate to fail must be the egress / authorityDerivation gate; got: {msg}"
                );
                // Telemetry gate would ALSO have failed, but main.rs would
                // never reach it because egress short-circuits.
                assert!(
                    !msg.contains("spec.telemetry"),
                    "egress error must NOT mention telemetry — confirms gate ordering; got: {msg}"
                );
            });
        });
    }

    /// Red-team D — when the egress gate passes (no authority bundle) but
    /// telemetry is missing, the telemetry gate must still fire. This is
    /// the "second gate isn't disabled by an earlier passing gate" check.
    #[test]
    fn redteam_d_telemetry_gate_still_fires_when_egress_gate_passes() {
        let _guard = ENV_MUTEX.lock().unwrap();
        // Empty authority bundle → egress gate passes. Telemetry missing →
        // telemetry gate must reject.
        let doc = doc_with_authority(AuthorityBundle::default());
        with_require_authority_derivation(true, || {
            with_require_telemetry_declared(true, || {
                enforce_authority_derivation_requirement(&doc)
                    .expect("empty authority bundle must pass egress gate");
                let err = super::enforce_telemetry_declared(&doc)
                    .expect_err("missing telemetry must fail the telemetry gate");
                assert!(err.to_string().contains("spec.telemetry"));
            });
        });
    }

    /// Red-team D — egress with empty Vec AND non-empty secret_refs must
    /// still fail the egress gate (secret_refs alone is sufficient to
    /// require a derivation token). Confirms the "either egress OR secrets
    /// triggers the gate" disjunction is not accidentally an AND.
    #[test]
    fn redteam_d_empty_egress_with_non_empty_secrets_still_requires_token() {
        let _guard = ENV_MUTEX.lock().unwrap();
        let doc = doc_with_authority(AuthorityBundle {
            egress_rules: Some(vec![]), // Some(empty)
            secret_refs: Some(vec!["AWS_ACCESS".into()]),
            ..AuthorityBundle::default()
        });
        with_require_authority_derivation(true, || {
            let err = enforce_authority_derivation_requirement(&doc)
                .expect_err("non-empty secret_refs + no token must still fail");
            assert!(err.to_string().contains("authorityDerivation"));
        });
    }

    /// Red-team D — non-empty egress_rules AND empty secret_refs must
    /// also fail (mirror of above; the OR is symmetric).
    #[test]
    fn redteam_d_non_empty_egress_with_empty_secrets_requires_token() {
        let _guard = ENV_MUTEX.lock().unwrap();
        let doc = doc_with_authority(AuthorityBundle {
            egress_rules: Some(vec![EgressRule {
                host: "api.example.com".into(),
                port: 443,
                protocol: Some("https".into()),
                dns_egress_justification: None,
            }]),
            secret_refs: Some(vec![]),
            ..AuthorityBundle::default()
        });
        with_require_authority_derivation(true, || {
            let err = enforce_authority_derivation_requirement(&doc)
                .expect_err("non-empty egress_rules + empty secret_refs + no token must fail");
            assert!(err.to_string().contains("authorityDerivation"));
        });
    }

    // ── T11-5 — enforce_kubernetes_namespace_placement ─────────────────────

    fn with_k8s_namespace<F: FnOnce() -> R, R>(value: Option<&str>, f: F) -> R {
        let prev = std::env::var_os("CELLOS_K8S_NAMESPACE");
        match value {
            Some(v) => std::env::set_var("CELLOS_K8S_NAMESPACE", v),
            None => std::env::remove_var("CELLOS_K8S_NAMESPACE"),
        }
        let out = f();
        match prev {
            Some(v) => std::env::set_var("CELLOS_K8S_NAMESPACE", v),
            None => std::env::remove_var("CELLOS_K8S_NAMESPACE"),
        }
        out
    }

    fn doc_with_namespace(namespace: Option<&str>) -> ExecutionCellDocument {
        let mut doc = doc_with_authority(AuthorityBundle::default());
        if let Some(ns) = namespace {
            doc.spec.placement = Some(cellos_core::PlacementSpec {
                pool_id: None,
                kubernetes_namespace: Some(ns.to_string()),
                queue_name: None,
            });
        }
        doc
    }

    #[test]
    fn k8s_namespace_unset_accepts_any_spec() {
        let _guard = ENV_MUTEX.lock().unwrap();
        let doc = doc_with_namespace(Some("cellos-prod"));
        with_k8s_namespace(None, || {
            super::enforce_kubernetes_namespace_placement(&doc)
                .expect("unset env must accept any spec");
        });
    }

    #[test]
    fn k8s_namespace_set_matches_spec() {
        let _guard = ENV_MUTEX.lock().unwrap();
        let doc = doc_with_namespace(Some("cellos-prod"));
        with_k8s_namespace(Some("cellos-prod"), || {
            super::enforce_kubernetes_namespace_placement(&doc)
                .expect("matching namespace must pass");
        });
    }

    #[test]
    fn k8s_namespace_set_rejects_mismatch() {
        let _guard = ENV_MUTEX.lock().unwrap();
        let doc = doc_with_namespace(Some("cellos-staging"));
        let err = with_k8s_namespace(Some("cellos-prod"), || {
            super::enforce_kubernetes_namespace_placement(&doc)
                .expect_err("mismatching namespace must fail")
        });
        let msg = err.to_string();
        assert!(
            msg.contains("cellos-staging") && msg.contains("cellos-prod"),
            "error must mention both namespaces; got: {msg}"
        );
    }

    #[test]
    fn k8s_namespace_set_accepts_spec_without_constraint() {
        // Portable spec (no kubernetesNamespace) is admissible everywhere.
        let _guard = ENV_MUTEX.lock().unwrap();
        let doc = doc_with_namespace(None);
        with_k8s_namespace(Some("cellos-prod"), || {
            super::enforce_kubernetes_namespace_placement(&doc)
                .expect("portable spec must pass on any namespace");
        });
    }

    #[test]
    fn strict_config_enabled_recognises_truthy_values() {
        let _guard = ENV_MUTEX.lock().unwrap();
        for v in ["1", "true", "TRUE", "Yes", "on"] {
            let got = with_strict_config(Some(v), strict_config_enabled);
            assert!(got, "value {v:?} should be recognised as truthy");
        }
    }

    #[test]
    fn strict_config_enabled_rejects_other_values() {
        let _guard = ENV_MUTEX.lock().unwrap();
        for v in ["0", "false", "no", "off", "", "maybe"] {
            let got = with_strict_config(Some(v), strict_config_enabled);
            assert!(!got, "value {v:?} should NOT be recognised as truthy");
        }
        let got = with_strict_config(None, strict_config_enabled);
        assert!(!got, "unset must be falsy");
    }

    #[test]
    fn warnings_finalize_ok_when_empty_regardless_of_strict() {
        let _guard = ENV_MUTEX.lock().unwrap();
        with_strict_config(Some("1"), || {
            let warnings = StartupConfigWarnings::new();
            warnings
                .finalize()
                .expect("empty warnings list must always succeed");
        });
    }

    #[test]
    fn warnings_finalize_ok_when_non_empty_and_strict_unset() {
        let _guard = ENV_MUTEX.lock().unwrap();
        with_strict_config(None, || {
            let mut warnings = StartupConfigWarnings::new();
            warnings.record(
                "CELLOS_CELL_BACKEND",
                "firecraker",
                "value didn't match enum",
            );
            warnings
                .finalize()
                .expect("non-empty warnings + strict unset must succeed (WARN-then-continue)");
        });
    }

    #[test]
    fn warnings_finalize_errors_when_non_empty_and_strict_enabled() {
        let _guard = ENV_MUTEX.lock().unwrap();
        with_strict_config(Some("1"), || {
            let mut warnings = StartupConfigWarnings::new();
            warnings.record(
                "CELLOS_CELL_BACKEND",
                "firecraker",
                "value didn't match enum {firecracker, gvisor, stub}",
            );
            warnings.record(
                "CELLOS_BROKER",
                "valt-approle",
                "value didn't match enum {env, file, github-oidc, vault-approle}",
            );
            let err = warnings
                .finalize()
                .expect_err("non-empty + strict must fail loud");
            let msg = err.to_string();
            assert!(
                msg.contains("CELLOS_STRICT_CONFIG"),
                "error must reference the strict flag; got: {msg}"
            );
            assert!(
                msg.contains("CELLOS_CELL_BACKEND") && msg.contains("CELLOS_BROKER"),
                "error must list every recorded warning; got: {msg}"
            );
        });
    }
}