cellos-supervisor 0.5.1

//! [`Supervisor`] — lifecycle orchestration extracted from the composition root.
//!
//! `main.rs` wires the adapters; `Supervisor::run` owns the lifecycle sequence:
//! network_scope → trust_plane_observability (optional) → secrets → started → run → export → destroy → revoke → destroyed.
//!
//! Teardown semantics: `destroy` and `revoke_for_cell` are called unconditionally
//! even when a phase error has already been captured.

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

use anyhow::Context;
use cellos_core::ports::{CellBackend, EventSink, ExportSink, SecretBroker};
use cellos_core::types::SecretDeliveryMode;
use cellos_core::{
    authz_rejected_data_v1, command_completed_data_v1, compliance_summary_data_v1,
    export_completed_data_v2, export_failed_data_v2, identity_materialized_data_v1,
    identity_revoked_data_v1, lifecycle_destroyed_data_v1, lifecycle_started_data_v1,
    observability_fs_touch_export_data_v1, observability_network_enforcement_data_v1,
    observability_network_scope_data_v1, observability_process_spawned_data_v1,
    policy_rejected_data_v1, validate_spec_against_policy, verify_authority_derivation,
    AuthorityDerivationToken, AuthorizationPolicy, CloudEventV1, Correlation, EgressRule,
    ExecutionCellDocument, ExecutionCellSpec, ExportArtifactMetadata, IdentityFailureOperation,
    LifecycleDestroyOutcome, LifecycleTerminalState, PolicyPackSpec, SecretView,
};
#[cfg(target_os = "linux")]
use cellos_core::{observability_container_security_data_v1, observability_network_policy_data_v1};

#[cfg(target_os = "linux")]
use crate::network_policy::read_proc_caps;
use crate::runtime_secret::RuntimeSecretSession;
use crate::supervisor_helpers::{
    argv0_allow_prefixes_error, cloud_event, configured_broker_name, effective_export_target_name,
    identity_resolve_key, logical_export_destination, resolve_export_target_kind,
    runtime_secret_entries_for_doc, runtime_secret_lease_requests_for_doc,
};

pub struct Supervisor {
    pub host: Arc<dyn CellBackend>,
    pub broker: Arc<dyn SecretBroker>,
    /// Primary event sink (JetStream or noop).
    pub event_sink: Arc<dyn EventSink>,
    /// Optional secondary JSONL sink — receives every event after the primary.
    pub jsonl_sink: Option<Arc<dyn EventSink>>,
    pub default_export_sink: Arc<dyn ExportSink>,
    pub named_export_sinks: HashMap<String, Arc<dyn ExportSink>>,
    /// Active policy pack for admission gate enforcement.  When set, every cell
    /// spec is validated before `host.create()` is called; violations produce a
    /// `policy.rejected` CloudEvent and an error return.
    pub policy_pack: Option<PolicyPackSpec>,
    /// T12 RBAC — operator-supplied authorization policy, loaded from
    /// `CELLOS_AUTHZ_POLICY_PATH` at startup. When set, every cell spec is
    /// checked against the policy's `subjects` / `allowedPools` /
    /// `allowedPolicyPacks` lists before `host.create()`; rejection produces a
    /// `cell.authz.v1.rejected` CloudEvent and a startup error. Independent
    /// of the policy-pack admission gate (`policy_pack` above) — both gates
    /// fire in series with their own rule namespaces.
    pub authz_policy: Option<AuthorizationPolicy>,
    /// Operator-supplied role → ED25519 verifying key map, sourced from
    /// `CELLOS_AUTHORITY_KEYS_PATH` and loaded once at startup. Plumbed through
    /// from `build_supervisor` so the `lifecycle.started` event's
    /// `derivationVerified` flag reflects the real cryptographic verification
    /// result (Lane B). Wrapped in `Arc` so cloning per cell run is cheap.
    /// Empty map is the legacy/default behaviour: tokens are present but no
    /// keys are configured → verification fails with `unknown role` →
    /// `derivationVerified: false`.
    pub authority_keys: Arc<HashMap<String, String>>,
    /// SEC-25 Phase 2: operator-supplied trust-keyset signer kid →
    /// `ed25519_dalek::VerifyingKey` map, sourced from
    /// `CELLOS_TRUST_VERIFY_KEYS_PATH` and loaded once at startup. Plumbed
    /// through from `build_supervisor` so the verifier
    /// (`cellos_core::verify_signed_trust_keyset_envelope`) has the keyring it
    /// needs at startup-time keyset establishment AND at any future per-cell
    /// envelope re-verification. Wrapped in `Arc` so cloning per cell run is
    /// cheap. Empty map is the legacy/default behaviour: any envelope
    /// reference under `CELLOS_TRUST_KEYSET_PATH` will surface a
    /// `keyset_verification_failed` event because no signature can verify
    /// against an empty keyring.
    pub trust_verify_keys: Arc<HashMap<String, ed25519_dalek::VerifyingKey>>,
    /// I6 / O6: tracks whether the supervisor was started in PERMISSIVE
    /// universal-token mode AND a universal token was accepted at admission
    /// time (`parentRunId: null`). When `true`, `Supervisor::run` emits a
    /// single `cell.identity.universal_token_accepted_in_permissive_mode`
    /// CloudEvent so the audit trail records the exposure.
    pub universal_token_accepted_in_permissive_mode: bool,
}

// I5 per-event signing has moved out of this lifecycle module and into the
// composition root: see `crate::event_signing::SigningEventSink`, which
// wraps `Arc<dyn EventSink>` and replaces every emit with a transport
// CloudEvent of type `dev.cellos.events.signed_envelope.v1`. The
// supervisor's `emit()` is back to a one-line forward to the configured
// sinks; whether the bytes that hit JetStream / JSONL are signed or not is
// decided in `composition::build_primary_event_sink` and
// `composition::build_jsonl_sink`.
//
// D7 zeroize hardening for the new SigningConfig lives at its current
// home `crate::event_signing::SigningConfig`.

/// SEAM-1 / L2-04 Phase 2b — resolved activation plan for the in-netns
/// DNS proxy. Built by [`Supervisor::build_dns_proxy_activation`] in the
/// async lifecycle path and consumed by `linux_run_cell_command_isolated`
/// in the sync `spawn_blocking` thread that owns the workload child PID.
///
/// Owns clones of all spec-derived state so the plan survives the
/// async/sync boundary without lifetime entanglement.
/// SEC-22 Phase 2 — resolved activation plan for the in-netns SNI-aware
/// L7 proxy. Built by [`Supervisor::build_sni_proxy_activation`] in the
/// async lifecycle path and consumed by `linux_run_cell_command_isolated`
/// once the workload child PID is known.
///
/// Owns clones of all spec-derived state so the plan survives the
/// async/sync boundary without lifetime entanglement (mirrors
/// [`DnsProxyActivation`]).
#[derive(Debug, Clone)]
pub(crate) struct SniProxyActivation {
    #[allow(dead_code)]
    pub cfg: crate::sni_proxy::SniProxyConfig,
    #[allow(dead_code)]
    pub listen_addr: std::net::SocketAddr,
}

#[derive(Debug, Clone)]
pub(crate) struct DnsProxyActivation {
    // The cfg + listen_addr fields are only consumed by the Linux spawn path.
    // On non-Linux the activation is built but never spawned (skip path); the
    // resolver_id is still useful for diagnostic logging on either platform.
    #[allow(dead_code)]
    pub cfg: crate::dns_proxy::DnsProxyConfig,
    #[allow(dead_code)]
    pub listen_addr: std::net::SocketAddr,
    #[allow(dead_code)]
    pub upstream_addr: std::net::SocketAddr,
    pub resolver_id: String,
}

impl Supervisor {
    pub fn new(
        host: Arc<dyn CellBackend>,
        broker: Arc<dyn SecretBroker>,
        event_sink: Arc<dyn EventSink>,
        jsonl_sink: Option<Arc<dyn EventSink>>,
        default_export_sink: Arc<dyn ExportSink>,
        named_export_sinks: HashMap<String, Arc<dyn ExportSink>>,
        authority_keys: Arc<HashMap<String, String>>,
    ) -> Self {
        Self {
            host,
            broker,
            event_sink,
            jsonl_sink,
            default_export_sink,
            named_export_sinks,
            policy_pack: None,
            authz_policy: None,
            authority_keys,
            trust_verify_keys: Arc::new(HashMap::new()),
            universal_token_accepted_in_permissive_mode: false,
        }
    }

