openlatch-client 0.6.5

//! Cursor's Connect-RPC streams, decoded on a mirror that can be dropped (Cursor I-2 plan 03).
//!
//! Cursor sends every model request to its own backend as Connect-RPC (protobuf), so the relay
//! can say which model ran, in which conversation and for how many tokens only by reading those
//! bytes. It reads a **copy**: the forward path offers each chunk to a bounded channel with
//! `try_send` and moves on ([`Teed`]); a full channel abandons the copy and marks the record
//! `partial`, and the stream completes untouched. A decode failure costs a field, never a turn
//! and never a record (D-13).
//!
//! | Route | Mirrored | Yields |
//! |---|---|---|
//! | `agent.v1.AgentService/Run` (HTTP/2 bidi) | request + response | model + conversation id from the first request frame, tokens from the response |
//! | `agent.v1.AgentService/RunSSE` | request + response | the pairing key from the request, tokens from the response |
//! | `aiserver.v1.BidiService/BidiAppend` (unary) | request | the pairing key → model + conversation id, into [`PairingTable`] |
//!
//! `RunSSE`'s request is mirrored too, which the plan's §2 did not list: the pairing key 3b
//! depends on lives in that request body, a single `BidiRequestId`.
//!
//! **No content.** Only the fields named in [`field`] are read, each into a model id, a
//! conversation id or a count; text deltas, the conversation state, tool traffic and every
//! credential sub-field of `model_details` are walked past, never copied. Raw frames are kept
//! only for a stream that decoded to nothing, response frames only, capped (D-17).
//!
//! **Who emits.** The decoder task owns the record: `GuardedBody`'s finalizer fires at end of
//! stream while the decoder may still be draining, and waiting for it there would put
//! backpressure on the forward (D-16). A panic inside the decoder surfaces as
//! `JoinError::is_panic` on its handle and still emits an `undecoded` record.

use std::collections::HashMap;
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll};
use std::time::{Duration, Instant};

use axum::http::{HeaderMap, StatusCode};
use bytes::Bytes;
use futures_core::Stream;
use tokio::sync::{mpsc, oneshot};

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

use super::capture::{CaptureDecode, CaptureGap};
use super::emit::{self, Observation};
use super::session::{self, SessionRegistry};
use super::tokenize::classify_model;
use super::ModelRelayState;

/// Recovered protobuf field numbers — the ONLY fields this module reads.
///
/// Source: `burpheart/cursor-tap`, `cursor_proto/agent_v1.proto` and `aiserver_v1.proto` at
/// commit `4c29ff5` (2026-01-31), extracted from the Cursor client's JavaScript bundle (the
/// January 2026 IDE release that commit captured). These are a third party's recovery, not a
/// published contract, and **they will drift**: a moved number reads as an absent field, so the
/// record degrades to `partial`/`undecoded` rather than failing. Change them here and nowhere
/// else.
pub mod field {
    /// `AgentClientMessage.run_request` (oneof) → `AgentRunRequest`.
    pub const CLIENT_RUN_REQUEST: u32 = 1;
    /// `AgentRunRequest.model_details` → `ModelDetails`.
    pub const RUN_MODEL_DETAILS: u32 = 3;
    /// `AgentRunRequest.conversation_id` (string).
    pub const RUN_CONVERSATION_ID: u32 = 5;
    /// `AgentRunRequest.requested_model` → `RequestedModel`. The model fallback: the plan names
    /// `providerOptions.cursor.modelName`, which the recovered protos carry no field for.
    pub const RUN_REQUESTED_MODEL: u32 = 9;
    /// `ModelDetails.model_id` (string). Its siblings 8–10 are credentials and are never read.
    pub const MODEL_DETAILS_MODEL_ID: u32 = 1;
    /// `RequestedModel.model_id` (string). Its siblings 4–6 are credentials and are never read.
    pub const REQUESTED_MODEL_MODEL_ID: u32 = 1;
    /// `AgentServerMessage.interaction_update` (oneof) → `InteractionUpdate`.
    pub const SERVER_INTERACTION_UPDATE: u32 = 1;
    /// `InteractionUpdate.token_delta` (oneof) → `TokenDeltaUpdate`.
    pub const UPDATE_TOKEN_DELTA: u32 = 8;
    /// `TokenDeltaUpdate.tokens` (int32).
    pub const TOKEN_DELTA_TOKENS: u32 = 1;
    /// `BidiAppendRequest.data` (string: hex of an `AgentClientMessage`).
    pub const APPEND_DATA: u32 = 1;
    /// `BidiAppendRequest.request_id` → `BidiRequestId`.
    pub const APPEND_REQUEST_ID: u32 = 2;
    /// `BidiRequestId.request_id` (string) — `RunSSE`'s whole request body.
    pub const BIDI_REQUEST_ID: u32 = 1;
}

/// A decoded frame is buffered whole, and never past this (D-14 step 2). Over it, the frame is
/// skipped and the record is `partial`.
pub const MAX_FRAME: usize = 4 << 20;
/// Mirror capacity in chunks (D-13).
pub const MIRROR_ITEMS: usize = 256;
/// Mirror capacity in queued bytes (D-13). Crossing either bound abandons the mirror.
pub const MIRROR_BYTES: usize = 8 << 20;
/// The response-only raw accumulator, and the size of a kept raw file (D-17).
pub const RAW_CAP: usize = 2 << 20;
/// At most this many raw files are kept (D-17).
pub const RAW_MAX_FILES: usize = 64;
/// Pairing table capacity (D-15).
pub const PAIRS_CAP: usize = 1024;
/// Pairing table entry lifetime (D-15).
pub const PAIRS_TTL: Duration = Duration::from_secs(10 * 60);
/// How long a finished response waits for a request mirror that is still open. The request of a
/// bidi `Run` can outlive the answer by a moment; it cannot hold the record back.
const REQUEST_GRACE: Duration = Duration::from_secs(5);
/// Longest conversation id or pairing key believed — the proxy's own session-id bound.
const MAX_ID_LEN: usize = 128;
/// Longest model id believed.
const MAX_MODEL_LEN: usize = 256;
/// Connect frame flag: the payload is compressed with `connect-content-encoding`.
const FLAG_COMPRESSED: u8 = 0x01;
/// Connect frame flag: end of stream (a JSON trailer, `{}` on success).
const FLAG_END_STREAM: u8 = 0x02;
/// The 5-byte payload Cursor opens a stream with — `0a03666f6f`, a field-1 string "foo". A
/// known non-message, not an error.
const INIT_PING: [u8; 5] = [0x0a, 0x03, b'f', b'o', b'o'];

// --- Routes -----------------------------------------------------------------------------------

/// The three Cursor routes the relay decodes (D-12).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum CursorRoute {
    /// `POST /agent.v1.AgentService/Run` — HTTP/2 full duplex.
    Run,
    /// `POST /agent.v1.AgentService/RunSSE` — the HTTP/1.1 fallback's response stream.
    RunSse,
    /// `POST /aiserver.v1.BidiService/BidiAppend` — the fallback's request messages.
    BidiAppend,
}

impl CursorRoute {
    /// The route `path` names, one trailing slash stripped as `WireFormat::resolve` strips it.
    pub fn of(path: &str) -> Option<Self> {
        let path = path
            .strip_suffix('/')
            .filter(|p| !p.is_empty())
            .unwrap_or(path);
        match path {
            "/agent.v1.AgentService/Run" => Some(Self::Run),
            "/agent.v1.AgentService/RunSSE" => Some(Self::RunSse),
            "/aiserver.v1.BidiService/BidiAppend" => Some(Self::BidiAppend),
            _ => None,
        }
    }
}

// --- Protobuf wire walking ----------------------------------------------------------------------

/// One field's value, as far as the wire says. Groups (wire types 3/4) end the walk.
enum Value<'a> {
    Varint(u64),
    Bytes(&'a [u8]),
    Fixed,
}

/// The fields of one message, in wire order. A malformed tail ends the iteration — a walk
/// never fails, it only stops early.
struct Fields<'a> {
    buf: &'a [u8],
    pos: usize,
}

fn fields(buf: &[u8]) -> Fields<'_> {
    Fields { buf, pos: 0 }
}

fn read_varint(buf: &[u8], pos: &mut usize) -> Option<u64> {
    let mut out = 0u64;
    for shift in (0..64).step_by(7) {
        let b = *buf.get(*pos)?;
        *pos += 1;
        out |= u64::from(b & 0x7f) << shift;
        if b & 0x80 == 0 {
            return Some(out);
        }
    }
    None
}

impl<'a> Iterator for Fields<'a> {
    type Item = (u32, Value<'a>);

    fn next(&mut self) -> Option<Self::Item> {
        if self.pos >= self.buf.len() {
            return None;
        }
        let key = read_varint(self.buf, &mut self.pos)?;
        let number = u32::try_from(key >> 3).ok()?;
        let value = match key & 0x07 {
            0 => Value::Varint(read_varint(self.buf, &mut self.pos)?),
            1 => {
                self.pos = self.pos.checked_add(8).filter(|&p| p <= self.buf.len())?;
                Value::Fixed
            }
            2 => {
                let len = usize::try_from(read_varint(self.buf, &mut self.pos)?).ok()?;
                let end = self.pos.checked_add(len).filter(|&e| e <= self.buf.len())?;
                let bytes = &self.buf[self.pos..end];
                self.pos = end;
                Value::Bytes(bytes)
            }
            5 => {
                self.pos = self.pos.checked_add(4).filter(|&p| p <= self.buf.len())?;
                Value::Fixed
            }
            _ => {
                self.pos = self.buf.len();
                return None;
            }
        };
        Some((number, value))
    }
}

/// The last length-delimited occurrence of `number` (proto3: the last one wins).
fn bytes_field(msg: &[u8], number: u32) -> Option<&[u8]> {
    fields(msg)
        .filter_map(|(n, v)| match v {
            Value::Bytes(b) if n == number => Some(b),
            _ => None,
        })
        .last()
}

/// A non-empty UTF-8 string field no longer than `max`.
fn string_field(msg: &[u8], number: u32, max: usize) -> Option<String> {
    let s = std::str::from_utf8(bytes_field(msg, number)?).ok()?.trim();
    (!s.is_empty() && s.len() <= max).then(|| s.to_string())
}

// --- Messages -----------------------------------------------------------------------------------

/// What a run request says about itself: the two facts the record carries, nothing else.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct RunFacts {
    pub conversation_id: Option<String>,
    pub model: Option<String>,
}

