openlatch-client 0.1.18

//! `openlatch bench <panic-isolation|port-stability>` — the two load-bearing
//! verification benches for plan 01. Each builds its own tokio runtime (the CLI
//! `dispatch` in `main.rs` is synchronous) and drives the real boundary code.
//!
//! - **panic-isolation (D-24):** inject a panic into `observe_request` and prove
//!   the request still completes UNMODIFIED, a pass-through failure is recorded,
//!   and the process PID is unchanged. Note (m-7): `main.rs` installs a global
//!   panic hook that fires `daemon_crashed` telemetry on ANY unwinding panic —
//!   including the one `catch_unwind` catches. That telemetry is a no-op here
//!   (no consent/handle in a bench) and is deliberately NOT the pass signal;
//!   this bench asserts PID stability directly and treats the
//!   `pass_through_failures` counter as the "failure recorded" signal.
//! - **port-stability (D-05/D-25):** prove the pinned port rebinds the SAME port
//!   within 2 s of release (no TIME_WAIT lockout, no silent re-probe) and that
//!   an occupied port is a LOUD failure, not a rebind elsewhere.

use std::path::Path;
use std::sync::Arc;
use std::time::{Duration, Instant};

use serde::{Deserialize, Serialize};
use serde_json::{json, Value};

use crate::error::{OlError, ERR_BOUNDARY_PORT_IN_USE, ERR_INVALID_CONFIG};

use super::capture::{CostBasis, Usage, UsageAccumulator};
use super::churn::ChurnTracker;
use super::emit::{assemble_data, assemble_event, Observation};
use super::session::{resolve_session, Assurance, Resolved, SessionRegistry};
use super::tokenize::Estimator;
use super::{bind_pinned, proxy, serve_ephemeral, BoundaryState};

/// Build a runtime for a bench.
fn runtime() -> Result<tokio::runtime::Runtime, OlError> {
    tokio::runtime::Runtime::new().map_err(|e| {
        OlError::new(
            ERR_INVALID_CONFIG,
            format!("failed to build bench runtime: {e}"),
        )
    })
}

/// `bench panic-isolation --inject observe`.
pub fn run_panic_isolation(inject: &str) -> Result<(), OlError> {
    let rt = runtime()?;
    rt.block_on(async move {
        let pid_before = std::process::id();

        // Mock upstream captures the forwarded body so we can prove it is
        // byte-identical despite the injected panic.
        let upstream = super::mock::spawn_capture_200().await;
        let upstream_base =
            reqwest::Url::parse(&format!("http://127.0.0.1:{}", upstream.port)).unwrap();
        let state = Arc::new(BoundaryState::new(upstream_base, 0, 4, &[]));
        let boundary_port = serve_ephemeral(state).await;

        // Arm the D-24 injection only for `--inject observe`.
        let armed = inject == "observe";
        proxy::set_inject_observe_panic(armed);

        let failures_before = proxy::pass_through_failures();
        let sent_body = br#"{"model":"claude-opus-4-8","messages":[]}"#.to_vec();

        let client = reqwest::Client::new();
        let resp = client
            .post(format!("http://127.0.0.1:{boundary_port}/v1/messages"))
            .header("content-type", "application/json")
            .header("x-api-key", "sk-ant-REDACTED")
            .body(sent_body.clone())
            .send()
            .await
            .map_err(|e| OlError::new(ERR_INVALID_CONFIG, format!("bench request failed: {e}")))?;

        let status = resp.status();
        let _ = resp.bytes().await;

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

        proxy::set_inject_observe_panic(false);

        let received = upstream.received_body.lock().unwrap().clone();
        let failures_after = proxy::pass_through_failures();
        let pid_after = std::process::id();

        // --- Assertions ---
        assert!(status.is_success(), "forward must complete (got {status})");
        assert_eq!(
            received.as_deref(),
            Some(sent_body.as_slice()),
            "forwarded body must be byte-identical despite the injected panic"
        );
        if armed {
            assert!(
                failures_after > failures_before,
                "an observe panic must be recorded as a pass-through failure"
            );
        }
        assert_eq!(
            pid_before, pid_after,
            "process PID must be unchanged (no crash)"
        );

        println!("PASS bench panic-isolation --inject {inject}");
        println!("  forward status        : {status}");
        println!("  body byte-identical   : yes");
        println!(
            "  pass-through failures : {} -> {} (recorded={})",
            failures_before, failures_after, armed
        );
        println!("  pid                   : {pid_before} (unchanged)");
        if armed {
            eprintln!(
                "  note: the panic backtrace printed above is the INJECTED panic — expected."
            );
        }
        Ok(())
    })
}

/// `bench port-stability`.
pub fn run_port_stability() -> Result<(), OlError> {
    let rt = runtime()?;
    rt.block_on(async move {
        // Stand in a currently-free port for the pinned port so the bench never
        // collides with a real running boundary.
        let probe = tokio::net::TcpListener::bind(("127.0.0.1", 0)).await?;
        let port = probe.local_addr()?.port();
        drop(probe);

        // (1)+(2): bind the pinned port, drop it, rebind the SAME port. The
        // whole cycle must complete well within 2 s — no TIME_WAIT lockout,
        // and crucially no re-probe onto a different port.
        let start = Instant::now();
        let l1 = bind_pinned(port).await?;
        assert_eq!(
            l1.local_addr()?.port(),
            port,
            "must bind the requested port"
        );
        drop(l1);
        let l2 = bind_pinned(port).await?;
        assert_eq!(
            l2.local_addr()?.port(),
            port,
            "rebind must be the SAME port"
        );
        let rebind_elapsed = start.elapsed();
        assert!(
            rebind_elapsed < Duration::from_secs(2),
            "rebind must complete within 2s (took {rebind_elapsed:?})"
        );

        // (3): occupied-at-startup → LOUD failure with the D-25 code, never a
        // silent rebind elsewhere. `l2` still holds the port.
        let occupied = bind_pinned(port).await;
        match occupied {
            Err(e) if e.code == ERR_BOUNDARY_PORT_IN_USE => {}
            Err(e) => {
                return Err(OlError::new(
                    ERR_INVALID_CONFIG,
                    format!("occupied bind failed with unexpected code {}", e.code),
                ))
            }
            Ok(_) => {
                return Err(OlError::new(
                    ERR_INVALID_CONFIG,
                    "occupied bind unexpectedly SUCCEEDED — silent re-probe risk",
                ))
            }
        }
        drop(l2);

        println!("PASS bench port-stability");
        println!("  pinned port           : {port}");
        println!("  rebind after release  : same port, {rebind_elapsed:?} (< 2s)");
        println!("  occupied-at-startup   : loud {ERR_BOUNDARY_PORT_IN_USE} (no re-probe)");
        Ok(())
    })
}

/// Build one representative measured `Observation` for the fixture corpus.
fn fixture_observation(event_id: &str, model: &str, model_known: bool) -> Observation {
    let mut obs = Observation::none();
    obs.measured = true;
    obs.event_id = event_id.to_string();
    obs.occurred_at = "2026-07-23T12:00:00Z".to_string();
    obs.model = Some(model.to_string());
    obs.model_known = model_known;
    obs.billing = super::billing::BillingMode::ApiKey;
    obs.install_id = "agt_fixture".to_string();
    obs.session = Resolved {
        agent_id: Some("agt_fixture".to_string()),
        source: Some("claude-code".to_string()),
        session_id: Some("sess_fixture".to_string()),
        assurance: Assurance::Attested,
    };
    obs.request_body_len = 4096;
    obs.has_breakpoint = true;
    obs
}