    /// T12 RBAC — admission gate that runs *before* the policy-pack gate and
    /// `host.create()`. Returns `Ok(())` when the spec is authorized;
    /// returns `Err` (after emitting a `cell.authz.v1.rejected` CloudEvent)
    /// when it is not.
    ///
    /// Evaluation order (each step short-circuits with its own rejection
    /// reason):
    /// 1. **Subject**: `spec.correlation.tenantId` must be present AND appear
    ///    in `authz_policy.subjects`. Absent tenant id is treated as
    ///    `subject_not_authorized` — the operator opted into the gate, so
    ///    every spec must declare a tenant.
    /// 2. **Pool**: when `allowed_pools` is non-empty, `spec.placement.poolId`
    ///    must be present and in the list. Absent pool is allowed only when
    ///    `allowed_pools` is empty.
    /// 3. **Policy pack**: when `allowed_policy_packs` is non-empty,
    ///    `spec.policy.packId` must be present and in the list.
    pub(crate) async fn enforce_authz_policy(
        &self,
        doc: &ExecutionCellDocument,
        authz_policy: &AuthorizationPolicy,
    ) -> anyhow::Result<()> {
        let spec = &doc.spec;
        let tenant_id = spec
            .correlation
            .as_ref()
            .and_then(|c| c.tenant_id.as_deref());

        // 1. Subject (tenant) check.
        let tenant = match tenant_id {
            Some(t) => t,
            None => {
                let msg = "spec.correlation.tenantId is required when an \
                           AuthorizationPolicy is configured but the spec did not declare one";
                self.emit_authz_rejection(spec, "subject_not_authorized", msg, None, None)
                    .await;
                return Err(anyhow::anyhow!(
                    "cell spec rejected by authorization policy: {msg}"
                ));
            }
        };
        if !authz_policy.subjects.iter().any(|s| s == tenant) {
            let msg =
                format!("tenantId {tenant:?} is not in the authorization policy subjects list");
            self.emit_authz_rejection(spec, "subject_not_authorized", &msg, None, None)
                .await;
            return Err(anyhow::anyhow!(
                "cell spec rejected by authorization policy: {msg}"
            ));
        }

        // 2. Pool check.
        if !authz_policy.allowed_pools.is_empty() {
            let pool_id = spec.placement.as_ref().and_then(|p| p.pool_id.as_deref());
            match pool_id {
                Some(p) if authz_policy.allowed_pools.iter().any(|x| x == p) => { /* allowed */ }
                Some(p) => {
                    let msg =
                        format!("pool {p:?} is not in the authorization policy allowedPools list");
                    self.emit_authz_rejection(spec, "pool_not_allowed", &msg, Some(p), None)
                        .await;
                    return Err(anyhow::anyhow!(
                        "cell spec rejected by authorization policy: {msg}"
                    ));
                }
                None => {
                    let msg = "spec.placement.poolId is required when the authorization \
                               policy declares allowedPools";
                    self.emit_authz_rejection(spec, "pool_not_allowed", msg, None, None)
                        .await;
                    return Err(anyhow::anyhow!(
                        "cell spec rejected by authorization policy: {msg}"
                    ));
                }
            }
        }

        // 3. Policy pack check.
        if !authz_policy.allowed_policy_packs.is_empty() {
            let pack_id = spec.policy.as_ref().and_then(|p| p.pack_id.as_deref());
            match pack_id {
                Some(p) if authz_policy.allowed_policy_packs.iter().any(|x| x == p) => {
                    /* allowed */
                }
                Some(p) => {
                    let msg = format!(
                        "policy pack {p:?} is not in the authorization policy \
                         allowedPolicyPacks list"
                    );
                    self.emit_authz_rejection(spec, "policy_pack_not_allowed", &msg, None, Some(p))
                        .await;
                    return Err(anyhow::anyhow!(
                        "cell spec rejected by authorization policy: {msg}"
                    ));
                }
                None => {
                    let msg = "spec.policy.packId is required when the authorization \
                               policy declares allowedPolicyPacks";
                    self.emit_authz_rejection(spec, "policy_pack_not_allowed", msg, None, None)
                        .await;
                    return Err(anyhow::anyhow!(
                        "cell spec rejected by authorization policy: {msg}"
                    ));
                }
            }
        }

        Ok(())
    }

    /// Best-effort `cell.authz.v1.rejected` event emit. Logs and discards
    /// transport failures — the admission error returned to `run()` is the
    /// authoritative outcome; we never mask it with an emit error.
    async fn emit_authz_rejection(
        &self,
        spec: &ExecutionCellSpec,
        reason: &str,
        message: &str,
        denied_pool_id: Option<&str>,
        denied_policy_pack_id: Option<&str>,
    ) {
        match authz_rejected_data_v1(spec, reason, message, denied_pool_id, denied_policy_pack_id) {
            Ok(data) => {
                let ev = cloud_event("dev.cellos.events.cell.authz.v1.rejected", data);
                if let Err(e) = self.emit(&ev).await {
                    tracing::warn!(
                        target: "cellos.supervisor.rbac",
                        error = %e,
                        "authz.rejected event emit failed"
                    );
                }
            }
            Err(e) => {
                tracing::warn!(
                    target: "cellos.supervisor.rbac",
                    error = %e,
                    "authz.rejected event build failed"
                );
            }
        }
    }

    /// Compute the `(derivationVerified, roleRoot, parentRunId)` triple
    /// emitted on `dev.cellos.events.cell.lifecycle.v1.started`.
    ///
    /// Returns `(None, None, None)` when the spec carries no derivation
    /// token — preserving the original lifecycle.started shape for specs
    /// that don't opt into authority derivation.
    ///
    /// When a token is present, `derivationVerified` is the boolean result
    /// of `verify_authority_derivation` against `self.authority_keys` —
    /// the same cryptographic check `main.rs` runs as a fail-closed gate
    /// before `build_supervisor`. This means the started event reports the
    /// real ED25519 verification result, not a structural-only stub.
    ///
    /// Extracted as a method so the wiring (Supervisor stores the keys → the
    /// started-event path uses them) can be unit-tested without standing up
    /// a full `Supervisor::run` lifecycle.
    pub(crate) fn compute_derivation_verification(
        &self,
        spec: &ExecutionCellSpec,
        token: Option<&AuthorityDerivationToken>,
    ) -> (Option<bool>, Option<String>, Option<String>) {
        match token {
            Some(token) => {
                let verified =
                    verify_authority_derivation(spec, token, self.authority_keys.as_ref()).is_ok();
                let role = token.role_root.0.clone();
                let parent = token.parent_run_id.clone();
                (Some(verified), Some(role), parent)
            }
            None => (None, None, None),
        }
    }

    /// Emit to the primary sink then optionally to the JSONL sink.
    ///
    /// I5 (per-event signing): when the operator opts in via
    /// `CELLOS_EVENT_SIGNING={ed25519|hmac}` plus a kid + key, the wrapping
    /// happens inside `event_sink` / `jsonl_sink` themselves — see
    /// `crate::event_signing::SigningEventSink`, attached in the
    /// composition root. From `Supervisor`'s perspective, an emit is just a
    /// forward to the configured sinks regardless of whether they're
    /// signing.
    pub(crate) async fn emit(&self, event: &CloudEventV1) -> anyhow::Result<()> {
        self.event_sink
            .emit(event)
            .await
            .map_err(|e| anyhow::anyhow!("primary sink emit: {e}"))?;
        if let Some(ref j) = self.jsonl_sink {
            j.emit(event)
                .await
                .map_err(|e| anyhow::anyhow!("jsonl sink emit: {e}"))?;
        }
        Ok(())
    }

