Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
100 changes: 45 additions & 55 deletions crates/llm/src/conversion/messages.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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);
Expand Down
51 changes: 43 additions & 8 deletions crates/llm/src/conversion/messages_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -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
Expand All @@ -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));
}
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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"
Expand Down
37 changes: 33 additions & 4 deletions crates/llm/src/types/messages.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<u64>, Option<u64>) {
let cache_creation = self.rest.get("cache_creation");
let five_m = cache_creation
Expand All @@ -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)
}
}
Expand Down Expand Up @@ -1168,6 +1174,29 @@ pub mod typed {
pub ephemeral_1h_input_tokens: Option<usize>,
}

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<usize>) -> (Option<u64>, Option<u64>) {
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.
Expand Down
Loading