/// `bench export-fixtures` — write the canonical `ai.openlatch.economics.*`
/// event stream (JSONL on stdout) that gates I-2 starting (I-1 §7 DoD).
pub fn run_export_fixtures() -> Result<(), OlError> {
    let filter = crate::privacy::PrivacyFilter::new(&[]);

    // 1) Clean provider-reported streaming call carrying the C-3 shape
    //    (input=50, cache_read=100000 → platform total input 100050).
    let obs1 = fixture_observation(
        "0190a000-0000-7000-8000-000000000001",
        "claude-opus-4-8",
        true,
    );
    let usage1 = Usage {
        input_tokens: 50,
        cache_read: 100_000,
        cache_write: 2_048,
        eph_5m: 2_048,
        eph_1h: 0,
        output_tokens: 321,
    };
    let data1 = assemble_data(&obs1, &usage1, CostBasis::ProviderReported, None, true);
    let ev1 = assemble_event(&obs1, data1, &filter);

    // 2) Interrupted stream → local estimate, tokenizer_estimated.
    let obs2 = fixture_observation(
        "0190a000-0000-7000-8000-000000000002",
        "claude-sonnet-5",
        true,
    );
    let est = Estimator.estimate("claude-sonnet-5", obs2.request_body_len);
    let usage2 = Usage {
        input_tokens: est.input_tokens,
        ..Usage::default()
    };
    let data2 = assemble_data(
        &obs2,
        &usage2,
        CostBasis::TokenizerEstimated,
        Some(super::capture::CaptureGap::StreamInterrupted),
        false,
    );
    let ev2 = assemble_event(&obs2, data2, &filter);

    // 3) Unattributed provider_error (no session) — must not vanish.
    let mut obs3 = fixture_observation(
        "0190a000-0000-7000-8000-000000000003",
        "claude-opus-4-8",
        true,
    );
    obs3.session = Resolved {
        agent_id: None,
        source: None,
        session_id: None,
        assurance: Assurance::Unknown,
    };
    let data3 = assemble_data(
        &obs3,
        &Usage::default(),
        CostBasis::ProviderReported,
        Some(super::capture::CaptureGap::ProviderError),
        false,
    );
    let ev3 = assemble_event(&obs3, data3, &filter);

    // 4) A churning-prefix event (timestamp class) — the prefix_* columns.
    let obs4 = fixture_observation(
        "0190a000-0000-7000-8000-000000000004",
        "claude-opus-4-8",
        true,
    );
    let prev = br#"{"system":[{"text":"as of 2026-07-22"}],"messages":[]}"#;
    let cur = br#"{"system":[{"text":"as of 2026-07-23"}],"messages":[]}"#;
    let churn = super::churn::classify_churn(prev, cur).expect("fixture churn diverges");
    let mut obs4 = obs4;
    obs4.churn = Some(churn);
    let usage4 = Usage {
        input_tokens: 128,
        output_tokens: 64,
        ..Usage::default()
    };
    let data4 = assemble_data(&obs4, &usage4, CostBasis::ProviderReported, None, true);
    let ev4 = assemble_event(&obs4, data4, &filter);

    for ev in [ev1, ev2, ev3, ev4] {
        // The wire shape I-2 builds against is the CloudEvents envelope.
        println!(
            "{}",
            serde_json::to_string(&ev.envelope).map_err(|e| OlError::new(
                ERR_INVALID_CONFIG,
                format!("fixture serialize failed: {e}")
            ))?
        );
    }
    Ok(())
}

/// Loopback egress canary shared by the transform-egress and replay benches. Bind
/// a `127.0.0.1:0` listener, spawn an accept-loop that counts any inbound
/// connection, run `workload`, then (after a 50 ms settle) return
/// `(canary_port, accepts, workload_output)`. A non-zero `accepts` means the
/// workload dialled out — both callers assert it is 0, each with its own message.
/// The workload's own `Err`/panic short-circuits before the count is read, exactly
/// as the inlined versions did.
async fn with_egress_canary<T>(
    workload: impl std::future::Future<Output = Result<T, OlError>>,
) -> Result<(u16, u64, T), OlError> {
    use std::sync::atomic::{AtomicU64, Ordering};

    // Canary: any accept here would mean the workload dialled out.
    let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)).await?;
    let canary_port = listener.local_addr()?.port();
    let accepts = Arc::new(AtomicU64::new(0));
    let accepts_task = accepts.clone();
    tokio::spawn(async move {
        loop {
            if listener.accept().await.is_ok() {
                accepts_task.fetch_add(1, Ordering::Relaxed);
            }
        }
    });

    let out = workload.await?;

    // Give any (non-existent) connection a beat to land, then read the count.
    tokio::time::sleep(Duration::from_millis(50)).await;
    let accepts = accepts.load(Ordering::Relaxed);
    Ok((canary_port, accepts, out))
}

