openlatch-client 0.3.3

//! The one hot path: `proxy_any`. Order matters.
//!
//! Two forwarding modes:
//! - **Opaque** (`stream_through_opaque`): anything we do not capture, or a body
//!   too big / of unknown length. The request body streams straight to the
//!   upstream with no permit and no materialization — it can never OOM.
//! - **Materialized** (`forward_streaming`): a bounded `/v1/messages` body of
//!   known length ≤ 32 MB. Read once as `Bytes`, observed on a **clone**, then
//!   forwarded **verbatim** so `cache_control` breakpoints round-trip
//!   byte-identically (D-06). Bounded by the in-flight semaphore (D-03).
//!
//! ## The one place different bytes can leave (D-28)
//!
//! Between observation and forward sits [`apply_transform`], the **only** step
//! in this file that can produce a body the agent did not send. It is inert
//! unless `[boundary] transforms_act` is on (it ships **off**) and an authored
//! `enforce` L-0 rule matched a viable request. Everything about it is built to
//! fail towards the original bytes:
//!
//! - It runs under its **own** `catch_unwind`, separate from observation's. A
//!   panic here degrades to forwarding the original — and, because the guard is
//!   separate, the measurement that already succeeded is *kept* rather than
//!   thrown away with it.
//! - It returns a complete `Bytes` or nothing. There is no path on which a
//!   partially-built buffer reaches the wire.
//! - The opaque path cannot mutate **by construction**: it never reads the body
//!   (F-17 / zero-buffer), so there is nothing to transform. That is recorded as
//!   the coverage gap it is — such a request already carries
//!   `unknown_wire_format` — rather than forcing materialization, which would
//!   trade a bounded missed saving for an unbounded memory risk.
//!
//! ### Observation is ordered before mutation, deliberately
//!
//! `churn.observe`, `request_body_len`, `has_breakpoint` and the pricing inputs
//! are all computed on the **original** body, and stay that way. Churn is the
//! load-bearing case: it measures what the *agent* does, which is the subject of
//! every prefix finding and of the savings recommendation those findings
//! produce. Feeding it our own edits would make the next request diff against a
//! body the agent never sent, and prefix findings would start describing — and
//! recommending fixes for — a problem we created. The mutation is recorded as a
//! **separate fact** instead: the `ai.openlatch.transform.*` tuple, where
//! `outcome = applied` and `ladder_stage = act` say exactly what we changed.
//! A consumer can then see "the agent sent no breakpoint **and** we inserted
//! one", which a mutated `has_breakpoint` would have hidden.
//!
//! Both hand the upstream byte stream straight to axum — the first response
//! byte reaches the agent before the last byte arrives from the provider
//! (D-06 / C-9b, zero buffering). On a connection failure there is no provider
//! response to pass through, so we synthesize the single unavoidable
//! OpenLatch-shaped signal: `502` + `x-openlatch-upstream: unreachable` (C-5b).

use std::pin::Pin;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::task::{Context, Poll};

use axum::body::Body;
use axum::extract::{Request, State};
use axum::http::header::{
    ACCEPT_ENCODING, CONNECTION, CONTENT_LENGTH, HOST, PROXY_AUTHENTICATE, PROXY_AUTHORIZATION, TE,
    TRAILER, TRANSFER_ENCODING, UPGRADE,
};
use axum::http::{request::Parts, HeaderMap, HeaderValue, StatusCode};
use axum::response::Response;
use axum::Json;
use bytes::Bytes;
use futures_core::Stream;
use tokio::sync::mpsc::Sender;
use tokio::sync::OwnedSemaphorePermit;

use crate::cloud::CloudEvent;
use crate::privacy::PrivacyFilter;

use super::billing::detect_billing;
use super::capture::{self, CaptureGap, CostBasis, Usage, UsageAccumulator};
use super::emit::{self, Observation};
use super::preflight::PREFLIGHT_HEADER;
use super::session::{self, resolve_session};
use super::tokenize::{classify_model, Estimator};
use super::wire_format::{self, AuthMode, WireFormat};
use super::{BoundaryState, MAX_MATERIALIZE_BYTES};

/// The request header carrying the stable per-install id (written at `init` into
/// `ANTHROPIC_CUSTOM_HEADERS`; it round-trips on every model request).
const INSTALL_ID_HEADER: &str = "x-openlatch-install-id";

/// Count of pass-through failures — a fallible step (body read, observe panic)
/// degraded to forwarding-unmodified. This is the "failure recorded" signal the
/// D-24 bench reads; it is NOT the global `daemon_crashed` telemetry event.
static PASS_THROUGH_FAILURES: AtomicU64 = AtomicU64::new(0);

/// Count of requests the forwarder could not forward **at all** — no upstream
/// response existed, so it answered with the synthetic 502 (C-5b).
///
/// Deliberately NOT folded into [`PASS_THROUGH_FAILURES`]. That counter means "a
/// fallible OpenLatch step degraded to forwarding unmodified", which is still a
/// success from the agent's side. This one means the agent got nothing back, and
/// it is the signal the daemon's wiring supervisor watches: real traffic
/// failing is what makes it re-probe, so a healthy boundary is never asked to
/// prove itself on a timer.
static UPSTREAM_FAILURES: AtomicU64 = AtomicU64::new(0);

/// Test/bench hook: when set, `observe_request` panics. Proves D-24 panic
/// isolation — a panic in the (future) measurement path must never corrupt the
/// forward or take down the listener. In production this is always `false`;
/// `observe_request` is a no-op stub until plan 02 fills it in.
static INJECT_OBSERVE_PANIC: AtomicBool = AtomicBool::new(false);

/// Test hook: when set, the response usage scanner ([`GuardedBody`]'s tee)
/// panics. Proves FIX 3 — a panic in the response-side scan degrades to an
/// UNMEASURED forward: the stream still flows byte-identically and no
/// corrupt/partial event is emitted. Its only caller is the unit test, so the
/// static + setter + poll-site load are `#[cfg(test)]`-gated out of release —
/// release ships neither the backdoor nor its per-chunk atomic load. The
/// production scan guard (`catch_unwind`) stays; only the injection trigger is
/// gated.
#[cfg(test)]
static INJECT_SCAN_PANIC: AtomicBool = AtomicBool::new(false);

/// Total pass-through failures recorded this process lifetime.
pub fn pass_through_failures() -> u64 {
    PASS_THROUGH_FAILURES.load(Ordering::Relaxed)
}

/// Total requests answered with the synthetic 502 this process lifetime — every
/// one of them a request the agent did not get an answer to.
pub fn upstream_failures() -> u64 {
    UPSTREAM_FAILURES.load(Ordering::Relaxed)
}

/// Arm/disarm the observe-panic injection (D-24 bench only).
pub fn set_inject_observe_panic(on: bool) {
    INJECT_OBSERVE_PANIC.store(on, Ordering::Relaxed);
}

// The mutation-panic injection is a **field on `BoundaryState`**
// (`inject_mutate_panic`), not a process-wide static like
// `INJECT_OBSERVE_PANIC` above. That static predates measurement state and is
// serialized by having exactly one caller; a second global flag would be armed
// for every listener in the process, and integration tests run in parallel
// threads sharing one — arming it in the panic-isolation test broke the
// unrelated acting tests running beside it. Per-state keeps each listener's
// fault injection to itself.

/// Arm/disarm the usage-scan-panic injection (FIX 3 regression test only).
#[cfg(test)]
pub fn set_inject_scan_panic(on: bool) {
    INJECT_SCAN_PANIC.store(on, Ordering::Relaxed);
}

fn record_pass_through_failure(reason: &'static str) {
    PASS_THROUGH_FAILURES.fetch_add(1, Ordering::Relaxed);
    // Never log the body or the credential — only the reason label (F-22).
    tracing::warn!(
        reason,
        "boundary pass-through failure — forwarding unmodified"
    );
}

/// Observe a request on a **clone** of its bytes (never mutating the original).
///
/// Stages every request-side fact of the economics event: model, billing mode,
/// the resolved attribution triple + assurance, the prefix-churn classification,
/// the pricing-input modifiers, the idempotency id, and the request-start
/// timestamp. The response side ([`GuardedBody`]) completes it with usage tokens.
///
/// **Synchronous, no `.await`** — it composes with `catch_unwind` (D-02): a panic
/// here is caught by the caller and degrades to pass-through, unmeasured (D-24).
/// It reads `bytes` only (never mutates them, D-06) and makes **no network call**
/// (D-08) — the zero-egress property the capture bench asserts.
fn observe_request(
    st: &BoundaryState,
    wire: WireFormat,
    headers: &HeaderMap,
    bytes: &Bytes,
) -> (Observation, Option<serde_json::Value>) {
    // D-24 injection point — arms only under the bench, never in production.
    if INJECT_OBSERVE_PANIC.load(Ordering::Relaxed) {
        panic!("injected observe panic (D-24 bench)");
    }

    let occurred_at = crate::envelope::current_timestamp();
    let event_id = uuid::Uuid::now_v7().to_string();

    // Parse the body on the clone. A malformed body leaves `model` None — the
    // provider will reject it (→ provider_error at finalize); we do not guess.
    let body: serde_json::Value = serde_json::from_slice(bytes).unwrap_or(serde_json::Value::Null);
    let model = capture::model_of(&body);
    let model_known = model
        .as_deref()
        .map(|m| classify_model(m).is_some())
        .unwrap_or(false);

    let billing = detect_billing(headers);
    let pricing = capture::derive_pricing_inputs(&body, headers);

    let install_id = headers
        .get(INSTALL_ID_HEADER)
        .and_then(|v| v.to_str().ok())
        .map(str::to_string)
        .unwrap_or_default();

    // Resolve the attribution triple + assurance in-process (B-2) against the
    // shared registry the hook side stamps. The request's own content picks
    // between concurrent sessions instead of "whichever hook fired last" (D-29).
    //
    // The route-resolved format is threaded from the handler: where a body keeps
    // its session id, its user turns and its tool results is an agent's
    // convention, so only the format can read them.
    let signals = request_signals(wire, &body);
    let session = session::resolve_session_with(&st.registry, &install_id, &signals);

    // Prefix-churn vs the previous request in this (install, session). The block
    // content is written to the LOCAL retention store; only the classification +
    // offsets travel on the wire (F-34).
    let session_key = session.session_id.clone().unwrap_or_default();
    let churn = st.churn.observe(&install_id, &session_key, bytes);
    if let Some(f) = &churn {
        retention_store(f, &occurred_at);
    }

    // Would-have transform eval (I-3-01): evaluate the L-1/L-2 baseline rules on the
    // parsed `body` CLONE — never on `bytes`. Synchronous, NO network (D-08), and
    // behind the caller's catch_unwind (D-24): a panic here degrades to an
    // unmeasured pass-through. Observe-only — it measures what a trim WOULD do; the
    // forwarded bytes are never touched (D-06). An L-0-only request matches no rule
    // → `None` → zero transform events (D-03).
    //
    // Bundle-authored request rules take precedence over the boundary-local
    // baseline — the baseline is the stand-in for "no bundle has arrived yet",
    // so once real rules exist it must not compete for the single decision
    // slot (C-17). One `ArcSwap` load per request, no clone of the rule set:
    // the same lock-free read the hook verdict path uses.
    let resident = st.resident_request_rules();
    let authored: &[crate::generated::types::PolicyRule] = resident
        .as_ref()
        .and_then(|guard| guard.as_ref().as_ref())
        .map(|bundle| bundle.request_rules.as_slice())
        .unwrap_or(&[]);
    let transform = super::transforms::evaluate_would_have_with(&body, authored);

    // The parsed body is handed on ONLY when this host may act — the mutation
    // step needs it and re-parsing up to 32 MB a second time would be a real
    // cost. With `transforms_act` off it is dropped here, so the default path
    // retains nothing extra and does exactly the work it did before D-28.
    let carry = st.transforms_act.then_some(body);

    (
        Observation {
            measured: true,
            event_id,
            occurred_at,
            model,
            model_known,
            billing,
            install_id: install_id.clone(),
            session,
            pricing,
            churn,
            // Deliberately the ORIGINAL body's facts — see the module doc's
            // "Observation is ordered before mutation".
            request_body_len: bytes.len(),
            has_breakpoint: capture::has_cache_breakpoint(bytes),
            transform,
            wire_format: wire,
            // Read here, where the wiring registry is in scope, so `emit` stays
            // a pure assembler over facts it was handed.
            attributable_agent: st.wiring.sole_wired_agent_for(wire),
        },
        carry,
    )
}

/// Evaluate the acting L-0 path, and rewrite the request when it fires (D-28).
///
/// Three distinguishable answers, because two of them are different facts:
///
/// | Return | Meaning |
/// | ------ | ------- |
/// | `None` | Nothing to say. The host did not opt in, no bundle is resident, or no `prefix_reorder` rule matched this request's shape. |
/// | `Some((None, decision))` | A rule matched but did not fire — net ≤ 0, structurally invalid, or authored `observe`. **Recorded**, so an operator who turned this on can see why it is not firing. Original bytes forwarded. |
/// | `Some((Some(bytes), decision))` | It fired. Forward these bytes. |
///
/// The middle row is why this returns the decision separately from the body: a
/// rule that measured a viable transform and then declined is information the
/// operator asked for by enabling the flag, and dropping it would leave the
/// console with nothing to explain the silence.
///
/// **Synchronous, network-free, and side-effect-free on failure.** It composes
/// with the caller's `catch_unwind` the same way `observe_request` does, and any
/// `Bytes` it returns are built completely before it returns — the caller can
/// only ever choose between two whole bodies.
#[allow(clippy::type_complexity)]
fn apply_transform(
    st: &BoundaryState,
    obs: &Observation,
    body: &serde_json::Value,
) -> Option<(Option<Bytes>, super::transforms::TransformDecision)> {
    // D-28 injection point — armed per listener by the bench/test, never in
    // production. Placed at the top deliberately: `apply_transform` builds its
    // `Bytes` whole and only at the very end, so there is no incremental buffer
    // for a panic to leave half-written. What the guard protects is the process
    // and the original bytes, and where inside the function it fires does not
    // change that proof.
    if st.inject_mutate_panic {
        panic!("injected mutate panic (D-28)");
    }

    let resident = st.resident_request_rules();
    let bundle = resident
        .as_ref()
        .and_then(|guard| guard.as_ref().as_ref())?;
    let authored = bundle.request_rules.as_slice();
    // No authored request rules ⇒ nothing here can act. Returning early rather
    // than falling through matters: `evaluate_acting` treats an empty slice as
    // "no bundle" and falls back to `BASELINE_RULES` (which carries no L-0
    // rule), so continuing would re-evaluate the same L-1/L-2 baseline the
    // observe step already ran and overwrite its own decision with a copy.
    if authored.is_empty() {
        return None;
    }

    // The organization-wide kill switch, honoured here for the same reason the
    // command plane honours it at evaluate time: `enforcement_enabled = false`
    // means every rule shadows, whatever its authored mode. An org that has
    // pulled that switch must not have its requests rewritten — and the switch
    // is a bundle property, so it cannot be baked into a rule at projection.
    let enforcing = bundle.enforcement_enabled;

    // The measured cross-request shape of this session's prefix — the horizon
    // L-0's net model needs, and the per-block stability a reorder needs. Folded
    // in here rather than during observation so that turning the flag off costs
    // nothing at all.
    let session_key = obs.session.session_id.clone().unwrap_or_default();
    let shape = st.prefix_shape.observe(&obs.install_id, &session_key, body);

    let evaluated = super::transforms::evaluate_acting(body, authored, &shape, enforcing)?;
    let Some(rewritten) = evaluated.rewritten else {
        // Matched, measured, declined. Record it; forward the original.
        return Some((None, evaluated.decision));
    };
    // Serialization is the last thing that can fail. It fails to "forward the
    // original and record nothing" rather than to a half-written buffer — and
    // deliberately not to a recorded `applied`, which would claim a rewrite that
    // never reached the wire.
    match serde_json::to_vec(&rewritten) {
        Ok(encoded) => Some((Some(Bytes::from(encoded)), evaluated.decision)),
        Err(_) => None,
    }
}

