fix(anthropic): map responses cached tokens to cache reads

litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py

@@ -7,6 +7,9 @@
 
 from litellm import verbose_logger
 from litellm._uuid import uuid
+from litellm.llms.anthropic.experimental_pass_through.responses_adapters.transformation import (
+    translate_responses_usage_to_anthropic_usage,
+)
 
 
 class AnthropicResponsesStreamWrapper:
@@ -226,24 +229,20 @@ def _process_event(self, event: Any) -> None:
                 event.get("response") if isinstance(event, dict) else None
             )
             stop_reason = "end_turn"
-            input_tokens = 0
-            output_tokens = 0
-            cache_creation_tokens = 0
-            cache_read_tokens = 0
+            anthropic_usage = {
+                "input_tokens": 0,
+                "output_tokens": 0,
+                "cache_creation_input_tokens": 0,
+                "cache_read_input_tokens": 0,
+            }
 
             if response_obj is not None:
                 status = getattr(response_obj, "status", None)
                 if status == "incomplete":
                     stop_reason = "max_tokens"
                 usage = getattr(response_obj, "usage", None)
                 if usage is not None:
-                    input_tokens = getattr(usage, "input_tokens", 0) or 0
-                    output_tokens = getattr(usage, "output_tokens", 0) or 0
-                    cache_creation_tokens = getattr(usage, "input_tokens_details", None)  # type: ignore[assignment]
-                    cache_read_tokens = getattr(usage, "output_tokens_details", None)  # type: ignore[assignment]
-                    # Prefer direct cache fields if present
-                    cache_creation_tokens = int(getattr(usage, "cache_creation_input_tokens", 0) or 0)
-                    cache_read_tokens = int(getattr(usage, "cache_read_input_tokens", 0) or 0)
+                    anthropic_usage = translate_responses_usage_to_anthropic_usage(usage)
 
             # Check if tool_use was in the output to override stop_reason
             if response_obj is not None:
@@ -257,13 +256,13 @@ def _process_event(self, event: Any) -> None:
                         break
 
             usage_delta: Dict[str, Any] = {
-                "input_tokens": input_tokens,
-                "output_tokens": output_tokens,
+                "input_tokens": anthropic_usage["input_tokens"],
+                "output_tokens": anthropic_usage["output_tokens"],
             }
