cellos-supervisor 0.6.0-pre

//! Composition root: wires host backend, secret broker, and event sinks from environment,
//! then hands control to [`Supervisor::run`].

mod command_runner;
mod composition;
#[allow(dead_code)]
mod destruction_evidence;
mod event_signing;
// `dns_proxy` is exercised by `lib.rs` consumers (tests + the dataplane
// module surface) and re-declared here so the binary build participates in
// the same module tree. The supervisor binary itself does not call
// `dns_proxy::run_one_shot` directly today — Phase 1 wires the activation
// predicate via `Supervisor::maybe_emit_dns_proxy_status` and exercises the
// proxy module end-to-end through the integration test under `tests/`.
// The `setns(2)` spawn that puts a real proxy thread inside the cell's
// network namespace is the runtime-integration follow-up slice.
#[allow(dead_code)]
mod dns_proxy;
// FC-38 Phase 2 — eBPF/nflog real-time per-flow listener (scaffolding).
// The binary does not call any of this module's surface today (Phase 2
// implementation is a future slice); the scaffold is reachable through
// `lib.rs` for unit tests. Mirrors the `dns_proxy` / `sni_proxy` pattern
// of feature-flagged modules that need to participate in the binary's
// module tree without contributing to the hot path.
mod ebpf_flow;
#[allow(dead_code)]
mod linux_cgroup;
mod linux_isolation;
#[cfg(target_os = "linux")]
mod linux_mount;
#[cfg(target_os = "linux")]
mod linux_net;
#[cfg(target_os = "linux")]
mod linux_seccomp;
mod network_policy;
mod nft_counters;
// Real-time per-flow events (nflog backend). Wired into
// `supervisor.rs::linux_run_cell_command_isolated` alongside dns_proxy and
// sni_proxy. Like those modules, the binary build only directly references
// a subset of the public surface; the rest is exercised through lib.rs +
// integration tests under `tests/`.
mod per_flow;
mod proxy_activation;
mod reconcile;
// SEC-21 Phase 3h.1 — `dns_proxy::dnssec` reuses `TrustAnchors` and
// `resolve_with_ttl_validated` from `resolver_refresh` (re-export, not
// copy), so the binary must include `resolver_refresh` in its module
// tree just like `lib.rs` does — without this, the binary build fails
// on `crate::resolver_refresh` paths in `dns_proxy/dnssec.rs`. The
// `#[allow(dead_code)]` mirrors the `dns_proxy` / `sni_proxy` pattern
// above: most of the surface is exercised through the lib by
// integration tests, not from `main.rs` directly.
#[allow(dead_code)]
mod resolver_refresh;
mod runtime_secret;
#[allow(dead_code)]
mod sni_proxy;
mod spec_input;
mod supervisor;
mod supervisor_helpers;
mod trust_keyset_load;
mod trust_plane_observability;

use std::ffi::OsString;
use std::path::PathBuf;