/// `AgentClientMessage` → its `run_request`'s conversation id and model id, or `None` when the
/// message is another member of the oneof (a heartbeat, a tool result, …).
pub fn run_facts(agent_client_message: &[u8]) -> Option<RunFacts> {
    let run = bytes_field(agent_client_message, field::CLIENT_RUN_REQUEST)?;
    let model = bytes_field(run, field::RUN_MODEL_DETAILS)
        .and_then(|d| string_field(d, field::MODEL_DETAILS_MODEL_ID, MAX_MODEL_LEN))
        .or_else(|| {
            bytes_field(run, field::RUN_REQUESTED_MODEL)
                .and_then(|r| string_field(r, field::REQUESTED_MODEL_MODEL_ID, MAX_MODEL_LEN))
        });
    Some(RunFacts {
        conversation_id: string_field(run, field::RUN_CONVERSATION_ID, MAX_ID_LEN),
        model,
    })
}

/// `BidiRequestId` → its request id: `RunSSE`'s request body, and the key a `BidiAppend`
/// carries.
pub fn bidi_request_id(msg: &[u8]) -> Option<String> {
    string_field(msg, field::BIDI_REQUEST_ID, MAX_ID_LEN)
}

/// `BidiAppendRequest` → `(pairing key, the run facts of its hex-encoded AgentClientMessage)`.
/// The facts are `None` for an append that carries anything but a run request.
pub fn bidi_append(msg: &[u8]) -> Option<(String, Option<RunFacts>)> {
    let key = bidi_request_id(bytes_field(msg, field::APPEND_REQUEST_ID)?)?;
    let facts = bytes_field(msg, field::APPEND_DATA)
        .and_then(|hex_data| hex::decode(hex_data).ok())
        .and_then(|client_message| run_facts(&client_message));
    Some((key, facts))
}

/// `AgentServerMessage` → the `tokenDelta` it carries, when it is one.
pub fn token_delta(agent_server_message: &[u8]) -> Option<u64> {
    let update = bytes_field(agent_server_message, field::SERVER_INTERACTION_UPDATE)?;
    let delta = bytes_field(update, field::UPDATE_TOKEN_DELTA)?;
    fields(delta)
        .filter_map(|(n, v)| match v {
            // int32 on the wire: a negative value is a 10-byte varint. Not a count.
            Value::Varint(t) if n == field::TOKEN_DELTA_TOKENS => u64::try_from(t as i64).ok(),
            _ => None,
        })
        .last()
        // proto3 omits a zero; a token delta with no field is a delta of zero.
        .or(Some(0))
}

// --- HTTP Content-Encoding (D-14 step 1) ----------------------------------------------------------

/// Undo the stream's HTTP `Content-Encoding` on the mirror. `gzip` and `deflate` only — any
/// other encoding decodes nothing, and the stream's record says so (`undecoded`, raw frames).
enum HttpLayer {
    Identity,
    Gzip(Box<flate2::write::GzDecoder<Vec<u8>>>),
    Deflate(Box<flate2::write::ZlibDecoder<Vec<u8>>>),
    /// `br`, `zstd`, anything unknown, or a gzip stream that failed mid-way.
    Undecodable,
}

/// Compressed input is fed in slices this size, so one chunk never inflates into more memory
/// than a slice's worst-case ratio allows.
const INFLATE_SLICE: usize = 4096;

impl HttpLayer {
    fn new(content_encoding: Option<&str>) -> Self {
        match content_encoding {
            None | Some("identity") => Self::Identity,
            Some("gzip") | Some("x-gzip") => {
                Self::Gzip(Box::new(flate2::write::GzDecoder::new(Vec::new())))
            }
            Some("deflate") => Self::Deflate(Box::new(flate2::write::ZlibDecoder::new(Vec::new()))),
            Some(_) => Self::Undecodable,
        }
    }

    fn is_undecodable(&self) -> bool {
        matches!(self, Self::Undecodable)
    }

    /// Feed one received chunk; `out` sees the decoded bytes in order. Returns `true` when this
    /// chunk turned the layer `Undecodable` — the stream failed mid-way.
    fn push(&mut self, chunk: &[u8], out: &mut dyn FnMut(&[u8])) -> bool {
        fn inflate<W: Write>(
            w: &mut W,
            drained: impl Fn(&mut W) -> Vec<u8>,
            chunk: &[u8],
            out: &mut dyn FnMut(&[u8]),
        ) -> bool {
            for slice in chunk.chunks(INFLATE_SLICE) {
                // `write` hands the PREVIOUS call's output on; `flush` drains this one.
                if w.write_all(slice).is_err() || w.flush().is_err() {
                    return false;
                }
                let decoded = drained(w);
                if !decoded.is_empty() {
                    out(&decoded);
                }
            }
            true
        }
        let ok = match self {
            Self::Identity => {
                out(chunk);
                true
            }
            Self::Gzip(d) => inflate(d.as_mut(), |d| std::mem::take(d.get_mut()), chunk, out),
            Self::Deflate(d) => inflate(d.as_mut(), |d| std::mem::take(d.get_mut()), chunk, out),
            Self::Undecodable => true,
        };
        if !ok {
            *self = Self::Undecodable;
        }
        !ok
    }
}

/// The encodings a request or response head declares.
#[derive(Clone, Debug, Default)]
struct Encodings {
    /// HTTP `Content-Encoding`, lower-cased.
    content: Option<String>,
    /// Connect's per-message `connect-content-encoding`, lower-cased.
    connect: Option<String>,
}

impl Encodings {
    fn of(headers: &HeaderMap) -> Self {
        let get = |name: &str| {
            headers
                .get(name)
                .and_then(|v| v.to_str().ok())
                .map(|v| v.trim().to_ascii_lowercase())
                .filter(|v| !v.is_empty())
        };
        Self {
            content: get("content-encoding"),
            connect: get("connect-content-encoding"),
        }
    }
}

/// Gunzip one compressed frame, never past [`MAX_FRAME`].
fn gunzip_capped(payload: &[u8]) -> Option<Vec<u8>> {
    let mut out = Vec::new();
    flate2::read::GzDecoder::new(payload)
        .take(MAX_FRAME as u64 + 1)
        .read_to_end(&mut out)
        .ok()?;
    (out.len() <= MAX_FRAME).then_some(out)
}

// --- Connect framing (D-14 steps 2–3) -------------------------------------------------------------

/// Splits `[flags:u8][len:u32 BE][payload]` frames out of a byte stream, whatever the chunk
/// boundaries. Buffers only the current frame, and skips one over [`MAX_FRAME`].
#[derive(Default)]
struct FrameReader {
    head: [u8; 5],
    head_len: usize,
    flags: u8,
    want: usize,
    in_payload: bool,
    payload: Vec<u8>,
    skip: usize,
}

impl FrameReader {
    /// Feed bytes; `frame` sees each complete frame and returns `false` to stop reading. Sets
    /// `oversize` for a skipped frame. Returns `false` once stopped.
    fn push(
        &mut self,
        mut data: &[u8],
        oversize: &mut bool,
        frame: &mut dyn FnMut(u8, &[u8]) -> bool,
    ) -> bool {
        while !data.is_empty() {
            if self.skip > 0 {
                let n = self.skip.min(data.len());
                self.skip -= n;
                data = &data[n..];
                continue;
            }
            if !self.in_payload {
                let n = (5 - self.head_len).min(data.len());
                self.head[self.head_len..self.head_len + n].copy_from_slice(&data[..n]);
                self.head_len += n;
                data = &data[n..];
                if self.head_len < 5 {
                    break;
                }
                self.head_len = 0;
                self.flags = self.head[0];
                let len =
                    u32::from_be_bytes([self.head[1], self.head[2], self.head[3], self.head[4]])
                        as usize;
                if len > MAX_FRAME {
                    *oversize = true;
                    self.skip = len;
                    continue;
                }
                self.want = len;
                self.in_payload = true;
                self.payload.clear();
            }
            // Falls through with `data` possibly empty, so a zero-length frame completes here.
            let n = (self.want - self.payload.len()).min(data.len());
            self.payload.extend_from_slice(&data[..n]);
            data = &data[n..];
            if self.payload.len() == self.want {
                self.in_payload = false;
                if !frame(self.flags, &self.payload) {
                    return false;
                }
            }
        }
        true
    }
}

/// One direction of a Connect stream: HTTP decoding, then framing, then per-message
/// decompression. `message` sees each message payload; nothing is kept.
struct StreamDecoder {
    http: HttpLayer,
    frames: FrameReader,
    connect_gzip: bool,
    connect_other: bool,
    /// A frame flag `0x02` arrived.
    ended: bool,
    /// The end-of-stream frame's JSON carried an `error`: Connect's streaming error, sent under
    /// HTTP 200.
    end_error: bool,
    /// A frame was skipped (oversize) or could not be decompressed, or the HTTP layer failed
    /// mid-stream.
    lossy: bool,
    stopped: bool,
}

impl StreamDecoder {
    fn new(enc: &Encodings) -> Self {
        let connect = enc.connect.as_deref();
        Self {
            http: HttpLayer::new(enc.content.as_deref()),
            frames: FrameReader::default(),
            connect_gzip: matches!(connect, Some("gzip")),
            connect_other: !matches!(connect, None | Some("identity") | Some("gzip")),
            ended: false,
            end_error: false,
            lossy: false,
            stopped: false,
        }
    }

