fix(streaming): keep the served service_tier on streamed chunks and spend rows

litellm/completion_extras/litellm_responses_transformation/transformation.py

@@ -1334,6 +1334,7 @@ def __init__(
     ):
         super().__init__(streaming_response, sync_stream, json_mode)
         self._chat_completion_id: str | None = None
+        self._served_service_tier: str | None = None
         self._tool_call_index_map: dict[int, int] = {}  # mutable-ok: per-stream accumulator state
 
     def _handle_string_chunk(
@@ -1598,6 +1599,7 @@ def translate_responses_chunk_to_openai_stream(
 
                 usage = ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(response_data.get("usage"))
             provider_metadata: Final = _provider_metadata(response_data)
+            served_service_tier: Final = response_data.get("service_tier")
             return ModelResponseStream(
                 choices=[
                     StreamingChoices(
@@ -1611,6 +1613,11 @@ def translate_responses_chunk_to_openai_stream(
                 ],
                 usage=usage,
                 provider_specific_fields=dict(provider_metadata) or None,  # mutable-ok: field is typed dict
+                **(
+                    MappingProxyType({"service_tier": served_service_tier})
+                    if isinstance(served_service_tier, str)
+                    else MappingProxyType({})
+                ),
             )
         else:
             pass
@@ -1639,12 +1646,28 @@ def chunk_parser(self, chunk: dict[str, object]) -> "ModelResponseStream":
             ModelResponseStream: OpenAI-formatted streaming chunk
         """
         verbose_logger.debug("Chat provider: transform_streaming_response called with chunk: %s", chunk)
-        return self._with_stream_scoped_id(
-            OpenAiResponsesToChatCompletionStreamIterator.translate_responses_chunk_to_openai_stream(
-                chunk, tool_call_index_map=self._tool_call_index_map
+        self._remember_served_service_tier(chunk)
+        return self._with_served_service_tier(
+            self._with_stream_scoped_id(
+                OpenAiResponsesToChatCompletionStreamIterator.translate_responses_chunk_to_openai_stream(
+                    chunk, tool_call_index_map=self._tool_call_index_map
+                )
             )
         )
 
+    def _remember_served_service_tier(self, chunk: dict[str, object]) -> None:
+        response_payload: Final = chunk.get("response")
+        if not isinstance(response_payload, dict):
+            return
+        served_tier: Final = response_payload.get("service_tier")
+        if isinstance(served_tier, str) and served_tier:
+            self._served_service_tier = served_tier
+
+    def _with_served_service_tier(self, chunk: "ModelResponseStream") -> "ModelResponseStream":
+        if self._served_service_tier is not None and chunk.model_dump().get("service_tier") is None:
+            setattr(chunk, "service_tier", self._served_service_tier)  # noqa: B010  # pydantic extra, not a declared field
+        return chunk
+
     def _with_stream_scoped_id(self, chunk: "ModelResponseStream") -> "ModelResponseStream":
         if self._chat_completion_id is None:
             self._chat_completion_id = chunk.id

litellm/litellm_core_utils/streaming_chunk_builder_utils.py

@@ -109,6 +109,7 @@ class _BaseChunk(TypedDict, total=False):
     created: ReadOnly[int]
     model: ReadOnly[str]
     system_fingerprint: ReadOnly[str | None]
+    service_tier: ReadOnly[str | None]
     choices: ReadOnly[Required[Sequence[StreamingChoices]]]
     _hidden_params: ReadOnly[_ChunkHiddenParams]
 
@@ -369,6 +370,13 @@ def _get_model_from_chunks(chunks: Sequence["_BaseChunk"], first_chunk_model: st
         # Fall back to first chunk's model if no different model found
         return first_chunk_model
 
+    @staticmethod
+    def _get_service_tier_from_chunks(chunks: Sequence["_BaseChunk"]) -> str | None:
+        return next(
+            (tier for chunk in reversed(chunks) if isinstance(tier := chunk.get("service_tier"), str) and tier),
+            None,
+        )
+
     def build_base_response(self, chunks: Sequence["_BaseChunk"]) -> ModelResponse:
         chunk = self.first_chunk
         id: Final = ChunkProcessor._get_chunk_id(chunks)
@@ -378,6 +386,7 @@ def build_base_response(self, chunks: Sequence["_BaseChunk"]) -> ModelResponse:
         # Get the actual model - for Azure Model Router, this finds the real model from later chunks
         model: Final = ChunkProcessor._get_model_from_chunks(chunks, first_chunk_model)
         system_fingerprint: Final = chunk.get("system_fingerprint", None)
+        service_tier: Final = ChunkProcessor._get_service_tier_from_chunks(chunks)
 
         role: Final = ChunkProcessor._get_role_from_chunks(chunks)
         finish_reason = "stop"
@@ -399,6 +408,11 @@ def build_base_response(self, chunks: Sequence["_BaseChunk"]) -> ModelResponse:
                 "created": created,
                 "model": model,
                 "system_fingerprint": system_fingerprint,
+                **(
+                    MappingProxyType({"service_tier": service_tier})
+                    if service_tier is not None
+                    else MappingProxyType({})
+                ),
                 "choices": [
                     {
                         "index": 0,

litellm/litellm_core_utils/streaming_handler.py

@@ -72,6 +72,12 @@ def _next_sync_or_exhausted(it: Any) -> object:
         return _SYNC_ITER_EXHAUSTED
 
 
+def _stamp_served_service_tier(response: ModelResponseStream, complete_streaming_response: ModelResponse) -> None:
+    served_tier: Final = complete_streaming_response.model_dump().get("service_tier")
+    if isinstance(served_tier, str) and served_tier:
+        setattr(response, "service_tier", served_tier)  # noqa: B010  # pydantic extra, not a declared field
+
+
 def is_async_iterable(obj: object) -> bool:
     """
     Check if an object is an async iterable (can be used with 'async for').
@@ -1873,6 +1879,7 @@ def __next__(self) -> "ModelResponseStream":
                         "usage",
                         getattr(complete_streaming_response, "usage"),
                     )
+                    _stamp_served_service_tier(response, complete_streaming_response)
                     try:
                         _cache_copy = complete_streaming_response.model_copy(deep=True)
                         _log_copy = complete_streaming_response.model_copy(deep=True)
@@ -2124,6 +2131,7 @@ async def _finalize_completed_stream(self, cache_hit: bool) -> "ModelResponseStr
                     "usage",
                     getattr(complete_streaming_response, "usage"),
                 )
+                _stamp_served_service_tier(response, complete_streaming_response)
                 try:
                     _copy = complete_streaming_response.model_copy(deep=True)
                 except RuntimeError:

litellm/proxy/proxy_server.py

@@ -9181,9 +9181,10 @@ def _fast_serialize_simple_model_response_stream(
         "object": getattr(chunk, "object", None),
         "created": getattr(chunk, "created", None),
         "model": model,
+        "service_tier": getattr(chunk, "service_tier", None),
         "choices": [choice_dict],
     }
-    for top_level_key in ("id", "object", "created"):
+    for top_level_key in ("id", "object", "created", "service_tier"):
         if payload[top_level_key] is None:
             payload.pop(top_level_key)
     return orjson.dumps(payload)

tests/e2e/coverage_registry/llm_conversational.yaml

@@ -13,6 +13,7 @@
 - {id: llm.chat_completions.openai.vision.nonstream.works, module: llm, tier: P0, subject_endpoint: chat_completions, route: openai, capability: vision, streaming: nonstream, assertions: [works], source: "model_prices json", rationale: "gpt-4o vision; high usage"}
 - {id: llm.chat_completions.openai.prompt_cache_5m.nonstream.works, module: llm, tier: P0, subject_endpoint: chat_completions, route: openai, capability: prompt_cache_5m, streaming: nonstream, assertions: [works], source: "model_prices json", rationale: "Prompt caching cost optimization"}
 - {id: llm.chat_completions.openai.service_tier.nonstream.works, module: llm, tier: P1, subject_endpoint: chat_completions, route: openai, capability: service_tier, streaming: nonstream, assertions: [works], source: "OpenAI service_tier param", rationale: "OpenAI scale-tier request option is forwarded and echoed"}
+- {id: llm.chat_completions.openai.service_tier.stream.echoes_served_tier, module: llm, tier: P1, subject_endpoint: chat_completions, route: openai, capability: service_tier, streaming: stream, assertions: [works], source: "litellm_core_utils/streaming_handler.py", fail_before_fix: proven, rationale: "Every relayed stream chunk carries the service_tier OpenAI stamped on it, so a streaming caller can see which tier served the request"}
 - {id: llm.chat_completions.openai.thinking.nonstream.works, module: llm, tier: P1, subject_endpoint: chat_completions, route: openai, capability: thinking, streaming: nonstream, assertions: [works], source: "model_prices json", rationale: "o-series reasoning; emerging"}
 - {id: llm.chat_completions.openai.structured_output.nonstream.works, module: llm, tier: P1, subject_endpoint: chat_completions, route: openai, capability: structured_output, streaming: nonstream, assertions: [works], source: "model_prices json", rationale: "response_schema extraction"}
 - {id: llm.chat_completions.anthropic.basic.nonstream.works, module: llm, tier: P0, subject_endpoint: chat_completions, route: anthropic, capability: basic, streaming: nonstream, assertions: [works], source: "proxy_server.py:8455", rationale: "P0 route translated to Anthropic"}

tests/e2e/coverage_registry/quota_management.yaml

@@ -60,6 +60,7 @@
 - {id: quota_management.spend_tracking.stream_cache_read.bills_cache_read_rate, module: quota_management, tier: P1, behavior: spend_tracking, variant: stream_cache_read, assertions: [bills_cache_read_rate], exercised_on: [chat_completions], source: "litellm_core_utils/streaming_chunk_builder_utils.py", rationale: "A streamed call's reassembled usage keeps the cached-token detail so cache reads bill at the cache-read discount, not full input price (#34812)"}
 - {id: quota_management.spend_tracking.messages_bridge.keeps_cache_tokens, module: quota_management, tier: P1, behavior: spend_tracking, variant: messages_bridge, assertions: [keeps_cache_tokens], exercised_on: [messages], source: "llms/anthropic/experimental_pass_through/responses_adapters/handler.py", rationale: "A /v1/messages request served by a Responses-only OpenAI model keeps its cache-read tokens and their discounted billing across the bridge (#34957)"}
 - {id: quota_management.spend_tracking.service_tier.bills_tier_rates, module: quota_management, tier: P1, behavior: spend_tracking, variant: service_tier, assertions: [bills_tier_rates], exercised_on: [chat_completions], source: "cost_calculator.py", rationale: "A priority service_tier call bills input, output, and reasoning at the deployment's *_priority rates and records the tier on the row (#35923, #35925)"}
+- {id: quota_management.spend_tracking.service_tier_stream.records_served_tier, module: quota_management, tier: P1, behavior: spend_tracking, variant: service_tier_stream, assertions: [records_served_tier], exercised_on: [chat_completions], source: "litellm_core_utils/streaming_chunk_builder_utils.py", fail_before_fix: proven, rationale: "A streamed call with no service_tier requested bills at the rates of the tier OpenAI stamps on its chunks and records that served tier on the row; the reassembled stream dropped the provider tier so the row recorded none and priced at the default rates"}
 - {id: quota_management.spend_tracking.cost_headers.additive_components, module: quota_management, tier: P1, behavior: spend_tracking, variant: cost_headers, assertions: [additive_components], exercised_on: [chat_completions], source: "proxy/common_request_processing.py", rationale: "The x-litellm-response-cost-* component headers sum to the total, input covers only fresh tokens, and reasoning stays a subset of output (#36965)"}
 - {id: quota_management.spend_tracking.passthrough_stream.injects_usage_cost, module: quota_management, tier: P1, behavior: spend_tracking, variant: passthrough_stream, assertions: [injects_usage_cost], exercised_on: [openai_passthrough], source: "proxy/pass_through_endpoints/streaming_handler.py", rationale: "With include_cost_in_streaming_usage on, the /openai passthrough's final streaming usage frame carries the proxy-computed cost (#36503). Uncovered: the flag is only settable in litellm_settings, and the shared e2e stack does not turn it on yet"}
 - {id: quota_management.spend_tracking.websearch_interception.bills_under_request_session, module: quota_management, tier: P1, behavior: spend_tracking, variant: websearch_interception, assertions: [bills_under_request_session], exercised_on: [messages], source: "integrations/websearch_interception/handler.py", fail_before_fix: proven, rationale: "A web_search server tool the proxy intercepts into litellm.asearch writes its own asearch spend row, and that row carries the parent request's session_id so the session view counts the search and its cost next to the turn that triggered it (LIT-8063)"}

tests/e2e/quota_management/spend_tracking/test_service_tier_pricing_e2e.py

@@ -13,27 +13,37 @@
 producing a vacuous rate comparison. Reasoning is requested explicitly with
 `reasoning_effort`, so the reasoning-rate assertion rests on a parameter the test
 sets rather than on whatever the model happens to do by default.
+
+The streaming cases pin the served-tier contract: OpenAI stamps the tier it actually
+used on every stream chunk, and that echo is what the caller sees and what the bill
+must be computed on. The request sets no service_tier, so the only place the tier
+can come from is the provider's response. The spend row must record the served tier
+and price input at that tier's rate, and every chunk the proxy relays must carry the
+same service_tier the provider sent.
 """
 
-import pytest
+import json
 
+import pytest
 from cost_rows import (
     approx_equal,
     assert_fresh_tokens_billed_at,
     assert_total_is_sum_of_components,
     poll_cost_row,
     register_priced_model,
 )
-from e2e_config import unique_marker
+from e2e_config import CHEAP_OPENAI_MODEL, unique_marker
 from e2e_http import unwrap
 from lifecycle import ResourceManager
-from models import ChatBody, ChatMessage, LiteLLMParamsBody
+from models import ChatBody, ChatMessage, ChatStreamOptions, LiteLLMParamsBody
+from pydantic import BaseModel
 from spend_e2e_client import SpendClient
 
 pytestmark = pytest.mark.e2e
 
 BACKEND = "openai/gpt-5.6-luna"
 OPENAI_API_KEY = "os.environ/OPENAI_API_KEY"
+STREAM_BACKEND = f"openai/{CHEAP_OPENAI_MODEL}"
 
 INPUT_RATE = 4e-05
 OUTPUT_RATE = 8e-05
@@ -42,6 +52,26 @@
 
 REASONING_EFFORT = "high"
 
+TIER_INPUT_RATES = {"default": INPUT_RATE, "priority": PRIORITY_INPUT_RATE}
+
+
+class _StreamChunk(BaseModel):
+    id: str | None = None
+    service_tier: str | None = None
+
+
+def _stream_chunks(events: list[str]) -> list[_StreamChunk]:
+    return [_StreamChunk.model_validate_json(event) for event in events if event.strip() != "[DONE]"]
+
+
+def _served_tier(chunks: list[_StreamChunk]) -> str:
+    tiers = {chunk.service_tier for chunk in chunks if chunk.service_tier}
+    assert len(tiers) == 1, (
+        f"the relayed stream carried {tiers or 'no'} service tier(s) across {len(chunks)} chunks; OpenAI stamps "
+        "the served tier on every chat chunk, so exactly one tier must reach the caller"
+    )
+    return tiers.pop()
+
 
 class TestServiceTierPricing:
     @pytest.mark.covers("quota_management.spend_tracking.service_tier.bills_tier_rates")
@@ -83,8 +113,7 @@ def test_priority_tier_bills_priority_rates(
             )
         )
         assert chat.service_tier == "priority", (
-            f"OpenAI served tier {chat.service_tier!r} instead of priority; "
-            "tier billing was never exercised"
+            f"OpenAI served tier {chat.service_tier!r} instead of priority; tier billing was never exercised"
         )
         assert chat.id, f"chat response carried no id: {chat}"
 
@@ -119,3 +148,78 @@ def test_priority_tier_bills_priority_rates(
         )
 
         assert_total_is_sum_of_components(row)
+
+    @pytest.mark.covers("quota_management.spend_tracking.service_tier_stream.records_served_tier")
+    def test_streamed_call_records_and_bills_the_served_tier(
+        self, client: SpendClient, resources: ResourceManager, scoped_key: str
+    ) -> None:
+        model = register_priced_model(
+            client.proxy,
+            resources,
+            "tier-priced-stream",
+            LiteLLMParamsBody(
+                model=BACKEND,
+                api_key=OPENAI_API_KEY,
+                input_cost_per_token=INPUT_RATE,
+                output_cost_per_token=OUTPUT_RATE,
+                input_cost_per_token_priority=PRIORITY_INPUT_RATE,
+                output_cost_per_token_priority=PRIORITY_OUTPUT_RATE,
+            ),
+        )
+
+        result = client.proxy.chat_stream(
+            scoped_key,
+            ChatBody(
+                model=model,
+                messages=[ChatMessage(role="user", content=f"{unique_marker()} reply with one word")],
+                max_completion_tokens=64,
+                stream=True,
+            ),
+        )
+        assert result.ok and result.stream_events, (
+            f"streamed chat failed (status {result.status_code}): {result.body[:300]}"
+        )
+        chunks = _stream_chunks(result.stream_events)
+        served_tier = _served_tier(chunks)
+        assert served_tier in TIER_INPUT_RATES, f"no custom rate registered for served tier {served_tier!r}"
+        stream_id = chunks[0].id
+        assert stream_id, f"first stream chunk carried no id: {result.stream_events[0][:200]}"
+
+        row = poll_cost_row(client.proxy, stream_id)
+        assert row is not None, f"no spend row with a cost breakdown landed for {stream_id}"
+        assert row.breakdown.service_tier == served_tier, (
+            f"the provider served tier {served_tier!r} on every chunk but the bill records "
+            f"pricing basis {row.breakdown.service_tier!r}"
+        )
+        assert_fresh_tokens_billed_at(row, TIER_INPUT_RATES[served_tier])
+        assert_total_is_sum_of_components(row)
+
+    @pytest.mark.covers("llm.chat_completions.openai.service_tier.stream.echoes_served_tier")
+    def test_every_streamed_chunk_carries_the_served_tier(
+        self, client: SpendClient, resources: ResourceManager, scoped_key: str
+    ) -> None:
+        model = register_priced_model(
+            client.proxy, resources, "tier-echo-stream", LiteLLMParamsBody(model=STREAM_BACKEND, api_key=OPENAI_API_KEY)
+        )
+        result = client.proxy.chat_stream(
+            scoped_key,
+            ChatBody(
+                model=model,
+                messages=[ChatMessage(role="user", content=f"{unique_marker()} reply with one word")],
+                max_completion_tokens=64,
+                stream=True,
+                stream_options=ChatStreamOptions(include_usage=True),
+            ),
+        )
+        assert result.ok and result.stream_events, (
+            f"streamed chat failed (status {result.status_code}): {result.body[:300]}"
+        )
+        chunks = _stream_chunks(result.stream_events)
+        served_tier = _served_tier(chunks)
+        missing = [
+            json.loads(event) for event, chunk in zip(result.stream_events, chunks) if chunk.service_tier is None
+        ]
+        assert not missing, (
+            f"{len(missing)} of {len(chunks)} relayed chunks dropped the provider's service_tier "
+            f"{served_tier!r}: {missing}"
+        )