cellos-supervisor 0.5.1

//! Trust-plane observability methods split out from `supervisor.rs` (P0-2).
//!
//! Hosts an additional `impl Supervisor` block carrying the six trust-plane /
//! authority audit methods invoked from `Supervisor::run`:
//!
//! - `emit_universal_token_permissive_warning` — I6/O6 single CloudEvent when
//!   admission accepts a `parentRunId: null` derivation token under permissive
//!   mode.
//! - `emit_identity_failed_best_effort` — central path for `cell.identity.failed`
//!   events (broker/runtime/leased failures).
//! - `maybe_emit_trust_plane_observability` — SEC-20/24 startup-time
//!   `dns_target_set` + `l7_egress_decision` audit events from the spec.
//! - `maybe_run_dns_authority_refresh` — SEC-21 host-controlled DNS resolver
//!   refresh single-shot at startup.
//! - `maybe_spawn_continuous_resolver_refresh` — SEC-21 Phase 2 continuous
//!   ticker daemon, spawned only when both env gates are on.
//! - `maybe_emit_l7_gate_observability` — SEC-22 Phase 1 startup audit events:
//!   one `l7_egress_decision` per declared hostname-allowlist entry, plus one
//!   `workload_direct_dns_blocked` deny event per port-53 egress rule whose
//!   destination is not a declared resolver.
//!
//! Pure code-move from `supervisor.rs`. No logic changes.

use std::collections::HashMap;
use std::time::Duration;

use anyhow::Context;
use cellos_core::{
    canonical_spec_hash, identity_failed_data_v1, observability_dns_resolution_data_v1,
    observability_dns_target_set_data_v1, observability_l7_egress_decision_data_v1,
    ExecutionCellDocument, IdentityFailureOperation, WorkloadIdentity,
    TRUST_PLANE_AGGREGATE_EGRESS_FQDN, TRUST_PLANE_BUILTIN_KEYSET_ID, TRUST_PLANE_BUILTIN_L7_KID,
    TRUST_PLANE_BUILTIN_RESOLVER_KID,
};

use crate::network_policy::{
    egress_destination_matches_declared_resolver, pick_resolver_socket_addr,
};
use crate::supervisor::Supervisor;
use crate::supervisor_helpers::{
    cloud_event, dns_authority_continuous_enabled, dns_authority_refresh_enabled,
    dns_authority_tick_interval, l7_gate_observability_enabled, trust_plane_observability_enabled,
};

impl Supervisor {
    /// I6 / O6: emit a single CloudEvent recording that admission accepted a
    /// universal (`parentRunId: null`) authority derivation token because the
    /// supervisor was running in permissive mode
    /// (`CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS=0`/`false`/`no`/`off`).
    ///
    /// Event type: `dev.cellos.events.cell.identity.v1.universal_token_accepted_in_permissive_mode`.
    /// Best-effort — emit failure does not abort the run; the audit trail's
    /// missing-event signal is itself a triage clue.
    pub(crate) async fn emit_universal_token_permissive_warning(
        &self,
        doc: &ExecutionCellDocument,
        cell_id: &str,
        run_id: &str,
        spec_hash: Option<&str>,
    ) {
        let role_root = doc
            .spec
            .authority
            .authority_derivation
            .as_ref()
            .map(|t| t.role_root.0.clone());
        let data = serde_json::json!({
            "cellId": cell_id,
            "runId": run_id,
            "specId": doc.spec.id,
            "specHash": spec_hash,
            "roleRoot": role_root,
            "reason": "universal authority derivation token (parentRunId: null) accepted in permissive mode",
            "remediation": "remove CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS=0 (or any falsy value) to restore the 1.0 strict default; or re-issue the token with an explicit parentRunId",
        });
        let ev = cloud_event(
            "dev.cellos.events.cell.identity.v1.universal_token_accepted_in_permissive_mode",
            data,
        );
        if let Err(e) = self.emit(&ev).await {
            tracing::warn!(
                target: "cellos.supervisor.authority",
                error = %e,
                "universal_token_accepted_in_permissive_mode event emit failed"
            );
        }
    }