    fn push(&mut self, chunk: &[u8], message: &mut dyn FnMut(&[u8])) {
        if self.stopped || self.http.is_undecodable() {
            return;
        }
        let Self {
            http,
            frames,
            connect_gzip,
            connect_other,
            ended,
            end_error,
            lossy,
            stopped,
        } = self;
        let mut oversize = false;
        let failed = http.push(chunk, &mut |decoded| {
            if *stopped {
                return;
            }
            let go_on = frames.push(decoded, &mut oversize, &mut |flags, payload| {
                if flags & FLAG_END_STREAM != 0 {
                    *ended = true;
                    *end_error = if flags & FLAG_COMPRESSED == 0 {
                        end_stream_error(payload)
                    } else {
                        *connect_gzip
                            && !*connect_other
                            && gunzip_capped(payload).is_some_and(|m| end_stream_error(&m))
                    };
                    return false;
                }
                if payload == INIT_PING {
                    return true;
                }
                if flags & FLAG_COMPRESSED == 0 {
                    message(payload);
                } else if *connect_gzip && !*connect_other {
                    match gunzip_capped(payload) {
                        Some(m) => message(&m),
                        None => *lossy = true,
                    }
                } else {
                    // A per-message codec other than gzip: this FRAME is undecoded.
                    *lossy = true;
                }
                true
            });
            if !go_on {
                *stopped = true;
            }
        });
        if oversize || failed {
            self.lossy = true;
        }
    }
}

/// Whether an end-of-stream frame's JSON (`{}` on success) carries an `error` object.
fn end_stream_error(payload: &[u8]) -> bool {
    serde_json::from_slice::<serde_json::Value>(payload)
        .is_ok_and(|v| v.get("error").is_some_and(|e| !e.is_null()))
}

/// A unary body (`BidiAppend`, `RunSSE`'s request) after HTTP decoding, never past
/// [`MAX_FRAME`]. `None` when it was over the cap or undecodable.
struct UnaryCollector {
    http: HttpLayer,
    body: Vec<u8>,
    over: bool,
}

impl UnaryCollector {
    fn new(enc: &Encodings) -> Self {
        Self {
            http: HttpLayer::new(enc.content.as_deref()),
            body: Vec::new(),
            over: false,
        }
    }

    fn push(&mut self, chunk: &[u8]) {
        let Self { http, body, over } = self;
        // A mid-stream failure leaves the layer `Undecodable`, which `finish` reads.
        http.push(chunk, &mut |decoded| {
            if *over || body.len() + decoded.len() > MAX_FRAME {
                *over = true;
                body.clear();
            } else {
                body.extend_from_slice(decoded);
            }
        });
    }

    /// The message: the body itself for unary `application/proto`, or its first frame's
    /// payload when the body is enveloped (`application/connect+proto`). A raw message opens
    /// with a field key (`0x0a` for these messages), which is never a valid flags byte.
    fn finish(self, enc: &Encodings) -> Option<Vec<u8>> {
        if self.over || self.http.is_undecodable() {
            return None;
        }
        let b = self.body;
        if b.len() >= 5 && b[0] & !(FLAG_COMPRESSED | FLAG_END_STREAM) == 0 {
            let len = u32::from_be_bytes([b[1], b[2], b[3], b[4]]) as usize;
            let payload = b.get(5..5 + len)?;
            return if b[0] & FLAG_COMPRESSED == 0 {
                Some(payload.to_vec())
            } else if enc.connect.as_deref() == Some("gzip") {
                gunzip_capped(payload)
            } else {
                None
            };
        }
        Some(b)
    }
}

// --- The mirror (D-13) ----------------------------------------------------------------------------

/// The forward path's end of a mirror. [`Mirror::offer`] never waits: over [`MIRROR_ITEMS`]
/// or [`MIRROR_BYTES`] the mirror is abandoned and the tap marked lossy (`partial`).
pub(crate) struct Mirror {
    tx: Option<mpsc::Sender<Bytes>>,
    queued: Arc<AtomicUsize>,
    lossy: Arc<AtomicBool>,
}

/// The decoder's end of a mirror.
pub(crate) struct MirrorRx {
    rx: mpsc::Receiver<Bytes>,
    queued: Arc<AtomicUsize>,
}

fn mirror(lossy: Arc<AtomicBool>) -> (Mirror, MirrorRx) {
    let (tx, rx) = mpsc::channel(MIRROR_ITEMS);
    let queued = Arc::new(AtomicUsize::new(0));
    (
        Mirror {
            tx: Some(tx),
            queued: queued.clone(),
            lossy,
        },
        MirrorRx { rx, queued },
    )
}

impl Mirror {
    /// Offer one forwarded chunk. The forward path never awaits the decoder.
    fn offer(&mut self, chunk: &Bytes) {
        let Some(tx) = self.tx.as_ref() else {
            return;
        };
        let n = chunk.len();
        // Counted before the send, so the receiver's subtraction can never run first.
        if self.queued.fetch_add(n, Ordering::AcqRel) + n > MIRROR_BYTES {
            self.queued.fetch_sub(n, Ordering::AcqRel);
            self.abandon(true);
            return;
        }
        match tx.try_send(chunk.clone()) {
            Ok(()) => {}
            Err(e) => {
                self.queued.fetch_sub(n, Ordering::AcqRel);
                // Full: the decoder fell behind — drop the mirror, keep the stream. Closed: the
                // decoder has what it needs from this side and stopped reading; nothing is lost.
                self.abandon(matches!(e, mpsc::error::TrySendError::Full(_)));
            }
        }
    }

    fn abandon(&mut self, lossy: bool) {
        self.tx = None;
        if lossy {
            self.lossy.store(true, Ordering::Relaxed);
        }
    }
}

impl MirrorRx {
    async fn recv(&mut self) -> Option<Bytes> {
        let chunk = self.rx.recv().await?;
        self.queued.fetch_sub(chunk.len(), Ordering::AcqRel);
        Some(chunk)
    }
}

/// A forwarded stream with a mirror attached: every `Ok` chunk is yielded unchanged **and**
/// offered to the mirror. End or error closes the mirror. With no mirror it is the stream.
pub(crate) struct Teed<S> {
    inner: Pin<Box<S>>,
    mirror: Option<Mirror>,
}

impl<S> Teed<S> {
    fn new(inner: S, mirror: Option<Mirror>) -> Self {
        Self {
            inner: Box::pin(inner),
            mirror,
        }
    }
}

impl<S, E> Stream for Teed<S>
where
    S: Stream<Item = Result<Bytes, E>>,
{
    type Item = Result<Bytes, E>;

    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))) => {
                if let Some(m) = this.mirror.as_mut() {
                    m.offer(chunk);
                }
            }
            Poll::Ready(None) | Poll::Ready(Some(Err(_))) => this.mirror = None,
            Poll::Pending => {}
        }
        polled
    }
}

// --- Pairing (D-15) ---------------------------------------------------------------------------------

/// `request_id` → the run facts a `BidiAppend` carried, for the `RunSSE` stream with the same
/// id. Bounded: [`PAIRS_CAP`] entries, [`PAIRS_TTL`] each, least-recently-used evicted first.
///
/// **Best-effort.** The key's two sides — `RunSSE`'s `BidiRequestId.request_id` and
/// `BidiAppendRequest.request_id.request_id` — are the recovered protos' shape; no captured
/// Cursor sample was available to confirm the values match on the wire. A miss costs the
/// record its model and conversation id (`partial`), nothing more.
#[derive(Default)]
pub struct PairingTable {
    inner: Mutex<HashMap<String, PairEntry>>,
}

struct PairEntry {
    facts: RunFacts,
    inserted: Instant,
    used: Instant,
}

impl PairingTable {
    pub fn insert(&self, key: String, facts: RunFacts) {
        self.insert_at(key, facts, Instant::now());
    }

    pub fn get(&self, key: &str) -> Option<RunFacts> {
        self.get_at(key, Instant::now())
    }

    fn insert_at(&self, key: String, facts: RunFacts, now: Instant) {
        let mut map = self.inner.lock().unwrap_or_else(|e| e.into_inner());
        map.retain(|_, e| now.duration_since(e.inserted) < PAIRS_TTL);
        if !map.contains_key(&key) && map.len() >= PAIRS_CAP {
            if let Some(lru) = map
                .iter()
                .min_by_key(|(_, e)| e.used)
                .map(|(k, _)| k.clone())
            {
                map.remove(&lru);
            }
        }
        map.insert(
            key,
            PairEntry {
                facts,
                inserted: now,
                used: now,
            },
        );
    }

    fn get_at(&self, key: &str, now: Instant) -> Option<RunFacts> {
        let mut map = self.inner.lock().unwrap_or_else(|e| e.into_inner());
        let entry = map.get_mut(key)?;
        if now.duration_since(entry.inserted) >= PAIRS_TTL {
            map.remove(key);
            return None;
        }
        entry.used = now;
        Some(entry.facts.clone())
    }
}

// --- The decoder --------------------------------------------------------------------------------

/// What one `Run`/`RunSSE` stream decoded to. `Default` is the panicked decoder: nothing.
#[derive(Debug, Default)]
pub(crate) struct Decoded {
    /// `None`: no response head arrived (the upstream failed) or the decoder panicked.
    status_ok: Option<bool>,
    model: Option<String>,
    conversation_id: Option<String>,
    output_tokens: Option<u64>,
    /// A frame was skipped or undecompressable.
    lossy: bool,
    /// The response carried its end-of-stream frame.
    ended: bool,
}

/// Response bytes as received, at most [`RAW_CAP`] (D-17). Shared between the decoder, which
/// fills it, and [`Finish`], which keeps it: a decoder panic unwinds the task, not this.
pub(crate) type RawBuf = Arc<Mutex<Vec<u8>>>;