use anyhow::Context;
use cellos_core::{
    admission_decision_data_v1, enforce_derivation_scope_policy, validate_execution_cell_document,
    verify_authority_derivation, AdmissionDecisionReason, AuthorityCapability, EnforcementBinding,
    ExecutionCellDocument, ExecutionCellSpec, ADMISSION_ALLOWED_TYPE, ADMISSION_REJECTED_TYPE,
};
use composition::{
    build_admission_receipt_signer, build_supervisor, emit_startup_banner,
    enforce_authority_derivation_requirement, enforce_deployment_profile, enforce_host_baseline,
    enforce_isolation_default, enforce_kubernetes_namespace_placement, enforce_telemetry_declared,
    load_authority_keys, resolve_deployment_mode,
};
use event_signing::StandaloneEventSigner;
use spec_input::{read_cell_spec, spec_sha256};
use supervisor_helpers::cloud_event;
use trust_keyset_load::{
    load_and_verify_ceiling_from_env, load_trust_verify_keys_from_env, CeilingLoadOutcome,
    CeilingVerifiedDetails,
};

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // HIGH-B5 (red-team wave-1, fixed wave-2.5): redacted filter on the fmt
    // layer suppresses reqwest/hyper TRACE events whose message bodies
    // include bearer tokens (the supervisor uses reqwest for OIDC token
    // fetch and Vault unwrap — both authorization-bearing flows).
    //
    // Output still goes to stderr — stdout stays clean for any structured
    // operator output a future subcommand prints. The I4 integration test
    // (`supervisor_i4_hardened_default`) asserts the banner lands on stderr.
    use tracing_subscriber::layer::SubscriberExt;
    use tracing_subscriber::util::SubscriberInitExt;
    use tracing_subscriber::Layer;

    let fmt_layer = tracing_subscriber::fmt::layer()
        .with_writer(std::io::stderr)
        .with_filter(cellos_core::observability::redacted_filter());

    tracing_subscriber::registry()
        .with(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
        )
        .with(fmt_layer)
        .init();

    // ── S09 (ADR-0027): power-on self-test of the active crypto provider BEFORE
    //    any signing path runs. A broken module is unconditionally fatal — the
    //    supervisor signs admission receipts and must not do so with an unproven
    //    provider. Under `--features fips` this exercises the aws-lc-fips module;
    //    on the default build, dalek.
    cellos_core::crypto::power_on_self_test()
        .map_err(|e| anyhow::anyhow!("crypto power-on self-test failed (fail-stop): {e}"))?;

    // ── Deployment profile: collapse three independent hardening opt-ins
    //    (CELLOS_AUTHORITY_KEYS_PATH, CELL_OS_REQUIRE_JETSTREAM,
    //    CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS) into a single
    //    CELLOS_DEPLOYMENT_PROFILE=hardened decision. Fails fast if the
    //    profile's preconditions aren't met. Must run before any code that
    //    reads those env vars (broker selection, derivation scope policy).
    enforce_deployment_profile()?;

    // ── I4: hardened-vs-portable startup banner + CELLOS_REQUIRE_ISOLATION=1
    //    as the 1.0 default. Refuse to start without unshare/seccomp/cgroup
    //    support unless CELLOS_DEPLOYMENT_PROFILE=portable is named explicitly.
    let deployment_mode = resolve_deployment_mode()?;
    enforce_isolation_default(deployment_mode)?;
    emit_startup_banner(deployment_mode);

    // ── C01 (S52, ADR-0033): probe the full host baseline (the three core
    //    isolation primitives plus nftables + kvm) and, under an isolation-
    //    enforcing profile with CELLOS_REQUIRE_HOST_BASELINE set, fail-stop on
    //    any missing control. Complements `enforce_isolation_default` (which
    //    already hard-gates the three core primitives under hardened) by adding
    //    the full five-control set and the offline-verifiable signed receipt
    //    emitted below once the standalone signer exists. Runs here so a missing
    //    control aborts before any spec parsing / run-loop wiring; the
    //    attestation is carried to the signed-receipt emit on the run path.
    let host_baseline = enforce_host_baseline(deployment_mode)?;

    let args: Vec<OsString> = std::env::args_os().skip(1).collect();
    if let Some(subcommand) = args.first().and_then(|s| s.to_str()) {
        match subcommand {
            "secret" => return runtime_secret::cli_main(&args[1..]).await,
            "--version" | "-V" => {
                println!("cellos-supervisor {}", env!("CARGO_PKG_VERSION"));
                return Ok(());
            }
            _ => {}
        }
    }

    // I4 — `--validate <spec>` runs spec parse + structural validation +
    // hardened/portable admission gates and exits 0 without entering the
    // supervisor run loop. Used by operators to lint a spec offline and by
    // CI to confirm a host's deployment-profile posture without launching a
    // workload.
    let validate_only = args
        .first()
        .and_then(|s| s.to_str())
        .map(|s| s == "--validate")
        .unwrap_or(false);
    let spec_arg_index = if validate_only { 1 } else { 0 };

    // ── Parse spec ────────────────────────────────────────────────────────
    let spec_path = args
        .get(spec_arg_index)
        .map(PathBuf::from)
        .unwrap_or_else(|| PathBuf::from("contracts/examples/execution-cell-minimal.valid.json"));

    let raw = read_cell_spec(&spec_path)?;
    let raw_spec_hash = spec_sha256(&raw);
    let doc: ExecutionCellDocument =
        serde_json::from_str(&raw).context("parse ExecutionCellDocument")?;
    validate_execution_cell_document(&doc).map_err(|e| anyhow::anyhow!("{e}"))?;

    // ── SEAM-3: when CELLOS_REQUIRE_AUTHORITY_DERIVATION is set, refuse to
    //    admit specs that declare egress or secrets without a derivation
    //    token. Runs after spec validation (which has no env access) and
    //    before signature verification, so the operator sees a clear
    //    "missing token" error instead of the spec slipping through into
    //    `build_supervisor`. Auto-enabled by `CELLOS_DEPLOYMENT_PROFILE=hardened`.
    enforce_authority_derivation_requirement(&doc)?;

    // ── F4a: when CELLOS_REQUIRE_TELEMETRY_DECLARED is set, refuse to admit
    //    specs that omit the `spec.telemetry` block. Auto-enabled by
    //    `CELLOS_DEPLOYMENT_PROFILE=hardened`. Runs immediately after the
    //    authority-derivation gate so operators see telemetry-intent
    //    enforcement on the same admission boundary.
    enforce_telemetry_declared(&doc)?;

    // ── T11-5: enforce Kubernetes-namespace placement. When
    //    `CELLOS_K8S_NAMESPACE` is set, any spec declaring
    //    `spec.placement.kubernetesNamespace` must match the runner's
    //    configured namespace exactly. Specs without that field still pass
    //    (a portable spec is admissible everywhere); unset env var disables
    //    the gate entirely.
    enforce_kubernetes_namespace_placement(&doc)?;

    // I4 — `--validate` exits here. The supervisor has now confirmed the
    // host's isolation posture (banner emitted), parsed + structurally
    // validated the spec, and applied the hardened/portable admission
    // gates. Anything past this point is run-loop wiring (build_supervisor
    // / supervisor.run), which is not what `--validate` should exercise.
    if validate_only {
        let validate_run_id = std::env::var("CELLOS_RUN_ID").unwrap_or_else(|_| "validate".into());
        let summary = composition::build_validation_summary(
            &doc,
            &validate_run_id,
            &raw_spec_hash,
            &spec_path,
        );
        println!(
            "{}",
            serde_json::to_string_pretty(&summary).unwrap_or_default()
        );
        return Ok(());
    }

    // ── Authority derivation: verify the grantor's ED25519 signature ──────
    // when a derivation token is attached. Keys come from
    // `CELLOS_AUTHORITY_KEYS_PATH` (operator-managed). Specs without a token
    // pass unchanged.
    let authority_keys = load_authority_keys()?;
    // I6 / O6: parentRunId scope policy. The 1.0 default is STRICT — a
    // token with `parentRunId: null` is rejected. Operators that need the
    // legacy permissive behaviour MUST opt out explicitly by setting
    // `CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS=0` (or `false`/`no`/`off`).
    // Permissive accepts emit a structured CloudEvent in `Supervisor::run`.
    let allow_universal = scoped_tokens_permissive_mode_enabled();
    let mut universal_token_accepted_in_permissive_mode = false;
    if let Some(token) = doc.spec.authority.authority_derivation.as_ref() {
        verify_authority_derivation(&doc.spec, token, &authority_keys)
            .map_err(|e| anyhow::anyhow!("{e}"))?;
        enforce_derivation_scope_policy(token, allow_universal)
            .map_err(|e| anyhow::anyhow!("{e}"))?;
        if allow_universal && token.parent_run_id.is_none() {
            universal_token_accepted_in_permissive_mode = true;
        }
    }

    // Lane B: hand the same loaded key map to the supervisor so the
    // `derivationVerified` field on `lifecycle.started` reflects the real
    // ED25519 verification result, not a structural-only stub. `Arc` so the
    // map is shared cheaply across the supervisor's lifetime.
    let authority_keys = std::sync::Arc::new(authority_keys);

    let run_id = std::env::var("CELLOS_RUN_ID").unwrap_or_else(|_| "run-local-001".into());

    // ── ADR-0025 §2: standalone admission-receipt signer ─────────────────
    // RESOLVES THE WIRE-BINDING CONTRADICTION. The ceiling gate (ADR-0024 §3)
    // is hooked here, BEFORE `build_supervisor` (below), so the `Supervisor`
    // and its composed `DlqSink(SigningEventSink(RedactingEventSink))` do not
    // yet exist — a receipt cannot reuse `Supervisor::emit`. This standalone
    // signer is constructed from the same `CELLOS_EVENT_SIGNING` config and
    // its own raw transport target, redacts before signing, and emits an
    // explicit `signed: false` record when signing is unconfigured. The
    // ceiling allow/reject receipts (below, T07) fire through
    // `admission_signer.emit_decision(..)` on this live, pre-build path.
    let admission_signer = build_admission_receipt_signer(&doc, &run_id).await?;

    // ── C01 (S52): emit the host-baseline attestation as a signed, offline-
    //    verifiable startup receipt through the same standalone signer used for
    //    admission receipts (redact-before-sign, same transport). The probe +
    //    fail-stop already ran at startup (above); this records the attested
    //    host posture under the org-root signing key so an auditor can verify it
    //    offline. Emission is advisory: a transient transport error must not
    //    block admission, mirroring the admission-receipt resilience posture.
    emit_host_baseline_receipt(&host_baseline, &admission_signer).await;

    // ── ADR-0024 §3: org-root authority-ceiling gate ─────────────────────
    // The fail-closed single-cell admission bound. Resolves the trust-root-
    // signed ceiling manifest (via the same SEC-25 envelope verifier and the
    // org-root keyring loaded from `CELLOS_TRUST_VERIFY_KEYS_PATH`) and rejects
    // any cell whose `spec.authority` is not a subset of a manifest ceiling.
    // Sits AFTER the derivation-scope block and BEFORE `build_supervisor`, so a
    // missing/invalid ceiling or an over-broad spec aborts admission before any
    // host wiring exists. The reject receipt is emitted through the standalone
    // signer BEFORE the `?` propagates the verdict error (receipt-before-Err).
    enforce_authority_ceiling(&doc.spec, &raw_spec_hash, &run_id, &admission_signer).await?;

    // ── Build runtime wiring and run ─────────────────────────────────────
    let mut supervisor = build_supervisor(&doc, &run_id, authority_keys).await?;

    // ── C03 (S17/S18): under the airgapped profile, reconcile any rows the
    //    durable spool accumulated while disconnected into the broker and emit
    //    the signed reconcile receipt — through the supervisor's already-composed
    //    primary sink, so there is a single writer to the spool. Off the
    //    admission path: a still-unreachable broker is a soft skip; only a
    //    corrupt durable chain is fatal (airgapped fail-closed).
    if deployment_mode.is_airgapped() {
        reconcile::run_startup_reconcile(&doc, &run_id, &supervisor.event_sink).await?;
    }
    // I6: hand the supervisor the permissive-mode acceptance flag so it can
    // emit a single structured warning CloudEvent during `run()`.
    supervisor.universal_token_accepted_in_permissive_mode =
        universal_token_accepted_in_permissive_mode;
    supervisor.run(&doc, &run_id, Some(&raw_spec_hash)).await
}