/// Persist a churn finding's block to the bounded, host-local retention store so
/// `openlatch boundary explain <finding_id>` can show it — never emitted.
fn retention_store(f: &super::churn::ChurnFinding, occurred_at: &str) {
    super::retention::store(&super::retention::FindingRecord {
        finding_id: f.finding_id.clone(),
        captured_at: occurred_at.to_string(),
        churn_layer: f.churn_layer.as_str().to_string(),
        churn_class: f.churn_class.as_str().to_string(),
        divergence_offset: f.divergence_offset,
        churn_byte_len: f.churn_byte_len,
        churn_block_index: f.churn_block_index,
        block: f.block.clone(),
    });
}

/// Build a METADATA-ONLY observation for a `/v1/messages` request that took the
/// OPAQUE path (Content-Length absent/invalid, body over the 32 MB ceiling, or a
/// saturated materialize semaphore). The body is **never** read here (F-17 /
/// zero-buffer), so every body-derived fact — model, pricing inputs,
/// prefix-churn — stays null; only the header-derived facts (session +
/// assurance, billing mode, install id) and the freshly minted event id +
/// timestamp are set. The `unknown_wire_format` gap is stamped in `finalize`
/// from [`Measure::wire_format_unknown`]. The response-side usage tee still
/// completes the token facts from the response stream.
fn observe_request_metadata_only(
    st: &BoundaryState,
    wire: WireFormat,
    headers: &HeaderMap,
) -> Observation {
    let install_id = headers
        .get(INSTALL_ID_HEADER)
        .and_then(|v| v.to_str().ok())
        .map(str::to_string)
        .unwrap_or_default();
    Observation {
        measured: true,
        event_id: uuid::Uuid::now_v7().to_string(),
        occurred_at: crate::envelope::current_timestamp(),
        billing: detect_billing(headers),
        session: resolve_session(&st.registry, &install_id),
        install_id,
        // The route IS known on this path even though the body is not, and
        // `emit.rs` reads `gen_ai.provider.name` off it — leaving it at
        // `none()`'s `Unknown` would silently relabel every forced-opaque
        // `/v1/messages` event's provider (D-18's own path).
        wire_format: wire,
        // Header-derived like the rest of this path: the route is known, and
        // the wiring registry is in memory. The opaque path is exactly where an
        // unattributed request is most likely, so omitting it here would leave
        // the fix out of the case that needs it most.
        attributable_agent: st.wiring.sole_wired_agent_for(wire),
        // model / model_known / pricing / churn / request_body_len / has_breakpoint
        // are all body-derived and unavailable on the opaque path → left at the
        // `none()` defaults (null / false / 0).
        ..Observation::none()
    }
}

/// How much text of a single message the prompt hash will look at. Bounds the
/// cost on a conversation carrying a large pasted blob.
const TEXT_SCAN_BYTES: usize = 4096;

/// How many of the conversation's most recent user turns to hash, on top of the
/// opening one.
///
/// The opening turn alone was the original design, and it is the one turn a
/// registry entry born mid-conversation is guaranteed NOT to hold — the entry is
/// in-memory, so a daemon restart or a `SESSION_QUIET_WINDOW` eviction rebuilds
/// it from whatever turn comes next. A few recent turns are what both sides
/// reliably share. Cost stays bounded at `(1 + PROMPT_SCAN_TURNS)` messages of at
/// most [`TEXT_SCAN_BYTES`] each.
const PROMPT_SCAN_TURNS: usize = 4;

/// Derive the cascade's request-side signals from the already-parsed body.
///
/// Reads the user turns that carry signal — the **first** (the conversation's
/// opening prompt) and the last few (the turn being answered right now) — plus
/// the tool-call ids the request is carrying back. Everything in between is
/// history that both concurrent sessions could share.
///
/// Only ever called when the cascade is on — off, no request is inspected for this.
///
/// Takes the request's [`WireFormat`] because every one of those reads is
/// format-specific, and the format answers all three: naming your own session
/// inside the request body is an agent's convention rather than a property of
/// HTTP ([`wire_format::declared_session_id`]), and the two protocols shape their
/// turns and their tool results differently
/// ([`wire_format::conversation_view`]). What stays here is what is the same for
/// both — which turns are worth hashing, how they are hashed, and the bounds.
fn request_signals(fmt: WireFormat, body: &serde_json::Value) -> session::RequestSignals {
    // Selector 0 — the session the agent names for itself. Read first because it
    // is the only signal that is an answer rather than a clue.
    let mut out = session::RequestSignals {
        declared_session_id: wire_format::declared_session_id(fmt, body),
        ..Default::default()
    };

    let Some(view) = wire_format::conversation_view(fmt, body) else {
        return out;
    };

    // Selector 2 — the user turns, hashed the same way the hook hashed them.
    //
    // Two candidates per turn, because the hook stored the hash of the prompt the
    // USER typed and the agent may have appended blocks of its own to that turn
    // before it reached us (Claude Code appends system reminders). The first text
    // block alone is what usually equals the typed prompt; the full concatenation
    // is what equals it when the agent appended nothing. Matching on either keeps
    // the test exact — no fuzzy compare, and still no stored text. Codex appends
    // nothing, so its two candidates coincide and the duplicate is deduped below.
    //
    // The opening turn AND the last few, because the registry's side of this join
    // is in-memory: a daemon restart or a quiet gap rebuilds the entry
    // mid-conversation, and it then holds recent turns but never the opening one.
    // Offering only the first turn made selector 2 unmatchable for every session
    // that outlived its own registry entry — which is every long-running session.
    let tail_from = view.user_turns.len().saturating_sub(PROMPT_SCAN_TURNS);
    for (i, msg) in view.user_turns.iter().enumerate() {
        if i != 0 && i < tail_from {
            continue;
        }
        for text in [
            first_text_block(msg, view.text_block_type, TEXT_SCAN_BYTES),
            message_text(msg, view.text_block_type, TEXT_SCAN_BYTES),
        ] {
            if text.trim().is_empty() {
                continue;
            }
            let h = session::text_hash(&text);
            if !out.prompt_hashes.contains(&h) {
                out.prompt_hashes.push(h);
            }
        }
    }

    // Selector 1 — the ids this turn is answering, already read in the format's
    // own shape: a `tool_result` block's `tool_use_id` on the Messages API, a
    // top-level item's `call_id` on the Responses API. Both are the same value the
    // hook recorded, because Codex passes the model's `call_id` through to its own
    // hook payload's `tool_use_id`.
    out.tool_use_ids = view.tool_result_ids;

    out
}

/// The message's **first** text block only, bounded to `limit` bytes.
///
/// `block_type` is the format's name for a text block — the caller reads it off
/// [`wire_format::ConversationView`] rather than naming a protocol here.
///
/// For a bare-string content this is the whole string, so it coincides with
/// [`message_text`] and the duplicate candidate is dropped by the caller.
fn first_text_block(msg: &serde_json::Value, block_type: &str, limit: usize) -> String {
    let mut out = String::new();
    match msg.get("content") {
        Some(serde_json::Value::String(s)) => push_bounded(&mut out, s, limit),
        Some(serde_json::Value::Array(blocks)) => {
            for b in blocks {
                if b.get("type").and_then(|t| t.as_str()) == Some(block_type) {
                    if let Some(t) = b.get("text").and_then(|v| v.as_str()) {
                        push_bounded(&mut out, t, limit);
                    }
                    break;
                }
            }
        }
        _ => {}
    }
    out
}

/// Concatenate a message's text, bounded to `limit` bytes.
///
/// Handles both content shapes either format accepts: a bare string, or an array
/// of typed blocks (only blocks of `block_type` contribute).
fn message_text(msg: &serde_json::Value, block_type: &str, limit: usize) -> String {
    let mut out = String::new();
    match msg.get("content") {
        Some(serde_json::Value::String(s)) => push_bounded(&mut out, s, limit),
        Some(serde_json::Value::Array(blocks)) => {
            for b in blocks {
                if b.get("type").and_then(|t| t.as_str()) == Some(block_type) {
                    if let Some(t) = b.get("text").and_then(|v| v.as_str()) {
                        push_bounded(&mut out, t, limit);
                    }
                }
                if out.len() >= limit {
                    break;
                }
            }
        }
        _ => {}
    }
    out
}

/// Append until `limit`, splitting only on a char boundary so the result stays
/// valid UTF-8 (a byte-index truncate would panic mid-codepoint).
fn push_bounded(out: &mut String, s: &str, limit: usize) {
    if out.len() >= limit {
        return;
    }
    let room = limit - out.len();
    if s.len() <= room {
        out.push_str(s);
    } else {
        let mut end = room;
        while end > 0 && !s.is_char_boundary(end) {
            end -= 1;
        }
        out.push_str(&s[..end]);
    }
}

/// Whether this request is OpenLatch's own preflight probe
/// ([`super::preflight::probe`]) rather than a caller's traffic.
///
/// A probe travels the whole forward path on purpose — that is the point of it —
/// but it is a synthetic request WE minted, so it must never become an
/// economics event on the customer's bill. Both measurement entry points
/// consult this; the header itself is stripped in [`forward_headers`] and never
/// reaches the provider.
fn is_preflight(headers: &HeaderMap) -> bool {
    headers.contains_key(PREFLIGHT_HEADER)
}

/// Assemble the measurement context for a `/v1/messages` request forced onto the
/// opaque path (FIX 2). `None` when measurement is disabled (no cloud sink) —
/// exactly the plan-01 behaviour, identical to the materialized path's guard —
/// and `None` for a preflight probe, which reaches this branch only if it
/// arrives while the materialize semaphore is saturated.
fn opaque_measure_ctx(
    st: &Arc<BoundaryState>,
    wire: WireFormat,
    headers: &HeaderMap,
) -> Option<MeasureCtx> {
    if is_preflight(headers) {
        return None;
    }
    st.cloud_tx.as_ref().map(|_| MeasureCtx {
        obs: observe_request_metadata_only(st, wire, headers),
        tokenizer: st.tokenizer,
        cloud_tx: st.cloud_tx.clone(),
        privacy: st.privacy.clone(),
        // A fact about the PATH, not about the decoder: the body was never
        // parsed, so nothing could have decoded it. Deriving this from
        // `has_decoder()` would make it `false` for an opaque `/v1/messages`
        // and kill D-18's gap on the one path D-18 is about.
        wire_format_unknown: true,
    })
}

/// The single boundary proxy handler. `fallback` routes ALL paths here.
pub async fn proxy_any(State(st): State<Arc<BoundaryState>>, req: Request) -> Response {
    let (parts, body) = req.into_parts();
    // Resolved from method + route, BEFORE the body is touched — no new parse,
    // no TTFT cost, and never from agent identity (PRD D-7).
    let wire = WireFormat::resolve(&parts);

    // --- Opaque path: anything we don't capture. NO permit, NO materialize. ---
    // GET /v1/models, POST /v1/messages/count_tokens, batch endpoints, etc. An
    // uncaptured request is never an economics event, so it carries NO
    // measurement context.
    if !wire.is_captured() {
        return stream_through_opaque(&st, parts, body, None).await;
    }

    // Decide BEFORE consuming `body`, using Content-Length. Over-limit or
    // unknown-length → opaque path with the body still INTACT (`to_bytes` would
    // consume it and leave nothing to forward). A `/v1/messages` request STILL
    // emits one economics event from a metadata-only observation (headers only,
    // body never parsed) with `unknown_wire_format` (FIX 2).
    let len = content_length(&parts.headers);
    if len.is_none_or(|n| n > MAX_MATERIALIZE_BYTES) {
        let ctx = opaque_measure_ctx(&st, wire, &parts.headers);
        return stream_through_opaque(&st, parts, body, ctx).await;
    }

    // --- Materialized path: bounded by the semaphore (D-03). ---
    // Non-blocking: never `acquire().await`, which would queue under load and
    // wedge the loop. Saturated → forward opaque — still correct, and still
    // measured from headers only (`unknown_wire_format`, FIX 2).
    let permit = match st.inflight.clone().try_acquire_owned() {
        Ok(p) => p,
        Err(_) => {
            let ctx = opaque_measure_ctx(&st, wire, &parts.headers);
            return stream_through_opaque(&st, parts, body, ctx).await;
        }
    };

    // Length is known ≤ 32 MB here, so to_bytes cannot over-run.
    let bytes = match axum::body::to_bytes(body, MAX_MATERIALIZE_BYTES).await {
        Ok(b) => b,
        Err(_) => {
            // Truly exceptional: the client hung up mid-body. No intact body
            // remains to forward, so the honest answer is a synthetic 502.
            record_pass_through_failure("body_read");
            return synth_502();
        }
    };

    // OBSERVE on the bytes via a clone-safe read — never mutate `bytes` (D-06).
    // Synchronous + wrapped in catch_unwind (D-02) so a panic in measurement
    // can't corrupt the forward — it degrades to an unmeasured pass-through.
    let (mut observation, parsed) = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        observe_request(&st, wire, &parts.headers, &bytes)
    }))
    .unwrap_or_else(|_| {
        record_pass_through_failure("observe_panic");
        (Observation::none(), None)
    });

    // A preflight probe is observed like anything else — deliberately, so a
    // panic in the observe stage is caught and recorded on the very path the
    // gate is meant to vouch for — and then dropped before it can be promoted
    // to an economics event. Suppressing measurement here rather than skipping
    // `observe_request` is what keeps the probe's coverage honest: it exercises
    // what a real request exercises, it just does not bill for it.
    if is_preflight(&parts.headers) {
        observation.measured = false;
    }

    // MUTATE — the only step that can produce bytes the agent did not send
    // (D-28). `parsed` is `Some` only when `[boundary] transforms_act` is on, so
    // with the flag off this block is a single `None` match and the forward
    // below is the pre-D-28 forward, unchanged.
    //
    // Its own `catch_unwind`, separate from observation's, for two reasons: a
    // panic here must forward the ORIGINAL bytes (never a partial buffer), and
    // it must not discard the measurement that already succeeded — sharing a
    // guard would have thrown away a good `Observation` because a transform
    // panicked.
    //
    // GATED ON THE FORMAT, and this is the most dangerous line in the seam. The
    // bundle's request rules are Anthropic-authored and read an Anthropic body
    // shape; making `/v1/responses` captured is what first brings it here.
    // `[boundary] transforms_act = true` is reachable in the field today, so
    // default-off is NOT sufficient protection. The question is asked of the
    // FORMAT (`WireFormat::transforms_apply`) rather than tested inline, so
    // adding a third format never means editing this shared decision code.
    let bytes = if !wire.transforms_apply() {
        bytes // Observe-only for every other format. PRD D-26, and AGENTS.md's perimeter.
    } else {
        match parsed {
            None => bytes,
            Some(body) => {
                let mutated = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                    apply_transform(&st, &observation, &body)
                }))
                .unwrap_or_else(|_| {
                    record_pass_through_failure("mutate_panic");
                    None
                });
                match mutated {
                    // The L-0 decision REPLACES the observe-stage would-have,
                    // whether or not it fired: recording what a removal lever
                    // would have done while forwarding a rewritten body would
                    // report the wrong event, and recording nothing when a rule
                    // matched and declined would leave the silence unexplained.
                    Some((rewritten, decision)) => {
                        observation.transform = Some(decision);
                        rewritten.unwrap_or(bytes)
                    }
                    None => bytes,
                }
            }
        }
    };

    // FORWARD `bytes` — the original verbatim (byte-identical, so cache_control
    // survives, D-06) unless the step above returned a validated rewrite.
    forward_streaming(&st, parts, bytes, permit, observation).await
}