/// `bench transform-egress` — prove the capture path makes ZERO outbound
/// connections (D-08). The observe / tokenize / session / churn functions hold
/// **no** network client by construction; this runs the full synchronous
/// pipeline while a canary loopback listener watches for any connection, and
/// asserts it accepted none.
pub fn run_transform_egress() -> Result<(), OlError> {
    let rt = runtime()?;
    rt.block_on(async move {
        let (canary_port, accepts, (iterations, estimated_tokens)) =
            with_egress_canary(async move {
                // Run the capture pipeline N times — the exact functions the
                // request path uses (observe → session → churn → tokenize →
                // usage scan).
                let reg = SessionRegistry::default();
                let tracker = ChurnTracker::default();
                let est = Estimator;
                let iterations = 500u64;
                let mut estimated_tokens = 0u64;
                for i in 0..iterations {
                    reg.upsert("agt_canary", "agt_canary", "claude-code", "sess_1");
                    let resolved = resolve_session(&reg, "agt_canary");
                    assert_eq!(resolved.assurance, Assurance::Attested);

                    let body =
                        format!(r#"{{"model":"claude-opus-4-8","messages":[{{"n":{i}}}]}}"#);
                    let _ = tracker.observe("agt_canary", "sess_1", body.as_bytes());

                    estimated_tokens += est.estimate("claude-opus-4-8", body.len()).input_tokens;

                    let mut acc = UsageAccumulator::default();
                    acc.scan_chunk(
                        br#"data: {"type":"message_delta","usage":{"output_tokens":7,"input_tokens":3}}"#,
                    );
                    assert!(acc.has_usage());
                }
                Ok((iterations, estimated_tokens))
            })
            .await?;

        // The capture path must have made zero outbound connections.
        if accepts != 0 {
            return Err(OlError::new(
                ERR_INVALID_CONFIG,
                format!("capture path made {accepts} outbound connection(s) — expected 0"),
            ));
        }

        println!("PASS bench transform-egress");
        println!("  capture iterations    : {iterations}");
        println!("  canary port           : 127.0.0.1:{canary_port}");
        println!("  outbound connections  : 0 (capture holds no network client)");
        println!("  estimator sanity      : {estimated_tokens} tokens estimated offline");
        Ok(())
    })
}

/// A canonical request body that matches exactly one baseline rule — the fixed
/// input the `replay` bench re-evaluates. `OL-ECO-001` is a trimmable conversation
/// (no strip marker, so L-2 never matches); `OL-ECO-002` carries a marked system
/// block with a short history (≤ keep, so L-1 never matches). Each is engineered to
/// a net-positive `skipped_stage` would-have so the printed tuple is illustrative.
fn replay_fixture(rule: &str) -> Result<Value, OlError> {
    match rule {
        "OL-ECO-001" => {
            // 8 messages: 2 large removed, 6 tiny retained → net-positive trim.
            let mut messages = vec![
                json!({ "role": "user", "content": "a".repeat(400) }),
                json!({ "role": "assistant", "content": "a".repeat(400) }),
            ];
            for _ in 0..6 {
                messages.push(json!({ "role": "user", "content": "hi" }));
            }
            Ok(json!({ "model": "claude-opus-4-8", "messages": messages }))
        }
        "OL-ECO-002" => Ok(json!({
            "model": "claude-opus-4-8",
            "system": [
                { "type": "text", "text": format!("{} {}", super::transforms::STRIP_MARKER, "z".repeat(400)) },
                { "type": "text", "text": "k" }
            ],
            "messages": [ { "role": "user", "content": "hi" } ]
        })),
        other => Err(OlError::new(
            ERR_INVALID_CONFIG,
            format!("unknown replay rule '{other}' — expected OL-ECO-001 or OL-ECO-002"),
        )),
    }
}

/// `bench replay --rule <id> --runs <N>` (D-05 determinism).
///
/// Re-evaluate the fixed [`replay_fixture`] body `runs` times and prove every
/// would-have output is **byte-identical** — a deterministic transform is the whole
/// point of the replay tuple. A canary loopback listener runs throughout and
/// asserts **zero** outbound connections during transform evaluation (no
/// network/model call, ever — D-04/D-05). Prints the single canonical tuple.
pub fn run_replay(rule: &str, runs: u64) -> Result<(), OlError> {
    let rt = runtime()?;
    rt.block_on(async move {
        let body = replay_fixture(rule)?;

        let (canary_port, outbound, (first, canonical, identical)) =
            with_egress_canary(async move {
                // Evaluate `runs` times; the serialized tuple must never vary.
                let first = super::transforms::evaluate_would_have(&body)
                    .ok_or_else(|| {
                        OlError::new(
                            ERR_INVALID_CONFIG,
                            format!("replay fixture for {rule} matched no baseline rule"),
                        )
                    })?
                    .to_wire_object();
                let canonical = serde_json::to_string(&first).map_err(|e| {
                    OlError::new(ERR_INVALID_CONFIG, format!("tuple serialize failed: {e}"))
                })?;
                let mut identical = 1u64;
                for _ in 1..runs {
                    let out = super::transforms::evaluate_would_have(&body)
                        .ok_or_else(|| {
                            OlError::new(
                                ERR_INVALID_CONFIG,
                                "replay fixture stopped matching".to_string(),
                            )
                        })?
                        .to_wire_object();
                    let s = serde_json::to_string(&out).map_err(|e| {
                        OlError::new(ERR_INVALID_CONFIG, format!("tuple serialize failed: {e}"))
                    })?;
                    if s != canonical {
                        return Err(OlError::new(
                            ERR_INVALID_CONFIG,
                            format!(
                                "would-have output diverged on run {identical}: {s} != {canonical}"
                            ),
                        ));
                    }
                    identical += 1;
                }
                Ok((first, canonical, identical))
            })
            .await?;

        // Transform evaluation must have made zero outbound connections.
        if outbound != 0 {
            return Err(OlError::new(
                ERR_INVALID_CONFIG,
                format!("transform eval made {outbound} outbound connection(s) — expected 0"),
            ));
        }

        let rule_id = first["ai.openlatch.transform.rule_id"]
            .as_str()
            .unwrap_or(rule);
        println!("PASS bench replay --rule {rule} --runs {runs}");
        println!("  byte-identical outputs: {identical}/{runs}");
        println!("  outbound connections  : 0 (canary 127.0.0.1:{canary_port} accepted none)");
        println!("  rule                  : {rule_id}");
        println!("  would-have tuple      : {canonical}");
        Ok(())
    })
}

// ---------------------------------------------------------------------------
// D-14 / D-19: A/B cache-preservation harness + gate.
//
// `bench cache-baseline` runs a FIXED scripted conversation against Anthropic
// (through the boundary listener, or `--direct` as the control) and records the
// provider-reported usage + TTFT per turn. `bench compare` reads two such reports
// and applies the two-part G-2 gate: the no-MCP case is the D-19 release gate
// (the artifact shown to a customer to prove we do not degrade their cache); the
// MCP case is MEASURED & DISCLOSED, never gated.
//
// The measurement is honest — it comes from the SAME `UsageAccumulator` the
// forwarder uses, so the harness's per-request cache-read signal is exactly the
// signal plan 02 infers `cache.preserved` from (D-15 close-out).
// ---------------------------------------------------------------------------

/// PASS threshold: the layer must not drop the token-weighted cache-hit rate by
/// more than this many percentage points (one-sided — any improvement passes).
const MAX_CACHE_DEGRADATION_PP: f64 = 1.0;
/// PASS threshold: the layer must not add more than this to p95 TTFT.
const MAX_TTFT_DELTA_MS: f64 = 10.0;
/// A p95 is only well-defined over a real distribution — the gate needs at least
/// this many requests per pass (a p95 over 20 is essentially the max).
const MIN_SAMPLE_SIZE: usize = 100;
/// Default model. Override with `OPENLATCH_BENCH_MODEL` to run the bench cheaply.
const BENCH_MODEL_DEFAULT: &str = "claude-opus-4-8";
/// Anthropic API version header the bench sends on every request.
const ANTHROPIC_VERSION: &str = "2023-06-01";
/// Small output cap — the bench measures the input/cache path, not generation.
const BENCH_MAX_TOKENS: u64 = 32;

const EXPERIMENT_NOTE: &str = "Each pass uses a UNIQUE nonce in its system-prompt \
prefix so neither pass warms the other's cache: both start cold and pay their own \
cache writes. Never compare two passes that share a nonce.";
const ENVIRONMENT_NOTE: &str = "cache_hit_rate is token-weighted (tracks cost, not \
request counts). ttft_ms is time-to-first-response-chunk; p95 TTFT is only \
meaningful on an otherwise-idle machine over >= 100 requests.";

/// One recorded turn: the four provider-reported token counts (C-3) plus the
/// per-request time-to-first-token in milliseconds.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct TurnRecord {
    /// 0-based turn index within the scripted conversation.
    pub turn: usize,
    /// `input_tokens` — post-last-breakpoint only (never the total).
    pub input_tokens: u64,
    /// `cache_creation_input_tokens` — tokens written to the cache this turn.
    pub cache_creation_input_tokens: u64,
    /// `cache_read_input_tokens` — tokens served from the cache this turn.
    pub cache_read_input_tokens: u64,
    /// `output_tokens`.
    pub output_tokens: u64,
    /// Time-to-first-token (first response chunk), milliseconds.
    pub ttft_ms: u64,
}

impl TurnRecord {
    /// View this turn as a plan-02 [`Usage`] so the harness signal can be checked
    /// against `infer_cache_preserved` on identical numbers (D-15).
    pub fn usage(&self) -> Usage {
        Usage {
            input_tokens: self.input_tokens,
            cache_read: self.cache_read_input_tokens,
            cache_write: self.cache_creation_input_tokens,
            output_tokens: self.output_tokens,
            ..Usage::default()
        }
    }
}

/// Metadata header written at the top of every cache-baseline report.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct BenchHeader {
    /// The distinct per-pass nonce (see `experiment_note`).
    pub nonce: String,
    /// True when this pass went straight to the provider (`--direct`, control).
    pub direct: bool,
    /// True when MCP tool definitions were injected — `compare` reads this to
    /// decide whether to GATE (no-MCP) or merely DISCLOSE (MCP).
    pub with_mcp: bool,
    /// Number of turns requested.
    pub turns: usize,
    /// Model id the bench drove.
    pub model: String,
    /// The base URL the pass sent to.
    pub base_url: String,
    /// Wall-clock generation time (unix milliseconds).
    pub generated_unix_ms: u64,
    /// Why the nonce exists — experiment-validity note.
    pub experiment_note: String,
    /// How to read the numbers honestly — environment note.
    pub environment_note: String,
}

/// A full cache-baseline report: header + per-turn usage/TTFT records.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct CacheBaselineReport {
    pub header: BenchHeader,
    pub turns: Vec<TurnRecord>,
}

/// The two-part gate outcome. `gated == false` means "MCP workload — disclosed,
/// not gated"; `pass` is only meaningful when `gated == true`.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Gate {
    pub gated: bool,
    pub pass: bool,
    pub cache_ok: bool,
    pub ttft_ok: bool,
    pub sample_ok: bool,
}

/// The computed comparison between two reports.
#[derive(Clone, Debug)]
pub struct Comparison {
    pub rate_a: f64,
    pub rate_b: f64,
    /// Signed, B − A, in percentage points. Positive = B improved cache reuse.
    pub delta_pp: f64,
    pub p95_ttft_a: f64,
    pub p95_ttft_b: f64,
    /// p95(TTFT_B) − p95(TTFT_A), milliseconds. Positive = B slower.
    pub ttft_delta_ms: f64,
    pub n_a: usize,
    pub n_b: usize,
    pub gate: Gate,
}

/// Token-weighted cache-hit rate (D-14, G-2):
///
/// ```text
/// cache_read / (input + cache_creation + cache_read)
/// ```
///
/// Token-weighted so it tracks cost, not request counts. Cold-start requests sit
/// in the denominator for both runs, so a fair comparison cancels them.
pub fn cache_hit_rate(turns: &[TurnRecord]) -> f64 {
    let mut num: u128 = 0;
    let mut den: u128 = 0;
    for t in turns {
        num += u128::from(t.cache_read_input_tokens);
        den += u128::from(t.input_tokens)
            + u128::from(t.cache_creation_input_tokens)
            + u128::from(t.cache_read_input_tokens);
    }
    if den == 0 {
        0.0
    } else {
        num as f64 / den as f64
    }
}

/// p95 by nearest-rank over a TTFT distribution. Well-defined only over a real
/// distribution — with `MIN_SAMPLE_SIZE` samples the 95th percentile is a genuine
/// tail figure, not the maximum.
fn p95(samples: &[u64]) -> f64 {
    if samples.is_empty() {
        return 0.0;
    }
    let mut sorted = samples.to_vec();
    sorted.sort_unstable();
    let rank = (0.95_f64 * sorted.len() as f64).ceil() as usize;
    let idx = rank.clamp(1, sorted.len()) - 1;
    sorted[idx] as f64
}