/// C01 (S52): emit the probed host-baseline attestation as a signed,
/// offline-verifiable startup receipt through the standalone admission signer
/// (redact-before-sign, same transport). Advisory by design — a build or
/// transport failure is logged but never blocks startup, matching the
/// admission-receipt resilience posture. When signing is unconfigured the
/// standalone signer still emits the advisory `signed: false` record, so the
/// attested posture is never silently dropped.
async fn emit_host_baseline_receipt(
    attestation: &cellos_core::HostBaselineAttestationV1,
    signer: &StandaloneEventSigner,
) {
    let event = match cellos_core::host_baseline_event(attestation) {
        Ok(event) => event,
        Err(e) => {
            tracing::warn!(
                target: "cellos.supervisor.baseline",
                error = %e,
                "host-baseline receipt: building the event failed (advisory — startup continues)"
            );
            return;
        }
    };
    if let Err(e) = signer.emit_decision(&event).await {
        tracing::warn!(
            target: "cellos.supervisor.baseline",
            error = %e,
            "host-baseline receipt: emit failed (advisory — startup continues)"
        );
    }
}

/// I6 / O6: parse `CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS` with a
/// **strict-by-default** policy.
///
/// Returns `true` iff the operator EXPLICITLY opted out of strict mode by
/// setting the flag to one of: `0`, `false`, `no`, `off`. Any other value
/// (including unset, empty, `1`, `true`, `yes`, `on`) keeps the strict
/// default — universal (`parentRunId: null`) tokens are rejected.
fn scoped_tokens_permissive_mode_enabled() -> bool {
    match std::env::var("CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS") {
        Ok(raw) => {
            let t = raw.trim().to_ascii_lowercase();
            matches!(t.as_str(), "0" | "false" | "no" | "off")
        }
        Err(_) => false, // unset → strict (1.0 default)
    }
}