impl Decoded {
    /// `capture.decode`: all three facts, with no frame skipped and nothing dropped by the
    /// mirror, is `decoded`; none of them is `undecoded` — unless the mirror itself was
    /// abandoned, where the copy is incomplete and `partial` is all that can be said; anything
    /// else is `partial`. A skipped frame on a stream that yielded nothing stays `undecoded`, so
    /// its raw frames are kept: that is exactly the drift D-17 keeps them for.
    fn outcome(&self, mirror_lost: bool) -> CaptureDecode {
        let have = [
            self.model.is_some(),
            self.conversation_id.is_some(),
            self.output_tokens.is_some(),
        ];
        if have.iter().all(|h| *h) && !self.lossy && !mirror_lost {
            CaptureDecode::Decoded
        } else if !have.iter().any(|h| *h) && !mirror_lost {
            CaptureDecode::Undecoded
        } else {
            CaptureDecode::Partial
        }
    }
}

/// The response head and its mirror, handed to the decoder when `relay` has them.
pub(crate) struct ResponseSide {
    status_ok: bool,
    enc: Encodings,
    rx: MirrorRx,
}

/// The first run request on a `Run` request stream. Returns as soon as it is found, dropping
/// the mirror — the rest of the request is tool traffic this module never reads.
async fn decode_run_request(mut rx: MirrorRx, enc: Encodings) -> (Option<RunFacts>, bool) {
    let mut d = StreamDecoder::new(&enc);
    let mut found = None;
    while let Some(chunk) = rx.recv().await {
        d.push(&chunk, &mut |msg| {
            if found.is_none() {
                found = run_facts(msg);
            }
        });
        if found.is_some() || d.stopped {
            break;
        }
    }
    (found, d.lossy)
}

/// A unary request body, whole.
async fn collect_unary(mut rx: MirrorRx, enc: Encodings) -> Option<Vec<u8>> {
    let mut c = UnaryCollector::new(&enc);
    while let Some(chunk) = rx.recv().await {
        c.push(&chunk);
    }
    c.finish(&enc)
}

/// Token deltas from a response stream, plus the raw bytes for D-17 into `raw`.
async fn decode_response(side: oneshot::Receiver<ResponseSide>, raw: RawBuf) -> Decoded {
    let Ok(ResponseSide {
        status_ok,
        enc,
        mut rx,
    }) = side.await
    else {
        return Decoded::default();
    };
    let mut d = StreamDecoder::new(&enc);
    let mut tokens: Option<u64> = None;
    while let Some(chunk) = rx.recv().await {
        {
            let mut raw = raw.lock().unwrap_or_else(|e| e.into_inner());
            if raw.len() < RAW_CAP {
                let n = (RAW_CAP - raw.len()).min(chunk.len());
                raw.extend_from_slice(&chunk[..n]);
            }
        }
        d.push(&chunk, &mut |msg| {
            if let Some(t) = token_delta(msg) {
                tokens = Some(tokens.unwrap_or(0).saturating_add(t));
            }
        });
    }
    Decoded {
        // Connect streaming reports a provider error under HTTP 200, in the end-of-stream frame.
        status_ok: Some(status_ok && !d.end_error),
        output_tokens: tokens,
        lossy: d.lossy,
        ended: d.ended,
        ..Decoded::default()
    }
}

/// What a request mirror yielded: `Run`'s facts directly, or `RunSSE`'s pairing key.
enum RequestOut {
    Facts(Option<RunFacts>, bool),
    Key(Option<String>),
}

/// One `Run`/`RunSSE` stream, both directions at once. The request side is abandoned
/// [`REQUEST_GRACE`] after the response ends.
async fn decode_turn(
    route: CursorRoute,
    req_rx: MirrorRx,
    req_enc: Encodings,
    resp: oneshot::Receiver<ResponseSide>,
    raw: RawBuf,
    pairs: Arc<PairingTable>,
) -> Decoded {
    let (done_tx, done_rx) = oneshot::channel::<()>();
    let request = async move {
        let req = async move {
            match route {
                CursorRoute::Run => {
                    let (facts, lossy) = decode_run_request(req_rx, req_enc).await;
                    RequestOut::Facts(facts, lossy)
                }
                _ => RequestOut::Key(
                    collect_unary(req_rx, req_enc)
                        .await
                        .and_then(|body| bidi_request_id(&body)),
                ),
            }
        };
        tokio::select! {
            out = req => Some(out),
            _ = async {
                let _ = done_rx.await;
                tokio::time::sleep(REQUEST_GRACE).await;
            } => None,
        }
    };
    let response = async move {
        let out = decode_response(resp, raw).await;
        let _ = done_tx.send(());
        out
    };
    let (req, mut out) = tokio::join!(request, response);
    let (facts, req_lossy) = match req {
        Some(RequestOut::Facts(facts, lossy)) => (facts, lossy),
        // Looked up at the END of the stream: the append carrying the run request is sent
        // after `RunSSE` opens.
        Some(RequestOut::Key(key)) => (key.and_then(|k| pairs.get(&k)), false),
        None => (None, false),
    };
    let facts = facts.unwrap_or_default();
    out.model = facts.model;
    out.conversation_id = facts.conversation_id;
    out.lossy |= req_lossy;
    out
}

/// A `BidiAppend` request: into the pairing table when it carries a run request.
async fn decode_append(rx: MirrorRx, enc: Encodings, pairs: Arc<PairingTable>) {
    let Some(body) = collect_unary(rx, enc).await else {
        return;
    };
    if let Some((key, Some(facts))) = bidi_append(&body) {
        pairs.insert(key, facts);
    }
}

/// Run `decode` as its own task and observe its handle: a panic inside it is
/// `JoinError::is_panic`, which `catch_unwind` around building the future would never see.
async fn supervise<F>(decode: F) -> Option<Decoded>
where
    F: std::future::Future<Output = Decoded> + Send + 'static,
{
    match tokio::spawn(decode).await {
        Ok(d) => Some(d),
        Err(e) => {
            if e.is_panic() {
                super::proxy::record_pass_through_failure("cursor_decode_panic");
            }
            None
        }
    }
}

// --- The record (D-16, D-18) ---------------------------------------------------------------------

/// Cursor-native models — Cursor's own and the ones it serves under its own name — have no
/// public per-token price: `unknown_model`, no estimate (D-18).
fn is_cursor_native(model: &str) -> bool {
    let m = model.trim().to_ascii_lowercase();
    m == "default" || m.starts_with("composer") || m.starts_with("grok") || m.starts_with("cursor-")
}

/// Whether the record may call its model known: a public model the tokenizer table carries.
fn model_known(model: &str) -> bool {
    !is_cursor_native(model) && classify_model(model).is_some()
}

/// `capture.gap`, from what the decoder saw — the existing values, most specific first.
fn capture_gap(
    d: &Decoded,
    panicked: bool,
    outcome: CaptureDecode,
    mirror_lost: bool,
) -> Option<CaptureGap> {
    if panicked {
        return Some(CaptureGap::UnknownWireFormat);
    }
    if d.status_ok != Some(true) {
        return Some(CaptureGap::ProviderError);
    }
    if outcome == CaptureDecode::Undecoded {
        return Some(CaptureGap::UnknownWireFormat);
    }
    if d.model.as_deref().is_some_and(|m| !model_known(m)) {
        return Some(CaptureGap::UnknownModel);
    }
    if !d.ended && !mirror_lost {
        return Some(CaptureGap::StreamInterrupted);
    }
    None
}

/// Everything the decoder task needs to emit, captured when the request arrived.
pub(crate) struct Finish {
    /// The request's metadata-only observation: event id, time, install, billing, attribution.
    pub(crate) obs: Observation,
    pub(crate) scope: Option<&'static str>,
    pub(crate) registry: Arc<SessionRegistry>,
    pub(crate) privacy: PrivacyFilter,
    pub(crate) cloud_tx: Option<mpsc::Sender<CloudEvent>>,
    pub(crate) raw_dir: PathBuf,
    /// The response's raw bytes, filled by the decoder and outliving it.
    pub(crate) raw: RawBuf,
    pub(crate) mirror_lost: Arc<AtomicBool>,
}

impl Finish {
    /// Emit the stream's one record; keep its raw response frames if it decoded to nothing.
    async fn run(self, decoded: Option<Decoded>) {
        let panicked = decoded.is_none();
        let d = decoded.unwrap_or_default();
        let mirror_lost = self.mirror_lost.load(Ordering::Relaxed);
        let outcome = if panicked {
            CaptureDecode::Undecoded
        } else {
            d.outcome(mirror_lost)
        };
        let gap = capture_gap(&d, panicked, outcome, mirror_lost);
        let mut obs = self.obs;
        obs.model_known = d.model.as_deref().is_some_and(model_known);
        obs.model = d.model;
        // The conversation id is the agent's own claim about its session (Attested even when
        // the hook has not registered it yet); absent, the request-time cascade stands.
        if let Some(conversation) = d.conversation_id {
            obs.session = session::resolve_session_scoped(
                &self.registry,
                &obs.install_id,
                &session::RequestSignals {
                    declared_session_id: Some(conversation),
                    ..Default::default()
                },
                self.scope,
            );
        }
        // The frames land before the record, so whoever reads the record finds them. A provider
        // error is the provider answering, not decoder drift: its frames are not kept (D-17).
        let raw = std::mem::take(&mut *self.raw.lock().unwrap_or_else(|e| e.into_inner()));
        if outcome == CaptureDecode::Undecoded
            && gap != Some(CaptureGap::ProviderError)
            && !raw.is_empty()
        {
            let (dir, id) = (self.raw_dir, obs.event_id.clone());
            let _ = tokio::task::spawn_blocking(move || write_raw(&dir, &id, &raw)).await;
        }
        emit::build_and_emit_decoded(
            &obs,
            d.output_tokens,
            outcome,
            gap,
            &self.privacy,
            self.cloud_tx.as_ref(),
        );
    }
}

// --- Raw frames on failure (D-17) ------------------------------------------------------------------

/// `$OPENLATCH_DIR/model_relay/cursor-raw/` — beside the findings store, same `model_relay`
/// spelling.
pub fn raw_dir() -> PathBuf {
    crate::config::openlatch_dir()
        .join("model_relay")
        .join("cursor-raw")
}

/// Serialises [`write_raw`]'s write-and-prune, so concurrent writers never each prune from a
/// stale listing. Held only on a blocking thread, never across an `.await`.
static RAW_WRITE: Mutex<()> = Mutex::new(());