/// Opaque forward: stream the request body straight through with no
/// materialization and no permit. Used for every non-captured path and for
/// bodies too big / of unknown length (D-08).
///
/// The body is **never** buffered or altered — `into_data_stream()` streams the
/// inbound chunks straight to the upstream. A `measure_ctx` is present only for a
/// `/v1/messages` request that took this path (a metadata-only observation, FIX
/// 2); its response usage tee STILL runs in `relay`, so response-side tokens are
/// captured and exactly one event is emitted with `unknown_wire_format`. For
/// every other opaque path `measure_ctx` is `None` and nothing is emitted.
async fn stream_through_opaque(
    st: &Arc<BoundaryState>,
    parts: Parts,
    body: Body,
    measure_ctx: Option<MeasureCtx>,
) -> Response {
    let url = match upstream_url(st, &parts) {
        Some(u) => u,
        None => {
            // A `/v1/messages` opaque call that can't resolve an upstream still
            // must not vanish — emit its terminal (provider_error) event.
            emit_terminal_error(measure_ctx);
            return synth_502();
        }
    };
    // THE LOAD POINT — once per forwarded request. `None` means no egress route is
    // permitted (`[proxy] allow_direct = false` with nothing working): answer 502 rather
    // than fall through to a direct connection to the provider (D-21).
    let Some(client) = st.client.current() else {
        emit_terminal_error(measure_ctx);
        return synth_502();
    };
    // Zero-buffer request side: wrap the inbound body as a reqwest stream.
    let req_body = reqwest::Body::wrap_stream(body.into_data_stream());
    let send = client
        // `url` outlives the send so `relay` can record the egress outcome
        // against the host this request ACTUALLY went to.
        .request(parts.method.clone(), url.clone())
        .headers(forward_headers(&parts.headers))
        .body(req_body)
        .send();
    // Bound ONLY the header wait (D-05 hang guard). A connected-but-silent
    // upstream must not wedge the request forever. The response BODY stream that
    // follows headers is never timed out (long SSE turns are legitimate, D-06).
    let upstream = match tokio::time::timeout(st.header_timeout, send).await {
        Ok(res) => res,
        Err(_elapsed) => {
            emit_terminal_error(measure_ctx);
            return synth_502();
        }
    };
    relay(st, &url, upstream, None, measure_ctx)
}

/// Materialized forward: send the exact `bytes` verbatim, then stream the
/// response back with zero buffering. The `permit` is held until the response
/// body is fully drained (moved into the stream guard). The staged `observation`
/// rides through so the response-side usage tee can complete + emit the event.
async fn forward_streaming(
    st: &Arc<BoundaryState>,
    parts: Parts,
    bytes: Bytes,
    permit: OwnedSemaphorePermit,
    observation: Observation,
) -> Response {
    // Assemble the measurement context on the request thread while `st` is in
    // scope: clones of the cheap sinks (the tokenizer is a ZST, the privacy
    // filter and cloud sender are Clone). `None` when measurement is disabled or
    // the observation was an unmeasured no-op — then this is exactly plan 01.
    // Materialized path: the request body WAS parsed, so the question is no
    // longer "did we read it?" but "can this build decode it?". A captured route
    // with no decoder — `/v1/responses` until plan 02 — must report
    // `unknown_wire_format`, the honest gap, and not `stream_interrupted`, which
    // describes the wrong failure. `:619`'s hard `true` on the OPAQUE path is a
    // fact about the path and is deliberately untouched.
    let undecoded = !observation.wire_format.has_decoder();
    let measure_ctx = if observation.measured && st.cloud_tx.is_some() {
        Some(MeasureCtx {
            obs: observation,
            tokenizer: st.tokenizer,
            cloud_tx: st.cloud_tx.clone(),
            privacy: st.privacy.clone(),
            wire_format_unknown: undecoded,
        })
    } else {
        None
    };

    let url = match upstream_url(st, &parts) {
        Some(u) => u,
        None => {
            // No provider response will exist — still record the failed call so
            // it does not vanish (F-36 spirit): provider_error, zero tokens.
            emit_terminal_error(measure_ctx);
            return synth_502();
        }
    };
    // THE LOAD POINT — once per forwarded request; same contract as the opaque path above.
    let Some(client) = st.client.current() else {
        emit_terminal_error(measure_ctx);
        return synth_502();
    };
    let send = client
        // Same reason as the opaque path: `relay` records against this URL.
        .request(parts.method.clone(), url.clone())
        .headers(forward_headers(&parts.headers))
        .body(bytes) // reqwest sends these exact bytes — byte-identical
        .send();
    // Bound ONLY the header wait. On elapse we return a synth 502 and `permit`
    // drops naturally at end of scope, so a silent upstream can never pin a
    // semaphore permit for the life of the process. The response BODY stream is
    // never timed out (D-06).
    let upstream = match tokio::time::timeout(st.header_timeout, send).await {
        Ok(res) => res,
        Err(_elapsed) => {
            emit_terminal_error(measure_ctx);
            return synth_502();
        }
    };
    relay(st, &url, upstream, Some(permit), measure_ctx)
}

/// Shared response relay: hand the reqwest byte stream straight to axum so the
/// first byte flows before the last arrives (D-06). On a connect/timeout error
/// there is no upstream response, so synthesize a 502 (C-5b).
///
/// # Egress recording
///
/// This is the **only** place in the boundary that records an egress outcome, and
/// that is a deliberate narrowing. Five other paths in this file also answer
/// `synth_502`, and none of them is evidence about the route:
///
/// | Other 502 site | What actually happened |
/// | --- | --- |
/// | `to_bytes` failed | the *client* hung up mid-upload |
/// | `upstream_url` returned `None` | a configuration fault, before any socket |
/// | the header wait elapsed | an upstream that is alive and merely slow |
///
/// Counting any of those would push a healthy proxy past the two-failure
/// threshold — and, once plan 02's self-heal lands, re-resolve the route out
/// from under a proxy that was never broken. A `reqwest::Error` here, by
/// contrast, means no response existed at all: connect refused, DNS gone, TLS
/// rejected. That is the transport, and only that is recorded.
/// `base` is the URL the two forwarders **already resolved** and sent to, so the
/// egress outcome is filed under the host that was actually used. With a
/// per-format upstream map a fixed read would file every Codex `api.openai.com`
/// transport failure under Anthropic's host — or skip it entirely when a
/// `no_proxy` entry covers one host and not the other — and the monitor driving
/// re-resolution would then be counting the wrong route.
fn relay(
    st: &Arc<BoundaryState>,
    base: &reqwest::Url,
    upstream: Result<reqwest::Response, reqwest::Error>,
    permit: Option<OwnedSemaphorePermit>,
    measure_ctx: Option<MeasureCtx>,
) -> Response {
    let resp = match upstream {
        Ok(r) => {
            st.egress.record_ok(base.as_str());
            r
        }
        // No upstream response exists — the one place an OpenLatch-shaped signal
        // is unavoidable. Record the failed call, then synth 502 (C-5b).
        Err(e) => {
            st.egress.record_failure(base.as_str(), &e);
            emit_terminal_error(measure_ctx);
            return synth_502();
        }
    };

    let status = resp.status();
    let mut headers = resp.headers().clone();
    strip_hop_by_hop_headers(&mut headers);

    // Promote the request-side context into a live measure that scans the
    // response stream for the terminal usage chunk and emits on completion.
    let measure = measure_ctx.map(|ctx| ctx.into_measure(status.is_success()));

    // reqwest StatusCode / HeaderMap are the SAME `http` crate types axum uses
    // (unified http 1.x), so no conversion is required.
    let guarded = GuardedBody {
        inner: Box::pin(resp.bytes_stream()),
        _permit: permit,
        measure,
    };
    let mut out = Response::new(Body::from_stream(guarded));
    *out.status_mut() = status;
    *out.headers_mut() = headers;
    out
}

/// Emit a terminal `provider_error` event for a call that never yielded a
/// provider response (connect/timeout failure) — so a failed call does not vanish
/// from the record. No-op when there is no measurement context.
fn emit_terminal_error(measure_ctx: Option<MeasureCtx>) {
    if let Some(ctx) = measure_ctx {
        // A call that never yielded a provider response is a terminal error.
        let mut m = ctx.into_measure(false);
        m.finalize();
    }
}

/// Build the upstream URL from the resolved base + the inbound path-and-query.
///
/// Two questions, both answered off the request head before the body is
/// touched: which format's upstream this follows
/// ([`WireFormat::resolve_upstream`] — the route, plus a promotion for an
/// uncaptured route that is provably Codex's), and which origin that format's
/// credential is addressed to ([`AuthMode::resolve`]).
fn upstream_url(st: &Arc<BoundaryState>, parts: &Parts) -> Option<reqwest::Url> {
    let pq = parts
        .uri
        .path_and_query()
        .map(|x| x.as_str())
        .unwrap_or("/");
    let base = st.upstream_for_request(
        WireFormat::resolve_upstream(parts),
        AuthMode::resolve(&parts.headers),
    );
    join_upstream(base, pq)
}

/// Join the inbound path-and-query onto the upstream base, **preserving any
/// path the base carries**.
///
/// `Url::join` on a leading-slash reference is an absolute-path reference: it
/// keeps the base's scheme and host and replaces path + query outright. That is
/// right for a bare origin and silently wrong for a based one — the ChatGPT
/// backend lives at `/backend-api/codex`, and a plain join drops that and posts
/// to `chatgpt.com/v1/responses`, a route which does not exist.
///
/// A base carrying a path is therefore the API root **including its version
/// segment**, so the inbound's own `/v1` is redundant and is dropped:
///
/// | base | inbound | forwarded |
/// |---|---|---|
/// | `https://api.anthropic.com` | `/v1/messages` | `/v1/messages` |
/// | `https://chatgpt.com/backend-api/codex` | `/v1/responses` | `/backend-api/codex/responses` |
/// | `https://chatgpt.com/backend-api/codex` | `/v1/models?c=1` | `/backend-api/codex/models?c=1` |
/// | `https://gw.example/proxy/v1` | `/v1/messages` | `/proxy/v1/messages` |
///
/// An operator fronting the provider writes the version they serve, which is
/// the reading that makes the last row come out right.
///
/// **A base with no path takes the original join untouched** — that is every
/// built-in but [`wire_format::CHATGPT_BASE`] and every value in the field
/// today, so this cannot change an existing install's forwarding.
///
/// Only a whole `/v1` segment is stripped: `/v1beta/x` keeps its path, because
/// `strip_prefix` alone would leave `beta/x` and invent a route.
fn join_upstream(base: &reqwest::Url, pq: &str) -> Option<reqwest::Url> {
    let prefix = base.path().trim_end_matches('/');
    if prefix.is_empty() {
        return base.join(pq).ok();
    }
    let suffix = match pq.strip_prefix("/v1") {
        Some(rest) if rest.is_empty() || rest.starts_with('/') || rest.starts_with('?') => rest,
        _ => pq,
    };
    base.join(&format!("{prefix}{suffix}")).ok()
}

/// Parse the `Content-Length` header as a byte count, if present and valid.
fn content_length(headers: &HeaderMap) -> Option<usize> {
    headers.get(CONTENT_LENGTH)?.to_str().ok()?.parse().ok()
}

/// Clone the caller's headers for the forward, stripping the hop-by-hop set
/// (plus `Host`, which reqwest re-derives from the upstream URL). The provider
/// credential (`x-api-key` / `Authorization`) and all Anthropic headers pass
/// VERBATIM.
fn forward_headers(src: &HeaderMap) -> HeaderMap {
    let mut h = src.clone();
    h.remove(HOST); // reqwest sets Host from the upstream URL (request side only)
                    // Force an identity-encoded response. The usage scanner (`capture::scan_chunk`)
                    // reads the SSE `message_start` / `message_delta` bytes as raw ASCII, so a
                    // gzip/br-compressed body would never match and every turn would degrade to
                    // `stream_interrupted` / `tokenizer_estimated` (output + cache lost). Dropping
                    // the client's `Accept-Encoding` makes upstream stream plaintext we can read.
    h.remove(ACCEPT_ENCODING);
    // Our own preflight marker is an internal signal between `preflight::probe`
    // and this process. The provider has no use for it and should never see a
    // header it did not agree to receive.
    h.remove(PREFLIGHT_HEADER);
    strip_hop_by_hop_headers(&mut h);
    h
}

