fix(protocols): preserve native cache usage counters

app/protocols/anthropic.py

@@ -37,6 +37,7 @@
 
 from app.protocols import clamp_stop_sequences, require_text
 from app.protocols.sse import finish_quietly
+from app.protocols.usage import cache_components
 
 logger = structlog.get_logger()
 
@@ -409,6 +410,17 @@ def _message_id(raw: str | None) -> str:
     return f"msg_{raw or uuid.uuid4().hex}"
 
 
+def _anthropic_usage(usage: dict) -> dict:
+    cache = cache_components(usage)
+    prompt = usage.get("prompt_tokens")
+    prompt = prompt if type(prompt) is int and prompt >= 0 else 0
+    return {
+        "input_tokens": prompt - sum(cache.values()),
+        "output_tokens": usage.get("completion_tokens", 0) or 0,
+        **cache,
+    }
+
+
 def to_anthropic_response(resp: dict) -> dict:
     choices = resp.get("choices") or [{}]
     choice = choices[0] if choices else {}
@@ -436,10 +448,7 @@ def to_anthropic_response(resp: dict) -> dict:
         "content": content,
         "stop_reason": _STOP_REASON_MAP.get(choice.get("finish_reason"), "end_turn"),
         "stop_sequence": None,
-        "usage": {
-            "input_tokens": usage.get("prompt_tokens", 0) or 0,
-            "output_tokens": usage.get("completion_tokens", 0) or 0,
-        },
+        "usage": _anthropic_usage(usage),
     }
 
 
@@ -690,9 +699,10 @@ def _tool_block_events(*, only_complete: bool = False) -> list[str]:
             yield _block_stop()
         for ev in _tool_block_events():
             yield ev
