openlatch-client 0.1.18

//! 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).
//!
//! 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, Method, 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::resolve_session;
use super::tokenize::{classify_model, Estimator};
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);
}

/// 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, headers: &HeaderMap, bytes: &Bytes) -> Observation {
    // 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.
    let session = resolve_session(&st.registry, &install_id);

    // 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);

    Observation {
        measured: true,
        event_id,
        occurred_at,
        model,
        model_known,
        billing,
        install_id,
        session,
        pricing,
        churn,
        request_body_len: bytes.len(),
        has_breakpoint: capture::has_cache_breakpoint(bytes),
        transform,
    }
}

/// 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, 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,
        // 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()
    }
}

/// 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>, headers: &HeaderMap) -> Option<MeasureCtx> {
    if is_preflight(headers) {
        return None;
    }
    st.cloud_tx.as_ref().map(|_| MeasureCtx {
        obs: observe_request_metadata_only(st, headers),
        tokenizer: st.tokenizer,
        cloud_tx: st.cloud_tx.clone(),
        privacy: st.privacy.clone(),
        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();
    let is_messages = parts.method == Method::POST && parts.uri.path() == "/v1/messages";

    // --- Opaque path: anything we don't capture. NO permit, NO materialize. ---
    // GET /v1/models, POST /v1/messages/count_tokens, batch endpoints, etc. A
    // non-`/v1/messages` request is never an economics event, so it carries NO
    // measurement context.
    if !is_messages {
        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, &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, &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 = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        observe_request(&st, &parts.headers, &bytes)
    }))
    .unwrap_or_else(|_| {
        record_pass_through_failure("observe_panic");
        Observation::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;
    }

    // FORWARD `bytes` verbatim — byte-identical, so cache_control survives (D-06).
    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();
        }
    };
    // 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 = st
        .client
        .request(parts.method.clone(), url)
        .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(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.
    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(),
            // Materialized path: the request body WAS parsed — full wire format.
            wire_format_unknown: false,
        })
    } 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();
        }
    };
    let send = st
        .client
        .request(parts.method.clone(), url)
        .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(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).
fn relay(
    upstream: Result<reqwest::Response, reqwest::Error>,
    permit: Option<OwnedSemaphorePermit>,
    measure_ctx: Option<MeasureCtx>,
) -> Response {
    let resp = match upstream {
        Ok(r) => r,
        // No upstream response exists — the one place an OpenLatch-shaped signal
        // is unavoidable. Record the failed call, then synth 502 (C-5b).
        Err(_) => {
            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 pinned base + the inbound path-and-query.
///
/// A leading-slash path-and-query is an absolute-path reference, so `join`
/// keeps the base's scheme + host and replaces only the path + query.
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("/");
    st.upstream_base.join(pq).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.
pub async fn boundary_status(State(st): State<Arc<BoundaryState>>) -> Json<serde_json::Value> {
    let verdict = st.wiring.verdict();
    Json(serde_json::json!({
        "status": "up",
        "port": st.port,
        "upstream": st.upstream_base.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(),
        "wired": st.wiring.is_wired(),
        "preflight": verdict.label(),
        "preflight_error": verdict.error(),
    }))
}

/// 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 / 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).
            (
                self.acc.usage(),
                CostBasis::ProviderReported,
                base_gap(&self.obs),
            )
        } 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() {
                    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(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, &[]))
    }

    #[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),
            reqwest::Client::new()
                .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 = reqwest::Client::new()
            .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 = reqwest::Client::new()
            .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 = reqwest::Client::new()
            .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 = reqwest::Client::new()
            .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 = reqwest::Client::new()
            .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 = reqwest::Client::new()
            .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() {
        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 = reqwest::Client::new()
            .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 = reqwest::Client::new()
            .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"
        );
    }

    /// 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() {
        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 = reqwest::Client::new()
            .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 note: no OTHER unit test in this binary wires measurement, so
        // it is the only one whose response tee scans a chunk. Arming the global
        // `INJECT_SCAN_PANIC` here therefore cannot cross-contaminate a sibling
        // test's scan (the integration tests run in a separate process).
        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 = reqwest::Client::new()
            .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 = reqwest::Client::new()
            .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)"
        );
    }
}