-            if cache_creation_tokens:
-                usage_delta["cache_creation_input_tokens"] = cache_creation_tokens
-            if cache_read_tokens:
-                usage_delta["cache_read_input_tokens"] = cache_read_tokens
+            if anthropic_usage["cache_creation_input_tokens"]:
+                usage_delta["cache_creation_input_tokens"] = anthropic_usage["cache_creation_input_tokens"]
+            if anthropic_usage["cache_read_input_tokens"]:
+                usage_delta["cache_read_input_tokens"] = anthropic_usage["cache_read_input_tokens"]
 
             self._chunk_queue.append(
                 {

litellm/llms/anthropic/experimental_pass_through/responses_adapters/transformation.py

@@ -6,6 +6,7 @@
 """
 
 import json
+from numbers import Number
 from typing import Any, Dict, List, Optional, Union, cast
 
 from litellm.litellm_core_utils.reasoning_effort_utils import (
@@ -32,6 +33,50 @@
 from litellm.types.llms.openai import ResponsesAPIResponse
 
 
+def _get_usage_value(usage: Any, key: str) -> Any:
+    if isinstance(usage, dict):
+        return usage.get(key)
+    return getattr(usage, key, None)
+
+
+def _get_int_usage_value(usage: Any, key: str) -> int:
+    value = _get_usage_value(usage, key)
+    if isinstance(value, Number):
+        return int(value)
+    if isinstance(value, str) and value.strip().isdigit():
+        return int(value)
+    return 0
+
+
+def _get_responses_cached_tokens(usage: Any) -> int:
+    input_tokens_details = _get_usage_value(usage, "input_tokens_details")
+    if input_tokens_details is None:
+        return 0
+    return _get_int_usage_value(input_tokens_details, "cached_tokens")
+
+
+def translate_responses_usage_to_anthropic_usage(usage: Any) -> AnthropicUsage:
+    input_tokens = _get_int_usage_value(usage, "input_tokens")
+    output_tokens = _get_int_usage_value(usage, "output_tokens")
+    cache_creation_tokens = _get_int_usage_value(usage, "cache_creation_input_tokens")
+
+    responses_cached_tokens = _get_responses_cached_tokens(usage)
+    if responses_cached_tokens:
+        cache_read_tokens = responses_cached_tokens
+        # OpenAI Responses input_tokens includes cached tokens. Anthropic Messages
+        # input_tokens excludes cache-read tokens, so only subtract for this source.
+        input_tokens = max(0, input_tokens - responses_cached_tokens)
+    else:
+        cache_read_tokens = _get_int_usage_value(usage, "cache_read_input_tokens")
+
+    return AnthropicUsage(
+        input_tokens=input_tokens,
+        output_tokens=output_tokens,
+        cache_creation_input_tokens=cache_creation_tokens,
+        cache_read_input_tokens=cache_read_tokens,
+    )
+
+
 class LiteLLMAnthropicToResponsesAPIAdapter:
     """
     Converts Anthropic /v1/messages requests to OpenAI Responses API format and
@@ -396,8 +441,6 @@ def translate_response(
             ResponseReasoningItem,
         )
 
-        from litellm.types.llms.openai import ResponseAPIUsage
-
         content: List[Dict[str, Any]] = []
         stop_reason: AnthropicFinishReason = "end_turn"
 
@@ -463,15 +506,7 @@ def translate_response(
         if response.status == "incomplete":
             stop_reason = "max_tokens"
 
-        # usage
-        raw_usage: Optional[ResponseAPIUsage] = response.usage
-        input_tokens = int(getattr(raw_usage, "input_tokens", 0) or 0)
-        output_tokens = int(getattr(raw_usage, "output_tokens", 0) or 0)
-
-        anthropic_usage = AnthropicUsage(
-            input_tokens=input_tokens,
-            output_tokens=output_tokens,
-        )
+        anthropic_usage = translate_responses_usage_to_anthropic_usage(response.usage)
 
         return AnthropicMessagesResponse(
             id=response.id,

tests/e2e/proxy_client.py

@@ -75,12 +75,130 @@
 
 RowsPredicate = Callable[[list[SpendLogRow]], bool]
 
+# After /model/new, the control-plane writer reloads itself immediately, but every
+# other gateway worker (and peer pod) only picks the model up on its add_deployment
+# job. That job runs every proxy_config_reload_interval_seconds (product default 30).
+# A single /v1/models hit can land on a hot worker while the next /chat hits a cold
+# one ("Invalid model name"). Wait for first listing within MODEL_SERVABLE_TIMEOUT,
+# then require continuous listing for MODEL_SERVABLE_DB_SYNC_SECONDS (the default
+# reload interval) so every worker has had a chance to sync from the DB.
+MODEL_SERVABLE_TIMEOUT = 40.0
+MODEL_SERVABLE_DB_SYNC_SECONDS = 30.0
+MODEL_SERVABLE_INTERVAL = 2.0
+# Cap each /v1/models poll so one slow request cannot outlast the remaining budget.
+MODEL_SERVABLE_REQUEST_TIMEOUT = 5.0
+
+
+@dataclass(frozen=True, slots=True)
+class Servable:
+    """The data plane listed the model within the deadline."""
+
+
+@dataclass(frozen=True, slots=True)
+class NotServable:
+    """The deadline passed without the data plane listing the model.
+
+    `last_result` is the final /v1/models read, so the caller can tell "the proxy
+    answered but omitted the model" (propagation) from "the read itself failed"
+    (network/auth) when reporting."""
+
+    last_result: Result[ModelsListResponse] | None
+
+
+ServableOutcome = Servable | NotServable
+
+
+def await_servable(
+    list_models: Callable[[float], Result[ModelsListResponse]],
+    *,
+    model_name: str,
+    timeout: float,
+    interval: float,
+    request_timeout: float,
+    db_sync_seconds: float,
+    now: Callable[[], float],
+    sleep: Callable[[float], None],
+) -> ServableOutcome:
+    """Poll until `model_name` is listed long enough for every worker to DB-sync.
+
+    First listing must happen within `timeout`. After that, the model must stay
+    listed continuously for `db_sync_seconds` (any miss resets the continuous
+    window). `db_sync_seconds=0` returns on the first listing. Each poll's request
+    timeout is clamped to the remaining budget. Sleeps only min(interval, time left)
+    so a final deadline-clamped poll is never skipped just because a full interval
+    does not fit. Clock and sleep are injected."""
+    started = now()
+    first_seen_at: float | None = None
+    last_result: Result[ModelsListResponse] | None = None
+    while True:
+        t = now()
+        phase_deadline = (
+            started + timeout if first_seen_at is None else first_seen_at + db_sync_seconds
+        )
+        remaining = phase_deadline - t
+        if remaining <= 0:
+            if (
+                last_result is not None
+                and first_seen_at is not None
+                and (db_sync_seconds <= 0 or t - first_seen_at >= db_sync_seconds)
+            ):
+                return Servable()
+            return NotServable(last_result=last_result)
+
+        poll_timeout = min(request_timeout, remaining)
+        last_result = list_models(poll_timeout)
+        listed = isinstance(last_result, Success) and any(
+            entry.id == model_name for entry in last_result.data.data
+        )
+        t = now()
+        if not listed:
+            first_seen_at = None
+        elif first_seen_at is None:
+            if t > started + timeout:
+                return NotServable(last_result=last_result)
+            first_seen_at = t
+            if db_sync_seconds <= 0:
+                return Servable()
+        elif t - first_seen_at >= db_sync_seconds:
+            return Servable()
+
+        phase_deadline = (
+            started + timeout if first_seen_at is None else first_seen_at + db_sync_seconds
+        )
+        wait = min(interval, phase_deadline - now())
+        if wait > 0:
+            sleep(wait)
+
+
+def servable_timeout_message(
+    *,
+    model_name: str,
+    timeout: float,
+    db_sync_seconds: float,
+    last_result: Result[ModelsListResponse] | None,
+) -> str:
+    last_error = (
+        f"; last /v1/models poll did not succeed: {last_result}"
+        if last_result is not None and not isinstance(last_result, Success)
+        else ""
+    )
+    return (
+        f"model {model_name!r} was created but never became servable on the data "
+        f"plane within {timeout}s of first listing (plus {db_sync_seconds}s continuous "
+        f"DB sync) after /model/new (control/data-plane propagation or "
+        f"STORE_MODEL_IN_DB reload issue){last_error}"
+    )
+
 
 @dataclass(frozen=True, slots=True)
 class ProxyClient:
     transport: Transport
     poll_timeout: float = 120.0
     poll_interval: float = 5.0
+    model_servable_timeout: float = MODEL_SERVABLE_TIMEOUT
+    model_servable_db_sync_seconds: float = MODEL_SERVABLE_DB_SYNC_SECONDS
+    model_servable_interval: float = MODEL_SERVABLE_INTERVAL
+    model_servable_request_timeout: float = MODEL_SERVABLE_REQUEST_TIMEOUT
 
     # ---- keys / customers (satisfies lifecycle.ResourceClient) ----------
 
@@ -167,7 +285,12 @@ def create_model(
         this returns can race the reload and 400 with "Invalid model name passed".
         We therefore poll the data-plane /v1/models until the model appears before
         handing back, so callers can invoke it immediately. In the monolithic case
-        it is already present on the first poll, so this adds one request."""
+        it is already present on the first poll, so this adds one request.
+
+        First listing must arrive within `model_servable_timeout` (not the longer
+        spend `poll_timeout`). The model must then stay listed for
+        `model_servable_db_sync_seconds` (product default DB reload interval) so every
+        gateway worker has run add_deployment before callers use the model."""
         model_id = unwrap(
             self.transport.post(
                 "/model/new",
@@ -184,33 +307,38 @@ def create_model(
         return model_id
 
     def _await_model_servable(self, model_name: str) -> None:
-        """Block until the data plane lists `model_name`, or fail loudly if it does
-        not within poll_timeout (a real propagation/config problem, surfaced here
-        instead of as a downstream "Invalid model name passed")."""
-        deadline = time.monotonic() + self.poll_timeout
-        last_result: Result[ModelsListResponse] | None = None
-        while time.monotonic() < deadline:
-            last_result = self.transport.get(
+        """Block until the data plane lists `model_name` long enough for DB sync.
+
+        Fails if first listing misses model_servable_timeout, or if continuous listing
+        for model_servable_db_sync_seconds never holds (multi-worker / peer reload)."""
+        outcome = await_servable(
+            lambda poll_timeout: self.transport.get(
                 "/v1/models",
                 headers=self.transport.master,
                 params=NoBody(),
                 response_type=ModelsListResponse,
-            )
-            if isinstance(last_result, Success) and any(
-                entry.id == model_name for entry in last_result.data.data
-            ):
-                return
-            time.sleep(self.poll_interval)
-        last_error = (
-            f"; last /v1/models poll did not succeed: {last_result}"
-            if last_result is not None and not isinstance(last_result, Success)
-            else ""
-        )
-        raise AssertionError(
-            f"model {model_name!r} was created but never became servable on the data "
-            f"plane within {self.poll_timeout}s of /model/new (control/data-plane "
-            f"propagation or STORE_MODEL_IN_DB reload issue){last_error}"
+                timeout=poll_timeout,
+            ),
+            model_name=model_name,
+            timeout=self.model_servable_timeout,
+            interval=self.model_servable_interval,
+            request_timeout=self.model_servable_request_timeout,
+            db_sync_seconds=self.model_servable_db_sync_seconds,
+            now=time.monotonic,
+            sleep=time.sleep,
         )
+        match outcome:
+            case Servable():
+                return
+            case NotServable(last_result=last_result):
+                raise AssertionError(
+                    servable_timeout_message(
+                        model_name=model_name,
+                        timeout=self.model_servable_timeout,
+                        db_sync_seconds=self.model_servable_db_sync_seconds,
+                        last_result=last_result,
+                    )
+                )
 
     def update_model(self, model_id: str, litellm_params: LiteLLMParamsBody) -> None:
         """Merge `litellm_params` over the deployment `model_id`'s stored params via

tests/e2e/transport.py

@@ -58,6 +58,7 @@ def get[R: BaseModel](
         headers: BaseModel,
         params: BaseModel,
         response_type: type[R],
+        timeout: float | None = None,
     ) -> Result[R]: ...
 
     def delete[R: BaseModel](
@@ -136,13 +137,16 @@ def get[R: BaseModel](
         headers: BaseModel,
         params: BaseModel,
         response_type: type[R],
+        timeout: float | None = None,
     ) -> Result[R]:
+        """`timeout` overrides the transport-wide request_timeout for this call, for
+        pollers whose own deadline is shorter than it."""
         return e2e_http.get(
             self._url(path),
             headers=headers,
             params=params,
             response_type=response_type,
-            timeout=self.request_timeout,
+            timeout=self.request_timeout if timeout is None else timeout,
         )
 
     def delete[R: BaseModel](
@@ -336,9 +340,14 @@ def get[R: BaseModel](
         headers: BaseModel,
         params: BaseModel,
         response_type: type[R],
+        timeout: float | None = None,
     ) -> Result[R]:
         return self._route(path).get(
-            path, headers=headers, params=params, response_type=response_type
+            path,
+            headers=headers,
+            params=params,
+            response_type=response_type,
+            timeout=timeout,
         )
 
     def delete[R: BaseModel](