[fix]: Anthropic provider - stop double-counting cache-read tokens in streaming usage - #5355
[fix]: Anthropic provider - stop double-counting cache-read tokens in streaming usage#5355is911 wants to merge 1 commit into
Conversation
… streaming usage The streaming accumulator in HandleAnthropicChatCompletionStreaming used a max-keep guard on input_tokens for both message_start and message_delta events indiscriminately. When message_delta arrived with its authoritative (smaller) uncached-tail input_tokens, the max-keep rejected it and retained the stale message_start full-prompt value; then normalizeCachedUsage folded cache_read_input_tokens on top at stream end, double-counting the cached portion (~2x prompt_tokens inflation on every cached request from turn 2 on). The fix makes message_delta.usage authoritative for all input-side fields via event-level overwrite (PromptTokens = delta.input + cache_creation + cache_read), preserves the existing message_start max-keep as the delta-absent fallback for early-terminated streams, adds an impossible-zero guard (PromptTokens == 0 && startSnapshot.InputTokens > 0 -> restore start-side) for non-conformant Anthropic-compatible providers that emit a partial-usage message_delta, and skips the end-of-stream normalizeCachedUsage fold when message_delta was processed. Affected packages: - core/providers/anthropic/anthropic.go - core/providers/anthropic/streamcacheusage_test.go (new) - core/changelog.md Tests: - TestAnthropicStreamCacheReadNotDoubleCounted: turn-2 fixture (start input=20532/cache_read=0; delta input=308/cache_read=20224). Pre-fix FAILS recording 40756; post-fix PASSES recording 20532. - TestAnthropicStreamNoCacheReadUnchanged: turn-1 non-regression (no cache). - TestAnthropicStreamUsage_NonConformantDeltaOmitsCacheFields: proves the impossible-zero guard is load-bearing for non-conformant providers. - Existing core/providers/anthropic/... tests: zero pass/fail diff. go vet, go build green. Closes maximhq#5354
📝 WalkthroughSummary by CodeRabbit
WalkthroughAnthropic streaming usage accumulation now overwrites prompt-token fields from authoritative ChangesAnthropic streaming usage
Estimated code review effort: 3 (Moderate) | ~25 minutes Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Warning There were issues while running some tools. Please review the errors and either fix the tool's configuration or disable the tool if it's a critical failure. 🔧 golangci-lint (2.12.2)level=error msg="[linters_context] typechecking error: pattern ./...: directory prefix . does not contain main module or its selected dependencies" Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Confidence Score: 4/5The changed usage accumulator needs fixes for partial deltas and stale cache-write details before merging.
core/providers/anthropic/anthropic.go Important Files Changed
Reviews (1): Last reviewed commit: "[fix]: Anthropic provider - stop double-..." | Re-trigger Greptile |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| if usageToProcess.CacheCreation.Ephemeral5mInputTokens > 0 || usageToProcess.CacheCreation.Ephemeral1hInputTokens > 0 { | ||
| if usage.PromptTokensDetails.CachedWriteTokenDetails == nil { | ||
| usage.PromptTokensDetails.CachedWriteTokenDetails = &schemas.ChatCachedWriteTokenDetails{} | ||
| } | ||
| if usageToProcess.CacheCreation.Ephemeral5mInputTokens > usage.PromptTokensDetails.CachedWriteTokenDetails.CachedWriteTokens5m { | ||
| usage.PromptTokensDetails.CachedWriteTokenDetails.CachedWriteTokens5m = usageToProcess.CacheCreation.Ephemeral5mInputTokens | ||
| } | ||
| if usageToProcess.CacheCreation.Ephemeral1hInputTokens > usage.PromptTokensDetails.CachedWriteTokenDetails.CachedWriteTokens1h { | ||
| usage.PromptTokensDetails.CachedWriteTokenDetails.CachedWriteTokens1h = usageToProcess.CacheCreation.Ephemeral1hInputTokens | ||
| } |
There was a problem hiding this comment.
The delta branch overwrites the authoritative cache totals but only max-keeps the 5m and 1h details. When message_start has cache-creation details and message_delta clears or omits them, the final usage can report CachedWriteTokens == 0 while retaining nonzero detail counts, producing inconsistent billing and telemetry.
| usage.PromptTokens = usageToProcess.InputTokens + usageToProcess.CacheReadInputTokens + usageToProcess.CacheCreationInputTokens | ||
| // OutputTokens on message_delta is the cumulative output count. | ||
| if usageToProcess.OutputTokens > usage.CompletionTokens { | ||
| usage.CompletionTokens = usageToProcess.OutputTokens | ||
| } | ||
| // Impossible-zero guard: restore start-side when a non-conformant | ||
| // provider emits a partial-usage delta (input=0 + cache fields omitted). | ||
| if usage.PromptTokens == 0 && startSnapshot != nil && startSnapshot.InputTokens > 0 { | ||
| usage.PromptTokens = startSnapshot.InputTokens | ||
| usage.PromptTokensDetails.CachedReadTokens = startSnapshot.CacheReadInputTokens | ||
| usage.PromptTokensDetails.CachedWriteTokens = startSnapshot.CacheCreationInputTokens | ||
| } | ||
| calculatedTotal := usage.PromptTokens + usage.CompletionTokens | ||
| if calculatedTotal > usage.TotalTokens { | ||
| usage.TotalTokens = calculatedTotal | ||
| } | ||
| return true |
There was a problem hiding this comment.
The fallback only detects an all-zero reconstructed prompt. If a compatible provider emits a partial delta with positive input_tokens but omits cache fields, this branch records the incomplete total and sets deltaProcessed, so normalization is skipped; for example, a 650-token start followed by {input_tokens: 500} is billed as 500 tokens.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@core/providers/anthropic/anthropic.go`:
- Around line 682-691: Update the cache-creation handling in the Anthropic usage
processing flow to replace the entire CachedWriteTokenDetails breakdown when
authoritative message_delta values are present, rather than merging with max
guards. Preserve the existing initialization path, but assign both
CachedWriteTokens5m and CachedWriteTokens1h directly from
usageToProcess.CacheCreation so stale message_start values, including nonzero
values when the delta reports zero, are overwritten.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 3320d7ee-f29d-4220-b760-1e1dbc8dd95f
📒 Files selected for processing (3)
core/changelog.mdcore/providers/anthropic/anthropic.gocore/providers/anthropic/streamcacheusage_test.go
| if usageToProcess.CacheCreation.Ephemeral5mInputTokens > 0 || usageToProcess.CacheCreation.Ephemeral1hInputTokens > 0 { | ||
| if usage.PromptTokensDetails.CachedWriteTokenDetails == nil { | ||
| usage.PromptTokensDetails.CachedWriteTokenDetails = &schemas.ChatCachedWriteTokenDetails{} | ||
| } | ||
| if usageToProcess.CacheCreation.Ephemeral5mInputTokens > usage.PromptTokensDetails.CachedWriteTokenDetails.CachedWriteTokens5m { | ||
| usage.PromptTokensDetails.CachedWriteTokenDetails.CachedWriteTokens5m = usageToProcess.CacheCreation.Ephemeral5mInputTokens | ||
| } | ||
| if usageToProcess.CacheCreation.Ephemeral1hInputTokens > usage.PromptTokensDetails.CachedWriteTokenDetails.CachedWriteTokens1h { | ||
| usage.PromptTokensDetails.CachedWriteTokenDetails.CachedWriteTokens1h = usageToProcess.CacheCreation.Ephemeral1hInputTokens | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Overwrite cache-write TTL details on message_delta.
These max guards retain preliminary message_start values when the authoritative delta reports smaller or zero TTL counts. Replace the nested breakdown wholesale alongside CachedWriteTokens; otherwise downstream usage contains stale cache details.
Proposed fix
- if usageToProcess.CacheCreation.Ephemeral5mInputTokens > 0 || usageToProcess.CacheCreation.Ephemeral1hInputTokens > 0 {
+ if usageToProcess.CacheCreation.Ephemeral5mInputTokens == 0 &&
+ usageToProcess.CacheCreation.Ephemeral1hInputTokens == 0 {
+ usage.PromptTokensDetails.CachedWriteTokenDetails = nil
+ } else {
if usage.PromptTokensDetails.CachedWriteTokenDetails == nil {
usage.PromptTokensDetails.CachedWriteTokenDetails = &schemas.ChatCachedWriteTokenDetails{}
}
- if usageToProcess.CacheCreation.Ephemeral5mInputTokens > usage.PromptTokensDetails.CachedWriteTokenDetails.CachedWriteTokens5m {
- usage.PromptTokensDetails.CachedWriteTokenDetails.CachedWriteTokens5m = usageToProcess.CacheCreation.Ephemeral5mInputTokens
- }
- if usageToProcess.CacheCreation.Ephemeral1hInputTokens > usage.PromptTokensDetails.CachedWriteTokenDetails.CachedWriteTokens1h {
- usage.PromptTokensDetails.CachedWriteTokenDetails.CachedWriteTokens1h = usageToProcess.CacheCreation.Ephemeral1hInputTokens
- }
+ usage.PromptTokensDetails.CachedWriteTokenDetails.CachedWriteTokens5m = usageToProcess.CacheCreation.Ephemeral5mInputTokens
+ usage.PromptTokensDetails.CachedWriteTokenDetails.CachedWriteTokens1h = usageToProcess.CacheCreation.Ephemeral1hInputTokens
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| if usageToProcess.CacheCreation.Ephemeral5mInputTokens > 0 || usageToProcess.CacheCreation.Ephemeral1hInputTokens > 0 { | |
| if usage.PromptTokensDetails.CachedWriteTokenDetails == nil { | |
| usage.PromptTokensDetails.CachedWriteTokenDetails = &schemas.ChatCachedWriteTokenDetails{} | |
| } | |
| if usageToProcess.CacheCreation.Ephemeral5mInputTokens > usage.PromptTokensDetails.CachedWriteTokenDetails.CachedWriteTokens5m { | |
| usage.PromptTokensDetails.CachedWriteTokenDetails.CachedWriteTokens5m = usageToProcess.CacheCreation.Ephemeral5mInputTokens | |
| } | |
| if usageToProcess.CacheCreation.Ephemeral1hInputTokens > usage.PromptTokensDetails.CachedWriteTokenDetails.CachedWriteTokens1h { | |
| usage.PromptTokensDetails.CachedWriteTokenDetails.CachedWriteTokens1h = usageToProcess.CacheCreation.Ephemeral1hInputTokens | |
| } | |
| if usageToProcess.CacheCreation.Ephemeral5mInputTokens == 0 && | |
| usageToProcess.CacheCreation.Ephemeral1hInputTokens == 0 { | |
| usage.PromptTokensDetails.CachedWriteTokenDetails = nil | |
| } else { | |
| if usage.PromptTokensDetails.CachedWriteTokenDetails == nil { | |
| usage.PromptTokensDetails.CachedWriteTokenDetails = &schemas.ChatCachedWriteTokenDetails{} | |
| } | |
| usage.PromptTokensDetails.CachedWriteTokenDetails.CachedWriteTokens5m = usageToProcess.CacheCreation.Ephemeral5mInputTokens | |
| usage.PromptTokensDetails.CachedWriteTokenDetails.CachedWriteTokens1h = usageToProcess.CacheCreation.Ephemeral1hInputTokens | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@core/providers/anthropic/anthropic.go` around lines 682 - 691, Update the
cache-creation handling in the Anthropic usage processing flow to replace the
entire CachedWriteTokenDetails breakdown when authoritative message_delta values
are present, rather than merging with max guards. Preserve the existing
initialization path, but assign both CachedWriteTokens5m and CachedWriteTokens1h
directly from usageToProcess.CacheCreation so stale message_start values,
including nonzero values when the delta reports zero, are overwritten.
|
@TejasGhatte have a look once |
|
Hey @is911 according to anthropic api input tokens field is cache exclusive. Also message delta usage counts are cumulative. |
244a01d to
ce1b2a6
Compare
|
Independent confirmation from a production deployment, plus data that I think bears directly on the open question in this thread.
Both halves of that are right — and the conclusion I'd draw is that Bifrost should be robust to it anyway, because this is not Kimi-specific. I hit the identical defect on QwenCloud, an unrelated vendor, and the two providers deviate in exactly the same way. Captured verbatim from // message_start — prompt-scale input_tokens, NO cache counters at all
{"type":"message_start","message":{"id":"msg_ba0b5ef1","model":"qwen3.7-plus","content":[],
"usage":{"input_tokens":8834,"output_tokens":0}}}
// message_delta — the authoritative, cache-aware split (NOT an increment)
{"type":"message_delta","delta":{"stop_reason":"end_turn"},
"usage":{"input_tokens":18,"output_tokens":426,"cache_creation_input_tokens":0,
"cache_read_input_tokens":8831,"cache_creation":{"ephemeral_5m_input_tokens":0}}}The non-streaming response for the same request agrees with For contrast, the compliant shape — MiniMax on the same code path, whose {"type":"message_start","message":{"usage":{"input_tokens":0,"output_tokens":0,"service_tier":"standard"}}}
{"type":"message_delta","usage":{"cache_read_input_tokens":8990,"input_tokens":1,"output_tokens":2}}So the trigger condition is precise and provider-shaped: On the undercount concern raised by the bot review
That is a real risk with "message_delta overwrites unconditionally". The variant I deployed avoids it by staying monotone — it never lowers a value — while still preferring the cache-aware number. An event is authoritative for // core/providers/anthropic/anthropic.go — accumulateAnthropicResponsesUsage
authoritative := usageToProcess.CacheReadInputTokens > 0 || usageToProcess.CacheCreationInputTokens > 0
authSeen := usage.InputTokensDetails != nil &&
(usage.InputTokensDetails.CachedReadTokens > 0 || usage.InputTokensDetails.CachedWriteTokens > 0)
if authoritative || (!authSeen && usageToProcess.InputTokens > usage.InputTokens) {
usage.InputTokens = usageToProcess.InputTokens
if billedUsage != nil {
billedUsage.PromptTokens = usageToProcess.InputTokens
}
}Compliant upstreams are provably unaffected: real Anthropic puts the cache counters on Production verificationDeployed to a live v1.6.11 gateway. Identical bytes,
Organic traffic at ~300K-token prompts went from One thing this PR does not coverThere is a second accumulator with the same flaw: Happy to open a PR for the passthrough half, or to rebase this one — it has been merge-conflicted since 08-19. |
…ly frame as cache-aware Review follow-ups on maximhq#6378. The type doc claimed a plain per-field max and parity with the native accumulator. Neither is true: input_tokens is no longer a plain max, and accumulateAnthropicResponsesUsage (anthropic.go) still takes an unconditional max and therefore still carries the double-count this type fixes. Point at maximhq#5354 / maximhq#5355 instead of claiming parity. "nothing here ever lowers a value" described the accumulators, not the emitted field. combined.InputTokens does drop, from the loose figure to the authoritative one, on the event where authSeen flips. That is safe because StreamPassthrough keeps the LAST non-nil observation and the authoritative message_delta is last — but the comment should say so rather than assert a monotonicity the emitted field does not have. Also: authority was decided on the two top-level cache counters only, so a frame reporting cache creation solely as the ephemeral 5m/1h breakdown was misread as cache-less and the prompt-scale input_tokens was kept. cache_creation is documented as the split OF cache_creation_input_tokens, so such a frame is cache-aware. Add it to the value test; the breakdown was already max-merged into the combined usage, so no other change is needed. Two tests added, both order-independent like the rest of the table: the breakdown-only frame (fails at 50000, passes at 1000), and a pin for the case where a cache-less event with a larger input_tokens follows an authoritative one (8931, where an unconditional max reports 9181) — no captured frame triggers that today, so the intent is pinned explicitly rather than left incidental. Refs maximhq#5510
4 participants
Summary
Fixes a ~2× prompt-token inflation in Anthropic-protocol streaming usage accounting. The streaming accumulator in
HandleAnthropicChatCompletionStreamingused amax-keepguard oninput_tokensthat did not branch on event type, so the authoritativemessage_delta.input_tokens(the uncached tail, smaller) was rejected in favor of the stalemessage_start.input_tokens(the full prompt). ThennormalizeCachedUsagefoldedcache_read_input_tokenson top at stream end →message_start.input_tokens + message_delta.cache_read_input_tokens→ double-count on every cached request from turn 2 onward.Closes #5354.
One-line root cause: the accumulator merged the two Anthropic usage snapshots via
max-keepinstead of treatingmessage_delta.usageas authoritative, then foldedcache_readon top of an already-inclusive prompt total.Before / after formula:
prompt_tokens = message_start.input_tokens + message_delta.cache_read_input_tokensprompt_tokens = message_delta.input_tokens + message_delta.cache_creation_input_tokens + message_delta.cache_read_input_tokensEvidence (3 consecutive turns, same session, provider: bifrost-kimi-anthropic / kimi-code/k3)
message_start(input/cache_read)message_delta(input/cache_read)Turn 1 was correct only because no cache read occurred (delta.cache_read == 0, so the buggy sum reduced to start.input). From turn 2 on, the buggy accumulator double-counted the cached portion. Reproduction signature: first request accurate, every subsequent request ~2×, growing by ~2× the real per-turn conversation growth.
Changes
core/providers/anthropic/anthropic.go— Extracted the event-loop usage-accumulation block into a new helperapplyStreamUsageEvent(usage, eventType, usageToProcess, startSnapshot)that branches on the Anthropic stream event type:message_delta: event-level unconditional overwrite ofInputTokens,CacheCreationInputTokens,CacheReadInputTokensfrom delta (NOT per-fieldif != 0checks — those would conflate absent-with-zero for non-pointer Gointfields); reconstructPromptTokens = delta.InputTokens + delta.CacheCreationInputTokens + delta.CacheReadInputTokens.message_start(or any non-delta event carrying usage): preserves existingmax-keepas the fallback for early-terminated streams.PromptTokens == 0 && startSnapshot.InputTokens > 0→ restore start-side values) for non-conformant Anthropic-compatible providers that emit a partial-usagemessage_delta(input=0 with cache fields omitted, deserializing to all-zeros). Dead code for spec-conformant providers — a real Anthropic prompt always hasPromptTokens >= 1.normalizeCachedUsagefold is skipped whenmessage_deltawas processed (deltaProcessedflag) — the cache breakdown is already folded intoPromptTokensby the overwrite.core/providers/anthropic/streamcacheusage_test.go(new) — Three fixtures (see How to test).core/changelog.md— Entry at the top per contributing guide.Design decision / trade-off
The fix uses an event-level overwrite rather than per-field
if != 0merge. Per-field checks would conflate absent-with-zero for non-pointer Gointfields (JSON deserialization of an omitted field produces 0), so a delta that legitimately omitscache_readwould be indistinguishable from a delta that reportscache_read=0. The event-level overwrite (with the impossible-zero guard as a safety net) is the only spec-correct option for the non-pointer field shape.Type of change
Affected areas
How to test
New fixtures:
TestAnthropicStreamCacheReadNotDoubleCounted— the primary regression guard. Turn-2 stream:message_startinput=20532/cache_read=0 →message_deltainput=308/cache_read=20224/cache_creation=0. Assertsprompt_tokens == 20532(i.e., 308 + 0 + 20224). Pre-fix this test FAILS recording 40756 (proves the test bites); post-fix PASSES recording 20532.TestAnthropicStreamNoCacheReadUnchanged— turn-1 non-regression (no cache).message_startinput=20458 →message_deltainput=20458/cache_read=0. Assertsprompt_tokens == 20458. Passes both pre- and post-fix.TestAnthropicStreamUsage_NonConformantDeltaOmitsCacheFields— proves the impossible-zero guard is load-bearing.message_startinput=20532/cache_read=20224 →message_deltainput=0 with cache fields omitted (→ deserialize to 0). Assertsprompt_tokens == 20532(start-side restored by the guard), NOT 0.Existing tests:
go test ./core/providers/anthropic/...shows zero pass/fail diff vsdevbaseline.Screenshots/Recordings
N/A (no UI changes).
Breaking changes
Behavior change:
prompt_tokensfor Anthropic-protocol streams with prompt caching will be ~50% lower than before (correct instead of ~2× inflated). Any downstream consumer that calibrated against the inflated values (cost baselines, governance budgets, telemetry dashboards) will see a one-time step-change to the correct values. The corrected values match the Anthropic Messages API spec.Related issues
Closes #5354
Security considerations
None. This is a usage-accounting fix; no auth, secrets, PII, or sandboxing surfaces are touched. The corrected
prompt_tokensflows into existing cost/governance/telemetry paths unchanged.Checklist
docs/contributing/README.mdand followed the guidelines (commit format[fix]:, affected packages listed,core/changelog.mdupdated)go build ./...,go vet ./...greengo test ./core/providers/anthropic/...greenFollow-up flags (out of scope for this PR)
HandleAnthropicResponsesStream+accumulateAnthropicResponsesUsage) uses a separate accumulator that was NOT audited for the same bug class. Worth a parallel audit if the Responses API exhibits similar double-counting.normalizeCachedUsagehas 0 matches incore/providers/openai/). No fix needed.core/providers/bedrock/bedrock.go) has its ownnormalizeCachedUsage— not audited; recommend a separate diagnostic if Bedrock routes Anthropic-protocol streams.cache_creation_input_tokensis consistently 0 from the kimi-code provider — upstream provider quirk, not a Bifrost bug. Not fixed.Known limitations (provider conformance assumption)
The fix assumes Anthropic-compatible providers either (a) repeat cache fields at
message_deltaper the Anthropic Messages API spec, or (b) omit the entiremessage_delta.usage(delta-absent fallback). Non-conformant providers that emit a PARTIAL-usagemessage_delta(input_tokens=0 with cache fields omitted — deserializing to all-zeros for non-pointer Goint) are handled by the impossible-zero guard. The guard is dead code for spec-conformant providers (zero behavior change). The guard does not log when it fires (the accumulator is currently logger-free; a structured-log line for non-conformance detection is a possible future enhancement).