// ── ADR-0024 §3: org-root authority-ceiling admission gate ──────────────────

/// Plain-string predicate recorded on every ceiling receipt. This is the
/// allowlist subset statement — **never** an environment label (`prod`/staging)
/// — so an auditor reading the receipt sees exactly what was checked
/// (ADR-0024 §Verdict legibility, Decision 2).
const CEILING_PREDICATE: &str = "ceiling.is_superset_of(spec.authority)";

/// `true` when the org-root authority ceiling is **required** — i.e. when
/// `CELLOS_REQUIRE_AUTHORITY_CEILING` is set to a truthy value OR the hardened
/// deployment profile is in effect (the require-var is auto-implied by
/// hardened). Mirrors the `trust_verify_keys_required` truthy parse and the
/// `resolve_deployment_mode` profile reading.
///
/// When this returns `false` (unset env + not hardened), a missing ceiling is
/// the single labeled fail-OPEN path (a signed `ceiling_degraded_unconfigured`
/// receipt then proceed); when `true`, a missing/invalid ceiling aborts
/// admission (ADR-0024 §Fail-closed posture).
fn authority_ceiling_required() -> bool {
    let explicit = std::env::var("CELLOS_REQUIRE_AUTHORITY_CEILING")
        .map(|v| {
            let t = v.trim().to_ascii_lowercase();
            matches!(t.as_str(), "1" | "true" | "yes" | "on")
        })
        .unwrap_or(false);
    explicit || deployment_profile_is_hardened()
}

/// `true` when `CELLOS_DEPLOYMENT_PROFILE` is unset/empty or `hardened` (the 1.0
/// default), mirroring `composition::resolve_deployment_mode`. `portable` and
/// any other value are not hardened.
fn deployment_profile_is_hardened() -> bool {
    let raw = std::env::var("CELLOS_DEPLOYMENT_PROFILE").unwrap_or_default();
    let t = raw.trim().to_ascii_lowercase();
    t.is_empty() || t == "hardened"
}

/// Where an admitted ceiling allow is actually bound. The subset **decision** is
/// platform-independent, but egress only binds in the kernel on Linux; on
/// Windows/portable hosts (ADR-0023) it is declare-and-audit. Recorded on the
/// receipt so a verifier on either OS knows whether an allow is kernel-enforced
/// (ADR-0024 §Linux-vs-Windows boundary).
fn ceiling_enforcement_binding() -> EnforcementBinding {
    if cfg!(target_os = "linux") {
        EnforcementBinding::LinuxNftables
    } else {
        EnforcementBinding::DeclaredAudit
    }
}

