cellos-supervisor 0.5.1

//! Proxy activation methods split out from `supervisor.rs` (P0-2).
//!
//! Hosts an additional `impl Supervisor` block carrying the five
//! activation-time methods:
//!
//! - `build_dns_proxy_activation` — SEAM-1 / L2-04 Phase 2b plan builder.
//! - `emit_dns_proxy_spawn_failure` — Phase 2b spawn-failure audit event.
//! - `build_sni_proxy_activation` — SEC-22 Phase 2 plan builder.
//! - `emit_sni_proxy_spawn_failure` — Phase 2 spawn-failure audit event.
//! - `emit_per_flow_decision_events` — FC-38 Phase 1 post-run counter →
//!   `network_flow_decision` fan-out.
//!
//! These were collocated with `Supervisor::run` in `supervisor.rs`; moving
//! them into their own module keeps `supervisor.rs` focused on lifecycle
//! orchestration. Pure code-move — no logic changes.

use cellos_core::{
    canonical_spec_hash, EgressRule, ExecutionCellDocument, TRUST_PLANE_BUILTIN_KEYSET_ID,
    TRUST_PLANE_BUILTIN_L7_KID, TRUST_PLANE_BUILTIN_RESOLVER_KID,
};

use crate::network_policy::parse_do53_endpoint;
use crate::supervisor::{DnsProxyActivation, SniProxyActivation, Supervisor};
use crate::supervisor_helpers::{
    cloud_event, dns_proxy_enabled, sni_proxy_bind_addr, sni_proxy_enabled,
    sni_proxy_upstream_override,
};

