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

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/completion_extras/litellm_responses_transformation/transformation.py

@@ -1598,6 +1598,7 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
 
                 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 +1612,7 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
                 ],
                 usage=usage,
                 provider_specific_fields=dict(provider_metadata) or None,  # mutable-ok: field is typed dict
+                **({"service_tier": served_service_tier} if isinstance(served_service_tier, str) else {}),
             )
         else:
             pass

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 @@ class ChunkProcessor:
         # 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 @@ class ChunkProcessor:
         # 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,7 @@ class ChunkProcessor:
                 "created": created,
                 "model": model,
                 "system_fingerprint": system_fingerprint,
+                **({"service_tier": service_tier} if service_tier is not None else {}),
                 "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)
+
+
 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 @@ class CustomStreamWrapper:
                         "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 @@ class CustomStreamWrapper:
                     "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,10 +13,18 @@ priority processing and served the default tier, the test fails there instead of
 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,
@@ -24,16 +32,18 @@ from cost_rows import (
     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 @@ PRIORITY_OUTPUT_RATE = 1.6e-04
 
 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 @@ class TestServiceTierPricing:
             )
         )
         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 @@ class TestServiceTierPricing:
         )
 
         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}"
+        )

tests/test_litellm/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py

@@ -4352,3 +4352,20 @@ def test_map_optional_params_verbosity_merges_into_text():
         verbosity_only_request,
     )
     assert verbosity_only_request["text"] == {"verbosity": "low"}
+
+
+def test_response_completed_carries_the_served_service_tier():
+    from litellm.completion_extras.litellm_responses_transformation.transformation import (
+        OpenAiResponsesToChatCompletionStreamIterator,
+    )
+
+    iterator = OpenAiResponsesToChatCompletionStreamIterator(streaming_response=None, sync_stream=True)
+
+    result = iterator.chunk_parser(
+        {
+            "type": "response.completed",
+            "response": {"id": "resp_1", "status": "completed", "output": [], "service_tier": "default"},
+        }
+    )
+
+    assert result.model_dump()["service_tier"] == "default"

tests/test_litellm/litellm_core_utils/test_streaming_chunk_builder_utils.py

@@ -1795,3 +1795,34 @@ def test_calculate_usage_keeps_a_reported_count_over_a_later_chunks_zero() -> No
     )
 
     assert (usage.prompt_tokens, usage.completion_tokens, usage.total_tokens) == (5, 17, 22)
+
+
+def _tier_chunk(content: str, service_tier: str | None, finish_reason: str | None = None) -> ModelResponseStream:
+    return ModelResponseStream(
+        id="chatcmpl-tier",
+        created=1,
+        model="gpt-4.1-mini",
+        object="chat.completion.chunk",
+        choices=[StreamingChoices(finish_reason=finish_reason, index=0, delta=Delta(content=content, role=None))],
+        **({"service_tier": service_tier} if service_tier is not None else {}),
+    )
+
+
+def test_stream_chunk_builder_records_the_last_service_tier_the_provider_stamped():
+    chunks = [
+        _tier_chunk("Hel", "auto"),
+        _tier_chunk("lo", None),
+        _tier_chunk("", "default", finish_reason="stop"),
+    ]
+
+    response = stream_chunk_builder(chunks=chunks)
+
+    assert response is not None
+    assert response.model_dump()["service_tier"] == "default"
+
+
+def test_stream_chunk_builder_omits_service_tier_when_no_chunk_carried_one():
+    response = stream_chunk_builder(chunks=[_tier_chunk("Hi", None, finish_reason="stop")])
+
+    assert response is not None
+    assert "service_tier" not in response.model_dump()

tests/test_litellm/litellm_core_utils/test_streaming_handler.py

@@ -4942,3 +4942,50 @@ async def test_async_stream_without_usage_counts_tokens_off_the_event_loop():
     assert chunks[-1].usage.prompt_tokens > 100_000
     assert chunks[-1].usage.completion_tokens > 100_000
     assert_loop_stayed_free(took, lags)
+
+
+@pytest.mark.parametrize("sync_mode", [True, False])
+@pytest.mark.asyncio
+async def test_openai_stream_relays_the_served_service_tier_on_every_chunk_including_usage(
+    logging_obj: Logging, sync_mode: bool
+):
+    from litellm.utils import ModelResponseListIterator
+
+    def _chunk(content: str, finish_reason: str | None, usage: Usage | None, choices: bool = True):
+        return ModelResponseStream(
+            id="chatcmpl-tier",
+            created=1742056047,
+            model="gpt-4.1-mini",
+            choices=[StreamingChoices(finish_reason=finish_reason, index=0, delta=Delta(content=content))]
+            if choices
+            else [],
+            usage=usage,
+            service_tier="default",
+        )
+
+    logging_obj.update_environment_variables(
+        model="gpt-4.1-mini",
+        optional_params={"stream_options": {"include_usage": True}},
+        litellm_params={},
+        custom_llm_provider="openai",
+    )
+    wrapper = CustomStreamWrapper(
+        completion_stream=ModelResponseListIterator(
+            model_responses=[
+                _chunk("Hi", None, None),
+                _chunk("", "stop", None),
+                _chunk("", None, Usage(prompt_tokens=10, completion_tokens=1, total_tokens=11), choices=False),
+            ]
+        ),
+        model="gpt-4.1-mini",
+        custom_llm_provider="openai",
+        logging_obj=logging_obj,
+        stream_options={"include_usage": True},
+    )
+
+    relayed = (
+        [chunk.model_dump() for chunk in wrapper] if sync_mode else [chunk.model_dump() async for chunk in wrapper]
+    )
+
+    assert [chunk.get("service_tier") for chunk in relayed] == ["default"] * len(relayed), relayed
+    assert relayed[-1]["usage"]["total_tokens"] == 11

tests/test_litellm/proxy/proxy_server/test_streaming_helpers.py

@@ -2058,3 +2058,13 @@ async def test_queue_request_stream_is_untouched_while_keepalives_are_unconfigur
 
     assert not any(chunk.startswith(b": ping") for chunk in chunks)
     assert chunks[-1] == b"data: [DONE]\n\n"
+
+
+def test_fast_serialize_simple_model_response_stream_keeps_served_service_tier():
+    chunk = _simple_chunk()
+    chunk.service_tier = "priority"
+
+    result = _fast_serialize_simple_model_response_stream(chunk)
+
+    assert result is not None
+    assert json.loads(result)["service_tier"] == "priority"