/// Strip hop-by-hop headers per RFC 7230 §6.1 from a header map, used on BOTH
/// the request-forward and response-relay paths (a proxy must not tunnel them).
///
/// Two parts:
/// 1. Every header named as a token in any `Connection` header value
///    (comma-split, trimmed, case-insensitive) — these are connection-specific
///    by the sender's own declaration and must not be forwarded.
/// 2. The fixed standard hop-by-hop set, plus `Content-Length` (reqwest/axum
///    re-frame the body themselves).
///
/// `Host` is deliberately NOT touched here — requests strip it separately (see
/// [`forward_headers`]); responses have no `Host` to strip.
fn strip_hop_by_hop_headers(h: &mut HeaderMap) {
    // (1) Remove every header named in a Connection token list.
    let connection_named: Vec<String> = h
        .get_all(CONNECTION)
        .iter()
        .filter_map(|v| v.to_str().ok())
        .flat_map(|v| v.split(','))
        .map(|t| t.trim().to_ascii_lowercase())
        .filter(|t| !t.is_empty())
        .collect();
    for name in connection_named {
        // `remove(&str)` is case-insensitive and a no-op on an unparsable name.
        h.remove(name.as_str());
    }

    // (2) Fixed hop-by-hop set. `keep-alive` has no `http` constant, so it is
    // removed by its lowercase name.
    h.remove(CONNECTION);
    h.remove("keep-alive");
    h.remove(TRANSFER_ENCODING);
    h.remove(TE);
    h.remove(TRAILER);
    h.remove(UPGRADE);
    h.remove(PROXY_AUTHENTICATE);
    h.remove(PROXY_AUTHORIZATION);
    h.remove(CONTENT_LENGTH);
}

/// Synthetic `502` for an unreachable upstream (C-5b). The single unavoidable
/// OpenLatch-shaped response — everything else is byte-transparent.
///
/// The single choke point for "the forward did not happen", which is why the
/// counter lives here rather than at the six call sites: a future seventh gets
/// counted for free, and the wiring supervisor's view of whether real traffic is
/// failing cannot silently go stale.
fn synth_502() -> Response {
    UPSTREAM_FAILURES.fetch_add(1, Ordering::Relaxed);
    let mut out = Response::new(Body::empty());
    *out.status_mut() = StatusCode::BAD_GATEWAY;
    out.headers_mut().insert(
        "x-openlatch-upstream",
        HeaderValue::from_static("unreachable"),
    );
    out
}

/// `GET /admin/boundary/status` — non-sensitive liveness for `openlatch status`
/// and the client-side admin surface (loopback only by construction). Reports
/// only port / uptime / capacity / failure count / wiring verdict — never a body
/// or a credential.
///
/// `wired` and `preflight` are what make "listening but not wired" a diagnosable
/// state rather than a mystery: the listener being up is no longer sufficient
/// for the agent to be pointed at it, so the reason it is not has to be
/// readable from outside the process. `init` blocks on `preflight` leaving
/// `pending`, and `doctor` prints `preflight_error` verbatim.
///
/// **All three are OBJECTS keyed by agent type**, because the wiring is. Two
/// agents share one listener and are probed in two different formats, so one
/// plane can be wired and green while the other's round trip fails; a scalar
/// here would report one of them for both. Every consumer reads
/// `field[agent]` — `classify_boundary`, `init`'s wait rule, and
/// `tests/boundary_wiring.rs`. An agent with no entry is one the supervisor has
/// not seen: `pending`, never a verdict borrowed from its neighbour.
pub async fn boundary_status(State(st): State<Arc<BoundaryState>>) -> Json<serde_json::Value> {
    let verdicts = st.wiring.verdicts();
    Json(serde_json::json!({
        "status": "up",
        "port": st.port,
        "upstream": WireFormat::ALL
            .iter()
            .map(|f| (f.as_str().to_string(), serde_json::json!(st.upstream_for(*f).as_str())))
            .collect::<serde_json::Map<_, _>>(),
        // The second destination for `openai-responses`, reached by a request
        // whose own credential is a ChatGPT subscription. A SIBLING rather than
        // an entry in the map above, so that map stays exactly one entry per
        // wire format — and `null`, not omitted, so a reader can tell "this
        // build does not pair" from "an operator configured that format".
        "upstream_chatgpt": st.chatgpt_upstream().map(reqwest::Url::as_str),
        "inflight_available": st.inflight.available_permits(),
        "uptime_secs": st.started_at.elapsed().as_secs(),
        "pass_through_failures": pass_through_failures(),
        "upstream_failures": upstream_failures(),
        // Which selector actually decided, live. The assurance on each emitted
        // row says how confident we were; this says *why*, which is the only
        // way to tell a working cascade from one whose joins stopped matching.
        "selector_wins": session::selector_wins()
            .into_iter()
            .map(|(k, v)| (k.to_string(), serde_json::json!(v)))
            .collect::<serde_json::Map<_, _>>(),
        "wired": st.wiring
            .wired_agents()
            .into_iter()
            .map(|(a, w)| (a.to_string(), serde_json::json!(w)))
            .collect::<serde_json::Map<_, _>>(),
        "preflight": verdicts
            .iter()
            .map(|(a, v)| (a.to_string(), serde_json::json!(v.label())))
            .collect::<serde_json::Map<_, _>>(),
        "preflight_error": verdicts
            .iter()
            .map(|(a, v)| (a.to_string(), serde_json::json!(v.error())))
            .collect::<serde_json::Map<_, _>>(),
    }))
}

/// Request-side facts + sinks handed from the forward thread into `relay`, where
/// the response status completes them into a live [`Measure`].
struct MeasureCtx {
    obs: Observation,
    tokenizer: Estimator,
    cloud_tx: Option<Sender<CloudEvent>>,
    privacy: PrivacyFilter,
    /// `true` when this rode in on the OPAQUE `/v1/messages` path (metadata-only
    /// observation, request body never parsed) — forces `unknown_wire_format`
    /// (FIX 2). `false` on the materialized path.
    wire_format_unknown: bool,
}

impl MeasureCtx {
    /// Promote the request-side context into a live [`Measure`] with a fresh
    /// accumulator. `status_ok` records whether the provider returned 2xx —
    /// `true` on a successful relay, `false` for a terminal `provider_error`.
    fn into_measure(self, status_ok: bool) -> Measure {
        Measure {
            obs: self.obs,
            acc: UsageAccumulator::default(),
            status_ok,
            emitted: false,
            wire_format_unknown: self.wire_format_unknown,
            tokenizer: self.tokenizer,
            cloud_tx: self.cloud_tx,
            privacy: self.privacy,
        }
    }
}

/// Live measurement carried by [`GuardedBody`]: accumulates usage across the
/// response stream and emits exactly one economics event when the stream ends.
struct Measure {
    obs: Observation,
    acc: UsageAccumulator,
    /// The provider returned a 2xx. `false` → `provider_error` (F-36), zero tokens.
    status_ok: bool,
    /// Idempotency guard so the event is emitted exactly once (poll-None vs Drop).
    emitted: bool,
    /// Set on the OPAQUE `/v1/messages` path — the request wire format was never
    /// captured, so `finalize` records `unknown_wire_format` (FIX 2).
    wire_format_unknown: bool,
    tokenizer: Estimator,
    cloud_tx: Option<Sender<CloudEvent>>,
    privacy: PrivacyFilter,
}

impl Measure {
    /// Assemble + emit the single economics event. Chooses the honest
    /// `cost_basis` + `capture_gap` from what capture actually saw:
    /// - non-2xx → `provider_error`, zero tokens, basis `provider_reported`;
    /// - clean terminal usage → `provider_reported`;
    /// - interrupted / unparsable → local estimate, `tokenizer_estimated`.
    fn finalize(&mut self) {
        if self.emitted {
            return;
        }
        self.emitted = true;

        let model = self.obs.model.clone().unwrap_or_default();
        let (usage, basis, gap) = if !self.status_ok {
            // A failed call consumed no billable tokens but must not vanish.
            (
                Usage::default(),
                CostBasis::ProviderReported,
                Some(CaptureGap::ProviderError),
            )
        } else if self.acc.is_terminal() {
            // The TERMINAL usage chunk (message_delta / response.completed /
            // non-streaming body) arrived cleanly — provider-reported. A
            // message_start alone is NOT terminal (its output_tokens=1 is
            // preliminary), so it falls to the estimate branch below rather than
            // being emitted as a real count (FIX 1).
            //
            // A 2xx whose usage arithmetic did not reconcile — the provider
            // reported `cached + cache_write > input_tokens`, so `input_tokens`
            // clamped to 0 — is `provider_error` too, but it KEEPS the measured
            // counts rather than zeroing them: only the input split is
            // untrustworthy, the output count is still the provider's own
            // number. Placed here, inside the terminal branch, it can fire only
            // on a DECODED terminal frame; the `wire_format_unknown` override
            // below keeps precedence over it.
            let gap = if self.acc.provider_arithmetic_bad() {
                Some(CaptureGap::ProviderError)
            } else {
                base_gap(&self.obs)
            };
            (self.acc.usage(), CostBasis::ProviderReported, gap)
        } else {
            // No terminal provider usage — local, network-free estimate (D-08).
            // NEVER provider_reported on an incomplete stream (C-4 honesty): a
            // stream that ended before message_delta is only partially measured,
            // and the preliminary message_start output is never emitted as final.
            let est = self.tokenizer.estimate(&model, self.obs.request_body_len);
            let usage = Usage {
                input_tokens: est.input_tokens,
                ..Usage::default()
            };
            // Same precedence as the inlined form: `unknown_model` when a model
            // string was present but off the known set (base_gap → Some), else
            // `stream_interrupted` (base_gap → None → the `.or` fallback).
            let gap = base_gap(&self.obs).or(Some(CaptureGap::StreamInterrupted));
            (usage, CostBasis::TokenizerEstimated, gap)
        };

        // Opaque-path override (FIX 2): a `/v1/messages` that bypassed request
        // capture (no/invalid Content-Length, over the 32 MB ceiling, or a
        // saturated materialize semaphore) never yielded a request-side wire
        // format — model/pricing/prefix are all null. That gap is recorded here.
        // A genuine `provider_error` (a failed call, `!status_ok`) keeps its more
        // specific gap so a failure never masquerades as a format gap.
        let gap = if self.wire_format_unknown && self.status_ok {
            Some(CaptureGap::UnknownWireFormat)
        } else {
            gap
        };

        let cache_preserved = capture::infer_cache_preserved(&usage);
        emit::build_and_emit(
            &self.obs,
            &usage,
            basis,
            gap,
            cache_preserved,
            &self.privacy,
            self.cloud_tx.as_ref(),
        );
    }
}

/// Base capture gap for an otherwise-clean call: `unknown_model` when a model
/// string was present but off the known (D-21) set; else none.
fn base_gap(obs: &Observation) -> Option<CaptureGap> {
    if obs.model.is_some() && !obs.model_known {
        Some(CaptureGap::UnknownModel)
    } else {
        None
    }
}

/// Wraps the upstream byte stream and holds the in-flight permit until the body
/// is fully drained, so the D-03 cap tracks the true concurrency of forwarded
/// requests (permit releases on `Drop`, i.e. when the response body ends or the
/// client hangs up). Unpin because all fields are Unpin.
///
/// When `measure` is `Some`, this is also the **usage tee** (D-10): every chunk
/// is scanned in passing for the terminal usage event and forwarded UNCHANGED —
/// it never buffers more than the current chunk. The scan runs OUTSIDE plan 01's
/// request-side `observe_request` guard, so it carries its **own** `catch_unwind`
/// (a panic there must not abort the response stream). On stream end (or drop /
/// error, i.e. a client hangup or interrupted stream) the event is emitted once.
struct GuardedBody {
    inner: Pin<Box<dyn Stream<Item = reqwest::Result<Bytes>> + Send>>,
    _permit: Option<OwnedSemaphorePermit>,
    measure: Option<Measure>,
}

impl Stream for GuardedBody {
    type Item = reqwest::Result<Bytes>;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let this = self.get_mut();
        let polled = this.inner.as_mut().poll_next(cx);
        match &polled {
            Poll::Ready(Some(Ok(chunk))) => {
                // The scan is genuinely guarded HERE (it runs outside the
                // request-side observe guard). A panic must not abort the
                // response stream — the chunk still flows unchanged below.
                // Capture the panic verdict into a local `bool` so the mutable
                // borrow of `this.measure` ends before we clear it (FIX 3).
                let scan_panicked = if let Some(m) = this.measure.as_mut() {
                    // The format the REQUEST resolved to (`WireFormat::resolve`,
                    // before the body was touched) is what selects the decoder —
                    // never what happened to come back. Copied out before the
                    // mutable borrow of `acc` so the two borrows stay disjoint.
                    let fmt = m.obs.wire_format;
                    let acc = &mut m.acc;
                    // FIX D: scan the borrowed `chunk` (&Bytes) directly — no clone.
                    // The ORIGINAL chunk is forwarded UNCHANGED below (never-buffer,
                    // byte-identical); scan_chunk reads it read-only as &[u8] (Bytes
                    // derefs to [u8]). `acc` (this.measure) and `chunk` (borrowed from
                    // the local `polled`) are disjoint, so this borrows cleanly.
                    std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                        // FIX B: the injection trigger is test-only — gated out of
                        // release so there is no per-chunk atomic load in production.
                        #[cfg(test)]
                        if INJECT_SCAN_PANIC.load(Ordering::Relaxed) {
                            panic!("injected usage-scan panic (FIX 3 test)");
                        }
                        acc.scan_chunk(fmt, chunk)
                    }))
                    .is_err()
                } else {
                    false
                };
                if scan_panicked {
                    // A caught panic may have left the accumulator partially
                    // mutated. Drop the measure so no corrupt/partial event is
                    // EVER emitted for this request — forwarding continues,
                    // the request is UNMEASURED (D-24). Cleared OUTSIDE the
                    // borrow above.
                    record_pass_through_failure("usage_scan_panic");
                    this.measure = None;
                }
            }
            // Stream ended, or errored mid-flight (interrupted) — emit once.
            Poll::Ready(None) | Poll::Ready(Some(Err(_))) => {
                if let Some(m) = this.measure.as_mut() {
                    m.finalize();
                }
            }
            Poll::Pending => {}
        }
        polled
    }
}