/// The four authority dimensions the ceiling **cannot** compare today
/// (`is_superset_of` covers only `egress_rules` + `secret_refs`). A cell that
/// populates any of them would otherwise pass the subset test vacuously, so the
/// gate detects population via `is_some()` and — under hardened — rejects
/// `ceiling_dimension_uncomparable` rather than silently admitting an uncompared
/// dimension (ADR-0024 §Advisory labelling). Returns the names of the populated
/// uncomparable dimensions, empty when none.
fn populated_uncomparable_dimensions(spec: &ExecutionCellSpec) -> Vec<&'static str> {
    let a = &spec.authority;
    let mut populated = Vec::new();
    if a.filesystem.is_some() {
        populated.push("filesystem");
    }
    if a.network.is_some() {
        populated.push("network");
    }
    if a.dns_authority.is_some() {
        populated.push("dns");
    }
    if a.cdn_authority.is_some() {
        populated.push("cdn");
    }
    populated
}

/// ADR-0024 §3 — fail-closed org-root authority-ceiling gate on the live
/// single-cell admission path.
///
/// Resolves the trust-root-signed ceiling manifest (`CELLOS_AUTHORITY_CEILING_PATH`,
/// verified by the SEC-25 envelope verifier against the org-root keyring from
/// `CELLOS_TRUST_VERIFY_KEYS_PATH`) and evaluates the **allowlist** predicate
/// `ceiling.is_superset_of(AuthorityCapability::from_bundle(&spec.authority))`
/// against each named ceiling. The predicate is the only admission gate — no
/// environment label is ever consulted (ADR-0024 Decision 2).
///
/// Every decision emits a signed admission-decision receipt (ADR-0025) through
/// `signer` BEFORE the verdict's `Err` is propagated, so a reject's receipt is
/// on the live transport even though the early-return aborts admission. Outcomes:
///
/// - **Manifest unset + not required** (`CeilingLoadOutcome::NotConfigured`) →
///   single signed `ceiling_degraded_unconfigured` receipt, then proceed. The
///   only fail-open path; reachable only when the gate is not required.
/// - **Manifest unset/invalid + required** → `load_and_verify_ceiling_from_env`
///   returns `Err` and admission aborts before this function emits (no host
///   wiring exists yet).
/// - **Populated uncomparable dimension** (filesystem/network/dns/cdn) → reject
///   `ceiling_dimension_uncomparable` under hardened; loud advisory warn and
///   continue to the subset check otherwise.
/// - **`spec.authority` ⊄ any ceiling** → reject `authority_exceeds_ceiling`.
/// - **`spec.authority` ⊆ some ceiling** → admit `admitted_within_ceiling`.
async fn enforce_authority_ceiling(
    spec: &ExecutionCellSpec,
    spec_hash: &str,
    run_id: &str,
    signer: &StandaloneEventSigner,
) -> anyhow::Result<()> {
    let required = authority_ceiling_required();
    let hardened = deployment_profile_is_hardened();
    let binding = ceiling_enforcement_binding();

    // The org-root keyring is loaded under the SAME require semantics as the
    // trust-keyset path: a configured-but-unloadable keys file is an operator
    // bug and fails closed regardless of the require flag (the loader enforces
    // that). `required` only governs the unset case.
    let keys = load_trust_verify_keys_from_env(required)?;

    match load_and_verify_ceiling_from_env(keys.as_ref(), required, std::time::SystemTime::now())? {
        CeilingLoadOutcome::NotConfigured => {
            // Labeled fail-OPEN path: no ceiling, not required. Emit a single
            // signed degraded receipt (never silent) and proceed. `admitted` is
            // true because the cell is admitted — but the reason names the
            // degraded posture so an auditor can see the ceiling did not bind.
            emit_ceiling_receipt(
                signer,
                ADMISSION_ALLOWED_TYPE,
                spec,
                true,
                AdmissionDecisionReason::CeilingDegradedUnconfigured,
                spec_hash,
                run_id,
                binding,
            )
            .await?;
            tracing::warn!(
                target: "cellos.supervisor.ceiling",
                receipt_signed = signer.is_signing_enabled(),
                "authority ceiling not configured and not required; admission proceeds via the \
                 labeled degraded fail-open path (ceiling_degraded_unconfigured receipt emitted; \
                 receipt is offline-verifiable only when CELLOS_EVENT_SIGNING is configured)"
            );
            Ok(())
        }
        CeilingLoadOutcome::Verified(details) => {
            enforce_verified_ceiling(spec, spec_hash, run_id, signer, &details, hardened, binding)
                .await
        }
    }
}

