Merge pull request #44404 from BerriAI/litellm_chore_35b6b0

chore(release): backport #44341, #44346 to stable/1.103.x

litellm/completion_extras/litellm_responses_transformation/transformation.py

@@ -188,7 +188,7 @@ def _reasoning_items_from_output_items(output_items: Sequence[object]) -> tuple[
 
 
 def _as_chat_reasoning_items(
-    reasoning_items: Sequence[_BuiltReasoningItem],
+    reasoning_items: Sequence[_BuiltReasoningItem | ChatCompletionReasoningItem],
 ) -> list[ChatCompletionReasoningItem] | None:
     if not reasoning_items:
         return None
@@ -782,7 +782,32 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
             else:
                 pass  # don't fail request if item in list is not supported
 
-        # If we accumulated tool calls, create a single choice with all of them
+        if accumulated_tool_calls and choices:
+            last_choice: Final = choices[-1]
+            last_reasoning_content: Final = getattr(last_choice.message, "reasoning_content", None)
+            last_reasoning_items: Final = getattr(last_choice.message, "reasoning_items", None)
+            merged_reasoning_content: Final = (
+                " ".join(value for value in (last_reasoning_content, reasoning_content) if value) or None
+            )
+            merged_reasoning_items: Final = _as_chat_reasoning_items(
+                (
+                    *(last_reasoning_items or ()),
+                    *(() if pending_reasoning_item is None else (pending_reasoning_item,)),
+                )
+            )
+            merged_message: Final = Message(
+                role=last_choice.message.role,
+                content=last_choice.message.content,
+                annotations=getattr(last_choice.message, "annotations", None),
+                tool_calls=accumulated_tool_calls,
+                reasoning_content=merged_reasoning_content,
+                reasoning_items=merged_reasoning_items,
+            )
+            return [
+                *choices[:-1],
+                Choices(message=merged_message, finish_reason="tool_calls", index=last_choice.index),
+            ]
+
         if accumulated_tool_calls:
             msg = Message(
                 content=None,

litellm/litellm_core_utils/litellm_logging.py

@@ -610,6 +610,9 @@ class Logging(LiteLLMLoggingBaseClass):
         self.truncated_messages_for_logging: str | list | dict | None = None  # mutable-ok: logged messages shape
         ## TIME TO FIRST TOKEN LOGGING ##
         self.completion_start_time: datetime.datetime | None = None
+        # The model the proxy shows the client on streamed chunks. The logged streamed response carries it
+        # once that response is priced, the same way a non-streamed response is logged
+        self.client_facing_stream_model: str | None = None
         self._llm_caching_handler: LLMCachingHandler | None = None
 
         # INITIAL LITELLM_PARAMS
@@ -2279,6 +2282,15 @@ class Logging(LiteLLMLoggingBaseClass):
             setattr(result, "usage", transformed_usage)
         return result
 
+    def _with_client_facing_stream_model(
+        self,
+        response: ModelResponse | TextCompletionResponse | ResponsesAPIResponse | InteractionsAPIResponse,
+    ) -> ModelResponse | TextCompletionResponse | ResponsesAPIResponse | InteractionsAPIResponse:
+        model: Final = self.client_facing_stream_model
+        if model is None or getattr(response, "model", None) in (None, model):
+            return response
+        return response.model_copy(update={"model": model})
+
     def _success_handler_helper_fn(
         self,
         result=None,
@@ -2592,9 +2604,11 @@ class Logging(LiteLLMLoggingBaseClass):
                     result=complete_streaming_response
                 )
                 self._merge_hidden_params_from_response_into_metadata(complete_streaming_response)
+                logged_streaming_response: Final = self._with_client_facing_stream_model(complete_streaming_response)
+                self.model_call_details["complete_streaming_response"] = logged_streaming_response
                 ## STANDARDIZED LOGGING PAYLOAD
                 self.model_call_details["standard_logging_object"] = self._build_standard_logging_payload(
-                    complete_streaming_response, start_time, end_time
+                    logged_streaming_response, start_time, end_time
                 )
                 standard_logging_payload: Final[StandardLoggingPayload | None] = self.model_call_details.get(
                     "standard_logging_object"
@@ -3130,10 +3144,13 @@ class Logging(LiteLLMLoggingBaseClass):
 
             await self._prepare_baseline_cache_estimate(complete_streaming_response)
 
+            logged_streaming_response: Final = self._with_client_facing_stream_model(complete_streaming_response)
+            self.model_call_details["async_complete_streaming_response"] = logged_streaming_response
+
             ## STANDARDIZED LOGGING PAYLOAD
             try:
                 self.model_call_details["standard_logging_object"] = self._build_standard_logging_payload(
-                    complete_streaming_response, start_time, end_time
+                    logged_streaming_response, start_time, end_time
                 )
             except Exception:  # noqa: BLE001  # payload build must never block later callbacks (slot release)
                 verbose_logger.exception(

litellm/proxy/proxy_server.py

@@ -8951,12 +8951,17 @@ def _restamp_streaming_chunk_model(
         )
         model_mismatch_logged = True
 
+    # The streaming wrapper keeps these same chunk objects to assemble the response it
+    # prices, so stamp a copy for the client and leave the provider's model for pricing.
+    # The logging object stamps the same model on the assembled response after pricing it.
+    logging_obj: Final = request_data.get("litellm_logging_obj")
+    if isinstance(logging_obj, LiteLLMLoggingObj):
+        logging_obj.client_facing_stream_model = target_model
     if isinstance(chunk, dict):
-        chunk["model"] = target_model
-        return chunk, model_mismatch_logged
+        return {**chunk, "model": target_model}, model_mismatch_logged
 
     try:
-        chunk.model = target_model
+        return chunk.model_copy(update={"model": target_model}), model_mismatch_logged
     except Exception as e:
         verbose_proxy_logger.error(
             "litellm_call_id=%s: failed to override chunk.model=%r on chunk_type=%s. error=%s",

tests/integration/contracts.json

@@ -1599,6 +1599,12 @@
     ],
     "tests/integration/spend/test_daily_activity_key_owner_traffic.py::test_owner_is_reported_on_every_route_while_a_burst_of_requests_waits_on_the_provider": [
       "quota_management.spend_tracking.key_attribution.every_route_under_provider_burst"
+    ],
+    "tests/integration/spend/test_stream_alias_billing.py::test_streamed_alias_matching_a_capability_rule_bills_the_deployment_price[vertex-mock-response]": [
+      "quota_management.spend_tracking.streamed_alias_bills_deployment_price"
+    ],
+    "tests/integration/spend/test_stream_alias_billing.py::test_streamed_alias_matching_a_capability_rule_bills_the_deployment_price[anthropic-upstream]": [
+      "quota_management.spend_tracking.streamed_alias_bills_deployment_price"
     ]
   },
   "browser": {

tests/integration/spend/test_stream_alias_billing.py

@@ -0,0 +1,191 @@
+"""A streamed alias never replaces the deployment's model for pricing (LIT-9065).
+
+The proxy shows the client's alias on every streamed chunk, but the chunks kept for end-of-stream cost calculation
+keep the deployment's model. "claude-opus-4.8-<digits>" is no cost-map key and only matches the claude capability
+rules, whose model info carries no prices, so a stream through that alias must bill exactly what the plain alias
+"integration-<hex>" bills at the same deployment rates, and the client must still see the alias it asked for
+"""
+
+import json
+from collections.abc import Callable
+from hashlib import sha256
+from typing import Final
+from uuid import uuid4
+
+import pytest
+from integration._support.client import (
+    Gateway,
+    Scenario,
+    eventually,
+    object_value,
+    string_value,
+)
+from integration._support.database import read_rows
+from integration._support.wire import Reply, Request, wire_server
+from pydantic import JsonValue
+
+
+def _sse_event(name: str, payload: dict[str, JsonValue]) -> bytes:
+    return f"event: {name}\ndata: {json.dumps(payload, separators=(',', ':'))}\n\n".encode()
+
+
+def _anthropic_reply(request: Request) -> Reply:
+    assert request.target.endswith("/v1/messages"), request.target
+    body: Final = json.loads(request.body)
+    assert body["model"] == "claude-opus-4-8", body
+    if body.get("stream") is not True:
+        return Reply(
+            body=json.dumps(
+                {
+                    "id": f"msg_{uuid4().hex[:12]}",
+                    "type": "message",
+                    "role": "assistant",
+                    "model": "claude-opus-4-8",
+                    "content": [{"type": "text", "text": "hi"}],
+                    "stop_reason": "end_turn",
+                    "stop_sequence": None,
+                    "usage": {"input_tokens": 30, "output_tokens": 40},
+                }
+            ).encode()
+        )
+    return Reply(
+        content_type="text/event-stream",
+        chunks=(
+            _sse_event(
+                "message_start",
+                {
+                    "type": "message_start",
+                    "message": {
+                        "id": f"msg_{uuid4().hex[:12]}",
+                        "type": "message",
+                        "role": "assistant",
+                        "model": "claude-opus-4-8",
+                        "content": [],
+                        "stop_reason": None,
+                        "stop_sequence": None,
+                        "usage": {"input_tokens": 30, "output_tokens": 1},
+                    },
+                },
+            ),
+            _sse_event(
+                "content_block_start",
+                {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}},
+            ),
+            _sse_event(
+                "content_block_delta",
+                {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "hi"}},
+            ),
+            _sse_event("content_block_stop", {"type": "content_block_stop", "index": 0}),
+            _sse_event(
+                "message_delta",
+                {
+                    "type": "message_delta",
+                    "delta": {"stop_reason": "end_turn", "stop_sequence": None},
+                    "usage": {"output_tokens": 40},
+                },
+            ),
+            _sse_event("message_stop", {"type": "message_stop"}),
+        ),
+    )
+
+
+def _deployment(
+    scenario: Scenario,
+    model_name: str,
+    litellm_params: dict[str, JsonValue],
+    model_info: dict[str, JsonValue] | None = None,
+) -> str:
+    created: Final = scenario.gateway.post(
+        "/model/new", {"model_name": model_name, "litellm_params": litellm_params, "model_info": model_info or {}}
+    )
+    identity: Final = string_value(object_value(created["model_info"])["id"])
+    scenario.cleanups.callback(scenario.delete_model, identity)
+    return model_name
+
+
+def _streamed_spend(gateway: Gateway, scenario: Scenario, model: str, content: str) -> dict[str, JsonValue]:
+    key: Final = scenario.key(models=[model])
+    response: Final = gateway.request(
+        "POST",
+        "/v1/chat/completions",
+        {
+            "model": model,
+            "messages": [{"role": "user", "content": content}],
+            "stream": True,
+            "stream_options": {"include_usage": True},
+        },
+        key=key,
+    )
+    assert response.status_code == 200, response.text
+    chunks: Final = tuple(
+        json.loads(line.removeprefix("data: "))
+        for line in response.text.splitlines()
+        if line.startswith("data: ") and line != "data: [DONE]"
+    )
+    assert chunks and {chunk["model"] for chunk in chunks} == {model}, response.text
+    rows: Final = eventually(
+        lambda: read_rows(
+            'SELECT spend, prompt_tokens, completion_tokens FROM "LiteLLM_SpendLogs" WHERE api_key=%s',
+            (sha256(key.encode()).hexdigest(),),
+        ),
+        lambda values: len(values) == 1,
+        seconds=70,
+    )
+    return rows[0]
+
+
+def _listed_deployments(gateway: Gateway, model_name: str) -> tuple[dict[str, JsonValue], ...]:
+    entries: Final = gateway.get("/model/info")["data"]
+    assert isinstance(entries, list)
+    return tuple(object_value(entry) for entry in entries if object_value(entry)["model_name"] == model_name)
+
+
+def _deployment_pricing(gateway: Gateway, model_name: str) -> dict[str, JsonValue]:
+    listed: Final = eventually(lambda: _listed_deployments(gateway, model_name), lambda found: len(found) == 1)
+    return object_value(listed[0]["model_info"])
+
+
+_BACKENDS: Final = (
+    pytest.param(
+        lambda _: {"model": "vertex_ai/claude-opus-4-8@default", "mock_response": "hi"},
+        id="vertex-mock-response",
+    ),
+    pytest.param(
+        lambda wire_url: {
+            "model": "anthropic/claude-opus-4-8",
+            "api_key": "integration-provider-key",
+            "api_base": wire_url,
+        },
+        id="anthropic-upstream",
+    ),
+)
+
+
+@pytest.mark.covers("quota_management.spend_tracking.streamed_alias_bills_deployment_price")
+@pytest.mark.parametrize("litellm_params", _BACKENDS)
+@pytest.mark.timeout(180)
+def test_streamed_alias_matching_a_capability_rule_bills_the_deployment_price(
+    gateway: Gateway, litellm_params: Callable[[str], dict[str, JsonValue]]
+) -> None:
+    with wire_server(_anthropic_reply) as wire, gateway.scenario() as scenario:
+        content: Final = f"alias billing {uuid4().hex}"
+        plain_alias: Final = f"integration-{uuid4().hex}"
+        rule_alias: Final = f"claude-opus-4.8-{uuid4().int % 10**8:08d}"
+        exact_row: Final = _streamed_spend(
+            gateway, scenario, _deployment(scenario, plain_alias, litellm_params(wire.url)), content
+        )
+        alias_row: Final = _streamed_spend(
+            gateway, scenario, _deployment(scenario, rule_alias, litellm_params(wire.url)), content
+        )
+
+        for model_name, row in ((plain_alias, exact_row), (rule_alias, alias_row)):
+            pricing: Final = _deployment_pricing(gateway, model_name)
+            input_rate: Final = float(str(pricing["input_cost_per_token"]))
+            output_rate: Final = float(str(pricing["output_cost_per_token"]))
+            uplift: Final = float(str(pricing["regional_endpoint_uplift_multiplier"] or 1))
+            assert input_rate > 0 and output_rate > 0, pricing
+            assert float(str(row["spend"])) == pytest.approx(
+                uplift
+                * (float(str(row["prompt_tokens"])) * input_rate + float(str(row["completion_tokens"])) * output_rate)
+            ), (model_name, row, pricing)
+

tests/test_litellm/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py

@@ -7,6 +7,15 @@ from unittest.mock import ANY, MagicMock, Mock, patch
 
 import httpx
 import pytest
+from openai.types.responses import (
+    ResponseFunctionToolCall,
+    ResponseOutputMessage,
+    ResponseOutputText,
+)
+from openai.types.responses.response_reasoning_item import (
+    ResponseReasoningItem,
+    Summary,
+)
 
 import litellm
 from litellm.completion_extras.litellm_responses_transformation.transformation import (
@@ -3275,6 +3284,148 @@ def test_convert_response_output_generic_pydantic_message_item():
     assert choices[0].finish_reason == "stop"
 
 
+def test_convert_response_output_merges_message_reasoning_and_function_call() -> None:
+    message: Final = ResponseOutputMessage(
+        id="msg_weather",
+        content=[
+            ResponseOutputText(
+                annotations=[
+                    {
+                        "type": "url_citation",
+                        "start_index": 0,
+                        "end_index": 5,
+                        "title": "Forecast",
+                        "url": "https://example.com/forecast",
+                    }
+                ],
+                text="Sunny.",
+                type="output_text",
+                logprobs=[],
+            )
+        ],
+        role="assistant",
+        status="completed",
+        type="message",
+    )
+    reasoning: Final = ResponseReasoningItem(
+        id="rs_before",
+        summary=[Summary(type="summary_text", text="Checking the forecast.")],
+        type="reasoning",
+        content=None,
+        encrypted_content=None,
+        status=None,
+    )
+    pending_reasoning: Final = ResponseReasoningItem(
+        id="rs_after",
+        summary=[Summary(type="summary_text", text="The location is Paris.")],
+        type="reasoning",
+        content=None,
+        encrypted_content=None,
+        status=None,
+    )
+    function_call: Final = ResponseFunctionToolCall(
+        id="fc_1",
+        type="function_call",
+        status="completed",
+        arguments='{"city":"Paris"}',
+        call_id="call_1",
+        name="get_weather",
+    )
+
+    message_and_call: Final = LiteLLMResponsesTransformationHandler._convert_response_output_to_choices(
+        (message, function_call)
+    )
+    assert len(message_and_call) == 1
+    assert message_and_call[0].index == 0
+    assert message_and_call[0].finish_reason == "tool_calls"
+    assert message_and_call[0].message.role == "assistant"
+    assert message_and_call[0].message.content == "Sunny."
+    assert message_and_call[0].message.annotations == [
+        {
+            "type": "url_citation",
+            "start_index": 0,
+            "end_index": 5,
+            "title": "Forecast",
+            "url": "https://example.com/forecast",
+        }
+    ]
+    function_calls: Final = message_and_call[0].message.tool_calls
+    assert function_calls is not None
+    assert len(function_calls) == 1
+    assert function_calls[0].function.name == "get_weather"
+    assert function_calls[0].function.arguments == '{"city":"Paris"}'
+
+    reasoning_before_message: Final = LiteLLMResponsesTransformationHandler._convert_response_output_to_choices(
+        (reasoning, message, function_call)
+    )
+    assert len(reasoning_before_message) == 1
+    assert reasoning_before_message[0].message.reasoning_content == "Checking the forecast."
+    reasoning_before_items: Final = reasoning_before_message[0].message.reasoning_items
+    assert reasoning_before_items is not None
+    assert reasoning_before_items[0]["id"] == "rs_before"
+
+    reasoning_after_message: Final = LiteLLMResponsesTransformationHandler._convert_response_output_to_choices(
+        (message, pending_reasoning, function_call)
+    )
+    assert len(reasoning_after_message) == 1
+    assert reasoning_after_message[0].message.reasoning_content == "The location is Paris."
+    reasoning_after_items: Final = reasoning_after_message[0].message.reasoning_items
+    assert reasoning_after_items is not None
+    assert reasoning_after_items[0]["id"] == "rs_after"
+
+    merged_reasoning: Final = LiteLLMResponsesTransformationHandler._convert_response_output_to_choices(
+        (reasoning, message, pending_reasoning, function_call)
+    )
+    assert len(merged_reasoning) == 1
+    assert merged_reasoning[0].message.reasoning_content == "Checking the forecast. The location is Paris."
+    merged_reasoning_items: Final = merged_reasoning[0].message.reasoning_items
+    assert merged_reasoning_items is not None
+    assert [item["id"] for item in merged_reasoning_items] == ["rs_before", "rs_after"]
+
+    tool_only: Final = LiteLLMResponsesTransformationHandler._convert_response_output_to_choices((function_call,))
+    assert len(tool_only) == 1
+    assert tool_only[0].index == 0
+    assert tool_only[0].finish_reason == "tool_calls"
+    assert tool_only[0].message.content is None
+    assert tool_only[0].message.tool_calls is not None
+    assert len(tool_only[0].message.tool_calls) == 1
+
+    message_only: Final = LiteLLMResponsesTransformationHandler._convert_response_output_to_choices((message,))
+    assert len(message_only) == 1
+    assert message_only[0].index == 0
+    assert message_only[0].finish_reason == "stop"
+    assert message_only[0].message.content == "Sunny."
+    assert message_only[0].message.tool_calls is None
+
+
+def test_convert_response_output_merges_raw_dict_message_and_function_call() -> None:
+    handler: Final = LiteLLMResponsesTransformationHandler()
+    raw_message: Final = {
+        "type": "message",
+        "role": "assistant",
+        "content": [{"type": "output_text", "text": "Let me check.", "annotations": []}],
+    }
+    raw_function_call: Final = {
+        "type": "function_call",
+        "id": "fc_1",
+        "call_id": "call_1",
+        "name": "get_weather",
+        "arguments": '{"city":"Paris"}',
+    }
+    choices: Final = LiteLLMResponsesTransformationHandler._convert_response_output_to_choices(
+        (raw_message, raw_function_call),
+        handle_raw_dict_callback=handler._handle_raw_dict_response_item,
+    )
+
+    assert len(choices) == 1
+    assert choices[0].index == 0
+    assert choices[0].finish_reason == "tool_calls"
+    assert choices[0].message.role == "assistant"
+    assert choices[0].message.content == "Let me check."
+    assert choices[0].message.tool_calls is not None
+    assert len(choices[0].message.tool_calls) == 1
+
+
 def test_convert_tools_to_responses_format_flattens_nested_custom_tool():
     from litellm.completion_extras.litellm_responses_transformation.transformation import (
         LiteLLMResponsesTransformationHandler,

tests/test_litellm/litellm_core_utils/test_litellm_logging.py

@@ -47,6 +47,18 @@ def logging_obj():
     )
 
 
+def test_with_client_facing_stream_model_stamps_a_copy_of_the_priced_response(logging_obj):
+    response = ModelResponse(model="claude-opus-4-6@default")
+    logging_obj.client_facing_stream_model = "claude-opus-4.6"
+    logged = logging_obj._with_client_facing_stream_model(response)
+    assert (logged.model, response.model) == ("claude-opus-4.6", "claude-opus-4-6@default")
+
+
+def test_with_client_facing_stream_model_keeps_the_response_when_the_proxy_set_no_model(logging_obj):
+    response = ModelResponse(model="claude-opus-4-6@default")
+    assert logging_obj._with_client_facing_stream_model(response) is response
+
+
 def test_get_combined_callback_list_preserves_insertion_order(logging_obj):
     assert logging_obj.get_combined_callback_list(
         dynamic_success_callbacks=["prometheus", "langfuse", "datadog", "otel", "s3"],

tests/test_litellm/proxy/proxy_server/test_streaming_helpers.py

@@ -283,8 +283,9 @@ def test_restamp_streaming_chunk_model_overrides_model_on_basemodel():
         "model": new_chunk.model,
         "logged": logged,
         "same_object": new_chunk is chunk,
+        "original_model": chunk.model,
     }
-    assert snapshot == {"model": "gpt-4", "logged": True, "same_object": True}
+    assert snapshot == {"model": "gpt-4", "logged": True, "same_object": False, "original_model": "openai/internal-x"}
 
 
 @pytest.mark.parametrize("return_raw_model_name", [False, True])
@@ -310,8 +311,7 @@ def test_restamp_streaming_chunk_model_overrides_model_on_dict():
         request_data={},
         model_mismatch_logged=True,
     )
-    assert new_chunk["model"] == "gpt-4"
-    assert logged is True
+    assert (new_chunk["model"], chunk["model"], logged) == ("gpt-4", "internal", True)
 
 
 def test_restamp_streaming_chunk_model_uses_fallback_model_from_metadata():
@@ -443,7 +443,7 @@ def test_restamp_streaming_chunk_model_fastest_response_preserves_model():
     assert logged is False
 
 
-def test_restamp_streaming_chunk_model_setattr_exception_logs_and_returns():
+def test_restamp_streaming_chunk_model_restamps_a_frozen_chunk_through_a_copy():
     from pydantic import ConfigDict
 
     class FrozenChunk(_simple_chunk().__class__):
@@ -462,8 +462,30 @@ def test_restamp_streaming_chunk_model_setattr_exception_logs_and_returns():
         request_data={"litellm_call_id": "test-id"},
         model_mismatch_logged=False,
     )
-    assert new_chunk.model == "openai/internal-x"
-    assert logged is True
+    assert (new_chunk.model, chunk.model, logged) == ("gpt-4", "openai/internal-x", True)
+
+
+def test_restamp_streaming_chunk_model_records_the_client_model_on_the_logging_object():
+    import time
+
+    from litellm.litellm_core_utils.litellm_logging import Logging
+
+    logging_obj = Logging(
+        model="openai/internal-x",
+        messages=[],
+        stream=True,
+        call_type="acompletion",
+        start_time=time.time(),
+        litellm_call_id="test-id",
+        function_id="test-id",
+    )
+    _restamp_streaming_chunk_model(
+        chunk=_simple_chunk(model="openai/internal-x"),
+        requested_model_from_client="gpt-4",
+        request_data={"litellm_call_id": "test-id", "litellm_logging_obj": logging_obj},
+        model_mismatch_logged=False,
+    )
+    assert logging_obj.client_facing_stream_model == "gpt-4"
 
 
 def test_format_fallback_metadata_sse_event():