    pub(crate) async fn emit_identity_failed_best_effort(
        &self,
        doc: &ExecutionCellDocument,
        cell_id: &str,
        run_id: &str,
        identity: &WorkloadIdentity,
        operation: IdentityFailureOperation,
        reason: &str,
    ) {
        match identity_failed_data_v1(
            &doc.spec,
            cell_id,
            Some(run_id),
            identity,
            operation,
            reason,
        )
        .context("build identity failed data")
        {
            Ok(data) => {
                let ev = cloud_event("dev.cellos.events.cell.identity.v1.failed", data);
                if let Err(e) = self.emit(&ev).await {
                    tracing::warn!(error = %e, "identity failed event emit failed");
                }
            }
            Err(e) => {
                tracing::warn!(error = %e, "identity failed event build failed");
            }
        }
    }

    /// Trust-plane observability (SEC-20/21/24): declared egress mirror + optional DNS resolution.
    ///
    /// Disabled when `CELLOS_TRUST_PLANE_EVENTS` is `0`, `false`, `no`, or `off`.
    /// Optional live DNS: set `CELLOS_RESOLVE_EGRESS_DNS=1` to resolve each declared egress host.
    pub(crate) async fn maybe_emit_trust_plane_observability(
        &self,
        doc: &ExecutionCellDocument,
        cell_id: &str,
        run_id: &str,
        phase_err: &mut Option<anyhow::Error>,
    ) {
        if !trust_plane_observability_enabled() {
            return;
        }
        let policy_digest = canonical_spec_hash(&doc.spec);
        let keyset_id = std::env::var("CELLOS_TRUST_KEYSET_ID")
            .unwrap_or_else(|_| TRUST_PLANE_BUILTIN_KEYSET_ID.to_string());
        let resolver_kid = std::env::var("CELLOS_TRUST_RESOLVER_KID")
            .unwrap_or_else(|_| TRUST_PLANE_BUILTIN_RESOLVER_KID.to_string());
        let l7_kid = std::env::var("CELLOS_TRUST_L7_KID")
            .unwrap_or_else(|_| TRUST_PLANE_BUILTIN_L7_KID.to_string());
        let updated_at = chrono::Utc::now().to_rfc3339();
        let resolve_dns = std::env::var("CELLOS_RESOLVE_EGRESS_DNS")
            .map(|v| {
                let t = v.trim().to_ascii_lowercase();
                matches!(t.as_str(), "1" | "true" | "yes" | "on")
            })
            .unwrap_or(false);

        match observability_dns_target_set_data_v1(
            &doc.spec,
            cell_id,
            Some(run_id),
            TRUST_PLANE_AGGREGATE_EGRESS_FQDN,
            "empty",
            policy_digest.as_str(),
            "refresh",
            &updated_at,
            keyset_id.as_str(),
            resolver_kid.as_str(),
        )
        .context("build dns_target_set data")
        {
            Err(e) => {
                if phase_err.is_none() {
                    *phase_err = Some(e);
                }
            }
            Ok(data) => {
                let ev = cloud_event(
                    "dev.cellos.events.cell.observability.v1.dns_target_set",
                    data,
                );
                if let Err(e) = self.emit(&ev).await {
                    if phase_err.is_none() {
                        *phase_err = Some(e);
                    }
                }
            }
        }

        if let Some(rules) = doc.spec.authority.egress_rules.as_deref() {
            for (idx, rule) in rules.iter().enumerate() {
                let decision_id = uuid::Uuid::new_v4().to_string();
                let rule_ref = format!("authority.egressRules[{idx}]");
                match observability_l7_egress_decision_data_v1(
                    &doc.spec,
                    cell_id,
                    Some(run_id),
                    decision_id.as_str(),
                    "allow",
                    rule.host.as_str(),
                    policy_digest.as_str(),
                    keyset_id.as_str(),
                    l7_kid.as_str(),
                    "egress_declared_allowlist",
                    Some(rule_ref.as_str()),
                    None,
                )
                .context("build l7_egress_decision data")
                {
                    Err(e) => {
                        if phase_err.is_none() {
                            *phase_err = Some(e);
                        }
                    }
                    Ok(data) => {
                        let ev = cloud_event(
                            "dev.cellos.events.cell.observability.v1.l7_egress_decision",
                            data,
                        );
                        if let Err(e) = self.emit(&ev).await {
                            if phase_err.is_none() {
                                *phase_err = Some(e);
                            }
                        }
                    }
                }

                if !resolve_dns {
                    continue;
                }
                let sock = format!("{}:{}", rule.host, rule.port);
                let resolved = tokio::time::timeout(
                    Duration::from_secs(3),
                    tokio::task::spawn_blocking(move || {
                        use std::net::ToSocketAddrs;
                        sock.to_socket_addrs()
                            .map(|i| i.take(24).collect::<Vec<std::net::SocketAddr>>())
                            .unwrap_or_default()
                    }),
                )
                .await;
                let addrs: Vec<std::net::SocketAddr> = match resolved {
                    Ok(Ok(addrs)) => addrs,
                    _ => Vec::new(),
                };
                let mut rows: Vec<(String, String, Option<u16>)> = Vec::with_capacity(addrs.len());
                for a in addrs {
                    match a {
                        std::net::SocketAddr::V4(v4) => {
                            rows.push((v4.ip().to_string(), "inet".to_string(), Some(rule.port)))
                        }
                        std::net::SocketAddr::V6(v6) => {
                            rows.push((v6.ip().to_string(), "inet6".to_string(), Some(rule.port)))
                        }
                    }
                }
                let row_refs: Vec<(&str, &str, Option<u16>)> = rows
                    .iter()
                    .map(|(a, f, p)| (a.as_str(), f.as_str(), *p))
                    .collect();
                let ttl_seconds: i64 = if rows.is_empty() { 0 } else { 300 };
                let rid = format!("rcp-{run_id}-{}", rule.host);
                let resolved_at = chrono::Utc::now().to_rfc3339();
                match observability_dns_resolution_data_v1(
                    &doc.spec,
                    cell_id,
                    Some(run_id),
                    rule.host.as_str(),
                    &resolved_at,
                    row_refs.as_slice(),
                    ttl_seconds,
                    policy_digest.as_str(),
                    keyset_id.as_str(),
                    resolver_kid.as_str(),
                    Some(rid.as_str()),
                )
                .context("build dns_resolution data")
                {
                    Err(e) => {
                        if phase_err.is_none() {
                            *phase_err = Some(e);
                        }
                    }
                    Ok(data) => {
                        let ev = cloud_event(
                            "dev.cellos.events.cell.observability.v1.dns_resolution",
                            data,
                        );
                        if let Err(e) = self.emit(&ev).await {
                            if phase_err.is_none() {
                                *phase_err = Some(e);
                            }
                        }
                    }
                }
            }
        }
    }