/// Evaluate the allowlist subset predicate against a verified manifest and emit
/// the receipt-before-Err decision. Split out so the uncomparable-dimension
/// short-circuit and the subset verdict share one receipt-emitting tail.
#[allow(clippy::too_many_arguments)]
async fn enforce_verified_ceiling(
    spec: &ExecutionCellSpec,
    spec_hash: &str,
    run_id: &str,
    signer: &StandaloneEventSigner,
    details: &CeilingVerifiedDetails,
    hardened: bool,
    binding: EnforcementBinding,
) -> anyhow::Result<()> {
    // Uncomparable dimensions first: a populated filesystem/network/dns/cdn
    // dimension would pass the subset test vacuously, so default-closed on the
    // unknown under hardened, loud advisory warn otherwise (ADR-0024 §Advisory).
    let uncomparable = populated_uncomparable_dimensions(spec);
    if !uncomparable.is_empty() {
        if hardened {
            emit_ceiling_receipt(
                signer,
                ADMISSION_REJECTED_TYPE,
                spec,
                false,
                AdmissionDecisionReason::CeilingDimensionUncomparable,
                spec_hash,
                run_id,
                binding,
            )
            .await?;
            return Err(anyhow::anyhow!(
                "authority ceiling: spec '{}' populates uncomparable authority dimension(s) {:?} \
                 the ceiling cannot bound; rejected ceiling_dimension_uncomparable under hardened \
                 (ceilingId {})",
                spec.id,
                uncomparable,
                details.manifest.ceiling_id,
            ));
        }
        tracing::warn!(
            target: "cellos.supervisor.ceiling",
            spec_id = %spec.id,
            dimensions = ?uncomparable,
            "spec populates authority dimension(s) the ceiling cannot compare; the ceiling does \
             NOT bound them (advisory — set CELLOS_DEPLOYMENT_PROFILE=hardened to reject)"
        );
    }

    // The allowlist predicate: admit iff the spec's comparable authority is a
    // subset of SOME named ceiling in the manifest. An empty `ceilings` list is
    // deny-all (the empty grant is a subset only of the empty claim).
    let claim = AuthorityCapability::from_bundle(&spec.authority);
    let admitted = details
        .manifest
        .ceilings
        .iter()
        .any(|named| named.ceiling.is_superset_of(&claim));

    if admitted {
        emit_ceiling_receipt(
            signer,
            ADMISSION_ALLOWED_TYPE,
            spec,
            true,
            AdmissionDecisionReason::AdmittedWithinCeiling,
            spec_hash,
            run_id,
            binding,
        )
        .await?;
        tracing::info!(
            target: "cellos.supervisor.ceiling",
            spec_id = %spec.id,
            ceiling_id = %details.manifest.ceiling_id,
            manifest_digest = %details.manifest_digest,
            "spec admitted within org-root authority ceiling"
        );
        return Ok(());
    }

    // Reject: emit the signed receipt BEFORE the Err propagates so the verdict
    // and its receipt share a single fate on the live transport.
    emit_ceiling_receipt(
        signer,
        ADMISSION_REJECTED_TYPE,
        spec,
        false,
        AdmissionDecisionReason::AuthorityExceedsCeiling,
        spec_hash,
        run_id,
        binding,
    )
    .await?;
    Err(anyhow::anyhow!(
        "authority ceiling: spec '{}' declared authority is not a subset of any ceiling in \
         manifest '{}'; rejected authority_exceeds_ceiling",
        spec.id,
        details.manifest.ceiling_id,
    ))
}

/// Build the admission-decision `data` payload, wrap it in a CloudEvent of
/// `event_ty`, and emit it through the standalone signer. The signer redacts
/// before signing and emits an explicit `signed:false` record when signing is
/// unconfigured (never a missing receipt — ADR-0025 §2).
#[allow(clippy::too_many_arguments)]
async fn emit_ceiling_receipt(
    signer: &StandaloneEventSigner,
    event_ty: &str,
    spec: &ExecutionCellSpec,
    admitted: bool,
    reason: AdmissionDecisionReason,
    spec_hash: &str,
    run_id: &str,
    binding: EnforcementBinding,
) -> anyhow::Result<()> {
    let data = admission_decision_data_v1(
        spec,
        admitted,
        reason,
        CEILING_PREDICATE,
        spec_hash,
        Some(run_id),
        binding,
        &[],
    )
    .context("build admission-decision receipt payload")?;
    let event = cloud_event(event_ty, data);
    signer
        .emit_decision(&event)
        .await
        .map_err(|e| anyhow::anyhow!("emit admission-decision receipt: {e}"))?;
    // Flush the transport so the receipt is on the wire before the gate either
    // aborts admission (reject) or proceeds. A buffering NATS publish would
    // otherwise be dropped when the reject path exits the process immediately —
    // ADR-0025: the decision and its receipt share a single fate on the wire.
    signer
        .flush()
        .await
        .map_err(|e| anyhow::anyhow!("flush admission-decision receipt: {e}"))?;
    Ok(())
}

#[cfg(test)]
mod ceiling_gate_tests {
    use super::*;
    use async_trait::async_trait;
    use cellos_core::ports::EventSink;
    use cellos_core::types::{NamedAuthorityCeiling, OrgAuthorityCeilingManifestV1};
    use cellos_core::{AuthorityBundle, CellosError, CloudEventV1, EgressRule};
    use std::sync::{Arc, Mutex};

