//! Capture from the terminal usage chunk (D-10, D-15) + pricing-input capture
//! (F-31) + the `unknown_model` gap (D-12).
//!
//! Two wire formats are decoded, dispatched on [`WireFormat`]:
//!
//! - **`anthropic-messages`** — usage in the terminal `message_delta` SSE event
//! (streaming) or the whole response body (non-streaming).
//! - **`openai-responses`** — usage at `event.response.usage` on the two
//! measured terminals, `response.completed` and `response.incomplete`.
//!
//! Usage is read **in passing** while the stream forwards. **The FORWARD path
//! never buffers**: `GuardedBody::poll_next` borrows each chunk, scans it and
//! yields it unchanged, and no forwarded byte ever waits for the scanner
//! (REJECTED: collecting the whole SSE stream and then parsing — it buffers and
//! kills TTFT).
//!
//! **The SCANNER is a different thing, and it carries over one partial line**
//! (D-15). It keeps the bytes after the last `\n` it has seen — bounded by
//! [`TAIL_CAP`] — and scans them prefixed to the next chunk, so a usage line
//! split across a chunk boundary still parses. That is not the buffer the
//! never-buffer invariant forbids: it is a copy of at most one partial line, on
//! the observer's side. It exists because a Responses `response.completed`
//! frame embeds the entire `Response` object — instructions, tools, every
//! output item — and therefore straddles a chunk boundary on **every** turn; a
//! one-chunk scanner misses the terminal frame every time and every Codex turn
//! degrades to `tokenizer_estimated`. Over the cap the tail is dropped and the
//! turn degrades, which is honest.
use serde_json::Value;
use super::wire_format::WireFormat;
/// Frozen enum `cost_basis = provider_reported | tokenizer_estimated | interpolated`.
///
/// A property of **capture**, not of pricing. `interpolated` is produced
/// platform-side (F-36); the client emits only the first two.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum CostBasis {
/// The terminal usage chunk arrived cleanly (2xx).
ProviderReported,
/// The stream was interrupted/unparseable — token counts are a local estimate.
TokenizerEstimated,
/// Tokens are provider-reported but no pricebook row matched (platform-set).
Interpolated,
}
impl CostBasis {
pub fn as_str(&self) -> &'static str {
match self {
CostBasis::ProviderReported => "provider_reported",
CostBasis::TokenizerEstimated => "tokenizer_estimated",
CostBasis::Interpolated => "interpolated",
}
}
}
/// Frozen enum `capture_gap = unknown_wire_format | unknown_model | provider_error | stream_interrupted`.
/// Nullable on the wire — set only when capture was incomplete.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum CaptureGap {
/// Body opaque / over the 32 MB ceiling (set by the forwarder, plan 01).
UnknownWireFormat,
/// The request model is not in the known (D-21) set.
UnknownModel,
/// Either the provider returned a non-2xx (F-36) — event emitted, tokens
/// **zero** — or it returned a 2xx whose usage arithmetic did not
/// reconcile (`cached + cache_write > input_tokens`), in which case the
/// measured tokens are **kept**: the output count is still trustworthy,
/// only the input split is not. A doc that promised zeros on a gap that
/// keeps them would be a trap for the next reader.
ProviderError,
/// The response stream ended before a usable terminal usage chunk.
StreamInterrupted,
}
impl CaptureGap {
pub fn as_str(&self) -> &'static str {
match self {
CaptureGap::UnknownWireFormat => "unknown_wire_format",
CaptureGap::UnknownModel => "unknown_model",
CaptureGap::ProviderError => "provider_error",
CaptureGap::StreamInterrupted => "stream_interrupted",
}
}
}
/// The five raw token counts the client emits (C-3). **`input_tokens` is
/// post-last-breakpoint only — never the total.** Total input is
/// `input_tokens + cache_creation + cache_read` and is computed platform-side.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct Usage {
/// `gen_ai.usage.input_tokens` — post-last-breakpoint only.
pub input_tokens: u64,
/// `gen_ai.usage.cache_read.input_tokens`.
pub cache_read: u64,
/// `gen_ai.usage.cache_creation.input_tokens` — the sum of the two buckets.
pub cache_write: u64,
/// `ai.openlatch.cache.ephemeral_5m_input_tokens` (priced 1.25×).
pub eph_5m: u64,
/// `ai.openlatch.cache.ephemeral_1h_input_tokens` (priced 2×).
pub eph_1h: u64,
/// `gen_ai.usage.output_tokens`.
pub output_tokens: u64,
}
impl Usage {
/// Field-wise **max** merge. Anthropic splits usage across `message_start`
/// (final input/cache, preliminary `output_tokens = 1`) and the terminal
/// `message_delta` (final cumulative output), so each field takes the larger
/// of the two — input/cache land once, output grows to its final value.
fn merged_max(self, other: Usage) -> Usage {
Usage {
input_tokens: self.input_tokens.max(other.input_tokens),
cache_read: self.cache_read.max(other.cache_read),
cache_write: self.cache_write.max(other.cache_write),
eph_5m: self.eph_5m.max(other.eph_5m),
eph_1h: self.eph_1h.max(other.eph_1h),
output_tokens: self.output_tokens.max(other.output_tokens),
}
}
}
/// Largest partial line the scanner will hold between chunks (D-15).
///
/// A Responses `response.completed` frame is the whole `Response` object —
/// instructions, tools, every output item — so it is tens to hundreds of KB;
/// `fixture_frame_fits_under_the_tail_cap` pins the real captured size at 4x
/// headroom under this. Memory bound: `TAIL_CAP` per in-flight stream
/// (`DEFAULT_INFLIGHT = 16` in `boundary/mod.rs` → 16 MiB worst case),
/// reachable only by a provider that sends a megabyte with no newline in it.
/// Over the cap the tail is **dropped** and the turn degrades to
/// `tokenizer_estimated` — honest, and bounded.
///
/// The cap is checked **before** the held tail and the incoming chunk are
/// joined, so the bound holds for the transient buffer too and not merely for
/// what is retained. Applying it only to what is retained would leave the real
/// bound at `TAIL_CAP + one chunk`, which is not what this constant promises.
pub const TAIL_CAP: usize = 1 << 20; // 1 MiB
/// Accumulates usage across streamed chunks. Anthropic splits usage across the
/// `message_start` event (input + cache fields, `output_tokens = 1`) and the
/// terminal `message_delta` event (final cumulative `output_tokens`), so fields
/// are merged by **max** — input/cache appear once (message_start), output grows
/// to its final value in message_delta. Non-streaming responses carry a single
/// top-level `usage` object, handled by the same merge. Responses delivers usage
/// once, on the terminal event, so the max-merge is a no-op there — but one
/// accumulator for both formats means the interrupted-stream, cost-basis and
/// gap logic has exactly one implementation.
///
/// **Not `Copy`** — it owns the carry-over tail (D-15). Nothing in the tree
/// copies one by value: every use is a `::default()` into a field or a local.
#[derive(Clone, Debug, Default)]
pub struct UsageAccumulator {
usage: Usage,
/// True once any usage object has been observed (message_start, message_delta,
/// or a non-streaming body) — used only for the "did this stream carry any
/// usage at all" diagnostic, NOT for the cost-basis decision.
seen: bool,
/// True once a **terminal** usage object has been observed: a streaming
/// `message_delta` (final cumulative output) or a complete non-streaming
/// response body. `message_start` — which carries FINAL input/cache but a
/// PRELIMINARY `output_tokens = 1` — deliberately does NOT set this. This flag
/// (not `seen`) is what separates `provider_reported` from
/// `tokenizer_estimated`: a stream that ends before `message_delta` is only
/// partially measured and must fall back to a local estimate.
terminal: bool,
/// True once a decoded usage object's input subtraction **saturated** —
/// the provider reported `cached + cache_write > input_tokens`, so the
/// input split does not reconcile. Read by `Measure::finalize`, which turns
/// it into `capture_gap = provider_error` while KEEPING the measured
/// counts: the output count is still trustworthy.
clamped: bool,
/// The bytes after the last `\n` the scanner has seen, carried into the
/// next chunk (D-15). Empty for every chunk that ends a line — which is
/// every SSE chunk except the ones that split an event — so the steady
/// state costs nothing. Bounded by [`TAIL_CAP`].
tail: Vec<u8>,
}
impl UsageAccumulator {
/// Scan one forwarded chunk — prefixed by the previous chunk's unfinished
/// tail (D-15) — for usage, and merge whatever is found. Returns `true` if
/// this chunk contributed usage.
///
/// **Read-only over the chunk**: the forwarded bytes are never mutated, and
/// the chunk itself is never retained. What is retained is a COPY of the
/// bytes after the view's last `\n` — at most one partial line, bounded by
/// [`TAIL_CAP`] — so a usage line split across a chunk boundary is scanned
/// complete on the chunk that finishes it. No forwarded byte waits for it.
///
/// Re-scanning a partial line is harmless: a truncated `data:` line fails
/// the serde parse and yields `None`, and `merged_max` makes a second
/// sighting of a COMPLETE usage line a no-op.
pub fn scan_chunk(&mut self, fmt: WireFormat, chunk: &[u8]) -> bool {
// CAP BEFORE JOINING, not after. Applying the cap only to what is
// RETAINED would allocate and scan `tail ++ chunk` first, making the
// real bound `TAIL_CAP + one chunk` rather than the `TAIL_CAP` the
// constant promises. Dropping the tail here is the same "over the cap →
// drop, the turn degrades" outcome the retain branch below specifies,
// reached without building the oversized buffer on the way.
if !self.tail.is_empty() && self.tail.len() + chunk.len() > TAIL_CAP {
tracing::debug!(
held = self.tail.len(),
incoming = chunk.len(),
"usage scanner: tail + chunk would exceed TAIL_CAP — tail dropped, turn degrades"
);
self.tail.clear();
}
// The view is `tail ++ chunk` when a tail is held, else the borrowed
// chunk — the zero-copy common case, since every SSE event ends `\n\n`.
let joined: Vec<u8>;
let view: &[u8] = if self.tail.is_empty() {
chunk
} else {
let mut j = std::mem::take(&mut self.tail);
j.extend_from_slice(chunk);
joined = j;
&joined
};
// Scan the WHOLE view, exactly as the one-chunk scan did — a compact
// non-streaming body with no `\n` in it is scanned here, as it always was.
let found = scan_usage(fmt, view);
// Hold back the bytes after the view's LAST `\n`: empty when the view
// ends in a newline, the whole view when it contains none.
let rest: &[u8] = match view.iter().rposition(|&b| b == b'\n') {
Some(i) => &view[i + 1..],
None => view,
};
if rest.len() <= TAIL_CAP {
self.tail = rest.to_vec();
} else {
tracing::debug!(
held = rest.len(),
"usage scanner: partial line exceeds TAIL_CAP — dropped, turn degrades"
);
self.tail.clear();
}
match found {
Some(found) => {
self.merge(found.usage);
self.seen = true;
if found.terminal {
self.terminal = true;
}
if found.clamped {
self.clamped = true;
}
true
}
None => false,
}
}
fn merge(&mut self, u: Usage) {
self.usage = self.usage.merged_max(u);
}
/// True once a usage object has been observed at least once.
pub fn has_usage(&self) -> bool {
self.seen
}
/// True once a **terminal** usage object has been observed — a streaming
/// `message_delta` (final cumulative output) or a complete non-streaming
/// response body. `message_start` (final input/cache but a preliminary
/// `output_tokens = 1`) does NOT set this, so a stream interrupted before the
/// terminal chunk correctly reports "not fully measured" and degrades to a
/// local estimate rather than emitting the preliminary output as final.
pub fn is_terminal(&self) -> bool {
self.terminal
}
/// The accumulated usage.
pub fn usage(&self) -> Usage {
self.usage
}
/// True when a decoded usage object's input subtraction **saturated** —
/// the provider reported `cached + cache_write > input_tokens`, so the
/// input split does not reconcile and `input_tokens` clamped to 0.
///
/// `Measure::finalize` turns this into `capture_gap = provider_error` on an
/// otherwise clean 2xx, and **keeps** the measured counts rather than
/// zeroing them: only the input split is untrustworthy, the output count is
/// still the provider's own number.
pub fn provider_arithmetic_bad(&self) -> bool {
self.clamped
}
}
/// The outcome of scanning one forwarded chunk: the merged usage found and
/// whether any of it came from a **terminal** usage object (a `message_delta` or
/// a non-streaming response body) rather than the preliminary `message_start`.
struct ScanResult {
usage: Usage,
terminal: bool,
/// True when the decoder's input subtraction saturated — the only place
/// that sees the raw counts is [`usage_and_terminal`], so it is the only
/// place that can know, and it reports it as its third member.
clamped: bool,
}
/// Extract a `Usage` from one scanner view, if it carries a usage object,
/// classify whether it carried **terminal** usage, and report whether the
/// decoder's input subtraction saturated.
///
/// Handles both SSE (`data: {…}` lines) and a raw non-streaming JSON body. What
/// a usage object looks like, and where it lives, is the format's business —
/// this is the shell, and it is format-agnostic.
fn scan_usage(fmt: WireFormat, chunk: &[u8]) -> Option<ScanResult> {
let text = std::str::from_utf8(chunk).ok()?;
let mut best: Option<Usage> = None;
let mut terminal = false;
let mut clamped = false;
// SSE data lines first.
for line in text.lines() {
let line = line.trim_start();
let payload = line.strip_prefix("data:").map(str::trim).unwrap_or(line);
if !payload.starts_with('{') {
continue;
}
// Cheap pre-filter before the serde parse: `usage` only appears in
// `message_start` / `message_delta`, so skip the bulk `content_block_delta`
// lines entirely rather than parse-and-throw-away. A line that does contain
// the literal "usage" still parses exactly as before — zero behavior change.
if !payload.contains("usage") {
continue;
}
if let Ok(v) = serde_json::from_str::<Value>(payload) {
if let Some((u, term, clamp)) = usage_and_terminal(fmt, &v) {
best = Some(merge_pick(best, u));
terminal |= term;
clamped |= clamp;
}
}
}
// Non-streaming: the whole chunk may be one JSON object with `.usage`. Guard
// the parse on a leading `{` so a non-JSON chunk is never fed to serde (a bare
// number/array/string could parse yet never carry `.usage`, so this is a pure
// cost cut — zero behavior change).
if best.is_none() && text.trim_start().starts_with('{') {
if let Ok(v) = serde_json::from_str::<Value>(text.trim()) {
if let Some((u, term, clamp)) = usage_and_terminal(fmt, &v) {
best = Some(u);
terminal |= term;
clamped |= clamp;
}
}
}
best.map(|usage| ScanResult {
usage,
terminal,
clamped,
})
}
/// Prefer the usage object carrying the most signal (larger output/input),
/// merging field-wise by max so message_start + message_delta both contribute.
fn merge_pick(prev: Option<Usage>, cur: Usage) -> Usage {
match prev {
None => cur,
Some(p) => p.merged_max(cur),
}
}
/// Pull a `Usage` out of one parsed SSE event or response body, classify
/// whether it is **terminal**, and report whether the mapping's input
/// subtraction saturated.
///
/// This is the one place per format that sees the raw counts, so it is the only
/// place that can know the subtraction clamped — which is why the third member
/// exists: `scan_usage` ORs it into [`ScanResult::clamped`] and the accumulator
/// carries it out to `Measure::finalize`.
///
/// # `anthropic-messages`
///
/// - `message_start` carries usage under `.message.usage` with FINAL input/cache
/// but a PRELIMINARY `output_tokens = 1` → **not terminal**. The stream is not
/// fully measured until the terminal chunk arrives.
/// - A streaming `message_delta` (`.usage`, final cumulative output) and a
/// non-streaming response body (top-level `.usage`, all-final) are **terminal**.
///
/// The `type` discriminator is what distinguishes the two: only `message_start`
/// is treated as preliminary; every other value carrying a top-level `.usage`
/// (message_delta and the non-streaming body, which has no `message_start` type)
/// is a complete measurement. Anthropic's mapping never clamps — its counts are
/// independent, not a total to subtract from — so its third member is always
/// `false`.
///
/// # `openai-responses`
///
/// [`responses_usage_and_terminal`] — keyed on the terminal type literal, with
/// usage read from `event.response.usage`.
///
/// # `unknown`
///
/// `None`. An uncaptured route never reaches a parsed body, and a captured
/// route with no decoder is reported as `unknown_wire_format` rather than
/// guessed at.
fn usage_and_terminal(fmt: WireFormat, v: &Value) -> Option<(Usage, bool, bool)> {
match fmt {
WireFormat::AnthropicMessages => {
if v.get("type").and_then(Value::as_str) == Some("message_start") {
let u = v.get("message").and_then(|m| m.get("usage"))?;
return Some((usage_fields(u), false, false));
}
let u = v.get("usage")?;
Some((usage_fields(u), true, false))
}
WireFormat::OpenAiResponses => responses_usage_and_terminal(v),
WireFormat::Unknown => None,
}
}
/// Read the six raw token fields out of a `usage` object.
fn usage_fields(u: &Value) -> Usage {
let cache_creation = u.get("cache_creation");
let eph_5m = cache_creation
.and_then(|c| c.get("ephemeral_5m_input_tokens"))
.and_then(Value::as_u64)
.unwrap_or(0);
let eph_1h = cache_creation
.and_then(|c| c.get("ephemeral_1h_input_tokens"))
.and_then(Value::as_u64)
.unwrap_or(0);
Usage {
input_tokens: u.get("input_tokens").and_then(Value::as_u64).unwrap_or(0),
cache_read: u
.get("cache_read_input_tokens")
.and_then(Value::as_u64)
.unwrap_or(0),
cache_write: u
.get("cache_creation_input_tokens")
.and_then(Value::as_u64)
.unwrap_or(0),
eph_5m,
eph_1h,
output_tokens: u.get("output_tokens").and_then(Value::as_u64).unwrap_or(0),
}
}
/// Pull a `Usage` out of a parsed **OpenAI Responses** stream event, classify
/// whether it is terminal, and report whether the input subtraction saturated.
///
/// # Key on the type literal, never on field presence
///
/// **Six** stream events embed a full `Response` object — `response.created`,
/// `response.in_progress`, `response.queued`, `response.completed`,
/// `response.failed`, `response.incomplete` — and `usage` is declared
/// *optional on the shared `Response` model*, not forbidden on the
/// non-terminal ones. A decoder that fires on "this chunk contains a usage
/// object" can therefore count the same request twice. The Anthropic scanner
/// above keys on `type` for exactly this reason; so does this one.
///
/// # There are four stream endings, and only two are measured
///
/// | Event | Outcome |
/// | ----- | ------- |
/// | `response.completed` | **terminal, measured** → `provider_reported` |
/// | `response.incomplete` | **terminal, measured** → `provider_reported` |
/// | `response.failed` | terminal, not measured → degrades to the estimate |
/// | `error` | terminal, not measured → degrades to the estimate |
///
/// `response.incomplete` is measured because it means the turn hit a cap
/// (`IncompleteDetails.reason` = `max_output_tokens` | `max_messages` |
/// `content_filter`) and carries FINAL usage. Those are the most expensive
/// turns on the plane; degrading them to a local estimate would throw away the
/// provider's real numbers exactly when they matter most.
///
/// `error` is the ONE member of the 58-event union whose type literal has no
/// `response.` prefix — a decoder matching on that prefix never recognises it
/// as an ending at all. It carries only `{code, message, param,
/// sequence_number}`, so there is no `response` and no usage to read. It and
/// `response.failed` take the same path an interrupted stream already takes,
/// and they get **no new `capture_gap` value**: why a measurement is missing
/// is not something this decoder surfaces, and `stream_interrupted` already
/// says "the stream ended without a measurement".
///
/// # Usage is optional even on `response.completed`
///
/// A well-formed `response.completed` can legally arrive with no `usage`. That
/// is **not measured** — `None`, degrading to `tokenizer_estimated`. It is not
/// zeros (which would report a free model call) and it is not a parse failure.
///
/// **An explicit `"usage": null` is the same answer.** That is the form the
/// wire actually uses — every non-terminal `Response`-bearing event carries
/// `"usage": null` — and `Value::get` answers `Some(Null)` for it, not `None`.
/// Reading the fields off a `Null` yields six zeros, which is exactly the free
/// model call this rule exists to refuse, so the usage must be an OBJECT.
///
/// # Usage is not top-level on the event
///
/// It lives at `event.response.usage`, never `event.usage`. Reading the event
/// root yields nothing on every request, and does so silently.
fn responses_usage_and_terminal(v: &Value) -> Option<(Usage, bool, bool)> {
match v.get("type").and_then(Value::as_str) {
Some("response.completed" | "response.incomplete") => {
let u = v.get("response")?.get("usage").filter(|u| u.is_object())?;
let (usage, clamped) = responses_usage_fields(u);
Some((usage, true, clamped))
}
// Every other literal — the deltas, the other four `Response`-bearing
// events, `response.failed` and the bare `error`.
_ => None,
}
}
/// Read the canonical token counts out of an OpenAI **Responses** `usage`
/// object, and report whether the input subtraction saturated.
///
/// The wire shape, taken from `codex-cli 0.150.1`'s own deserializer (it
/// consumes exactly this payload) and corroborated by OpenAI's published
/// `ResponseUsage` type and the openai-node / openai-python type files:
///
/// ```text
/// usage {
/// input_tokens,
/// input_tokens_details -> { cached_tokens, cache_write_tokens },
/// output_tokens,
/// output_tokens_details -> { reasoning_tokens }, <-- NOT DECODED
/// total_tokens,
/// }
/// ```
///
/// | [`Usage`] field | Responses source |
/// | --------------- | ---------------- |
/// | `input_tokens` | `input_tokens − cached_tokens − cache_write_tokens` |
/// | `cache_read` | `input_tokens_details.cached_tokens` |
/// | `cache_write` | `input_tokens_details.cache_write_tokens` |
/// | `output_tokens` | `output_tokens` |
/// | `eph_5m`, `eph_1h` | always 0 — Anthropic-only TTL buckets, never inferred |
///
/// This is the canonical [`Usage`] contract unchanged — *`input_tokens` is
/// post-last-breakpoint only, never the total* — which is what lets one struct
/// serve both formats with no new fields.
///
/// **Both cache counts are nested under `input_tokens_details`.** The nesting
/// is not asymmetric. Reading `cache_write_tokens` off the top level of
/// `usage` yields `None` on every request, which is indistinguishable from "no
/// cache write happened" — a silent zero rather than a visible failure.
///
/// **`cache_write_tokens` is optional and defaults to 0.** It is documented for
/// GPT-5.6 and later only; on earlier models the field is absent and the
/// formula degrades correctly to `input − cached`. Making it required reads
/// `None` and zeroes the whole subtraction on every older model.
///
/// **Reasoning tokens are deliberately not decoded.**
/// `output_tokens_details.reasoning_tokens` is a *subset* of `output_tokens`,
/// so omitting it under-counts nothing, and it is a number the product does
/// not act on. [`Usage`] gains no field for it. (The name is a trap in its own
/// right: `reasoning_output_tokens` exists in the Codex binary as *internal
/// telemetry* naming, so a grep appears to confirm the wrong path.)
///
/// # The subtraction saturates, and the clamp is a wire contract
///
/// A provider reporting `cached + cache_write > input_tokens` must not produce
/// `u64::MAX`. It produces **0** and sets the returned `clamped` flag, which
/// the caller turns into `capture_gap = provider_error` while **keeping** the
/// measured counts: the output count is still trustworthy, only the input
/// split is not. A negative count is never emitted.
fn responses_usage_fields(u: &Value) -> (Usage, bool) {
let input_tokens = u.get("input_tokens").and_then(Value::as_u64).unwrap_or(0);
let details = u.get("input_tokens_details");
let cache_read = details
.and_then(|d| d.get("cached_tokens"))
.and_then(Value::as_u64)
.unwrap_or(0);
let cache_write = details
.and_then(|d| d.get("cache_write_tokens"))
.and_then(Value::as_u64)
.unwrap_or(0);
// Fresh (post-cache) input, in THREE terms. OpenAI's prompt-caching guide
// computes exactly this — `ordinaryInputTokens = inputTokens -
// cachedTokens - cacheWriteTokens` — which only type-checks if both are
// subsets of the input total; Codex's own telemetry corroborates it by
// emitting a DERIVED `non_cached_input_tokens` alongside the two raw
// counts, a metric that only needs to exist if `input_tokens` is the total.
let fresh = input_tokens
.saturating_sub(cache_read)
.saturating_sub(cache_write);
// `saturating_add` so a provider reporting two enormous counts cannot wrap
// the comparison itself into a false "reconciles".
let clamped = cache_read.saturating_add(cache_write) > input_tokens;
(
Usage {
input_tokens: fresh,
cache_read,
cache_write,
// Anthropic-only TTL buckets. Always 0 for Responses, never inferred.
eph_5m: 0,
eph_1h: 0,
output_tokens: u.get("output_tokens").and_then(Value::as_u64).unwrap_or(0),
},
clamped,
)
}
/// A **real captured** OpenAI Responses `response.completed` frame, verbatim.
///
/// Provenance — this is a recording of the wire, not a hand-written shape. A
/// hand-written fixture encodes the author's belief about the payload, which is
/// exactly what PRD C-10 got wrong in two places:
///
/// | | |
/// | --- | --- |
/// | Source | `dlants/magenta.nvim`, `node/core/src/providers/fixtures/openai/search-cache-ab.json` — recorded live against the OpenAI Responses API |
/// | Commit | `0a02676dc7f5c59412575c8fb10665df263e799c` (2026-08-02) |
/// | Frame | turn 0's `response.completed` event |
/// | Model | `gpt-5.4` |
/// | **Byte size** | **1632 bytes** — the number `fixture_frame_fits_under_the_tail_cap` turns into a gate |
///
/// It carries a real prompt-cache hit (`cached_tokens = 2688`), real reasoning
/// tokens (`153`, which this decoder deliberately does not read) and the full
/// `Response` object, so it exercises the mapping rather than illustrating it.
///
/// The wire shape matches `codex-cli 0.150.1`'s own deserializer field for
/// field — `ResponseCompletedUsage { input_tokens, input_tokens_details -> {
/// cached_tokens, cache_write_tokens }, output_tokens, output_tokens_details ->
/// { reasoning_tokens }, total_tokens }` at
/// `codex-rs/codex-api/src/sse/responses.rs` (tag `rust-v0.150.1`) — which is
/// the contract, since that deserializer consumes exactly this payload.
///
/// **Known limitation, recorded honestly.** This is a capture of the Responses
/// API, not of a Codex turn: a Codex `response.completed` additionally carries
/// Codex's 60–200 KB `instructions` and its `tools`, so a live frame is one to
/// two orders of magnitude larger than this one. The size gate below is
/// therefore a floor, not a ceiling, and the live acceptance block is what
/// proves a real Codex frame fits under [`TAIL_CAP`].
pub const RESPONSES_COMPLETED_FIXTURE: &str = r#"{"type":"response.completed","response":{"id":"resp_095f33e0857d63e1016a6f83dcc6688199a13a9d57edf88690","object":"response","created_at":1785693148,"status":"completed","background":false,"completed_at":1785693170,"error":null,"frequency_penalty":0,"incomplete_details":null,"instructions":"You are a terse assistant. Answer in as few words as possible.","max_output_tokens":null,"max_tool_calls":null,"model":"gpt-5.4","moderation":null,"output":[],"parallel_tool_calls":true,"presence_penalty":0,"previous_response_id":null,"prompt_cache_key":"24b01870-5302-4bcf-afb1-77918396dd62","prompt_cache_retention":"24h","reasoning":{"context":"current_turn","effort":"none","mode":"standard","summary":null},"safety_identifier":"user-PSrNP3YsMUMJKpUurGyolmsy","service_tier":"default","store":false,"temperature":1,"text":{"format":{"type":"text"},"verbosity":"medium"},"tool_choice":"auto","tool_usage":{"image_gen":{"input_tokens":0,"input_tokens_details":{"image_tokens":0,"text_tokens":0},"output_tokens":0,"output_tokens_details":{"image_tokens":0,"text_tokens":0},"total_tokens":0},"web_search":{"num_requests":1}},"tools":[{"type":"web_search","return_token_budget":"default","search_content_types":["text"],"search_context_size":"medium","user_location":{"type":"approximate","city":null,"country":"US","region":null,"timezone":null}}],"top_logprobs":0,"top_p":0.98,"truncation":"disabled","usage":{"input_tokens":14342,"input_tokens_details":{"cache_write_tokens":0,"cached_tokens":2688},"output_tokens":916,"output_tokens_details":{"reasoning_tokens":153},"total_tokens":15258},"user":null,"metadata":{}},"sequence_number":170}"#;
/// The `usage` object exactly as it appears inside
/// [`RESPONSES_COMPLETED_FIXTURE`], so a caller can substitute its own counts
/// into the real frame instead of writing a second, made-up one.
///
/// [`super::mock::spawn_capture_responses_sse`] is the caller.
pub const RESPONSES_COMPLETED_FIXTURE_USAGE: &str = r#""usage":{"input_tokens":14342,"input_tokens_details":{"cache_write_tokens":0,"cached_tokens":2688},"output_tokens":916,"output_tokens_details":{"reasoning_tokens":153},"total_tokens":15258}"#;
/// The pricing-input modifiers derived from the request (F-31). `batch` and
/// `fast_mode` are NOT-NULL wire booleans; `inference_geo` is nullable.
#[derive(Clone, Debug, Default)]
pub struct PricingInputs {
pub batch: bool,
pub fast_mode: bool,
pub inference_geo: Option<String>,
}
/// Derive the pricing inputs from the request body + headers.
///
/// ⚠️ Conservative by design. `/v1/messages` (the captured path) is not a batch
/// endpoint, so `batch` is essentially always false (the PRD flags whether batch
/// traffic transits the listener at all as unverified). `fast_mode` and
/// `inference_geo` have **no confirmed wire source**; they default false/None and
/// are only set when an explicit, unambiguous signal is present.
pub fn derive_pricing_inputs(body: &Value, headers: &axum::http::HeaderMap) -> PricingInputs {
// batch: only true on an explicit request-body flag (defensive — normally
// false on /v1/messages).
let batch = body.get("batch").and_then(Value::as_bool).unwrap_or(false);
// fast_mode: Anthropic exposes no confirmed "fast" flag on /v1/messages.
// Recognise only an explicit body boolean; default false otherwise.
let fast_mode = body
.get("fast_mode")
.and_then(Value::as_bool)
.unwrap_or(false);
// inference_geo: no confirmed source. Read an explicit header if a deployment
// sets one, else null.
let inference_geo = headers
.get("x-openlatch-inference-geo")
.and_then(|v| v.to_str().ok())
.map(|s| s.trim().to_ascii_lowercase())
.filter(|s| !s.is_empty());
PricingInputs {
batch,
fast_mode,
inference_geo,
}
}
/// Extract the `model` string from the request body.
pub fn model_of(body: &Value) -> Option<String> {
body.get("model")
.and_then(Value::as_str)
.map(str::to_string)
}
/// Whether the request body carries at least one `cache_control` breakpoint.
/// Used as context for the (weak) `cache.preserved` signal (D-15).
pub fn has_cache_breakpoint(raw_body: &[u8]) -> bool {
// A substring scan is sufficient and avoids re-parsing the whole body; the
// key only appears as a JSON object key on a real breakpoint.
memmem(raw_body, b"\"cache_control\"")
}
/// Infer `cache.preserved` (D-15) — a **weak/open** signal.
///
/// ⚠️ Cold start, 5-minute TTL expiry, and genuine customer churn all produce
/// `cache_read = 0` legitimately, so a `false` here does not prove the breakpoint
/// was lost. Recorded as an open question (I-1 OQ2); this is a plan-03 release
/// gate, not a settled fact. Inferred `true` only when we actually read cache.
pub fn infer_cache_preserved(usage: &Usage) -> bool {
usage.cache_read > 0
}
/// Tiny substring search (no `memchr` dependency needed for this hot-but-small path).
fn memmem(haystack: &[u8], needle: &[u8]) -> bool {
if needle.is_empty() || haystack.len() < needle.len() {
return false;
}
haystack.windows(needle.len()).any(|w| w == needle)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn c3_input_is_not_the_total() {
// C-3: input_tokens=50, cache_read=100000 → total input MUST be 100050.
// A test that fails if anyone treats input_tokens as the total.
let chunk = br#"data: {"type":"message_start","message":{"usage":{"input_tokens":50,"cache_read_input_tokens":100000,"cache_creation_input_tokens":0,"output_tokens":1}}}"#;
let mut acc = UsageAccumulator::default();
assert!(acc.scan_chunk(WireFormat::AnthropicMessages, chunk));
let u = acc.usage();
assert_eq!(u.input_tokens, 50);
assert_eq!(u.cache_read, 100_000);
let total_input = u.input_tokens + u.cache_write + u.cache_read;
assert_eq!(
total_input, 100_050,
"total input must be input + cache_creation + cache_read (C-3)"
);
}
#[test]
fn merges_message_start_and_message_delta() {
// message_start carries input+cache, output=1; message_delta carries the
// final output. Merge-by-max yields the complete usage.
// The `\n` terminators are what the wire sends, and D-15's carry-over
// needs them: the scanner holds back everything after a view's last
// newline, so an unterminated chunk is by definition a PARTIAL line and
// is prefixed to the next one. Every expected value below is unchanged.
let start = br#"data: {"type":"message_start","message":{"usage":{"input_tokens":10,"cache_read_input_tokens":5,"cache_creation_input_tokens":8,"cache_creation":{"ephemeral_5m_input_tokens":6,"ephemeral_1h_input_tokens":2},"output_tokens":1}}}
"#;
let delta = br#"data: {"type":"message_delta","usage":{"output_tokens":321}}
"#;
let mut acc = UsageAccumulator::default();
acc.scan_chunk(WireFormat::AnthropicMessages, start);
acc.scan_chunk(WireFormat::AnthropicMessages, delta);
let u = acc.usage();
assert_eq!(u.input_tokens, 10);
assert_eq!(u.cache_read, 5);
assert_eq!(u.cache_write, 8);
assert_eq!(u.eph_5m, 6);
assert_eq!(u.eph_1h, 2);
assert_eq!(u.output_tokens, 321);
assert!(acc.has_usage());
}
#[test]
fn message_start_is_not_terminal_until_message_delta() {
// FIX 1: message_start carries FINAL input/cache but a PRELIMINARY
// output_tokens=1, so it must NOT count as terminal. A stream that ends
// here is only partially measured (→ tokenizer_estimated in finalize).
// Terminated, as the wire terminates it — see
// `merges_message_start_and_message_delta`. Values unchanged.
let start = br#"data: {"type":"message_start","message":{"usage":{"input_tokens":10,"cache_read_input_tokens":5,"output_tokens":1}}}
"#;
let mut acc = UsageAccumulator::default();
assert!(
acc.scan_chunk(WireFormat::AnthropicMessages, start),
"message_start contributes input/cache usage"
);
assert!(acc.has_usage(), "usage WAS observed");
assert!(
!acc.is_terminal(),
"but message_start is NOT terminal — output is preliminary (=1)"
);
// The terminal message_delta flips the flag and carries the final output.
let delta = br#"data: {"type":"message_delta","usage":{"output_tokens":321}}
"#;
acc.scan_chunk(WireFormat::AnthropicMessages, delta);
assert!(acc.is_terminal(), "message_delta IS terminal");
assert_eq!(
acc.usage().output_tokens,
321,
"the terminal output overrides the preliminary 1"
);
}
#[test]
fn non_streaming_body_is_terminal() {
// A complete non-streaming response body (top-level .usage, no
// message_start type) is a full measurement → terminal.
let body =
br#"{"id":"msg_1","type":"message","usage":{"input_tokens":42,"output_tokens":7}}"#;
let mut acc = UsageAccumulator::default();
assert!(acc.scan_chunk(WireFormat::AnthropicMessages, body));
assert!(acc.is_terminal());
}
#[test]
fn ephemeral_5m_1h_split_captured() {
let chunk = br#"data: {"usage":{"input_tokens":0,"cache_creation_input_tokens":100,"cache_creation":{"ephemeral_5m_input_tokens":80,"ephemeral_1h_input_tokens":20},"output_tokens":0}}"#;
let mut acc = UsageAccumulator::default();
acc.scan_chunk(WireFormat::AnthropicMessages, chunk);
let u = acc.usage();
assert_eq!(u.eph_5m, 80);
assert_eq!(u.eph_1h, 20);
assert_eq!(u.eph_5m + u.eph_1h, u.cache_write);
}
#[test]
fn non_streaming_body_usage() {
let body = br#"{"id":"msg_1","usage":{"input_tokens":42,"output_tokens":7}}"#;
let mut acc = UsageAccumulator::default();
assert!(acc.scan_chunk(WireFormat::AnthropicMessages, body));
assert_eq!(acc.usage().input_tokens, 42);
assert_eq!(acc.usage().output_tokens, 7);
}
#[test]
fn non_usage_chunk_is_ignored() {
let mut acc = UsageAccumulator::default();
assert!(!acc.scan_chunk(
WireFormat::AnthropicMessages,
b"data: {\"type\":\"content_block_delta\"}\n\n"
));
assert!(!acc.has_usage());
}
#[test]
fn cache_preserved_is_read_gated() {
assert!(infer_cache_preserved(&Usage {
cache_read: 1,
..Default::default()
}));
assert!(!infer_cache_preserved(&Usage::default()));
}
#[test]
fn pricing_inputs_default_conservative() {
let body = serde_json::json!({"model":"claude-opus-4-8","messages":[]});
let p = derive_pricing_inputs(&body, &axum::http::HeaderMap::new());
assert!(!p.batch);
assert!(!p.fast_mode);
assert!(p.inference_geo.is_none());
}
// ---- openai-responses (plan 02) ---------------------------------------
/// The captured frame as it arrives on the wire: one SSE event, one
/// `data:` line, terminated by the blank line.
fn responses_event(body: &str) -> Vec<u8> {
format!("event: response.completed\ndata: {body}\n\n").into_bytes()
}
/// A `data:`-framed Responses event carrying an arbitrary usage object.
fn responses_usage_event(kind: &str, usage: &str) -> Vec<u8> {
format!("data: {{\"type\":\"{kind}\",\"response\":{{\"id\":\"resp_1\",\"usage\":{usage}}}}}\n\n")
.into_bytes()
}
fn responses_acc(chunk: &[u8]) -> UsageAccumulator {
let mut acc = UsageAccumulator::default();
acc.scan_chunk(WireFormat::OpenAiResponses, chunk);
acc
}
#[test]
fn responses_completed_maps_every_field() {
// THE REAL CAPTURED FRAME, verbatim — see RESPONSES_COMPLETED_FIXTURE's
// provenance table. Its recorded counts are input 14342, cached 2688,
// cache_write 0 (the recording predates GPT-5.6, where the field first
// appears), output 916, reasoning 153.
let acc = responses_acc(&responses_event(RESPONSES_COMPLETED_FIXTURE));
let u = acc.usage();
assert!(acc.is_terminal(), "response.completed is terminal");
assert!(acc.has_usage());
// input_tokens - cached_tokens - cache_write_tokens, with cache_write 0
// in this recording (the `- 0` term is written out in
// `responses_cache_write_is_read_from_input_details`, which carries a
// non-zero one).
assert_eq!(
u.input_tokens,
14_342 - 2_688,
"input is FRESH input: input_tokens - cached_tokens - cache_write_tokens"
);
assert_eq!(
u.cache_read, 2_688,
"from input_tokens_details.cached_tokens"
);
assert_eq!(
u.cache_write, 0,
"from input_tokens_details.cache_write_tokens — absent in this model's payload. \
The NESTING is guarded by responses_cache_write_is_read_from_input_details"
);
assert_eq!(u.output_tokens, 916);
assert_eq!(
(u.eph_5m, u.eph_1h),
(0, 0),
"the ephemeral TTL buckets are Anthropic-only and are never inferred"
);
assert!(
!acc.provider_arithmetic_bad(),
"2688 + 0 <= 14342 — this capture reconciles"
);
}
#[test]
fn responses_cache_write_is_read_from_input_details() {
// C-10 asserts, as an observed fact, that `cache_write_tokens` is
// top-level on `usage` while `cached_tokens` is nested. It is not
// asymmetric — both are nested. This fixture carries BOTH, with
// different values, so a C-10-literal mapper reads 7 and reds.
let acc = responses_acc(&responses_usage_event(
"response.completed",
r#"{"input_tokens":1000,"cache_write_tokens":7,"input_tokens_details":{"cached_tokens":100,"cache_write_tokens":250},"output_tokens":9,"total_tokens":1009}"#,
));
let u = acc.usage();
assert_eq!(
u.cache_write, 250,
"the NESTED value wins; 7 is C-10's trap"
);
assert_eq!(u.cache_read, 100);
assert_eq!(
u.input_tokens,
1000 - 100 - 250,
"and the subtraction has THREE terms"
);
}
#[test]
fn responses_absent_cache_write_degrades_to_two_terms() {
// Pre-GPT-5.6: the field simply is not there. It defaults to 0 and the
// formula degrades to `input - cached` — never to zero.
let acc = responses_acc(&responses_usage_event(
"response.completed",
r#"{"input_tokens":1000,"input_tokens_details":{"cached_tokens":400},"output_tokens":9,"total_tokens":1009}"#,
));
let u = acc.usage();
assert_eq!(u.input_tokens, 600, "input - cached, not zero");
assert_eq!(u.cache_write, 0);
assert!(!acc.provider_arithmetic_bad());
}
#[test]
fn responses_cached_exceeding_input_clamps_to_zero() {
// A provider whose counts do not reconcile must produce 0, never a
// wrapped u64::MAX, and must say the capture was wrong.
let acc = responses_acc(&responses_usage_event(
"response.completed",
r#"{"input_tokens":10,"input_tokens_details":{"cached_tokens":8,"cache_write_tokens":5},"output_tokens":4,"total_tokens":14}"#,
));
let u = acc.usage();
assert_eq!(u.input_tokens, 0, "saturating, not wrapping");
assert_ne!(u.input_tokens, u64::MAX);
assert!(
acc.provider_arithmetic_bad(),
"8 + 5 > 10 — the input split does not reconcile"
);
assert_eq!(
u.output_tokens, 4,
"the output count is still the provider's own number and is KEPT"
);
// The gap itself is computed in `Measure::finalize`, one module away —
// `responses_clamp_sets_provider_error` in proxy.rs asserts it.
}
#[test]
fn responses_incomplete_is_terminal_and_measured() {
// D-08: `incomplete` means the turn hit a cap and carries FINAL usage.
// Those are the most expensive turns on the plane.
let acc = responses_acc(&responses_usage_event(
"response.incomplete",
r#"{"input_tokens":900,"input_tokens_details":{"cached_tokens":100,"cache_write_tokens":50},"output_tokens":4096,"total_tokens":4996}"#,
));
let u = acc.usage();
assert!(acc.is_terminal(), "response.incomplete IS terminal");
assert_eq!(u.input_tokens, 750);
assert_eq!(u.cache_read, 100);
assert_eq!(u.cache_write, 50);
assert_eq!(u.output_tokens, 4096);
assert_eq!((u.eph_5m, u.eph_1h), (0, 0));
}
#[test]
fn responses_failed_and_bare_error_are_not_measured() {
// Both carry a usage object here ON PURPOSE: the decoder keys on the
// TYPE LITERAL, so a payload that would be measurable if it were keyed
// on field presence must still not be measured.
let failed = responses_acc(&responses_usage_event(
"response.failed",
r#"{"input_tokens":5,"output_tokens":5,"total_tokens":10}"#,
));
assert!(
!failed.is_terminal(),
"response.failed degrades to the estimate"
);
assert!(!failed.has_usage());
// `error` is the ONE member of the union with no `response.` prefix — a
// prefix-matching decoder never recognises it as an ending at all.
let bare = responses_acc(
br#"data: {"type":"error","code":"server_error","message":"boom","sequence_number":3,"usage":{"input_tokens":5,"output_tokens":5}}"#,
);
assert!(!bare.is_terminal(), "the bare error event degrades too");
assert!(!bare.has_usage());
}
#[test]
fn responses_usage_on_a_non_terminal_event_is_ignored() {
// D-07's double-count guard. Six stream events embed a full `Response`
// and `usage` is optional on the shared model, not forbidden on the
// non-terminal ones — so a presence-keyed decoder counts twice.
//
// The in-progress numbers are strictly GREATER than the terminal's on
// every mapped field: `merged_max` is a field-wise MAX, so smaller
// decoy values would let a broken decoder produce the right answer.
let mut acc = UsageAccumulator::default();
acc.scan_chunk(
WireFormat::OpenAiResponses,
&responses_usage_event(
"response.in_progress",
r#"{"input_tokens":999999,"input_tokens_details":{"cached_tokens":999999,"cache_write_tokens":999999},"output_tokens":999999,"total_tokens":999999}"#,
),
);
assert!(
!acc.has_usage(),
"a non-terminal event contributes NOTHING, not even to `seen`"
);
acc.scan_chunk(
WireFormat::OpenAiResponses,
&responses_usage_event(
"response.completed",
r#"{"input_tokens":300,"input_tokens_details":{"cached_tokens":100,"cache_write_tokens":50},"output_tokens":7,"total_tokens":307}"#,
),
);
let u = acc.usage();
assert!(acc.is_terminal());
assert_eq!(u.input_tokens, 150);
assert_eq!(u.cache_read, 100);
assert_eq!(u.cache_write, 50);
assert_eq!(
u.output_tokens, 7,
"the terminal's exact numbers, not 999999"
);
}
#[test]
fn responses_completed_without_usage_is_not_measured() {
// A well-formed terminal can legally arrive with no usage. That is "not
// measured" — emitting zeros would report a free model call.
let mut acc = UsageAccumulator::default();
let found = acc.scan_chunk(
WireFormat::OpenAiResponses,
br#"data: {"type":"response.completed","response":{"id":"resp_1","status":"completed","usage":null}}"#,
);
assert!(!found);
assert!(!acc.has_usage());
assert!(
!acc.is_terminal(),
"not measured — not zeros, not a parse failure"
);
assert_eq!(acc.usage(), Usage::default());
}
#[test]
fn responses_usage_is_not_read_from_the_event_root() {
// Usage lives at `event.response.usage`, never `event.usage`. Reading
// the root yields nothing on every request, and does so silently.
let mut acc = UsageAccumulator::default();
let found = acc.scan_chunk(
WireFormat::OpenAiResponses,
br#"data: {"type":"response.completed","usage":{"input_tokens":500,"input_tokens_details":{"cached_tokens":10},"output_tokens":20,"total_tokens":520}}"#,
);
assert!(
!found,
"nothing lives under .response, so nothing is measured"
);
assert!(!acc.is_terminal());
assert_eq!(acc.usage(), Usage::default());
}
#[test]
fn responses_split_terminal_line_is_reassembled_by_the_scanner() {
// THE D-15 GATE. It reds on a one-chunk scanner, and on the round-5
// split-at-last-newline form that dropped the held tail whenever the
// incoming chunk carried no newline.
let body = responses_event(RESPONSES_COMPLETED_FIXTURE);
let json_at = body
.windows(6)
.position(|w| w == b"data: ")
.expect("one data: line")
+ 6;
let expect = |acc: &UsageAccumulator, what: &str| {
let u = acc.usage();
assert!(
acc.is_terminal(),
"{what}: the terminal frame must reassemble"
);
assert_eq!(u.input_tokens, 14_342 - 2_688, "{what}");
assert_eq!(u.cache_read, 2_688, "{what}");
assert_eq!(u.cache_write, 0, "{what}");
assert_eq!(u.output_tokens, 916, "{what}");
assert_eq!((u.eph_5m, u.eph_1h), (0, 0), "{what}");
};
// Two chunks, cut at an arbitrary byte inside the JSON line.
let cut = json_at + 200;
let mut two = UsageAccumulator::default();
two.scan_chunk(WireFormat::OpenAiResponses, &body[..cut]);
two.scan_chunk(WireFormat::OpenAiResponses, &body[cut..]);
expect(&two, "two-way cut");
// Five chunks, with newline-free middles — the shape a live turn
// arrives in. A five-way cut whose every chunk happens to carry a `\n`
// passes on the round-5 snippet, so the property is ASSERTED, not
// assumed.
let cuts = [json_at + 10, json_at + 30, json_at + 55, json_at + 80];
let pieces: Vec<&[u8]> = vec![
&body[..cuts[0]],
&body[cuts[0]..cuts[1]],
&body[cuts[1]..cuts[2]],
&body[cuts[2]..cuts[3]],
&body[cuts[3]..],
];
for (i, p) in pieces.iter().enumerate().take(4).skip(1) {
assert!(
!p.contains(&b'\n'),
"middle chunk {i} must be newline-free — that is the case D-15 exists for"
);
}
let mut five = UsageAccumulator::default();
for p in &pieces {
five.scan_chunk(WireFormat::OpenAiResponses, p);
}
expect(&five, "five-way cut with newline-free middles");
}
#[test]
fn anthropic_split_line_is_also_reassembled() {
// The carry-over lives in the SHELL, not in either decoder — so
// Anthropic gets it too. Strictly better than the one-chunk scanner,
// which missed this line entirely.
let line =
br#"data: {"type":"message_delta","usage":{"output_tokens":321,"input_tokens":11}}
"#;
let cut = 30;
let mut acc = UsageAccumulator::default();
acc.scan_chunk(WireFormat::AnthropicMessages, &line[..cut]);
assert!(!acc.is_terminal(), "half a line carries no usage yet");
acc.scan_chunk(WireFormat::AnthropicMessages, &line[cut..]);
assert!(
acc.is_terminal(),
"the completed line parses on the second chunk"
);
assert_eq!(acc.usage().output_tokens, 321);
assert_eq!(acc.usage().input_tokens, 11);
}
/// A `data:` line of exactly `total_len` bytes carrying a COMPLETE,
/// parseable `response.completed` usage object, padded out in
/// `instructions` the way a real Codex frame is.
fn padded_completed_line(total_len: usize) -> Vec<u8> {
let make = |pad: &str| {
format!(
"data: {{\"type\":\"response.completed\",\"response\":{{\"instructions\":\"{pad}\",\"usage\":{{\"input_tokens\":10,\"input_tokens_details\":{{\"cached_tokens\":0,\"cache_write_tokens\":0}},\"output_tokens\":4,\"total_tokens\":14}}}}}}"
)
};
let overhead = make("").len();
make(&"x".repeat(total_len - overhead)).into_bytes()
}
#[test]
fn oversized_tail_is_dropped_and_degrades() {
// (a) A `data:` line longer than TAIL_CAP fed with no newline at all.
// The tail must be RELEASED, not held — an unbounded tail is how a
// provider that never sends a newline grows memory without limit.
let huge: Vec<u8> = b"data: "
.iter()
.copied()
.chain(std::iter::repeat_n(b'x', TAIL_CAP * 3))
.collect();
let mut acc = UsageAccumulator::default();
for piece in [&huge[..50], &huge[50..120], &huge[120..]] {
acc.scan_chunk(WireFormat::OpenAiResponses, piece);
}
assert!(
!acc.is_terminal(),
"an unparseable fragment is not a measurement"
);
assert!(!acc.has_usage());
assert!(
acc.tail.is_empty(),
"over the cap the tail is dropped — memory released, not held"
);
// (b) THE CASE (a) CANNOT SEE. A HELD tail near the cap, then a chunk
// that pushes `tail + chunk` past it. (a) only ever inspects RETAINED
// state, so it passes on an implementation that caps AFTER joining —
// one that allocates and scans `tail ++ chunk` first, making the real
// bound `TAIL_CAP + one chunk`.
//
// The line below is COMPLETE and parseable, so if the joined buffer
// were built the scan would find the usage and set `terminal`. It must
// not: the cap is checked BEFORE the join.
let line = padded_completed_line(TAIL_CAP + 64);
// Control first — the same line in one chunk DOES measure, which is
// what makes the assertion below evidence about the cap rather than
// about an unparseable payload.
let mut control = UsageAccumulator::default();
control.scan_chunk(WireFormat::OpenAiResponses, &line);
assert!(
control.is_terminal(),
"the padded line is genuinely parseable"
);
assert_eq!(control.usage().output_tokens, 4);
let split = TAIL_CAP - 64;
let mut acc = UsageAccumulator::default();
acc.scan_chunk(WireFormat::OpenAiResponses, &line[..split]);
assert_eq!(
acc.tail.len(),
split,
"a newline-free chunk under the cap is held whole"
);
acc.scan_chunk(WireFormat::OpenAiResponses, &line[split..]);
assert!(
!acc.is_terminal(),
"tail + chunk exceeds TAIL_CAP, so the held tail is dropped and the \
joined buffer is never built — the turn degrades"
);
assert_eq!(
acc.tail.len(),
line.len() - split,
"only the incoming chunk is retained; the oversized pair was never joined"
);
}
#[test]
fn fixture_frame_fits_under_the_tail_cap() {
// The byte size recorded in RESPONSES_COMPLETED_FIXTURE's provenance
// table, turned into a gate: 4x headroom for a turn whose `output`
// items are longer than the captured one's.
assert_eq!(
RESPONSES_COMPLETED_FIXTURE.len(),
1632,
"the recorded frame size is part of the fixture's provenance — \
update the doc comment if the fixture is ever re-captured"
);
assert!(RESPONSES_COMPLETED_FIXTURE.len() * 4 <= TAIL_CAP);
}
#[test]
fn anthropic_mapping_is_unchanged_apart_from_the_new_argument() {
// Every Anthropic assertion above keeps its EXPECTED VALUES; the only
// edit those tests took is the added `WireFormat::AnthropicMessages`
// argument. This one re-states the canonical mapping through the new
// signature so the non-regression has a name of its own.
let start = br#"data: {"type":"message_start","message":{"usage":{"input_tokens":10,"cache_read_input_tokens":5,"cache_creation_input_tokens":8,"cache_creation":{"ephemeral_5m_input_tokens":6,"ephemeral_1h_input_tokens":2},"output_tokens":1}}}
"#;
let delta = br#"data: {"type":"message_delta","usage":{"output_tokens":321}}
"#;
let mut acc = UsageAccumulator::default();
assert!(acc.scan_chunk(WireFormat::AnthropicMessages, start));
assert!(
!acc.is_terminal(),
"message_start is still preliminary — output_tokens = 1"
);
assert!(acc.scan_chunk(WireFormat::AnthropicMessages, delta));
assert!(acc.is_terminal());
let u = acc.usage();
assert_eq!(u.input_tokens, 10);
assert_eq!(u.cache_read, 5);
assert_eq!(u.cache_write, 8);
assert_eq!(u.eph_5m, 6);
assert_eq!(u.eph_1h, 2);
assert_eq!(u.output_tokens, 321);
assert!(
!acc.provider_arithmetic_bad(),
"Anthropic's counts are independent, not a total to subtract from — it never clamps"
);
// The non-streaming body, same mapping, still terminal.
let mut body_acc = UsageAccumulator::default();
assert!(body_acc.scan_chunk(
WireFormat::AnthropicMessages,
br#"{"id":"msg_1","type":"message","usage":{"input_tokens":42,"output_tokens":7}}"#
));
assert!(body_acc.is_terminal());
assert_eq!(body_acc.usage().input_tokens, 42);
assert_eq!(body_acc.usage().output_tokens, 7);
}
#[test]
fn breakpoint_detection() {
assert!(has_cache_breakpoint(
br#"{"system":[{"type":"text","cache_control":{"type":"ephemeral"}}]}"#
));
assert!(!has_cache_breakpoint(br#"{"messages":[]}"#));
}
}