cellos-supervisor 0.5.1

//! SEC-22 Phase 2 SNI-aware egress proxy.
//!
//! Inspects the first bytes of every TCP connection a workload opens and
//! enforces `dnsAuthority.hostnameAllowlist` at L7 by parsing either:
//!
//! - the TLS ClientHello's `server_name` extension ([`sni`]), or
//! - the HTTP/1.x `Host` header on cleartext requests ([`http`]),
//!
//! and matching the extracted hostname against the shared
//! [`cellos_core::hostname_allowlist::matches_allowlist`] helper. Allowed
//! flows are forwarded to a declared upstream via
//! `tokio::io::copy_bidirectional` (the proxy NEVER terminates TLS — the
//! ClientHello bytes that produced the SNI decision are written through to
//! the upstream as the first bytes of the forwarded stream). Denied flows
//! are dropped (TLS) or answered with `HTTP/1.1 403 Forbidden` (cleartext
//! HTTP). Each decision emits exactly one
//! `dev.cellos.events.cell.observability.v1.l7_egress_decision` CloudEvent
//! carrying one of the eight Phase 2 reason codes:
//!
//! - `l7_sni_allowlist_match`           — TLS allow
//! - `l7_sni_allowlist_miss`            — TLS deny (SNI not in allowlist)
//! - `l7_sni_missing`                   — TLS deny (no SNI in ClientHello)
//! - `l7_http_host_allowlist_match`     — HTTP allow
//! - `l7_http_host_allowlist_miss`      — HTTP deny (Host not in allowlist)
//! - `l7_http_host_missing`             — HTTP deny (no Host header)
//! - `l7_unknown_protocol`              — first bytes neither TLS nor HTTP/1.x
//! - `l7_peek_timeout`                  — workload sent no bytes within the
//!   peek window (defensive shutdown)
//!
//! ### Phase 3a (h2c) — additional reason codes
//!
//! - `l7_h2_authority_allowlist_match`  — h2c HEADERS-frame `:authority`
//!   matched the allowlist; bytes forwarded.
//! - `l7_h2_authority_allowlist_miss`   — h2c HEADERS-frame `:authority`
//!   did NOT match the allowlist; deny + GOAWAY (PROTOCOL_ERROR).
//! - `l7_h2_authority_missing`          — well-formed h2c frames but no
//!   `:authority` pseudo-header in the first HEADERS frame.
//! - `l7_h2_unparseable_headers`        — HPACK decode failure,
//!   CONTINUATION-fragmented HEADERS, oversized frame, or any other h2
//!   parse error Phase 3a does not handle. Deny + GOAWAY.
//!
//! The reason-code list extends the existing
//! `cell-observability-l7-egress-decision-v1.schema.json` `reasonCode`
//! description without breaking shape — only new string values are added.
//!
//! ## Honest scope
//!
//! - **TLS not terminated.** SNI ↔ Host alignment cannot be checked in
//!   Phase 2; that requires MITM termination and a CA the workload trusts.
//!   Fronting via mismatched SNI vs. Host on the same TLS session remains a
//!   Phase 3 residual risk per [`docs/sec22-residual-risk.md`].
//! - **HTTP/2 cleartext (h2c) inspected (Phase 3a).** When a workload
//!   sends the canonical 24-byte `PRI * HTTP/2.0` preface, the proxy
//!   peels it + the optional SETTINGS frame and parses the first HEADERS
//!   frame's `:authority` pseudo-header through the static-table-only
//!   HPACK decoder in [`h2`]. The same `dnsAuthority.hostnameAllowlist`
//!   that gates SNI + HTTP/1.x Host gates h2c. Allow path forwards the
//!   buffered preface + HEADERS verbatim; deny path emits an HTTP/2
//!   GOAWAY (PROTOCOL_ERROR) and closes.
//! - **HTTP/2 over TLS still residual.** Inspecting HEADERS inside an
//!   encrypted stream requires TLS termination — Phase 4. HTTP/3 / QUIC
//!   are also out of scope.
//! - **DoH residual.** When an operator allowlists a hostname that itself
//!   hosts DoH, the workload can tunnel DNS inside an allowed flow. This is
//!   an operator-policy residual; the proxy enforces what the allowlist
//!   declares, not the semantic intent.
//!
//! ## Module layout
//!
//! - [`sni`] — pure-byte ClientHello SNI extractor.
//! - [`http`] — HTTP/1.x request-line + Host extractor.
//! - [`run_one_shot`] (in this file) — async per-connection accept/probe/
//!   forward loop, platform-neutral.
//! - [`spawn`] — Linux-only `setns(2)` helper that places `run_one_shot`'s
//!   listener inside the cell's netns; mirrors [`crate::dns_proxy::spawn`].
//!
//! Reuses the shared
//! [`cellos_core::hostname_allowlist::matches_allowlist`] helper that the
//! W8 commit extracted from the W3 SEAM-1 dataplane proxy.

pub mod h2;
pub mod http;
pub mod sni;
pub mod spawn;

use std::net::SocketAddr;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;

use cellos_core::{CdnProvider, CloudEventV1, ExecutionCellSpec};
use serde_json::{json, Map, Value};
use tokio::io::AsyncWriteExt;
use tokio::net::{TcpListener, TcpStream};

/// Maximum bytes the proxy peeks before deciding probe/allow/deny.
///
/// scope: 16 KiB ceiling. The TLS / HTTP/1.x paths always decide well
/// under 4 KiB; the ceiling is sized for h2c HEADERS+CONTINUATION
/// reassembly. RFC 7540 §6.5.2 default `SETTINGS_MAX_FRAME_SIZE` is
/// 16 KiB so the worst-case single HEADERS-with-no-CONTINUATION frame
/// plus the leading SETTINGS frame fits in this buffer; longer header
/// blocks fragment over CONTINUATION up to [`h2::MAX_HEADER_BLOCK_SIZE`]
/// (64 KiB) — at which point the reassembler returns `Ok(None)` (need
/// more bytes) and the peek_timeout path triggers, so the connection is
/// denied with a `peek_timeout` rather than over-buffering memory per
/// connection.
pub const PEEK_BUF_LEN: usize = 16 * 1024;

/// Subset of [`ExecutionCellSpec`] that the proxy needs at run-time. Owned
/// so the supervisor can build it once and pass it through the
/// async-to-sync-thread boundary without lifetime gymnastics (mirrors
/// [`crate::dns_proxy::DnsProxyConfig`]).
#[derive(Debug, Clone)]
pub struct SniProxyConfig {
    /// Address the proxy listener is bound to inside the cell's netns.
    /// Used for diagnostics; the caller passes the actual pre-bound listener.
    pub bind_addr: SocketAddr,
    /// Address the proxy forwards allowed connections to. In a real
    /// deployment this is a transparent next-hop (egress NAT, sidecar) that
    /// re-emits the bytes onto the wire. In the unit tests it is a
    /// localhost echo / handshake stub.
    pub upstream_addr: SocketAddr,
    /// Hostname allowlist (literal or single-leading-`*.` wildcard). Same
    /// shape as [`crate::dns_proxy::DnsProxyConfig::hostname_allowlist`] —
    /// shared matcher in `cellos_core::hostname_allowlist`.
    pub hostname_allowlist: Vec<String>,
    /// CDN providers the workload's spec declares (`spec.authority.cdnAuthority.providers`).
    /// scope: retained for diagnostics and future SNI ↔ Host fronting
    /// detection; the current decision logic does not branch on it.
    pub cdn_providers: Vec<CdnProvider>,
    /// Cell identifier (mirrors `lifecycle.started.cellId`).
    pub cell_id: String,
    /// Run identifier (mirrors `lifecycle.started.runId`).
    pub run_id: String,
    /// Optional `policyDigest` to bind into emitted events.
    pub policy_digest: Option<String>,
    /// Optional `keysetId` to bind into emitted events.
    pub keyset_id: Option<String>,
    /// Optional `issuerKid` to bind into emitted events.
    pub issuer_kid: Option<String>,
    /// Optional `correlationId` to bind into emitted events.
    pub correlation_id: Option<String>,
    /// Resolver / proxy identifier stamped into events for audit trail.
    pub upstream_resolver_id: String,
    /// Maximum time to wait for first bytes from the workload. On timeout
    /// the connection is dropped and one `l7_peek_timeout` event is emitted.
    pub peek_timeout: Duration,
}

/// Sink the proxy uses to publish per-decision CloudEvents. Trait-erased
/// so unit tests can plug in an in-memory collector. Mirrors
/// [`crate::dns_proxy::DnsQueryEmitter`] — the same fire-and-forget shape
/// keeps the per-connection task synchronous from the emit-call's POV.
pub trait L7DecisionEmitter: Send + Sync + 'static {
    /// Publish a single CloudEvent. Implementations should not block.
    fn emit(&self, event: CloudEventV1);
}

/// Aggregate counters returned by [`run_one_shot`] when the loop terminates.
/// Mirrors [`crate::dns_proxy::DnsProxyStats`]; logged at teardown for
/// sanity-checking ratios.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct ProxyStats {
    pub connections_total: u64,
    pub connections_allowed: u64,
    pub connections_denied: u64,
    pub peek_timeouts: u64,
    pub upstream_failures: u64,
}

/// Reason codes for `l7_egress_decision` events emitted by the SNI proxy.
/// All twelve values extend the existing schema's open-ended `reasonCode`
/// description — no schema break.
mod reason_code {
    pub const SNI_ALLOWLIST_MATCH: &str = "l7_sni_allowlist_match";
    pub const SNI_ALLOWLIST_MISS: &str = "l7_sni_allowlist_miss";
    pub const SNI_MISSING: &str = "l7_sni_missing";
    pub const HTTP_HOST_ALLOWLIST_MATCH: &str = "l7_http_host_allowlist_match";
    pub const HTTP_HOST_ALLOWLIST_MISS: &str = "l7_http_host_allowlist_miss";
    pub const HTTP_HOST_MISSING: &str = "l7_http_host_missing";
    pub const UNKNOWN_PROTOCOL: &str = "l7_unknown_protocol";
    pub const PEEK_TIMEOUT: &str = "l7_peek_timeout";
    // ── Phase 3a (h2c HEADERS-frame :authority) ──────────────────────
    pub const H2_AUTHORITY_ALLOWLIST_MATCH: &str = "l7_h2_authority_allowlist_match";
    pub const H2_AUTHORITY_ALLOWLIST_MISS: &str = "l7_h2_authority_allowlist_miss";
    pub const H2_AUTHORITY_MISSING: &str = "l7_h2_authority_missing";
    pub const H2_UNPARSEABLE_HEADERS: &str = "l7_h2_unparseable_headers";
    // ── Phase 3g (full HPACK: dynamic-table state + Huffman) ─────────
    // Differentiate audit trail by HOW the authority was decoded so
    // operators can spot adversaries who deliberately use rare encoder
    // paths (Huffman + dynamic table) to obfuscate.
    pub const H2_AUTHORITY_ALLOWLIST_MATCH_HUFFMAN: &str =
        "l7_h2_authority_allowlist_match_huffman";
    pub const H2_AUTHORITY_ALLOWLIST_MISS_HUFFMAN: &str = "l7_h2_authority_allowlist_miss_huffman";
    pub const H2_AUTHORITY_ALLOWLIST_MATCH_DYNAMIC_INDEXED: &str =
        "l7_h2_authority_allowlist_match_dynamic_indexed";
    pub const H2_AUTHORITY_ALLOWLIST_MISS_DYNAMIC_INDEXED: &str =
        "l7_h2_authority_allowlist_miss_dynamic_indexed";
}

/// Map an HPACK [`h2::AuthorityProvenance`] to the right (allow, miss)
/// reason-code pair so the audit trail differentiates static-table refs
/// (P3c attacker simplicity), dynamic-table refs (Phase 3g — adversary
/// established prior context), and Huffman literals (Phase 3g — adversary
/// used encoder compression).
fn h2_reason_codes_for(provenance: h2::AuthorityProvenance) -> (&'static str, &'static str) {
    match provenance {
        h2::AuthorityProvenance::StaticIndexed | h2::AuthorityProvenance::StaticLiteral => (
            reason_code::H2_AUTHORITY_ALLOWLIST_MATCH,
            reason_code::H2_AUTHORITY_ALLOWLIST_MISS,
        ),
        h2::AuthorityProvenance::DynamicIndexed => (
            reason_code::H2_AUTHORITY_ALLOWLIST_MATCH_DYNAMIC_INDEXED,
            reason_code::H2_AUTHORITY_ALLOWLIST_MISS_DYNAMIC_INDEXED,
        ),
        h2::AuthorityProvenance::Huffman => (
            reason_code::H2_AUTHORITY_ALLOWLIST_MATCH_HUFFMAN,
            reason_code::H2_AUTHORITY_ALLOWLIST_MISS_HUFFMAN,
        ),
    }
}