-        delta_usage: dict = {"output_tokens": usage.get("completion_tokens", 0) or 0}
-        if usage.get("prompt_tokens"):
-            delta_usage["input_tokens"] = usage["prompt_tokens"]
+        delta_usage = _anthropic_usage(usage)
+        if usage.get("prompt_tokens") is None:
+            # No measured count: do not overwrite message_start's estimate.
+            delta_usage.pop("input_tokens")
         yield _ev("message_delta", {
             "type": "message_delta",
             "delta": {"stop_reason": stop_reason or "end_turn", "stop_sequence": None},

app/protocols/gemini.py

@@ -35,6 +35,7 @@
 from app.protocols import clamp_stop_sequences, require_text
 from app.protocols.anthropic import _args_complete, _parse_json_object, flatten_openai_content
 from app.protocols.sse import finish_quietly
+from app.protocols.usage import cache_components
 
 logger = structlog.get_logger()
 
@@ -500,13 +501,18 @@ def to_openai_request(req: GeminiGenerateRequest, *, model: str, stream: bool) -
 
 
 def _usage_metadata(usage: dict) -> dict:
-    prompt = usage.get("prompt_tokens", 0) or 0
+    prompt = usage.get("prompt_tokens")
+    prompt = prompt if type(prompt) is int and prompt >= 0 else 0
     completion = usage.get("completion_tokens", 0) or 0
-    return {
+    result = {
         "promptTokenCount": prompt,
         "candidatesTokenCount": completion,
         "totalTokenCount": usage.get("total_tokens") or (prompt + completion),
     }
+    cache = cache_components(usage)
+    if "cache_read_input_tokens" in cache:
+        result["cachedContentTokenCount"] = cache["cache_read_input_tokens"]
+    return result
 
 
 def to_gemini_response(resp: dict) -> dict:

app/protocols/usage.py

@@ -0,0 +1,30 @@
+"""Cache components of LiteLLM's inclusive prompt_tokens.
+
+Only observed nonnegative integer counters are projected. None/missing is
+unknown, not zero. Aliases describe the same tokens, so they must agree.
+An invalid or out-of-bounds breakdown is omitted as a whole; callers keep
+the inclusive input count rather than emit a fabricated negative remainder.
+"""
+
+
+def cache_components(usage: dict) -> dict[str, int]:
+    prompt = usage.get("prompt_tokens")
+    if type(prompt) is not int or prompt < 0:
+        return {}
+    details = usage.get("prompt_tokens_details")
+    if not isinstance(details, dict):
+        details = {}
+    components = {}
+    for target, aliases in (
+        ("cache_read_input_tokens", ("cached_tokens",)),
+        ("cache_creation_input_tokens", ("cache_write_tokens", "cache_creation_tokens", "cache_creation_input_tokens")),
+    ):
+        values = [source[key] for source, keys in ((usage, (target,)), (details, aliases))
+                  for key in keys if source.get(key) is not None]
+        if any(type(value) is not int or value < 0 for value in values):
+            return {}
+        if values:
+            if any(value != values[0] for value in values):
+                return {}
+            components[target] = values[0]
+    return components if sum(components.values()) <= prompt else {}

integrations/claude-code.md

@@ -53,6 +53,13 @@ Streaming (`client.messages.stream(...)`) and tool use work as usual.
 - Streaming `message_start.usage.input_tokens` is the same estimate
   `/v1/messages/count_tokens` returns — the upstream only reports the
   exact count at end-of-stream, where it is sent in `message_delta`.
+  Blocking and terminal streaming usage split LiteLLM's inclusive prompt
+  count into ordinary `input_tokens` and observed `cache_read_input_tokens` /
+  `cache_creation_input_tokens`, following [Anthropic's additive usage contract](https://platform.claude.com/docs/en/build-with-claude/prompt-caching#tracking-cache-performance).
+  Missing/null cache counters stay omitted; explicit zeros are preserved.
+  Negative, non-integer, conflicting aliases, or cache totals exceeding the
+  prompt count cause the cache breakdown to be omitted and the inclusive
+  count retained. An unusable prompt count falls back to zero.
 - Server-side tools (web search, computer use), PDF `document` blocks, and
   the Files API return 400. Custom function tools are fully supported.
 - `POST /v1/messages/count_tokens` returns an estimate

integrations/gemini-sdk.md

@@ -49,6 +49,12 @@ logs; the header form doesn't.
 
 ## Known limitations (v1)
 
+- Blocking and terminal streaming `usageMetadata.cachedContentTokenCount`
+  is included when upstream reports a valid cache-read count, including an
+  explicit zero. `promptTokenCount` remains inclusive, per the [Gemini usage contract](https://ai.google.dev/api/generate-content#UsageMetadata).
+  Missing/null cache counters stay omitted. Negative, non-integer,
+  conflicting aliases, or a cache breakdown exceeding the prompt count
+  is omitted; an unusable prompt count falls back to zero.
 - `candidateCount > 1`, `cachedContent`, Files API (`fileData`), and
   server-side tools (`googleSearch`, `codeExecution`, `urlContext`)
   return 400. Function declarations are fully supported (uppercase proto

tests/integration/test_anthropic_messages.py

@@ -182,6 +182,35 @@ async def test_invalid_key_renders_anthropic_envelope(native_client):
 # ── blocking ──
 
 
+@pytest.mark.parametrize("stream", [False, True])
+async def test_native_cache_usage_reaches_blocking_and_terminal_response(native_client, stream):
+    from litellm.llms.anthropic.chat.transformation import AnthropicConfig
+
+    client, fake, key = native_client
+    usage = AnthropicConfig().calculate_usage({
+        "input_tokens": 50, "output_tokens": 10,
+        "cache_read_input_tokens": 9000, "cache_creation_input_tokens": 1000,
+    }, reasoning_content=None).model_dump(exclude_none=True)
+
+    async def chunks():
+        for chunk in _stream_chunks():
+            if "usage" in chunk:
+                chunk["usage"] = usage
+            yield chunk
+
+    async def upstream(**kwargs):
+        if kwargs.get("stream"):
+            return chunks()
+        return {**_openai_response(), "usage": usage}
+
+    fake.acompletion.side_effect = upstream
+    response = await client.post("/v1/messages", json=_messages_payload(stream=stream), headers={"x-api-key": key})
+    assert response.status_code == 200
+    actual = _parse_anthropic_sse(response.text)[-2][1]["usage"] if stream else response.json()["usage"]
+    assert actual == {"input_tokens": 50, "output_tokens": 10,
+                      "cache_read_input_tokens": 9000, "cache_creation_input_tokens": 1000}
+
+
 async def test_blocking_response_shape_and_orca_headers(native_client):
     client, fake, key = native_client
     r = await client.post("/v1/messages", json=_messages_payload(),

tests/integration/test_gemini_generate.py

@@ -15,6 +15,36 @@
 import pytest
 
 
+@pytest.mark.parametrize("action", ["generateContent", "streamGenerateContent"])
+async def test_cached_usage_reaches_native_http_response(native_client, action):
+    from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import VertexGeminiConfig
+
+    client, fake, key = native_client
+    usage = VertexGeminiConfig.calculate_usage({"usageMetadata": {
+        "promptTokenCount": 100, "candidatesTokenCount": 2,
+        "totalTokenCount": 102, "cachedContentTokenCount": 90,
+    }}).model_dump(exclude_none=True)
+
+    async def chunks():
+        for chunk in _stream_chunks():
+            if "usage" in chunk:
+                chunk["usage"] = usage
+            yield chunk
+
+    async def upstream(**kwargs):
+        if kwargs.get("stream"):
+            return chunks()
+        return {**_openai_response(), "usage": usage}
+
+    fake.acompletion.side_effect = upstream
+    response = await client.post(f"/v1beta/models/gemini-1.5-flash:{action}",
+                                 json={"contents": "hi"}, headers={"x-goog-api-key": key})
+    assert response.status_code == 200
+    body = response.json()[-1] if action == "streamGenerateContent" else response.json()
+    assert body["usageMetadata"] == {"promptTokenCount": 100, "candidatesTokenCount": 2,
+                                     "totalTokenCount": 102, "cachedContentTokenCount": 90}
+
+
 def _openai_response(**kwargs) -> dict:
     return {
         "id": "chatcmpl-1",

tests/unit/test_native_cache_usage.py

@@ -0,0 +1,104 @@
+"""Native response usage must preserve the inclusive LiteLLM cache breakdown."""
+
+import json
+
+import pytest
+
+from app.protocols.anthropic import stream_events, to_anthropic_response
+from app.protocols.gemini import stream_chunks, to_gemini_response
+
+
+async def _frames(usage):
+    yield {"choices": [], "usage": usage}
+
+
+@pytest.mark.parametrize("details,cache", [
+    ({"cached_tokens": 90, "cache_write_tokens": 10},
+     {"cache_read_input_tokens": 90, "cache_creation_input_tokens": 10}),
+    ({"cached_tokens": 90, "cache_creation_tokens": 10},
+     {"cache_read_input_tokens": 90, "cache_creation_input_tokens": 10}),
+    ({"cached_tokens": 0}, {"cache_read_input_tokens": 0}),
+    ({"cache_write_tokens": 0}, {"cache_creation_input_tokens": 0}),
+    ({"cached_tokens": None, "cache_creation_tokens": None}, {}),
+    ({}, {}),
+])
+async def test_blocking_and_terminal_stream_preserve_known_cache_components(details, cache):
+    usage = {"prompt_tokens": 105, "completion_tokens": 2, "total_tokens": 107,
+             "prompt_tokens_details": details}
+    expected = {"input_tokens": 105 - sum(cache.values()), "output_tokens": 2, **cache}
+    assert to_anthropic_response({"usage": usage})["usage"] == expected
+    events = [json.loads(event.split("data: ")[1]) async for event in stream_events(_frames(usage))]
+    assert events[-2]["usage"] == expected
+
+    gemini = {"promptTokenCount": 105, "candidatesTokenCount": 2, "totalTokenCount": 107}
+    if "cache_read_input_tokens" in cache:
+        gemini["cachedContentTokenCount"] = cache["cache_read_input_tokens"]
+    assert to_gemini_response({"usage": usage})["usageMetadata"] == gemini
+    chunks = [chunk async for chunk in stream_chunks(_frames(usage))]
+    assert chunks[-1]["usageMetadata"] == gemini
+
+
+@pytest.mark.parametrize("details", [
+    {"cached_tokens": -1}, {"cached_tokens": 106},
+    {"cached_tokens": 100, "cache_write_tokens": 6},
+    {"cached_tokens": True}, {"cached_tokens": "90"},
+    {"cached_tokens": 90.5},
+    {"cache_write_tokens": 10, "cache_creation_tokens": 11},
+])
+async def test_inconsistent_cache_breakdown_is_omitted_without_negative_input(details):
+    usage = {"prompt_tokens": 105, "completion_tokens": 2, "prompt_tokens_details": details}
+    assert to_anthropic_response({"usage": usage})["usage"] == {"input_tokens": 105, "output_tokens": 2}
+    assert "cachedContentTokenCount" not in to_gemini_response({"usage": usage})["usageMetadata"]
+    events = [json.loads(event.split("data: ")[1]) async for event in stream_events(_frames(usage))]
+    assert events[-2]["usage"] == {"input_tokens": 105, "output_tokens": 2}
+
+
+async def test_known_zero_ordinary_input_replaces_stream_estimate():
+    usage = {"prompt_tokens": 90, "completion_tokens": 2,
+             "prompt_tokens_details": {"cached_tokens": 90}}
+    events = [json.loads(event.split("data: ")[1]) async for event in stream_events(_frames(usage), input_tokens=37)]
+    assert events[0]["message"]["usage"]["input_tokens"] == 37
+    assert events[-2]["usage"] == {"input_tokens": 0, "output_tokens": 2, "cache_read_input_tokens": 90}
+
+
+def test_real_anthropic_normalizer_aliases_are_not_counted_twice():
+    from litellm.llms.anthropic.chat.transformation import AnthropicConfig
+
+    usage = AnthropicConfig().calculate_usage({
+        "input_tokens": 50, "output_tokens": 10,
+        "cache_read_input_tokens": 9000, "cache_creation_input_tokens": 1000,
+    }, reasoning_content=None).model_dump(exclude_none=True)
+    assert usage["prompt_tokens"] == 10050
+    assert to_anthropic_response({"usage": usage})["usage"] == {
+        "input_tokens": 50, "output_tokens": 10,
+        "cache_read_input_tokens": 9000, "cache_creation_input_tokens": 1000,
+    }
+
+
+@pytest.mark.parametrize("usage", [
+    {"prompt_tokens": 10, "cache_read_input_tokens": 7},
+    {"prompt_tokens": 10, "cache_creation_input_tokens": 7},
+])
+def test_legacy_top_level_litellm_components(usage):
+    expected = {"input_tokens": 3, "output_tokens": 0}
+    expected.update({key: value for key, value in usage.items() if key.startswith("cache_")})
+    assert to_anthropic_response({"usage": usage})["usage"] == expected
+
+
+@pytest.mark.parametrize("prompt", [-1, None, "10"])
+def test_unusable_prompt_count_cannot_establish_a_cache_breakdown(prompt):
+    usage = {"prompt_tokens": prompt, "completion_tokens": 2,
+             "prompt_tokens_details": {"cached_tokens": 1}}
+    assert to_anthropic_response({"usage": usage})["usage"] == {"input_tokens": 0, "output_tokens": 2}
+    assert to_gemini_response({"usage": usage})["usageMetadata"] == {
+        "promptTokenCount": 0, "candidatesTokenCount": 2, "totalTokenCount": 2,
+    }
+
+
+def test_real_gemini_normalizer_preserves_inclusive_cached_count():
+    from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import VertexGeminiConfig
+
+    raw = {"promptTokenCount": 100, "candidatesTokenCount": 2,
+           "totalTokenCount": 102, "cachedContentTokenCount": 90}
+    usage = VertexGeminiConfig.calculate_usage({"usageMetadata": raw}).model_dump(exclude_none=True)
+    assert to_gemini_response({"usage": usage})["usageMetadata"] == raw