diff --git a/crates/llm/src/conversion/messages.rs b/crates/llm/src/conversion/messages.rs index bcccc76331..7a499d7c36 100644 --- a/crates/llm/src/conversion/messages.rs +++ b/crates/llm/src/conversion/messages.rs @@ -801,22 +801,15 @@ pub mod from_completions { message.usage.cache_read_input_tokens.map(|i| i as u64); r.response.cache_creation_input_tokens = message.usage.cache_creation_input_tokens.map(|i| i as u64); - // Input-side usage (cache write split included) is final at - // `message_start`; a reported all-zero split with a - // small aggregate is the provider's authoritative - // sub-minimum no-op write evidence. - r.response.cache_creation_5m_input_tokens = message - .usage - .cache_creation - .as_ref() - .and_then(|c| c.ephemeral_5m_input_tokens) - .map(|i| i as u64); - r.response.cache_creation_1h_input_tokens = message - .usage - .cache_creation - .as_ref() - .and_then(|c| c.ephemeral_1h_input_tokens) - .map(|i| i as u64); + // Input-side usage (cache write split included) is + // final at `message_start`; only a split consistent + // with its own aggregate becomes evidence. + let (five_m, one_h) = match &message.usage.cache_creation { + Some(c) => c.evidence_against(message.usage.cache_creation_input_tokens), + None => (None, None), + }; + r.response.cache_creation_5m_input_tokens = five_m; + r.response.cache_creation_1h_input_tokens = one_h; // Provisional evidence: the cumulative totals only // arrive on the terminal `message_delta`. r.response.usage_complete = Some(false); @@ -958,19 +951,23 @@ pub mod from_completions { if let Some(cwt) = usage.cache_creation_input_tokens { r.response.cache_creation_input_tokens = Some(cwt as u64); } - // The cumulative delta wins when it carries the split; - // absent leaves earlier evidence untouched. + // The cumulative delta wins when it carries a CONSISTENT + // split; absent or contradictory leaves earlier evidence + // untouched. if usage.cache_creation.is_some() { - r.response.cache_creation_5m_input_tokens = usage - .cache_creation - .as_ref() - .and_then(|c| c.ephemeral_5m_input_tokens) - .map(|i| i as u64); - r.response.cache_creation_1h_input_tokens = usage + let (five_m, one_h) = usage .cache_creation .as_ref() - .and_then(|c| c.ephemeral_1h_input_tokens) - .map(|i| i as u64); + .map(|c| c.evidence_against(usage.cache_creation_input_tokens)) + .unwrap_or((None, None)); + // A consistent delta (including an explicit + // consistent zero) wins; a contradictory one + // degrades to unknown and leaves earlier + // evidence untouched. + if five_m.is_some() || one_h.is_some() { + r.response.cache_creation_5m_input_tokens = five_m; + r.response.cache_creation_1h_input_tokens = one_h; + } } if let Some(o) = usage.output_tokens { r.response.output_tokens = Some(o as u64); @@ -1216,25 +1213,14 @@ pub fn passthrough_stream( r.response.cache_creation_input_tokens = message.usage.cache_creation_input_tokens.map(|i| i as u64); // Input-side usage (cache write split included) is - // final at `message_start`; the terminal - // `message_delta` is cumulative-final. Only - // self-consistent splits become evidence. - // Input-side usage (cache write split included) is final at - // `message_start`; a reported all-zero split with a small - // aggregate is the provider's authoritative sub-minimum - // no-op write evidence. - r.response.cache_creation_5m_input_tokens = message - .usage - .cache_creation - .as_ref() - .and_then(|c| c.ephemeral_5m_input_tokens) - .map(|i| i as u64); - r.response.cache_creation_1h_input_tokens = message - .usage - .cache_creation - .as_ref() - .and_then(|c| c.ephemeral_1h_input_tokens) - .map(|i| i as u64); + // final at `message_start`; only a split consistent + // with its own aggregate becomes evidence. + let (five_m, one_h) = match &message.usage.cache_creation { + Some(c) => c.evidence_against(message.usage.cache_creation_input_tokens), + None => (None, None), + }; + r.response.cache_creation_5m_input_tokens = five_m; + r.response.cache_creation_1h_input_tokens = one_h; // Provisional evidence: the cumulative totals only // arrive on the terminal `message_delta`. r.response.usage_complete = Some(false); @@ -1298,19 +1284,23 @@ pub fn passthrough_stream( if let Some(cwt) = usage.cache_creation_input_tokens { r.response.cache_creation_input_tokens = Some(cwt as u64); } - // The cumulative delta wins when it carries the split; - // absent leaves earlier evidence untouched. + // The cumulative delta wins when it carries a CONSISTENT + // split; absent or contradictory leaves earlier evidence + // untouched. if usage.cache_creation.is_some() { - r.response.cache_creation_5m_input_tokens = usage - .cache_creation - .as_ref() - .and_then(|c| c.ephemeral_5m_input_tokens) - .map(|i| i as u64); - r.response.cache_creation_1h_input_tokens = usage + let (five_m, one_h) = usage .cache_creation .as_ref() - .and_then(|c| c.ephemeral_1h_input_tokens) - .map(|i| i as u64); + .map(|c| c.evidence_against(usage.cache_creation_input_tokens)) + .unwrap_or((None, None)); + // A consistent delta (including an explicit + // consistent zero) wins; a contradictory one + // degrades to unknown and leaves earlier + // evidence untouched. + if five_m.is_some() || one_h.is_some() { + r.response.cache_creation_5m_input_tokens = five_m; + r.response.cache_creation_1h_input_tokens = one_h; + } } if let Some(o) = usage.output_tokens { r.response.output_tokens = Some(o as u64); diff --git a/crates/llm/src/conversion/messages_tests.rs b/crates/llm/src/conversion/messages_tests.rs index 16a1e902dd..221f7ed557 100644 --- a/crates/llm/src/conversion/messages_tests.rs +++ b/crates/llm/src/conversion/messages_tests.rs @@ -494,13 +494,13 @@ async fn passthrough_stream_cache_usage_is_preserved() { } /// Captured production `message_start` shape (golden stream_basic): a -/// small aggregate alongside an all-zero split — the provider's -/// authoritative sub-minimum no-op write evidence. The split buckets are -/// forwarded EXACTLY as reported (Some(0)/Some(0) — real evidence, not -/// absence) and the terminal delta, which carries no split here, leaves -/// that evidence untouched instead of clobbering it. +/// small aggregate alongside an all-zero split. This VIOLATES the +/// Anthropic documented invariant (aggregate == 5m + 1h) for a final +/// snapshot, so it is not zero-cost evidence: the guard degrades the split +/// to unknown and downstream settlement keeps failing closed on the +/// collapsed meter instead of silently unpricing the residual aggregate. #[tokio::test] -async fn passthrough_stream_sub_minimum_zero_split_is_authoritative_evidence() { +async fn passthrough_stream_contradictory_message_start_split_degrades_to_unknown() { let input = format!( "{}\n\n{}\n\n", message_start_with_usage( @@ -510,8 +510,8 @@ async fn passthrough_stream_sub_minimum_zero_split_is_authoritative_evidence() { ); let info = run_passthrough(input).await; assert_eq!(info.response.cache_creation_input_tokens, Some(13)); - assert_eq!(info.response.cache_creation_5m_input_tokens, Some(0)); - assert_eq!(info.response.cache_creation_1h_input_tokens, Some(0)); + assert_eq!(info.response.cache_creation_5m_input_tokens, None); + assert_eq!(info.response.cache_creation_1h_input_tokens, None); } /// Real Anthropic streaming shape (production canary): the input-side @@ -532,3 +532,38 @@ async fn passthrough_stream_consistent_message_start_split_becomes_evidence() { assert_eq!(info.response.cache_creation_5m_input_tokens, Some(67759)); assert_eq!(info.response.cache_creation_1h_input_tokens, Some(0)); } + +/// Precedence: a CONSISTENT terminal delta split wins over the +/// message_start evidence (cumulative-final per protocol). +#[tokio::test] +async fn passthrough_stream_consistent_delta_split_wins_over_start() { + let input = format!( + "{}\n\n{}\n\n", + message_start_with_usage( + r#"{"input_tokens":10,"output_tokens":1,"cache_creation_input_tokens":500,"cache_read_input_tokens":0,"cache_creation":{"ephemeral_5m_input_tokens":500,"ephemeral_1h_input_tokens":0}}"# + ), + r#"data: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":44,"cache_creation_input_tokens":1000,"cache_creation":{"ephemeral_5m_input_tokens":600,"ephemeral_1h_input_tokens":400}}}"#, + ); + let info = run_passthrough(input).await; + assert_eq!(info.response.cache_creation_input_tokens, Some(1000)); + assert_eq!(info.response.cache_creation_5m_input_tokens, Some(600)); + assert_eq!(info.response.cache_creation_1h_input_tokens, Some(400)); +} + +/// A CONTRADICTORY delta split must NOT clobber earlier consistent +/// evidence: the guard degrades the delta's split to unknown and the +/// message_start evidence survives. +#[tokio::test] +async fn passthrough_stream_contradictory_delta_split_leaves_start_evidence() { + let input = format!( + "{}\n\n{}\n\n", + message_start_with_usage( + r#"{"input_tokens":10,"output_tokens":1,"cache_creation_input_tokens":500,"cache_read_input_tokens":0,"cache_creation":{"ephemeral_5m_input_tokens":500,"ephemeral_1h_input_tokens":0}}"# + ), + r#"data: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":44,"cache_creation_input_tokens":1000,"cache_creation":{"ephemeral_5m_input_tokens":0,"ephemeral_1h_input_tokens":0}}}"#, + ); + let info = run_passthrough(input).await; + assert_eq!(info.response.cache_creation_input_tokens, Some(1000)); + assert_eq!(info.response.cache_creation_5m_input_tokens, Some(500)); + assert_eq!(info.response.cache_creation_1h_input_tokens, Some(0)); +} diff --git a/crates/llm/src/tests/response/anthropic/stream_basic.messages-completions-streaming.snap b/crates/llm/src/tests/response/anthropic/stream_basic.messages-completions-streaming.snap index 86ae6528af..c15f292e9a 100644 --- a/crates/llm/src/tests/response/anthropic/stream_basic.messages-completions-streaming.snap +++ b/crates/llm/src/tests/response/anthropic/stream_basic.messages-completions-streaming.snap @@ -24,8 +24,6 @@ data: [DONE] "output_tokens": 21, "total_tokens": 36, "cache_creation_input_tokens": 13, - "cache_creation_5m_input_tokens": 0, - "cache_creation_1h_input_tokens": 0, "cached_input_tokens": 12, "service_tier": "standard", "provider_model": "claude-haiku-4-5-20251001", diff --git a/crates/llm/src/tests/response/anthropic/stream_basic.messages-messages-streaming.snap b/crates/llm/src/tests/response/anthropic/stream_basic.messages-messages-streaming.snap index 6d00ca167a..36922cd4f4 100644 --- a/crates/llm/src/tests/response/anthropic/stream_basic.messages-messages-streaming.snap +++ b/crates/llm/src/tests/response/anthropic/stream_basic.messages-messages-streaming.snap @@ -43,8 +43,6 @@ data: {"type":"message_stop" } "output_tokens": 21, "total_tokens": 36, "cache_creation_input_tokens": 13, - "cache_creation_5m_input_tokens": 0, - "cache_creation_1h_input_tokens": 0, "cached_input_tokens": 12, "service_tier": "standard", "provider_model": "claude-haiku-4-5-20251001", diff --git a/crates/llm/src/tests/response/anthropic/thinking.messages-messages.snap b/crates/llm/src/tests/response/anthropic/thinking.messages-messages.snap index 80cb4bf981..6aa27bf237 100644 --- a/crates/llm/src/tests/response/anthropic/thinking.messages-messages.snap +++ b/crates/llm/src/tests/response/anthropic/thinking.messages-messages.snap @@ -1,5 +1,6 @@ --- source: crates/llm/src/golden_tests.rs +assertion_line: 754 description: src/tests/response/anthropic/thinking.json info: id: msg_01BE9GBzjs95TqdG8FeUanUn @@ -60,8 +61,6 @@ info: "output_tokens": 251, "total_tokens": 298, "cache_creation_input_tokens": 1, - "cache_creation_5m_input_tokens": 0, - "cache_creation_1h_input_tokens": 0, "cached_input_tokens": 2, "service_tier": "standard", "provider_model": "claude-opus-4-6" diff --git a/crates/llm/src/types/messages.rs b/crates/llm/src/types/messages.rs index ce98dc751e..5e16641d9f 100644 --- a/crates/llm/src/types/messages.rs +++ b/crates/llm/src/types/messages.rs @@ -170,10 +170,10 @@ impl Usage { /// reordering. `None` = the provider reported no split; consumers must /// not guess the class from the aggregate. /// - /// A reported all-zero split next to a small aggregate is the - /// provider's authoritative sub-minimum no-op write evidence (observed - /// on captured production usage): the classes are authoritative and the - /// residual aggregate meters at zero — not an inconsistency to hide. + /// Same documented-invariant guard as the streaming path + /// ([`typed::CacheCreationSplit::evidence_against`]): a split whose + /// buckets do not sum to this usage object's own aggregate is not + /// evidence and degrades to unknown — never silently zero-cost. pub fn cache_ttl_split(&self) -> (Option, Option) { let cache_creation = self.rest.get("cache_creation"); let five_m = cache_creation @@ -182,6 +182,12 @@ impl Usage { let one_h = cache_creation .and_then(|c| c.get("ephemeral_1h_input_tokens")) .and_then(|v| v.as_u64()); + if five_m.is_some() || one_h.is_some() { + let sum = five_m.unwrap_or(0).saturating_add(one_h.unwrap_or(0)); + if sum != self.cache_creation_input_tokens.unwrap_or(0) { + return (None, None); + } + } (five_m, one_h) } } @@ -1168,6 +1174,29 @@ pub mod typed { pub ephemeral_1h_input_tokens: Option, } + impl CacheCreationSplit { + /// Per-TTL evidence for telemetry, gated on the Anthropic documented + /// invariant `cache_creation_input_tokens == ephemeral_5m + + /// ephemeral_1h` for the SAME usage snapshot. A split that + /// contradicts its own aggregate is not zero-cost evidence: captured + /// production snapshots report `{5m: 0, 1h: 0}` next to a small + /// positive aggregate (sub-minimum cache markers), and reading that + /// as authoritative would silently unprice the residual aggregate. + /// A contradictory split degrades to unknown — downstream settlement + /// keeps failing closed on the collapsed meter instead. + pub fn evidence_against(&self, aggregate: Option) -> (Option, Option) { + let five_m = self.ephemeral_5m_input_tokens.map(|i| i as u64); + let one_h = self.ephemeral_1h_input_tokens.map(|i| i as u64); + if five_m.is_some() || one_h.is_some() { + let sum = five_m.unwrap_or(0).saturating_add(one_h.unwrap_or(0)); + if sum != aggregate.unwrap_or(0) as u64 { + return (None, None); + } + } + (five_m, one_h) + } + } + #[derive(Clone, Serialize, Deserialize, Debug, Eq, PartialEq)] pub struct MessageDelta { /// The reason that we stopped.