/// What the first peeked bytes look like.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ProtocolGuess {
    Tls,
    H2c,
    Http1,
    Unknown,
}

fn guess_protocol(buf: &[u8]) -> ProtocolGuess {
    if buf.is_empty() {
        return ProtocolGuess::Unknown;
    }
    if buf[0] == 22 {
        return ProtocolGuess::Tls;
    }
    // The h2c connection preface starts with `PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n`.
    // Probe BEFORE the HTTP/1.x method dispatch because `PRI` is a
    // "request method" only on h2 and any HTTP/1.x parser would mistake
    // the bytes that follow for a malformed request line.
    if h2::is_h2c_preface(buf) {
        return ProtocolGuess::H2c;
    }
    // HTTP/1.x request lines start with an ASCII method token. Cover the
    // common verbs the brief enumerates; anything else is unknown.
    const METHODS: &[&[u8]] = &[
        b"GET ",
        b"POST ",
        b"HEAD ",
        b"PUT ",
        b"DELETE ",
        b"OPTIONS ",
        b"PATCH ",
        b"CONNECT ",
    ];
    for m in METHODS {
        if buf.len() >= m.len() && &buf[..m.len()] == *m {
            return ProtocolGuess::Http1;
        }
    }
    ProtocolGuess::Unknown
}

/// Run the SNI proxy accept loop until `shutdown` is set.
///
/// Each accepted TCP connection is handed to a dedicated `tokio::spawn`
/// task that:
///
/// 1. Reads up to [`PEEK_BUF_LEN`] bytes within `cfg.peek_timeout`.
/// 2. Probes the protocol byte (TLS=22 / HTTP method / unknown).
/// 3. Calls the appropriate parser ([`sni::extract_sni`] or
///    [`http::extract_http_host`]) on the buffered preamble.
/// 4. Matches the extracted host against `cfg.hostname_allowlist`.
/// 5. On allow: connects to `cfg.upstream_addr`, writes the buffered
///    preamble, then `tokio::io::copy_bidirectional` until either side
///    closes.
/// 6. On deny: closes the client (TLS) or writes
///    `HTTP/1.1 403 Forbidden\r\nContent-Length: 0\r\nConnection: close\r\n\r\n`
///    (HTTP) and drops the stream.
/// 7. Emits exactly one `l7_egress_decision` CloudEvent regardless of path.
///
/// Each spawned task increments [`ProxyStats`] via shared atomics; the
/// returned snapshot reflects cumulative counts at loop exit.
///
/// Termination: the accept loop awakens at most once per `shutdown` flip
/// because `accept` blocks. The spawn helper's `signal_sni_proxy_shutdown`
/// connects-and-drops a TCP socket to wake `accept`; the loop then observes
/// `shutdown` and returns.
pub async fn run_one_shot(
    cfg: &SniProxyConfig,
    listener: TcpListener,
    emitter: Arc<dyn L7DecisionEmitter>,
    shutdown: Arc<AtomicBool>,
) -> std::io::Result<ProxyStats> {
    let total = Arc::new(std::sync::atomic::AtomicU64::new(0));
    let allowed = Arc::new(std::sync::atomic::AtomicU64::new(0));
    let denied = Arc::new(std::sync::atomic::AtomicU64::new(0));
    let timeouts = Arc::new(std::sync::atomic::AtomicU64::new(0));
    let upstream_failures = Arc::new(std::sync::atomic::AtomicU64::new(0));

    // Build the per-event spec stub once; the supervisor passes us
    // pre-derived fields, but `observability_l7_egress_decision_data_v1`
    // takes a `&ExecutionCellSpec` for `correlation` and `id`. We synthesise
    // a minimal spec carrying just the fields the builder reads.
    let event_spec = build_event_spec(cfg);

    while !shutdown.load(Ordering::SeqCst) {
        let (stream, peer) = match listener.accept().await {
            Ok(t) => t,
            Err(e) => {
                if shutdown.load(Ordering::SeqCst) {
                    break;
                }
                tracing::warn!(
                    target: "cellos.supervisor.sni_proxy",
                    error = %e,
                    "accept() failed"
                );
                continue;
            }
        };
        if shutdown.load(Ordering::SeqCst) {
            // Wake-connection from `signal_sni_proxy_shutdown` — drop and exit.
            drop(stream);
            break;
        }
        total.fetch_add(1, Ordering::SeqCst);

        let cfg = cfg.clone();
        let emitter = emitter.clone();
        let event_spec = event_spec.clone();
        let allowed = allowed.clone();
        let denied = denied.clone();
        let timeouts = timeouts.clone();
        let upstream_failures = upstream_failures.clone();

        tokio::spawn(async move {
            handle_connection(
                stream,
                peer,
                cfg,
                emitter,
                event_spec,
                allowed,
                denied,
                timeouts,
                upstream_failures,
            )
            .await;
        });
    }

    Ok(ProxyStats {
        connections_total: total.load(Ordering::SeqCst),
        connections_allowed: allowed.load(Ordering::SeqCst),
        connections_denied: denied.load(Ordering::SeqCst),
        peek_timeouts: timeouts.load(Ordering::SeqCst),
        upstream_failures: upstream_failures.load(Ordering::SeqCst),
    })
}

/// Per-connection worker. Encapsulated so [`run_one_shot`] stays focused on
/// the accept loop. Updates the shared atomics + emits exactly one event
/// per call.
#[allow(clippy::too_many_arguments)]
async fn handle_connection(
    mut stream: TcpStream,
    _peer: SocketAddr,
    cfg: SniProxyConfig,
    emitter: Arc<dyn L7DecisionEmitter>,
    event_spec: ExecutionCellSpec,
    allowed: Arc<std::sync::atomic::AtomicU64>,
    denied: Arc<std::sync::atomic::AtomicU64>,
    timeouts: Arc<std::sync::atomic::AtomicU64>,
    upstream_failures: Arc<std::sync::atomic::AtomicU64>,
) {
    // ── Step 1: peek with a bounded timeout. ──────────────────────────────
    let mut buf = vec![0u8; PEEK_BUF_LEN];
    let peek_result = tokio::time::timeout(cfg.peek_timeout, stream.peek(&mut buf)).await;

    let n = match peek_result {
        Err(_elapsed) => {
            // Peek timed out before any bytes arrived. Drop the client and
            // emit a peek_timeout event — this is the fallback path when
            // the workload opens a TCP connection but never sends bytes.
            timeouts.fetch_add(1, Ordering::SeqCst);
            denied.fetch_add(1, Ordering::SeqCst);
            emit_decision(
                &emitter,
                &cfg,
                &event_spec,
                "deny",
                "",
                reason_code::PEEK_TIMEOUT,
                None,
                None,
            );
            return;
        }
        Ok(Err(e)) => {
            tracing::debug!(
                target: "cellos.supervisor.sni_proxy",
                error = %e,
                "peek() error before any bytes"
            );
            denied.fetch_add(1, Ordering::SeqCst);
            emit_decision(
                &emitter,
                &cfg,
                &event_spec,
                "deny",
                "",
                reason_code::UNKNOWN_PROTOCOL,
                None,
                None,
            );
            return;
        }
        Ok(Ok(0)) => {
            // EOF before any data — treat as unknown.
            denied.fetch_add(1, Ordering::SeqCst);
            emit_decision(
                &emitter,
                &cfg,
                &event_spec,
                "deny",
                "",
                reason_code::UNKNOWN_PROTOCOL,
                None,
                None,
            );
            return;
        }
        Ok(Ok(n)) => n,
    };
    let preamble = &buf[..n];

    // ── Step 2: probe protocol + extract host. ────────────────────────────
    let guess = guess_protocol(preamble);
    let (host_opt, allow_reason, miss_reason, missing_reason, deny_response): (
        Option<String>,
        &'static str,
        &'static str,
        &'static str,
        DenyResponse,
    ) = match guess {
        ProtocolGuess::Tls => match sni::extract_sni(preamble) {
            Ok(opt) => (
                opt,
                reason_code::SNI_ALLOWLIST_MATCH,
                reason_code::SNI_ALLOWLIST_MISS,
                reason_code::SNI_MISSING,
                DenyResponse::Drop,
            ),
            Err(_e) => {
                denied.fetch_add(1, Ordering::SeqCst);
                emit_decision(
                    &emitter,
                    &cfg,
                    &event_spec,
                    "deny",
                    "",
                    reason_code::UNKNOWN_PROTOCOL,
                    None,
                    None,
                );
                return;
            }
        },
        ProtocolGuess::H2c => {
            // scope: per-stream allow/deny. Hand off to a dedicated
            // frame-by-frame handler that:
            //   - connects to upstream
            //   - parses every frame on both directions (frame layer is
            //     pure-byte; no h2 crate)
            //   - decides allow/deny per HEADERS+CONTINUATION block via
            //     H2ConnectionDecoder, and emits one l7_egress_decision
            //     per stream decision (each event carries streamId)
            //   - on deny: writes RST_STREAM(REFUSED_STREAM=7) to the
            //     workload, drops further frames on that stream, but
            //     keeps the connection alive for OTHER streams
            //   - allow path forwards verbatim to upstream
            //
            // The single-connection allow/deny lock from Phase 3g is
            // gone; the connection itself is no longer either "allowed"
            // or "denied" — only individual streams are.
            handle_h2c_connection(
                stream,
                preamble.to_vec(),
                cfg.clone(),
                emitter.clone(),
                event_spec.clone(),
                allowed.clone(),
                denied.clone(),
                upstream_failures.clone(),
            )
            .await;
            return;
        }
        ProtocolGuess::Http1 => match http::extract_http_host(preamble) {
            Ok(opt) => (
                opt,
                reason_code::HTTP_HOST_ALLOWLIST_MATCH,
                reason_code::HTTP_HOST_ALLOWLIST_MISS,
                reason_code::HTTP_HOST_MISSING,
                DenyResponse::Http403,
            ),
            Err(_e) => {
                denied.fetch_add(1, Ordering::SeqCst);
                emit_decision(
                    &emitter,
                    &cfg,
                    &event_spec,
                    "deny",
                    "",
                    reason_code::UNKNOWN_PROTOCOL,
                    None,
                    None,
                );
                return;
            }
        },
        ProtocolGuess::Unknown => {
            denied.fetch_add(1, Ordering::SeqCst);
            emit_decision(
                &emitter,
                &cfg,
                &event_spec,
                "deny",
                "",
                reason_code::UNKNOWN_PROTOCOL,
                None,
                None,
            );
            return;
        }
    };

    let host = match host_opt {
        Some(h) if !h.is_empty() => h,
        _ => {
            // TLS without SNI / HTTP without Host. Deny per protocol.
            denied.fetch_add(1, Ordering::SeqCst);
            send_deny_response(&mut stream, deny_response).await;
            emit_decision(
                &emitter,
                &cfg,
                &event_spec,
                "deny",
                "",
                missing_reason,
                None,
                None,
            );
            return;
        }
    };

    // ── Step 3: match against allowlist. ──────────────────────────────────
    if !cellos_core::hostname_allowlist::matches_allowlist(&host, &cfg.hostname_allowlist) {
        denied.fetch_add(1, Ordering::SeqCst);
        send_deny_response(&mut stream, deny_response).await;
        emit_decision(
            &emitter,
            &cfg,
            &event_spec,
            "deny",
            host.as_str(),
            miss_reason,
            None,
            None,
        );
        return;
    }

    // ── Step 4: allow → forward to upstream. ──────────────────────────────
    let upstream = match TcpStream::connect(cfg.upstream_addr).await {
        Ok(s) => s,
        Err(e) => {
            tracing::warn!(
                target: "cellos.supervisor.sni_proxy",
                error = %e,
                upstream = %cfg.upstream_addr,
                "upstream connect failed"
            );
            upstream_failures.fetch_add(1, Ordering::SeqCst);
            denied.fetch_add(1, Ordering::SeqCst);
            // Allowlist matched but upstream unreachable: send deny response so
            // the workload sees deterministic failure rather than hanging.
            send_deny_response(&mut stream, deny_response).await;
            emit_decision(
                &emitter,
                &cfg,
                &event_spec,
                "deny",
                host.as_str(),
                reason_code::UNKNOWN_PROTOCOL,
                None,
                None,
            );
            return;
        }
    };
    allowed.fetch_add(1, Ordering::SeqCst);
    emit_decision(
        &emitter,
        &cfg,
        &event_spec,
        "allow",
        host.as_str(),
        allow_reason,
        None,
        None,
    );

    // The peeked bytes are still in the kernel socket buffer (peek doesn't
    // consume), so `copy_bidirectional` will replay them to the upstream
    // automatically. This preserves the ClientHello / Host-bearing request
    // line so the upstream sees the original wire bytes verbatim.
    let mut client = stream;
    let mut up = upstream;
    if let Err(e) = tokio::io::copy_bidirectional(&mut client, &mut up).await {
        tracing::debug!(
            target: "cellos.supervisor.sni_proxy",
            error = %e,
            host = %host,
            "copy_bidirectional ended with error"
        );
    }
}