    /// Capture sink — records every event emitted so a test can read back the
    /// receipt the gate produced and assert its `type`/`reason`.
    struct CaptureSink(Mutex<Vec<CloudEventV1>>);

    impl CaptureSink {
        fn new() -> Arc<Self> {
            Arc::new(Self(Mutex::new(Vec::new())))
        }
        fn events(&self) -> Vec<CloudEventV1> {
            self.0.lock().unwrap().clone()
        }
    }

    #[async_trait]
    impl EventSink for CaptureSink {
        async fn emit(&self, event: &CloudEventV1) -> Result<(), CellosError> {
            self.0.lock().unwrap().push(event.clone());
            Ok(())
        }
    }

    /// Build an unsigned standalone signer over a capture sink. Unsigned mode
    /// still emits an advisory `signed:false` record carrying the decision
    /// `data`, which is all these ordering/reason assertions need (the signed
    /// path is covered by `event_signing` unit tests + the T17 black-box test).
    fn unsigned_signer(capture: Arc<CaptureSink>) -> StandaloneEventSigner {
        let _g = ENV_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
        std::env::remove_var("CELLOS_EVENT_SIGNING");
        std::env::remove_var("CELLOS_REDACT_EVENT_FIELDS");
        StandaloneEventSigner::from_env(capture as Arc<dyn EventSink>).0
    }

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

    fn egress(host: &str, port: u16) -> EgressRule {
        EgressRule {
            host: host.into(),
            port,
            protocol: Some("tcp".into()),
            dns_egress_justification: None,
        }
    }

    fn spec_with_authority(authority: AuthorityBundle) -> ExecutionCellSpec {
        ExecutionCellSpec {
            id: "spec-under-test".into(),
            authority,
            ..Default::default()
        }
    }

    fn ceiling_details(ceiling: AuthorityCapability) -> CeilingVerifiedDetails {
        CeilingVerifiedDetails {
            manifest: OrgAuthorityCeilingManifestV1 {
                schema_version: "1.0.0".into(),
                ceiling_id: "ceiling-test".into(),
                ceiling_epoch: 1,
                ceilings: vec![NamedAuthorityCeiling {
                    name: "default".into(),
                    ceiling,
                }],
                not_before: None,
                not_after: None,
            },
            manifest_digest: "sha256:deadbeef".into(),
            verified_signer_kid: "org-root".into(),
        }
    }

    /// The reason discriminant on the single emitted receipt's `data`.
    fn emitted_reason(capture: &CaptureSink) -> String {
        let events = capture.events();
        assert_eq!(events.len(), 1, "exactly one receipt expected");
        events[0]
            .data
            .as_ref()
            .and_then(|d| d.get("reason"))
            .and_then(|v| v.as_str())
            .expect("receipt carries a reason")
            .to_string()
    }