    /// Execute the full cell lifecycle for `doc` with the given `run_id`.
    ///
    /// Returns `Ok(())` on success, `Err` if any phase failed.
    /// Teardown (destroy + revoke) is always called regardless of phase errors.
    pub async fn run(
        &self,
        doc: &ExecutionCellDocument,
        run_id: &str,
        spec_hash: Option<&str>,
    ) -> anyhow::Result<()> {
        // ── T12 authorization gate ─────────────────────────────────────────
        // Runs BEFORE the policy-pack gate so an unauthorized subject can
        // never reach pack evaluation, host.create(), or any secret broker
        // call. Independent rule namespace from the policy pack.
        if let Some(ref authz) = self.authz_policy {
            self.enforce_authz_policy(doc, authz).await?;
        }

        // ── policy admission gate ──────────────────────────────────────────
        // Validate before calling host.create(); a rejected cell never gets a VM.
        if let Some(ref pack) = self.policy_pack {
            let violations = validate_spec_against_policy(&doc.spec, pack);
            if !violations.is_empty() {
                let violation_msgs: Vec<String> = violations
                    .iter()
                    .map(|v| format!("{}: {}", v.rule, v.message))
                    .collect();
                // Emit the rejection event (best-effort — don't mask the admission error).
                if let Ok(data) = policy_rejected_data_v1(&doc.spec, &violations) {
                    let ev = cloud_event("dev.cellos.events.cell.policy.v1.rejected", data);
                    if let Err(e) = self.emit(&ev).await {
                        tracing::warn!(error = %e, "policy.rejected event emit failed");
                    }
                }
                return Err(anyhow::anyhow!(
                    "cell spec rejected by policy pack '{}': {}",
                    pack.id,
                    violation_msgs.join("; ")
                ));
            }
        }

        // SEC-16: capture any broker-supplied correlation ID before the host
        // is created. We forward it into the compliance summary CloudEvent
        // when the operator did not set `spec.correlation.correlationId`,
        // so audit pipelines (taudit, etc.) can stitch a CellOS run together
        // with the broker's exec session ID without an operator hint.
        let broker_correlation_id = self.broker.broker_correlation_id();

        let handle = self
            .host
            .create(doc)
            .await
            .map_err(|e| anyhow::anyhow!("cell create: {e}"))?;

        let ttl = doc.spec.lifetime.ttl_seconds;

        tracing::info!(
            cell_id = %handle.cell_id,
            spec_id = %doc.spec.id,
            run_id = %run_id,
            "cell created"
        );

        let mut phase_err: Option<anyhow::Error> = None;
        // L5-15 — shared FlowAccumulator wired between the in-netns nflog
        // listener (which records `Opened` events as the workload exercises
        // 5-tuples) and the homeostasis emit site (which reads
        // `unique_flow_count()` to stamp `exercised_egress_connections`).
        //
        // Created here, before any subprocess spawn, so the listener's
        // `setns(2)` thread receives a clone that's already alive. Only
        // populated when `CELLOS_PER_FLOW_REALTIME=1` (or the legacy
        // `_FIRECRACKER_PER_FLOW_EBPF=1`) is on — when the env gate is off
        // we keep `None` to preserve the E2-06 contract at the emit site.
        let flow_accumulator: Option<
            std::sync::Arc<
                std::sync::Mutex<crate::ebpf_flow::connection_tracking::FlowAccumulator>,
            >,
        > = {
            let realtime_on = std::env::var(crate::per_flow::ENV_PER_FLOW_REALTIME).as_deref()
                == Ok("1")
                || std::env::var(crate::per_flow::ENV_PER_FLOW_EBPF).as_deref() == Ok("1");
            if realtime_on {
                Some(std::sync::Arc::new(std::sync::Mutex::new(
                    crate::ebpf_flow::connection_tracking::FlowAccumulator::new(),
                )))
            } else {
                None
            }
        };
        let mut identity_materialized = false;
        let mut resolved_secrets: Vec<SecretView> = Vec::new();
        // Captured after the command phase; forwarded to the compliance summary event.
        let mut command_exit_code: Option<i32> = None;
        // Tracks whether the in-VM exit bridge produced an authenticated exit
        // code. None = bridge wasn't used (host-side spawn) so the clean/forced
        // distinction is not meaningful for this run. Some(true) = clean
        // (4-byte exit code received). Some(false) = forced (bridge errored;
        // any recorded exit code is synthetic — supervisor likely SIGKILLed
        // the VMM during teardown).
        let mut in_vm_exit_authenticated: Option<bool> = None;
        // P0-5 / G2 — captured from the lifecycle.started CloudEvent envelope
        // so we can populate `provenance.parent` on revoke + export receipt
        // events. `None` until lifecycle.started has been emitted; downstream
        // call sites tolerate absence (no provenance block in that case).
        let mut started_event_id: Option<String> = None;

        // ── network_scope ──────────────────────────────────────────────────
        let egress_rule_count = doc
            .spec
            .authority
            .egress_rules
            .as_ref()
            .map(|v| v.len())
            .unwrap_or(0);
        let has_opaque_network = doc.spec.authority.network.is_some();
        match observability_network_scope_data_v1(
            &doc.spec,
            &handle.cell_id,
            Some(run_id),
            egress_rule_count,
            has_opaque_network,
        )
        .context("build network_scope data")
        {
            Err(e) => phase_err = Some(e),
            Ok(data) => {
                let ev = cloud_event(
                    "dev.cellos.events.cell.observability.v1.network_scope",
                    data,
                );
                if let Err(e) = self.emit(&ev).await {
                    phase_err = Some(e);
                }
            }
        }

        self.maybe_emit_trust_plane_observability(doc, &handle.cell_id, run_id, &mut phase_err)
            .await;

        // SEC-21: host-controlled resolver refresh + drift events. Disabled
        // unless `CELLOS_DNS_AUTHORITY_REFRESH=1`; runs one tick per cell run.
        self.maybe_run_dns_authority_refresh(doc, &handle.cell_id, run_id, &mut phase_err)
            .await;

        // SEC-21 Phase 2: continuous resolver-refresh ticker. Disabled unless
        // BOTH `CELLOS_DNS_AUTHORITY_REFRESH=1` AND
        // `CELLOS_DNS_AUTHORITY_CONTINUOUS=1`. Runs alongside the workload
        // for the lifetime of the cell; teardown is at the end of `run()`
        // before the destroyed event fires.
        let mut continuous_resolver_ticker: Option<
            cellos_supervisor::resolver_refresh::ticker::TickerHandle,
        > = self.maybe_spawn_continuous_resolver_refresh(doc, &handle.cell_id, run_id);

        // SEC-22 Phase 1: L7 hostname-allowlist evaluation + direct-DNS-block
        // observability. Disabled unless `CELLOS_L7_GATE_OBSERVABILITY=1`.
        // Kernel-level direct-DNS containment runs separately in
        // `generate_nft_ruleset` and is not gated by this flag — only the
        // observable signal is.
        self.maybe_emit_l7_gate_observability(doc, &handle.cell_id, run_id, &mut phase_err)
            .await;

        // SEAM-1 / L2-04 Phase 2b: in-netns DNS proxy activation.
        // Disabled unless `CELLOS_DNS_PROXY=1` AND the spec carries a
        // populated `dnsAuthority.hostnameAllowlist` AND at least one
        // `do53-*` resolver with a parseable `host:port` endpoint. The
        // actual `setns(2)` spawn happens later in
        // `linux_run_cell_command_isolated` once the workload child PID
        // is known; this call only resolves the activation predicate +
        // captures the runtime handle so the spawn can run later.
        let dns_proxy_activation = self.build_dns_proxy_activation(doc, &handle.cell_id, run_id);
        let dns_proxy_emitter = if dns_proxy_activation.is_some() {
            #[cfg(target_os = "linux")]
            {
                Some(std::sync::Arc::new(
                    crate::dns_proxy::spawn::EventSinkEmitter::capture_current(
                        self.event_sink.clone(),
                        self.jsonl_sink.clone(),
                    ),
                )
                    as std::sync::Arc<dyn crate::dns_proxy::DnsQueryEmitter>)
            }
            #[cfg(not(target_os = "linux"))]
            {
                tracing::warn!(
                    target: "cellos.supervisor.dns_proxy",
                    cell_id = %handle.cell_id,
                    run_id = %run_id,
                    "SEAM-1 DNS proxy: non-Linux platform — spawn no-op (predicate held but setns unavailable)"
                );
                None
            }
        } else {
            None
        };

        // SEC-22 Phase 2: in-netns SNI-aware proxy activation. Disabled
        // unless `CELLOS_SNI_PROXY=1` AND the spec carries a populated
        // `dnsAuthority.hostnameAllowlist`. Like the DNS proxy, the actual
        // `setns(2)` spawn happens later in `linux_run_cell_command_isolated`
        // once the workload child PID is known; this slot only resolves the
        // activation predicate + captures the runtime handle.
        let sni_proxy_activation = self.build_sni_proxy_activation(doc, &handle.cell_id, run_id);
        let sni_proxy_emitter = if sni_proxy_activation.is_some() {
            #[cfg(target_os = "linux")]
            {
                Some(std::sync::Arc::new(
                    crate::sni_proxy::spawn::EventSinkEmitter::capture_current(
                        self.event_sink.clone(),
                        self.jsonl_sink.clone(),
                    ),
                )
                    as std::sync::Arc<dyn crate::sni_proxy::L7DecisionEmitter>)
            }
            #[cfg(not(target_os = "linux"))]
            {
                tracing::warn!(
                    target: "cellos.supervisor.sni_proxy",
                    cell_id = %handle.cell_id,
                    run_id = %run_id,
                    "SEC-22 Phase 2 SNI proxy: non-Linux platform — spawn no-op (predicate held but setns unavailable)"
                );
                None
            }
        } else {
            None
        };

        // ── secrets ────────────────────────────────────────────────────────
        let secret_delivery = doc
            .spec
            .run
            .as_ref()
            .map(|run| run.secret_delivery.clone())
            .unwrap_or(SecretDeliveryMode::Env);
        let uses_runtime_leased_broker =
            matches!(secret_delivery, SecretDeliveryMode::RuntimeLeasedBroker);

        if uses_runtime_leased_broker {
            if phase_err.is_none() {
                let requests = runtime_secret_lease_requests_for_doc(doc);
                if !requests.is_empty() {
                    match self
                        .broker
                        .prepare_runtime_secret_lease(&handle.cell_id, &requests)
                        .await
                    {
                        Ok(()) => {
                            if let Some(identity) = &doc.spec.identity {
                                match identity_materialized_data_v1(
                                    &doc.spec,
                                    &handle.cell_id,
                                    Some(run_id),
                                    identity,
                                )
                                .context("build identity materialized data")
                                {
                                    Err(e) => phase_err = Some(e),
                                    Ok(data) => {
                                        let ev = cloud_event(
                                            "dev.cellos.events.cell.identity.v1.materialized",
                                            data,
                                        );
                                        if let Err(e) = self.emit(&ev).await {
                                            phase_err = Some(e);
                                        } else {
                                            identity_materialized = true;
                                        }
                                    }
                                }
                            }
                        }
                        Err(e) => {
                            if let Some(identity) = &doc.spec.identity {
                                let err_text = format!(
                                    "runtime leased broker prepare {} -> {}: {e}",
                                    identity_resolve_key(identity),
                                    identity.secret_ref
                                );
                                self.emit_identity_failed_best_effort(
                                    doc,
                                    &handle.cell_id,
                                    run_id,
                                    identity,
                                    IdentityFailureOperation::Materialize,
                                    &err_text,
                                )
                                .await;
                                phase_err = Some(anyhow::anyhow!("{err_text}"));
                            } else {
                                phase_err =
                                    Some(anyhow::anyhow!("runtime leased broker prepare: {e}"));
                            }
                        }
                    }
                }
            }
        } else {
            if phase_err.is_none() {
                if let Some(identity) = &doc.spec.identity {
                    let identity_ttl = identity.ttl_seconds.unwrap_or(ttl);
                    let resolve_key = identity_resolve_key(identity);
                    let broker_name = configured_broker_name();
                    match self
                        .broker
                        .resolve(resolve_key, &handle.cell_id, identity_ttl)
                        .await
                    {
                        Ok(mut view) if view.value.is_empty() => {
                            // Fail-fast: an empty identity credential is a silent runtime
                            // wedge waiting to happen — surface it now with the same
                            // structured error path used for outright resolve failures.
                            tracing::error!(
                                target: "cellos.supervisor.secrets",
                                key = %identity.secret_ref,
                                resolve_key = %resolve_key,
                                broker = %broker_name,
                                "secret unresolvable"
                            );
                            view.key = identity.secret_ref.clone();
                            let err_text = format!(
                                "identity {} unresolvable via {} broker (empty value for {})",
                                identity.secret_ref, broker_name, resolve_key
                            );
                            self.emit_identity_failed_best_effort(
                                doc,
                                &handle.cell_id,
                                run_id,
                                identity,
                                IdentityFailureOperation::Materialize,
                                &err_text,
                            )
                            .await;
                            phase_err = Some(anyhow::anyhow!("{err_text}"));
                        }
                        Ok(mut view) => {
                            // The broker resolves using the provider-specific lookup key
                            // (for GitHub OIDC, the audience), but the workload contract
                            // exposes the credential under the logical in-cell alias.
                            view.key = identity.secret_ref.clone();
                            resolved_secrets.push(view);
                            match identity_materialized_data_v1(
                                &doc.spec,
                                &handle.cell_id,
                                Some(run_id),
                                identity,
                            )
                            .context("build identity materialized data")
                            {
                                Err(e) => phase_err = Some(e),
                                Ok(data) => {
                                    let ev = cloud_event(
                                        "dev.cellos.events.cell.identity.v1.materialized",
                                        data,
                                    );
                                    if let Err(e) = self.emit(&ev).await {
                                        phase_err = Some(e);
                                    } else {
                                        identity_materialized = true;
                                    }
                                }
                            }
                        }
                        Err(e) => {
                            tracing::error!(
                                target: "cellos.supervisor.secrets",
                                key = %identity.secret_ref,
                                resolve_key = %resolve_key,
                                broker = %broker_name,
                                error = %e,
                                "secret unresolvable"
                            );
                            let err_text = format!(
                                "identity {} unresolvable via {} broker: {e}",
                                identity.secret_ref, broker_name
                            );
                            self.emit_identity_failed_best_effort(
                                doc,
                                &handle.cell_id,
                                run_id,
                                identity,
                                IdentityFailureOperation::Materialize,
                                &err_text,
                            )
                            .await;
                            phase_err = Some(anyhow::anyhow!("{err_text}"));
                        }
                    }
                }
            }

            if phase_err.is_none() {
                if let Some(refs) = &doc.spec.authority.secret_refs {
                    let broker_name = configured_broker_name();
                    for key in refs {
                        if doc
                            .spec
                            .identity
                            .as_ref()
                            .is_some_and(|identity| key == &identity.secret_ref)
                        {
                            continue;
                        }
                        match self.broker.resolve(key, &handle.cell_id, ttl).await {
                            Ok(view) => {
                                // Fail-fast on empty resolution: a broker that returns an empty
                                // string for a declared secretRef is indistinguishable from a
                                // missing secret to the workload, and silently injecting "" leads
                                // to subprocesses hanging on resources they can't open (e.g. an
                                // empty $DISPLAY). Treat empty as unresolvable so the cell is
                                // aborted before spawn rather than wedging at runtime.
                                if view.value.is_empty() {
                                    tracing::error!(
                                        target: "cellos.supervisor.secrets",
                                        key = %key,
                                        broker = %broker_name,
                                        "secret unresolvable"
                                    );
                                    phase_err = Some(anyhow::anyhow!(
                                        "secret {key} unresolvable via {broker_name} broker"
                                    ));
                                    break;
                                }
                                resolved_secrets.push(view);
                            }
                            Err(e) => {
                                tracing::error!(
                                    target: "cellos.supervisor.secrets",
                                    key = %key,
                                    broker = %broker_name,
                                    error = %e,
                                    "secret unresolvable"
                                );
                                phase_err = Some(anyhow::anyhow!(
                                    "secret {key} unresolvable via {broker_name} broker: {e}"
                                ));
                                break;
                            }
                        }
                    }
                }
            }
        }

        // ── started ────────────────────────────────────────────────────────
        if phase_err.is_none() {
            // L5-05 / Lane B: surface authority-derivation verification result on
            // the started event so taudit can answer "prove every prod cell
            // derived authority from role X" by querying lifecycle.started
            // events alone.
            //
            // Non-fatal: a failed verification records derivationVerified=false
            // and lets the run proceed — taudit aggregates the boolean across
            // runs to detect drift. The real fail-closed gate runs in `main.rs`
            // before `build_supervisor` and rejects invalid tokens outright.
            //
            // The role-keys map is plumbed in via `Supervisor::authority_keys`
            // (set by `build_supervisor` from `load_authority_keys`), so the
            // `derivationVerified` flag now reflects the actual ED25519
            // signature check rather than a structural-only stub. Specs
            // without a derivation token emit none of the three new fields,
            // preserving the original lifecycle.started shape.
            let (derivation_verified, role_root_str, parent_run_id_str) = self
                .compute_derivation_verification(
                    &doc.spec,
                    doc.spec.authority.authority_derivation.as_ref(),
                );

            match lifecycle_started_data_v1(
                &doc.spec,
                &handle.cell_id,
                Some(run_id),
                derivation_verified,
                role_root_str.as_deref(),
                parent_run_id_str.as_deref(),
                spec_hash,
                handle.kernel_digest_sha256.as_deref(),
                handle.rootfs_digest_sha256.as_deref(),
                handle.firecracker_digest_sha256.as_deref(),
            )
            .context("build lifecycle started data")
            {
                Err(e) => phase_err = Some(e),
                Ok(data) => {
                    let ev = cloud_event("dev.cellos.events.cell.lifecycle.v1.started", data);
                    // P0-5 / G2: record the started envelope's CloudEvent id so
                    // revoke + export receipt events can stamp it as
                    // `provenance.parent`. We capture even if `emit` fails —
                    // the envelope was constructed and the ID identifies the
                    // attempted event. Downstream provenance is best-effort.
                    started_event_id = Some(ev.id.clone());
                    if let Err(e) = self.emit(&ev).await {
                        phase_err = Some(e);
                    }
                }
            }
        }

        // I6 / O6: when admission accepted a universal (parentRunId: null)
        // authority derivation token in PERMISSIVE mode, emit one structured
        // warning event so the audit trail records the exposure. Strict mode
        // (the 1.0 default) rejects universal tokens at admission time and
        // never reaches this path.
        if self.universal_token_accepted_in_permissive_mode {
            self.emit_universal_token_permissive_warning(doc, &handle.cell_id, run_id, spec_hash)
                .await;
        }

        // ── container_security (Linux only) ───────────────────────────────
        // Emit capability bitmasks read from /proc/self/status immediately after
        // lifecycle.started so the audit trail captures the concrete capability
        // set in place when the cell began — rather than "deepce could not
        // determine capabilities because capsh was not installed."
        #[cfg(target_os = "linux")]
        if phase_err.is_none() {
            let caps = read_proc_caps();
            let data = observability_container_security_data_v1(
                &handle.cell_id,
                Some(run_id),
                caps[0].as_deref(),
                caps[1].as_deref(),
                caps[2].as_deref(),
                caps[3].as_deref(),
                caps[4].as_deref(),
            );
            let ev = cloud_event(
                "dev.cellos.events.cell.observability.v1.container_security",
                data,
            );
            if let Err(e) = self.emit(&ev).await {
                tracing::warn!(error = %e, "container_security event emit failed (non-fatal)");
            }
        }

        // ── network_policy (Linux + CLONE_NEWNET) ─────────────────────────
        #[cfg(target_os = "linux")]
        if phase_err.is_none() {
            if let Some(flags) = linux_subprocess_unshare_flags() {
                if flags & libc::CLONE_NEWNET != 0 {
                    let egress_rules = doc.spec.authority.egress_rules.as_deref().unwrap_or(&[]);
                    match observability_network_policy_data_v1(
                        &doc.spec,
                        &handle.cell_id,
                        Some(run_id),
                        "clone_newnet",
                        egress_rules,
                    )
                    .context("build network_policy data")
                    {
                        Err(e) => phase_err = Some(e),
                        Ok(data) => {
                            let ev = cloud_event(
                                "dev.cellos.events.cell.observability.v1.network_policy",
                                data,
                            );
                            if let Err(e) = self.emit(&ev).await {
                                phase_err = Some(e);
                            }
                        }
                    }
                }
            }
        }

        // ── run ────────────────────────────────────────────────────────────
        let egress_rules_for_run: Vec<EgressRule> =
            doc.spec.authority.egress_rules.clone().unwrap_or_default();

        if phase_err.is_none() {
            if let Some(ref run) = doc.spec.run {
                if let Some(msg) = argv0_allow_prefixes_error(&run.argv) {
                    phase_err = Some(anyhow::anyhow!("{msg}"));
                }
            }
        }

        if phase_err.is_none() {
            if let Some(ref run) = doc.spec.run {
                let program = run.argv.first().map(String::as_str).unwrap_or("");
                let argc = run.argv.len();
                match observability_process_spawned_data_v1(
                    &doc.spec,
                    &handle.cell_id,
                    Some(run_id),
                    program,
                    argc,
                )
                .context("build process_spawned data")
                {
                    Err(e) => phase_err = Some(e),
                    Ok(data) => {
                        let ev = cloud_event(
                            "dev.cellos.events.cell.observability.v1.process_spawned",
                            data,
                        );
                        if let Err(e) = self.emit(&ev).await {
                            phase_err = Some(e);
                        }
                    }
                }

                if phase_err.is_none() {
                    let mut runtime_secret_session: Option<RuntimeSecretSession> = None;
                    if run.secret_delivery == SecretDeliveryMode::RuntimeBroker {
                        let entries = runtime_secret_entries_for_doc(doc, &resolved_secrets);
                        match RuntimeSecretSession::start_snapshot(&handle.cell_id, &entries).await
                        {
                            Ok(session) => {
                                runtime_secret_session = Some(session);
                            }
                            Err(e) => {
                                phase_err =
                                    Some(e.context("start runtimeBroker secret delivery session"));
                            }
                        }
                    } else if run.secret_delivery == SecretDeliveryMode::RuntimeLeasedBroker {
                        let requests = runtime_secret_lease_requests_for_doc(doc);
                        match RuntimeSecretSession::start_leased(
                            &handle.cell_id,
                            self.broker.clone(),
                            &requests,
                        )
                        .await
                        {
                            Ok(session) => {
                                runtime_secret_session = Some(session);
                            }
                            Err(e) => {
                                phase_err = Some(
                                    e.context("start runtimeLeasedBroker secret delivery session"),
                                );
                            }
                        }
                    }

                    let (
                        exit_code,
                        duration_ms,
                        spawn_err,
                        net_enforcement_nft,
                        dns_proxy_spawn_err,
                        sni_proxy_spawn_err,
                        per_flow_rows,
                    ) = if phase_err.is_none() {
                        // Check if the backend manages in-VM execution (e.g. Firecracker
                        // + cellos-init vsock bridge). If so, skip the host-side spawn.
                        if let Some(result) = self.host.wait_for_in_vm_exit(&handle.cell_id).await {
                            let code = match result {
                                Ok(c) => {
                                    // Authenticated exit code via vsock bridge —
                                    // terminal_state will be Clean for the
                                    // destroyed event.
                                    in_vm_exit_authenticated = Some(true);
                                    c
                                }
                                Err(e) => {
                                    // Bridge errored before delivering 4 bytes;
                                    // teardown will SIGKILL the VMM and the -1
                                    // recorded here is synthetic. Distinguish
                                    // this in the destroyed event so audit can
                                    // tell "clean exit" from "we gave up".
                                    in_vm_exit_authenticated = Some(false);
                                    phase_err = Some(anyhow::anyhow!("in-VM exit bridge: {e}"));
                                    -1
                                }
                            };
                            // Surface the backend's nft enforcement signal so the
                            // `network_enforcement` CloudEvent is emitted on the
                            // in-VM exit path with parity to the host-subprocess
                            // path. Backends that don't own enforcement leave
                            // this `None` and the event is suppressed (as before).
                            //
                            // scope (FC-38): in-VM exit path does not surface
                            // per-flow rows yet (the supervisor cannot scrape
                            // counters from a Firecracker guest's nft table).
                            // Future work tracks this with eBPF on the host
                            // bridge or guest-side telemetry.
                            (
                                code,
                                0u64,
                                None,
                                handle.nft_rules_applied,
                                None,
                                None,
                                Vec::new(),
                            )
                        } else {
                            crate::command_runner::run_cell_command(
                                run,
                                doc.spec.lifetime.ttl_seconds,
                                &handle,
                                &egress_rules_for_run,
                                doc.spec.authority.dns_authority.as_ref(),
                                &resolved_secrets,
                                runtime_secret_session.as_ref(),
                                dns_proxy_activation.clone(),
                                dns_proxy_emitter.clone(),
                                sni_proxy_activation.clone(),
                                sni_proxy_emitter.clone(),
                                // Per-flow real-time listener
                                // sink + run_id for the in-netns nflog
                                // listener. None when the env gate is off
                                // or self.event_sink isn't set up; the
                                // closure spawn checks the env again
                                // before binding the netlink socket.
                                Some((self.event_sink.clone(), run_id.to_string())),
                                // L5-15 — clone the shared accumulator into
                                // the listener path. The supervisor keeps
                                // its own clone for the homeostasis emit
                                // read below.
                                flow_accumulator.clone(),
                            )
                            .await
                        }
                    } else {
                        (-1, 0, None, None, None, None, Vec::new())
                    };
                    command_exit_code = Some(exit_code);

                    // SEAM-1 / L2-04 Phase 2b: surface any proxy spawn error
                    // as a single `dns_query` event with `reasonCode:
                    // upstream_failure`. Preserves the audit trail W3 designed
                    // for the activation stub — without this, a setns/bind
                    // failure would silently leave the cell unprotected.
                    if let Some(err_text) = dns_proxy_spawn_err {
                        self.emit_dns_proxy_spawn_failure(
                            doc,
                            &handle.cell_id,
                            run_id,
                            dns_proxy_activation
                                .as_ref()
                                .map(|a| a.resolver_id.as_str()),
                            &err_text,
                        )
                        .await;
                    }

                    // SEC-22 Phase 2 spawn-error audit: wired in W9 finalize.
                    // The closure-return tuple was extended to carry
                    // `sni_proxy_spawn_err` (mirroring the dns_proxy_spawn_err
                    // channel just above), and a setns/bind failure on the L7
                    // proxy now surfaces as one `l7_egress_decision` event with
                    // `reasonCode: l7_unknown_protocol` + an `errorText` field
                    // identifying the spawn-side cause. The cell itself is NOT
                    // killed: SEC-22's kernel-level direct-DNS block remains in
                    // place; this proxy is the L7 audit layer on top, not the
                    // only line of defence.
                    if let Some(err_text) = sni_proxy_spawn_err {
                        self.emit_sni_proxy_spawn_failure(doc, &handle.cell_id, run_id, &err_text)
                            .await;
                    }

                    if let Some(session) = runtime_secret_session.take() {
                        session.shutdown().await;
                    }

                    if let Some(nft_rules_applied) = net_enforcement_nft {
                        match observability_network_enforcement_data_v1(
                            &doc.spec,
                            &handle.cell_id,
                            Some(run_id),
                            nft_rules_applied,
                            egress_rules_for_run.len(),
                            exit_code,
                            spawn_err.as_deref(),
                        )
                        .context("build network_enforcement data")
                        {
                            Err(e) => phase_err = Some(e),
                            Ok(data) => {
                                let ev = cloud_event(
                                    "dev.cellos.events.cell.observability.v1.network_enforcement",
                                    data,
                                );
                                if let Err(e) = self.emit(&ev).await {
                                    phase_err = Some(e);
                                }
                            }
                        }
                    }

                    // FC-38 Phase 1: per-rule attribution events emitted from
                    // the post-run nft counter scan. `per_flow_rows` is empty
                    // unless `CELLOS_PER_FLOW_ENFORCEMENT_EVENTS=1` was set
                    // AND nft applied successfully on the isolated path; on
                    // non-Linux supervisor hosts this loop is a no-op (the
                    // Vec is empty). One event per rule whose decision is
                    // `deny` regardless of count, OR whose decision is
                    // `allow` and packet_count > 0 (allow rules with zero
                    // traffic are intentionally skipped — the operator
                    // already saw the rule declared in `network_policy`).
                    if phase_err.is_none() && !per_flow_rows.is_empty() {
                        self.emit_per_flow_decision_events(
                            doc,
                            &handle.cell_id,
                            run_id,
                            &egress_rules_for_run,
                            &per_flow_rows,
                        )
                        .await;
                    }

                    if phase_err.is_none() {
                        match command_completed_data_v1(
                            &doc.spec,
                            &handle.cell_id,
                            Some(run_id),
                            &run.argv,
                            exit_code,
                            duration_ms,
                            spawn_err.as_deref(),
                        )
                        .context("build command completed data")
                        {
                            Err(e) => phase_err = Some(e),
                            Ok(data) => {
                                let ev = cloud_event(
                                    "dev.cellos.events.cell.command.v1.completed",
                                    data,
                                );
                                if let Err(e) = self.emit(&ev).await {
                                    phase_err = Some(e);
                                } else if let Some(msg) = spawn_err {
                                    phase_err = Some(anyhow::anyhow!("command spawn: {msg}"));
                                } else if exit_code != 0 {
                                    phase_err = Some(anyhow::anyhow!(
                                        "command exited with status {exit_code}"
                                    ));
                                }
                            }
                        }
                    }
                }
            }
        }

        // Authority shaping: secret material is no longer needed after the subprocess exits.
        // Drop now so ZeroizeOnDrop wipes the values before the export / destroy phases run.
        drop(resolved_secrets);

        // P0-4 / G1 — build a spec view whose `Correlation.correlation_id` is
        // back-filled from the broker (`broker_correlation_id`) when the
        // operator did not supply one. We compute this once here and reuse it
        // for export + revoke + compliance summary so audit consumers see the
        // same `correlationId` on every event in the teardown phase, not just
        // on `compliance.summary`. Cloning `doc.spec` keeps the merge scoped:
        // it never mutates the inbound document, so a parallel call sharing
        // the same `&ExecutionCellDocument` keeps its own correlation view.
        let teardown_effective_correlation = {
            let mut c = doc.spec.correlation.clone().unwrap_or_default();
            if c.correlation_id.is_none() {
                c.correlation_id = broker_correlation_id.clone();
            }
            if c == Correlation::default() {
                None
            } else {
                Some(c)
            }
        };
        let teardown_spec = if teardown_effective_correlation == doc.spec.correlation {
            // No change — avoid the clone when the operator already supplied
            // a correlation_id (or no broker correlation is available).
            None
        } else {
            let mut s = doc.spec.clone();
            s.correlation = teardown_effective_correlation;
            Some(s)
        };
        let teardown_spec_ref = teardown_spec.as_ref().unwrap_or(&doc.spec);

        // P0-5 / G2 — pre-compute the `provenance.parent` block we stamp on
        // revoke + export receipt events. Walks back to
        // `cell.lifecycle.v1.started`, which is the receiver-side root of
        // the run for taudit's graph build. `None` only when the started
        // event was never emitted (e.g. an early phase_err); downstream call
        // sites tolerate the absence by emitting no `provenance` block,
        // preserving v1/v2 schema backward-compat.
        let teardown_provenance: Option<cellos_core::Provenance> =
            started_event_id.as_ref().map(|id| cellos_core::Provenance {
                parent: id.clone(),
                parent_type: "dev.cellos.events.cell.lifecycle.v1.started".to_string(),
            });

        // ── export ─────────────────────────────────────────────────────────
        if phase_err.is_none() {
            if let Some(ref export) = doc.spec.export {
                if let Some(artifacts) = &export.artifacts {
                    for art in artifacts {
                        match observability_fs_touch_export_data_v1(
                            &doc.spec,
                            &handle.cell_id,
                            Some(run_id),
                            &art.path,
                            &art.name,
                        )
                        .context("build fs_touch data")
                        {
                            Err(e) => {
                                phase_err = Some(e);
                                break;
                            }
                            Ok(data) => {
                                let ev = cloud_event(
                                    "dev.cellos.events.cell.observability.v1.fs_touch",
                                    data,
                                );
                                if let Err(e) = self.emit(&ev).await {
                                    phase_err = Some(e);
                                    break;
                                }
                            }
                        }

                        if phase_err.is_some() {
                            break;
                        }

                        let effective_target_name = effective_export_target_name(doc, art);
                        let sink = effective_target_name
                            .and_then(|name| self.named_export_sinks.get(name))
                            .unwrap_or(&self.default_export_sink);
                        let target_kind = resolve_export_target_kind(doc, art, sink.target_kind());
                        let destination_hint = logical_export_destination(
                            doc,
                            art,
                            sink.destination_hint(&art.name).as_deref(),
                        );
                        let export_metadata = ExportArtifactMetadata {
                            content_type: art.content_type.clone(),
                        };

                        match sink.push(&art.name, &art.path, &export_metadata).await {
                            Ok(mut receipt) => {
                                receipt.target_kind = target_kind.clone();
                                if receipt.target_name.is_none() {
                                    receipt.target_name = effective_target_name.map(str::to_string);
                                }
                                if let Some(destination) =
                                    logical_export_destination(doc, art, Some(&receipt.destination))
                                {
                                    receipt.destination = destination;
                                }

                                // P0-4 / P0-5: stamp broker-merged correlation
                                // (G1) and lifecycle-started provenance (G2)
                                // so taudit can stitch artifact → started →
                                // spec → derivation token without operator
                                // hints. `teardown_spec_ref` falls back to
                                // `&doc.spec` when no merge was needed.
                                match export_completed_data_v2(
                                    teardown_spec_ref,
                                    &handle.cell_id,
                                    Some(run_id),
                                    &art.name,
                                    &receipt,
                                    teardown_provenance.as_ref(),
                                )
                                .context("build export_completed v2 data")
                                {
                                    Err(e) => {
                                        phase_err = Some(e);
                                        break;
                                    }
                                    Ok(data) => {
                                        let ev = cloud_event(
                                            "dev.cellos.events.cell.export.v2.completed",
                                            data,
                                        );
                                        if let Err(e) = self.emit(&ev).await {
                                            phase_err = Some(e);
                                            break;
                                        }
                                    }
                                }
                            }
                            Err(e) => {
                                let err_text = e.to_string();
                                // P0-4 / P0-5: same broker-merged correlation
                                // + lifecycle-started provenance as the
                                // success path so failed receipts route
                                // through the same audit graph as completed.
                                match export_failed_data_v2(
                                    teardown_spec_ref,
                                    &handle.cell_id,
                                    Some(run_id),
                                    &art.name,
                                    target_kind,
                                    effective_target_name,
                                    destination_hint.as_deref(),
                                    &err_text,
                                    teardown_provenance.as_ref(),
                                )
                                .context("build export_failed v2 data")
                                {
                                    Ok(data) => {
                                        let ev = cloud_event(
                                            "dev.cellos.events.cell.export.v2.failed",
                                            data,
                                        );
                                        if let Err(emit_err) = self.emit(&ev).await {
                                            tracing::warn!(
                                                error = %emit_err,
                                                artifact = %art.name,
                                                "export failed event emit failed"
                                            );
                                        }
                                    }
                                    Err(build_err) => {
                                        tracing::warn!(
                                            error = %build_err,
                                            artifact = %art.name,
                                            "export failed event build failed"
                                        );
                                    }
                                }
                                phase_err = Some(anyhow::anyhow!("{err_text}"));
                                break;
                            }
                        }
                    }
                }
            }
        }

        // ── teardown: destroy + revoke (always) ────────────────────────────
        let destroy_r = self.host.destroy(&handle).await;
        let revoke_r = self.broker.revoke_for_cell(&handle.cell_id).await;

        if let Some(identity) = &doc.spec.identity {
            match &revoke_r {
                Ok(()) if identity_materialized => {
                    // P0-4 / P0-5: revoke is the receiver-side end of the
                    // secret-broker round trip — stamp broker-merged
                    // correlation (G1) and lifecycle-started provenance (G2)
                    // so taudit can answer "which run consumed this lease"
                    // by walking provenance.parent back to the started
                    // envelope without operator joins.
                    match identity_revoked_data_v1(
                        teardown_spec_ref,
                        &handle.cell_id,
                        Some(run_id),
                        identity,
                        Some("teardown"),
                        teardown_provenance.as_ref(),
                    )
                    .context("build identity revoked data")
                    {
                        Err(e) => {
                            if phase_err.is_none() {
                                phase_err = Some(e);
                            } else {
                                tracing::warn!(
                                    error = %e,
                                    "identity revoked event build failed (keeping prior phase error)"
                                );
                            }
                        }
                        Ok(data) => {
                            let ev =
                                cloud_event("dev.cellos.events.cell.identity.v1.revoked", data);
                            if let Err(e) = self.emit(&ev).await {
                                if phase_err.is_none() {
                                    phase_err = Some(e);
                                } else {
                                    tracing::warn!(
                                        error = %e,
                                        "identity revoked event emit failed (keeping prior phase error)"
                                    );
                                }
                            }
                        }
                    }
                }
                Err(e) if identity_materialized => {
                    let err_text = format!("identity revoke {}: {e}", identity.secret_ref);
                    self.emit_identity_failed_best_effort(
                        doc,
                        &handle.cell_id,
                        run_id,
                        identity,
                        IdentityFailureOperation::Revoke,
                        &err_text,
                    )
                    .await;
                }
                _ => {}
            }
        }

        if let Err(e) = destroy_r {
            let err = anyhow::anyhow!("cell destroy: {e}");
            if phase_err.is_none() {
                phase_err = Some(err);
            } else {
                tracing::warn!(error = %e, "destroy failed (keeping prior phase error)");
            }
        }
        if let Err(e) = revoke_r {
            let err = anyhow::anyhow!("secret revoke: {e}");
            if phase_err.is_none() {
                phase_err = Some(err);
            } else {
                tracing::warn!(error = %e, "revoke failed (keeping prior phase error)");
            }
        }

        // ── compliance summary ─────────────────────────────────────────────
        // SEC-16 / P0-4 — the broker-supplied correlation ID has already
        // been merged into `teardown_spec_ref` above (G1), so the same
        // correlation surface that revoke + export receipts saw is the one
        // compliance summary publishes. Single source of truth keeps audit
        // consumers from observing inconsistent `correlation.correlationId`
        // across teardown-phase events for the same run.
        match compliance_summary_data_v1(
            teardown_spec_ref,
            &handle.cell_id,
            Some(run_id),
            command_exit_code,
        )
        .context("build compliance summary data")
        {
            Err(e) => {
                if phase_err.is_none() {
                    phase_err = Some(e);
                }
            }
            Ok(data) => {
                let ev = cloud_event("dev.cellos.events.cell.compliance.v1.summary", data);
                if let Err(e) = self.emit(&ev).await {
                    tracing::warn!(error = %e, "compliance summary emit failed (non-fatal)");
                }
            }
        }

        // ── authority homeostasis ──────────────────────────────────────────────────
        // Best-effort, non-fatal — same pattern as compliance.summary.
        // MVP: egress declared count from spec; exercised = proxy via nft signal.
        // Real per-connection count requires eBPF (L5-12 full).
        {
            let declared_egress = doc
                .spec
                .authority
                .egress_rules
                .as_ref()
                .map(|v| v.len() as u32)
                .unwrap_or(0);
            let declared_secrets = doc
                .spec
                .authority
                .secret_refs
                .as_ref()
                .map(|v| v.len() as u32)
                .unwrap_or(0);

            // L5-15: when `CELLOS_PER_FLOW_REALTIME=1` is on, a per-cell flow
            // accumulator (`crate::ebpf_flow::connection_tracking::FlowAccumulator`)
            // counts UNIQUE 5-tuples observed by the kernel-side backend
            // and the count rides into the homeostasis signal as a real
            // `u32`. The accumulator is constructed at the top of `run()`,
            // a clone is handed to the in-netns nflog listener via
            // `linux_run_cell_command_isolated`, and the listener records
            // an `Opened` event per `cellos-flow accept` datagram it
            // decodes. By the time we reach this emit site `child.wait()`
            // has returned and the listener thread has been joined, so
            // the accumulator holds the real count from nflog or eBPF
            // (whichever backend was wired).
            //
            // The env-off branch preserves the E2-06 contract: `None` +
            // the `telemetry_pending_L2-04` diagnostic so consumers can
            // distinguish "operator opted out" from "operator opted in
            // but observed zero flows."
            //
            // The single source of truth for the env gate is
            // `crate::per_flow::build_activation_from_env`, which honours
            // both the legacy `CELLOS_FIRECRACKER_PER_FLOW_EBPF=1` name and
            // the backend-neutral `CELLOS_PER_FLOW_REALTIME=1` alias.
            let (exercised_egress, exercised_egress_reason): (Option<u32>, Option<String>) =
                if let Some(acc) = flow_accumulator.as_ref() {
                    // Real count from nflog or eBPF — whichever backend
                    // fed the accumulator during the run.
                    let count = match acc.lock() {
                        Ok(guard) => guard.unique_flow_count(),
                        Err(poisoned) => poisoned.into_inner().unique_flow_count(),
                    };
                    let count_u32 = if count > u32::MAX as u64 {
                        u32::MAX
                    } else {
                        count as u32
                    };
                    (Some(count_u32), None)
                } else {
                    (None, Some("telemetry_pending_L2-04".to_string()))
                };

            let efficiency = cellos_core::HomeostasisSignal::compute_efficiency(
                declared_egress,
                exercised_egress,
            );
            let spec_hash = cellos_core::canonical_spec_hash(&doc.spec);

            let signal = cellos_core::HomeostasisSignal {
                spec_hash,
                declared_egress_rules: declared_egress,
                exercised_egress_connections: exercised_egress,
                exercised_egress_reason,
                declared_mount_paths: 0,
                accessed_mount_paths: 0,
                declared_secret_count: declared_secrets,
                authority_efficiency: efficiency,
                recommended_removals: vec![],
            };

            match cellos_core::homeostasis_signal_data_v1(
                &doc.spec,
                &handle.cell_id,
                Some(run_id),
                &signal,
            )
            .context("build homeostasis signal data")
            {
                Err(e) => {
                    tracing::warn!(error = %e, "homeostasis signal build failed (non-fatal)");
                }
                Ok(data) => {
                    let ev = cloud_event("dev.cellos.events.cell.authority.v1.homeostasis", data);
                    if let Err(e) = self.emit(&ev).await {
                        tracing::warn!(error = %e, "homeostasis signal emit failed (non-fatal)");
                    }
                }
            }

            // L5-15 enforcement: when real flow telemetry is wired AND the
            // observed unique-connection count exceeds the declared egress-rule
            // count, emit a sibling violation event. This is a security
            // invariant breach — either the spec under-declares (operator
            // action: widen egressRules) or enforcement leaked (operator
            // action: audit nftables / eBPF rule application). We only fire
            // when `exercised_egress` is `Some(_)`; a `None` count means
            // "telemetry pending" (per E2-06) and cannot prove a violation.
            if let Some(exercised_count) = exercised_egress {
                let declared_u64 = declared_egress as u64;
                let exercised_u64 = exercised_count as u64;
                if exercised_u64 > declared_u64 {
                    tracing::warn!(
                        cell_id = %handle.cell_id,
                        declared = declared_u64,
                        exercised = exercised_u64,
                        "homeostasis violation: exercised egress exceeds declared authority"
                    );
                    let violation_data = cellos_core::homeostasis_violation_data_v1(
                        &handle.cell_id,
                        declared_u64,
                        exercised_u64,
                        &signal.spec_hash,
                    );
                    let violation_event = cloud_event(
                        "dev.cellos.events.cell.authority.v1.homeostasis_violation",
                        violation_data,
                    );
                    if let Err(e) = self.emit(&violation_event).await {
                        tracing::warn!(
                            error = %e,
                            "homeostasis violation emit failed (non-fatal)"
                        );
                    }
                }
            }
        }

        // SEC-21 Phase 2: tear down the continuous resolver-refresh ticker
        // before the destroyed event fires. Bounded join: 2s timeout matches
        // the W4 SEAM-1 Phase 2b DNS proxy shutdown pattern.
        if let Some(ticker_handle) = continuous_resolver_ticker.take() {
            ticker_handle
                .shutdown
                .store(true, std::sync::atomic::Ordering::SeqCst);
            match tokio::time::timeout(std::time::Duration::from_secs(2), ticker_handle.task).await
            {
                Ok(Ok(stats)) => {
                    tracing::info!(
                        target: "cellos.supervisor.resolver_refresh.ticker",
                        cell_id = %handle.cell_id,
                        run_id = %run_id,
                        tick_count = stats.tick_count,
                        events_emitted = stats.events_emitted,
                        resolver_errors = stats.resolver_errors,
                        "SEC-21 Phase 2 continuous resolver-refresh ticker exited"
                    );
                }
                Ok(Err(join_err)) => {
                    tracing::warn!(
                        target: "cellos.supervisor.resolver_refresh.ticker",
                        cell_id = %handle.cell_id,
                        run_id = %run_id,
                        error = %join_err,
                        "SEC-21 Phase 2 continuous resolver-refresh ticker panicked"
                    );
                }
                Err(_) => {
                    tracing::warn!(
                        target: "cellos.supervisor.resolver_refresh.ticker",
                        cell_id = %handle.cell_id,
                        run_id = %run_id,
                        "SEC-21 Phase 2 continuous resolver-refresh ticker join timeout (>2s); abandoning"
                    );
                }
            }
        }

        // ── destroyed ──────────────────────────────────────────────────────
        let outcome = if phase_err.is_none() {
            LifecycleDestroyOutcome::Succeeded
        } else {
            LifecycleDestroyOutcome::Failed
        };
        let reason = phase_err.as_ref().map(|e| e.to_string());
        // terminal_state is populated only when the in-VM bridge was actually
        // consulted for this run. Host-side spawn paths emit no terminalState
        // — the clean/forced distinction only describes whether the
        // authenticated vsock exit-code path delivered a result.
        let terminal_state = in_vm_exit_authenticated.map(|authenticated| {
            if authenticated {
                LifecycleTerminalState::Clean
            } else {
                LifecycleTerminalState::Forced
            }
        });
        // F5 (D5 destruction-evidence): pass `None, None` until F1b wires
        // the per-run evidence-bundle aggregator (see
        // `destruction_evidence::DestructionEvidenceAggregator`). Once F1b
        // lands, the supervisor will look up the URN of the bundle for
        // `run_id` and the host backend's `TeardownReport`-derived residue
        // class and pass them here. The function is additive: today's
        // `None, None` produces the byte-identical legacy payload.
        match lifecycle_destroyed_data_v1(
            &doc.spec,
            &handle.cell_id,
            Some(run_id),
            outcome,
            reason.as_deref(),
            terminal_state,
            None,
            None,
        )
        .context("build lifecycle destroyed data")
        {
            Err(e) => {
                if phase_err.is_none() {
                    phase_err = Some(e);
                }
            }
            Ok(data) => {
                let ev = cloud_event("dev.cellos.events.cell.lifecycle.v1.destroyed", data);
                if let Err(e) = self.emit(&ev).await {
                    if phase_err.is_none() {
                        phase_err = Some(e);
                    } else {
                        tracing::warn!(error = %e, "destroyed event emit failed (keeping prior phase error)");
                    }
                }
            }
        }

        if let Some(e) = phase_err {
            return Err(e);
        }

        tracing::info!(cell_id = %handle.cell_id, "lifecycle complete");
        Ok(())
    }
}