/// scope: dedicated h2c connection handler with per-stream
/// allow/deny semantics.
///
/// Supersedes a prior `tokio::io::copy_bidirectional` short-circuit
/// (which locked the entire connection to the FIRST `:authority`
/// observed) with a frame-by-frame parser:
///
/// 1. Connect to upstream; forward the buffered preface bytes.
/// 2. In one task, copy upstream → client verbatim (we don't inspect
///    server-emitted HEADERS — server-PUSH inspection is Phase 4).
/// 3. In the main task, read client; parse frames; for HEADERS or
///    CONTINUATION feed to `H2ConnectionDecoder.feed_frame`. For each
///    completed block, match `:authority` against the allowlist and
///    either forward verbatim (allow) or RST_STREAM(REFUSED_STREAM=7)
///    plus drop subsequent stream-bound frames (deny). Emit one
///    `l7_egress_decision` event per stream decision; each event
///    carries `streamId`.
/// 4. Control frames (SETTINGS / PING / WINDOW_UPDATE / GOAWAY) are
///    forwarded verbatim regardless of any per-stream deny — they
///    govern the connection itself.
///
/// Honest scope: server-PUSH'd stream `:authority` (PUSH_PROMISE
/// HEADERS) and TRAILERS-frame inspection both remain Phase 4 — this
/// handler reads only client→server HEADERS/CONTINUATION. TLS-encrypted
/// h2 (h2 over TLS) is still ADR-0004 territory.
///
/// CONTINUATION-fragmented HEADERS across MULTIPLE streams on the same
/// connection: per RFC 7540 §6.10, only one stream may have an open
/// HEADERS at a time on a connection (the §6.10 PROTOCOL_ERROR contract
/// is enforced inside `H2StreamReassembler`). For the single-stream
/// CONTINUATION case the handler buffers the raw HEADERS+CONTINUATION
/// frame bytes against the stream id and forwards them on allow / drops
/// them on deny.
#[allow(clippy::too_many_arguments)]
/// Lazy upstream connection for the h2c handler.
///
/// **Zero-byte invariant on deny.** The handler MUST NOT open a TCP
/// connection to the upstream until the first `:authority`-bearing
/// HEADERS frame has been parsed AND the authority decision is `allow`.
/// Until then, every byte the parse loop wants to forward (the 24-byte
/// HTTP/2 preface, the workload's SETTINGS, WINDOW_UPDATE, and any
/// other control frames that arrive on stream 0) is buffered in
/// memory. If the first authority decision is `deny`, the buffer is
/// dropped and no connection is ever made — the
/// `break_attempt_sni_mismatch_h2c::h2c_authority_mismatch_is_denied_with_zero_upstream_bytes`
/// regression guard locks this in.
enum UpstreamSink {
    /// No upstream socket yet. `buffer` accumulates bytes the parse
    /// loop wanted to forward; on the first allow they are flushed in
    /// order.
    Pending { addr: SocketAddr, buffer: Vec<u8> },
    /// Connected: bytes flow through the split write half; the read
    /// half is exposed to the bidirectional select! loop.
    Open {
        rd: tokio::net::tcp::OwnedReadHalf,
        wr: tokio::net::tcp::OwnedWriteHalf,
    },
}

impl UpstreamSink {
    fn pending(addr: SocketAddr) -> Self {
        Self::Pending {
            addr,
            buffer: Vec::new(),
        }
    }

    /// Buffer or write — never opens the connection.
    async fn write(&mut self, bytes: &[u8]) -> std::io::Result<()> {
        match self {
            Self::Pending { buffer, .. } => {
                buffer.extend_from_slice(bytes);
                Ok(())
            }
            Self::Open { wr, .. } => wr.write_all(bytes).await,
        }
    }

    /// Connect (idempotent). On first call: opens the TCP socket and
    /// flushes the pending buffer. On subsequent calls: no-op.
    async fn commit(&mut self) -> std::io::Result<()> {
        if let Self::Pending { addr, buffer } = self {
            let stream = TcpStream::connect(*addr).await?;
            let (rd, mut wr) = stream.into_split();
            if !buffer.is_empty() {
                wr.write_all(buffer).await?;
            }
            *self = Self::Open { rd, wr };
        }
        Ok(())
    }

    async fn shutdown(&mut self) {
        if let Self::Open { wr, .. } = self {
            let _ = wr.shutdown().await;
        }
    }
}

/// `select!`-friendly read: returns the upstream's next bytes when
/// connected, otherwise an unfullfillable future so the select branch
/// is effectively dormant until the first allow opens the socket.
async fn read_upstream(sink: &mut UpstreamSink, buf: &mut [u8]) -> std::io::Result<usize> {
    use tokio::io::AsyncReadExt;
    match sink {
        UpstreamSink::Open { rd, .. } => rd.read(buf).await,
        UpstreamSink::Pending { .. } => std::future::pending().await,
    }
}

#[allow(clippy::too_many_arguments)]
async fn handle_h2c_connection(
    stream: TcpStream,
    preamble: Vec<u8>,
    cfg: SniProxyConfig,
    emitter: Arc<dyn L7DecisionEmitter>,
    event_spec: ExecutionCellSpec,
    allowed: Arc<std::sync::atomic::AtomicU64>,
    denied: Arc<std::sync::atomic::AtomicU64>,
    upstream_failures: Arc<std::sync::atomic::AtomicU64>,
) {
    use tokio::io::AsyncReadExt;

    // Step 1: stage the upstream sink in `Pending` mode. The TCP
    // connection is NOT opened yet — only the first allow decision
    // commits it. Until then every "to upstream" write goes into the
    // sink's buffer; on a connection that only ever sees denied
    // streams we never connect, never leak the preface, never leak
    // SETTINGS.
    let mut upstream = UpstreamSink::pending(cfg.upstream_addr);

    // Step 2: consume the buffered preamble from the kernel socket.
    // The peek didn't consume; do a real read so subsequent reads
    // start AFTER the preamble.
    let mut client = stream;
    let mut consumed = vec![0u8; preamble.len()];
    if let Err(e) = client.read_exact(&mut consumed).await {
        tracing::debug!(
            target: "cellos.supervisor.sni_proxy",
            error = %e,
            "h2c read_exact(preamble) failed"
        );
        return;
    }
    debug_assert_eq!(
        consumed, preamble,
        "kernel returned different bytes than peek"
    );

    // Step 3: split the client side now (we own both halves) and queue
    // the 24-byte preface against the deferred upstream sink. The
    // preface is part of every h2 connection's wire shape, but it is
    // not committed to the upstream until the first allow.
    let (mut client_rd, mut client_wr) = client.into_split();

    if let Err(e) = upstream.write(&preamble[..h2::HTTP2_PREFACE.len()]).await {
        tracing::debug!(
            target: "cellos.supervisor.sni_proxy",
            error = %e,
            "h2c buffer(preface) failed"
        );
        return;
    }

    // Bidirectional copy is driven inline via tokio::select! below so
    // the same task that owns `client_wr` can both forward upstream→
    // client bytes verbatim AND emit RST_STREAM frames on the workload
    // side when a per-stream allow/deny decision lands. Server-PUSH
    // HEADERS inspection is Phase 4.
    let mut decoder = h2::H2ConnectionDecoder::new();
    let mut denied_streams: std::collections::HashSet<u32> = std::collections::HashSet::new();
    // Per-stream pending HEADERS+CONTINUATION raw frame buffer. The
    // handler buffers raw frame bytes against the stream id and
    // forwards them on the block-level allow decision (or drops on
    // deny). RFC 7540 §6.10 forbids interleaving multiple streams'
    // HEADERS+CONTINUATION sequences, so only ONE entry will ever be
    // live at a time.
    let mut pending_headers: std::collections::HashMap<u32, Vec<u8>> =
        std::collections::HashMap::new();
    let mut frame_buf: Vec<u8> = preamble[h2::HTTP2_PREFACE.len()..].to_vec();
    let mut read_chunk = vec![0u8; 16 * 1024];
    let mut up_read_chunk = vec![0u8; 16 * 1024];

    'outer: loop {
        // Try to parse as many full frames as the buffer holds.
        loop {
            match h2::frame::parse_one_frame(&frame_buf) {
                Ok(Some((header, payload, _rest))) => {
                    let frame_total = 9 + header.length as usize;
                    let frame_bytes = frame_buf[..frame_total].to_vec();
                    let payload_owned = payload.to_vec();
                    let frame_type = header.frame_type;
                    let stream_id = header.stream_id;
                    let is_stream_bound = stream_id != 0;
                    let is_headers_or_continuation = frame_type == h2::frame::FRAME_TYPE_HEADERS
                        || frame_type == h2::frame::FRAME_TYPE_CONTINUATION;

                    if is_headers_or_continuation {
                        // Buffer the raw frame bytes against stream id;
                        // they'll be forwarded together on allow.
                        pending_headers
                            .entry(stream_id)
                            .or_default()
                            .extend_from_slice(&frame_bytes);

                        match decoder.feed_frame(&header, &payload_owned) {
                            Ok(Some(decoded)) => {
                                let sid = decoded.stream_id;
                                let host_norm = decoded.authority.clone();
                                let provenance = if decoded.via_huffman {
                                    h2::AuthorityProvenance::Huffman
                                } else if decoded.via_dynamic_table {
                                    h2::AuthorityProvenance::DynamicIndexed
                                } else {
                                    h2::AuthorityProvenance::StaticLiteral
                                };
                                let (allow_r, miss_r) = h2_reason_codes_for(provenance);
                                let allow = match host_norm.as_deref() {
                                    Some(h) if !h.is_empty() => {
                                        cellos_core::hostname_allowlist::matches_allowlist(
                                            h,
                                            &cfg.hostname_allowlist,
                                        )
                                    }
                                    _ => false,
                                };
                                let pending = pending_headers.remove(&sid).unwrap_or_default();
                                if host_norm.is_none() {
                                    denied.fetch_add(1, Ordering::SeqCst);
                                    denied_streams.insert(sid);
                                    let rst = build_rst_stream_refused(sid);
                                    let _ = client_wr.write_all(&rst).await;
                                    emit_decision(
                                        &emitter,
                                        &cfg,
                                        &event_spec,
                                        "deny",
                                        "",
                                        reason_code::H2_AUTHORITY_MISSING,
                                        None,
                                        Some(sid),
                                    );
                                } else if allow {
                                    allowed.fetch_add(1, Ordering::SeqCst);
                                    // First allow gates the TCP commit:
                                    // open the upstream socket and flush
                                    // the preface + any buffered control
                                    // frames (SETTINGS, WINDOW_UPDATE).
                                    // Idempotent — second + subsequent
                                    // allows just write through.
                                    if let Err(e) = upstream.commit().await {
                                        tracing::warn!(
                                            target: "cellos.supervisor.sni_proxy",
                                            error = %e,
                                            upstream = %cfg.upstream_addr,
                                            "h2c upstream connect failed (deferred)"
                                        );
                                        upstream_failures.fetch_add(1, Ordering::SeqCst);
                                        denied.fetch_add(1, Ordering::SeqCst);
                                        let _ = client_wr.write_all(H2_GOAWAY_PROTOCOL_ERROR).await;
                                        let _ = client_wr.shutdown().await;
                                        emit_decision(
                                            &emitter,
                                            &cfg,
                                            &event_spec,
                                            "deny",
                                            "",
                                            reason_code::UNKNOWN_PROTOCOL,
                                            None,
                                            None,
                                        );
                                        break 'outer;
                                    }
                                    if let Err(e) = upstream.write(&pending).await {
                                        tracing::debug!(
                                            target: "cellos.supervisor.sni_proxy",
                                            error = %e,
                                            "h2c forward(allowed HEADERS block) failed"
                                        );
                                        break 'outer;
                                    }
                                    emit_decision(
                                        &emitter,
                                        &cfg,
                                        &event_spec,
                                        "allow",
                                        host_norm.as_deref().unwrap_or(""),
                                        allow_r,
                                        None,
                                        Some(sid),
                                    );
                                } else {
                                    denied.fetch_add(1, Ordering::SeqCst);
                                    denied_streams.insert(sid);
                                    let rst = build_rst_stream_refused(sid);
                                    let _ = client_wr.write_all(&rst).await;
                                    emit_decision(
                                        &emitter,
                                        &cfg,
                                        &event_spec,
                                        "deny",
                                        host_norm.as_deref().unwrap_or(""),
                                        miss_r,
                                        None,
                                        Some(sid),
                                    );
                                }
                            }
                            Ok(None) => {
                                // Block not yet complete; keep buffering.
                            }
                            Err(_e) => {
                                denied.fetch_add(1, Ordering::SeqCst);
                                denied_streams.insert(stream_id);
                                pending_headers.remove(&stream_id);
                                let rst = build_rst_stream_refused(stream_id);
                                let _ = client_wr.write_all(&rst).await;
                                emit_decision(
                                    &emitter,
                                    &cfg,
                                    &event_spec,
                                    "deny",
                                    "",
                                    reason_code::H2_UNPARSEABLE_HEADERS,
                                    None,
                                    Some(stream_id),
                                );
                            }
                        }
                    } else if is_stream_bound && denied_streams.contains(&stream_id) {
                        // Drop stream-bound frames on a denied stream.
                    } else {
                        // Control frames (stream id 0) + stream-bound
                        // on undenied streams: queue or forward. The
                        // sink buffers if upstream isn't connected yet
                        // (deferred-connect invariant for the deny path)
                        // and writes through once the first allow has
                        // committed the connection.
                        if let Err(e) = upstream.write(&frame_bytes).await {
                            tracing::debug!(
                                target: "cellos.supervisor.sni_proxy",
                                error = %e,
                                "h2c forward(frame) failed"
                            );
                            break 'outer;
                        }
                    }

                    frame_buf.drain(..frame_total);
                    continue;
                }
                Ok(None) => break, // need more bytes
                Err(_e) => {
                    // Connection-level frame parse error (oversized
                    // frame, malformed header). Send GOAWAY and bail.
                    let _ = client_wr.write_all(H2_GOAWAY_PROTOCOL_ERROR).await;
                    upstream.shutdown().await;
                    break 'outer;
                }
            }
        }

        // Bidirectional read: prefer client (so we keep parsing) but
        // also drain upstream → client to avoid stalling the upstream
        // when it sends responses (HEADERS / DATA / WINDOW_UPDATE).
        tokio::select! {
            biased;
            r = client_rd.read(&mut read_chunk) => match r {
                Ok(0) => break 'outer,
                Ok(n) => frame_buf.extend_from_slice(&read_chunk[..n]),
                Err(e) => {
                    tracing::debug!(
                        target: "cellos.supervisor.sni_proxy",
                        error = %e,
                        "h2c client read error"
                    );
                    break 'outer;
                }
            },
            r = read_upstream(&mut upstream, &mut up_read_chunk) => match r {
                Ok(0) => break 'outer,
                Ok(n) => {
                    if let Err(e) = client_wr.write_all(&up_read_chunk[..n]).await {
                        tracing::debug!(
                            target: "cellos.supervisor.sni_proxy",
                            error = %e,
                            "h2c upstream→client write failed"
                        );
                        break 'outer;
                    }
                }
                Err(e) => {
                    tracing::debug!(
                        target: "cellos.supervisor.sni_proxy",
                        error = %e,
                        "h2c upstream read error"
                    );
                    break 'outer;
                }
            },
        }
    }

    upstream.shutdown().await;
    let _ = client_wr.shutdown().await;
}