    /// A spec whose authority is a subset of the ceiling is admitted with
    /// reason `admitted_within_ceiling` and a single allowed receipt.
    #[tokio::test]
    async fn admitted_within_ceiling() {
        let capture = CaptureSink::new();
        let signer = unsigned_signer(capture.clone());
        let spec = spec_with_authority(AuthorityBundle {
            egress_rules: Some(vec![egress("api.internal", 443)]),
            ..Default::default()
        });
        let details = ceiling_details(AuthorityCapability {
            egress_rules: vec![egress("api.internal", 443)],
            secret_refs: vec![],
        });

        enforce_verified_ceiling(
            &spec,
            "sha256:abc",
            "run-1",
            &signer,
            &details,
            true,
            EnforcementBinding::LinuxNftables,
        )
        .await
        .expect("subset spec must be admitted");

        let events = capture.events();
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].ty, ADMISSION_ALLOWED_TYPE);
        assert_eq!(emitted_reason(&capture), "admitted_within_ceiling");
    }

    /// A spec whose egress exceeds the ceiling is rejected
    /// `authority_exceeds_ceiling`, and the receipt is emitted BEFORE the Err
    /// propagates (the receipt-before-early-return ordering property).
    #[tokio::test]
    async fn authority_exceeds_ceiling_rejects_after_receipt() {
        let capture = CaptureSink::new();
        let signer = unsigned_signer(capture.clone());
        let spec = spec_with_authority(AuthorityBundle {
            egress_rules: Some(vec![egress("0.0.0.0/0", 443)]),
            ..Default::default()
        });
        let details = ceiling_details(AuthorityCapability {
            egress_rules: vec![egress("api.internal", 443)],
            secret_refs: vec![],
        });

        let err = enforce_verified_ceiling(
            &spec,
            "sha256:abc",
            "run-1",
            &signer,
            &details,
            true,
            EnforcementBinding::LinuxNftables,
        )
        .await
        .expect_err("over-broad spec must be rejected");
        assert!(format!("{err}").contains("authority_exceeds_ceiling"));

        // Receipt was already on the transport at the moment the Err returned:
        // the capture sink holds exactly the reject receipt.
        assert_eq!(emitted_reason(&capture), "authority_exceeds_ceiling");
        assert_eq!(capture.events()[0].ty, ADMISSION_REJECTED_TYPE);
    }

    /// A populated uncomparable dimension under hardened is rejected
    /// `ceiling_dimension_uncomparable` (default-closed on the unknown).
    #[tokio::test]
    async fn uncomparable_dimension_rejected_under_hardened() {
        let capture = CaptureSink::new();
        let signer = unsigned_signer(capture.clone());
        let spec = spec_with_authority(AuthorityBundle {
            filesystem: Some(serde_json::json!({"mounts": ["/data"]})),
            ..Default::default()
        });
        let details = ceiling_details(AuthorityCapability::default());

        let err = enforce_verified_ceiling(
            &spec,
            "sha256:abc",
            "run-1",
            &signer,
            &details,
            true,
            EnforcementBinding::LinuxNftables,
        )
        .await
        .expect_err("populated filesystem under hardened must reject");
        assert!(format!("{err}").contains("ceiling_dimension_uncomparable"));
        assert_eq!(emitted_reason(&capture), "ceiling_dimension_uncomparable");
        assert_eq!(capture.events()[0].ty, ADMISSION_REJECTED_TYPE);
    }

    /// Outside hardened a populated uncomparable dimension is a loud advisory
    /// warn, NOT a reject: the gate falls through to the subset verdict and (for
    /// an empty comparable claim) admits.
    #[tokio::test]
    async fn uncomparable_dimension_warns_outside_hardened() {
        let capture = CaptureSink::new();
        let signer = unsigned_signer(capture.clone());
        let spec = spec_with_authority(AuthorityBundle {
            network: Some(serde_json::json!({"any": true})),
            ..Default::default()
        });
        // Empty comparable claim is a subset of an empty ceiling, so it admits.
        let details = ceiling_details(AuthorityCapability::default());

        enforce_verified_ceiling(
            &spec,
            "sha256:abc",
            "run-1",
            &signer,
            &details,
            false,
            EnforcementBinding::DeclaredAudit,
        )
        .await
        .expect("outside hardened, uncomparable dimension is advisory only");
        assert_eq!(emitted_reason(&capture), "admitted_within_ceiling");
    }

    /// An empty `ceilings` list is deny-all: a non-empty comparable claim has no
    /// ceiling to be a subset of, so it is rejected `authority_exceeds_ceiling`.
    #[tokio::test]
    async fn empty_ceiling_list_is_deny_all() {
        let capture = CaptureSink::new();
        let signer = unsigned_signer(capture.clone());
        let spec = spec_with_authority(AuthorityBundle {
            secret_refs: Some(vec!["prod-db".into()]),
            ..Default::default()
        });
        let details = CeilingVerifiedDetails {
            manifest: OrgAuthorityCeilingManifestV1 {
                schema_version: "1.0.0".into(),
                ceiling_id: "deny-all".into(),
                ceiling_epoch: 1,
                ceilings: vec![],
                not_before: None,
                not_after: None,
            },
            manifest_digest: "sha256:00".into(),
            verified_signer_kid: "org-root".into(),
        };

        let err = enforce_verified_ceiling(
            &spec,
            "sha256:abc",
            "run-1",
            &signer,
            &details,
            true,
            EnforcementBinding::LinuxNftables,
        )
        .await
        .expect_err("deny-all ceiling must reject a non-empty claim");
        assert!(format!("{err}").contains("authority_exceeds_ceiling"));
        assert_eq!(emitted_reason(&capture), "authority_exceeds_ceiling");
    }

    /// The labeled degraded fail-open path emits an `allowed`-typed receipt
    /// carrying reason `ceiling_degraded_unconfigured` — the receipt the
    /// `NotConfigured` arm of `enforce_authority_ceiling` produces before it
    /// proceeds. Exercised through the same `emit_ceiling_receipt` tail the gate
    /// uses, so the discriminant and event type stay in lockstep without
    /// mutating env across an await point.
    #[tokio::test]
    async fn degraded_unconfigured_receipt_shape() {
        let capture = CaptureSink::new();
        let signer = unsigned_signer(capture.clone());
        let spec = spec_with_authority(AuthorityBundle::default());

        emit_ceiling_receipt(
            &signer,
            ADMISSION_ALLOWED_TYPE,
            &spec,
            true,
            AdmissionDecisionReason::CeilingDegradedUnconfigured,
            "sha256:abc",
            "run-1",
            EnforcementBinding::DeclaredAudit,
        )
        .await
        .expect("degraded receipt emits");

        assert_eq!(emitted_reason(&capture), "ceiling_degraded_unconfigured");
        let events = capture.events();
        assert_eq!(events[0].ty, ADMISSION_ALLOWED_TYPE);
        assert_eq!(
            events[0].data.as_ref().and_then(|d| d.get("admitted")),
            Some(&serde_json::Value::Bool(true)),
            "degraded path admits (proceeds)"
        );
    }
}