Merge pull request #25846 from BerriAI/litellm_bedrock_cache_start_negative_cost

fix(bedrock): prevent negative streaming costs for start-only cache usage

litellm/litellm_core_utils/llm_cost_calc/utils.py

@@ -684,6 +684,9 @@ def generic_cost_per_token(  # noqa: PLR0915
             - cache_creation
             - image_tokens
         )
+        # Clamp to zero: inconsistent streaming usage 
+        if text_tokens < 0:
+            text_tokens = 0
         prompt_tokens_details["text_tokens"] = text_tokens
 
     (

litellm/llms/bedrock/messages/invoke_transformations/anthropic_claude3_transformation.py

@@ -591,11 +591,14 @@ class AmazonAnthropicClaudeMessagesConfig(
         """
         Bedrock invoke does not return SSE formatted data. This function is a wrapper to ensure litellm chunks are SSE formatted.
 
-        Bedrock's Anthropic-compatible streaming puts cache usage fields
-        (cache_creation_input_tokens, cache_read_input_tokens) only on
-        message_stop, not on message_start or message_delta. Claude Code's
-        SDK only merges usage from message_delta, so we promote those fields
-        from message_stop onto message_delta before yielding.
+        Bedrock's Anthropic-compatible streaming usually puts cache usage fields
+        (cache_creation_input_tokens, cache_read_input_tokens) on message_stop.
+        Some deployments (including GovCloud) emit the cache breakdown only on
+        ``message_start.message.usage``; ``message_delta`` / ``message_stop`` then
+        repeat uncached ``input_tokens`` only. We promote cache fields from
+        ``message_stop`` onto ``message_delta``, and when those are absent we
+        merge them from ``message_start`` so logging/cost sees a consistent usage
+        object (fixes negative input costs: LIT-2411).
         """
         from litellm.llms.anthropic.experimental_pass_through.messages.streaming_iterator import (
             BaseAnthropicMessagesStreamingIterator,
@@ -611,6 +614,27 @@ class AmazonAnthropicClaudeMessagesConfig(
         async for chunk in handler.async_sse_wrapper(patched_stream):
             yield chunk
 
+    @staticmethod
+    def _merge_message_start_cache_into_delta_usage(
+        delta_usage: Dict[str, Any],
+        start_usage: Optional[Dict[str, Any]],
+    ) -> None:
+        """
+        Copy cache breakdown from message_start onto message_delta usage when
+        those keys are missing on the delta (GovCloud / some Bedrock streams).
+        """
+        if not start_usage:
+            return
+        for field in ("cache_creation_input_tokens", "cache_read_input_tokens"):
+            if field not in delta_usage:
+                val = start_usage.get(field)
+                if val is not None:
+                    delta_usage[field] = val
+        if "cache_creation" not in delta_usage:
+            cc = start_usage.get("cache_creation")
+            if cc is not None:
+                delta_usage["cache_creation"] = cc
+
     @staticmethod
     async def _promote_message_stop_usage(
         completion_stream: AsyncIterator[
@@ -618,20 +642,13 @@ class AmazonAnthropicClaudeMessagesConfig(
         ],
     ) -> AsyncIterator[Union[bytes, GenericStreamingChunk, ModelResponseStream, dict]]:
         """
-        Promote cache usage fields from message_stop onto message_delta.
-
-        Bedrock reports input_tokens (uncached only) on message_start, and
-        the full breakdown (input_tokens, cache_creation_input_tokens,
-        cache_read_input_tokens) only on message_stop. Claude Code's SDK
-        merges usage from message_start and message_delta but ignores
-        message_stop. This method buffers message_delta and, when
-        message_stop arrives with cache usage, merges those fields into the
-        message_delta usage. input_tokens is kept as the uncached-only
-        count; downstream calculate_usage adds cache tokens to
-        prompt_tokens.
+        Promote cache usage fields onto message_delta from message_stop (and,
+        when stop lacks them, from message_start).  Ensures the final usage
+        chunk that logging/cost sees is always self-consistent.
         """
         _CACHE_FIELDS = ("cache_creation_input_tokens", "cache_read_input_tokens")
-        pending_delta = None
+        pending_delta: Optional[Dict[str, Any]] = None
+        start_usage_snapshot: Optional[Dict[str, Any]] = None
 
         async for chunk in completion_stream:
             if not isinstance(chunk, dict):
@@ -643,8 +660,19 @@ class AmazonAnthropicClaudeMessagesConfig(
 
             chunk_type = chunk.get("type")
 
+            if chunk_type == "message_start":
+                msg: Dict[str, Any] = cast(Dict[str, Any], chunk.get("message") or {})
+                u = msg.get("usage")
+                if isinstance(u, dict):
+                    start_usage_snapshot = dict(u)
+                if pending_delta is not None:
+                    yield pending_delta
+                    pending_delta = None
+                yield chunk
+                continue
+
             if chunk_type == "message_delta":
-                pending_delta = chunk
+                pending_delta = cast(Dict[str, Any], chunk)
                 continue
 
             if chunk_type == "message_stop" and pending_delta is not None:
@@ -661,6 +689,10 @@ class AmazonAnthropicClaudeMessagesConfig(
                         raw_input if isinstance(raw_input, int) else 0
                     )
 
+                AmazonAnthropicClaudeMessagesConfig._merge_message_start_cache_into_delta_usage(
+                    delta_usage, start_usage_snapshot
+                )
+
                 if delta_usage:
                     pending_delta["usage"] = delta_usage  # type: ignore[arg-type]
 
@@ -676,6 +708,12 @@ class AmazonAnthropicClaudeMessagesConfig(
             yield chunk
 
         if pending_delta is not None:
+            delta_usage = dict(pending_delta.get("usage") or {})
+            AmazonAnthropicClaudeMessagesConfig._merge_message_start_cache_into_delta_usage(
+                delta_usage, start_usage_snapshot
+            )
+            if delta_usage:
+                pending_delta["usage"] = delta_usage  # type: ignore[arg-type]
             yield pending_delta
 
 

tests/test_litellm/llms/bedrock/messages/invoke_transformations/test_anthropic_claude3_transformation.py

@@ -618,6 +618,147 @@ async def test_promote_message_stop_usage_preserves_message_delta_output_tokens(
     assert delta_out["usage"]["input_tokens"] == 3
 
 
+@pytest.mark.asyncio
+async def test_promote_message_start_cache_when_message_stop_omits_cache_fields():
+    """
+    GovCloud / some Bedrock streams put cache_read only on message_start; delta and
+    stop repeat uncached input_tokens only. Merging start cache onto message_delta
+    avoids inconsistent usage and negative input costs (LIT-2411).
+    """
+    cfg = AmazonAnthropicClaudeMessagesConfig()
+
+    async def _stream():  # type: ignore[return-type]
+        yield {
+            "type": "message_start",
+            "message": {
+                "id": "msg_1",
+                "type": "message",
+                "role": "assistant",
+                "content": [],
+                "model": "claude-sonnet-4-5-20250929",
+                "usage": {
+                    "input_tokens": 10,
+                    "cache_creation_input_tokens": 0,
+                    "cache_read_input_tokens": 22167,
+                    "cache_creation": {
+                        "ephemeral_5m_input_tokens": 0,
+                        "ephemeral_1h_input_tokens": 0,
+                    },
+                    "output_tokens": 4,
+                },
+            },
+        }
+        yield {
+            "type": "message_delta",
+            "delta": {"stop_reason": "end_turn", "stop_sequence": None},
+            "usage": {"input_tokens": 10, "output_tokens": 181},
+        }
+        yield {"type": "message_stop", "usage": {"input_tokens": 10, "output_tokens": 181}}
+
+    merged: list[dict] = []
+    async for chunk in cfg._promote_message_stop_usage(_stream()):
+        if isinstance(chunk, dict):
+            merged.append(chunk)
+
+    delta_chunks = [c for c in merged if c.get("type") == "message_delta"]
+    assert len(delta_chunks) == 1
+    u = delta_chunks[0]["usage"]
+    assert u["input_tokens"] == 10
+    assert u["output_tokens"] == 181
+    assert u["cache_read_input_tokens"] == 22167
+    assert u["cache_creation_input_tokens"] == 0
+
+
+@pytest.mark.asyncio
+async def test_unified_bedrock_messages_cache_on_start_only_never_negative_cost():
+    """
+    Regression guard for LIT-2411:
+    If cache usage is present only on message_start (and omitted from
+    message_delta/message_stop), final reconstructed usage + cost must still
+    be consistent and non-negative.
+    """
+    from litellm import completion_cost
+    from litellm.proxy.pass_through_endpoints.llm_provider_handlers.anthropic_passthrough_logging_handler import (
+        AnthropicPassthroughLoggingHandler,
+    )
+
+    cfg = AmazonAnthropicClaudeMessagesConfig()
+
+    async def _stream():  # type: ignore[return-type]
+        yield {
+            "type": "message_start",
+            "message": {
+                "id": "msg_bdrk_01WuFzkDbE9KWgiWakMRNKcA",
+                "type": "message",
+                "role": "assistant",
+                "content": [],
+                "model": "claude-sonnet-4-5-20250929",
+                "stop_reason": None,
+                "stop_sequence": None,
+                "usage": {
+                    "input_tokens": 10,
+                    "cache_creation_input_tokens": 0,
+                    "cache_read_input_tokens": 22167,
+                    "cache_creation": {
+                        "ephemeral_5m_input_tokens": 0,
+                        "ephemeral_1h_input_tokens": 0,
+                    },
+                    "output_tokens": 4,
+                },
+            },
+        }
+        yield {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}
+        yield {
+            "type": "content_block_delta",
+            "index": 0,
+            "delta": {"type": "text_delta", "text": "Hello from regression test"},
+        }
+        yield {"type": "content_block_stop", "index": 0}
+        yield {
+            "type": "message_delta",
+            "delta": {"stop_reason": "end_turn", "stop_sequence": None},
+            "usage": {"output_tokens": 181, "input_tokens": 10},
+        }
+        yield {"type": "message_stop", "usage": {"input_tokens": 10, "output_tokens": 181}}
+
+    logging_obj = LiteLLMLoggingObj(
+        model="bedrock/anthropic.claude-3-5-sonnet-20240620-v1:0",
+        messages=[{"role": "user", "content": "Hi"}],
+        stream=True,
+        call_type="chat",
+        start_time=datetime.now(),
+        litellm_call_id="test_cache_on_start_only_never_negative_cost",
+        function_id="test_cache_on_start_only_never_negative_cost",
+    )
+
+    collected: list[bytes] = []
+    async for sse in cfg.bedrock_sse_wrapper(
+        completion_stream=_stream(),
+        litellm_logging_obj=logging_obj,
+        request_body={"model": "anthropic.claude-3-5-sonnet-20240620-v1:0"},
+    ):
+        collected.append(sse)
+
+    built = AnthropicPassthroughLoggingHandler._build_complete_streaming_response(
+        all_chunks=collected,
+        model="anthropic.claude-3-5-sonnet-20240620-v1:0",
+        litellm_logging_obj=Mock(),
+    )
+    assert built.usage is not None
+    assert built.usage.prompt_tokens == 22177
+    assert built.usage.completion_tokens == 181
+    assert built.usage.cache_creation_input_tokens == 0
+    assert built.usage.cache_read_input_tokens == 22167
+
+    cost = completion_cost(
+        completion_response=built,
+        model="bedrock/anthropic.claude-3-5-sonnet-20240620-v1:0",
+        custom_llm_provider="bedrock",
+    )
+    assert cost > 0
+    assert cost == pytest.approx(0.0093951, rel=0, abs=1e-9)
+
+
 @pytest.mark.asyncio
 async def test_unified_bedrock_messages_sse_usage_and_cost_claude_sonnet_46():
     """