/// Build a 13-byte RST_STREAM frame (RFC 7540 §6.4) on `stream_id` with
/// error code `REFUSED_STREAM` (0x7). Layout:
///
/// ```text
///   00 00 04          length = 4
///   03                type = RST_STREAM
///   00                flags = 0
///   ?? ?? ?? ??       stream id (R bit clear)
///   00 00 00 07       error code = REFUSED_STREAM
/// ```
fn build_rst_stream_refused(stream_id: u32) -> [u8; 13] {
    let mut out = [0u8; 13];
    out[0] = 0x00;
    out[1] = 0x00;
    out[2] = 0x04; // length
    out[3] = 0x03; // type = RST_STREAM
    out[4] = 0x00; // flags
    let sid = stream_id & 0x7FFF_FFFF;
    out[5..9].copy_from_slice(&sid.to_be_bytes());
    out[9..13].copy_from_slice(&7u32.to_be_bytes()); // REFUSED_STREAM
    out
}

#[derive(Debug, Clone, Copy)]
enum DenyResponse {
    Drop,
    Http403,
    /// scope: retained for connection-level h2c errors that bail out
    /// before the per-stream handler — see [`handle_h2c_connection`].
    /// The per-stream allow/deny path uses
    /// `RST_STREAM(REFUSED_STREAM=7)` directly via
    /// [`build_rst_stream_refused`] instead so the connection stays
    /// alive for other streams.
    #[allow(dead_code)]
    H2Goaway,
}

/// Minimal HTTP/2 GOAWAY frame (RFC 7540 §6.8) carrying error code
/// `PROTOCOL_ERROR` (0x1) on stream 0 with `last-stream-id = 1`. Layout:
///
/// ```text
///   00 00 08    length = 8
///   07          type = GOAWAY
///   00          flags = 0
///   00 00 00 00 stream id = 0 (R bit clear)
///   00 00 00 01 last-stream-id = 1
///   00 00 00 01 error code = PROTOCOL_ERROR
/// ```
const H2_GOAWAY_PROTOCOL_ERROR: &[u8; 17] = &[
    0x00, 0x00, 0x08, // length
    0x07, // type=GOAWAY
    0x00, // flags
    0x00, 0x00, 0x00, 0x00, // stream id=0
    0x00, 0x00, 0x00, 0x01, // last-stream-id=1
    0x00, 0x00, 0x00, 0x01, // error code = PROTOCOL_ERROR
];

async fn send_deny_response(stream: &mut TcpStream, mode: DenyResponse) {
    match mode {
        DenyResponse::Http403 => {
            const RESP: &[u8] =
                b"HTTP/1.1 403 Forbidden\r\nContent-Length: 0\r\nConnection: close\r\n\r\n";
            if let Err(e) = stream.write_all(RESP).await {
                tracing::debug!(
                    target: "cellos.supervisor.sni_proxy",
                    error = %e,
                    "writing 403 response failed"
                );
            }
            let _ = stream.shutdown().await;
        }
        DenyResponse::H2Goaway => {
            if let Err(e) = stream.write_all(H2_GOAWAY_PROTOCOL_ERROR).await {
                tracing::debug!(
                    target: "cellos.supervisor.sni_proxy",
                    error = %e,
                    "writing h2 GOAWAY failed"
                );
            }
            let _ = stream.shutdown().await;
        }
        DenyResponse::Drop => {
            // Nothing to write — letting `stream` drop closes it.
        }
    }
}

/// Build a synthetic [`ExecutionCellSpec`] carrying just the fields
/// [`cellos_core::observability_l7_egress_decision_data_v1`] reads (id +
/// correlation). The supervisor's full spec is large and we do not need it
/// inside the proxy thread; a thin shim keeps the API surface narrow.
fn build_event_spec(cfg: &SniProxyConfig) -> ExecutionCellSpec {
    use cellos_core::{AuthorityBundle, Correlation, Lifetime};
    let correlation = cfg.correlation_id.as_ref().map(|c| Correlation {
        correlation_id: Some(c.clone()),
        ..Default::default()
    });
    ExecutionCellSpec {
        id: format!("sni-proxy/{}/{}", cfg.cell_id, cfg.run_id),
        correlation,
        ingress: None,
        environment: None,
        placement: None,
        policy: None,
        identity: None,
        run: None,
        authority: AuthorityBundle::default(),
        lifetime: Lifetime { ttl_seconds: 0 },
        export: None,
        telemetry: None,
    }
}

/// Build + emit one `l7_egress_decision` CloudEvent.
#[allow(clippy::too_many_arguments)]
fn emit_decision(
    emitter: &Arc<dyn L7DecisionEmitter>,
    cfg: &SniProxyConfig,
    spec: &ExecutionCellSpec,
    action: &str,
    sni_host: &str,
    reason_code_str: &str,
    rule_ref: Option<&str>,
    // scope: per-stream correlation id for h2 paths. None for
    // SNI / HTTP/1.x / unknown / peek-timeout paths.
    stream_id: Option<u32>,
) {
    let decision_id = uuid::Uuid::new_v4().to_string();
    let policy_digest = cfg.policy_digest.clone().unwrap_or_default();
    let keyset_id = cfg.keyset_id.clone().unwrap_or_default();
    let issuer_kid = cfg.issuer_kid.clone().unwrap_or_default();
    let data = match cellos_core::observability_l7_egress_decision_data_v1(
        spec,
        cfg.cell_id.as_str(),
        Some(cfg.run_id.as_str()),
        decision_id.as_str(),
        action,
        // Schema requires sniHost.minLength=1; substitute a conventional
        // placeholder when the proxy could not extract a hostname (peek
        // timeout, missing SNI / Host, unknown protocol). The reasonCode
        // preserves the underlying signal.
        if sni_host.is_empty() {
            "(unknown)"
        } else {
            sni_host
        },
        policy_digest.as_str(),
        keyset_id.as_str(),
        issuer_kid.as_str(),
        reason_code_str,
        rule_ref,
        stream_id,
    ) {
        Ok(v) => v,
        Err(e) => {
            tracing::warn!(
                target: "cellos.supervisor.sni_proxy",
                error = %e,
                "build l7_egress_decision data failed"
            );
            return;
        }
    };
    let observed_at = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true);
    let event = CloudEventV1 {
        specversion: "1.0".into(),
        id: uuid::Uuid::new_v4().to_string(),
        source: "cellos-sni-proxy".into(),
        ty: "dev.cellos.events.cell.observability.v1.l7_egress_decision".into(),
        datacontenttype: Some("application/json".into()),
        data: Some(data),
        time: Some(observed_at),
        traceparent: None,
    };
    emitter.emit(event);
}

/// Best-effort field-value extractor used by tests + supervisor diagnostics.
/// Walks the event's `data` object and returns the JSON value at `key`.
#[allow(dead_code)]
fn event_data_get<'a>(event: &'a CloudEventV1, key: &str) -> Option<&'a Value> {
    event.data.as_ref()?.as_object()?.get(key)
}

/// Construct an empty `Map<String, Value>` with the standard event-data
/// shape; reserved for future use if tests need to fabricate events without
/// going through the emitter path.
#[allow(dead_code)]
fn empty_data_map() -> Map<String, Value> {
    let mut m = Map::new();
    m.insert("decisionId".into(), json!(uuid::Uuid::new_v4().to_string()));
    m
}