/// Apply the two-part G-2 gate. MCP workloads are never gated (disclosure only).
fn evaluate_gate(
    delta_pp: f64,
    ttft_delta_ms: f64,
    n_a: usize,
    n_b: usize,
    with_mcp: bool,
) -> Gate {
    if with_mcp {
        return Gate {
            gated: false,
            pass: false,
            cache_ok: false,
            ttft_ok: false,
            sample_ok: false,
        };
    }
    // One-sided: an improvement (delta_pp > 0) always passes; only a degradation
    // beyond the threshold fails.
    let cache_ok = delta_pp >= -MAX_CACHE_DEGRADATION_PP;
    let ttft_ok = ttft_delta_ms <= MAX_TTFT_DELTA_MS;
    let sample_ok = n_a >= MIN_SAMPLE_SIZE && n_b >= MIN_SAMPLE_SIZE;
    Gate {
        gated: true,
        pass: cache_ok && ttft_ok && sample_ok,
        cache_ok,
        ttft_ok,
        sample_ok,
    }
}

/// Compute the full comparison between two reports. Rejects any pair that is not
/// a valid like-for-like experiment before computing: mismatched MCP workload,
/// a shared nonce (passes would warm each other's cache), reversed orientation
/// (A must be direct/baseline, B through-layer — the gate is directional),
/// mismatched models, or a turn-count shape where the header disagrees with the
/// recorded turns. These guards apply to the disclosed MCP path too; only the
/// *gating* is skipped for MCP.
pub fn compare_reports(
    a: &CacheBaselineReport,
    b: &CacheBaselineReport,
) -> Result<Comparison, OlError> {
    // --- Validity guards (D-19). These apply to BOTH the gated no-MCP case and
    // the MCP disclosure path: a disclosed number must still be a valid
    // like-for-like measurement — only the *gating* differs, never the shape,
    // orientation, nonce, or model checks. ---

    // Same workload shape: comparing a no-MCP pass against an MCP pass is not
    // like-for-like (MCP forces tool defs into the invalidatable prefix).
    if a.header.with_mcp != b.header.with_mcp {
        return Err(OlError::new(
            ERR_INVALID_CONFIG,
            format!(
                "bench compare: A (mcp={}) and B (mcp={}) are different workloads — \
                 comparing them is invalid; compare like-for-like.",
                a.header.with_mcp, b.header.with_mcp
            ),
        ));
    }

    // FIX 1 — distinct cache-key nonces. Each pass MUST use a different nonce so
    // neither warms the other's cache; equal nonces share a cache namespace and
    // make the comparison meaningless (one pass reads what the other wrote).
    if a.header.nonce == b.header.nonce {
        return Err(OlError::new(
            ERR_INVALID_CONFIG,
            format!(
                "bench compare: A and B share nonce {:?} — the two passes MUST use \
                 distinct cache-key nonces so neither warms the other's cache; a shared \
                 nonce means one pass reads the other's cache writes, so the comparison \
                 is invalid.",
                a.header.nonce
            ),
        ));
    }

    // FIX 2 — like-for-like orientation. A must be the direct/baseline/control
    // pass and B the through-layer pass (the `compare baseline.json withlayer.json`
    // order). The one-sided cache gate is DIRECTIONAL — swapping the files would
    // launder a real regression into a PASS — so a reversed pair is rejected.
    let orientation_ok = a.header.direct && !b.header.direct;
    if !orientation_ok {
        return Err(OlError::new(
            ERR_INVALID_CONFIG,
            format!(
                "bench compare: expected A=direct/baseline (direct=true) and \
                 B=through-layer (direct=false), got direct_a={} direct_b={} — the \
                 one-sided gate is directional; pass `compare baseline.json \
                 withlayer.json` in that order so a regression cannot be laundered by \
                 swapping the files.",
                a.header.direct, b.header.direct
            ),
        ));
    }

    // FIX 2 — same model on both passes. Different models have different
    // tokenizers and cache economics, so a delta between them is not attributable
    // to the layer.
    if a.header.model != b.header.model {
        return Err(OlError::new(
            ERR_INVALID_CONFIG,
            format!(
                "bench compare: A model {:?} != B model {:?} — both passes must drive \
                 the same model for the delta to be attributable to the layer.",
                a.header.model, b.header.model
            ),
        ));
    }

    // FIX 2 — shape: requested turn count must match across passes AND equal the
    // actual recorded turn count on both sides. A header whose stated `turns`
    // disagrees with `turns.len()` is a damaged report.
    if a.header.turns != b.header.turns
        || a.header.turns != a.turns.len()
        || b.header.turns != b.turns.len()
    {
        return Err(OlError::new(
            ERR_INVALID_CONFIG,
            format!(
                "bench compare: turn-count shape is invalid — header.turns A={} B={}, \
                 recorded turns A={} B={}; requested must equal actual on both sides and \
                 match across passes.",
                a.header.turns,
                b.header.turns,
                a.turns.len(),
                b.turns.len()
            ),
        ));
    }

    let rate_a = cache_hit_rate(&a.turns);
    let rate_b = cache_hit_rate(&b.turns);
    let delta_pp = (rate_b - rate_a) * 100.0;

    let ttft_a: Vec<u64> = a.turns.iter().map(|t| t.ttft_ms).collect();
    let ttft_b: Vec<u64> = b.turns.iter().map(|t| t.ttft_ms).collect();
    let p95_ttft_a = p95(&ttft_a);
    let p95_ttft_b = p95(&ttft_b);
    let ttft_delta_ms = p95_ttft_b - p95_ttft_a;

    let n_a = a.turns.len();
    let n_b = b.turns.len();
    let with_mcp = a.header.with_mcp;
    let gate = evaluate_gate(delta_pp, ttft_delta_ms, n_a, n_b, with_mcp);

    Ok(Comparison {
        rate_a,
        rate_b,
        delta_pp,
        p95_ttft_a,
        p95_ttft_b,
        ttft_delta_ms,
        n_a,
        n_b,
        gate,
    })
}

/// Render the comparison as the human report `bench compare` prints.
pub fn format_comparison(
    path_a: &Path,
    path_b: &Path,
    a: &CacheBaselineReport,
    b: &CacheBaselineReport,
    c: &Comparison,
) -> String {
    use std::fmt::Write as _;
    let mut s = String::new();
    let _ = writeln!(s, "bench compare — cache-preservation gate (D-19)");
    let _ = writeln!(
        s,
        "  A {}  (direct={}, mcp={}, n={}, nonce={})",
        path_a.display(),
        a.header.direct,
        a.header.with_mcp,
        c.n_a,
        a.header.nonce
    );
    let _ = writeln!(
        s,
        "  B {}  (direct={}, mcp={}, n={}, nonce={})",
        path_b.display(),
        b.header.direct,
        b.header.with_mcp,
        c.n_b,
        b.header.nonce
    );
    let _ = writeln!(s);
    let _ = writeln!(
        s,
        "  cache_hit_rate(A) : {:.4}  ({:.2}%)",
        c.rate_a,
        c.rate_a * 100.0
    );
    let _ = writeln!(
        s,
        "  cache_hit_rate(B) : {:.4}  ({:.2}%)",
        c.rate_b,
        c.rate_b * 100.0
    );
    let _ = writeln!(s, "  delta_pp (B-A)    : {:+.2} pp", c.delta_pp);
    let _ = writeln!(s, "  p95 TTFT(A)       : {:.0} ms", c.p95_ttft_a);
    let _ = writeln!(s, "  p95 TTFT(B)       : {:.0} ms", c.p95_ttft_b);
    let _ = writeln!(
        s,
        "  ttft_delta_ms     : {:+.1} ms  (p95_B - p95_A)",
        c.ttft_delta_ms
    );
    let _ = writeln!(s);
    let _ = writeln!(
        s,
        "  sample size       : A={} requests, B={} requests",
        c.n_a, c.n_b
    );
    let _ = writeln!(
        s,
        "  validity          : token-weighted rate; p95 valid only on an \
         otherwise-idle machine over >= {MIN_SAMPLE_SIZE} requests"
    );
    let _ = writeln!(s);
    if c.gate.gated {
        let verdict = if c.gate.pass { "PASS" } else { "FAIL" };
        let _ = writeln!(s, "  gate (no-MCP)     : {verdict}");
        let _ = writeln!(
            s,
            "    - cache degradation {:.2}pp {} {:.1}pp threshold",
            (-c.delta_pp).max(0.0),
            if c.gate.cache_ok { "<=" } else { ">" },
            MAX_CACHE_DEGRADATION_PP
        );
        let _ = writeln!(
            s,
            "    - p95 ttft delta {:+.1}ms {} {:.0}ms threshold",
            c.ttft_delta_ms,
            if c.gate.ttft_ok { "<=" } else { ">" },
            MAX_TTFT_DELTA_MS
        );
        let _ = writeln!(
            s,
            "    - sample size {} {} {MIN_SAMPLE_SIZE} minimum",
            c.n_a.min(c.n_b),
            if c.gate.sample_ok { ">=" } else { "<" },
        );
    } else {
        let _ = writeln!(
            s,
            "  gate              : DISCLOSED, NOT GATED (MCP workload)"
        );
        let _ = writeln!(
            s,
            "    MCP forces tool definitions into the invalidatable prefix behind a custom"
        );
        let _ = writeln!(
            s,
            "    base URL — a product-level disclosure, not an implementation bug. The delta"
        );
        let _ = writeln!(s, "    above is reported for the customer, never gated.");
    }
    s
}