    /// SEC-21 host-controlled resolver refresh — one tick per cell run.
    ///
    /// Gated behind `CELLOS_DNS_AUTHORITY_REFRESH=1` (disabled by default,
    /// mirroring `CELLOS_RESOLVE_EGRESS_DNS`). When enabled and the spec
    /// carries `authority.dnsAuthority.refreshPolicy`, runs a single
    /// [`cellos_supervisor::resolver_refresh::ResolverRefresh::tick`] over the declared
    /// hostname allowlist (or, if empty, the egress-rule hosts) and emits any
    /// resulting `dns_authority_drift` CloudEvents through the configured
    /// [`EventSink`].
    ///
    /// "One tick per run" satisfies the SEC-21 acceptance contract (deterministic
    /// refresh loop with bounded stale window). A continuous-ticker daemon mode
    /// — emitting drift events while the cell is running — is deliberately a
    /// follow-up.
    pub(crate) async fn maybe_run_dns_authority_refresh(
        &self,
        doc: &ExecutionCellDocument,
        cell_id: &str,
        run_id: &str,
        phase_err: &mut Option<anyhow::Error>,
    ) {
        if !dns_authority_refresh_enabled() {
            return;
        }
        let Some(dns_authority) = doc.spec.authority.dns_authority.as_ref() else {
            return;
        };
        if dns_authority.refresh_policy.is_none() {
            return;
        }

        // Hostname pool: prefer the explicit allowlist; fall back to the hosts
        // implied by declared egress rules so a spec without a hostname allowlist
        // still gets drift coverage for what it can talk to.
        let hostnames: Vec<String> = if !dns_authority.hostname_allowlist.is_empty() {
            dns_authority.hostname_allowlist.clone()
        } else {
            doc.spec
                .authority
                .egress_rules
                .as_deref()
                .map(|rules| rules.iter().map(|r| r.host.clone()).collect())
                .unwrap_or_default()
        };
        if hostnames.is_empty() {
            return;
        }

        let policy_digest = canonical_spec_hash(&doc.spec);
        let keyset_id = std::env::var("CELLOS_TRUST_KEYSET_ID")
            .unwrap_or_else(|_| TRUST_PLANE_BUILTIN_KEYSET_ID.to_string());
        let resolver_kid = std::env::var("CELLOS_TRUST_RESOLVER_KID")
            .unwrap_or_else(|_| TRUST_PLANE_BUILTIN_RESOLVER_KID.to_string());

        let refresher = cellos_supervisor::resolver_refresh::ResolverRefresh {
            policy: dns_authority.refresh_policy.as_ref(),
            // SEC-21 Phase 3e — DNS rebinding mitigation. Threaded through
            // from the spec; `None` when the operator hasn't declared a
            // policy → no per-hostname response-IP tracking. The startup
            // tick uses the same wiring as the continuous ticker; both
            // share the same per-cell rebinding state via the ticker
            // handle (the startup tick uses a throwaway state because it
            // runs once before the ticker spawns).
            rebinding_policy: dns_authority.rebinding_policy.as_ref(),
            resolvers: dns_authority.resolvers.as_slice(),
            hostnames: hostnames.as_slice(),
            keyset_id: Some(keyset_id.as_str()),
            issuer_kid: Some(resolver_kid.as_str()),
            policy_digest: Some(policy_digest.as_str()),
            correlation_id: doc
                .spec
                .correlation
                .as_ref()
                .and_then(|c| c.correlation_id.as_deref()),
            source: Some("cellos-supervisor"),
            // SEC-21 Phase 3h — startup tick is single-shot and not the
            // intended DNSSEC enforcement point; the continuous ticker is
            // the production validation site. Leaving these `None` here
            // preserves the startup tick's existing behaviour and keeps the
            // P3a-only path unchanged for callers who don't yet wire DNSSEC.
            dnssec_policy: None,
            trust_anchors: None,
        };

        // SEC-21 Phase 3 — production resolver routes through hickory-resolver
        // so the emitted `dns_authority_drift` events carry real upstream TTL.
        // Pre-resolve every hostname on a spawn_blocking thread (so the
        // supervisor's async runtime never stalls on a slow DNS lookup), then
        // the sync closure passed to `tick()` reads from the pre-resolved map.
        let upstream_addr = pick_resolver_socket_addr(dns_authority.resolvers.as_slice());
        let resolver_timeout = std::time::Duration::from_secs(2);
        let hostnames_for_resolve: Vec<String> = hostnames.clone();
        let resolved: HashMap<String, Option<cellos_supervisor::resolver_refresh::ResolvedAnswer>> =
            match tokio::task::spawn_blocking(move || {
                let runtime_handle = tokio::runtime::Handle::current();
                let mut out: HashMap<
                    String,
                    Option<cellos_supervisor::resolver_refresh::ResolvedAnswer>,
                > = HashMap::new();
                for hostname in hostnames_for_resolve {
                    let answer = runtime_handle.block_on(
                        cellos_supervisor::resolver_refresh::resolve_with_ttl(
                            &hostname,
                            upstream_addr,
                            resolver_timeout,
                        ),
                    );
                    out.insert(hostname, answer.ok());
                }
                out
            })
            .await
            {
                Ok(map) => map,
                Err(_) => {
                    // spawn_blocking join error — bail rather than block-emit on
                    // unknown state. No drift events; supervisor.run continues.
                    return;
                }
            };

        let mut state = cellos_supervisor::resolver_refresh::ResolverState::new();
        let resolver = |hostname: &str| -> std::io::Result<
            cellos_supervisor::resolver_refresh::ResolvedAnswer,
        > {
            match resolved.get(hostname) {
                Some(Some(answer)) => Ok(answer.clone()),
                _ => Err(std::io::Error::other(
                    "dns resolve failed or hostname not pre-resolved",
                )),
            }
        };

        let events = refresher.tick(
            &mut state,
            &resolver,
            std::time::SystemTime::now(),
            cell_id,
            run_id,
        );

        for ev in events {
            if let Err(e) = self.emit(&ev).await {
                if phase_err.is_none() {
                    *phase_err = Some(e);
                }
            }
        }
    }