/// Keep `raw` (response frames, as received) as `<event_id>.bin`, owner-only and at most
/// [`RAW_CAP`], then prune the directory. Runs on a blocking thread.
fn write_raw(dir: &Path, event_id: &str, raw: &[u8]) -> std::io::Result<PathBuf> {
    let name = super::retention::sanitize(event_id);
    if name.is_empty() {
        return Err(std::io::Error::new(
            std::io::ErrorKind::InvalidInput,
            "empty event id",
        ));
    }
    let _serial = RAW_WRITE.lock().unwrap_or_else(|e| e.into_inner());
    crate::fs_secure::create_dir_owner_only(dir)?;
    let path = dir.join(format!("{name}.bin"));
    crate::fs_secure::write_owner_only_bytes(&path, &raw[..raw.len().min(RAW_CAP)])?;
    super::retention::prune(dir, "bin", RAW_MAX_FILES);
    Ok(path)
}

// --- The tap: what the proxy holds ----------------------------------------------------------------

/// One Cursor request's mirrors, held by the forward path. Dropping it (an upstream that never
/// answered) closes both, and the decoder still emits — a `provider_error` record.
pub(crate) struct Tap {
    request: Option<Mirror>,
    response: Option<(oneshot::Sender<ResponseSide>, Arc<AtomicBool>)>,
}

impl Tap {
    /// Start the decoder task for one Cursor request (Cursor I-2 D-13), or `None` for our own
    /// probe, without an event sink (nothing would be recorded, so nothing is decoded), or on a
    /// path that is not one of the three routes. `obs` builds the request's observation, and
    /// runs only once the request qualifies; the decoder completes it.
    pub(crate) fn for_request(
        st: &ModelRelayState,
        headers: &HeaderMap,
        path: &str,
        scope: Option<&'static str>,
        obs: impl FnOnce() -> Observation,
    ) -> Option<Self> {
        if super::proxy::is_preflight(headers) || st.cloud_tx.is_none() {
            return None;
        }
        let route = CursorRoute::of(path)?;
        Self::start_with(st, route, headers, obs(), scope, raw_dir())
    }

    /// [`Tap::for_request`] for a known route, keeping raw frames under `raw_dir` — the seam tests use to keep them out
    /// of the real `$OPENLATCH_DIR`.
    fn start_with(
        st: &ModelRelayState,
        route: CursorRoute,
        headers: &HeaderMap,
        obs: Observation,
        scope: Option<&'static str>,
        raw_dir: PathBuf,
    ) -> Option<Self> {
        st.cloud_tx.as_ref()?;
        let lossy = Arc::new(AtomicBool::new(false));
        let (req_mirror, req_rx) = mirror(lossy.clone());
        let req_enc = Encodings::of(headers);
        let pairs = st.cursor_pairs.clone();
        if route == CursorRoute::BidiAppend {
            tokio::spawn(decode_append(req_rx, req_enc, pairs));
            return Some(Self {
                request: Some(req_mirror),
                response: None,
            });
        }
        let (resp_tx, resp_rx) = oneshot::channel();
        let raw = RawBuf::default();
        let finish = Finish {
            obs,
            scope,
            registry: st.registry.clone(),
            privacy: st.privacy.clone(),
            cloud_tx: st.cloud_tx.clone(),
            raw_dir,
            raw: raw.clone(),
            mirror_lost: lossy.clone(),
        };
        tokio::spawn(async move {
            let decoded = supervise(decode_turn(route, req_rx, req_enc, resp_rx, raw, pairs)).await;
            finish.run(decoded).await;
        });
        Some(Self {
            request: Some(req_mirror),
            response: Some((resp_tx, lossy)),
        })
    }

    /// The request body, mirrored.
    pub(crate) fn tee_request<S>(&mut self, body: S) -> Teed<S> {
        Teed::new(body, self.request.take())
    }

    /// The response body, mirrored — and the response head handed to the decoder.
    pub(crate) fn tee_response<S>(
        self,
        status: StatusCode,
        headers: &HeaderMap,
        body: S,
    ) -> Teed<S> {
        let Some((tx, lossy)) = self.response else {
            return Teed::new(body, None);
        };
        let (m, rx) = mirror(lossy);
        let side = ResponseSide {
            status_ok: status.is_success(),
            enc: Encodings::of(headers),
            rx,
        };
        match tx.send(side) {
            Ok(()) => Teed::new(body, Some(m)),
            Err(_) => Teed::new(body, None),
        }
    }
}

#[cfg(test)]
mod tests {
    //! Fixtures under `tests/fixtures/cursor/relay/` are **synthetic**: no captured Cursor wire
    //! sample was available, so each is built from the recovered protos' documented shape (the
    //! [`field`] numbers) with every content field replaced by an `OL-SENTINEL-*` marker. The
    //! pairing test therefore proves both sides parse to the same key under that shape — not that
    //! Cursor's live traffic has it, which is why the pairing is best-effort.

    use super::*;
    use crate::model_relay::wire_format::WireFormat;
    use futures_util::StreamExt;
    use std::time::SystemTime;

    const APPEND: &[u8] =
        include_bytes!("../../tests/fixtures/cursor/relay/synthetic-bidi-append-request.bin");
    const RUNSSE_REQUEST: &[u8] =
        include_bytes!("../../tests/fixtures/cursor/relay/synthetic-runsse-request.bin");
    const RUN_REQUEST: &[u8] =
        include_bytes!("../../tests/fixtures/cursor/relay/synthetic-run-request.bin");
    const RUN_RESPONSE: &[u8] =
        include_bytes!("../../tests/fixtures/cursor/relay/synthetic-run-response.bin");

    const CONVERSATION: &str = "3f0c2a9e-0000-4000-8000-00000000c0de";
    const PAIR_KEY: &str = "7b1d3c55-0000-4000-8000-0000000b1d1a";

    // --- encoders (tests only) ---

    fn varint(mut n: u64) -> Vec<u8> {
        let mut out = Vec::new();
        loop {
            let b = (n & 0x7f) as u8;
            n >>= 7;
            if n == 0 {
                out.push(b);
                return out;
            }
            out.push(b | 0x80);
        }
    }

    fn ld(number: u32, data: &[u8]) -> Vec<u8> {
        let mut out = varint(u64::from(number) << 3 | 2);
        out.extend(varint(data.len() as u64));
        out.extend_from_slice(data);
        out
    }

    fn vi(number: u32, v: u64) -> Vec<u8> {
        let mut out = varint(u64::from(number) << 3);
        out.extend(varint(v));
        out
    }

    fn frame(flags: u8, payload: &[u8]) -> Vec<u8> {
        let mut out = vec![flags];
        out.extend((payload.len() as u32).to_be_bytes());
        out.extend_from_slice(payload);
        out
    }

    fn token_msg(n: u64) -> Vec<u8> {
        ld(1, &ld(8, &vi(1, n)))
    }

    fn gzip(data: &[u8]) -> Vec<u8> {
        let mut e = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
        e.write_all(data).unwrap();
        e.finish().unwrap()
    }

    fn enc(content: Option<&str>, connect: Option<&str>) -> Encodings {
        Encodings {
            content: content.map(str::to_string),
            connect: connect.map(str::to_string),
        }
    }

    /// Every message a direction yields, and the decoder's end state.
    fn messages(chunks: &[&[u8]], e: &Encodings) -> (Vec<Vec<u8>>, StreamDecoder) {
        let mut d = StreamDecoder::new(e);
        let mut out = Vec::new();
        for c in chunks {
            d.push(c, &mut |m| out.push(m.to_vec()));
        }
        (out, d)
    }

    fn tokens_of(msgs: &[Vec<u8>]) -> Option<u64> {
        msgs.iter()
            .filter_map(|m| token_delta(m))
            .fold(None, |acc, t| Some(acc.unwrap_or(0) + t))
    }

    // --- framing ---

    #[test]
    fn frames_split_across_chunk_boundaries() {
        let whole = messages(&[RUN_RESPONSE], &Encodings::default());
        for size in [1, 2, 3, 5, 7, 64] {
            let chunks: Vec<&[u8]> = RUN_RESPONSE.chunks(size).collect();
            let (msgs, d) = messages(&chunks, &Encodings::default());
            assert_eq!(msgs, whole.0, "chunk size {size}");
            assert_eq!(tokens_of(&msgs), Some(42), "chunk size {size}");
            assert!(d.ended && !d.lossy, "chunk size {size}");
        }
    }

    #[test]
    fn gzip_frame_decodes() {
        // Per message: flag 0x01 with `connect-content-encoding: gzip`.
        let stream = [
            frame(FLAG_COMPRESSED, &gzip(&token_msg(7))),
            frame(FLAG_END_STREAM, b"{}"),
        ]
        .concat();
        let (msgs, d) = messages(&[&stream], &enc(None, Some("gzip")));
        assert_eq!(tokens_of(&msgs), Some(7));
        assert!(!d.lossy);

        // Whole stream: HTTP `Content-Encoding: gzip` on top of plain frames.
        let (msgs, d) = messages(&[&gzip(RUN_RESPONSE)], &enc(Some("gzip"), None));
        assert_eq!(tokens_of(&msgs), Some(42));
        assert!(d.ended);

        // A compressed frame whose codec is not gzip is that FRAME undecoded, and the record lossy.
        let (msgs, d) = messages(&[&stream], &enc(None, Some("zstd")));
        assert!(msgs.is_empty());
        assert!(d.lossy);
    }

    #[test]
    fn end_stream_stops() {
        let stream = [
            frame(0, &token_msg(5)),
            frame(FLAG_END_STREAM, b"{}"),
            frame(0, &token_msg(1000)),
        ]
        .concat();
        let (msgs, d) = messages(&[&stream], &Encodings::default());
        assert_eq!(
            tokens_of(&msgs),
            Some(5),
            "nothing after end-of-stream is read"
        );
        assert!(d.ended);
    }