// Network-policy helpers moved to `crate::network_policy`:
//   generate_nft_ruleset, pick_resolver_socket_addr,
//   parse_do53_endpoint, parse_udp_resolver_endpoint,
//   apply_nft_in_ns, scan_nft_counters_in_ns, read_proc_caps.

#[cfg(test)]
mod tests {
    // ── Lane B: derivationVerified wiring ─────────────────────────────────
    //
    // Proves the role-keys map plumbed in via `Supervisor::authority_keys`
    // is the one consulted when computing the `derivationVerified` flag on
    // `dev.cellos.events.cell.lifecycle.v1.started`. We exercise the helper
    // directly rather than running the full `Supervisor::run` lifecycle —
    // the inline match block at the started-event call site delegates here,
    // so this is the smallest assertion that proves the wiring is honest.
    mod derivation_verified {
        use super::super::Supervisor;
        use cellos_core::ports::{NoopEventSink, NoopExportSink};
        use cellos_core::spec_validation::authority_derivation_signing_payload;
        use cellos_core::{
            AuthorityBundle, AuthorityCapability, AuthorityDerivationToken, AuthoritySignature,
            EgressRule, ExecutionCellSpec, Lifetime, RoleId,
        };
        use cellos_host_cellos::MemorySecretBroker;
        use cellos_host_stub::StubCellBackend;
        use std::collections::HashMap;
        use std::sync::Arc;