    /// SEC-21 Phase 2 — spawn the continuous-ticker daemon for the cell's
    /// lifetime.
    ///
    /// Activation predicate (all required):
    ///
    /// 1. `CELLOS_DNS_AUTHORITY_REFRESH=1` — Phase 1 master gate.
    /// 2. `CELLOS_DNS_AUTHORITY_CONTINUOUS=1` — Phase 2 specific gate.
    /// 3. The spec carries `authority.dnsAuthority.refreshPolicy`.
    /// 4. There is at least one hostname to refresh (allowlist or
    ///    egress-rule-derived).
    ///
    /// Returns `Some(TickerHandle)` when activation held and the spawn
    /// succeeded; `None` otherwise. The supervisor is responsible for
    /// joining the handle at cell destroy.
    ///
    /// **Additivity:** this method does NOT replace
    /// [`Self::maybe_run_dns_authority_refresh`]. The startup one-tick
    /// still fires at boot to give the operator a baseline observation
    /// even when the continuous ticker is off; the ticker (when on) runs
    /// on top of that, with its own independent
    /// [`cellos_supervisor::resolver_refresh::ResolverState`]. Practical
    /// consequence: the **first** continuous-tick observation per
    /// hostname emits a baseline drift event with `previousDigest:
    /// empty`, even when the startup tick already observed the same
    /// hostname.
    pub(crate) fn maybe_spawn_continuous_resolver_refresh(
        &self,
        doc: &ExecutionCellDocument,
        cell_id: &str,
        run_id: &str,
    ) -> Option<cellos_supervisor::resolver_refresh::ticker::TickerHandle> {
        if !dns_authority_refresh_enabled() {
            return None;
        }
        if !dns_authority_continuous_enabled() {
            return None;
        }
        let dns_authority = doc.spec.authority.dns_authority.as_ref()?;
        dns_authority.refresh_policy.as_ref()?;

        let hostnames: Vec<String> = if !dns_authority.hostname_allowlist.is_empty() {
            dns_authority.hostname_allowlist.clone()
        } else {
            doc.spec
                .authority
                .egress_rules
                .as_deref()
                .map(|rules| rules.iter().map(|r| r.host.clone()).collect())
                .unwrap_or_default()
        };
        if hostnames.is_empty() {
            return None;
        }

        let interval = dns_authority_tick_interval(dns_authority.refresh_policy.as_ref());
        let policy_digest = canonical_spec_hash(&doc.spec);
        let keyset_id = std::env::var("CELLOS_TRUST_KEYSET_ID")
            .unwrap_or_else(|_| TRUST_PLANE_BUILTIN_KEYSET_ID.to_string());
        let resolver_kid = std::env::var("CELLOS_TRUST_RESOLVER_KID")
            .unwrap_or_else(|_| TRUST_PLANE_BUILTIN_RESOLVER_KID.to_string());
        let correlation_id = doc
            .spec
            .correlation
            .as_ref()
            .and_then(|c| c.correlation_id.clone());

        // SEC-21 Phase 3 — production resolver routes through hickory-resolver
        // so the ticker emits `dns_authority_drift` events carrying real
        // upstream TTL. Pick the upstream + timeout once and share between
        // the unvalidated (P3a) and validating (P3h) closures so they hit
        // the same nameserver and the same deadline.
        let upstream_addr = pick_resolver_socket_addr(dns_authority.resolvers.as_slice());
        let resolver_timeout = std::time::Duration::from_secs(2);

        // SEC-21 Phase 3h — load trust anchors once per ticker (env >
        // spec > IANA-default); failed loads surface as
        // `dns_authority_dnssec_failed{reason: trust_anchor_missing}`
        // when the operator opts in via the resolver `dnssec` field.
        // When no resolver has opted in, the loader is never called and
        // we leave both `dnssec_policy` and `trust_anchors` as `None`.
        let dnssec_policy_for_ticker: Option<cellos_core::DnsResolverDnssecPolicy> = dns_authority
            .resolvers
            .iter()
            .find_map(|r| r.dnssec.clone());
        let trust_anchors_for_ticker: Option<cellos_supervisor::resolver_refresh::TrustAnchors> =
            dnssec_policy_for_ticker.as_ref().and_then(|p| {
                cellos_supervisor::resolver_refresh::TrustAnchors::load(
                    p.trust_anchors_path.as_deref(),
                )
                .ok()
            });

        let cfg = cellos_supervisor::resolver_refresh::ticker::TickerConfig {
            interval,
            policy: dns_authority.refresh_policy.clone(),
            // SEC-21 Phase 3e — propagate the operator's DNS rebinding
            // mitigation policy so the continuous ticker's per-cell
            // `RebindingState` enforces it across the cell's lifetime.
            // `None` (no policy declared) → no rebinding events fire.
            rebinding_policy: dns_authority.rebinding_policy.clone(),
            resolvers: dns_authority.resolvers.clone(),
            hostnames,
            keyset_id: Some(keyset_id),
            issuer_kid: Some(resolver_kid),
            policy_digest: Some(policy_digest),
            correlation_id,
            source: "cellos-supervisor".into(),
            cell_id: cell_id.to_string(),
            run_id: run_id.to_string(),
            // SEC-21 Phase 3h — pick the first resolver that has opted
            // into DNSSEC validation; the trust anchors are loaded once
            // per ticker (above) and stamped into every emitted
            // `dns_authority_dnssec_failed` event. When no resolver has
            // a `dnssec` block, both fields are `None` → P3a/P3e
            // behaviour is preserved exactly.
            dnssec_policy: dnssec_policy_for_ticker.clone(),
            trust_anchors: trust_anchors_for_ticker.clone(),
            // Production validating closure mirrors the unvalidated one
            // below: same `Handle::block_on` discipline, same
            // `pick_resolver_socket_addr` upstream, same timeout. Built
            // unconditionally so the ticker's `dnssec_active` predicate
            // (which gates on `dnssec_policy.is_some() &&
            // validated_resolver.is_some()`) only flips when the spec
            // opted in. Trust anchors are cloned per-call inside the
            // closure (cheap — Vec<u8> + String) so the `Send + Sync`
            // bound on `SharedValidatedResolverFn` is satisfied.
            validated_resolver: Some(std::sync::Arc::new(
                move |hostname: &str| -> std::io::Result<
                    cellos_supervisor::resolver_refresh::ValidatedResolvedAnswer,
                > {
                    let runtime_handle = tokio::runtime::Handle::current();
                    let anchors = cellos_supervisor::resolver_refresh::TrustAnchors::iana_default();
                    runtime_handle.block_on(
                        cellos_supervisor::resolver_refresh::resolve_with_ttl_validated(
                            hostname,
                            upstream_addr,
                            resolver_timeout,
                            &anchors,
                        ),
                    )
                },
            )),
        };

        // P3a unvalidated closure stays — used when `dnssec_policy is
        // None` (every resolver in the spec). The ticker's `dnssec_active`
        // predicate decides which closure runs each tick.
        let resolver: cellos_supervisor::resolver_refresh::ticker::SharedResolverFn =
            std::sync::Arc::new(move |hostname: &str| -> std::io::Result<
                cellos_supervisor::resolver_refresh::ResolvedAnswer,
            > {
                let runtime_handle = tokio::runtime::Handle::current();
                runtime_handle.block_on(
                    cellos_supervisor::resolver_refresh::resolve_with_ttl(
                        hostname,
                        upstream_addr,
                        resolver_timeout,
                    ),
                )
            });

        let emitter: std::sync::Arc<
            dyn cellos_supervisor::resolver_refresh::ticker::DriftEmitter,
        > = std::sync::Arc::new(
            cellos_supervisor::resolver_refresh::sink_emitter::EventSinkEmitter::capture_current(
                self.event_sink.clone(),
                self.jsonl_sink.clone(),
            ),
        );

        let handle = cellos_supervisor::resolver_refresh::ticker::spawn_continuous_ticker(
            cfg, emitter, resolver,
        );
        tracing::info!(
            target: "cellos.supervisor.resolver_refresh.ticker",
            cell_id = %cell_id,
            run_id = %run_id,
            interval_secs = interval.as_secs(),
            "SEC-21 Phase 2 continuous resolver-refresh ticker spawned"
        );
        Some(handle)
    }