    #[test]
    fn init_ping_is_ignored() {
        let stream = [frame(0, &INIT_PING), frame(0, &token_msg(3))].concat();
        let (msgs, d) = messages(&[&stream], &Encodings::default());
        assert_eq!(msgs, vec![token_msg(3)], "the ping is not a message");
        assert!(!d.lossy, "and not an error");
    }

    #[test]
    fn an_oversize_frame_is_skipped_and_marks_partial() {
        let mut stream = vec![0u8];
        stream.extend(((MAX_FRAME + 1) as u32).to_be_bytes());
        stream.extend(std::iter::repeat_n(0u8, MAX_FRAME + 1));
        stream.extend(frame(0, &token_msg(9)));
        let chunks: Vec<&[u8]> = stream.chunks(64 * 1024).collect();
        let (msgs, d) = messages(&chunks, &Encodings::default());
        assert_eq!(tokens_of(&msgs), Some(9), "the next frame header is found");
        assert!(d.lossy);
    }

    #[test]
    fn unknown_tags_do_not_fail_the_record() {
        // A server message with only an unknown tag, an update with an unknown member, a varint
        // where a message was expected, and trailing garbage.
        let drift = [
            frame(0, &ld(99, b"\x08\x01")),
            frame(0, &ld(1, &ld(42, b"drift"))),
            frame(0, &vi(1, 5)),
            frame(0, &[0xff, 0xff, 0xff]),
            frame(0, &token_msg(11)),
        ]
        .concat();
        let (msgs, d) = messages(&[&drift], &Encodings::default());
        assert_eq!(tokens_of(&msgs), Some(11));
        assert!(!d.lossy);
    }

    // --- messages ---

    #[test]
    fn bidi_append_yields_model_and_conversation() {
        let (key, facts) = bidi_append(APPEND).expect("an append");
        assert_eq!(key, PAIR_KEY);
        let facts = facts.expect("a run request");
        assert_eq!(facts.conversation_id.as_deref(), Some(CONVERSATION));
        assert_eq!(facts.model.as_deref(), Some("claude-4.5-sonnet"));
        // The fixture carries a prompt and an API key; neither is read.
        let seen = format!("{facts:?}");
        assert!(!seen.contains("SENTINEL"), "{seen}");
        assert!(
            APPEND
                .windows(8)
                .any(|w| w == b"53454e54".as_slice() || w == b"SENTINEL".as_slice()),
            "premise: the fixture carries content"
        );

        // The model falls back to `requested_model` when `model_details` names none.
        let run = [ld(9, &ld(1, b"gpt-5")), ld(5, b"c1")].concat();
        let facts = run_facts(&ld(1, &run)).expect("a run request");
        assert_eq!(facts.model.as_deref(), Some("gpt-5"));
        // Any other oneof member is not a run request.
        assert_eq!(run_facts(&ld(7, b"")), None);
    }

    #[test]
    fn runsse_and_bidi_append_share_the_pairing_key() {
        let mut c = UnaryCollector::new(&Encodings::default());
        c.push(RUNSSE_REQUEST);
        let body = c.finish(&Encodings::default()).expect("a body");
        let runsse_key = bidi_request_id(&body).expect("RunSSE names its request");
        let (append_key, _) = bidi_append(APPEND).expect("an append");
        assert_eq!(runsse_key, append_key);

        // A raw (unenveloped) `BidiRequestId` reads the same.
        let raw = ld(1, PAIR_KEY.as_bytes());
        let mut c = UnaryCollector::new(&Encodings::default());
        c.push(&raw);
        assert_eq!(
            bidi_request_id(&c.finish(&Encodings::default()).unwrap()).as_deref(),
            Some(PAIR_KEY)
        );
    }

    #[test]
    fn the_pairing_table_is_bounded_by_ttl_and_lru() {
        let t = PairingTable::default();
        let t0 = Instant::now();
        let facts = |m: &str| RunFacts {
            conversation_id: None,
            model: Some(m.to_string()),
        };
        t.insert_at("a".into(), facts("a"), t0);
        assert!(t
            .get_at("a", t0 + PAIRS_TTL - Duration::from_secs(1))
            .is_some());
        assert!(t.get_at("a", t0 + PAIRS_TTL).is_none(), "expired");

        let at = |ms: u64| t0 + Duration::from_millis(ms);
        for i in 0..PAIRS_CAP as u64 {
            t.insert_at(format!("k{i}"), facts("x"), at(i));
        }
        // Touch k0, so k1 is the least recently used.
        assert!(t.get_at("k0", at(5_000)).is_some());
        t.insert_at("new".into(), facts("n"), at(6_000));
        assert!(t.get_at("k1", at(7_000)).is_none(), "LRU evicted");
        assert!(t.get_at("k0", at(7_000)).is_some());
        assert!(t.get_at("new", at(7_000)).is_some());
    }

    // --- the task ---

    fn state() -> (Arc<ModelRelayState>, mpsc::Receiver<CloudEvent>) {
        let (tx, rx) = mpsc::channel(16);
        let st = ModelRelayState::new(
            reqwest::Url::parse("http://127.0.0.1:1").unwrap(),
            0,
            8,
            &[],
        )
        .with_measurement(Arc::new(SessionRegistry::default()), Some(tx));
        (Arc::new(st), rx)
    }

    fn observation() -> Observation {
        let mut o = Observation::none();
        o.measured = true;
        o.event_id = uuid::Uuid::now_v7().to_string();
        o.occurred_at = "2026-09-30T12:00:00Z".to_string();
        o.install_id = "agt_test".to_string();
        o.wire_format = WireFormat::CursorConnectRpc;
        o.attributable_agent = Some("cursor");
        o
    }

    fn chunks(bytes: &[u8], size: usize) -> Vec<Result<Bytes, std::io::Error>> {
        bytes
            .chunks(size)
            .map(|c| Ok(Bytes::copy_from_slice(c)))
            .collect()
    }

    fn headers(pairs: &[(&'static str, &'static str)]) -> HeaderMap {
        let mut h = HeaderMap::new();
        for (k, v) in pairs {
            h.insert(*k, v.parse().unwrap());
        }
        h
    }

    async fn forward<S: Stream<Item = Result<Bytes, std::io::Error>>>(s: Teed<S>) -> Vec<u8> {
        s.map(|c| c.unwrap().to_vec())
            .collect::<Vec<_>>()
            .await
            .concat()
    }

    async fn next_event(rx: &mut mpsc::Receiver<CloudEvent>) -> serde_json::Value {
        tokio::time::timeout(Duration::from_secs(10), rx.recv())
            .await
            .expect("an event within 10s")
            .expect("the channel is open")
            .envelope
    }

    /// One RunSSE stream through a tap: request, then response, both forwarded and compared.
    async fn run_sse(
        st: &Arc<ModelRelayState>,
        dir: &Path,
        resp_headers: HeaderMap,
        response: &[u8],
    ) -> Observation {
        let obs = observation();
        let mut tap = Tap::start_with(
            st,
            CursorRoute::RunSse,
            &HeaderMap::new(),
            obs.clone(),
            Some("cursor"),
            dir.to_path_buf(),
        )
        .expect("a tap");
        let req = tap.tee_request(futures_util::stream::iter(chunks(RUNSSE_REQUEST, 7)));
        assert_eq!(forward(req).await, RUNSSE_REQUEST);
        let resp = tap.tee_response(
            StatusCode::OK,
            &resp_headers,
            futures_util::stream::iter(chunks(response, 3)),
        );
        assert_eq!(forward(resp).await, response, "forwarded byte for byte");
        obs
    }