        fn signed_spec_and_token(
            seed: u8,
            tamper_signature: bool,
        ) -> (ExecutionCellSpec, AuthorityDerivationToken, String) {
            use base64::engine::general_purpose::STANDARD;
            use base64::Engine as _;
            use ed25519_dalek::{Signer as _, SigningKey};

            let signing_key = SigningKey::from_bytes(&[seed; 32]);
            let verifying_key_b64 = STANDARD.encode(signing_key.verifying_key().to_bytes());

            let egress = vec![EgressRule {
                host: "api.example.com".into(),
                port: 443,
                protocol: Some("https".into()),
                dns_egress_justification: None,
            }];

            let spec = ExecutionCellSpec {
                id: "lane-b-cell".into(),
                correlation: None,
                ingress: None,
                environment: None,
                placement: None,
                policy: None,
                identity: None,
                run: None,
                authority: AuthorityBundle {
                    egress_rules: Some(egress.clone()),
                    secret_refs: Some(vec!["api-key".into()]),
                    // authority_derivation attached on the caller side
                    ..AuthorityBundle::default()
                },
                lifetime: Lifetime { ttl_seconds: 60 },
                export: None,
                telemetry: None,
            };

            let mut token = AuthorityDerivationToken {
                role_root: RoleId("role-test".into()),
                parent_run_id: Some("run-lane-b".into()),
                derivation_steps: vec![],
                leaf_capability: AuthorityCapability {
                    egress_rules: egress,
                    secret_refs: vec!["api-key".into()],
                },
                grantor_signature: AuthoritySignature {
                    algorithm: "ed25519".into(),
                    bytes: String::new(),
                },
            };
            let payload = authority_derivation_signing_payload(&token).expect("payload encode");
            let signature = signing_key.sign(&payload);
            let mut sig_bytes = signature.to_bytes();
            if tamper_signature {
                sig_bytes[0] ^= 0xFF; // flip a byte to invalidate
            }
            token.grantor_signature.bytes = STANDARD.encode(sig_bytes);

            (spec, token, verifying_key_b64)
        }