#[cfg(test)]
pub(crate) mod test_helpers {
    /// Minimal but well-formed TLS ClientHello bytes carrying the given
    /// SNI value(s). Mirrors the synth fixture in `sni::tests` so other
    /// tests can build records without re-implementing the wire format.
    pub fn build_client_hello(snis: &[&str]) -> Vec<u8> {
        let mut body = Vec::new();
        body.extend_from_slice(&[0x03, 0x03]); // legacy_version TLS 1.2
        body.extend_from_slice(&[0u8; 32]); // random
        body.push(0); // session id length
        body.extend_from_slice(&[0x00, 0x02, 0x13, 0x01]); // 1 cipher suite
        body.extend_from_slice(&[0x01, 0x00]); // compression methods

        let mut ext_section = Vec::new();
        if !snis.is_empty() {
            let mut sn_body = Vec::new();
            let mut inner = Vec::new();
            for s in snis {
                inner.push(0u8); // host_name type
                inner.extend_from_slice(&(s.len() as u16).to_be_bytes());
                inner.extend_from_slice(s.as_bytes());
            }
            sn_body.extend_from_slice(&(inner.len() as u16).to_be_bytes());
            sn_body.extend_from_slice(&inner);
            ext_section.extend_from_slice(&[0x00, 0x00]);
            ext_section.extend_from_slice(&(sn_body.len() as u16).to_be_bytes());
            ext_section.extend_from_slice(&sn_body);
        }
        body.extend_from_slice(&(ext_section.len() as u16).to_be_bytes());
        body.extend_from_slice(&ext_section);

        let mut hs = Vec::new();
        hs.push(1);
        let body_len_bytes = (body.len() as u32).to_be_bytes();
        hs.extend_from_slice(&body_len_bytes[1..]);
        hs.extend_from_slice(&body);

        let mut rec = Vec::new();
        rec.push(22);
        rec.extend_from_slice(&[0x03, 0x01]);
        rec.extend_from_slice(&(hs.len() as u16).to_be_bytes());
        rec.extend_from_slice(&hs);
        rec
    }
}

#[cfg(test)]
mod tests {
    use super::test_helpers::build_client_hello;
    use super::*;
    use std::sync::Mutex;
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    use tokio::net::{TcpListener, TcpStream};

    /// In-memory event collector for proxy unit tests.
    #[derive(Default)]
    struct MemEmitter {
        events: Mutex<Vec<CloudEventV1>>,
    }
    impl L7DecisionEmitter for MemEmitter {
        fn emit(&self, event: CloudEventV1) {
            self.events.lock().unwrap().push(event);
        }
    }

    fn cfg_with(allowlist: &[&str], upstream: SocketAddr, peek_ms: u64) -> SniProxyConfig {
        SniProxyConfig {
            bind_addr: "127.0.0.1:0".parse().unwrap(),
            upstream_addr: upstream,
            hostname_allowlist: allowlist.iter().map(|s| s.to_string()).collect(),
            cdn_providers: vec![],
            cell_id: "test-cell".into(),
            run_id: "test-run".into(),
            policy_digest: Some("digest-test".into()),
            keyset_id: Some("keyset-test".into()),
            issuer_kid: Some("kid-test".into()),
            correlation_id: None,
            upstream_resolver_id: "sni-proxy-test".into(),
            peek_timeout: Duration::from_millis(peek_ms),
        }
    }

    /// Spawn a tiny localhost echo server: accepts one connection, reads
    /// everything until EOF, echoes it back. Returns the bound address +
    /// a join handle. Used as the upstream for allow-path tests.
    async fn spawn_echo_upstream() -> (SocketAddr, tokio::task::JoinHandle<Vec<u8>>) {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let h = tokio::spawn(async move {
            let (mut s, _) = listener.accept().await.unwrap();
            let mut buf = Vec::new();
            // Read until EOF or first 64KiB; the proxy's
            // copy_bidirectional may keep the half-open stream alive
            // longer than the client wants, so cap.
            let mut tmp = [0u8; 4096];
            for _ in 0..32 {
                match s.read(&mut tmp).await {
                    Ok(0) => break,
                    Ok(n) => buf.extend_from_slice(&tmp[..n]),
                    Err(_) => break,
                }
            }
            buf
        });
        (addr, h)
    }

    /// Spawn the proxy; return its listen addr + a shutdown flag + the
    /// accept-loop join handle.
    async fn spawn_proxy(
        cfg: SniProxyConfig,
        emitter: Arc<MemEmitter>,
    ) -> (
        SocketAddr,
        Arc<AtomicBool>,
        tokio::task::JoinHandle<std::io::Result<ProxyStats>>,
    ) {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let shutdown = Arc::new(AtomicBool::new(false));
        let shutdown2 = shutdown.clone();
        let h = tokio::spawn(async move {
            run_one_shot(
                &cfg,
                listener,
                emitter as Arc<dyn L7DecisionEmitter>,
                shutdown2,
            )
            .await
        });
        (addr, shutdown, h)
    }

    fn poke_shutdown(addr: SocketAddr) {
        // Connect-and-drop wakes accept().
        let _ = std::net::TcpStream::connect_timeout(&addr, Duration::from_millis(200));
    }

    #[tokio::test]
    async fn proxy_allows_tls_with_matching_sni() {
        let (upstream, upstream_h) = spawn_echo_upstream().await;
        let emitter = Arc::new(MemEmitter::default());
        let cfg = cfg_with(&["api.example.com"], upstream, 500);
        let (listen, shutdown, h) = spawn_proxy(cfg, emitter.clone()).await;

        let ch = build_client_hello(&["api.example.com"]);
        let mut s = TcpStream::connect(listen).await.unwrap();
        s.write_all(&ch).await.unwrap();
        s.shutdown().await.ok();
        // Give upstream time to drain.
        let upstream_bytes = tokio::time::timeout(Duration::from_secs(2), upstream_h)
            .await
            .expect("upstream task")
            .expect("upstream join");
        // Upstream should have received exactly the ClientHello bytes
        // (peek didn't consume; copy_bidirectional replayed them).
        assert_eq!(
            upstream_bytes, ch,
            "upstream did not receive forwarded ClientHello bytes verbatim"
        );

        shutdown.store(true, Ordering::SeqCst);
        poke_shutdown(listen);
        let _ = tokio::time::timeout(Duration::from_secs(2), h).await;

        let evs = emitter.events.lock().unwrap();
        assert_eq!(evs.len(), 1, "expected exactly one event");
        let data = evs[0].data.as_ref().unwrap();
        assert_eq!(data["action"], "allow");
        assert_eq!(data["reasonCode"], "l7_sni_allowlist_match");
        assert_eq!(data["sniHost"], "api.example.com");
    }

    #[tokio::test]
    async fn proxy_denies_tls_with_unmatched_sni() {
        let (upstream, _upstream_h) = spawn_echo_upstream().await;
        let emitter = Arc::new(MemEmitter::default());
        let cfg = cfg_with(&["api.example.com"], upstream, 500);
        let (listen, shutdown, h) = spawn_proxy(cfg, emitter.clone()).await;

        let ch = build_client_hello(&["evil.example.com"]);
        let mut s = TcpStream::connect(listen).await.unwrap();
        s.write_all(&ch).await.unwrap();
        // For TLS deny, the proxy drops — read should return EOF quickly.
        let mut sink = Vec::new();
        let read_timeout =
            tokio::time::timeout(Duration::from_millis(800), s.read_to_end(&mut sink)).await;
        assert!(
            read_timeout.is_ok(),
            "TLS deny should close the stream promptly; got peek timeout"
        );

        shutdown.store(true, Ordering::SeqCst);
        poke_shutdown(listen);
        let _ = tokio::time::timeout(Duration::from_secs(2), h).await;

        let evs = emitter.events.lock().unwrap();
        assert_eq!(evs.len(), 1);
        let data = evs[0].data.as_ref().unwrap();
        assert_eq!(data["action"], "deny");
        assert_eq!(data["reasonCode"], "l7_sni_allowlist_miss");
        assert_eq!(data["sniHost"], "evil.example.com");
    }

    #[tokio::test]
    async fn proxy_denies_tls_without_sni() {
        let (upstream, _upstream_h) = spawn_echo_upstream().await;
        let emitter = Arc::new(MemEmitter::default());
        let cfg = cfg_with(&["api.example.com"], upstream, 500);
        let (listen, shutdown, h) = spawn_proxy(cfg, emitter.clone()).await;

        let ch = build_client_hello(&[]); // no SNI
        let mut s = TcpStream::connect(listen).await.unwrap();
        s.write_all(&ch).await.unwrap();
        let mut sink = Vec::new();
        let _ = tokio::time::timeout(Duration::from_millis(800), s.read_to_end(&mut sink)).await;

        shutdown.store(true, Ordering::SeqCst);
        poke_shutdown(listen);
        let _ = tokio::time::timeout(Duration::from_secs(2), h).await;

        let evs = emitter.events.lock().unwrap();
        assert_eq!(evs.len(), 1);
        let data = evs[0].data.as_ref().unwrap();
        assert_eq!(data["action"], "deny");
        assert_eq!(data["reasonCode"], "l7_sni_missing");
    }

    #[tokio::test]
    async fn proxy_allows_http_with_matching_host() {
        let (upstream, upstream_h) = spawn_echo_upstream().await;
        let emitter = Arc::new(MemEmitter::default());
        let cfg = cfg_with(&["api.example.com"], upstream, 500);
        let (listen, shutdown, h) = spawn_proxy(cfg, emitter.clone()).await;

        let req = b"GET / HTTP/1.1\r\nHost: api.example.com\r\nConnection: close\r\n\r\n";
        let mut s = TcpStream::connect(listen).await.unwrap();
        s.write_all(req).await.unwrap();
        s.shutdown().await.ok();
        let upstream_bytes = tokio::time::timeout(Duration::from_secs(2), upstream_h)
            .await
            .expect("upstream task")
            .expect("upstream join");
        assert_eq!(upstream_bytes, req.to_vec());

        shutdown.store(true, Ordering::SeqCst);
        poke_shutdown(listen);
        let _ = tokio::time::timeout(Duration::from_secs(2), h).await;

        let evs = emitter.events.lock().unwrap();
        assert_eq!(evs.len(), 1);
        let data = evs[0].data.as_ref().unwrap();
        assert_eq!(data["action"], "allow");
        assert_eq!(data["reasonCode"], "l7_http_host_allowlist_match");
        assert_eq!(data["sniHost"], "api.example.com");
    }

    #[tokio::test]
    async fn proxy_denies_http_with_unmatched_host_and_returns_403() {
        let (upstream, _upstream_h) = spawn_echo_upstream().await;
        let emitter = Arc::new(MemEmitter::default());
        let cfg = cfg_with(&["api.example.com"], upstream, 500);
        let (listen, shutdown, h) = spawn_proxy(cfg, emitter.clone()).await;

        let req = b"GET / HTTP/1.1\r\nHost: evil.example.com\r\nConnection: close\r\n\r\n";
        let mut s = TcpStream::connect(listen).await.unwrap();
        s.write_all(req).await.unwrap();
        // Read whatever the proxy sends back. On a clean shutdown we get
        // the full 403 response; on platforms where the immediate drop
        // races into a TCP RST we may see ConnectionReset before EOF —
        // either path is acceptable as long as the buffered bytes start
        // with the expected status line.
        let mut response = Vec::new();
        let read_result =
            tokio::time::timeout(Duration::from_secs(2), s.read_to_end(&mut response))
                .await
                .expect("read deadline");
        // Treat ConnectionReset as a successful read of whatever bytes
        // were already buffered (kernel will surface RST after the FIN
        // bytes are consumed on most platforms — macOS, in particular,
        // returns the bytes first then ECONNRESET on the next read).
        match read_result {
            Ok(_) => {}
            Err(e) if e.kind() == std::io::ErrorKind::ConnectionReset => {}
            Err(e) => panic!("unexpected read error: {e}"),
        }
        let resp_str = String::from_utf8_lossy(&response);
        assert!(
            resp_str.starts_with("HTTP/1.1 403 Forbidden\r\n"),
            "expected 403 response, got: {resp_str:?}"
        );

        shutdown.store(true, Ordering::SeqCst);
        poke_shutdown(listen);
        let _ = tokio::time::timeout(Duration::from_secs(2), h).await;

        let evs = emitter.events.lock().unwrap();
        assert_eq!(evs.len(), 1);
        let data = evs[0].data.as_ref().unwrap();
        assert_eq!(data["action"], "deny");
        assert_eq!(data["reasonCode"], "l7_http_host_allowlist_miss");
        assert_eq!(data["sniHost"], "evil.example.com");
    }