    /// SEC-22 Phase 1 L7 gate observability — emits extra `l7_egress_decision`
    /// events recording the policy's hostname allowlist commitment AND
    /// per-port-53 egress rule whose destination isn't a declared resolver.
    ///
    /// Disabled unless `CELLOS_L7_GATE_OBSERVABILITY=1`. The kernel-level
    /// direct-DNS backstop in [`generate_nft_ruleset`] runs regardless of this
    /// flag — this method only governs the *observable signal*.
    ///
    /// scope: this method is **observability-only at L7** because real SNI /
    /// Host / CONNECT-method enforcement requires a hostname-aware proxy
    /// (handled by the SNI proxy). The events emitted here record the
    /// *declared policy commitment* so audit pipelines can assert "the policy
    /// was supposed to gate `example.com`" alongside the per-flow allow/deny
    /// signal from the SNI proxy itself.
    ///
    /// Emitted reason codes:
    /// - `hostname_allowlist_evaluated` — one per
    ///   `dnsAuthority.hostnameAllowlist[]` entry; `action: "allow"`.
    /// - `workload_direct_dns_blocked` — one per `egressRules[]` entry whose
    ///   port is 53 and whose host is *not* a declared do53 resolver, when
    ///   `dnsAuthority.blockDirectWorkloadDns == true`; `action: "deny"`.
    pub(crate) async fn maybe_emit_l7_gate_observability(
        &self,
        doc: &ExecutionCellDocument,
        cell_id: &str,
        run_id: &str,
        phase_err: &mut Option<anyhow::Error>,
    ) {
        if !l7_gate_observability_enabled() {
            return;
        }
        let Some(dns_authority) = doc.spec.authority.dns_authority.as_ref() else {
            return;
        };

        let policy_digest = canonical_spec_hash(&doc.spec);
        let keyset_id = std::env::var("CELLOS_TRUST_KEYSET_ID")
            .unwrap_or_else(|_| TRUST_PLANE_BUILTIN_KEYSET_ID.to_string());
        let l7_kid = std::env::var("CELLOS_TRUST_L7_KID")
            .unwrap_or_else(|_| TRUST_PLANE_BUILTIN_L7_KID.to_string());

        // Per-hostname-allowlist evaluation event. Records the policy
        // commitment for each declared hostname; Phase 2 SNI-aware proxy will
        // turn this into an actual L7 allow/deny decision per request.
        for (idx, hostname) in dns_authority.hostname_allowlist.iter().enumerate() {
            let decision_id = uuid::Uuid::new_v4().to_string();
            let rule_ref = format!("authority.dnsAuthority.hostnameAllowlist[{idx}]");
            match cellos_core::observability_l7_egress_decision_data_v1(
                &doc.spec,
                cell_id,
                Some(run_id),
                decision_id.as_str(),
                "allow",
                hostname.as_str(),
                policy_digest.as_str(),
                keyset_id.as_str(),
                l7_kid.as_str(),
                "hostname_allowlist_evaluated",
                Some(rule_ref.as_str()),
                None,
            )
            .context("build l7_egress_decision (hostname_allowlist_evaluated) data")
            {
                Err(e) => {
                    if phase_err.is_none() {
                        *phase_err = Some(e);
                    }
                }
                Ok(data) => {
                    let ev = cloud_event(
                        "dev.cellos.events.cell.observability.v1.l7_egress_decision",
                        data,
                    );
                    if let Err(e) = self.emit(&ev).await {
                        if phase_err.is_none() {
                            *phase_err = Some(e);
                        }
                    }
                }
            }
        }

        // Direct-DNS-block deny events. For each declared egress rule on port
        // 53, check whether the destination matches a declared do53 resolver;
        // if not, this is the policy demanding kernel-level DNS containment.
        if dns_authority.block_direct_workload_dns {
            if let Some(rules) = doc.spec.authority.egress_rules.as_deref() {
                for (idx, rule) in rules.iter().enumerate() {
                    if rule.port != 53 {
                        continue;
                    }
                    if egress_destination_matches_declared_resolver(
                        rule.host.as_str(),
                        rule.port,
                        dns_authority.resolvers.as_slice(),
                    ) {
                        continue;
                    }
                    let decision_id = uuid::Uuid::new_v4().to_string();
                    let rule_ref = format!("authority.egressRules[{idx}]");
                    match cellos_core::observability_l7_egress_decision_data_v1(
                        &doc.spec,
                        cell_id,
                        Some(run_id),
                        decision_id.as_str(),
                        "deny",
                        rule.host.as_str(),
                        policy_digest.as_str(),
                        keyset_id.as_str(),
                        l7_kid.as_str(),
                        "workload_direct_dns_blocked",
                        Some(rule_ref.as_str()),
                        None,
                    )
                    .context("build l7_egress_decision (workload_direct_dns_blocked) data")
                    {
                        Err(e) => {
                            if phase_err.is_none() {
                                *phase_err = Some(e);
                            }
                        }
                        Ok(data) => {
                            let ev = cloud_event(
                                "dev.cellos.events.cell.observability.v1.l7_egress_decision",
                                data,
                            );
                            if let Err(e) = self.emit(&ev).await {
                                if phase_err.is_none() {
                                    *phase_err = Some(e);
                                }
                            }
                        }
                    }
                }
            }
        }
    }
}