    #[tokio::test]
    async fn a_runsse_stream_pairs_and_decodes() {
        let (st, mut rx) = state();
        let dir = tempfile::tempdir().unwrap();
        // The append first, as Cursor sends it, through its own tap.
        let mut append = Tap::start_with(
            &st,
            CursorRoute::BidiAppend,
            &HeaderMap::new(),
            observation(),
            None,
            dir.path().to_path_buf(),
        )
        .expect("a tap");
        assert_eq!(
            forward(append.tee_request(futures_util::stream::iter(chunks(APPEND, 11)))).await,
            APPEND
        );
        for _ in 0..100 {
            if st.cursor_pairs.get(PAIR_KEY).is_some() {
                break;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        run_sse(&st, dir.path(), HeaderMap::new(), RUN_RESPONSE).await;
        let ev = next_event(&mut rx).await;
        assert_eq!(ev["data"][emit::CAPTURE_DECODE_KEY], "decoded");
        assert_eq!(ev["olmodelslug"], "claude-4.5-sonnet");
        assert_eq!(ev["olmodelprovider"], "cursor");
        assert_eq!(ev["subject"], CONVERSATION);
        assert_eq!(ev["source"], "cursor");
        assert_eq!(ev["data"]["ai.openlatch.session.assurance"], "attested");
        assert_eq!(ev["data"]["gen_ai.usage.output_tokens"], 42);
        assert!(ev["data"].get("gen_ai.usage.input_tokens").is_none());
        let s = ev.to_string();
        assert!(!s.contains("SENTINEL"), "no content in the record: {s}");
        assert_eq!(
            std::fs::read_dir(dir.path())
                .map(|d| d.count())
                .unwrap_or(0),
            0
        );
    }

    #[tokio::test]
    async fn the_decoder_task_emits_after_the_mirror_closes() {
        let (st, mut rx) = state();
        let dir = tempfile::tempdir().unwrap();
        let mut tap = Tap::start_with(
            &st,
            CursorRoute::Run,
            &HeaderMap::new(),
            observation(),
            Some("cursor"),
            dir.path().to_path_buf(),
        )
        .expect("a tap");
        assert_eq!(
            forward(tap.tee_request(futures_util::stream::iter(chunks(RUN_REQUEST, 5)))).await,
            RUN_REQUEST
        );
        let (tx, body_rx) = mpsc::channel::<Result<Bytes, std::io::Error>>(8);
        let body = futures_util::stream::unfold(body_rx, |mut rx| async move {
            rx.recv().await.map(|c| (c, rx))
        });
        let teed = tap.tee_response(StatusCode::OK, &HeaderMap::new(), body);
        let forwarder = tokio::spawn(forward(teed));
        tx.send(Ok(Bytes::from_static(RUN_RESPONSE))).await.unwrap();
        tokio::time::sleep(Duration::from_millis(200)).await;
        assert!(rx.try_recv().is_err(), "no record while the stream is open");
        drop(tx);
        assert_eq!(forwarder.await.unwrap(), RUN_RESPONSE);
        let ev = next_event(&mut rx).await;
        assert_eq!(ev["data"][emit::CAPTURE_DECODE_KEY], "decoded");
        assert_eq!(ev["subject"], CONVERSATION);
    }

    #[tokio::test]
    async fn a_panicking_decoder_still_emits_undecoded() {
        let (st, mut rx) = state();
        let dir = tempfile::tempdir().unwrap();
        let obs = observation();
        let raw = RawBuf::default();
        let finish = Finish {
            obs: obs.clone(),
            scope: Some("cursor"),
            registry: st.registry.clone(),
            privacy: st.privacy.clone(),
            cloud_tx: st.cloud_tx.clone(),
            raw_dir: dir.path().to_path_buf(),
            raw: raw.clone(),
            mirror_lost: Arc::new(AtomicBool::new(false)),
        };
        // A real response mirror, held open, with more than the cap sent through it.
        let (side_tx, side_rx) = oneshot::channel();
        let (m, mrx) = mirror(Arc::new(AtomicBool::new(false)));
        assert!(side_tx
            .send(ResponseSide {
                status_ok: true,
                enc: Encodings::default(),
                rx: mrx,
            })
            .is_ok());
        let (tx, body_rx) = mpsc::channel::<Result<Bytes, std::io::Error>>(1);
        let body = futures_util::stream::unfold(body_rx, |mut rx| async move {
            rx.recv().await.map(|c| (c, rx))
        });
        let forwarder = tokio::spawn(forward(Teed::new(body, Some(m))));
        let big = vec![0xabu8; RAW_CAP + 4096];
        let sent = big.clone();
        let sender = tokio::spawn(async move {
            for c in sent.chunks(64 * 1024) {
                tx.send(Ok(Bytes::copy_from_slice(c))).await.unwrap();
            }
            tx // the stream stays open: the decoder is mid-stream when it panics
        });
        let watched = raw.clone();
        let decoded = supervise(async move {
            let crash = async move {
                for _ in 0..500 {
                    if watched.lock().unwrap().len() >= RAW_CAP {
                        break;
                    }
                    tokio::time::sleep(Duration::from_millis(10)).await;
                }
                panic!("injected decoder panic");
            };
            tokio::join!(decode_response(side_rx, raw), crash).0
        })
        .await;
        assert!(decoded.is_none(), "the panic is observed on the handle");
        finish.run(decoded).await;
        let ev = next_event(&mut rx).await;
        assert_eq!(ev["data"][emit::CAPTURE_DECODE_KEY], "undecoded");
        assert_eq!(
            ev["data"]["ai.openlatch.capture.gap"],
            "unknown_wire_format"
        );
        // D-17: the response bytes the decoder had received survive its unwind, capped.
        let kept = std::fs::read(dir.path().join(format!("{}.bin", obs.event_id)))
            .expect("raw frames kept after a panic");
        assert_eq!(kept, big[..RAW_CAP]);
        drop(sender.await.unwrap());
        assert_eq!(forwarder.await.unwrap(), big, "forwarding is unaffected");
    }

    #[tokio::test]
    async fn a_gzip_stream_failing_midway_is_partial() {
        let (st, mut rx) = state();
        let dir = tempfile::tempdir().unwrap();
        // Valid frames up to a sync-flush point, then bytes no deflate stream can hold.
        let end = frame(FLAG_END_STREAM, b"{}");
        assert!(
            RUN_RESPONSE.ends_with(&end),
            "premise: the fixture ends the stream"
        );
        let mut e = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
        e.write_all(&RUN_RESPONSE[..RUN_RESPONSE.len() - end.len()])
            .unwrap();
        e.flush().unwrap();
        let valid = e.get_ref().clone();
        let tail = [0xffu8; 64];
        let response = [valid.as_slice(), &tail].concat();
        let obs = observation();
        let mut tap = Tap::start_with(
            &st,
            CursorRoute::Run,
            &HeaderMap::new(),
            obs.clone(),
            Some("cursor"),
            dir.path().to_path_buf(),
        )
        .expect("a tap");
        assert_eq!(
            forward(tap.tee_request(futures_util::stream::iter(chunks(RUN_REQUEST, 5)))).await,
            RUN_REQUEST
        );
        let resp = tap.tee_response(
            StatusCode::OK,
            &headers(&[("content-encoding", "gzip")]),
            // The corrupt tail in a chunk of its own, after every valid byte has decoded.
            futures_util::stream::iter(chunks(&valid, 16).into_iter().chain(chunks(&tail, 64))),
        );
        assert_eq!(forward(resp).await, response, "forwarded byte for byte");
        let ev = next_event(&mut rx).await;
        // All three facts arrived, but the tail did not decode: `partial`, not `decoded`.
        assert_eq!(ev["data"][emit::CAPTURE_DECODE_KEY], "partial");
        assert_eq!(ev["olmodelslug"], "claude-4.5-sonnet");
        assert_eq!(ev["subject"], CONVERSATION);
        assert_eq!(ev["data"]["gen_ai.usage.output_tokens"], 42);
        // D-17 keeps raw frames for `undecoded` only, so a `partial` stream writes none.
        assert!(!dir.path().join(format!("{}.bin", obs.event_id)).exists());
        assert_eq!(
            std::fs::read_dir(dir.path())
                .map(|d| d.count())
                .unwrap_or(0),
            0
        );
    }

    #[tokio::test]
    async fn a_br_encoded_stream_is_undecoded_and_still_forwarded() {
        let (st, mut rx) = state();
        let dir = tempfile::tempdir().unwrap();
        // No append was seen, so RunSSE's facts can only come from its own stream — and `br` is
        // not decoded. The bytes stand in for a brotli body: the decoder must not read them.
        let obs = run_sse(
            &st,
            dir.path(),
            headers(&[("content-encoding", "br")]),
            RUN_RESPONSE,
        )
        .await;
        let ev = next_event(&mut rx).await;
        assert_eq!(ev["data"][emit::CAPTURE_DECODE_KEY], "undecoded");
        assert!(ev["data"].get("gen_ai.usage.output_tokens").is_none());
        // D-17: the response frames, as received, and nothing of the request.
        let kept = std::fs::read(dir.path().join(format!("{}.bin", obs.event_id)))
            .expect("raw frames kept");
        assert_eq!(kept, RUN_RESPONSE);
    }

    #[tokio::test]
    async fn raw_frames_only_on_undecoded_and_only_responses() {
        let (st, mut rx) = state();
        let dir = tempfile::tempdir().unwrap();
        // Partial (tokens, no pairing): nothing kept.
        run_sse(&st, dir.path(), HeaderMap::new(), RUN_RESPONSE).await;
        let ev = next_event(&mut rx).await;
        assert_eq!(ev["data"][emit::CAPTURE_DECODE_KEY], "partial");
        assert_eq!(
            std::fs::read_dir(dir.path())
                .map(|d| d.count())
                .unwrap_or(0),
            0
        );

        // Undecoded: the response only — the request's key is not in the file.
        let junk = b"not a connect stream at all".to_vec();
        let obs = run_sse(&st, dir.path(), HeaderMap::new(), &junk).await;
        let ev = next_event(&mut rx).await;
        assert_eq!(ev["data"][emit::CAPTURE_DECODE_KEY], "undecoded");
        let kept = std::fs::read(dir.path().join(format!("{}.bin", obs.event_id))).unwrap();
        assert_eq!(kept, junk);
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            let mode = std::fs::metadata(dir.path().join(format!("{}.bin", obs.event_id)))
                .unwrap()
                .permissions()
                .mode()
                & 0o777;
            assert_eq!(mode, 0o600);
        }
    }

    /// Connect streaming answers a provider error under HTTP 200, in the end-of-stream frame
    /// (live: every credential-less `Run`/`RunSSE` to api2.cursor.sh). That is `provider_error`,
    /// and its frames are the provider's answer, not decoder drift: nothing is kept.
    #[tokio::test]
    async fn an_end_stream_error_is_a_provider_error_and_keeps_no_frames() {
        let (st, mut rx) = state();
        let dir = tempfile::tempdir().unwrap();
        let error = frame(
            FLAG_END_STREAM,
            br#"{"error":{"code":"unauthenticated","message":"OL-SENTINEL"}}"#,
        );
        let obs = run_sse(&st, dir.path(), HeaderMap::new(), &error).await;
        let ev = next_event(&mut rx).await;
        assert_eq!(ev["data"]["ai.openlatch.capture.gap"], "provider_error");
        assert_eq!(ev["data"][emit::CAPTURE_DECODE_KEY], "undecoded");
        assert!(
            !dir.path().join(format!("{}.bin", obs.event_id)).exists(),
            "a provider error is not decoder drift"
        );

        // A normal end-of-stream `{}` after the same (undecoded) nothing is unchanged: drift.
        let ok = frame(FLAG_END_STREAM, b"{}");
        let obs = run_sse(&st, dir.path(), HeaderMap::new(), &ok).await;
        let ev = next_event(&mut rx).await;
        assert_eq!(
            ev["data"]["ai.openlatch.capture.gap"],
            "unknown_wire_format"
        );
        let kept = std::fs::read(dir.path().join(format!("{}.bin", obs.event_id))).unwrap();
        assert_eq!(kept, ok);
    }

    #[test]
    fn end_stream_error_is_read_from_the_end_frame() {
        let err = frame(FLAG_END_STREAM, br#"{"error":{"code":"unauthenticated"}}"#);
        let (_, d) = messages(&[&err], &Encodings::default());
        assert!(d.ended && d.end_error);
        let gz = frame(
            FLAG_END_STREAM | FLAG_COMPRESSED,
            &gzip(br#"{"error":{"code":"internal"}}"#),
        );
        let (_, d) = messages(&[&gz], &enc(None, Some("gzip")));
        assert!(d.end_error);
        for ok in [&b"{}"[..], br#"{"metadata":{}}"#, br#"{"error":null}"#] {
            let (_, d) = messages(&[&frame(FLAG_END_STREAM, ok)], &Encodings::default());
            assert!(d.ended && !d.end_error);
        }
    }

    #[test]
    fn raw_files_are_pruned_by_age_and_count() {
        let dir = tempfile::tempdir().unwrap();
        let old = dir.path().join("old.bin");
        std::fs::write(&old, b"x").unwrap();
        let eight_days = SystemTime::now() - Duration::from_secs(8 * 24 * 60 * 60);
        std::fs::File::options()
            .write(true)
            .open(&old)
            .unwrap()
            .set_modified(eight_days)
            .unwrap();
        for i in 0..RAW_MAX_FILES + 5 {
            let p = dir.path().join(format!("f{i:03}.bin"));
            std::fs::write(&p, b"x").unwrap();
            let t = SystemTime::now() - Duration::from_secs(3600 + i as u64);
            std::fs::File::options()
                .write(true)
                .open(&p)
                .unwrap()
                .set_modified(t)
                .unwrap();
        }
        let written = write_raw(dir.path(), "evt-new", &vec![7u8; RAW_CAP + 10]).unwrap();
        assert_eq!(
            std::fs::metadata(&written).unwrap().len(),
            RAW_CAP as u64,
            "truncated at the cap"
        );
        assert!(!old.exists(), "older than MAX_AGE");
        let left: Vec<_> = std::fs::read_dir(dir.path()).unwrap().flatten().collect();
        assert_eq!(left.len(), RAW_MAX_FILES);
        assert!(written.exists(), "the newest survives");
    }

    #[test]
    fn concurrent_raw_writes_keep_at_most_the_cap() {
        let dir = tempfile::tempdir().unwrap();
        let bins = |dir: &Path| {
            std::fs::read_dir(dir)
                .unwrap()
                .flatten()
                .filter(|e| e.path().extension().and_then(|x| x.to_str()) == Some("bin"))
                .count()
        };
        // Full already, so every write takes the directory over the cap until its prune runs.
        for i in 0..RAW_MAX_FILES {
            write_raw(dir.path(), &format!("evt-seed-{i}"), b"x").unwrap();
        }
        let running = Arc::new(AtomicUsize::new(8));
        // Between two write-and-prune sections the directory is never over the cap.
        let observer = {
            let (dir, running) = (dir.path().to_path_buf(), running.clone());
            std::thread::spawn(move || {
                let mut most = 0;
                while running.load(Ordering::Acquire) > 0 {
                    {
                        let _between = RAW_WRITE.lock().unwrap();
                        most = most.max(bins(&dir));
                    }
                    std::thread::yield_now();
                }
                most
            })
        };
        let threads: Vec<_> = (0..8)
            .map(|t| {
                let (dir, running) = (dir.path().to_path_buf(), running.clone());
                std::thread::spawn(move || {
                    for i in 0..16 {
                        write_raw(&dir, &format!("evt-{t}-{i}"), b"x").unwrap();
                    }
                    running.fetch_sub(1, Ordering::Release);
                })
            })
            .collect();
        for t in threads {
            t.join().unwrap();
        }
        let most = observer.join().unwrap();
        assert!(most <= RAW_MAX_FILES, "{most} files seen between writes");
        let left = bins(dir.path());
        assert!(left <= RAW_MAX_FILES, "{left} files left");
    }

    #[tokio::test]
    async fn raw_accumulator_stops_at_2mib() {
        let (st, mut rx) = state();
        let dir = tempfile::tempdir().unwrap();
        let big = vec![0xabu8; RAW_CAP + 4096];
        let obs = observation();
        let mut tap = Tap::start_with(
            &st,
            CursorRoute::RunSse,
            &HeaderMap::new(),
            obs.clone(),
            None,
            dir.path().to_path_buf(),
        )
        .unwrap();
        let _ =
            forward(tap.tee_request(futures_util::stream::iter(chunks(RUNSSE_REQUEST, 64)))).await;
        // Chunks the decoder keeps up with, so nothing is dropped on the mirror side.
        let (tx, body_rx) = mpsc::channel::<Result<Bytes, std::io::Error>>(1);
        let body = futures_util::stream::unfold(body_rx, |mut rx| async move {
            rx.recv().await.map(|c| (c, rx))
        });
        let forwarder = tokio::spawn(forward(tap.tee_response(
            StatusCode::OK,
            &HeaderMap::new(),
            body,
        )));
        for c in big.chunks(64 * 1024) {
            tx.send(Ok(Bytes::copy_from_slice(c))).await.unwrap();
        }
        drop(tx);
        assert_eq!(
            forwarder.await.unwrap().len(),
            big.len(),
            "forwarding is unaffected"
        );
        let ev = next_event(&mut rx).await;
        assert_eq!(ev["data"][emit::CAPTURE_DECODE_KEY], "undecoded");
        let kept = std::fs::read(dir.path().join(format!("{}.bin", obs.event_id))).unwrap();
        assert_eq!(kept.len(), RAW_CAP);
    }

    #[tokio::test]
    async fn a_full_mirror_marks_partial_and_the_stream_completes() {
        let (st, mut rx) = state();
        let dir = tempfile::tempdir().unwrap();
        let mut tap = Tap::start_with(
            &st,
            CursorRoute::Run,
            &HeaderMap::new(),
            observation(),
            None,
            dir.path().to_path_buf(),
        )
        .unwrap();
        let _ =
            forward(tap.tee_request(futures_util::stream::iter(chunks(RUN_REQUEST, 1024)))).await;
        // Every chunk is ready at once, so the forward outruns the (unscheduled) decoder: more
        // chunks than the mirror holds.
        let stream = RUN_RESPONSE.repeat(MIRROR_ITEMS + 50);
        let chunks_in = chunks(&stream, RUN_RESPONSE.len());
        assert!(chunks_in.len() > MIRROR_ITEMS);
        let out = forward(tap.tee_response(
            StatusCode::OK,
            &HeaderMap::new(),
            futures_util::stream::iter(chunks_in),
        ))
        .await;
        assert_eq!(out, stream, "the stream completes untouched");
        let ev = next_event(&mut rx).await;
        assert_eq!(ev["data"][emit::CAPTURE_DECODE_KEY], "partial");
    }

    #[tokio::test]
    async fn an_upstream_that_never_answered_still_records() {
        let (st, mut rx) = state();
        let dir = tempfile::tempdir().unwrap();
        let tap = Tap::start_with(
            &st,
            CursorRoute::Run,
            &HeaderMap::new(),
            observation(),
            None,
            dir.path().to_path_buf(),
        )
        .unwrap();
        drop(tap);
        let ev = next_event(&mut rx).await;
        assert_eq!(ev["data"]["ai.openlatch.capture.gap"], "provider_error");
    }

    #[test]
    fn cursor_native_models_are_unknown_to_pricing() {
        for m in [
            "composer-1",
            "grok-4.6",
            "cursor-grok-4.6-medium",
            "default",
            "Composer-2",
        ] {
            assert!(!model_known(m), "{m}");
        }
        let d = Decoded {
            status_ok: Some(true),
            model: Some("composer-1".into()),
            conversation_id: Some("c".into()),
            output_tokens: Some(1),
            ended: true,
            ..Decoded::default()
        };
        assert_eq!(
            capture_gap(&d, false, d.outcome(false), false),
            Some(CaptureGap::UnknownModel)
        );
    }

    /// End to end through the relay: a Connect origin streaming frames slowly, a `Run` request
    /// through `proxy_any`, the body forwarded as it arrives and one decoded record after.
    #[tokio::test(flavor = "multi_thread")]
    async fn a_run_stream_is_recorded_through_the_relay() {
        let frames: Vec<Vec<u8>> = {
            let mut out = Vec::new();
            let mut rest = RUN_RESPONSE;
            while !rest.is_empty() {
                let len = u32::from_be_bytes([rest[1], rest[2], rest[3], rest[4]]) as usize;
                out.push(rest[..5 + len].to_vec());
                rest = &rest[5 + len..];
            }
            out
        };
        let origin =
            crate::model_relay::mock::spawn_connect_frames(frames, Duration::from_millis(40)).await;
        let (tx, mut rx) = mpsc::channel(16);
        let base = reqwest::Url::parse(&format!("http://127.0.0.1:{}", origin.port)).unwrap();
        let st = Arc::new(
            ModelRelayState::new(base, 0, 8, &[])
                .with_measurement(Arc::new(SessionRegistry::default()), Some(tx)),
        );
        let req = axum::http::Request::builder()
            .method(axum::http::Method::POST)
            .uri("/agent.v1.AgentService/Run")
            .header("content-type", "application/connect+proto")
            .body(axum::body::Body::from(RUN_REQUEST))
            .unwrap();
        let resp = crate::model_relay::proxy::proxy_any(axum::extract::State(st), req).await;
        assert_eq!(resp.status(), StatusCode::OK);
        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
            .await
            .unwrap();
        assert_eq!(&body[..], RUN_RESPONSE, "forwarded byte for byte");
        assert_eq!(
            origin.received_body.lock().unwrap().as_deref(),
            Some(RUN_REQUEST)
        );
        let ev = next_event(&mut rx).await;
        assert_eq!(ev["data"][emit::CAPTURE_DECODE_KEY], "decoded");
        assert_eq!(ev["wireformat"], "cursor-connect-rpc");
        assert_eq!(ev["olmodelslug"], "claude-4.5-sonnet");
        assert_eq!(ev["subject"], CONVERSATION);
        assert_eq!(ev["data"]["gen_ai.usage.output_tokens"], 42);
        assert!(!ev.to_string().contains("SENTINEL"));
    }
}