/// `bench compare A.json B.json` — read two reports, print the gate report, and
/// exit non-zero if the (no-MCP) D-19 gate fails.
pub fn run_compare(a: &Path, b: &Path) -> Result<(), OlError> {
    let report_a = read_report(a)?;
    let report_b = read_report(b)?;
    let cmp = compare_reports(&report_a, &report_b)?;
    print!("{}", format_comparison(a, b, &report_a, &report_b, &cmp));

    if cmp.gate.gated && !cmp.gate.pass {
        return Err(OlError::new(
            ERR_INVALID_CONFIG,
            format!(
                "D-19 cache-preservation gate FAILED (delta_pp={:+.2}, ttft_delta_ms={:+.1}, \
                 n_a={}, n_b={})",
                cmp.delta_pp, cmp.ttft_delta_ms, cmp.n_a, cmp.n_b
            ),
        ));
    }
    Ok(())
}

fn read_report(path: &Path) -> Result<CacheBaselineReport, OlError> {
    let bytes = std::fs::read(path).map_err(|e| {
        OlError::new(
            ERR_INVALID_CONFIG,
            format!("bench compare: cannot read {}: {e}", path.display()),
        )
    })?;
    serde_json::from_slice(&bytes).map_err(|e| {
        OlError::new(
            ERR_INVALID_CONFIG,
            format!(
                "bench compare: {} is not a valid cache-baseline report: {e}",
                path.display()
            ),
        )
    })
}

/// Resolve the model id, honouring the `OPENLATCH_BENCH_MODEL` override.
fn bench_model() -> String {
    std::env::var("OPENLATCH_BENCH_MODEL")
        .ok()
        .filter(|m| !m.trim().is_empty())
        .unwrap_or_else(|| BENCH_MODEL_DEFAULT.to_string())
}

/// A distinct nonce per invocation (pid + nanosecond clock) so this pass never
/// warms another pass's cache. Dependency-free by design.
fn unique_nonce() -> String {
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    format!("olbench-{:x}-{nanos:x}", std::process::id())
}

fn now_unix_ms() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as u64)
        .unwrap_or(0)
}

/// Build the fixed, deterministic, cacheable system prefix. The nonce lives in
/// the prefix so this pass's cache key never collides with the other pass's; the
/// prefix is padded past the model's minimum cacheable size so it is actually
/// cached.
fn system_prefix(nonce: &str) -> String {
    let mut s = format!(
        "OpenLatch cache-preservation bench. Experiment nonce: {nonce}. This system \
         prompt is a fixed, deterministic, cacheable prefix. Answer every question \
         with only the number requested.\n\n"
    );
    let filler = "The quick brown fox jumps over the lazy dog. ";
    while s.len() < 12_000 {
        s.push_str(filler);
    }
    s
}

/// The MCP tool-definition block injected on `--with-mcp` — an approximation of
/// the tool defs an MCP-enabled agent forces into the invalidatable prefix.
fn mcp_tools() -> Value {
    json!([
        {
            "name": "read_file",
            "description": "Read a file from the workspace by path.",
            "input_schema": {
                "type": "object",
                "properties": {"path": {"type": "string"}},
                "required": ["path"]
            }
        },
        {
            "name": "search",
            "description": "Search the codebase for a query string.",
            "input_schema": {
                "type": "object",
                "properties": {"query": {"type": "string"}},
                "required": ["query"]
            }
        }
    ])
}

/// Assemble one `/v1/messages` request body for a turn.
fn build_request_body(model: &str, prefix: &str, user: &str, tools: Option<&Value>) -> Value {
    let mut body = json!({
        "model": model,
        "max_tokens": BENCH_MAX_TOKENS,
        "stream": true,
        "system": [
            {"type": "text", "text": prefix, "cache_control": {"type": "ephemeral"}}
        ],
        "messages": [
            {"role": "user", "content": user}
        ]
    });
    if let Some(t) = tools {
        body["tools"] = t.clone();
    }
    body
}

/// True if a COMPLETE SSE line carries a generated text token — a
/// `content_block_delta` event whose `delta.type == "text_delta"` (equivalently,
/// a `delta.text` string). This is the first-GENERATED-TOKEN signal TTFT must be
/// timed from (FIX 3): `message_start` / `ping` / `content_block_start` all arrive
/// earlier but carry no token, so timing TTFT at them understates true
/// time-to-first-token and can hide a token-latency regression.
fn line_is_text_delta(line: &[u8]) -> bool {
    let Ok(text) = std::str::from_utf8(line) else {
        return false;
    };
    for l in text.lines() {
        let l = l.trim_start();
        let payload = l.strip_prefix("data:").map(str::trim).unwrap_or(l);
        if !payload.starts_with('{') {
            continue;
        }
        // Cheap pre-filter before the serde parse: a text token line always
        // carries the literal "text_delta".
        if !payload.contains("text_delta") {
            continue;
        }
        if let Ok(v) = serde_json::from_str::<Value>(payload) {
            if v.get("type").and_then(Value::as_str) != Some("content_block_delta") {
                continue;
            }
            let delta = v.get("delta");
            let is_text_delta =
                delta.and_then(|d| d.get("type")).and_then(Value::as_str) == Some("text_delta");
            let has_text = delta
                .and_then(|d| d.get("text"))
                .and_then(Value::as_str)
                .is_some();
            if is_text_delta || has_text {
                return true;
            }
        }
    }
    false
}

/// Accumulates one turn's SSE stream. Buffers partial `data:` lines across network
/// chunk boundaries (FIX 4 — a usage JSON line split across two chunks is
/// otherwise unparsable), scans every COMPLETE line for provider usage with the
/// SAME [`UsageAccumulator`] the forwarder uses, and records TTFT at the first
/// generated text token (FIX 3).
#[derive(Default)]
struct TurnScanner {
    /// Bytes not yet terminated by a newline — carried to the next chunk.
    line_buf: Vec<u8>,
    acc: UsageAccumulator,
    ttft_ms: Option<u64>,
    /// Elapsed time of the most recent chunk — used to timestamp a trailing
    /// partial line flushed at end-of-stream.
    last_elapsed_ms: u64,
}

impl TurnScanner {
    /// Feed one network chunk that arrived `elapsed_ms` after the request started.
    fn push(&mut self, chunk: &[u8], elapsed_ms: u64) {
        self.last_elapsed_ms = elapsed_ms;
        self.line_buf.extend_from_slice(chunk);
        // Scan only COMPLETE lines; keep the trailing partial buffered so a JSON
        // line straddling a chunk boundary is scanned once fully assembled.
        while let Some(pos) = self.line_buf.iter().position(|&b| b == b'\n') {
            let line: Vec<u8> = self.line_buf.drain(..=pos).collect();
            self.scan_line(&line, elapsed_ms);
        }
    }

    fn scan_line(&mut self, line: &[u8], elapsed_ms: u64) {
        self.acc.scan_chunk(line);
        if self.ttft_ms.is_none() && line_is_text_delta(line) {
            self.ttft_ms = Some(elapsed_ms);
        }
    }