impl Supervisor {
    /// SEAM-1 / L2-04 DNS proxy activation — Phase 2b.
    ///
    /// Build a [`DnsProxyActivation`] when the activation predicate holds, so
    /// `linux_run_cell_command_isolated` can `setns(2)` the proxy thread into
    /// the cell's netns the moment the workload child PID becomes known.
    ///
    /// **Activation predicate** (all three must hold):
    /// 1. `CELLOS_DNS_PROXY=1` is set in the supervisor environment.
    /// 2. `spec.authority.dnsAuthority.hostnameAllowlist` is non-empty.
    /// 3. At least one resolver in `spec.authority.dnsAuthority.resolvers[]`
    ///    has `protocol ∈ {do53-udp, do53-tcp}` AND a parseable `host:port`
    ///    endpoint.
    ///
    /// Returns `None` when the predicate fails OR when running on a platform
    /// other than Linux (the only platform where `setns(2)` + per-cell
    /// netns isolation is supported). On non-Linux the supervisor logs once
    /// and proceeds without DNS proxy enforcement — the Phase 2a stub event
    /// is intentionally not preserved because Phase 2b's audit signal is
    /// per-real-query events from the live proxy, not a one-shot status.
    ///
    /// The returned plan owns clones of `cellos-core` types so it survives
    /// crossing the `run_cell_command` → `linux_run_cell_command_isolated`
    /// async/sync boundary without lifetime entanglement.
    pub(crate) fn build_dns_proxy_activation(
        &self,
        doc: &ExecutionCellDocument,
        cell_id: &str,
        run_id: &str,
    ) -> Option<DnsProxyActivation> {
        if !dns_proxy_enabled() {
            return None;
        }
        let dns_authority = doc.spec.authority.dns_authority.as_ref()?;
        if dns_authority.hostname_allowlist.is_empty() {
            return None;
        }
        // Choose the first do53-{udp,tcp} resolver with a parseable endpoint.
        // scope: bind a single UDP listener on that endpoint; the proxy
        // then forwards over the same do53-udp upstream (per the dataplane's
        // do53-only contract). DoH/DoT/DoQ remain residual risk per
        // `docs/sec22-residual-risk.md`.
        let (resolver, host, port) = dns_authority.resolvers.iter().find_map(|r| {
            if !matches!(
                r.protocol,
                cellos_core::DnsResolverProtocol::Do53Udp
                    | cellos_core::DnsResolverProtocol::Do53Tcp
            ) {
                return None;
            }
            let (h, p) = parse_do53_endpoint(r)?;
            Some((r, h, p))
        })?;

        // Parse `host:port` to a SocketAddr — scope: host MUST be a literal
        // IP because the workload's `/etc/resolv.conf` cannot rely on host-DNS
        // to resolve its own resolver. Hostname-form resolvers fall back to
        // None here and the supervisor proceeds without proxy enforcement —
        // the operator gets a tracing warn so they can fix the spec.
        let listen_addr: std::net::SocketAddr = match format!("{host}:{port}").parse() {
            Ok(addr) => addr,
            Err(e) => {
                tracing::warn!(
                    target: "cellos.supervisor.dns_proxy",
                    cell_id = %cell_id,
                    run_id = %run_id,
                    resolver_id = %resolver.resolver_id,
                    endpoint = %resolver.endpoint,
                    error = %e,
                    "do53 resolver endpoint did not parse as IP:port — DNS proxy NOT activated"
                );
                return None;
            }
        };
        // scope: forward to the same declared resolver address. Real-world
        // deployments separate listen-side and upstream-side resolvers — the
        // contract layer (`DnsAuthority::resolvers[]`) admits that, and the
        // supervisor's plan struct keeps them as distinct fields so the
        // forwarding-target choice can evolve without churning the spawn API.
        //
        // Test/operator escape hatch: `CELLOS_DNS_PROXY_UPSTREAM_OVERRIDE=ip:port`
        // overrides the upstream address while leaving the listener on the
        // declared resolver endpoint. Used by the e2e integration test to
        // run a stub upstream on a different port without extending the
        // contract; in production deployments the override is ignored unless
        // explicitly set.
        let upstream_addr = match std::env::var("CELLOS_DNS_PROXY_UPSTREAM_OVERRIDE") {
            Ok(raw) => match raw.trim().parse::<std::net::SocketAddr>() {
                Ok(addr) => {
                    tracing::info!(
                        target: "cellos.supervisor.dns_proxy",
                        cell_id = %cell_id,
                        run_id = %run_id,
                        override_addr = %addr,
                        "CELLOS_DNS_PROXY_UPSTREAM_OVERRIDE applied"
                    );
                    addr
                }
                Err(e) => {
                    tracing::warn!(
                        target: "cellos.supervisor.dns_proxy",
                        error = %e,
                        raw = %raw,
                        "CELLOS_DNS_PROXY_UPSTREAM_OVERRIDE parse failed — falling back to listen_addr"
                    );
                    listen_addr
                }
            },
            Err(_) => listen_addr,
        };

        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 3h.1 — construct the dataplane DNSSEC validator
        // from the cell's DnsAuthority. `from_authority` returns:
        //   Ok(None)  when mode = off (no dnssec block on the first
        //             resolver, OR validate=false). Hot path unchanged.
        //   Ok(Some)  when mode = require/best_effort. Validator runs on
        //             every allowlisted query in the proxy.
        //   Err(...)  when an operator-supplied trust_anchors_path is
        //             missing/symlinked/oversized — surfaces at activation
        //             rather than per-query, mirroring the W6 SEC-25 /
        //             P3h discipline. We log and FAIL the activation if
        //             the operator asked for DNSSEC but mis-wired it,
        //             rather than silently downgrading.
        let dnssec_validator =
            match crate::dns_proxy::dnssec::DataplaneDnssecValidator::from_authority(dns_authority)
            {
                Ok(opt) => opt.map(std::sync::Arc::new),
                Err(e) => {
                    tracing::error!(
                        target: "cellos.supervisor.dns_proxy",
                        cell_id = %cell_id,
                        run_id = %run_id,
                        error = %e,
                        "dataplane DNSSEC validator construction FAILED — refusing to activate proxy without the requested DNSSEC enforcement"
                    );
                    return None;
                }
            };

        // Map the spec-declared resolver protocol onto
        // the proxy's UpstreamTransport. CELLOS_DNS_UPSTREAM_TRANSPORT
        // env var, when set, overrides the spec-derived selector — used
        // by integration tests to exercise the DoT/DoH/DoQ paths against
        // a stub upstream without re-spinning the spec.
        let spec_transport = match resolver.protocol {
            cellos_core::DnsResolverProtocol::Do53Udp => {
                crate::dns_proxy::upstream::UpstreamTransport::Do53Udp
            }
            cellos_core::DnsResolverProtocol::Do53Tcp => {
                crate::dns_proxy::upstream::UpstreamTransport::Do53Tcp
            }
            cellos_core::DnsResolverProtocol::Dot => {
                crate::dns_proxy::upstream::UpstreamTransport::Dot
            }
            cellos_core::DnsResolverProtocol::Doh => {
                crate::dns_proxy::upstream::UpstreamTransport::Doh
            }
            cellos_core::DnsResolverProtocol::Doq => {
                crate::dns_proxy::upstream::UpstreamTransport::Doq
            }
        };
        // A6 — operator override resolved from env. Two names are accepted
        // (in priority order):
        //   1. `CELLOS_DNS_UPSTREAM_PROTOCOL` — the contract-aligned name
        //      (matches `DnsAuthority::resolvers[].protocol`). Preferred.
        //   2. `CELLOS_DNS_UPSTREAM_TRANSPORT` — the legacy name, preserved
        //      so existing operator scripts and integration tests keep
        //      working byte-for-byte.
        let (env_var_name, raw_env) = match (
            std::env::var("CELLOS_DNS_UPSTREAM_PROTOCOL"),
            std::env::var("CELLOS_DNS_UPSTREAM_TRANSPORT"),
        ) {
            (Ok(raw), _) if !raw.trim().is_empty() => ("CELLOS_DNS_UPSTREAM_PROTOCOL", Some(raw)),
            (_, Ok(raw)) if !raw.trim().is_empty() => ("CELLOS_DNS_UPSTREAM_TRANSPORT", Some(raw)),
            _ => ("CELLOS_DNS_UPSTREAM_PROTOCOL", None),
        };
        let transport = match raw_env {
            Some(raw) => match crate::dns_proxy::upstream::UpstreamTransport::parse(&raw) {
                Some(t) => {
                    tracing::info!(
                        target: "cellos.supervisor.dns_proxy",
                        spec = %spec_transport.label(),
                        env = %t.label(),
                        var = %env_var_name,
                        "DNS upstream transport env override applied — overriding spec resolver protocol"
                    );
                    t
                }
                None => {
                    tracing::warn!(
                        target: "cellos.supervisor.dns_proxy",
                        raw = %raw,
                        var = %env_var_name,
                        "DNS upstream transport env override did not parse — falling back to spec resolver protocol"
                    );
                    spec_transport
                }
            },
            None => spec_transport,
        };

        // A6 / DNS-DOH-2 / DNS-DOQ-2 — populate per-transport extras when a
        // non-do53 transport is in play. Reads
        // `CELLOS_DNS_UPSTREAM_DOT_SERVER` / `_DOT_PORT` / `_DOT_SNI`,
        // `CELLOS_DNS_UPSTREAM_DOH_URL`, and
        // `CELLOS_DNS_UPSTREAM_DOQ_SERVER` / `_DOQ_PORT`. For the do53 paths
        // extras are inert (no-cost defaults), so we gate the env read on
        // whichever transport is active to keep noise out of operator logs
        // that don't care.
        let upstream_extras = match transport {
            crate::dns_proxy::upstream::UpstreamTransport::Dot => {
                let extras = crate::dns_proxy::upstream::UpstreamExtras::from_env();
                tracing::info!(
                    target: "cellos.supervisor.dns_proxy",
                    dot_server = extras.dot_server.as_deref().unwrap_or("<spec>"),
                    dot_port = extras.dot_port.unwrap_or(853),
                    dot_sni = extras.dot_sni.as_deref().unwrap_or("<ip-literal>"),
                    "DoT upstream extras resolved from env"
                );
                extras
            }
            crate::dns_proxy::upstream::UpstreamTransport::Doh => {
                let extras = crate::dns_proxy::upstream::UpstreamExtras::from_env();
                tracing::info!(
                    target: "cellos.supervisor.dns_proxy",
                    doh_url = extras.doh_url.as_deref().unwrap_or("<default>"),
                    "DoH upstream extras resolved from env"
                );
                extras
            }
            crate::dns_proxy::upstream::UpstreamTransport::Doq => {
                let extras = crate::dns_proxy::upstream::UpstreamExtras::from_env();
                tracing::info!(
                    target: "cellos.supervisor.dns_proxy",
                    doq_server = extras.doq_server.as_deref().unwrap_or("<default>"),
                    doq_port = extras.doq_port.unwrap_or(853),
                    "DoQ upstream extras resolved from env"
                );
                extras
            }
            _ => crate::dns_proxy::upstream::UpstreamExtras::default(),
        };

        let cfg = crate::dns_proxy::DnsProxyConfig {
            bind_addr: listen_addr,
            upstream_addr,
            hostname_allowlist: dns_authority.hostname_allowlist.clone(),
            allowed_query_types: dns_authority.allowed_query_types.clone(),
            cell_id: cell_id.to_string(),
            run_id: run_id.to_string(),
            policy_digest: Some(policy_digest),
            keyset_id: Some(keyset_id),
            issuer_kid: Some(resolver_kid),
            correlation_id,
            upstream_resolver_id: resolver.resolver_id.clone(),
            upstream_timeout: std::time::Duration::from_millis(400),
            // A5 — workload-facing TCP/53 listener idle timeout. 30s
            // mirrors the UDP SERVFAIL ceiling so a stuck workload
            // cannot pin a TCP worker thread for longer than the UDP
            // equivalent.
            tcp_idle_timeout: std::time::Duration::from_secs(30),
            dnssec_validator,
            transport,
            upstream_extras,
        };

        tracing::info!(
            target: "cellos.supervisor.dns_proxy",
            cell_id = %cell_id,
            run_id = %run_id,
            resolver_id = %resolver.resolver_id,
            listen = %listen_addr,
            allowlist_count = dns_authority.hostname_allowlist.len(),
            "SEAM-1 DNS proxy activation plan built — spawn deferred to netns child"
        );

        Some(DnsProxyActivation {
            cfg,
            listen_addr,
            upstream_addr,
            resolver_id: resolver.resolver_id.clone(),
        })
    }

