//! Capture from the terminal usage chunk (D-10, D-15) + pricing-input capture
//! (F-31) + the `unknown_model` gap (D-12).
//!
//! Four 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`.
//! - **`openai-chat-completions`** — usage at a top-level `.usage` object on
//! the last chunk before the `[DONE]` sentinel.
//! - **`google-generate-content`** — usage at a top-level `usageMetadata`
//! object on **every** chunk, CUMULATIVE, with `candidates[].finishReason`
//! as the end-of-turn marker.
//!
//! 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 `model_relay/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.
/// The literal every usage-bearing line of `fmt` contains: the key the scan
/// gate in [`scan_usage`] pre-filters on.
///
/// Ollama's native frames carry no `usage` object — the counts are top-level
/// `prompt_eval_count` / `eval_count` on the `done: true` line — so under the
/// `usage` literal every one of its lines was skipped. A turn still decoded
/// when that line arrived as a chunk of its own, through the whole-body
/// fallback, and decoded NOTHING when it shared a chunk with the line before
/// it. Spelled as a `match` so a new format has to choose.
fn usage_marker(fmt: WireFormat) -> &'static str {
match fmt {
WireFormat::AnthropicMessages
| WireFormat::OpenAiResponses
| WireFormat::OpenAiChatCompletions
| WireFormat::GoogleGenerateContent
| WireFormat::Unknown => "usage",
WireFormat::OllamaNative => "eval_count",
}
}
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.
//
// R-9 — THIS GATE AND THE `{` ONE ABOVE ARE LOAD-BEARING FOR EVERY
// FORMAT, and both were written for Anthropic alone. Both OpenAI
// formats survive them BY ACCIDENT OF NAMING: their frames arrive as
// `data: {…}` and their counts live under a key spelled literally
// `usage`. A format that spells that key differently is filtered out
// HERE, before its decoder is ever reached, and the turn degrades in
// silence — so adding such a format means widening this gate, not only
// writing the arm. `chat_completions_survives_both_scan_gates` goes
// through `scan_usage` rather than calling `usage_and_terminal`
// directly for exactly that reason: a direct call bypasses both gates
// and proves nothing about them.
//
// **Google is ONE CHARACTER from not surviving, and it is worth naming
// the character.** Gemini spells its key `usageMetadata`, which
// CONTAINS the literal `usage` and therefore clears this gate — by
// accident of Google's naming, not by anything either side agreed.
// Had the field been called `tokenMetadata`, every chunk of every
// Gemini turn would be skipped HERE, the decoder below would never run,
// and the turn would degrade in silence with all of the decoder's own
// unit tests still green. `google_survives_both_scan_gates` goes
// through `scan_usage` for that reason, and this filter is now
// load-bearing for FOUR formats on reasoning written for one.
//
// The `{` gate has a Google-shaped hole the `usage` gate does not:
// `streamGenerateContent` WITHOUT `alt=sse` answers a pretty-printed
// JSON ARRAY, whose lines are `[{`, indented `"key": value,` and `}`.
// None of them clears this pair of gates, and the non-streaming
// fallback below rejects the whole body for its leading `[`, so such a
// turn decodes NOTHING.
//
// That is a degrade, not a miscount: no terminal usage means
// `tokenizer_estimated` + `stream_interrupted`, which is the honest
// report of a turn we could not measure — never a wrong number.
// `google_array_framing_degrades_rather_than_miscounts` pins exactly
// that, so the shape is a measured case rather than one discovered in
// staging. Widening the gates to read the array form is deliberately
// NOT done here: it is reachable only from a caller that requests it,
// Google's own SDKs request `alt=sse` for every streaming call, and
// widening a filter three shipped formats depend on to serve a caller
// we have not observed is the trade this comment exists to refuse.
// Recorded as an outbox OQ.
//
// Ollama's native format is the first that does NOT spell the key
// `usage`, and so the first that needed this gate widened rather than
// surviving it by accident — see `usage_marker`.
if !payload.contains(usage_marker(fmt)) {
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`.
///
/// # `openai-chat-completions`
///
/// [`chat_completions_usage_fields`] — a top-level `.usage` OBJECT, terminal
/// whenever one is present. There is no discriminator to key on and none is
/// needed: the usage chunk is the last one before the `data: [DONE]` sentinel
/// and there is no preliminary-usage case, unlike `message_start`. The chunk
/// carrying it has an EMPTY `choices` array, so nothing on this path indexes
/// `choices[0]`.
///
/// # `google-generate-content`
///
/// [`google_usage_fields`] — a top-level `usageMetadata` OBJECT, which Gemini
/// emits on **EVERY** chunk and which is **CUMULATIVE**: `candidatesTokenCount`
/// grows as the stream progresses.
///
/// "This chunk carries usage" therefore cannot mean terminal here, the way it
/// legitimately does for the format above — it would mark every chunk terminal.
/// The discriminator is `candidates[].finishReason`, Google's own end-of-turn
/// marker, which puts this arm structurally beside Anthropic's `message_start`
/// test rather than beside chat-completions' presence test.
///
/// **The `?` on `usageMetadata` runs BEFORE `terminal` is computed, and that is
/// a real, bounded gap.** A `finishReason` chunk arriving with NO
/// `usageMetadata` answers `None`, so [`scan_usage`] never reads `terminal` out
/// of it and the turn never registers as terminal. That is acceptable only
/// because Gemini documents `usageMetadata` on every chunk including the last;
/// `google_terminal_chunk_without_usage_is_not_a_measurement` pins the
/// assumption so a vendor change fails loudly rather than under-reporting in
/// silence. Note that gate 2 in [`scan_usage`] would have skipped such a line
/// anyway — it carries no literal `usage` — so widening this arm alone would
/// not recover it.
///
/// # `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::OllamaNative => {
// Ollama reports the turn's totals ONCE, on the terminal chunk, and
// identically whether the caller asked for streaming or not —
// verified against a live server: `done`, `done_reason`,
// `prompt_eval_count`, `eval_count` and `prompt_eval_cached_count`
// appear on the final SSE line exactly as they do on a
// non-streaming body.
//
// So `done: true` is the terminal discriminator, and there is no
// running total to clamp: unlike chat-completions, no intermediate
// chunk carries a partial count to be superseded.
if v.get("done").and_then(Value::as_bool) != Some(true) {
return None;
}
let n = |k: &str| v.get(k).and_then(Value::as_u64).unwrap_or(0);
Some((
Usage {
// `prompt_eval_count` is the WHOLE prompt, cached portion
// included — it is not a post-cache remainder the way
// Anthropic's `input_tokens` is. `cache_read` is reported
// beside it for visibility and is NOT subtracted here;
// doing so would silently re-interpret the provider's own
// number.
input_tokens: n("prompt_eval_count"),
cache_read: n("prompt_eval_cached_count"),
// Ollama exposes no cache-creation accounting at all, and a
// zero is the honest answer rather than a guess.
cache_write: 0,
eph_5m: 0,
eph_1h: 0,
output_tokens: n("eval_count"),
},
true,
false,
))
}
WireFormat::OpenAiResponses => responses_usage_and_terminal(v),
WireFormat::OpenAiChatCompletions => {
// `usage` is present-and-NULL on api.openai.com's intermediate
// chunks, so `.get("usage")` is not enough: a JSON null reads as
// "found" and maps to six zeros, which `merged_max` then KEEPS —
// a free model call, reported as a measurement. Require an OBJECT.
// (An OpenAI-compatible server omits the key entirely instead;
// both shapes answer `None` here, by the same filter.)
let u = v.get("usage").filter(|u| u.is_object())?;
let (usage, clamped) = chat_completions_usage_fields(u);
Some((usage, true, clamped))
}
WireFormat::GoogleGenerateContent => {
// An OBJECT, for the reason the arm above gives: a present-and-null
// key reads as "found" and maps to six zeros, which `merged_max`
// then KEEPS — a free model call, reported as a measurement.
let u = v.get("usageMetadata").filter(|u| u.is_object())?;
let (usage, clamped) = google_usage_fields(u);
// DD-06. `usageMetadata` rides EVERY chunk and is CUMULATIVE, so
// the flag CANNOT be "we found usage": that marks every chunk
// terminal, and the first one would then be enough to call the turn
// measured. It is Google's own end-of-turn marker instead.
let terminal = v
.get("candidates")
.and_then(Value::as_array)
.is_some_and(|c| c.iter().any(|x| x.get("finishReason").is_some()));
Some((usage, terminal, clamped))
}
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}"#;
/// Read the canonical token counts out of an OpenAI **chat-completions**
/// `usage` object, and report whether the input subtraction saturated.
///
/// The wire shape, verified 2026-09-14 against OpenAI's published
/// `CompletionUsage` type and against the live capture in
/// [`CHAT_COMPLETIONS_CHUNK_FIXTURE`]:
///
/// ```text
/// usage {
/// prompt_tokens,
/// prompt_tokens_details -> { cached_tokens },
/// completion_tokens,
/// completion_tokens_details -> { reasoning_tokens }, <-- NOT DECODED
/// total_tokens,
/// }
/// ```
///
/// | [`Usage`] field | chat-completions source |
/// | --------------- | ----------------------- |
/// | `input_tokens` | `prompt_tokens − cached_tokens` |
/// | `cache_read` | `prompt_tokens_details.cached_tokens` |
/// | `cache_write` | always 0 — the format has no cache-write count |
/// | `output_tokens` | `completion_tokens` |
/// | `eph_5m`, `eph_1h` | always 0 — Anthropic-only TTL buckets, never inferred |
///
/// # `reasoning_tokens` is INSIDE `completion_tokens`, so adding it double-counts
///
/// `completion_tokens_details.reasoning_tokens` is a *subset* of
/// `completion_tokens`, exactly as the Responses API's
/// `output_tokens_details.reasoning_tokens` is a subset of its `output_tokens`.
/// A decoder that adds it reports more output than the provider billed, on
/// every reasoning turn, in the direction that costs the customer money.
///
/// **This is the opposite of the ruling a format whose reasoning count is a
/// SEPARATE field needs**, where the equivalent number has to be ADDED. The
/// asymmetry is the vendors’, not ours; it is stated wherever a decoder makes
/// the choice so that no reader copies the neighbouring answer.
///
/// # The subtraction has THREE terms, mirroring [`responses_usage_fields`]
///
/// `prompt_tokens` is the TOTAL input *including* both cache buckets, and
/// `prompt_tokens_details` carries them: `cached_tokens` (read from cache) and
/// `cache_write_tokens` (written to it). Fresh input is what remains.
///
/// *Corrected 2026-09-14 by review.* This read two terms and hardcoded
/// `cache_write: 0`, on the reasoning that only the Responses API grew the
/// field. That is not a safe asymmetry to assume: OpenAI's generated schema for
/// this endpoint defines `cache_write_tokens` as "the unadjusted number of
/// prompt tokens written to cache", so hardcoding zero DROPS A REAL PROVIDER
/// COUNT and overstates fresh input by exactly it.
///
/// Reading a field that turns out to be absent costs nothing — it yields `0`
/// and the extra subtraction is a no-op, which is precisely how
/// [`responses_usage_fields`] already treats it on pre-GPT-5.6 models. The
/// asymmetric risk runs the other way: a hardcoded zero is wrong the moment the
/// field IS sent, and wrong silently, because the shape still parses.
///
/// The subtraction **saturates**, and the clamp is reported. A provider
/// claiming `cached_tokens > prompt_tokens` produces **0**, never a wrapped
/// `u64::MAX`, and sets the returned flag — which the caller turns into
/// `capture_gap = provider_error` while KEEPING the measured counts: the output
/// count is still the provider’s own number, only the input split is wrong.
fn chat_completions_usage_fields(u: &Value) -> (Usage, bool) {
let prompt_tokens = u.get("prompt_tokens").and_then(Value::as_u64).unwrap_or(0);
let details = u.get("prompt_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);
(
Usage {
// Fresh (post-cache) input, in THREE terms — same shape as
// `responses_usage_fields`, and absent fields subtract zero.
input_tokens: prompt_tokens
.saturating_sub(cache_read)
.saturating_sub(cache_write),
cache_read,
cache_write,
// Anthropic-only TTL buckets. Always 0 here, never inferred.
eph_5m: 0,
eph_1h: 0,
// ALREADY INCLUDES `completion_tokens_details.reasoning_tokens`.
output_tokens: u
.get("completion_tokens")
.and_then(Value::as_u64)
.unwrap_or(0),
},
// The clamp reports the COMBINED subtraction, not just the read
// bucket — `saturating_add` so a provider reporting two enormous
// counts cannot wrap into a false pass, exactly as the Responses
// decoder does it.
cache_read.saturating_add(cache_write) > prompt_tokens,
)
}
/// A **real captured** chat-completions **terminal usage chunk**, verbatim.
///
/// Provenance — a recording of the wire, not a hand-written shape:
///
/// | | |
/// | --- | --- |
/// | Source | a live `POST /v1/chat/completions` against `ollama 0.34.0`’s OpenAI-compatible surface on loopback, captured 2026-09-14 |
/// | Request | `stream: true`, `stream_options: {include_usage: true}` — the flag without which NO usage object is emitted at all (DD-07) |
/// | Frame | the last `data:` chunk before the `data: [DONE]` sentinel |
/// | Model | `qwen2.5-coder:14b` — an OpenAI-compatible local provider, which is the traffic this route exists for |
/// | **Byte size** | **262 bytes** — the number `chat_completions_fixture_fits_under_the_tail_cap` turns into a gate |
///
/// It carries the trap the decoder is written against, which is why it is a
/// recording rather than an illustration: **`"choices":[]`** — the terminal
/// chunk’s `choices` array is EMPTY, so an implementation validating
/// `choices[0]` fails on the one chunk that carries the measurement.
///
/// **Known limitations, recorded honestly.** An OpenAI-compatible server is not
/// api.openai.com, and this capture omits three things the decoder is written
/// for and this fixture therefore cannot prove:
///
/// - api.openai.com sends `"usage": null` on every INTERMEDIATE chunk; this
/// server omits the key entirely. The `.is_object()` filter answers both, and
/// `chat_completions_null_usage_is_not_a_measurement` carries the null form.
/// - api.openai.com adds `completion_tokens_details.reasoning_tokens`;
/// `chat_completions_reasoning_is_not_added_to_output` carries that.
/// - api.openai.com pads chunks with an `obfuscation` string to normalise their
/// sizes, so nothing may assume anything about chunk *content* shape. Nothing
/// here does — the scanner is line-oriented and the decoder reads `.usage`.
pub const CHAT_COMPLETIONS_CHUNK_FIXTURE: &str = r#"{"id":"chatcmpl-398","object":"chat.completion.chunk","created":1789410009,"model":"qwen2.5-coder:14b","system_fingerprint":"fp_ollama","choices":[],"usage":{"prompt_tokens":34,"prompt_tokens_details":{"cached_tokens":0},"completion_tokens":2,"total_tokens":36}}"#;
/// The `usage` object exactly as it appears inside
/// [`CHAT_COMPLETIONS_CHUNK_FIXTURE`], so a caller can substitute its own
/// counts into the real chunk instead of writing a second, made-up one.
///
/// [`super::mock::spawn_capture_chat_completions_sse`] is the caller.
pub const CHAT_COMPLETIONS_CHUNK_FIXTURE_USAGE: &str = r#""usage":{"prompt_tokens":34,"prompt_tokens_details":{"cached_tokens":0},"completion_tokens":2,"total_tokens":36}"#;
/// Read the canonical token counts out of a Gemini `usageMetadata` object, and
/// report whether the input subtraction saturated.
///
/// The wire shape, from Google's published `GenerateContentResponse` /
/// `UsageMetadata` schema. Every key is camelCase, and only three of the six
/// are ever guaranteed present:
///
/// ```text
/// usageMetadata {
/// promptTokenCount, // required. The TOTAL input, cache INCLUDED
/// cachedContentTokenCount, // optional. Absent means 0, NOT an error
/// candidatesTokenCount, // required. The one that GROWS
/// thoughtsTokenCount, // optional. Reasoning — billed as OUTPUT
/// toolUsePromptTokenCount, // optional. Additional INPUT
/// totalTokenCount, // required. Cross-check only, mapped nowhere
/// }
/// ```
///
/// | [`Usage`] field | generate-content source |
/// | --------------- | ----------------------- |
/// | `input_tokens` | `(promptTokenCount + toolUsePromptTokenCount) − cachedContentTokenCount` |
/// | `cache_read` | `cachedContentTokenCount` |
/// | `cache_write` | always 0 — the format reports no cache-WRITE count, and one is never inferred |
/// | `output_tokens` | `candidatesTokenCount + thoughtsTokenCount` |
/// | `eph_5m`, `eph_1h` | always 0 — Anthropic-only TTL buckets, never inferred |
///
/// # `thoughtsTokenCount` is a SEPARATE field, so it is ADDED
///
/// **This is the INVERSE of the ruling the OpenAI chat-completions decoder
/// above takes, and the two sites sit a few hundred lines apart.** There the
/// reasoning count is a *subset* of the output total and adding it
/// over-reports every reasoning turn; here Google reports its thinking tokens
/// OUTSIDE `candidatesTokenCount` and bills them as output, so *not* adding it
/// under-reports every thinking turn. Same error, opposite direction, and the
/// asymmetry is the vendors' rather than ours — which is why both decoders
/// state it at the site instead of leaving a reader to copy the neighbouring
/// answer across.
///
/// # The ORDER of the three input terms is load-bearing
///
/// `toolUsePromptTokenCount` is additional *input*, so it joins the total
/// BEFORE the cache subtraction. Subtracting first and adding after does two
/// things, both wrong on the same turn: it under-reports a turn whose
/// subtraction saturated, and it computes the clamp against the smaller total —
/// so the turn reports `clamped == false` while having clamped.
///
/// # `totalTokenCount` is a cross-check, not a field
///
/// Google's total is `promptTokenCount + toolUsePromptTokenCount +
/// candidatesTokenCount + thoughtsTokenCount`, which after the mapping above is
/// exactly `input_tokens + cache_read + output_tokens` — the same identity C-3
/// already states about [`Usage`]. `google_total_token_count_reconciles`
/// asserts it rather than storing the number, because a seventh raw count on
/// the wire would be a field the platform has to learn to ignore.
///
/// # Every optional field absent is a ZERO, never an error
///
/// An ordinary turn — no cached content, no tools, no thinking — carries
/// `promptTokenCount`, `candidatesTokenCount` and `totalTokenCount` and nothing
/// else. Treating any of the other three as required would answer `None` for
/// most Gemini turns ever made.
///
/// # The subtraction saturates, and the clamp is reported
///
/// A provider claiming `cachedContentTokenCount > promptTokenCount +
/// toolUsePromptTokenCount` produces **0**, never a wrapped `u64::MAX`, and
/// sets the returned flag — which the caller turns into `capture_gap =
/// provider_error` while KEEPING the measured counts: only the input split is
/// untrustworthy, the output count is still the provider's own number.
fn google_usage_fields(u: &Value) -> (Usage, bool) {
let count = |k: &str| u.get(k).and_then(Value::as_u64).unwrap_or(0);
// Tool-use tokens are additional INPUT and join the total BEFORE the cache
// subtraction — see the doc above; the order is not cosmetic.
// `saturating_add` so two enormous counts cannot wrap the comparison into a
// false "reconciles", exactly as the two decoders above do it.
let gross_input = count("promptTokenCount").saturating_add(count("toolUsePromptTokenCount"));
let cache_read = count("cachedContentTokenCount");
(
Usage {
input_tokens: gross_input.saturating_sub(cache_read),
cache_read,
// No cache-write count exists on this format.
cache_write: 0,
// Anthropic-only TTL buckets. Always 0 here, never inferred.
eph_5m: 0,
eph_1h: 0,
// ADDED, not already included — the inverse of the OpenAI
// chat-completions rule a few hundred lines above.
output_tokens: count("candidatesTokenCount")
.saturating_add(count("thoughtsTokenCount")),
},
cache_read > gross_input,
)
}
/// A **SCHEMA_DERIVED** `google-generate-content` terminal chunk — the JSON of
/// one `data:` frame, verbatim as `alt=sse` sends it.
///
/// Provenance, stated plainly because it is weaker than its two neighbours':
///
/// | | |
/// | --- | --- |
/// | Source | Google's published `GenerateContentResponse` / `UsageMetadata` schema — **constructed, not recorded** |
/// | Written | 2026-09-14 |
/// | Model | `gemini-2.5-pro` |
/// | Shape | the terminal chunk: `candidates[0].finishReason = "STOP"` **and** a `usageMetadata` carrying all six counts |
///
/// **This is NOT a capture, and saying so is the point.** Nobody in this unit
/// can produce a real one: the fake-local-`model_provider` trick that captured
/// Codex's wire yields *requests*, and a real `usageMetadata` needs a real
/// Gemini turn against a real key. Every test in this module works on a
/// schema-derived fixture — the field names, the optionality and the
/// cumulative behaviour are all schema facts — and what a recording would add
/// is a byte-size pin and confidence in the member ORDER.
///
/// **Outbox: replace this with a real capture once any Gemini key is
/// available**, and this comment is what tells the next reader the debt is
/// still open.
///
/// The counts reconcile: `(1024 + 16) − 256 = 784` fresh input, `256` cache
/// read, `64 + 32 = 96` output, and `784 + 256 + 96 = 1136 = totalTokenCount`.
pub const GOOGLE_GENERATE_CONTENT_FIXTURE: &str = r#"{"candidates":[{"content":{"parts":[{"text":"The parser lives in src/parse.rs; start at fn parse_module."}],"role":"model"},"finishReason":"STOP","index":0}],"usageMetadata":{"promptTokenCount":1024,"cachedContentTokenCount":256,"candidatesTokenCount":64,"thoughtsTokenCount":32,"toolUsePromptTokenCount":16,"totalTokenCount":1136},"modelVersion":"gemini-2.5-pro","responseId":"y8nHaPeaDvqfmecPmLXzsAo"}"#;
/// The `usageMetadata` object exactly as it appears inside
/// [`GOOGLE_GENERATE_CONTENT_FIXTURE`], so a caller can substitute its own
/// counts into the same chunk instead of writing a second, made-up one.
///
/// [`super::mock::spawn_capture_google_sse`] is the caller.
pub const GOOGLE_GENERATE_CONTENT_FIXTURE_USAGE: &str = r#""usageMetadata":{"promptTokenCount":1024,"cachedContentTokenCount":256,"candidatesTokenCount":64,"thoughtsTokenCount":32,"toolUsePromptTokenCount":16,"totalTokenCount":1136}"#;
/// 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)
}
/// Extract the model from the request **PATH**, for the one format that puts it
/// there.
///
/// Gemini names the model in the route rather than the body —
/// `/v1beta/models/gemini-2.5-pro:streamGenerateContent` — so [`model_of`]
/// answers `None` for every generate-content turn, the observation stamps
/// `unknown_model`, and the economics event prices **null no matter what the
/// pricebook holds**. Landing pricebook rows while model resolution stays
/// broken buys nothing; this is the other half.
///
/// # Additive on purpose — [`model_of`] is NOT touched
///
/// This is a separate function reached through an `.or_else`, rather than a
/// `WireFormat` parameter added to [`model_of`]. `model_of` is on the hot path
/// of both shipped formats and has one call site; changing its signature to
/// serve a third format costs a shipped-behaviour edit for no gain over a
/// fallback that only ever fires where the body answered nothing.
///
/// # Take the PATH, never the path-and-query
///
/// The caller must pass `uri.path()`. With the query attached,
/// `…:streamGenerateContent?alt=sse&key=AIza...:xyz` puts the LAST colon inside
/// the query string, so `rsplit_once(':')` splits THERE and the "model" this
/// returns is **a fragment of the caller's API key** — which then travels to
/// the platform inside the economics record.
/// `model_from_path_ignores_the_query_string` is that guard.
///
/// # The trace, both surfaces
///
/// `/v1beta/models/gemini-2.5-pro:streamGenerateContent`
/// → `rsplit_once(':')` → `("/v1beta/models/gemini-2.5-pro", …)`
/// → `rsplit_once('/')` → `("/v1beta/models", "gemini-2.5-pro")`.
/// The Vertex path
/// `/v1/projects/p/locations/l/publishers/google/models/gemini-2.5-pro:generateContent`
/// traces identically: the model segment sits in the same position on both
/// surfaces, which is the same reason the route arm matches on the colon
/// suffix (DD-03). A colon in an EARLIER segment is harmless — `rsplit_once`
/// takes the last one.
///
/// # A SILENT site, recorded
///
/// The `_ => None` catch-all means a fifth format whose model also lives in its
/// route falls here and resolves nothing, with no compiler error — the same
/// stale-by-default shape the route-arm plan's NOT-caught table records for
/// `has_decoder` and `transforms_apply`. It is listed there.
pub fn model_from_path(fmt: WireFormat, path: &str) -> Option<String> {
match fmt {
// Between the last '/' and the ':' method suffix.
WireFormat::GoogleGenerateContent => path
.rsplit_once(':')
.map(|(head, _)| head)
.and_then(|head| head.rsplit_once('/'))
.map(|(_, model)| model.to_string())
.filter(|m| !m.is_empty()),
_ => None,
}
}
/// 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 {
/// **Ollama's native totals decode, in both modes.**
///
/// Body shape taken from a live `qwen2.5-coder:7b` on Ollama, not from
/// documentation: the terminal SSE line and the non-streaming body carry the
/// same keys, which is why one arm serves both.
#[test]
fn ollama_native_usage_decodes_from_the_done_chunk() {
let done = serde_json::json!({
"model": "qwen2.5-coder:7b",
"done": true,
"done_reason": "stop",
"prompt_eval_count": 30,
"prompt_eval_cached_count": 4,
"eval_count": 10
});
let (u, terminal, clamp) =
usage_and_terminal(WireFormat::OllamaNative, &done).expect("the done chunk decodes");
assert_eq!(u.input_tokens, 30);
assert_eq!(u.cache_read, 4);
assert_eq!(u.output_tokens, 10);
assert_eq!(
u.cache_write, 0,
"Ollama exposes no cache-creation accounting"
);
assert!(terminal, "`done: true` ends the turn");
assert!(
!clamp,
"totals are reported once, so there is nothing to clamp"
);
// An intermediate streaming chunk carries no totals and must not be
// read as terminal — doing so would end the turn on its first token.
let mid = serde_json::json!({ "model": "qwen2.5-coder:7b", "done": false });
assert!(usage_and_terminal(WireFormat::OllamaNative, &mid).is_none());
}
/// **R-9, for the first format that does not spell its key `usage`.**
///
/// Through `scan_usage`, and on a chunk holding the last content line AND
/// the terminal line — the lines of a live `qwen2.5-coder:7b` stream,
/// newline-delimited JSON with no `data:` framing. That chunk failed both
/// paths before `usage_marker`: the line gate skipped each line for lacking
/// the literal `usage`, and the whole-body fallback could not parse two
/// objects as one. A terminal line arriving alone decoded through the
/// fallback, which is why the live turns looked measured.
#[test]
fn ollama_native_survives_both_scan_gates_when_lines_coalesce() {
let chunk = concat!(
r#"{"model":"qwen2.5-coder:7b","message":{"role":"assistant","content":"ong"},"done":false}"#,
"\n",
r#"{"model":"qwen2.5-coder:7b","message":{"role":"assistant","content":""},"done":true,"done_reason":"stop","prompt_eval_count":36,"prompt_eval_cached_count":35,"eval_count":3}"#,
"\n",
);
assert!(
!chunk.contains("usage"),
"the premise: nothing here survives the Anthropic-shaped literal"
);
let found = scan_usage(WireFormat::OllamaNative, chunk.as_bytes())
.expect("the terminal line clears both gates and decodes");
assert!(found.terminal);
assert_eq!(found.usage.input_tokens, 36);
assert_eq!(found.usage.cache_read, 35);
assert_eq!(found.usage.output_tokens, 3);
}
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());
}
/// **The stubs are gone, and this test outlived them.**
///
/// It was written for the formats that shipped a route before a decoder,
/// narrowed to generate-content when the chat-completions decoder landed,
/// and generate-content was the last one — the `google_*` block below is
/// what replaced it. The name is kept because the commit that shipped the
/// routes gates it by name; deleting it would turn that acceptance red for
/// a reason unrelated to what it checks.
///
/// What it asserts is now the INVARIANT the stubs were an instance of, on
/// the one variant that will always answer it: `None` is this contract's
/// only failure shape and means "not a measurement", which is exactly
/// consistent with `has_decoder() == false`.
///
/// Lives HERE and not beside its twin in `wire_format.rs`:
/// `usage_and_terminal` is private to this module, so naming it from
/// another one is E0603. One test per module, not one test across two.
#[test]
fn unknown_format_is_never_a_measurement() {
// Real terminal frames from two captured formats, read under `Unknown`
// — so the `None` is the arm answering and not a body that carried no
// usage in the first place.
let google = serde_json::json!({
"candidates": [{"finishReason": "STOP"}],
"usageMetadata": {"promptTokenCount": 12, "candidatesTokenCount": 34}
});
let anthropic = serde_json::json!({
"type": "message",
"usage": {"input_tokens": 12, "output_tokens": 34}
});
assert!(usage_and_terminal(WireFormat::Unknown, &google).is_none());
assert!(usage_and_terminal(WireFormat::Unknown, &anthropic).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);
}
/// Extended when the fourth decoder landed. The `observe_request` signature
/// gained a `path` argument in the same commit — the shipped formats do not
/// read it, and the model they resolve still comes from `body["model"]` via
/// an untouched `model_of` (`model_of_is_unchanged_for_shipped_formats` is
/// the other half of that guard). What is re-stated here is the decoding
/// side: three arms now sit beside Anthropic's in `usage_and_terminal`, and
/// its expected values are the ones it had when it was the only one.
#[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);
// And the Responses half of "shipped", through its own real captured
// frame: a fourth arm in the same `match` changed neither its values
// nor its terminal classification.
let responses = responses_acc(&responses_event(RESPONSES_COMPLETED_FIXTURE));
assert!(responses.is_terminal());
assert_eq!(responses.usage().input_tokens, 14_342 - 2_688);
assert_eq!(responses.usage().cache_read, 2_688);
assert_eq!(responses.usage().output_tokens, 916);
assert!(!responses.provider_arithmetic_bad());
}
// ---- openai-chat-completions (plan 03) --------------------------------
/// One SSE event as it arrives on the wire: a `data:` line terminated by a
/// blank line. The capture in the fixture is the JSON only.
fn chat_event(body: &str) -> Vec<u8> {
format!("data: {body}\n\n").into_bytes()
}
/// A `data:`-framed chunk carrying an arbitrary usage value — an object,
/// `null`, or anything else JSON can hold.
fn chat_usage_event(usage: &str) -> Vec<u8> {
format!(
"data: {{\"object\":\"chat.completion.chunk\",\"choices\":[],\"usage\":{usage}}}\n\n"
)
.into_bytes()
}
fn chat_acc(chunk: &[u8]) -> UsageAccumulator {
let mut acc = UsageAccumulator::default();
acc.scan_chunk(WireFormat::OpenAiChatCompletions, chunk);
acc
}
#[test]
fn chat_completions_terminal_usage_is_decoded() {
// THE REAL CAPTURED CHUNK, verbatim — see CHAT_COMPLETIONS_CHUNK_FIXTURE's
// provenance table. Its recorded counts are prompt 34, cached 0,
// completion 2.
let acc = chat_acc(&chat_event(CHAT_COMPLETIONS_CHUNK_FIXTURE));
let u = acc.usage();
assert!(
acc.is_terminal(),
"the usage chunk is the last before [DONE]; there is no preliminary form"
);
assert!(acc.has_usage());
assert_eq!(
u.input_tokens, 34,
"prompt_tokens - cached_tokens, cached 0 here"
);
assert_eq!(u.output_tokens, 2, "completion_tokens");
assert_eq!(u.cache_read, 0);
assert_eq!(
u.cache_write, 0,
"chat-completions has no cache-write count — zero, not invented"
);
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(),
"0 <= 34 — this capture reconciles"
);
}
#[test]
fn chat_completions_null_usage_is_not_a_measurement() {
// api.openai.com puts `"usage": null` on EVERY intermediate chunk. The
// key is present, so `.get("usage")` answers `Some(Null)`, and reading
// six fields off a Null yields six zeros — which `merged_max` then KEEPS
// for the whole stream. A free model call, reported as a measurement.
let acc = chat_acc(&chat_usage_event("null"));
assert!(!acc.has_usage(), "a null usage is not a usage object");
assert!(!acc.is_terminal());
assert_eq!(acc.usage(), Usage::default());
// And the null must not survive a later real one either: the real chunk
// is what the stream is measured by.
let mut acc = UsageAccumulator::default();
acc.scan_chunk(WireFormat::OpenAiChatCompletions, &chat_usage_event("null"));
acc.scan_chunk(
WireFormat::OpenAiChatCompletions,
&chat_event(CHAT_COMPLETIONS_CHUNK_FIXTURE),
);
assert!(acc.is_terminal());
assert_eq!(acc.usage().output_tokens, 2);
}
#[test]
fn chat_completions_empty_choices_is_not_an_error() {
// The captured terminal chunk carries `"choices":[]`. An implementation
// reaching for `choices[0]` — to read a finish_reason, say — panics or
// returns None on the ONE chunk that carries the measurement, and the
// turn degrades on every request.
assert!(
CHAT_COMPLETIONS_CHUNK_FIXTURE.contains(r#""choices":[]"#),
"the fixture must actually carry the empty array, or this proves nothing"
);
let acc = chat_acc(&chat_event(CHAT_COMPLETIONS_CHUNK_FIXTURE));
assert!(acc.is_terminal());
assert_eq!(acc.usage().output_tokens, 2);
}
#[test]
fn chat_completions_cache_write_is_read_not_assumed_zero() {
// REGRESSION 2026-09-14. This decoder hardcoded `cache_write: 0` on the
// reasoning that only the Responses API grew the field. OpenAI's
// generated schema for THIS endpoint defines `cache_write_tokens` as
// "the unadjusted number of prompt tokens written to cache", so a
// hardcoded zero drops a real provider count and overstates fresh input
// by exactly it — silently, because the shape still parses.
//
// SYNTHETIC ON PURPOSE: the live Ollama capture this plan's fixtures
// come from does not emit the field, so it cannot hold this claim.
let u = serde_json::json!({
"prompt_tokens": 100,
"completion_tokens": 7,
"prompt_tokens_details": { "cached_tokens": 30, "cache_write_tokens": 20 },
});
let (usage, clamped) = chat_completions_usage_fields(&u);
assert_eq!(usage.cache_write, 20, "the written bucket must be carried");
assert_eq!(usage.cache_read, 30);
assert_eq!(
usage.input_tokens, 50,
"fresh input is prompt MINUS BOTH buckets: 100 - 30 - 20"
);
assert!(!clamped, "the arithmetic reconciles, so nothing is clamped");
// Absence stays a no-op — which is the whole reason reading it is safe.
let without = serde_json::json!({
"prompt_tokens": 100,
"completion_tokens": 7,
"prompt_tokens_details": { "cached_tokens": 30 },
});
let (usage, clamped) = chat_completions_usage_fields(&without);
assert_eq!(usage.cache_write, 0);
assert_eq!(usage.input_tokens, 70, "one absent term subtracts nothing");
assert!(!clamped);
// And the clamp reports the COMBINED overflow, not just the read bucket.
let over = serde_json::json!({
"prompt_tokens": 40,
"completion_tokens": 1,
"prompt_tokens_details": { "cached_tokens": 30, "cache_write_tokens": 30 },
});
let (usage, clamped) = chat_completions_usage_fields(&over);
assert_eq!(usage.input_tokens, 0, "saturates at zero, never wraps");
assert!(
clamped,
"60 > 40 must be reported even though neither bucket alone is"
);
}
#[test]
fn chat_completions_cached_subtraction_clamps() {
// Never observed, but the mapping cannot prove it cannot happen: a
// provider claiming more cached input than input must produce 0, never a
// wrapped u64::MAX, and must SAY the split did not reconcile.
let acc = chat_acc(&chat_usage_event(
r#"{"prompt_tokens":10,"prompt_tokens_details":{"cached_tokens":12},"completion_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(),
"12 > 10 — the third member is COMPUTED from the subtraction, never hardcoded false"
);
assert_eq!(
u.output_tokens, 4,
"the output count is still the provider's own number and is KEPT"
);
assert_eq!(
u.cache_read, 12,
"and the reported cache count is not rewritten"
);
// The gap itself is computed in `Measure::finalize`, one module away.
}
#[test]
fn chat_completions_reasoning_is_not_added_to_output() {
// OpenAI counts reasoning INSIDE `completion_tokens`, so adding
// `completion_tokens_details.reasoning_tokens` double-counts every
// reasoning turn — in the direction that overstates spend. This is the
// opposite of the ruling a format whose reasoning count is a SEPARATE
// field needs; the asymmetry is the vendors'.
let acc = chat_acc(&chat_usage_event(
r#"{"prompt_tokens":100,"prompt_tokens_details":{"cached_tokens":40},"completion_tokens":900,"completion_tokens_details":{"reasoning_tokens":700},"total_tokens":1000}"#,
));
let u = acc.usage();
assert_eq!(
u.output_tokens, 900,
"completion_tokens verbatim — 1600 would be the double count"
);
assert_eq!(u.input_tokens, 60, "100 - 40");
assert_eq!(
u.cache_read, 40,
"the cache split is the field this record exists for"
);
}
#[test]
fn chat_completions_non_streaming_body_decodes() {
// A REAL captured non-streaming response body — same request as the
// fixture's, without `stream`, against ollama 0.34.0 on 2026-09-14. It
// carries a genuine prompt-cache hit (33 of 34), so the subtraction is
// exercised rather than illustrated.
let body = br#"{"id":"chatcmpl-948","object":"chat.completion","created":1789410027,"model":"qwen2.5-coder:14b","system_fingerprint":"fp_ollama","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}],"usage":{"prompt_tokens":34,"prompt_tokens_details":{"cached_tokens":33},"completion_tokens":2,"total_tokens":36}}"#;
let acc = chat_acc(body);
let u = acc.usage();
assert!(
acc.is_terminal(),
"a complete non-streaming body is a complete measurement"
);
assert_eq!(
u.input_tokens, 1,
"34 - 33: fresh input only, never the total"
);
assert_eq!(u.cache_read, 33);
assert_eq!(u.output_tokens, 2);
assert!(!acc.provider_arithmetic_bad());
}
#[test]
fn chat_completions_survives_both_scan_gates() {
// R-9. `scan_usage` filters every SSE line on `starts_with('{')` and on
// `contains("usage")` BEFORE any parse, and both gates were written for
// Anthropic. Chat-completions clears them by accident of naming, so the
// dependency is pinned HERE — through `scan_usage`, which applies them.
// Calling `usage_and_terminal` directly bypasses both and would stay
// green while every real turn was being filtered out upstream.
let frame = chat_event(CHAT_COMPLETIONS_CHUNK_FIXTURE);
// The two gates, as the scanner applies them, on the real payload.
let text = std::str::from_utf8(&frame).unwrap();
let payload = text
.lines()
.map(str::trim_start)
.find_map(|l| l.strip_prefix("data:").map(str::trim))
.expect("the frame carries one data: line");
assert!(
payload.starts_with('{'),
"gate 1: the payload is a JSON object"
);
assert!(
payload.contains("usage"),
"gate 2: the counts live under a key spelled literally `usage`"
);
let found = scan_usage(WireFormat::OpenAiChatCompletions, &frame)
.expect("the frame clears both gates and decodes");
assert!(found.terminal);
assert_eq!(found.usage.output_tokens, 2);
}
#[test]
fn chat_completions_fixture_fits_under_the_tail_cap() {
// The byte size recorded in CHAT_COMPLETIONS_CHUNK_FIXTURE's provenance
// table, turned into a gate.
assert_eq!(
CHAT_COMPLETIONS_CHUNK_FIXTURE.len(),
262,
"the recorded chunk size is part of the fixture's provenance — \
update the doc comment if the fixture is ever re-captured"
);
// A chat-completions chunk carries no request echo, so it is three
// orders of magnitude under the cap rather than the Responses frame's
// four times. The gate is the same one either way.
assert!(CHAT_COMPLETIONS_CHUNK_FIXTURE.len() * 4 <= TAIL_CAP);
}
#[test]
fn shipped_formats_decode_unchanged_after_chat_completions() {
// The new arm is additive: the two formats that already shipped decode
// their own captured fixtures to exactly the values they did before.
// Named for this plan because the next decoder needs a test of the same
// purpose, and two `shipped_formats_decode_unchanged` in one `mod tests`
// is E0428.
let responses = responses_acc(&responses_event(RESPONSES_COMPLETED_FIXTURE));
assert!(responses.is_terminal());
assert_eq!(responses.usage().input_tokens, 14_342 - 2_688);
assert_eq!(responses.usage().cache_read, 2_688);
assert_eq!(responses.usage().output_tokens, 916);
let mut anthropic = UsageAccumulator::default();
anthropic.scan_chunk(
WireFormat::AnthropicMessages,
br#"{"id":"msg_1","type":"message","usage":{"input_tokens":42,"cache_read_input_tokens":5,"output_tokens":7}}"#,
);
assert!(anthropic.is_terminal());
assert_eq!(
anthropic.usage().input_tokens,
42,
"Anthropic never subtracts"
);
assert_eq!(anthropic.usage().cache_read, 5);
assert_eq!(anthropic.usage().output_tokens, 7);
assert!(
!anthropic.provider_arithmetic_bad(),
"Anthropic's counts are independent — its third member is always false"
);
}
// ---- google-generate-content (plan 04) --------------------------------
/// One SSE event as `alt=sse` frames it: a `data:` line and a blank line.
fn google_event(body: &str) -> Vec<u8> {
format!("data: {body}\n\n").into_bytes()
}
/// A `usageMetadata` object with all six counts, Google's own total.
///
/// `totalTokenCount` is prompt + tool-use + candidates + thoughts, which is
/// how Google computes it — `cachedContentTokenCount` is a SUBSET of the
/// prompt and so is never added again.
fn google_usage(
prompt: u64,
cached: u64,
candidates: u64,
thoughts: u64,
tool_use: u64,
) -> String {
let total = prompt + tool_use + candidates + thoughts;
format!(
r#""usageMetadata":{{"promptTokenCount":{prompt},"cachedContentTokenCount":{cached},"candidatesTokenCount":{candidates},"thoughtsTokenCount":{thoughts},"toolUsePromptTokenCount":{tool_use},"totalTokenCount":{total}}}"#
)
}
/// One generate-content chunk carrying an arbitrary usage fragment, with or
/// without the end-of-turn marker.
fn google_chunk(usage: &str, finish: bool) -> String {
let fin = if finish {
r#","finishReason":"STOP""#
} else {
""
};
format!(
r#"{{"candidates":[{{"content":{{"parts":[{{"text":"hi"}}],"role":"model"}}{fin},"index":0}}],{usage},"modelVersion":"gemini-2.5-pro"}}"#
)
}
fn google_acc(chunks: &[String]) -> UsageAccumulator {
let mut acc = UsageAccumulator::default();
for c in chunks {
acc.scan_chunk(WireFormat::GoogleGenerateContent, &google_event(c));
}
acc
}
/// **DD-06, and the reason this plan exists.**
///
/// Gemini puts `usageMetadata` on EVERY chunk and it is CUMULATIVE. An
/// accumulator that sums — the correct instinct for Anthropic and for
/// OpenAI, both of which report usage once — multiplies the turn by roughly
/// the chunk count and inflates the customer's bill by the same factor.
#[test]
fn google_usage_is_cumulative_take_the_last() {
let acc = google_acc(&[
google_chunk(&google_usage(1024, 256, 10, 4, 16), false),
google_chunk(&google_usage(1024, 256, 25, 12, 16), false),
google_chunk(&google_usage(1024, 256, 64, 32, 16), true),
]);
let u = acc.usage();
assert_eq!(
u.output_tokens,
64 + 32,
"the LAST cumulative counts, not the sum"
);
assert_ne!(
u.output_tokens,
(10 + 25 + 64) + (4 + 12 + 32),
"summing a cumulative counter bills this turn three times"
);
// The input side is constant across the stream and must not triple either.
assert_eq!(u.input_tokens, (1024 + 16) - 256);
assert_eq!(u.cache_read, 256);
}
/// The accident, pinned.
///
/// `merge_pick` → `merged_max` is element-wise max, designed for Anthropic's
/// split of usage across `message_start` and `message_delta`. Against a
/// MONOTONICALLY GROWING cumulative counter it happens to land on the final
/// value — which is why the arm above needs no special accumulation. That is
/// luck, not intent: a future change to `merge_pick` (say, to sum, or to
/// prefer the first sighting) would break Google alone, silently, with every
/// Anthropic and OpenAI test still green.
#[test]
fn merged_max_lands_on_final_cumulative_value() {
let chunks = [
google_usage(1024, 256, 10, 4, 16),
google_usage(1024, 256, 25, 12, 16),
google_usage(1024, 256, 64, 32, 16),
];
let decoded: Vec<Usage> = chunks
.iter()
.map(|u| {
let v: Value = serde_json::from_str(&google_chunk(u, false)).unwrap();
usage_and_terminal(WireFormat::GoogleGenerateContent, &v)
.expect("every chunk carries usage")
.0
})
.collect();
let merged = decoded
.iter()
.fold(None, |acc, u| Some(merge_pick(acc, *u)))
.expect("three chunks merge to something");
assert_eq!(
merged,
*decoded.last().unwrap(),
"element-wise max over a growing counter IS the last value — the whole \
Google mapping rides on this, so a change to merge_pick fails here first"
);
}
/// Since EVERY chunk carries usage, "has usage" cannot mean terminal.
/// Google's own end-of-turn marker is `candidates[].finishReason`.
#[test]
fn google_terminal_only_on_finish_reason() {
let mid = google_acc(&[google_chunk(&google_usage(1024, 0, 10, 0, 0), false)]);
assert!(mid.has_usage(), "the chunk DID carry usage");
assert!(
!mid.is_terminal(),
"a usage-bearing mid-stream chunk is not the end of the turn"
);
let done = google_acc(&[
google_chunk(&google_usage(1024, 0, 10, 0, 0), false),
google_chunk(&google_usage(1024, 0, 64, 0, 0), true),
]);
assert!(done.is_terminal(), "finishReason IS the marker");
assert_eq!(done.usage().output_tokens, 64);
}
/// A stream that ends with no `finishReason`.
///
/// **What the accumulator holds and what the event reports are different
/// things, and only the first is asserted here.** The accumulator carries
/// the last cumulative counts with `is_terminal() == false`. `finalize`
/// (`proxy.rs`) then takes the non-terminal branch and **DISCARDS those
/// counts entirely**: it emits a local tokenizer estimate with
/// `cost_basis = tokenizer_estimated` and `capture_gap = stream_interrupted`
/// (or `unknown_model`, when a model string resolved off the known set).
///
/// That is deliberate — C-4 honesty: an incomplete stream is never reported
/// `provider_reported` — and it is stated here because an earlier draft of
/// this plan claimed the counts were "kept, correctly flagged incomplete".
/// They are not kept. Read `Measure::finalize` before repeating it.
#[test]
fn google_interrupted_stream_yields_non_terminal_measurement() {
let acc = google_acc(&[
google_chunk(&google_usage(1024, 256, 10, 4, 16), false),
google_chunk(&google_usage(1024, 256, 25, 12, 16), false),
]);
assert!(acc.has_usage());
assert!(
!acc.is_terminal(),
"no finishReason arrived, so the turn is not measured"
);
assert_eq!(
acc.usage().output_tokens,
25 + 12,
"the accumulator holds the last cumulative counts…"
);
// …and `finalize` throws them away for a tokenizer estimate. The proxy
// half of that is `google_split_across_chunk_boundary`'s neighbourhood.
}
/// The counterpart gap, pinned so a vendor change fails loudly.
///
/// The `?` on `usageMetadata` runs BEFORE `terminal` is computed, so a
/// `finishReason` chunk arriving with no usage answers `None` and the turn
/// never registers as terminal. Gate 2 in `scan_usage` would have skipped
/// the line anyway — it carries no literal `usage` — so this is not
/// recoverable by widening the arm alone. It is acceptable only because
/// Gemini documents `usageMetadata` on every chunk INCLUDING the last.
#[test]
fn google_terminal_chunk_without_usage_is_not_a_measurement() {
let bare = r#"{"candidates":[{"content":{"parts":[{"text":"hi"}],"role":"model"},"finishReason":"STOP","index":0}],"modelVersion":"gemini-2.5-pro"}"#;
let v: Value = serde_json::from_str(bare).unwrap();
assert!(
usage_and_terminal(WireFormat::GoogleGenerateContent, &v).is_none(),
"no usage object, so no measurement — and therefore no terminal flag"
);
let acc = google_acc(&[bare.to_string()]);
assert!(!acc.has_usage());
assert!(!acc.is_terminal());
}
/// An ordinary turn: no cached content, no tools, no thinking. Three of the
/// six fields are absent, and every one of them is a ZERO rather than an
/// error — treating any as required answers `None` for most Gemini turns
/// ever made.
#[test]
fn google_optional_fields_absent_is_zero_not_error() {
let minimal = r#""usageMetadata":{"promptTokenCount":1024,"candidatesTokenCount":64,"totalTokenCount":1088}"#;
let acc = google_acc(&[google_chunk(minimal, true)]);
assert!(acc.is_terminal(), "still a measurement");
let u = acc.usage();
assert_eq!(u.input_tokens, 1024);
assert_eq!(u.cache_read, 0);
assert_eq!(u.output_tokens, 64);
assert_eq!(u.cache_write, 0);
assert_eq!(u.eph_5m, 0);
assert_eq!(u.eph_1h, 0);
assert!(!acc.provider_arithmetic_bad());
}
/// `thoughtsTokenCount` is a SEPARATE field and must be ADDED to output —
/// the exact inverse of chat-completions' `reasoning_tokens`, which is a
/// subset of `completion_tokens` and must NOT be. Not adding it silently
/// under-bills every thinking turn.
#[test]
fn google_thoughts_tokens_count_as_output() {
let acc = google_acc(&[google_chunk(&google_usage(100, 0, 64, 32, 0), true)]);
assert_eq!(
acc.usage().output_tokens,
96,
"candidates + thoughts, because Google bills reasoning as output"
);
assert_ne!(
acc.usage().output_tokens,
64,
"reading candidatesTokenCount alone drops every reasoning token"
);
}
/// The order of the three input terms, which was unstated in an earlier
/// draft and is not cosmetic.
///
/// `toolUsePromptTokenCount` is additional INPUT, so it joins the total
/// BEFORE the cache subtraction. Subtracting first and adding after gives
/// `(10 − 12).saturating → 0`, then `+6 = 6`, and reports `clamped == false`
/// on a turn that clamped.
#[test]
fn google_tool_use_tokens_join_input_before_the_subtraction() {
let acc = google_acc(&[google_chunk(&google_usage(10, 12, 5, 0, 6), true)]);
assert_eq!(
acc.usage().input_tokens,
(10 + 6) - 12,
"tool-use tokens join the gross input, then the cache is subtracted"
);
assert!(
!acc.provider_arithmetic_bad(),
"16 ≥ 12 reconciles — the wrong order would have reported a clamp here"
);
}
/// The subtraction saturates at zero AND says so. `clamped` is COMPUTED —
/// an earlier draft of this arm passed a hardcoded `false` while its own
/// prose required the saturation to be reported.
#[test]
fn google_cached_subtraction_clamps() {
let acc = google_acc(&[google_chunk(&google_usage(100, 500, 64, 32, 0), true)]);
assert_eq!(acc.usage().input_tokens, 0, "saturates, never wraps");
assert!(
acc.provider_arithmetic_bad(),
"the clamp is the third member, and finalize turns it into provider_error"
);
assert_eq!(
acc.usage().output_tokens,
96,
"the output count is still the provider's own number and is KEPT"
);
}
/// `totalTokenCount` is a cross-check, not a field.
///
/// After the mapping, `input_tokens + cache_read + output_tokens` is exactly
/// Google's total — the same identity C-3 states about `Usage`. Asserted
/// rather than stored, so the platform never has to learn to ignore a
/// seventh raw count.
#[test]
fn google_total_token_count_reconciles() {
let acc = google_acc(&[GOOGLE_GENERATE_CONTENT_FIXTURE.to_string()]);
let u = acc.usage();
assert_eq!(
u.input_tokens + u.cache_read + u.output_tokens,
1136,
"the fixture's totalTokenCount"
);
}
/// The whole schema-derived fixture, every field.
#[test]
fn google_fixture_decodes_every_field() {
let acc = google_acc(&[GOOGLE_GENERATE_CONTENT_FIXTURE.to_string()]);
let u = acc.usage();
assert!(acc.is_terminal(), "the fixture carries finishReason: STOP");
assert_eq!(u.input_tokens, (1024 + 16) - 256);
assert_eq!(u.cache_read, 256);
assert_eq!(u.output_tokens, 64 + 32);
assert_eq!(u.cache_write, 0, "the format reports no cache-write count");
assert_eq!(u.eph_5m, 0);
assert_eq!(u.eph_1h, 0);
assert!(!acc.provider_arithmetic_bad());
}
/// **R-9.** Through `scan_usage`, never through `usage_and_terminal`.
///
/// `scan_usage` applies two filters before any parse, both written for
/// Anthropic: the payload must start `{` after the `data:` strip, and it
/// must contain the literal `usage`. Google clears the second only because
/// its key is spelled `usageMetadata` — had Google called it
/// `tokenMetadata`, every chunk would be skipped HERE, the decoder would
/// never run, and the turn would degrade in silence with all of the
/// decoder's own unit tests green. Calling `usage_and_terminal` directly
/// bypasses both gates and proves nothing about them.
#[test]
fn google_survives_both_scan_gates() {
let found = scan_usage(
WireFormat::GoogleGenerateContent,
&google_event(GOOGLE_GENERATE_CONTENT_FIXTURE),
)
.expect("the chunk clears both gates and decodes");
assert!(found.terminal);
assert_eq!(found.usage.output_tokens, 64 + 32);
assert_eq!(found.usage.cache_read, 256);
}
/// The `{` gate's Google-shaped hole, measured rather than discovered in
/// staging.
///
/// `streamGenerateContent` WITHOUT `alt=sse` answers a pretty-printed JSON
/// ARRAY. No line of it clears gate 1 (`[{`, indented `"key": value,`, `}`),
/// and the non-streaming fallback rejects the whole body for its leading
/// `[`. So the turn decodes NOTHING — which is a DEGRADE, not a miscount:
/// no terminal usage means `tokenizer_estimated` + `stream_interrupted`,
/// the honest report of a turn we could not measure. The counts are never
/// wrong, only absent.
#[test]
fn google_array_framing_degrades_rather_than_miscounts() {
let array = format!(
"[{{\n \"candidates\": [],\n {}\n}}\n,\n{{\n \"candidates\": []\n}}\n]",
r#""usageMetadata": {"promptTokenCount": 1024, "candidatesTokenCount": 64}"#
);
let mut acc = UsageAccumulator::default();
acc.scan_chunk(WireFormat::GoogleGenerateContent, array.as_bytes());
assert!(
!acc.has_usage(),
"the array framing decodes to nothing — absent, never wrong"
);
assert!(!acc.is_terminal());
assert_eq!(acc.usage(), Usage::default());
}
/// **The tail carry-over, asserted directly.**
///
/// This replaces the byte-size pin its two sibling fixtures carry. "A
/// hand-authored const is under 1 MiB" is unfalsifiable for a chunk of a few
/// hundred bytes; what the Responses cap test actually protects is D-15's
/// carry-over, and this asserts that — the chunk is fed to the scanner split
/// at EVERY byte offset, and must decode identically every time.
#[test]
fn google_usage_survives_a_tail_split_at_every_offset() {
let event = google_event(GOOGLE_GENERATE_CONTENT_FIXTURE);
let whole = google_acc(&[GOOGLE_GENERATE_CONTENT_FIXTURE.to_string()]);
for i in 1..event.len() {
let mut acc = UsageAccumulator::default();
acc.scan_chunk(WireFormat::GoogleGenerateContent, &event[..i]);
acc.scan_chunk(WireFormat::GoogleGenerateContent, &event[i..]);
assert_eq!(
acc.usage(),
whole.usage(),
"usage differs when the chunk is split at byte {i}"
);
assert!(
acc.is_terminal(),
"terminal lost when the chunk is split at byte {i}"
);
}
}
/// A present-and-NULL `usageMetadata` is not a measurement — six zeros that
/// `merged_max` would KEEP, reported as a free model call. The same
/// `.is_object()` filter the two OpenAI arms carry.
#[test]
fn google_null_usage_is_not_a_measurement() {
let chunk = google_chunk(r#""usageMetadata":null"#, true);
let v: Value = serde_json::from_str(&chunk).unwrap();
assert!(usage_and_terminal(WireFormat::GoogleGenerateContent, &v).is_none());
}
// ---- DD-08: the model lives in the path ------------------------------
/// Both surfaces, both method suffixes. A `rsplit_once('/')` without the
/// `:` strip returns `gemini-2.5-pro:streamGenerateContent`, which is not a
/// model and matches no pricebook row.
#[test]
fn model_from_path_extracts_gemini_and_vertex() {
for (path, want) in [
(
"/v1beta/models/gemini-2.5-pro:streamGenerateContent",
"gemini-2.5-pro",
),
("/v1beta/models/gemini-2.5-flash:generateContent", "gemini-2.5-flash"),
(
"/v1/projects/bea/locations/us-central1/publishers/google/models/gemini-2.5-pro:streamGenerateContent",
"gemini-2.5-pro",
),
(
"/v1/projects/bea/locations/us-central1/publishers/google/models/gemini-2.5-pro:generateContent",
"gemini-2.5-pro",
),
] {
assert_eq!(
model_from_path(WireFormat::GoogleGenerateContent, path).as_deref(),
Some(want),
"path {path}"
);
}
// Degenerate shapes resolve nothing rather than junk.
for path in [
"/v1beta/models/:streamGenerateContent",
"/v1beta/models/gemini-2.5-pro",
"",
] {
assert_eq!(
model_from_path(WireFormat::GoogleGenerateContent, path),
None,
"path {path:?}"
);
}
}
/// **A credential leak, and the reason the caller must pass `uri.path()`.**
///
/// With the query attached, the LAST colon of
/// `…:streamGenerateContent?alt=sse&key=AIza…:xyz` is inside the QUERY, so
/// `rsplit_once(':')` splits there and what comes back is a fragment of the
/// caller's API key — which then travels to the platform inside the
/// economics record as the model.
#[test]
fn model_from_path_ignores_the_query_string() {
let path = "/v1beta/models/gemini-2.5-pro:streamGenerateContent";
assert_eq!(
model_from_path(WireFormat::GoogleGenerateContent, path).as_deref(),
Some("gemini-2.5-pro"),
);
// The trap, spelled out: the same function handed `path_and_query()`.
let leaked = model_from_path(
WireFormat::GoogleGenerateContent,
&format!("{path}?alt=sse&key=AIzaSyFAKEKEY:v1"),
)
.expect("it resolves SOMETHING, which is the whole problem");
assert_ne!(leaked, "gemini-2.5-pro");
assert!(
leaked.contains("AIzaSyFAKEKEY"),
"this is the customer's key, not a model — pass uri.path(), got {leaked:?}"
);
}
/// **R-8.** `model_of` was not touched, so this passes trivially — which is
/// exactly what it pins. DD-08 changed `observe_request`'s signature instead
/// of this function's, because `model_of` is on the hot path of both shipped
/// formats and the fallback only ever fires where the body answered nothing.
#[test]
fn model_of_is_unchanged_for_shipped_formats() {
let anthropic = serde_json::json!({"model": "claude-opus-4-8", "messages": []});
let responses = serde_json::json!({"model": "gpt-5-codex", "input": "hi"});
assert_eq!(model_of(&anthropic).as_deref(), Some("claude-opus-4-8"));
assert_eq!(model_of(&responses).as_deref(), Some("gpt-5-codex"));
// And the new fallback is inert for every format but Google, so the
// `.or_else` in `observe_request` cannot change what they resolve.
for fmt in [
WireFormat::AnthropicMessages,
WireFormat::OpenAiResponses,
WireFormat::OpenAiChatCompletions,
WireFormat::Unknown,
] {
assert_eq!(
model_from_path(fmt, "/v1beta/models/gemini-2.5-pro:generateContent"),
None,
"{fmt:?} must resolve no model from a path"
);
}
// The other direction: a Gemini body carries no `model` key, which is
// why the fallback had to exist at all.
let google: Value =
serde_json::from_str(r#"{"contents":[{"role":"user","parts":[{"text":"hi"}]}]}"#)
.unwrap();
assert_eq!(model_of(&google), None);
}
#[test]
fn breakpoint_detection() {
assert!(has_cache_breakpoint(
br#"{"system":[{"type":"text","cache_control":{"type":"ephemeral"}}]}"#
));
assert!(!has_cache_breakpoint(br#"{"messages":[]}"#));
}
}