    /// Finalize the stream into `(usage, ttft_ms)` or a LOUD error. A release gate
    /// must fail on damaged measurement, never record a zero-filled turn (FIX 4):
    /// the turn is rejected unless a terminal provider-usage object was captured
    /// AND the cache denominator `input + cache_creation + cache_read > 0`.
    fn resolve(mut self, who: &str) -> Result<(Usage, u64), OlError> {
        // Flush any trailing partial line (a stream that ended without a final
        // newline) before deciding.
        if !self.line_buf.is_empty() {
            let line = std::mem::take(&mut self.line_buf);
            self.scan_line(&line, self.last_elapsed_ms);
        }

        let usage = self.acc.usage();
        let denom = usage.input_tokens + usage.cache_write + usage.cache_read;
        if !self.acc.is_terminal() || denom == 0 {
            return Err(OlError::new(
                ERR_INVALID_CONFIG,
                format!(
                    "bench turn from {who} produced no usable provider usage \
                     (terminal={}, input+cache_creation+cache_read={denom}) — a release \
                     gate must fail loudly on damaged measurement, never record a \
                     zero-filled turn.",
                    self.acc.is_terminal()
                ),
            ));
        }

        // TTFT is only set at the first generated text token; a stream that carried
        // terminal usage but never a text token has no valid time-to-first-token.
        let ttft = self.ttft_ms.ok_or_else(|| {
            OlError::new(
                ERR_INVALID_CONFIG,
                format!(
                    "bench turn from {who} recorded provider usage but never a generated \
                     text token — no valid time-to-first-token."
                ),
            )
        })?;
        Ok((usage, ttft))
    }
}

/// Send one turn and return its provider-reported [`Usage`] + TTFT (ms). Uses the
/// SAME [`UsageAccumulator`] the forwarder uses, so the harness signal equals the
/// captured signal (D-15).
async fn send_turn(
    client: &reqwest::Client,
    base_url: &str,
    api_key: &str,
    body: &Value,
) -> Result<(Usage, u64), OlError> {
    let url = format!("{}/v1/messages", base_url.trim_end_matches('/'));
    let payload = serde_json::to_vec(body).map_err(|e| {
        OlError::new(
            ERR_INVALID_CONFIG,
            format!("bench body serialize failed: {e}"),
        )
    })?;

    let started = Instant::now();
    let mut resp = client
        .post(&url)
        .header("x-api-key", api_key)
        .header("anthropic-version", ANTHROPIC_VERSION)
        .header("content-type", "application/json")
        .body(payload)
        .send()
        .await
        .map_err(|e| {
            OlError::new(
                ERR_INVALID_CONFIG,
                format!("bench request to {url} failed: {e}"),
            )
        })?;

    let status = resp.status();
    if !status.is_success() {
        let text = resp.text().await.unwrap_or_default();
        let snippet: String = text.chars().take(300).collect();
        return Err(OlError::new(
            ERR_INVALID_CONFIG,
            format!("bench turn got HTTP {status} from {url}: {snippet}"),
        ));
    }

    let mut scanner = TurnScanner::default();
    while let Some(chunk) = resp
        .chunk()
        .await
        .map_err(|e| OlError::new(ERR_INVALID_CONFIG, format!("bench stream read failed: {e}")))?
    {
        // Timestamp per network read; TTFT is later attributed to the first chunk
        // that completes a generated-text-token line (FIX 3). Keep scanning ALL
        // chunks for usage — never stop at the first token.
        let elapsed_ms = started.elapsed().as_millis() as u64;
        scanner.push(&chunk, elapsed_ms);
    }

    scanner.resolve(&url)
}