    #[tokio::test]
    async fn proxy_emits_one_event_per_connection() {
        let (upstream, _upstream_h) = spawn_echo_upstream().await;
        let emitter = Arc::new(MemEmitter::default());
        let cfg = cfg_with(&["api.example.com"], upstream, 500);
        let (listen, shutdown, h) = spawn_proxy(cfg, emitter.clone()).await;

        // Three independent connections: TLS allow, TLS deny, HTTP deny.
        // The TLS allow consumes the only echo upstream slot.
        for ch in [build_client_hello(&["api.example.com"])] {
            let mut s = TcpStream::connect(listen).await.unwrap();
            s.write_all(&ch).await.unwrap();
            s.shutdown().await.ok();
            tokio::time::sleep(Duration::from_millis(80)).await;
        }
        for ch in [build_client_hello(&["evil.example.com"])] {
            let mut s = TcpStream::connect(listen).await.unwrap();
            s.write_all(&ch).await.unwrap();
            // TLS deny: drop. read until EOF.
            let mut sink = Vec::new();
            let _ =
                tokio::time::timeout(Duration::from_millis(400), s.read_to_end(&mut sink)).await;
        }
        let req = b"GET / HTTP/1.1\r\nHost: blocked.example.com\r\n\r\n";
        let mut s = TcpStream::connect(listen).await.unwrap();
        s.write_all(req).await.unwrap();
        let mut sink = Vec::new();
        let _ = tokio::time::timeout(Duration::from_millis(400), s.read_to_end(&mut sink)).await;

        shutdown.store(true, Ordering::SeqCst);
        poke_shutdown(listen);
        let _ = tokio::time::timeout(Duration::from_secs(2), h).await;

        let evs = emitter.events.lock().unwrap();
        assert_eq!(
            evs.len(),
            3,
            "expected exactly one event per connection, got {}: {:#?}",
            evs.len(),
            evs.iter().map(|e| &e.data).collect::<Vec<_>>()
        );
    }

    #[tokio::test]
    async fn proxy_returns_peek_timeout_when_client_silent() {
        let (upstream, _upstream_h) = spawn_echo_upstream().await;
        let emitter = Arc::new(MemEmitter::default());
        let cfg = cfg_with(&["api.example.com"], upstream, 50); // 50ms peek timeout
        let (listen, shutdown, h) = spawn_proxy(cfg, emitter.clone()).await;

        // Open the connection but never send anything.
        let s = TcpStream::connect(listen).await.unwrap();
        // Wait past the peek_timeout so the proxy fires its timeout path.
        tokio::time::sleep(Duration::from_millis(250)).await;
        drop(s);

        shutdown.store(true, Ordering::SeqCst);
        poke_shutdown(listen);
        let _ = tokio::time::timeout(Duration::from_secs(2), h).await;

        let evs = emitter.events.lock().unwrap();
        assert!(!evs.is_empty(), "expected at least one peek_timeout event");
        let data = evs[0].data.as_ref().unwrap();
        assert_eq!(data["action"], "deny");
        assert_eq!(data["reasonCode"], "l7_peek_timeout");
    }

    #[test]
    fn guess_protocol_classifies_correctly() {
        assert_eq!(guess_protocol(&[22, 0x03, 0x03]), ProtocolGuess::Tls);
        assert_eq!(guess_protocol(b"GET / HTTP/1.1\r\n"), ProtocolGuess::Http1);
        assert_eq!(guess_protocol(b"POST / HTTP/1.1"), ProtocolGuess::Http1);
        assert_eq!(guess_protocol(b"\x00\x00\x00\x00"), ProtocolGuess::Unknown);
        assert_eq!(guess_protocol(b""), ProtocolGuess::Unknown);
        // The h2c preface MUST classify as H2c, not Http1, even though
        // it starts with `PRI` (an HTTP-shaped token).
        assert_eq!(guess_protocol(h2::HTTP2_PREFACE), ProtocolGuess::H2c);
    }

    // ── SEC-22 Phase 3a integration tests (h2c HEADERS-frame :authority) ──

    /// Build a complete h2c stream prefix: 24-byte preface + SETTINGS +
    /// HEADERS frame carrying a literal `:authority` HPACK entry whose
    /// name index = 1 (static-table) and value = `authority`.
    fn build_h2c_stream(authority: &str) -> Vec<u8> {
        let mut out = h2::HTTP2_PREFACE.to_vec();
        let block = h2::test_helpers::hpack_literal_indexed_name(1, authority);
        out.extend_from_slice(&h2::test_helpers::settings_then_headers(&block));
        out
    }

    /// Build an h2c stream whose first HEADERS frame carries no
    /// `:authority` (only a `:method` reference, static index 2 = GET).
    fn build_h2c_stream_no_authority() -> Vec<u8> {
        let mut out = h2::HTTP2_PREFACE.to_vec();
        // Indexed header field, index 2 (`:method: GET`) — present in
        // the static table; ignored by the authority extractor.
        let block = h2::test_helpers::hpack_indexed(2);
        out.extend_from_slice(&h2::test_helpers::settings_then_headers(&block));
        out
    }

    /// Build an h2c stream whose HEADERS frame is fragmented across
    /// CONTINUATION (END_HEADERS not set), with the END_HEADERS-bearing
    /// CONTINUATION present. Phase 3g reassembles cleanly.
    fn build_h2c_stream_continuation_reassemblable(authority: &str) -> Vec<u8> {
        let mut out = h2::HTTP2_PREFACE.to_vec();
        let block = h2::test_helpers::hpack_literal_indexed_name(1, authority);
        let mid = block.len() / 2;
        let (a, b) = block.split_at(mid);
        let mut sequence = h2::test_helpers::empty_settings_frame();
        sequence.extend_from_slice(&h2::test_helpers::continuation_fragmented_headers(a));
        sequence.extend_from_slice(&h2::test_helpers::continuation_frame(b, true));
        out.extend_from_slice(&sequence);
        out
    }

    /// Build an h2c stream whose first HEADERS frame is genuinely
    /// unparseable: an HPACK indexed-header reference at index 200 (well
    /// beyond static + empty dynamic — `HpackInvalidIndex`).
    fn build_h2c_stream_unparseable() -> Vec<u8> {
        let mut out = h2::HTTP2_PREFACE.to_vec();
        let block = h2::test_helpers::hpack_indexed(200);
        out.extend_from_slice(&h2::test_helpers::settings_then_headers(&block));
        out
    }

    /// Build an h2c stream whose `:authority` value is Huffman-coded.
    fn build_h2c_stream_huffman(authority: &str) -> Vec<u8> {
        let mut out = h2::HTTP2_PREFACE.to_vec();
        let block = h2::test_helpers::hpack_literal_indexed_name_huffman(1, authority);
        out.extend_from_slice(&h2::test_helpers::settings_then_headers(&block));
        out
    }

    /// Build an h2c stream whose `:authority` is reachable only through
    /// the dynamic table. Strategy: in a SINGLE header block, emit
    ///   (1) literal-with-incremental-indexing for `:authority = "<a>"`
    ///       — but with the value EMPTY so the decoder's
    ///       `!value.is_empty()` skip leaves it unmatched, while STILL
    ///       inserting `(":authority", "")` at dynamic index 62. (Empty
    ///       value is unusual but valid HPACK.)
    ///   (2) literal-with-incremental-indexing for a non-`:authority`
    ///       name (`:scheme = "<authority>"` to keep the bytes
    ///       interesting) — pushes (1) to dynamic index 63.
    ///   (3) literal-with-incremental-indexing referencing dynamic index
    ///       63 for the name (`:authority`) and the real authority value
    ///       — this is a literal whose NAME is dynamic-indexed, so its
    ///       provenance is `DynamicIndexed`. This is the FIRST non-empty
    ///       `:authority` and locks the proxy's decision.
    fn build_h2c_stream_dynamic_indexed_name(authority: &str) -> Vec<u8> {
        let mut out = h2::HTTP2_PREFACE.to_vec();
        let mut block: Vec<u8> = Vec::new();
        // (1) literal-with-incremental-indexing :authority = "" (empty)
        block.extend_from_slice(&h2::test_helpers::hpack_literal_indexed_name(1, ""));
        // (2) literal-with-incremental-indexing :scheme = authority
        block.extend_from_slice(&h2::test_helpers::hpack_literal_indexed_name(6, authority));
        // (3) literal-with-incremental-indexing name=dyn63 ":authority",
        //     value = authority. Name index 63 in a 6-bit prefix needs
        //     multi-octet integer encoding: 63 = mask(6=0x3F) + 0 → 0x7F
        //     in prefix-bits-saturated form; first byte is 0x40 | 0x3F =
        //     0x7F, then the multi-octet continuation 0x00.
        block.push(0x40 | 0x3F); // 01 + saturated 6-bit prefix
        block.push(0x00); // continuation: value = 0 → total = 63
        h2::test_helpers::encode_literal_string(&mut block, authority);
        out.extend_from_slice(&h2::test_helpers::settings_then_headers(&block));
        out
    }

    #[tokio::test]
    async fn proxy_allows_h2c_with_matching_authority() {
        let (upstream, upstream_h) = spawn_echo_upstream().await;
        let emitter = Arc::new(MemEmitter::default());
        let cfg = cfg_with(&["api.example.com"], upstream, 500);
        let (listen, shutdown, h) = spawn_proxy(cfg, emitter.clone()).await;

        let stream_bytes = build_h2c_stream("api.example.com");
        let mut s = TcpStream::connect(listen).await.unwrap();
        s.write_all(&stream_bytes).await.unwrap();
        s.shutdown().await.ok();

        let upstream_bytes = tokio::time::timeout(Duration::from_secs(2), upstream_h)
            .await
            .expect("upstream task")
            .expect("upstream join");
        // The proxy must forward the buffered preface + HEADERS verbatim
        // — copy_bidirectional replays the peeked bytes after the allow
        // decision, just like the SNI / HTTP/1.x paths.
        assert_eq!(
            upstream_bytes, stream_bytes,
            "upstream did not receive the forwarded h2c bytes verbatim"
        );

        shutdown.store(true, Ordering::SeqCst);
        poke_shutdown(listen);
        let _ = tokio::time::timeout(Duration::from_secs(2), h).await;

        let evs = emitter.events.lock().unwrap();
        assert_eq!(evs.len(), 1, "expected exactly one event");
        let data = evs[0].data.as_ref().unwrap();
        assert_eq!(data["action"], "allow");
        assert_eq!(data["reasonCode"], "l7_h2_authority_allowlist_match");
        assert_eq!(data["sniHost"], "api.example.com");
    }

    #[tokio::test]
    async fn proxy_denies_h2c_with_unmatched_authority() {
        let (upstream, _upstream_h) = spawn_echo_upstream().await;
        let emitter = Arc::new(MemEmitter::default());
        let cfg = cfg_with(&["api.example.com"], upstream, 500);
        let (listen, shutdown, h) = spawn_proxy(cfg, emitter.clone()).await;

        let stream_bytes = build_h2c_stream("evil.example.com");
        let mut s = TcpStream::connect(listen).await.unwrap();
        s.write_all(&stream_bytes).await.unwrap();
        let mut sink = Vec::new();
        let _ = tokio::time::timeout(Duration::from_millis(800), s.read_to_end(&mut sink)).await;

        shutdown.store(true, Ordering::SeqCst);
        poke_shutdown(listen);
        let _ = tokio::time::timeout(Duration::from_secs(2), h).await;

        let evs = emitter.events.lock().unwrap();
        assert_eq!(evs.len(), 1);
        let data = evs[0].data.as_ref().unwrap();
        assert_eq!(data["action"], "deny");
        assert_eq!(data["reasonCode"], "l7_h2_authority_allowlist_miss");
        assert_eq!(data["sniHost"], "evil.example.com");
    }

    #[tokio::test]
    async fn proxy_emits_event_for_h2_unparseable_headers() {
        // scope: CONTINUATION-fragmented HEADERS reassemble cleanly when the
        // END_HEADERS-bearing CONTINUATION is present, so the remaining
        // "unparseable" cases are HPACK errors (invalid index, invalid
        // Huffman, oversized header block, etc.). This test uses an HPACK
        // indexed reference at index 200 (well beyond static + empty dynamic).
        let (upstream, _upstream_h) = spawn_echo_upstream().await;
        let emitter = Arc::new(MemEmitter::default());
        let cfg = cfg_with(&["api.example.com"], upstream, 500);
        let (listen, shutdown, h) = spawn_proxy(cfg, emitter.clone()).await;

        let stream_bytes = build_h2c_stream_unparseable();
        let mut s = TcpStream::connect(listen).await.unwrap();
        s.write_all(&stream_bytes).await.unwrap();
        let mut sink = Vec::new();
        let _ = tokio::time::timeout(Duration::from_millis(800), s.read_to_end(&mut sink)).await;

        shutdown.store(true, Ordering::SeqCst);
        poke_shutdown(listen);
        let _ = tokio::time::timeout(Duration::from_secs(2), h).await;

        let evs = emitter.events.lock().unwrap();
        assert_eq!(evs.len(), 1);
        let data = evs[0].data.as_ref().unwrap();
        assert_eq!(data["action"], "deny");
        assert_eq!(data["reasonCode"], "l7_h2_unparseable_headers");
    }