    /// Emit a single `dns_query` event with `reasonCode: upstream_failure`
    /// when the netns spawn failed. Preserves the audit trail W3 designed
    /// for the activation stub — without this, a spawn failure (e.g. EPERM
    /// on `setns`) would silently leave the cell unprotected.
    pub(crate) async fn emit_dns_proxy_spawn_failure(
        &self,
        doc: &ExecutionCellDocument,
        cell_id: &str,
        run_id: &str,
        resolver_id: Option<&str>,
        error_text: &str,
    ) {
        let observed_at = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true);
        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());
        // Truncate the error text to a sane length so a libc message storm
        // can't blow the JSONL line size.
        let truncated = if error_text.len() > 200 {
            format!("{}…", &error_text[..200])
        } else {
            error_text.to_string()
        };
        let event = cellos_core::DnsQueryEvent {
            schema_version: "1.0.0".into(),
            cell_id: cell_id.to_string(),
            run_id: run_id.to_string(),
            query_id: uuid::Uuid::new_v4().to_string(),
            query_name: "(spawn-error)".into(),
            query_type: cellos_core::DnsQueryType::A,
            decision: cellos_core::DnsQueryDecision::Deny,
            reason_code: cellos_core::DnsQueryReasonCode::UpstreamFailure,
            response_rcode: None,
            upstream_resolver_id: resolver_id.map(str::to_string),
            upstream_latency_ms: None,
            response_target_count: None,
            keyset_id: Some(keyset_id),
            issuer_kid: Some(resolver_kid),
            policy_digest: Some(policy_digest),
            correlation_id,
            observed_at: observed_at.clone(),
        };
        tracing::warn!(
            target: "cellos.supervisor.dns_proxy",
            cell_id = %cell_id,
            run_id = %run_id,
            error = %truncated,
            "SEAM-1 DNS proxy spawn failed — emitting upstream_failure audit event"
        );
        match cellos_core::cloud_event_v1_dns_query("cellos-supervisor", &observed_at, &event) {
            Ok(env) => {
                if let Err(e) = self.emit(&env).await {
                    tracing::warn!(error = %e, "dns_query spawn-error event emit failed");
                }
            }
            Err(e) => {
                tracing::warn!(error = %e, "dns_query spawn-error event build failed");
            }
        }
    }

    /// SEC-22 Phase 2 — resolve the SNI-aware L7 proxy activation plan.
    ///
    /// **Activation predicate** (all three must hold):
    /// 1. `CELLOS_SNI_PROXY=1` is set in the supervisor environment.
    /// 2. `spec.authority.dnsAuthority.hostnameAllowlist` is non-empty.
    /// 3. The platform is Linux (the only platform where `setns(2)` is
    ///    available; the activation is built on macOS for unit tests but
    ///    the spawn path is gated behind `cfg!(target_os = "linux")`).
    ///
    /// Returns `None` when the predicate fails. The bind address comes from
    /// `CELLOS_SNI_PROXY_BIND` (default `127.0.0.1:8443`); the upstream
    /// address comes from `CELLOS_SNI_PROXY_UPSTREAM_OVERRIDE` (test-only)
    /// and falls back to the bind address — in production this would be a
    /// transparent next-hop egress NAT.
    pub(crate) fn build_sni_proxy_activation(
        &self,
        doc: &ExecutionCellDocument,
        cell_id: &str,
        run_id: &str,
    ) -> Option<SniProxyActivation> {
        if !sni_proxy_enabled() {
            return None;
        }
        let dns_authority = doc.spec.authority.dns_authority.as_ref()?;
        if dns_authority.hostname_allowlist.is_empty() {
            return None;
        }
        let listen_addr = sni_proxy_bind_addr();
        let upstream_addr = sni_proxy_upstream_override().unwrap_or(listen_addr);

        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());
        let correlation_id = doc
            .spec
            .correlation
            .as_ref()
            .and_then(|c| c.correlation_id.clone());

        let cdn_providers = doc
            .spec
            .authority
            .cdn_authority
            .as_ref()
            .map(|c| c.providers.clone())
            .unwrap_or_default();

        let cfg = crate::sni_proxy::SniProxyConfig {
            bind_addr: listen_addr,
            upstream_addr,
            hostname_allowlist: dns_authority.hostname_allowlist.clone(),
            cdn_providers,
            cell_id: cell_id.to_string(),
            run_id: run_id.to_string(),
            policy_digest: Some(policy_digest),
            keyset_id: Some(keyset_id),
            issuer_kid: Some(l7_kid),
            correlation_id,
            upstream_resolver_id: "sni-proxy-l7".to_string(),
            peek_timeout: std::time::Duration::from_millis(500),
        };

        tracing::info!(
            target: "cellos.supervisor.sni_proxy",
            cell_id = %cell_id,
            run_id = %run_id,
            listen = %listen_addr,
            upstream = %upstream_addr,
            allowlist_count = dns_authority.hostname_allowlist.len(),
            "SEC-22 Phase 2 SNI proxy activation plan built — spawn deferred to netns child"
        );

        Some(SniProxyActivation { cfg, listen_addr })
    }

    /// SEC-22 Phase 2 — emit a single `l7_egress_decision` event with
    /// `reasonCode: l7_unknown_protocol` when the proxy spawn failed, so
    /// the audit trail records the gap.
    ///
    /// Wired in W9 finalize: `linux_run_cell_command_isolated` propagates
    /// `sni_proxy_spawn_err` out of its `spawn_blocking` closure (5-tuple
    /// Ok / 4-tuple Err) into `Supervisor::run`'s 7-tuple destructure, and
    /// this method is invoked when the value is `Some`. Mirrors
    /// `emit_dns_proxy_spawn_failure` above.
    pub(crate) async fn emit_sni_proxy_spawn_failure(
        &self,
        doc: &ExecutionCellDocument,
        cell_id: &str,
        run_id: &str,
        error_text: &str,
    ) {
        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());
        let truncated = if error_text.len() > 200 {
            format!("{}…", &error_text[..200])
        } else {
            error_text.to_string()
        };
        let decision_id = uuid::Uuid::new_v4().to_string();
        match cellos_core::observability_l7_egress_decision_data_v1(
            &doc.spec,
            cell_id,
            Some(run_id),
            decision_id.as_str(),
            "deny",
            "(spawn-error)",
            policy_digest.as_str(),
            keyset_id.as_str(),
            l7_kid.as_str(),
            "l7_unknown_protocol",
            Some(truncated.as_str()),
            None,
        ) {
            Ok(data) => {
                let ev = cloud_event(
                    "dev.cellos.events.cell.observability.v1.l7_egress_decision",
                    data,
                );
                if let Err(e) = self.emit(&ev).await {
                    tracing::warn!(error = %e, "sni_proxy spawn-error event emit failed");
                }
            }
            Err(e) => {
                tracing::warn!(error = %e, "sni_proxy spawn-error event build failed");
            }
        }
    }

    /// FC-38 Phase 1: emit one `network_flow_decision` CloudEvent per scraped
    /// nft counter row. Skips allow rules with `packet_count == 0` (operator
    /// already saw them in `network_policy`); always emits deny rules so a
    /// "policy applied, nothing matched" state is visible in the audit
    /// stream.
    ///
    /// `egress_rules` is used to attribute ordinary egress accept rows back
    /// to their declaration site (`authority.egressRules[i]`) by matching
    /// the row's `dst_addr` + `dst_port` + protocol against each declared
    /// rule. When no match is found, the event still emits without a
    /// `nft_rule_ref` (the operator can correlate via `dst_addr`/`dst_port`).
    pub(crate) async fn emit_per_flow_decision_events(
        &self,
        doc: &ExecutionCellDocument,
        cell_id: &str,
        run_id: &str,
        egress_rules: &[EgressRule],
        rows: &[crate::nft_counters::NftCounterRow],
    ) {
        let observed_at = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true);
        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());

        for row in rows {
            let classified = crate::nft_counters::classify_rule_repr(&row.rule_repr);
            let outcome = match classified.decision {
                crate::nft_counters::NftDecision::Allow => {
                    cellos_core::NetworkFlowDecisionOutcome::Allow
                }
                crate::nft_counters::NftDecision::Deny => {
                    cellos_core::NetworkFlowDecisionOutcome::Deny
                }
            };
            // Skip allow rules with no traffic — declared-policy declaration
            // is already captured in the `network_policy` event. Honest scope
            // call-out: deny rules with 0 counters STILL emit, because an
            // operator wants to know "nothing was blocked" is the actual
            // observed state, not a missing event.
            if matches!(outcome, cellos_core::NetworkFlowDecisionOutcome::Allow)
                && row.packet_count == 0
            {
                continue;
            }
            // Best-effort: tie an ordinary egress accept row back to the
            // declared `authority.egressRules[i]` index by matching daddr +
            // dport + protocol. Resolver-allow / workload-DNS-block /
            // default-drop rows already carry their synthetic refs from the
            // classifier.
            let mut nft_rule_ref = classified.nft_rule_ref.clone();
            if classified.reason_code == "nft_egress_rule_match" && nft_rule_ref.is_none() {
                if let (Some(addr), Some(port)) =
                    (classified.dst_addr.as_deref(), classified.dst_port)
                {
                    let proto = classified.protocol.as_deref();
                    for (idx, rule) in egress_rules.iter().enumerate() {
                        let rule_proto = rule.protocol.as_deref().unwrap_or("tcp");
                        if rule.host == addr
                            && rule.port == port
                            && proto.is_some_and(|p| p == rule_proto)
                        {
                            nft_rule_ref = Some(format!("authority.egressRules[{idx}]"));
                            break;
                        }
                    }
                }
            }
            let payload = cellos_core::NetworkFlowDecision {
                schema_version: "1.0.0".into(),
                cell_id: cell_id.to_string(),
                run_id: run_id.to_string(),
                decision_id: uuid::Uuid::new_v4().to_string(),
                direction: cellos_core::NetworkFlowDirection::Egress,
                decision: outcome,
                reason_code: classified.reason_code.to_string(),
                nft_rule_ref,
                dst_addr: classified.dst_addr.clone(),
                dst_port: classified.dst_port,
                protocol: classified.protocol.clone(),
                packet_count: Some(row.packet_count),
                byte_count: Some(row.byte_count),
                policy_digest: Some(policy_digest.clone()),
                keyset_id: Some(keyset_id.clone()),
                issuer_kid: Some(resolver_kid.clone()),
                correlation_id: correlation_id.clone(),
                observed_at: observed_at.clone(),
            };
            match cellos_core::cloud_event_v1_network_flow_decision(
                "cellos-supervisor",
                &observed_at,
                &payload,
            ) {
                Ok(env) => {
                    if let Err(e) = self.emit(&env).await {
                        tracing::warn!(
                            target: "cellos.supervisor.per_flow_enforcement",
                            error = %e,
                            "network_flow_decision event emit failed (FC-38)"
                        );
                    }
                }
                Err(e) => {
                    tracing::warn!(
                        target: "cellos.supervisor.per_flow_enforcement",
                        error = %e,
                        "network_flow_decision event build failed (FC-38)"
                    );
                }
            }
        }
    }
}