/// `bench cache-baseline --turns N [--with-mcp] [--direct] [--out FILE]`.
///
/// Runs a fixed scripted conversation and writes per-turn provider usage + TTFT
/// as JSON (to `out`, or stdout when omitted). Makes REAL network calls — needs a
/// real `ANTHROPIC_API_KEY`; dev/CI-only.
pub fn run_cache_baseline(
    turns: usize,
    with_mcp: bool,
    direct: bool,
    out: Option<&Path>,
) -> Result<(), OlError> {
    let api_key = std::env::var("ANTHROPIC_API_KEY")
        .ok()
        .filter(|k| !k.trim().is_empty())
        .ok_or_else(|| {
            OlError::new(
                ERR_INVALID_CONFIG,
                "bench cache-baseline needs a real ANTHROPIC_API_KEY — it makes REAL \
                 network calls and is dev/CI-only. Export ANTHROPIC_API_KEY and retry.",
            )
        })?;

    if turns == 0 {
        return Err(OlError::new(
            ERR_INVALID_CONFIG,
            "bench cache-baseline needs --turns >= 1 (the D-19 gate needs >= 100)",
        ));
    }

    let model = bench_model();
    let base_url = if direct {
        super::ANTHROPIC_BASE.to_string()
    } else {
        format!("http://127.0.0.1:{}", super::resolve_boundary_port())
    };
    let nonce = unique_nonce();
    let header = BenchHeader {
        nonce: nonce.clone(),
        direct,
        with_mcp,
        turns,
        model: model.clone(),
        base_url: base_url.clone(),
        generated_unix_ms: now_unix_ms(),
        experiment_note: EXPERIMENT_NOTE.to_string(),
        environment_note: ENVIRONMENT_NOTE.to_string(),
    };

    let prefix = system_prefix(&nonce);
    let tools = if with_mcp { Some(mcp_tools()) } else { None };

    let rt = runtime()?;
    let records = rt.block_on(async move {
        // Use the canonical boundary client (rustls, connect-timeout, NO overall
        // stream timeout) so the bench measures the same transport production
        // forwards over — fidelity matters for a cache-preservation gate.
        let client = super::build_boundary_client();
        let mut records = Vec::with_capacity(turns);
        for i in 0..turns {
            let user = format!("Question {i}: reply with only the number {i}.");
            let body = build_request_body(&model, &prefix, &user, tools.as_ref());
            let (usage, ttft_ms) = send_turn(&client, &base_url, &api_key, &body).await?;
            records.push(TurnRecord {
                turn: i,
                input_tokens: usage.input_tokens,
                cache_creation_input_tokens: usage.cache_write,
                cache_read_input_tokens: usage.cache_read,
                output_tokens: usage.output_tokens,
                ttft_ms,
            });
        }
        Ok::<_, OlError>(records)
    })?;

    let report = CacheBaselineReport {
        header,
        turns: records,
    };
    let json = serde_json::to_string_pretty(&report).map_err(|e| {
        OlError::new(
            ERR_INVALID_CONFIG,
            format!("bench report serialize failed: {e}"),
        )
    })?;

    match out {
        Some(path) => {
            std::fs::write(path, json.as_bytes()).map_err(|e| {
                OlError::new(
                    ERR_INVALID_CONFIG,
                    format!(
                        "bench cache-baseline: failed to write {}: {e}",
                        path.display()
                    ),
                )
            })?;
            eprintln!(
                "bench cache-baseline: wrote {} turns to {} (direct={direct}, mcp={with_mcp})",
                report.turns.len(),
                path.display()
            );
        }
        None => {
            println!("{json}");
            eprintln!(
                "bench cache-baseline: {} turns (direct={direct}, mcp={with_mcp}) — JSON on stdout",
                report.turns.len()
            );
        }
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::boundary::capture::infer_cache_preserved;

    /// Uniform turns: `n` requests, each with the given token counts + TTFT.
    fn uniform_turns(n: usize, input: u64, cache_read: u64, ttft: u64) -> Vec<TurnRecord> {
        (0..n)
            .map(|i| TurnRecord {
                turn: i,
                input_tokens: input,
                cache_creation_input_tokens: 0,
                cache_read_input_tokens: cache_read,
                output_tokens: 8,
                ttft_ms: ttft,
            })
            .collect()
    }

    fn report(turns: Vec<TurnRecord>, with_mcp: bool, direct: bool) -> CacheBaselineReport {
        let n = turns.len();
        CacheBaselineReport {
            header: BenchHeader {
                nonce: format!("test-{}", u64::from(direct)),
                direct,
                with_mcp,
                turns: n,
                model: "test-model".to_string(),
                base_url: "test".to_string(),
                generated_unix_ms: 0,
                experiment_note: String::new(),
                environment_note: String::new(),
            },
            turns,
        }
    }

    #[test]
    fn c3_cache_hit_rate_fixture() {
        // C-3: input=50, cache_read=100000 → rate = 100000/100050 ≈ 0.9995.
        let turns = vec![TurnRecord {
            turn: 0,
            input_tokens: 50,
            cache_creation_input_tokens: 0,
            cache_read_input_tokens: 100_000,
            output_tokens: 7,
            ttft_ms: 100,
        }];
        let rate = cache_hit_rate(&turns);
        assert!(
            (rate - 100_000.0 / 100_050.0).abs() < 1e-9,
            "exact fixture rate: {rate}"
        );
        assert!((rate - 0.9995).abs() < 1e-3, "≈ 0.9995: {rate}");
    }

    #[test]
    fn empty_rate_is_zero_not_nan() {
        assert_eq!(cache_hit_rate(&[]), 0.0);
    }

    #[test]
    fn delta_sign_positive_when_b_improves() {
        // A: cold everywhere (no cache reads → rate 0). B: heavy cache reads.
        let a = report(uniform_turns(100, 100, 0, 100), false, true);
        let b = report(uniform_turns(100, 100, 900, 100), false, false);
        let c = compare_reports(&a, &b).unwrap();
        assert!(c.rate_a.abs() < 1e-9, "A rate is 0: {}", c.rate_a);
        assert!(c.rate_b > 0.89, "B rate is high: {}", c.rate_b);
        assert!(c.delta_pp > 0.0, "B>A ⇒ positive delta: {}", c.delta_pp);
    }

    #[test]
    fn token_weighting_big_requests_dominate() {
        // 100 tiny cold requests + ONE huge cache-read request. Token-weighted the
        // rate is dominated by the big request (~0.999); a request-count average
        // would be ~1/101 ≈ 0.01.
        let mut turns = uniform_turns(100, 10, 0, 5);
        turns.push(TurnRecord {
            turn: 100,
            input_tokens: 10,
            cache_creation_input_tokens: 0,
            cache_read_input_tokens: 1_000_000,
            output_tokens: 8,
            ttft_ms: 5,
        });
        let rate = cache_hit_rate(&turns);
        assert!(
            rate > 0.99,
            "a few big-token requests dominate a token-weighted rate: {rate}"
        );
    }

    #[test]
    fn p95_over_100_is_well_defined_not_max() {
        // Distribution 1..=100. p95 nearest-rank = ceil(0.95*100)=95 → value 95,
        // strictly below the maximum (100). A p95 over 20 would be ~the max.
        let samples: Vec<u64> = (1..=100).collect();
        let v = p95(&samples);
        assert!((v - 95.0).abs() < 1e-9, "p95(1..=100) = 95: {v}");
        assert!(v < 100.0, "p95 is NOT the maximum");

        // Contrast: over only 20 samples the p95 collapses toward the max.
        let small: Vec<u64> = (1..=20).collect();
        assert!(
            p95(&small) >= 19.0,
            "p95 over 20 is essentially the max — why the gate needs >= 100"
        );
    }

    #[test]
    fn gate_passes_small_regression() {
        // 0.5pp cache degradation + 5ms p95 TTFT delta over 100 requests → PASS.
        let a = report(uniform_turns(100, 100, 900, 100), false, true); // rate 0.900
        let b = report(uniform_turns(100, 105, 895, 105), false, false); // rate 0.895
        let c = compare_reports(&a, &b).unwrap();
        assert!(
            (c.delta_pp - (-0.5)).abs() < 1e-9,
            "delta_pp: {}",
            c.delta_pp
        );
        assert!(
            (c.ttft_delta_ms - 5.0).abs() < 1e-9,
            "ttft: {}",
            c.ttft_delta_ms
        );
        assert!(
            c.gate.gated && c.gate.pass,
            "0.5pp/5ms case PASSES: {:?}",
            c.gate
        );
    }

    #[test]
    fn gate_fails_large_cache_regression() {
        // 2pp cache degradation → FAIL (TTFT within budget).
        let a = report(uniform_turns(100, 100, 900, 100), false, true); // 0.900
        let b = report(uniform_turns(100, 120, 880, 100), false, false); // 0.880
        let c = compare_reports(&a, &b).unwrap();
        assert!(
            (c.delta_pp - (-2.0)).abs() < 1e-9,
            "delta_pp: {}",
            c.delta_pp
        );
        assert!(c.gate.gated && !c.gate.pass, "2pp case FAILS: {:?}", c.gate);
        assert!(
            !c.gate.cache_ok && c.gate.ttft_ok,
            "cache fails, ttft ok: {:?}",
            c.gate
        );
    }

    #[test]
    fn gate_fails_ttft_regression() {
        // Cache identical, but 15ms p95 TTFT delta → FAIL.
        let a = report(uniform_turns(100, 100, 900, 100), false, true);
        let b = report(uniform_turns(100, 100, 900, 115), false, false);
        let c = compare_reports(&a, &b).unwrap();
        assert!(c.delta_pp.abs() < 1e-9, "no cache change: {}", c.delta_pp);
        assert!(
            (c.ttft_delta_ms - 15.0).abs() < 1e-9,
            "ttft: {}",
            c.ttft_delta_ms
        );
        assert!(
            c.gate.gated && !c.gate.pass,
            "15ms case FAILS: {:?}",
            c.gate
        );
        assert!(
            c.gate.cache_ok && !c.gate.ttft_ok,
            "cache ok, ttft fails: {:?}",
            c.gate
        );
    }

    #[test]
    fn improvement_passes_regardless_of_magnitude() {
        // One-sided: even a large improvement (B much better than A) passes.
        let a = report(uniform_turns(100, 500, 500, 100), false, true); // 0.5
        let b = report(uniform_turns(100, 50, 950, 100), false, false); // 0.95
        let c = compare_reports(&a, &b).unwrap();
        assert!(c.delta_pp > 40.0, "large improvement: {}", c.delta_pp);
        assert!(
            c.gate.gated && c.gate.pass,
            "improvement passes: {:?}",
            c.gate
        );
    }

    #[test]
    fn mcp_is_never_gated() {
        // A catastrophic MCP regression is DISCLOSED, never gated.
        let a = report(uniform_turns(100, 100, 900, 100), true, true);
        let b = report(uniform_turns(100, 1000, 0, 200), true, false);
        let c = compare_reports(&a, &b).unwrap();
        assert!(!c.gate.gated, "MCP workload is never gated: {:?}", c.gate);
        // run_compare must not error even on a huge MCP delta.
        let out = format_comparison(Path::new("a"), Path::new("b"), &a, &b, &c);
        assert!(
            out.contains("DISCLOSED, NOT GATED"),
            "disclosure label present"
        );
    }

    #[test]
    fn insufficient_sample_cannot_pass() {
        // Even a perfect delta cannot PASS with fewer than MIN_SAMPLE_SIZE requests.
        let a = report(uniform_turns(20, 100, 900, 100), false, true);
        let b = report(uniform_turns(20, 100, 900, 100), false, false);
        let c = compare_reports(&a, &b).unwrap();
        assert!(!c.gate.sample_ok, "20 < 100");
        assert!(!c.gate.pass, "cannot PASS under-sampled: {:?}", c.gate);
    }

    #[test]
    fn mismatched_mcp_is_an_error() {
        let a = report(uniform_turns(100, 100, 900, 100), false, true);
        let b = report(uniform_turns(100, 100, 900, 100), true, false);
        assert!(compare_reports(&a, &b).is_err(), "no-MCP vs MCP is invalid");
    }

    #[test]
    fn d15_harness_signal_agrees_with_cache_preserved() {
        // D-15 close-out: the harness's per-request cache-read signal is EXACTLY
        // what plan 02 infers cache.preserved from — cache_read>0 ⇒ preserved true.
        let hit = TurnRecord {
            turn: 0,
            input_tokens: 50,
            cache_creation_input_tokens: 0,
            cache_read_input_tokens: 100_000, // C-3 shape
            output_tokens: 7,
            ttft_ms: 100,
        };
        let miss = TurnRecord {
            turn: 1,
            input_tokens: 4096,
            cache_creation_input_tokens: 2048,
            cache_read_input_tokens: 0,
            output_tokens: 7,
            ttft_ms: 100,
        };
        for t in [&hit, &miss] {
            assert_eq!(
                infer_cache_preserved(&t.usage()),
                t.cache_read_input_tokens > 0,
                "harness cache-read signal must agree with cache.preserved"
            );
        }
        assert!(infer_cache_preserved(&hit.usage()));
        assert!(!infer_cache_preserved(&miss.usage()));
    }

    #[test]
    fn report_json_roundtrips() {
        let original = report(uniform_turns(3, 100, 900, 100), false, true);
        let json = serde_json::to_string_pretty(&original).unwrap();
        let parsed: CacheBaselineReport = serde_json::from_str(&json).unwrap();
        assert_eq!(original, parsed, "report survives a JSON round-trip");
        // The wire field names match provider terminology.
        assert!(json.contains("cache_read_input_tokens"));
        assert!(json.contains("cache_creation_input_tokens"));
    }

    #[test]
    fn build_request_body_has_cacheable_prefix_and_optional_tools() {
        let prefix = system_prefix("nonce-abc");
        assert!(
            prefix.len() >= 12_000,
            "prefix padded past the cache minimum"
        );
        assert!(prefix.contains("nonce-abc"), "nonce is in the prefix");

        let no_tools = build_request_body("m", &prefix, "hi", None);
        assert!(
            no_tools.get("tools").is_none(),
            "no tools without --with-mcp"
        );
        assert_eq!(
            no_tools["system"][0]["cache_control"]["type"], "ephemeral",
            "the prefix carries a cache breakpoint"
        );

        let tools = mcp_tools();
        let with_tools = build_request_body("m", &prefix, "hi", Some(&tools));
        assert!(
            with_tools.get("tools").is_some(),
            "tools injected on --with-mcp"
        );
    }

    #[test]
    fn distinct_nonces_per_invocation() {
        // Experiment validity: two invocations never share a nonce, so neither
        // pass warms the other's cache.
        let a = unique_nonce();
        let b = unique_nonce();
        assert_ne!(a, b, "each invocation gets a distinct nonce");
    }

    // --- FIX 1: compare rejects EQUAL nonces ---

    #[test]
    fn compare_rejects_equal_nonces() {
        // Equal nonces share a cache namespace — one pass warms the other, so the
        // comparison is invalid and must error.
        let mut a = report(uniform_turns(100, 100, 900, 100), false, true);
        let mut b = report(uniform_turns(100, 100, 900, 100), false, false);
        a.header.nonce = "shared-nonce".to_string();
        b.header.nonce = "shared-nonce".to_string();
        assert!(
            compare_reports(&a, &b).is_err(),
            "identical nonces must be rejected — the passes share a cache namespace"
        );
    }

    // --- FIX 2: compare enforces like-for-like orientation + shape ---

    #[test]
    fn compare_rejects_reversed_orientation() {
        // A=through-layer (direct=false), B=direct (direct=true) — reversed. The
        // one-sided gate is directional; a reversed pair could launder a real
        // regression into a PASS, so it must error.
        let a = report(uniform_turns(100, 100, 900, 100), false, false); // nonce test-0
        let b = report(uniform_turns(100, 100, 900, 100), false, true); // nonce test-1
        assert!(
            compare_reports(&a, &b).is_err(),
            "reversed orientation (A=layer, B=direct) must error"
        );
    }

    #[test]
    fn compare_rejects_mismatched_models() {
        // Same orientation + distinct nonces, but different models → not like-for-like.
        let mut a = report(uniform_turns(100, 100, 900, 100), false, true);
        let b = report(uniform_turns(100, 100, 900, 100), false, false);
        a.header.model = "claude-opus-4-8".to_string(); // b stays "test-model"
        assert!(
            compare_reports(&a, &b).is_err(),
            "different models are not a like-for-like comparison"
        );
    }

    #[test]
    fn compare_rejects_header_turns_disagreeing_with_body() {
        // header.turns claims 101 but the body has 100 recorded turns → damaged report.
        let mut a = report(uniform_turns(100, 100, 900, 100), false, true);
        let b = report(uniform_turns(100, 100, 900, 100), false, false);
        a.header.turns = a.turns.len() + 1;
        assert!(
            compare_reports(&a, &b).is_err(),
            "header.turns must equal turns.len() on both sides"
        );
    }

    #[test]
    fn compare_accepts_correctly_oriented_no_mcp_pair() {
        // A=direct/baseline, B=through-layer, distinct nonces, same model, matching
        // turns → a valid like-for-like comparison that still gates.
        let a = report(uniform_turns(100, 100, 900, 100), false, true);
        let b = report(uniform_turns(100, 100, 900, 100), false, false);
        let c = compare_reports(&a, &b).expect("correctly-oriented no-MCP pair is valid");
        assert!(c.gate.gated, "a valid no-MCP pair is gated: {:?}", c.gate);
    }

    // --- FIX 3: TTFT is the first GENERATED text token, not message_start ---

    #[test]
    fn ttft_is_first_text_delta_not_message_start() {
        let mut sc = TurnScanner::default();
        // message_start arrives first (setup, no token) at t=5ms.
        sc.push(
            br#"data: {"type":"message_start","message":{"usage":{"input_tokens":10,"cache_read_input_tokens":5,"output_tokens":1}}}
"#,
            5,
        );
        // ping at t=7ms — still no token.
        sc.push(
            br#"data: {"type":"ping"}
"#,
            7,
        );
        // First generated text token at t=42ms — THIS is TTFT.
        sc.push(
            br#"data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"4"}}
"#,
            42,
        );
        // More deltas + terminal usage at t=50ms (keep scanning past first token).
        sc.push(
            br#"data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"2"}}
data: {"type":"message_delta","usage":{"output_tokens":2}}
"#,
            50,
        );
        let (usage, ttft) = sc.resolve("test").expect("valid stream resolves");
        assert_eq!(
            ttft, 42,
            "TTFT is the first content_block_delta text token (42ms), NOT message_start (5ms)"
        );
        assert_eq!(usage.input_tokens, 10, "usage still captured");
        assert_eq!(usage.cache_read, 5);
        assert_eq!(
            usage.output_tokens, 2,
            "output grew via the terminal message_delta"
        );
    }

    // --- FIX 4: split-line usage is captured; missing terminal usage ERRORS ---

    #[test]
    fn split_usage_line_across_chunks_is_captured() {
        // The message_start usage JSON line is split mid-object across two network
        // chunks. Line-buffering must reassemble it so usage is still captured.
        let mut sc = TurnScanner::default();
        let full = r#"data: {"type":"message_start","message":{"usage":{"input_tokens":123,"cache_read_input_tokens":456,"cache_creation_input_tokens":7,"output_tokens":1}}}
"#;
        let split_at = 40; // mid-JSON, no newline in the first half
        let bytes = full.as_bytes();
        sc.push(&bytes[..split_at], 1);
        sc.push(&bytes[split_at..], 2);
        // A generated token (for TTFT) and a terminal usage event.
        sc.push(
            br#"data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"x"}}
"#,
            3,
        );
        sc.push(
            br#"data: {"type":"message_delta","usage":{"output_tokens":9}}
"#,
            4,
        );
        let (usage, ttft) = sc.resolve("test").expect("split usage line still resolves");
        assert_eq!(
            usage.input_tokens, 123,
            "input captured despite the split line"
        );
        assert_eq!(
            usage.cache_read, 456,
            "cache_read captured despite the split line"
        );
        assert_eq!(
            usage.cache_write, 7,
            "cache_creation captured despite the split line"
        );
        assert_eq!(ttft, 3, "TTFT at the text token");
    }

    #[test]
    fn no_terminal_usage_errors_not_zero_turn() {
        // Only message_start (preliminary, NOT terminal) — the stream is damaged
        // (no message_delta). A release gate must ERROR, never record a zero- or
        // partially-filled turn.
        let mut sc = TurnScanner::default();
        sc.push(
            br#"data: {"type":"message_start","message":{"usage":{"input_tokens":10,"cache_read_input_tokens":5,"output_tokens":1}}}
"#,
            1,
        );
        sc.push(
            br#"data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"x"}}
"#,
            2,
        );
        assert!(
            sc.resolve("test").is_err(),
            "no terminal provider usage must error, never a zero-filled turn"
        );
    }

    #[test]
    fn empty_stream_errors_not_zero_turn() {
        // A stream with no usage at all (e.g. a provider error body that never
        // parsed) must error rather than record an all-zero turn.
        let mut sc = TurnScanner::default();
        sc.push(
            br#"data: {"type":"ping"}
"#,
            1,
        );
        assert!(
            sc.resolve("test").is_err(),
            "a usage-less stream must error, never record zeros"
        );
    }
}