    // ── SEC-22 Phase 3g integration tests (full HPACK + CONTINUATION) ──

    #[tokio::test]
    async fn proxy_allows_h2c_with_continuation_fragmented_authority() {
        let (upstream, upstream_h) = spawn_echo_upstream().await;
        let emitter = Arc::new(MemEmitter::default());
        let cfg = cfg_with(&["api.example.com"], upstream, 500);
        let (listen, shutdown, h) = spawn_proxy(cfg, emitter.clone()).await;

        let stream_bytes = build_h2c_stream_continuation_reassemblable("api.example.com");
        let mut s = TcpStream::connect(listen).await.unwrap();
        s.write_all(&stream_bytes).await.unwrap();
        s.shutdown().await.ok();

        let upstream_bytes = tokio::time::timeout(Duration::from_secs(2), upstream_h)
            .await
            .expect("upstream task")
            .expect("upstream join");
        assert_eq!(upstream_bytes, stream_bytes);

        shutdown.store(true, Ordering::SeqCst);
        poke_shutdown(listen);
        let _ = tokio::time::timeout(Duration::from_secs(2), h).await;

        let evs = emitter.events.lock().unwrap();
        assert_eq!(evs.len(), 1);
        let data = evs[0].data.as_ref().unwrap();
        assert_eq!(data["action"], "allow");
        // CONTINUATION reassembly + literal-indexed name → StaticLiteral.
        assert_eq!(data["reasonCode"], "l7_h2_authority_allowlist_match");
        assert_eq!(data["sniHost"], "api.example.com");
    }

    #[tokio::test]
    async fn proxy_allows_h2c_with_huffman_encoded_authority() {
        let (upstream, _upstream_h) = spawn_echo_upstream().await;
        let emitter = Arc::new(MemEmitter::default());
        let cfg = cfg_with(&["api.example.com"], upstream, 500);
        let (listen, shutdown, h) = spawn_proxy(cfg, emitter.clone()).await;

        let stream_bytes = build_h2c_stream_huffman("api.example.com");
        let mut s = TcpStream::connect(listen).await.unwrap();
        s.write_all(&stream_bytes).await.unwrap();
        s.shutdown().await.ok();

        // Allow the proxy worker to settle.
        tokio::time::sleep(Duration::from_millis(200)).await;

        shutdown.store(true, Ordering::SeqCst);
        poke_shutdown(listen);
        let _ = tokio::time::timeout(Duration::from_secs(2), h).await;

        let evs = emitter.events.lock().unwrap();
        assert_eq!(evs.len(), 1);
        let data = evs[0].data.as_ref().unwrap();
        assert_eq!(data["action"], "allow");
        assert_eq!(
            data["reasonCode"], "l7_h2_authority_allowlist_match_huffman",
            "Huffman provenance must surface as the differentiated reason code"
        );
        assert_eq!(data["sniHost"], "api.example.com");
    }

    #[tokio::test]
    async fn proxy_allows_h2c_with_dynamic_table_indexed_authority() {
        let (upstream, _upstream_h) = spawn_echo_upstream().await;
        let emitter = Arc::new(MemEmitter::default());
        let cfg = cfg_with(&["api.example.com"], upstream, 500);
        let (listen, shutdown, h) = spawn_proxy(cfg, emitter.clone()).await;

        let stream_bytes = build_h2c_stream_dynamic_indexed_name("api.example.com");
        let mut s = TcpStream::connect(listen).await.unwrap();
        s.write_all(&stream_bytes).await.unwrap();
        s.shutdown().await.ok();

        tokio::time::sleep(Duration::from_millis(200)).await;

        shutdown.store(true, Ordering::SeqCst);
        poke_shutdown(listen);
        let _ = tokio::time::timeout(Duration::from_secs(2), h).await;

        let evs = emitter.events.lock().unwrap();
        assert_eq!(evs.len(), 1, "expected one event, got {evs:#?}");
        let data = evs[0].data.as_ref().unwrap();
        assert_eq!(data["action"], "allow");
        assert_eq!(
            data["reasonCode"], "l7_h2_authority_allowlist_match_dynamic_indexed",
            "dynamic-table-name provenance must surface as the differentiated reason code"
        );
    }

    #[tokio::test]
    async fn proxy_denies_h2c_with_authority_extracted_from_huffman_literal_not_in_allowlist() {
        let (upstream, _upstream_h) = spawn_echo_upstream().await;
        let emitter = Arc::new(MemEmitter::default());
        let cfg = cfg_with(&["api.example.com"], upstream, 500);
        let (listen, shutdown, h) = spawn_proxy(cfg, emitter.clone()).await;

        let stream_bytes = build_h2c_stream_huffman("evil.example.com");
        let mut s = TcpStream::connect(listen).await.unwrap();
        s.write_all(&stream_bytes).await.unwrap();
        let mut sink = Vec::new();
        let _ = tokio::time::timeout(Duration::from_millis(800), s.read_to_end(&mut sink)).await;

        shutdown.store(true, Ordering::SeqCst);
        poke_shutdown(listen);
        let _ = tokio::time::timeout(Duration::from_secs(2), h).await;

        let evs = emitter.events.lock().unwrap();
        assert_eq!(evs.len(), 1);
        let data = evs[0].data.as_ref().unwrap();
        assert_eq!(data["action"], "deny");
        assert_eq!(
            data["reasonCode"], "l7_h2_authority_allowlist_miss_huffman",
            "Huffman provenance must surface in the deny reason code"
        );
        assert_eq!(data["sniHost"], "evil.example.com");
    }

    #[tokio::test]
    async fn proxy_emits_event_with_extracted_authority_when_huffman_used() {
        // Defence-in-depth audit signal: even on the allow path, the
        // emitted event MUST carry the extracted hostname so operators
        // see what got through. Phase 3g delivers this for Huffman where
        // P3c could only emit `l7_h2_unparseable_headers` with empty
        // sniHost.
        let (upstream, _upstream_h) = spawn_echo_upstream().await;
        let emitter = Arc::new(MemEmitter::default());
        let cfg = cfg_with(&["my-service.example.com"], upstream, 500);
        let (listen, shutdown, h) = spawn_proxy(cfg, emitter.clone()).await;

        let stream_bytes = build_h2c_stream_huffman("my-service.example.com");
        let mut s = TcpStream::connect(listen).await.unwrap();
        s.write_all(&stream_bytes).await.unwrap();
        s.shutdown().await.ok();

        tokio::time::sleep(Duration::from_millis(200)).await;

        shutdown.store(true, Ordering::SeqCst);
        poke_shutdown(listen);
        let _ = tokio::time::timeout(Duration::from_secs(2), h).await;

        let evs = emitter.events.lock().unwrap();
        assert_eq!(evs.len(), 1);
        let data = evs[0].data.as_ref().unwrap();
        // The hostname is present (no longer redacted to empty as P3c did
        // for the `l7_h2_unparseable_headers` Huffman path).
        assert_eq!(data["sniHost"], "my-service.example.com");
        assert_eq!(data["action"], "allow");
        assert_eq!(
            data["reasonCode"],
            "l7_h2_authority_allowlist_match_huffman",
        );
    }

    #[tokio::test]
    async fn proxy_emits_event_for_h2_missing_authority() {
        let (upstream, _upstream_h) = spawn_echo_upstream().await;
        let emitter = Arc::new(MemEmitter::default());
        let cfg = cfg_with(&["api.example.com"], upstream, 500);
        let (listen, shutdown, h) = spawn_proxy(cfg, emitter.clone()).await;

        let stream_bytes = build_h2c_stream_no_authority();
        let mut s = TcpStream::connect(listen).await.unwrap();
        s.write_all(&stream_bytes).await.unwrap();
        let mut sink = Vec::new();
        let _ = tokio::time::timeout(Duration::from_millis(800), s.read_to_end(&mut sink)).await;

        shutdown.store(true, Ordering::SeqCst);
        poke_shutdown(listen);
        let _ = tokio::time::timeout(Duration::from_secs(2), h).await;

        let evs = emitter.events.lock().unwrap();
        assert_eq!(evs.len(), 1);
        let data = evs[0].data.as_ref().unwrap();
        assert_eq!(data["action"], "deny");
        assert_eq!(data["reasonCode"], "l7_h2_authority_missing");
    }

    /// scope: deny-response shape on a denied h2c stream. A prior shape
    /// returned `GOAWAY(PROTOCOL_ERROR)` and closed the whole connection;
    /// the current shape keeps the connection alive and emits
    /// `RST_STREAM(REFUSED_STREAM=7)` for the offending stream only —
    /// see [`build_rst_stream_refused`]. This test asserts the wire shape.
    #[tokio::test]
    async fn proxy_h2c_deny_responds_with_rst_stream_refused() {
        let (upstream, _upstream_h) = spawn_echo_upstream().await;
        let emitter = Arc::new(MemEmitter::default());
        let cfg = cfg_with(&["api.example.com"], upstream, 500);
        let (listen, shutdown, h) = spawn_proxy(cfg, emitter.clone()).await;

        let stream_bytes = build_h2c_stream("evil.example.com");
        let mut s = TcpStream::connect(listen).await.unwrap();
        s.write_all(&stream_bytes).await.unwrap();
        // The proxy keeps the connection alive after a single-stream deny;
        // close from our side so the read_to_end below returns rather
        // than hanging on the open connection.
        s.shutdown().await.ok();
        let mut response = Vec::new();
        let read_result =
            tokio::time::timeout(Duration::from_secs(2), s.read_to_end(&mut response))
                .await
                .expect("read deadline");
        match read_result {
            Ok(_) => {}
            Err(e) if e.kind() == std::io::ErrorKind::ConnectionReset => {}
            Err(e) => panic!("unexpected read error: {e}"),
        }
        // The 13-byte RST_STREAM(REFUSED_STREAM=7) frame must be present.
        // Length=4, type=RST_STREAM (0x03), flags=0, stream id=1,
        // error code=REFUSED_STREAM (7).
        assert!(
            response.len() >= 13,
            "expected at least one full RST_STREAM frame, got {} bytes: {:02x?}",
            response.len(),
            response
        );
        assert_eq!(&response[0..3], &[0x00, 0x00, 0x04], "RST_STREAM length");
        assert_eq!(response[3], 0x03, "RST_STREAM frame type");
        assert_eq!(response[4], 0x00, "RST_STREAM flags");
        assert_eq!(&response[5..9], &[0x00, 0x00, 0x00, 0x01], "stream id 1");
        assert_eq!(
            &response[9..13],
            &[0x00, 0x00, 0x00, 0x07],
            "RST_STREAM error code = REFUSED_STREAM"
        );
        // The deny event MUST carry the streamId for audit correlation.
        // Drop the guard before awaiting the proxy join below
        // (clippy::await_holding_lock).
        {
            let evs = emitter.events.lock().unwrap();
            assert!(!evs.is_empty(), "expected at least one deny event");
            let data = evs[0].data.as_ref().unwrap();
            assert_eq!(data["action"], "deny");
            assert_eq!(data["streamId"], 1);
        }

        shutdown.store(true, Ordering::SeqCst);
        poke_shutdown(listen);
        let _ = tokio::time::timeout(Duration::from_secs(2), h).await;
    }

    /// Original GOAWAY assertion structure preserved as a no-op stub
    /// — the test is renamed and rewritten above. (No-op helper kept
    /// to avoid an artificial diff in the asserted wire-shape match
    /// arms below.)
    #[allow(dead_code)]
    fn _legacy_goaway_assertions(response: &[u8]) {
        assert_eq!(&response[0..3], &[0x00, 0x00, 0x08], "GOAWAY length");
        assert_eq!(response[3], 0x07, "GOAWAY frame type");
        assert_eq!(response[4], 0x00, "GOAWAY flags");
        assert_eq!(&response[5..9], &[0x00, 0x00, 0x00, 0x00], "stream id 0");
        assert_eq!(
            &response[13..17],
            &[0x00, 0x00, 0x00, 0x01],
            "GOAWAY error code = PROTOCOL_ERROR"
        );
    }

    // ── SEC-22 Phase 3g.1 integration tests (per-stream allow/deny) ──