        fn supervisor_with_keys(keys: HashMap<String, String>) -> Supervisor {
            Supervisor::new(
                Arc::new(StubCellBackend),
                Arc::new(MemorySecretBroker::new()),
                Arc::new(NoopEventSink),
                None,
                Arc::new(NoopExportSink),
                HashMap::new(),
                Arc::new(keys),
            )
        }

        #[test]
        fn returns_none_triple_when_no_token_attached() {
            let (spec, _token, _vk) = signed_spec_and_token(0xA1, false);
            let sup = supervisor_with_keys(HashMap::new());
            let (verified, role, parent) = sup.compute_derivation_verification(&spec, None);
            assert_eq!(verified, None);
            assert_eq!(role, None);
            assert_eq!(parent, None);
        }

        #[test]
        fn reports_true_when_keys_authorize_valid_signature() {
            let (spec, token, verifying_key_b64) = signed_spec_and_token(0xA2, false);
            let mut keys = HashMap::new();
            keys.insert("role-test".to_string(), verifying_key_b64);
            let sup = supervisor_with_keys(keys);
            let (verified, role, parent) = sup.compute_derivation_verification(&spec, Some(&token));
            assert_eq!(
                verified,
                Some(true),
                "valid signature + matching role key must verify"
            );
            assert_eq!(role.as_deref(), Some("role-test"));
            assert_eq!(parent.as_deref(), Some("run-lane-b"));
        }