impl Drop for GuardedBody {
    fn drop(&mut self) {
        // Client hangup before the stream drained → still emit once (interrupted
        // path). The `emitted` guard makes this idempotent with poll-None.
        if let Some(m) = self.measure.as_mut() {
            m.finalize();
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::boundary::{mock, serve_ephemeral, BoundaryState};
    use futures_util::StreamExt;
    use std::time::{Duration, Instant};

    fn state_for(upstream_port: u16) -> Arc<BoundaryState> {
        let base = reqwest::Url::parse(&format!("http://127.0.0.1:{upstream_port}")).unwrap();
        Arc::new(BoundaryState::new(base, 0, 8, &[]))
    }

    // --- upstream joining ----------------------------------------------------

    fn joined(base: &str, pq: &str) -> String {
        let url = reqwest::Url::parse(base).expect("base parses");
        join_upstream(&url, pq).expect("join succeeds").to_string()
    }

    /// A base with no path must forward byte for byte as it always has — this
    /// is every value in the field today, so a regression here is a regression
    /// for every existing install.
    #[test]
    fn a_bare_origin_joins_exactly_as_before() {
        assert_eq!(
            joined("https://api.anthropic.com", "/v1/messages"),
            "https://api.anthropic.com/v1/messages"
        );
        assert_eq!(
            joined("https://api.openai.com", "/v1/responses"),
            "https://api.openai.com/v1/responses"
        );
        assert_eq!(
            joined("https://api.anthropic.com", "/v1/models?beta=true"),
            "https://api.anthropic.com/v1/models?beta=true"
        );
        // A trailing slash is the same origin, not a path.
        assert_eq!(
            joined("https://api.anthropic.com/", "/v1/messages"),
            "https://api.anthropic.com/v1/messages"
        );
    }

    /// Format resolution, auth-mode selection and the join, COMPOSED — the URL
    /// a real Codex install's requests actually leave for.
    ///
    /// The three unit tests above each prove one half; this is the one that
    /// would have caught the shipped behaviour, where every row below left for
    /// `api.anthropic.com` or for a route that does not exist.
    #[test]
    fn a_codex_install_forwards_each_route_to_the_right_origin() {
        let st = Arc::new(
            BoundaryState::new(crate::boundary::default_upstream(), 0, 8, &[])
                .with_upstream_map(std::collections::BTreeMap::new()),
        );
        use axum::http::Method;
        let sent = |method: Method, uri: &str, headers: &[(&str, &str)]| {
            let mut req = axum::http::Request::builder().method(method).uri(uri);
            for (k, v) in headers {
                req = req.header(*k, *v);
            }
            let (parts, ()) = req.body(()).expect("build request").into_parts();
            upstream_url(&st, &parts)
                .expect("an upstream resolves")
                .to_string()
        };

        const ACCOUNT: (&str, &str) = ("chatgpt-account-id", "ef1a0c98");
        const ORIGINATOR: (&str, &str) = ("originator", "codex_exec");
        const PLATFORM_KEY: (&str, &str) = ("authorization", "Bearer sk-proj-abc");

        // A ChatGPT-plan turn — the case that 403'd against api.openai.com,
        // because that token carries `api.connectors.*` scopes and nothing else.
        assert_eq!(
            sent(Method::POST, "/v1/responses", &[ACCOUNT, ORIGINATOR]),
            "https://chatgpt.com/backend-api/codex/responses"
        );
        // The same route on a platform key keeps OpenAI's first-party base.
        assert_eq!(
            sent(Method::POST, "/v1/responses", &[PLATFORM_KEY, ORIGINATOR]),
            "https://api.openai.com/v1/responses"
        );
        // Uncaptured, and Codex issues it every session. It used to leave for
        // Anthropic bearing an OpenAI credential.
        assert_eq!(
            sent(
                Method::GET,
                "/v1/models?client_version=0.150.1",
                &[ACCOUNT, ORIGINATOR]
            ),
            "https://chatgpt.com/backend-api/codex/models?client_version=0.150.1"
        );
        assert_eq!(
            sent(
                Method::GET,
                "/v1/models?client_version=0.150.1",
                &[PLATFORM_KEY, ORIGINATOR]
            ),
            "https://api.openai.com/v1/models?client_version=0.150.1"
        );
        // Claude Code is untouched, on both its captured and uncaptured routes
        // — including under the bearer credential its subscription mode sends,
        // which is why the promotion reads a Codex marker and not a token shape.
        assert_eq!(
            sent(Method::POST, "/v1/messages", &[("x-api-key", "sk-ant-abc")]),
            "https://api.anthropic.com/v1/messages"
        );
        assert_eq!(
            sent(
                Method::GET,
                "/v1/models",
                &[("authorization", "Bearer sk-ant-oat01-abc")]
            ),
            "https://api.anthropic.com/v1/models"
        );
    }

    /// The ChatGPT backend lives under a prefix. `Url::join` alone drops it and
    /// posts to a route that does not exist — the defect this function fixes.
    #[test]
    fn a_based_upstream_keeps_its_prefix() {
        assert_eq!(
            joined(wire_format::CHATGPT_BASE, "/v1/responses"),
            "https://chatgpt.com/backend-api/codex/responses"
        );
        assert_eq!(
            joined(wire_format::CHATGPT_BASE, "/v1/models?client_version=0.1"),
            "https://chatgpt.com/backend-api/codex/models?client_version=0.1"
        );
        // What the old `join` did, kept here so the regression is legible.
        let base = reqwest::Url::parse(wire_format::CHATGPT_BASE).unwrap();
        assert_eq!(
            base.join("/v1/responses").unwrap().to_string(),
            "https://chatgpt.com/v1/responses",
            "plain join drops the prefix — that is the bug"
        );
    }

    /// An operator fronting the provider writes the version they serve, so the
    /// inbound's own `/v1` is dropped rather than doubled.
    #[test]
    fn a_gateway_prefix_absorbs_the_version_segment() {
        assert_eq!(
            joined("https://gw.example/proxy/v1", "/v1/messages"),
            "https://gw.example/proxy/v1/messages"
        );
        assert_eq!(
            joined("https://gw.example/proxy/v1/", "/v1/messages"),
            "https://gw.example/proxy/v1/messages"
        );
    }

    /// Only a WHOLE `/v1` segment is a version prefix. A bare `strip_prefix`
    /// would turn `/v1beta/x` into `beta/x` and invent a route.
    #[test]
    fn only_a_whole_version_segment_is_stripped() {
        assert_eq!(
            joined(wire_format::CHATGPT_BASE, "/v1beta/models"),
            "https://chatgpt.com/backend-api/codex/v1beta/models"
        );
        assert_eq!(
            joined(wire_format::CHATGPT_BASE, "/v2/responses"),
            "https://chatgpt.com/backend-api/codex/v2/responses"
        );
        // A path that is exactly the version segment collapses to the root.
        assert_eq!(
            joined(wire_format::CHATGPT_BASE, "/v1"),
            "https://chatgpt.com/backend-api/codex"
        );
    }

    // --- egress recording ----------------------------------------------------

    /// A boundary whose upstream is a NAMED authority, wired to report.
    ///
    /// The name is load-bearing. A loopback destination is hard-bypassed, never
    /// travels the egress route, and is therefore deliberately never recorded —
    /// which is also why the *other* `synth_502` paths cannot be covered end to
    /// end in-process: every one of them needs a reachable upstream, and every
    /// reachable upstream in a hermetic test is loopback. `relay` is called
    /// directly here, and the narrowing itself is asserted on the source below.
    fn reporting_state() -> (Arc<BoundaryState>, crate::egress::EgressState) {
        let cfg = crate::egress::EgressConfig::direct();
        let health = crate::egress::EgressState::new(&cfg);
        let state = BoundaryState::new(
            reqwest::Url::parse("https://api.anthropic.test/").expect("url"),
            0,
            8,
            &[],
        )
        .with_egress_reporter(crate::egress::EgressReporter::recording(
            &cfg,
            health.clone(),
        ));
        (Arc::new(state), health)
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_forwarded_response_records_egress_ok() {
        let (st, health) = reporting_state();
        health.record_failure(crate::error::ERR_PROXY_UNREACHABLE, "an earlier outage");
        health.record_failure(crate::error::ERR_PROXY_UNREACHABLE, "an earlier outage");

        let upstream = mock::spawn_capture_200().await;
        let response = crate::egress::client()
            .get(format!("http://127.0.0.1:{}/", upstream.port))
            .send()
            .await
            .expect("the mock answers 200");

        let base = st.upstream_for(WireFormat::AnthropicMessages).clone();
        let _ = relay(&st, &base, Ok(response), None, None);
        assert_eq!(
            health.consecutive_failures(),
            0,
            "a forwarded response is proof the route works"
        );
        assert!(health.last_ok_at().is_some());
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn the_connect_level_error_is_the_one_that_records_a_failure() {
        let (st, health) = reporting_state();
        let dead = mock::closed_port().await;
        let err = crate::egress::client()
            .get(format!("http://127.0.0.1:{dead}/"))
            .send()
            .await
            .expect_err("a closed port cannot answer");

        let base = st.upstream_for(WireFormat::AnthropicMessages).clone();
        let _ = relay(&st, &base, Err(err), None, None);
        assert_eq!(health.consecutive_failures(), 1);
        let recorded = health.last_error().expect("a code and a message");
        assert_eq!(
            recorded.code,
            crate::error::ERR_EGRESS_UNREACHABLE,
            "a direct route names the platform side, not a proxy"
        );
        assert!(
            !recorded.message.is_empty(),
            "a code without a message is not a diagnosis"
        );
    }

    /// A measured boundary in front of a usage-bearing SSE mock, plus the
    /// channel its economics events land on.
    fn measured_state(
        upstream_port: u16,
    ) -> (Arc<BoundaryState>, tokio::sync::mpsc::Receiver<CloudEvent>) {
        let (tx, rx) = tokio::sync::mpsc::channel::<CloudEvent>(8);
        let base = reqwest::Url::parse(&format!("http://127.0.0.1:{upstream_port}")).unwrap();
        let registry = Arc::new(session::SessionRegistry::default());
        registry.upsert("agt_1", "agt_1", "claude-code", "sess_a");
        let st = Arc::new(BoundaryState::new(base, 0, 8, &[]).with_measurement(registry, Some(tx)));
        (st, rx)
    }

    /// An UNCAPTURED route is the normal answer, not an error path: it forwards
    /// opaque, byte-identically, and emits **nothing**.
    ///
    /// Not a D-18 test. D-18 is about a `/v1/messages` forced opaque, which
    /// still emits one event — `tests/boundary_measurement.rs` guards that.
    /// Threading a measurement context onto this branch instead would open an
    /// economics row for every `GET /v1/models` and every batch call, a stream
    /// nothing in this unit sanctions.
    #[tokio::test(flavor = "multi_thread")]
    async fn unknown_route_forwards_byte_identically_and_emits_nothing() {
        let up = mock::spawn_capture_200().await;
        let (st, mut rx) = measured_state(up.port);
        let port = serve_ephemeral(st).await;

        let body = br#"{"anything":"at all"}"#.to_vec();
        let resp = crate::egress::client()
            .post(format!("http://127.0.0.1:{port}/v1/unknown"))
            .header("content-type", "application/json")
            .header(INSTALL_ID_HEADER, "agt_1")
            .body(body.clone())
            .send()
            .await
            .expect("the forward answers");
        assert!(resp.status().is_success());
        let _ = resp.bytes().await.unwrap();

        assert_eq!(
            up.received_body.lock().unwrap().clone(),
            Some(body),
            "an uncaptured route forwards the bytes verbatim"
        );
        assert!(
            tokio::time::timeout(Duration::from_millis(400), rx.recv())
                .await
                .ok()
                .flatten()
                .is_none(),
            "an uncaptured route must emit NO economics event"
        );
    }

    /// The D-14 flip, and the only unit-level gate on it.
    ///
    /// This REPLACES plan 01's captured-route-without-a-decoder test, which the
    /// flip makes impossible: `WireFormat` has three variants, `Unknown` is
    /// never captured, and after this commit both captured formats have a
    /// decoder — so no captured route can report `unknown_wire_format` any more,
    /// and that test could not be re-pointed either. It was deleted here, in the
    /// same commit as the flip.
    ///
    /// **Asserts `!= unknown_wire_format`, NOT absence.** A correct Codex turn
    /// always carries `unknown_model`: `classify_model` selects a *tokenizer*
    /// for local estimation and is Anthropic-only, while a Codex measurement is
    /// `provider_reported` and needs no tokenizer. An absence assertion would
    /// red on correct code — and dropping `model` from the fixture to make it
    /// pass would leave a green test that proves nothing.
    #[tokio::test(flavor = "multi_thread")]
    async fn responses_decoded_turn_is_not_unknown_wire_format() {
        // A REAL `response.completed` frame, sent in three chunks with a
        // newline-free middle one — so this exercises D-15's carry-over through
        // the real tee, not just the unit scanner.
        let up = mock::spawn_capture_responses_sse(14_342, 2_688, 0, 916).await;
        let (st, mut rx) = measured_state(up.port);
        let port = serve_ephemeral(st).await;

        let body = br#"{"model":"gpt-5-codex","input":"hello"}"#.to_vec();
        let resp = crate::egress::client()
            .post(format!("http://127.0.0.1:{port}/v1/responses"))
            .header("content-type", "application/json")
            .header(INSTALL_ID_HEADER, "agt_1")
            .body(body)
            .send()
            .await
            .expect("the forward answers");
        assert!(resp.status().is_success());
        let _ = resp.bytes().await.unwrap();

        let env = tokio::time::timeout(Duration::from_secs(3), rx.recv())
            .await
            .ok()
            .flatten()
            .expect("a captured route emits one economics event")
            .envelope;
        let data = &env["data"];

        assert_ne!(
            data["ai.openlatch.capture.gap"], "unknown_wire_format",
            "the decoder ran — labelling a real measurement unmeasured is the D-14 defect"
        );
        // Was `unknown_model` until the Codex family got a measured byte ratio
        // in `tokenize.rs`. `base_gap` reads tokenizer-table membership and not
        // whether the estimator ran, so an exact provider-reported Codex turn
        // was labelled gapped on a path that never estimated anything. Nothing
        // is missing from this measurement, and it now says so.
        assert!(
            data["ai.openlatch.capture.gap"].is_null(),
            "a fully provider-reported turn on a known model has no gap to report, got {}",
            data["ai.openlatch.capture.gap"]
        );
        assert_eq!(data["ai.openlatch.cost.basis"], "provider_reported");
        assert_eq!(data["gen_ai.provider.name"], "openai");
        assert_eq!(data["gen_ai.usage.output_tokens"], 916);
        assert_eq!(data["gen_ai.usage.cache_read.input_tokens"], 2_688);
        assert_eq!(
            data["gen_ai.usage.input_tokens"],
            14_342 - 2_688,
            "fresh input: the three-term subtraction, through the real proxy"
        );
        assert_eq!(data["ai.openlatch.cache.ephemeral_5m_input_tokens"], 0);
        assert_eq!(data["ai.openlatch.cache.ephemeral_1h_input_tokens"], 0);
    }

    /// The clamp's gap, asserted where it is computed.
    ///
    /// `cached + cache_write > input_tokens` cannot be observed from
    /// `capture.rs`: the accumulator only carries the flag, and `finalize` —
    /// one module away — is what turns it into `capture_gap = provider_error`.
    /// The measured OUTPUT count must survive: only the input split is
    /// untrustworthy.
    #[tokio::test(flavor = "multi_thread")]
    async fn responses_clamp_sets_provider_error() {
        // 8 + 5 > 10.
        let up = mock::spawn_capture_responses_sse(10, 8, 5, 4).await;
        let (st, mut rx) = measured_state(up.port);
        let port = serve_ephemeral(st).await;

        let body = br#"{"model":"gpt-5-codex","input":"hello"}"#.to_vec();
        let resp = crate::egress::client()
            .post(format!("http://127.0.0.1:{port}/v1/responses"))
            .header("content-type", "application/json")
            .header(INSTALL_ID_HEADER, "agt_1")
            .body(body)
            .send()
            .await
            .expect("the forward answers");
        assert!(resp.status().is_success());
        let _ = resp.bytes().await.unwrap();

        let env = tokio::time::timeout(Duration::from_secs(3), rx.recv())
            .await
            .ok()
            .flatten()
            .expect("a captured route emits one economics event")
            .envelope;
        let data = &env["data"];

        assert_eq!(
            data["ai.openlatch.capture.gap"], "provider_error",
            "a 2xx whose usage arithmetic does not reconcile is a provider error"
        );
        assert_eq!(
            data["gen_ai.usage.output_tokens"], 4,
            "the output count is KEPT — it is still the provider's own number"
        );
        assert_eq!(
            data["gen_ai.usage.input_tokens"], 0,
            "clamped to zero, never wrapped"
        );
        assert_eq!(
            data["ai.openlatch.cost.basis"], "provider_reported",
            "the terminal frame did arrive; it is the input SPLIT that is wrong"
        );
    }

    /// The narrowing, asserted where it can be: on the source.
    ///
    /// Five paths in this file answer `synth_502` and only ONE is evidence about
    /// the route. A client that hung up mid-upload, an unresolvable upstream URL,
    /// and a header deadline blown by a slow-but-alive upstream are all
    /// *someone else's* problem — counting any of them would push a healthy
    /// proxy past the two-failure threshold and, once self-heal lands, re-resolve
    /// the route out from under it.
    ///
    /// Two recording calls, both inside `relay`: one `record_ok`, one
    /// `record_failure`.
    #[test]
    fn only_the_relay_records_an_egress_outcome() {
        // Production code only — this test's own mention of the needle lives
        // below the `cfg(test)` line and must not count itself.
        let production = include_str!("proxy.rs")
            .split("mod tests {")
            .next()
            .expect("the file has a production half");
        let sites = production.matches("st.egress.record_").count();
        assert_eq!(
            sites, 2,
            "egress outcomes must be recorded ONLY at relay's two branches —              a third call site means another synth_502 path started reporting"
        );
    }

    /// A Messages-shaped body: opening user turn, the assistant's tool call, and
    /// the user turn carrying the `tool_result` that answers it.
    fn conversation() -> serde_json::Value {
        serde_json::json!({
            "messages": [
                {"role": "user", "content": [{"type": "text", "text": "build the parser"}]},
                {"role": "assistant", "content": [
                    {"type": "tool_use", "id": "toolu_aaa", "name": "Bash", "input": {}}
                ]},
                {"role": "user", "content": [
                    {"type": "tool_result", "tool_use_id": "toolu_aaa", "content": "ok"},
                    {"type": "text", "text": "now check /repo/alpha/src/main.rs please"}
                ]}
            ]
        })
    }

    /// The exact `metadata.user_id` shape Claude Code 2.1.241 sends, captured
    /// off the wire on 2026-08-24.
    fn claude_code_metadata(session_id: &str) -> serde_json::Value {
        serde_json::json!({
            "metadata": {
                "user_id": format!(
                    r#"{{"device_id":"600058d8","account_uuid":"212c36cb","session_id":"{session_id}"}}"#
                )
            }
        })
    }

    #[test]
    fn the_request_names_its_own_session() {
        let body = claude_code_metadata("5c7d9833-5e3b-4f88-9ece-cf371fb48c6a");
        assert_eq!(
            request_signals(WireFormat::AnthropicMessages, &body)
                .declared_session_id
                .as_deref(),
            Some("5c7d9833-5e3b-4f88-9ece-cf371fb48c6a")
        );
    }

    #[test]
    fn anything_that_is_not_that_shape_is_silently_ignored() {
        // The field is another program's internal convention, not a contract.
        // Every one of these must degrade to "no declared id" so the heuristics
        // below carry on — never to an error, and never to a bogus label.
        let cases = [
            serde_json::json!({}),                                    // no metadata
            serde_json::json!({"metadata": {}}),                      // no user_id
            serde_json::json!({"metadata": {"user_id": "plain-id"}}), // not JSON
            serde_json::json!({"metadata": {"user_id": 42}}),         // not a string
            serde_json::json!({"metadata": {"user_id": "{\"account_uuid\":\"x\"}"}}), // no session_id
            serde_json::json!({"metadata": {"user_id": "{\"session_id\":\"\"}"}}),    // blank
            serde_json::json!({"metadata": {"user_id": "{\"session_id\":\"a\\u0000b\"}"}}), // control char
        ];
        for body in cases {
            assert_eq!(
                request_signals(WireFormat::AnthropicMessages, &body).declared_session_id,
                None,
                "must not trust {body}"
            );
        }
    }

    #[test]
    fn an_absurdly_long_declared_session_id_is_refused() {
        // It becomes a label in the economics record and a registry key; an
        // unbounded caller-controlled string must not become either.
        let body = serde_json::json!({
            "metadata": {"user_id": format!(r#"{{"session_id":"{}"}}"#, "x".repeat(500))}
        });
        assert_eq!(
            request_signals(WireFormat::AnthropicMessages, &body).declared_session_id,
            None
        );
    }

    #[test]
    fn declared_session_id_is_unchanged_for_anthropic() {
        // The function moved to `wire_format.rs`; its answers did not. Driven
        // against the same fixtures the pre-move tests use — the exact Claude
        // Code shape, and the table of everything that is not it.
        let body = claude_code_metadata("5c7d9833-5e3b-4f88-9ece-cf371fb48c6a");
        assert_eq!(
            wire_format::declared_session_id(WireFormat::AnthropicMessages, &body).as_deref(),
            Some("5c7d9833-5e3b-4f88-9ece-cf371fb48c6a")
        );

        let refused = [
            serde_json::json!({}),                                    // no metadata
            serde_json::json!({"metadata": {}}),                      // no user_id
            serde_json::json!({"metadata": {"user_id": "plain-id"}}), // not JSON
            serde_json::json!({"metadata": {"user_id": 42}}),         // not a string
            serde_json::json!({"metadata": {"user_id": "{\"account_uuid\":\"x\"}"}}), // no session_id
            serde_json::json!({"metadata": {"user_id": "{\"session_id\":\"\"}"}}),    // blank
            serde_json::json!({"metadata": {"user_id": "{\"session_id\":\"a\\u0000b\"}"}}), // control char
            serde_json::json!({
                "metadata": {"user_id": format!(r#"{{"session_id":"{}"}}"#, "x".repeat(500))}
            }), // over the 128-char cap
        ];
        for body in refused {
            assert_eq!(
                wire_format::declared_session_id(WireFormat::AnthropicMessages, &body),
                None,
                "must not trust {body}"
            );
        }
    }

    #[test]
    fn observe_request_believes_the_session_the_request_names() {
        // End to end through the function the request path calls: two stale
        // sessions in the registry, and a request naming a third that the
        // registry has never seen. This is the field failure of 2026-08-24 —
        // before selector 0 it landed on whichever stale session moved last.
        let base = reqwest::Url::parse("http://127.0.0.1:1").unwrap();
        let st = BoundaryState::new(base, 0, 8, &[]);
        for zombie in ["6693e5f6", "a5a39553"] {
            st.registry
                .upsert("agt_test", "agt_test", "claude-code", zombie);
        }

        let mut body = conversation();
        body["metadata"] = claude_code_metadata("5c7d9833")["metadata"].clone();

        let mut headers = HeaderMap::new();
        headers.insert(INSTALL_ID_HEADER, "agt_test".parse().unwrap());
        let bytes = Bytes::from(serde_json::to_vec(&body).unwrap());

        let obs = observe_request(&st, WireFormat::AnthropicMessages, &headers, &bytes).0;
        assert_eq!(obs.session.session_id.as_deref(), Some("5c7d9833"));
        assert_eq!(
            obs.session.assurance,
            crate::boundary::session::Assurance::Attested
        );
    }

    #[test]
    fn request_signals_reads_the_first_and_last_user_turns() {
        let s = request_signals(WireFormat::AnthropicMessages, &conversation());
        // Selector 1 — the id this turn answers.
        assert_eq!(s.tool_use_ids, vec!["toolu_aaa".to_string()]);
        // Selector 2 — hashed identically to the hook side, so the join works.
        assert!(s
            .prompt_hashes
            .contains(&crate::boundary::session::text_hash("build the parser")));
    }

    #[test]
    fn request_signals_offers_recent_turns_not_only_the_opening_one() {
        // The regression guard for the mid-conversation registry entry: a daemon
        // that started after the conversation opened only ever recorded the LATER
        // turns, so a request offering `messages[0]` alone could not join.
        let body = serde_json::json!({"messages": [
            {"role": "user", "content": "hello this a session without a cache"},
            {"role": "assistant", "content": "hi"},
            {"role": "user", "content": "what is session Id"}
        ]});
        let s = request_signals(WireFormat::AnthropicMessages, &body);
        for p in ["hello this a session without a cache", "what is session Id"] {
            assert!(
                s.prompt_hashes
                    .contains(&crate::boundary::session::text_hash(p)),
                "missing candidate for {p:?} — an entry born mid-conversation \
                 holds only the later turns"
            );
        }
    }

    #[test]
    fn request_signals_handles_a_bare_string_content() {
        // The Messages API accepts a plain string as content; the first-prompt
        // selector must not silently only work for the block form.
        let body = serde_json::json!({
            "messages": [{"role": "user", "content": "build the parser"}]
        });
        let s = request_signals(WireFormat::AnthropicMessages, &body);
        assert!(s
            .prompt_hashes
            .contains(&crate::boundary::session::text_hash("build the parser")));
        // A bare string yields one candidate, not a duplicate pair.
        assert_eq!(s.prompt_hashes.len(), 1);
    }

    #[test]
    fn first_prompt_survives_a_block_the_agent_appended() {
        // The case selector 2 exists to survive: the hook stored the hash of the
        // typed prompt, but the turn that reached the provider carries an extra
        // block the agent added. The typed prompt must still be a candidate.
        let body = serde_json::json!({
            "messages": [{"role": "user", "content": [
                {"type": "text", "text": "build the parser"},
                {"type": "text", "text": "<system-reminder>appended by the agent</system-reminder>"}
            ]}]
        });
        let s = request_signals(WireFormat::AnthropicMessages, &body);
        assert!(
            s.prompt_hashes
                .contains(&crate::boundary::session::text_hash("build the parser")),
            "the typed prompt must survive as a candidate, else selector 2 never fires"
        );
    }

    #[test]
    fn request_signals_is_empty_for_a_body_with_no_messages() {
        // Anything that is not a conversation yields no signals, and an empty
        // signal set is what makes the cascade fall through rather than misfire.
        for body in [serde_json::json!({}), serde_json::json!({"messages": []})] {
            let s = request_signals(WireFormat::AnthropicMessages, &body);
            assert!(s.tool_use_ids.is_empty());
            assert!(s.prompt_hashes.is_empty());
        }
    }

    /// A Codex turn on the Responses wire, in the shape codex-cli 0.150.1 sends:
    /// the `developer` preamble, the `<environment_context>` Codex prepends as a
    /// user turn of its own, the typed prompt verbatim, and the call/output pair
    /// for the tool it just ran.
    fn codex_turn() -> serde_json::Value {
        serde_json::json!({
            "model": "gpt-5-codex",
            "input": [
                {"type": "message", "id": "msg_1", "role": "developer",
                 "content": [{"type": "input_text", "text": "<skills_instructions>…"}]},
                {"type": "message", "id": "msg_2", "role": "user",
                 "content": [{"type": "input_text", "text": "<environment_context>\n  <cwd>/repo</cwd>\n</environment_context>"}]},
                {"type": "message", "id": "msg_3", "role": "user",
                 "content": [{"type": "input_text", "text": "build the parser"}]},
                {"type": "function_call", "id": "fc_1", "call_id": "call_aaa",
                 "name": "shell", "arguments": "{}"},
                {"type": "function_call_output", "id": "fco_1", "call_id": "call_aaa",
                 "output": "ok"}
            ]
        })
    }

    #[test]
    fn request_signals_reads_a_codex_responses_turn() {
        // Before this, `request_signals` looked for `messages` and a Responses
        // body returned empty: selectors 1 and 2 were unreachable for Codex and
        // the cascade collapsed to "only one live session" or a guess.
        let s = request_signals(WireFormat::OpenAiResponses, &codex_turn());

        // Selector 1. Codex passes the model's `call_id` straight through to its
        // own hook payload's `tool_use_id` (`run_pre_tool_use_hooks(…,
        // invocation.call_id, …)` in codex-rs), so this is the same string the
        // registry holds — an exact join, not a resemblance.
        assert_eq!(s.tool_use_ids, vec!["call_aaa".to_string()]);

        // Selector 2. The typed prompt rides verbatim in its own `input_text`
        // block, which is byte-for-byte what `UserPromptSubmit` reports as
        // `prompt`.
        assert!(
            s.prompt_hashes
                .contains(&crate::boundary::session::text_hash("build the parser")),
            "the typed prompt must survive as a candidate, else selector 2 never fires"
        );

        // And nothing from the `developer` preamble, which is identical across
        // every Codex session and would match sessions this request never touched.
        assert!(
            !s.prompt_hashes
                .contains(&crate::boundary::session::text_hash(
                    "<skills_instructions>…"
                )),
            "the agent's own preamble is not a signal about WHICH session this is"
        );
    }

    #[test]
    fn a_codex_body_read_as_the_wrong_format_yields_nothing() {
        // The route picks the format. Read as Messages, a Responses body has no
        // `messages` to find — and the honest answer is no signals, so the
        // cascade falls through rather than misfiring on a mis-resolved route.
        let s = request_signals(WireFormat::AnthropicMessages, &codex_turn());
        assert!(s.tool_use_ids.is_empty());
        assert!(s.prompt_hashes.is_empty());

        let s = request_signals(WireFormat::OpenAiResponses, &conversation());
        assert!(s.tool_use_ids.is_empty());
        assert!(s.prompt_hashes.is_empty());
    }

    #[test]
    fn message_text_truncates_on_a_char_boundary() {
        // A byte-index split mid-codepoint would panic; multi-byte text is
        // ordinary in real prompts, so this is a correctness guard, not a nicety.
        let body = serde_json::json!({
            "messages": [{"role": "user", "content": "é".repeat(4096)}]
        });
        let msg = &body["messages"][0];
        let text = message_text(msg, "text", 101);
        assert!(text.len() <= 101);
        assert!(text.chars().all(|c| c == 'é'));
    }

    /// The cascade end to end, through the function the request path actually
    /// calls.
    ///
    /// The unit tests above prove the two halves separately — `request_signals`
    /// reads the body, `resolve_session_with` picks the session — but neither
    /// proves `observe_request` wires one into the other. That wiring is what a
    /// refactor breaks silently: the cascade would be dead in production while
    /// every unit test stayed green.
    fn observed_session() -> Observation {
        let base = reqwest::Url::parse("http://127.0.0.1:1").unwrap();
        let st = BoundaryState::new(base, 0, 8, &[]);

        // Two live sessions: `sess_a` ran the tool call this request answers,
        // `sess_b` acted afterwards. Most-recently-active therefore says `sess_b`
        // and the content says `sess_a` — the disagreement is the whole test.
        st.registry.upsert_signals(
            "agt_test",
            "agt_test",
            "claude-code",
            "sess_a",
            &session::SessionSignals {
                tool_use_id: Some("toolu_aaa"),
                prompt: Some("build the parser"),
            },
        );
        std::thread::sleep(Duration::from_millis(5));
        st.registry
            .upsert("agt_test", "agt_test", "claude-code", "sess_b");

        let mut headers = HeaderMap::new();
        headers.insert(INSTALL_ID_HEADER, "agt_test".parse().unwrap());
        let bytes = Bytes::from(serde_json::to_vec(&conversation()).unwrap());

        observe_request(&st, WireFormat::AnthropicMessages, &headers, &bytes).0
    }

    #[test]
    fn observe_request_attributes_to_the_session_the_content_identifies() {
        let obs = observed_session();
        assert_eq!(obs.session.session_id.as_deref(), Some("sess_a"));
        assert_eq!(
            obs.session.assurance,
            crate::boundary::session::Assurance::Attested,
            "an exact tool-result join identifies the session; it does not guess"
        );
    }

    #[test]
    fn the_opaque_path_still_falls_back_and_says_it_guessed() {
        // The other half of the contract, and the only remaining way to reach the
        // pre-cascade pick now that the flag is gone. A body over the 32 MB
        // ceiling, or one with no usable Content-Length, is never read (F-17
        // zero-buffer) — so there are no signals to cascade on and recency
        // decides. It must still resolve, and must still be flagged a guess.
        let base = reqwest::Url::parse("http://127.0.0.1:1").unwrap();
        let st = BoundaryState::new(base, 0, 8, &[]);
        st.registry.upsert_signals(
            "agt_test",
            "agt_test",
            "claude-code",
            "sess_a",
            &session::SessionSignals {
                tool_use_id: Some("toolu_aaa"),
                prompt: Some("build the parser"),
            },
        );
        std::thread::sleep(Duration::from_millis(5));
        st.registry
            .upsert("agt_test", "agt_test", "claude-code", "sess_b");

        let mut headers = HeaderMap::new();
        headers.insert(INSTALL_ID_HEADER, "agt_test".parse().unwrap());
        let obs = observe_request_metadata_only(&st, WireFormat::AnthropicMessages, &headers);

        assert_eq!(obs.session.session_id.as_deref(), Some("sess_b"));
        assert_eq!(
            obs.session.assurance,
            crate::boundary::session::Assurance::Inferred
        );
    }

    /// The same disagreement as `observed_session`, on the Codex wire.
    ///
    /// Two live Codex sessions: `sess_a` ran the tool call this request carries
    /// back, `sess_b` acted afterwards. Most-recently-active therefore says
    /// `sess_b` and the content says `sess_a`.
    ///
    /// The hook half needs no change to make this work and that is the point: the
    /// registry is stamped from the raw hook payload's `tool_use_id` and `prompt`
    /// (`daemon::handlers`), and codex-cli 0.150.1 names both fields exactly as
    /// Claude Code does — so the join was always half-built, waiting on a request
    /// side that could read `input`.
    #[test]
    fn observe_request_attributes_a_codex_turn_to_the_session_that_ran_the_tool() {
        let base = reqwest::Url::parse("http://127.0.0.1:1").unwrap();
        let st = BoundaryState::new(base, 0, 8, &[]);

        st.registry.upsert_signals(
            "agt_test",
            "agt_test",
            "codex-cli",
            "sess_a",
            &session::SessionSignals {
                tool_use_id: Some("call_aaa"),
                prompt: Some("build the parser"),
            },
        );
        std::thread::sleep(Duration::from_millis(5));
        st.registry
            .upsert("agt_test", "agt_test", "codex-cli", "sess_b");

        let mut headers = HeaderMap::new();
        headers.insert(INSTALL_ID_HEADER, "agt_test".parse().unwrap());
        let bytes = Bytes::from(serde_json::to_vec(&codex_turn()).unwrap());
        let obs = observe_request(&st, WireFormat::OpenAiResponses, &headers, &bytes).0;

        assert_eq!(obs.session.session_id.as_deref(), Some("sess_a"));
        assert_eq!(
            obs.session.assurance,
            crate::boundary::session::Assurance::Attested,
            "an exact call-id join identifies the session; it does not guess"
        );
    }

    /// The prompt join alone, with no tool call in the request at all — the first
    /// turn of a Codex session, which is exactly when two concurrent sessions are
    /// most likely to be confused and no tool has run yet to tell them apart.
    #[test]
    fn a_codex_opening_turn_is_attributed_by_its_prompt() {
        let base = reqwest::Url::parse("http://127.0.0.1:1").unwrap();
        let st = BoundaryState::new(base, 0, 8, &[]);

        st.registry.upsert_signals(
            "agt_test",
            "agt_test",
            "codex-cli",
            "sess_a",
            &session::SessionSignals {
                tool_use_id: None,
                prompt: Some("build the parser"),
            },
        );
        std::thread::sleep(Duration::from_millis(5));
        st.registry
            .upsert("agt_test", "agt_test", "codex-cli", "sess_b");

        let body = serde_json::json!({"input": [
            {"type": "message", "role": "user",
             "content": [{"type": "input_text", "text": "build the parser"}]}
        ]});
        let mut headers = HeaderMap::new();
        headers.insert(INSTALL_ID_HEADER, "agt_test".parse().unwrap());
        let bytes = Bytes::from(serde_json::to_vec(&body).unwrap());
        let obs = observe_request(&st, WireFormat::OpenAiResponses, &headers, &bytes).0;

        assert_eq!(obs.session.session_id.as_deref(), Some("sess_a"));
        assert_eq!(
            obs.session.assurance,
            crate::boundary::session::Assurance::Attested
        );
    }

    /// Selector 0 for Codex, end to end, on the shape a real turn sends.
    ///
    /// The id is not corroborated by anything else here — `sess_a` and `sess_b`
    /// are both live, neither ran the tool this request carries, and the declared
    /// id names a session the registry has never heard of. It still wins, because
    /// an agent naming its own session is a claim about itself that needs no
    /// corroboration: an agent's FIRST model call races the `SessionStart` hook
    /// that would register it, which is exactly when no heuristic can help.
    #[test]
    fn a_codex_request_that_names_its_session_is_believed_over_the_joins() {
        let base = reqwest::Url::parse("http://127.0.0.1:1").unwrap();
        let st = BoundaryState::new(base, 0, 8, &[]);
        st.registry
            .upsert("agt_test", "agt_test", "codex-cli", "sess_a");
        std::thread::sleep(Duration::from_millis(5));
        st.registry
            .upsert("agt_test", "agt_test", "codex-cli", "sess_b");

        // `client_metadata` exactly as captured off codex-cli 0.150.1.
        let mut body = codex_turn();
        body["client_metadata"] = serde_json::json!({
            "session_id": "01a07677-988f-7901-af48-225c7386aabc",
            "thread_id": "01a07677-988f-7901-af48-225c7386aabc",
            "turn_id": "01a07677-7334-74a2-be87-bdd0bbe8fb77",
            "x-codex-installation-id": "5532dc79-4512-4bae-ab73-0dd3e814d329"
        });

        let mut headers = HeaderMap::new();
        headers.insert(INSTALL_ID_HEADER, "agt_test".parse().unwrap());
        let bytes = Bytes::from(serde_json::to_vec(&body).unwrap());
        let obs = observe_request(&st, WireFormat::OpenAiResponses, &headers, &bytes).0;

        assert_eq!(
            obs.session.session_id.as_deref(),
            Some("01a07677-988f-7901-af48-225c7386aabc"),
            "the request named its own session; nothing else gets a vote"
        );
        assert_eq!(
            obs.session.assurance,
            crate::boundary::session::Assurance::Attested
        );
    }

    #[test]
    fn content_length_parses_and_defaults() {
        let mut h = HeaderMap::new();
        assert_eq!(content_length(&h), None); // unknown length → opaque branch
        h.insert(CONTENT_LENGTH, HeaderValue::from_static("42"));
        assert_eq!(content_length(&h), Some(42));
    }

    #[test]
    fn forward_headers_strips_framing_keeps_credential() {
        let mut h = HeaderMap::new();
        h.insert(HOST, HeaderValue::from_static("127.0.0.1:7600"));
        h.insert(CONTENT_LENGTH, HeaderValue::from_static("10"));
        h.insert("x-api-key", HeaderValue::from_static("sk-ant-xyz"));
        h.insert("anthropic-version", HeaderValue::from_static("2023-06-01"));
        let out = forward_headers(&h);
        assert!(
            out.get(HOST).is_none(),
            "host must be stripped (reqwest sets it)"
        );
        assert!(
            out.get(CONTENT_LENGTH).is_none(),
            "content-length must be stripped"
        );
        assert_eq!(out.get("x-api-key").unwrap(), "sk-ant-xyz");
        assert_eq!(out.get("anthropic-version").unwrap(), "2023-06-01");
    }

    #[test]
    fn forward_headers_strips_accept_encoding_for_readable_usage() {
        // The usage scanner reads the SSE body as raw ASCII, so a compressed
        // response would never match and every turn would degrade to
        // `stream_interrupted`. Accept-Encoding must be dropped so upstream
        // returns identity-encoded bytes — while credentials pass through.
        let mut h = HeaderMap::new();
        h.insert(ACCEPT_ENCODING, HeaderValue::from_static("gzip, br, zstd"));
        h.insert("x-api-key", HeaderValue::from_static("sk-ant-xyz"));
        let out = forward_headers(&h);
        assert!(
            out.get(ACCEPT_ENCODING).is_none(),
            "accept-encoding must be stripped so the response body is scannable"
        );
        assert_eq!(out.get("x-api-key").unwrap(), "sk-ant-xyz");
    }

    #[test]
    fn forward_headers_strips_connection_listed_and_hop_by_hop() {
        // RFC 7230 §6.1: a proxy must drop every header named in the Connection
        // token list, plus the fixed hop-by-hop set — but pass credentials and
        // provider headers through verbatim.
        let mut h = HeaderMap::new();
        h.insert(CONNECTION, HeaderValue::from_static("x-internal-foo"));
        h.insert("x-internal-foo", HeaderValue::from_static("secret"));
        h.insert("keep-alive", HeaderValue::from_static("timeout=5"));
        h.insert("x-api-key", HeaderValue::from_static("sk-ant-xyz"));
        h.insert("authorization", HeaderValue::from_static("Bearer tok"));
        let out = forward_headers(&h);
        assert!(
            out.get("x-internal-foo").is_none(),
            "a Connection-listed header must be stripped"
        );
        assert!(
            out.get("keep-alive").is_none(),
            "Keep-Alive is hop-by-hop and must be stripped"
        );
        assert!(
            out.get(CONNECTION).is_none(),
            "Connection itself is stripped"
        );
        // Credentials and provider headers survive untouched.
        assert_eq!(out.get("x-api-key").unwrap(), "sk-ant-xyz");
        assert_eq!(out.get("authorization").unwrap(), "Bearer tok");
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn header_wait_times_out_to_synth_502() {
        // FIX 1 (D-05 hang guard): an upstream that accepts the connection but
        // never returns response headers must not wedge the request forever
        // (and must not pin a semaphore permit). With a short injected header
        // timeout, the materialized path degrades to a synthetic 502 and the
        // request RETURNS instead of hanging.
        let hang_port = mock::spawn_hang_after_accept().await;
        let base = reqwest::Url::parse(&format!("http://127.0.0.1:{hang_port}")).unwrap();
        let state = Arc::new(
            BoundaryState::new(base, 0, 8, &[]).with_header_timeout(Duration::from_millis(200)),
        );
        let port = serve_ephemeral(state).await;

        let resp = tokio::time::timeout(
            Duration::from_secs(5),
            crate::egress::client()
                .post(format!("http://127.0.0.1:{port}/v1/messages"))
                .header("content-type", "application/json")
                .body(br#"{"model":"x","messages":[]}"#.to_vec())
                .send(),
        )
        .await
        .expect("request must return within 5s, not hang")
        .unwrap();

        assert_eq!(resp.status(), StatusCode::BAD_GATEWAY);
        assert_eq!(
            resp.headers().get("x-openlatch-upstream").unwrap(),
            "unreachable"
        );
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn admin_status_endpoint_reports_the_listener() {
        // Acceptance #7: GET /admin/boundary/status is served locally (route
        // wins over the proxy fallback) and reports the listener's liveness.
        let port = serve_ephemeral(state_for(0)).await;
        let resp = crate::egress::client()
            .get(format!("http://127.0.0.1:{port}/admin/boundary/status"))
            .send()
            .await
            .unwrap();
        assert!(resp.status().is_success());
        let v: serde_json::Value = resp.json().await.unwrap();
        assert_eq!(v["status"], "up");
        assert!(v["port"].is_number());
        assert!(v["pass_through_failures"].is_number());
    }

    #[tokio::test]
    async fn synth_502_shape() {
        let r = synth_502();
        assert_eq!(r.status(), StatusCode::BAD_GATEWAY);
        assert_eq!(
            r.headers().get("x-openlatch-upstream").unwrap(),
            "unreachable"
        );
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn opaque_get_models_forwarded() {
        let up = mock::spawn_capture_200().await;
        let port = serve_ephemeral(state_for(up.port)).await;
        let resp = crate::egress::client()
            .get(format!("http://127.0.0.1:{port}/v1/models"))
            .send()
            .await
            .unwrap();
        assert!(resp.status().is_success());
        assert_eq!(resp.bytes().await.unwrap().as_ref(), b"ok");
        let line = up.received_request_line.lock().unwrap().clone().unwrap();
        assert!(
            line.starts_with("GET /v1/models"),
            "opaque path forwards verbatim: {line}"
        );
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn materialized_messages_forwards_body_and_credential() {
        let up = mock::spawn_capture_200().await;
        let port = serve_ephemeral(state_for(up.port)).await;
        let body = br#"{"model":"claude-opus-4-8","messages":[{"role":"user","content":"hi"}]}"#;
        let resp = crate::egress::client()
            .post(format!("http://127.0.0.1:{port}/v1/messages"))
            .header("content-type", "application/json")
            .header("x-api-key", "sk-ant-test123")
            .body(body.to_vec())
            .send()
            .await
            .unwrap();
        assert!(resp.status().is_success());
        // Body forwarded byte-identical (materialized path).
        assert_eq!(
            up.received_body.lock().unwrap().clone().unwrap(),
            body.to_vec()
        );
        // Caller credential passed VERBATIM.
        assert_eq!(up.header("x-api-key").as_deref(), Some("sk-ant-test123"));
        // Host rewritten to the upstream, not the boundary's host.
        assert_eq!(
            up.header("host").as_deref(),
            Some(format!("127.0.0.1:{}", up.port).as_str())
        );
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn breakpoint_body_round_trips_byte_identical() {
        let up = mock::spawn_capture_200().await;
        let port = serve_ephemeral(state_for(up.port)).await;
        // A body carrying cache_control breakpoints (D-06 / Acceptance #6).
        let body = br#"{"model":"claude-opus-4-8","system":[{"type":"text","text":"x","cache_control":{"type":"ephemeral"}}],"messages":[]}"#;
        let resp = crate::egress::client()
            .post(format!("http://127.0.0.1:{port}/v1/messages"))
            .header("content-type", "application/json")
            .body(body.to_vec())
            .send()
            .await
            .unwrap();
        assert!(resp.status().is_success());
        let received = up.received_body.lock().unwrap().clone().unwrap();
        // Hash equality == byte identity; cache_control survives untouched.
        assert_eq!(
            received,
            body.to_vec(),
            "cache_control breakpoint must round-trip byte-identically"
        );
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn count_tokens_forwarded_opaque_with_body() {
        // POST /v1/messages/count_tokens is NOT the captured path → opaque
        // forward, body still intact (F-16). Exercises stream_through_opaque
        // with a request body.
        let up = mock::spawn_capture_200().await;
        let port = serve_ephemeral(state_for(up.port)).await;
        let body = br#"{"model":"claude-opus-4-8","messages":[]}"#;
        let resp = crate::egress::client()
            .post(format!("http://127.0.0.1:{port}/v1/messages/count_tokens"))
            .header("content-type", "application/json")
            .body(body.to_vec())
            .send()
            .await
            .unwrap();
        assert!(resp.status().is_success());
        assert_eq!(
            up.received_body.lock().unwrap().clone().unwrap(),
            body.to_vec()
        );
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn unreachable_upstream_yields_synth_502() {
        // Point the boundary at a guaranteed-closed port → connect refused →
        // synthetic 502 + x-openlatch-upstream: unreachable (C-5b).
        let dead = mock::closed_port().await;
        let port = serve_ephemeral(state_for(dead)).await;
        let resp = crate::egress::client()
            .post(format!("http://127.0.0.1:{port}/v1/messages"))
            .header("content-type", "application/json")
            .body(br#"{"model":"x","messages":[]}"#.to_vec())
            .send()
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_GATEWAY);
        assert_eq!(
            resp.headers().get("x-openlatch-upstream").unwrap(),
            "unreachable"
        );
    }

    /// A failed forward must be counted somewhere the wiring supervisor can see.
    ///
    /// It watches this counter, not `pass_through_failures` — which counts
    /// fallible OpenLatch steps that degraded to forwarding unmodified, and is
    /// NOT touched by an unreachable upstream. Wiring the watchdog to that one
    /// would have left it blind to the exact failure it exists for: a provider
    /// that goes away while the agent is already pointed at us.
    #[tokio::test(flavor = "multi_thread")]
    async fn an_unreachable_upstream_is_counted_as_an_upstream_failure() {
        // Guards the "pass_through_failures did not move" assertion below.
        let _serialised = SCAN_PANIC_LOCK.lock().await;
        let dead = mock::closed_port().await;
        let port = serve_ephemeral(state_for(dead)).await;

        let upstream_before = upstream_failures();
        let pass_through_before = pass_through_failures();

        let resp = crate::egress::client()
            .post(format!("http://127.0.0.1:{port}/v1/messages"))
            .header("content-type", "application/json")
            .body(br#"{"model":"x","messages":[]}"#.to_vec())
            .send()
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_GATEWAY);

        assert!(
            upstream_failures() > upstream_before,
            "a request the agent got no answer to must be counted"
        );
        assert_eq!(
            pass_through_failures(),
            pass_through_before,
            "an unreachable upstream is not a degraded-to-pass-through step"
        );
    }

    /// The preflight probe travels the real forward path — that is the whole
    /// point of it — but must never land in the customer's economics data, and
    /// its marker must never reach the provider. An event would bill them for
    /// our health check; a header we invented is not one Anthropic agreed to
    /// receive.
    #[tokio::test(flavor = "multi_thread")]
    async fn preflight_probe_forwards_without_measuring_or_leaking_its_marker() {
        use crate::boundary::preflight::PREFLIGHT_HEADER;
        use crate::boundary::session::SessionRegistry;

        let up = mock::spawn_capture_200().await;
        let base = reqwest::Url::parse(&format!("http://127.0.0.1:{}", up.port)).unwrap();
        let (tx, mut rx) = tokio::sync::mpsc::channel(8);
        let reg = Arc::new(SessionRegistry::default());
        reg.upsert("agt_1", "agt_1", "claude-code", "sess_a");
        let state = Arc::new(BoundaryState::new(base, 0, 8, &[]).with_measurement(reg, Some(tx)));
        let port = serve_ephemeral(state).await;

        let resp = crate::egress::client()
            .post(format!("http://127.0.0.1:{port}/v1/messages"))
            .header("content-type", "application/json")
            .header("x-openlatch-install-id", "agt_1")
            .header(PREFLIGHT_HEADER, "1")
            .body(
                br#"{"model":"claude-opus-4-8","max_tokens":1,"messages":[{"role":"user","content":"ping"}]}"#
                    .to_vec(),
            )
            .send()
            .await
            .unwrap();
        assert!(
            resp.status().is_success(),
            "the probe must be forwarded like any other request"
        );
        // Drain: the economics event, if there were one, is emitted when the
        // response body guard drops — so asserting before this proves nothing.
        let _ = resp.bytes().await;

        for _ in 0..50 {
            if up.received_headers.lock().unwrap().is_some() {
                break;
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        let sent = up.received_headers.lock().unwrap().clone().unwrap();
        assert!(
            !sent.to_ascii_lowercase().contains(PREFLIGHT_HEADER),
            "the preflight marker must be stripped before the request leaves for the provider, \
             got headers: {sent}"
        );

        tokio::time::sleep(Duration::from_millis(50)).await;
        assert!(
            rx.try_recv().is_err(),
            "a preflight probe must emit ZERO economics events"
        );
    }

    /// Serialises every test whose response tee scans a chunk, because
    /// `INJECT_SCAN_PANIC` is a process-global and one of them arms it.
    ///
    /// It also covers tests that assert `pass_through_failures()` holds *still*:
    /// arming the scan panic bumps that same process-global counter, so an
    /// unsynchronised "must not move" assertion fails only in the full suite.
    ///
    /// A `tokio::sync::Mutex` rather than `std`: these are async tests and the
    /// guard is held across awaits.
    static SCAN_PANIC_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());

    /// The control for the test above: the SAME request without the marker is
    /// measured. Without this, "no event" would also pass if measurement were
    /// broken outright.
    #[tokio::test(flavor = "multi_thread")]
    async fn an_unmarked_request_on_the_same_path_is_measured() {
        // Held for the same reason the injecting test holds it: this one scans
        // chunks, and a scan under an armed `INJECT_SCAN_PANIC` panics.
        let _serialised = SCAN_PANIC_LOCK.lock().await;
        use crate::boundary::session::SessionRegistry;

        let up = mock::spawn_capture_200().await;
        let base = reqwest::Url::parse(&format!("http://127.0.0.1:{}", up.port)).unwrap();
        let (tx, mut rx) = tokio::sync::mpsc::channel(8);
        let reg = Arc::new(SessionRegistry::default());
        reg.upsert("agt_1", "agt_1", "claude-code", "sess_a");
        let state = Arc::new(BoundaryState::new(base, 0, 8, &[]).with_measurement(reg, Some(tx)));
        let port = serve_ephemeral(state).await;

        let resp = crate::egress::client()
            .post(format!("http://127.0.0.1:{port}/v1/messages"))
            .header("content-type", "application/json")
            .header("x-openlatch-install-id", "agt_1")
            .body(
                br#"{"model":"claude-opus-4-8","max_tokens":1,"messages":[{"role":"user","content":"ping"}]}"#
                    .to_vec(),
            )
            .send()
            .await
            .unwrap();
        let _ = resp.bytes().await;

        tokio::time::sleep(Duration::from_millis(50)).await;
        assert!(
            rx.try_recv().is_ok(),
            "an ordinary /v1/messages request must still emit its economics event"
        );
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn usage_scan_panic_degrades_to_unmeasured() {
        // FIX 3 (D-24): a panic inside the response usage scanner must NOT abort
        // the stream and must NOT emit a corrupt/partial event. Forwarding
        // continues, the request body is byte-identical, the response drains
        // cleanly, a pass-through failure is recorded, and the request is
        // UNMEASURED — exactly ZERO economics events.
        //
        // Isolation: `INJECT_SCAN_PANIC` is process-global, so every test whose
        // response tee scans a chunk has to be serialised against this one.
        //
        // This used to claim no sibling wired measurement, and one does —
        // `an_unmarked_request_on_the_same_path_is_measured`, the control for
        // this very test. Run in the same slice of the suite it inherited the
        // armed flag, panicked in its scan, and failed with "must still emit
        // its economics event". Reproducible with `cargo test --lib
        // boundary::proxy`, invisible in a full run, which is the worst shape a
        // flake can take.
        let _serialised = SCAN_PANIC_LOCK.lock().await;
        use crate::boundary::session::SessionRegistry;

        let up = mock::spawn_capture_usage_sse(10, 0, 0, 0, 0, 20).await;
        let base = reqwest::Url::parse(&format!("http://127.0.0.1:{}", up.port)).unwrap();
        let (tx, mut rx) = tokio::sync::mpsc::channel(8);
        let reg = Arc::new(SessionRegistry::default());
        reg.upsert("agt_1", "agt_1", "claude-code", "sess_a");
        let state = Arc::new(BoundaryState::new(base, 0, 8, &[]).with_measurement(reg, Some(tx)));
        let port = serve_ephemeral(state).await;

        let failures_before = pass_through_failures();
        set_inject_scan_panic(true);

        let body = br#"{"model":"claude-opus-4-8","stream":true,"messages":[{"role":"user","content":"hi"}]}"#.to_vec();
        let resp = crate::egress::client()
            .post(format!("http://127.0.0.1:{port}/v1/messages"))
            .header("content-type", "application/json")
            .header("x-openlatch-install-id", "agt_1")
            .body(body.clone())
            .send()
            .await
            .unwrap();
        assert!(
            resp.status().is_success(),
            "forward must complete despite the scan panic"
        );
        // The stream drains cleanly (not aborted) — the response bytes arrive.
        let received = resp.bytes().await.expect("response body drains cleanly");
        assert!(!received.is_empty(), "response body still flows through");

        set_inject_scan_panic(false);

        // Give the mock a moment to store the captured request body.
        for _ in 0..50 {
            if up.received_body.lock().unwrap().is_some() {
                break;
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }

        // Request body forwarded byte-identical (never mutated by the tee).
        assert_eq!(
            up.received_body.lock().unwrap().clone().unwrap(),
            body,
            "request body must be byte-identical despite the scan panic"
        );
        // The failure was recorded.
        assert!(
            pass_through_failures() > failures_before,
            "a usage-scan panic must be recorded as a pass-through failure"
        );
        // ZERO economics events — the request is UNMEASURED, not partial/corrupt.
        assert!(
            tokio::time::timeout(Duration::from_millis(500), rx.recv())
                .await
                .ok()
                .flatten()
                .is_none(),
            "a usage-scan panic must emit NO event (unmeasured, not corrupt)"
        );
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn streaming_is_zero_buffer() {
        // C-9b: the mock trickles 3 chunks with 60ms gaps. A streaming forward
        // delivers byte one to the client BEFORE the mock writes its final
        // chunk; a buffering forward would deliver nothing until the end.
        let gap = Duration::from_millis(60);
        let trickle = mock::spawn_trickle_sse(3, gap).await;
        let port = serve_ephemeral(state_for(trickle.port)).await;

        let resp = crate::egress::client()
            .post(format!("http://127.0.0.1:{port}/v1/messages"))
            .header("content-type", "application/json")
            .body(br#"{"model":"x","messages":[],"stream":true}"#.to_vec())
            .send()
            .await
            .unwrap();
        assert!(resp.status().is_success());

        let mut stream = resp.bytes_stream();
        let first = stream.next().await;
        let first_byte_at = Instant::now();
        assert!(first.is_some(), "expected at least one streamed chunk");
        assert!(first.unwrap().is_ok());

        // Drain the rest so the mock finishes and stamps final_written_at.
        while stream.next().await.is_some() {}

        let final_written_at = trickle
            .final_written_at
            .lock()
            .unwrap()
            .expect("mock must have finished writing");
        assert!(
            first_byte_at < final_written_at,
            "first client byte must arrive BEFORE the upstream stream completes (zero-buffer)"
        );
    }
}