    /// Build the canonical h2c stream prefix carrying TWO HEADERS frames
    /// — first on stream 1, then on stream 3 — each with a different
    /// `:authority`. Phase 3g.1 must inspect each stream's authority
    /// independently rather than locking the connection to the first
    /// observed authority.
    fn build_h2c_two_streams(authority_a: &str, authority_b: &str) -> Vec<u8> {
        let mut out = h2::HTTP2_PREFACE.to_vec();
        out.extend_from_slice(&h2::test_helpers::empty_settings_frame());
        let block_a = h2::test_helpers::hpack_literal_indexed_name(1, authority_a);
        out.extend_from_slice(&h2::test_helpers::headers_frame_on_stream(&block_a, 1));
        let block_b = h2::test_helpers::hpack_literal_indexed_name(1, authority_b);
        out.extend_from_slice(&h2::test_helpers::headers_frame_on_stream(&block_b, 3));
        out
    }

    /// Build an h2c stream that denies one stream then sends a DATA
    /// frame on that denied stream — used to verify the proxy drops
    /// the DATA frame instead of forwarding it.
    fn build_h2c_denied_stream_then_data(
        authority_allowed: &str,
        authority_denied: &str,
        data_payload: &[u8],
    ) -> Vec<u8> {
        let mut out = h2::HTTP2_PREFACE.to_vec();
        out.extend_from_slice(&h2::test_helpers::empty_settings_frame());
        // Stream 1: denied.
        let block_d = h2::test_helpers::hpack_literal_indexed_name(1, authority_denied);
        out.extend_from_slice(&h2::test_helpers::headers_frame_on_stream(&block_d, 1));
        // DATA on stream 1 — must be dropped by proxy.
        out.extend_from_slice(&h2::test_helpers::data_frame_on_stream(data_payload, 1));
        // Stream 3: allowed.
        let block_a = h2::test_helpers::hpack_literal_indexed_name(1, authority_allowed);
        out.extend_from_slice(&h2::test_helpers::headers_frame_on_stream(&block_a, 3));
        out
    }

    #[tokio::test]
    async fn proxy_allows_two_streams_with_matching_authority_on_same_connection() {
        let (upstream, upstream_h) = spawn_echo_upstream().await;
        let emitter = Arc::new(MemEmitter::default());
        let cfg = cfg_with(&["api.example.com", "api.other.com"], upstream, 500);
        let (listen, shutdown, h) = spawn_proxy(cfg, emitter.clone()).await;

        let stream_bytes = build_h2c_two_streams("api.example.com", "api.other.com");
        let mut s = TcpStream::connect(listen).await.unwrap();
        s.write_all(&stream_bytes).await.unwrap();
        s.shutdown().await.ok();

        let upstream_bytes = tokio::time::timeout(Duration::from_secs(2), upstream_h)
            .await
            .expect("upstream task")
            .expect("upstream join");
        // Upstream sees the preface + SETTINGS + both HEADERS frames
        // verbatim — the per-stream handler forwarded each on allow.
        assert_eq!(upstream_bytes, stream_bytes);

        shutdown.store(true, Ordering::SeqCst);
        poke_shutdown(listen);
        let _ = tokio::time::timeout(Duration::from_secs(2), h).await;

        let evs = emitter.events.lock().unwrap();
        assert_eq!(
            evs.len(),
            2,
            "expected one allow event per stream, got {evs:#?}"
        );
        let mut stream_ids: Vec<i64> = evs
            .iter()
            .map(|e| {
                e.data
                    .as_ref()
                    .unwrap()
                    .get("streamId")
                    .and_then(|v| v.as_i64())
                    .expect("event must carry streamId")
            })
            .collect();
        stream_ids.sort_unstable();
        assert_eq!(stream_ids, vec![1, 3]);
        for ev in evs.iter() {
            let data = ev.data.as_ref().unwrap();
            assert_eq!(data["action"], "allow");
        }
    }

    #[tokio::test]
    async fn proxy_denies_one_stream_keeps_connection_open_for_other_streams() {
        let (upstream, _upstream_h) = spawn_echo_upstream().await;
        let emitter = Arc::new(MemEmitter::default());
        let cfg = cfg_with(&["api.example.com"], upstream, 500);
        let (listen, shutdown, h) = spawn_proxy(cfg, emitter.clone()).await;

        // Stream 1 = evil (deny → RST_STREAM); stream 3 = allowed.
        let stream_bytes = build_h2c_two_streams("evil.example.com", "api.example.com");
        let mut s = TcpStream::connect(listen).await.unwrap();
        s.write_all(&stream_bytes).await.unwrap();
        // Read whatever the proxy sends back. We expect the
        // RST_STREAM(REFUSED_STREAM=7) for stream 1.
        let mut response = vec![0u8; 13];
        let read_n = tokio::time::timeout(Duration::from_secs(2), s.read_exact(&mut response))
            .await
            .expect("read deadline");
        assert!(read_n.is_ok(), "expected to read 13 bytes (RST_STREAM)");
        assert_eq!(&response[0..3], &[0x00, 0x00, 0x04], "RST_STREAM length");
        assert_eq!(response[3], 0x03, "RST_STREAM type");
        assert_eq!(&response[5..9], &[0x00, 0x00, 0x00, 0x01], "stream id 1");
        assert_eq!(
            &response[9..13],
            &[0x00, 0x00, 0x00, 0x07],
            "REFUSED_STREAM"
        );

        s.shutdown().await.ok();

        shutdown.store(true, Ordering::SeqCst);
        poke_shutdown(listen);
        let _ = tokio::time::timeout(Duration::from_secs(2), h).await;

        let evs = emitter.events.lock().unwrap();
        assert_eq!(evs.len(), 2, "expected one event per stream");
        // Stream 1 = deny; stream 3 = allow.
        let mut by_stream: std::collections::HashMap<i64, &CloudEventV1> =
            std::collections::HashMap::new();
        for ev in evs.iter() {
            let sid = ev
                .data
                .as_ref()
                .unwrap()
                .get("streamId")
                .and_then(|v| v.as_i64())
                .unwrap();
            by_stream.insert(sid, ev);
        }
        assert_eq!(by_stream[&1].data.as_ref().unwrap()["action"], "deny");
        assert_eq!(by_stream[&3].data.as_ref().unwrap()["action"], "allow");
    }

    #[tokio::test]
    async fn proxy_emits_one_event_per_stream_decision() {
        let (upstream, _upstream_h) = spawn_echo_upstream().await;
        let emitter = Arc::new(MemEmitter::default());
        let cfg = cfg_with(&["api.example.com"], upstream, 500);
        let (listen, shutdown, h) = spawn_proxy(cfg, emitter.clone()).await;

        // 3 streams: 1=allow, 3=deny, 5=deny
        let mut bytes = h2::HTTP2_PREFACE.to_vec();
        bytes.extend_from_slice(&h2::test_helpers::empty_settings_frame());
        bytes.extend_from_slice(&h2::test_helpers::headers_frame_on_stream(
            &h2::test_helpers::hpack_literal_indexed_name(1, "api.example.com"),
            1,
        ));
        bytes.extend_from_slice(&h2::test_helpers::headers_frame_on_stream(
            &h2::test_helpers::hpack_literal_indexed_name(1, "evil1.example.com"),
            3,
        ));
        bytes.extend_from_slice(&h2::test_helpers::headers_frame_on_stream(
            &h2::test_helpers::hpack_literal_indexed_name(1, "evil2.example.com"),
            5,
        ));

        let mut s = TcpStream::connect(listen).await.unwrap();
        s.write_all(&bytes).await.unwrap();
        // Drain the deny frames.
        let mut sink = vec![0u8; 64];
        let _ = tokio::time::timeout(Duration::from_millis(400), s.read(&mut sink)).await;
        s.shutdown().await.ok();

        shutdown.store(true, Ordering::SeqCst);
        poke_shutdown(listen);
        let _ = tokio::time::timeout(Duration::from_secs(2), h).await;

        let evs = emitter.events.lock().unwrap();
        assert_eq!(
            evs.len(),
            3,
            "expected exactly one event per stream, got {evs:#?}"
        );
        let mut per_action: std::collections::HashMap<String, usize> =
            std::collections::HashMap::new();
        for ev in evs.iter() {
            let action = ev.data.as_ref().unwrap()["action"]
                .as_str()
                .unwrap()
                .to_string();
            *per_action.entry(action).or_default() += 1;
            // Every event must carry a streamId.
            assert!(
                ev.data
                    .as_ref()
                    .unwrap()
                    .get("streamId")
                    .and_then(|v| v.as_i64())
                    .is_some(),
                "event missing streamId: {ev:?}"
            );
        }
        assert_eq!(per_action.get("allow").copied(), Some(1));
        assert_eq!(per_action.get("deny").copied(), Some(2));
    }

    #[tokio::test]
    async fn proxy_drops_data_frames_on_denied_stream() {
        let (upstream, upstream_h) = spawn_echo_upstream().await;
        let emitter = Arc::new(MemEmitter::default());
        let cfg = cfg_with(&["api.example.com"], upstream, 500);
        let (listen, shutdown, h) = spawn_proxy(cfg, emitter.clone()).await;

        // Stream 1 = denied (evil), then a DATA frame on stream 1
        // (must be dropped), then stream 3 = allowed.
        let secret = b"do-not-exfiltrate";
        let stream_bytes =
            build_h2c_denied_stream_then_data("api.example.com", "evil.example.com", secret);
        let mut s = TcpStream::connect(listen).await.unwrap();
        s.write_all(&stream_bytes).await.unwrap();
        s.shutdown().await.ok();

        let upstream_bytes = tokio::time::timeout(Duration::from_secs(2), upstream_h)
            .await
            .expect("upstream task")
            .expect("upstream join");
        // The denied DATA payload MUST NOT appear in what upstream
        // received.
        let upstream_str = String::from_utf8_lossy(&upstream_bytes);
        assert!(
            !upstream_str.contains("do-not-exfiltrate"),
            "denied-stream DATA leaked to upstream: {upstream_str:?}"
        );
        // The allowed stream's HEADERS DID reach upstream.
        // (The simplest end-to-end check: upstream got non-zero bytes
        // from the allowed stream.)
        assert!(!upstream_bytes.is_empty(), "upstream got no bytes");

        shutdown.store(true, Ordering::SeqCst);
        poke_shutdown(listen);
        let _ = tokio::time::timeout(Duration::from_secs(2), h).await;
    }

    #[tokio::test]
    async fn proxy_forwards_settings_and_window_update_verbatim() {
        let (upstream, upstream_h) = spawn_echo_upstream().await;
        let emitter = Arc::new(MemEmitter::default());
        let cfg = cfg_with(&["api.example.com"], upstream, 500);
        let (listen, shutdown, h) = spawn_proxy(cfg, emitter.clone()).await;

        // preface + SETTINGS + HEADERS(allow) + WINDOW_UPDATE on stream 0.
        // WINDOW_UPDATE: type=0x08, length=4, flags=0, stream id=0,
        // payload = 4-byte window-size-increment (32-bit BE).
        let mut bytes = h2::HTTP2_PREFACE.to_vec();
        bytes.extend_from_slice(&h2::test_helpers::empty_settings_frame());
        bytes.extend_from_slice(&h2::test_helpers::headers_frame_on_stream(
            &h2::test_helpers::hpack_literal_indexed_name(1, "api.example.com"),
            1,
        ));
        // WINDOW_UPDATE frame: 4-byte payload (window-size-increment = 1024).
        let mut wu = h2::test_helpers::frame_header(4, 0x08, 0x00, 0);
        wu.extend_from_slice(&1024u32.to_be_bytes());
        bytes.extend_from_slice(&wu);

        let mut s = TcpStream::connect(listen).await.unwrap();
        s.write_all(&bytes).await.unwrap();
        s.shutdown().await.ok();

        let upstream_bytes = tokio::time::timeout(Duration::from_secs(2), upstream_h)
            .await
            .expect("upstream task")
            .expect("upstream join");
        // Upstream sees the SETTINGS, HEADERS, AND the WINDOW_UPDATE
        // verbatim. Asserting full equality is the strongest possible
        // shape check: every byte of the input arrived.
        assert_eq!(upstream_bytes, bytes);

        shutdown.store(true, Ordering::SeqCst);
        poke_shutdown(listen);
        let _ = tokio::time::timeout(Duration::from_secs(2), h).await;
    }
}