        #[test]
        fn reports_false_when_signature_is_tampered() {
            let (spec, token, verifying_key_b64) = signed_spec_and_token(0xA3, true);
            let mut keys = HashMap::new();
            keys.insert("role-test".to_string(), verifying_key_b64);
            let sup = supervisor_with_keys(keys);
            let (verified, role, parent) = sup.compute_derivation_verification(&spec, Some(&token));
            assert_eq!(
                verified,
                Some(false),
                "tampered signature must fail verification"
            );
            // Role + parent are still echoed so taudit can group failures.
            assert_eq!(role.as_deref(), Some("role-test"));
            assert_eq!(parent.as_deref(), Some("run-lane-b"));
        }

        #[test]
        fn reports_false_when_keys_map_is_empty() {
            // Pre-Lane-B behaviour: missing operator keys → unknown role →
            // derivationVerified=false. Locks in the contract that an empty
            // map (e.g. CELLOS_AUTHORITY_KEYS_PATH unset) does NOT silently
            // claim verification succeeded.
            let (spec, token, _vk) = signed_spec_and_token(0xA4, false);
            let sup = supervisor_with_keys(HashMap::new());
            let (verified, _role, _parent) =
                sup.compute_derivation_verification(&spec, Some(&token));
            assert_eq!(verified, Some(false));
        }
    }
}