Merge 0e0c895c55 into 9007e0fa70

litellm/cost_calculator.py

@@ -5,7 +5,7 @@ import math
 import time
 from collections.abc import Mapping, Sequence
 from datetime import datetime
-from functools import lru_cache
+from functools import lru_cache, partial
 from types import MappingProxyType
 from typing import TYPE_CHECKING, Any, Final, Literal, TypeAlias, cast
 
@@ -22,7 +22,7 @@ from litellm.constants import (
     OPENAI_LIVE_TERMINAL_RESPONSE_EVENT_TYPES,
     REALTIME_METERED_USAGE_KEY,
 )
-from litellm.litellm_core_utils.core_helpers import as_str_mapping
+from litellm.litellm_core_utils.core_helpers import as_str_mapping, provider_reported_usage_cost
 from litellm.litellm_core_utils.cost_discount import resolve_cost_discount
 from litellm.litellm_core_utils.decisions_pricing import decisions_pricing_fields, is_decisions_call_type
 from litellm.litellm_core_utils.hidden_params import HIDDEN_PARAMS_ATTR
@@ -408,13 +408,114 @@ def _per_second_pricing_cost(
     return resolved_cost_per_second * seconds, 0.0
 
 
+_PROVIDERS_WITH_OWN_REPORTED_COST_READER: Final = frozenset(
+    {LlmProviders.XAI.value, LlmProviders.PERPLEXITY.value, LlmProviders.OPENCODE_GO.value}
+)
+
+
+def _billable_reported_cost(usage_object: Usage | None, custom_llm_provider: str | None) -> float | None:
+    if custom_llm_provider in _PROVIDERS_WITH_OWN_REPORTED_COST_READER:
+        return None
+    return _billable_header_cost(provider_reported_usage_cost(usage_object, custom_llm_provider))
+
+
+def _split_reported_cost(reported_cost: float, priced: tuple[float, float] | None) -> tuple[float, float]:
+    """Providers report one total, so attribute it to input and output in the ratio the token math gives."""
+    priced_total: Final = (priced[0] + priced[1]) if priced is not None else 0.0
+    if priced is None or priced_total <= 0:
+        return 0.0, reported_cost
+    input_cost: Final = reported_cost * priced[0] / priced_total
+    return input_cost, reported_cost - input_cost
+
+
 def cost_per_token(
     model: str = "",
     prompt_tokens: int = 0,
     completion_tokens: int = 0,
     response_time_ms: float | None = 0.0,
     custom_llm_provider: str | None = None,
-    region_name=None,
+    region_name: str | None = None,
+    ### CHARACTER PRICING ###
+    prompt_characters: int | None = None,
+    completion_characters: int | None = None,
+    ### PROMPT CACHING PRICING ### - used for anthropic
+    cache_creation_input_tokens: int | None = 0,
+    cache_read_input_tokens: int | None = 0,
+    ### CUSTOM PRICING ###
+    custom_cost_per_token: CostPerToken | None = None,
+    custom_cost_per_second: float | None = None,
+    ### NUMBER OF QUERIES ###
+    number_of_queries: int | None = None,
+    ### USAGE OBJECT ###
+    usage_object: Usage | None = None,  # just read the usage object if provided
+    ### BILLED UNITS ###
+    rerank_billed_units: RerankBilledUnits | None = None,
+    ### CALL TYPE ###
+    call_type: CallTypesLiteral = "completion",
+    audio_transcription_file_duration: float = 0.0,  # for audio transcription calls - the file time in seconds
+    ### SERVICE TIER ###
+    service_tier: str | None = None,  # for OpenAI service tier pricing
+    ### DATA RESIDENCY ###
+    data_residency: str | None = None,  # for OpenAI regional-processing uplift (e.g. "eu", "us")
+    ### VERTEX LOCATION ###
+    vertex_location: str | None = None,  # for Vertex AI regional-endpoint uplift (e.g. "us-east5", "global")
+    response: object | None = None,
+    ### REQUEST MODEL ###
+    request_model: str | None = None,  # original request model for router detection
+    custom_model_info: OCRPricing | None = None,
+    bedrock_model_info: ModelInfo | None = None,
+) -> tuple[float, float]:
+    """Cost in USD for prompt and completion tokens.
+
+    A numeric ``usage.cost`` the provider reported is the bill, split by the ratio of our own token math,
+    unless the caller priced the call itself with ``custom_cost_per_token`` or ``custom_cost_per_second``.
+    Without a reported cost the token math is the bill.
+    """
+    priced_by_caller: Final = custom_cost_per_token is not None or custom_cost_per_second is not None
+    reported_cost: Final = None if priced_by_caller else _billable_reported_cost(usage_object, custom_llm_provider)
+    rated: Final = partial(
+        _cost_per_token_from_rates,
+        model=model,
+        prompt_tokens=prompt_tokens,
+        completion_tokens=completion_tokens,
+        response_time_ms=response_time_ms,
+        custom_llm_provider=custom_llm_provider,
+        region_name=region_name,
+        prompt_characters=prompt_characters,
+        completion_characters=completion_characters,
+        cache_creation_input_tokens=cache_creation_input_tokens,
+        cache_read_input_tokens=cache_read_input_tokens,
+        custom_cost_per_token=custom_cost_per_token,
+        custom_cost_per_second=custom_cost_per_second,
+        number_of_queries=number_of_queries,
+        usage_object=usage_object,
+        rerank_billed_units=rerank_billed_units,
+        call_type=call_type,
+        audio_transcription_file_duration=audio_transcription_file_duration,
+        service_tier=service_tier,
+        data_residency=data_residency,
+        vertex_location=vertex_location,
+        response=response,
+        request_model=request_model,
+        custom_model_info=custom_model_info,
+        bedrock_model_info=bedrock_model_info,
+    )
+    if reported_cost is None:
+        return rated()
+    try:
+        return _split_reported_cost(reported_cost, rated())
+    except Exception:  # noqa: BLE001  # an unmapped model has no rates to split by, the reported total still bills
+        verbose_logger.debug("No rates for model=%s, billing the provider reported cost %s", model, reported_cost)
+        return _split_reported_cost(reported_cost, None)
+
+
+def _cost_per_token_from_rates(
+    model: str = "",
+    prompt_tokens: int = 0,
+    completion_tokens: int = 0,
+    response_time_ms: float | None = 0.0,
+    custom_llm_provider: str | None = None,
+    region_name: str | None = None,
     ### CHARACTER PRICING ###
     prompt_characters: int | None = None,
     completion_characters: int | None = None,
@@ -1515,7 +1616,7 @@ def completion_cost(
     call_type: CallTypesLiteral | None = None,
     ### REGION ###
     custom_llm_provider: str | None = None,
-    region_name=None,  # used for bedrock pricing
+    region_name: str | None = None,  # used for bedrock pricing
     ### IMAGE GEN ###
     size: str | None = None,
     quality: str | None = None,
@@ -1610,7 +1711,9 @@ def completion_cost(
         audio_transcription_file_duration: float = 0.0
         provider_usage_object: Final = get_usage_object(completion_response=completion_response)
         cost_per_token_usage_object: Final[Usage | None] = (
-            _without_provider_stated_cost(provider_usage_object) if custom_pricing else provider_usage_object
+            _without_provider_stated_cost(provider_usage_object)
+            if custom_pricing or base_model is not None
+            else provider_usage_object
         )
         rerank_billed_units: RerankBilledUnits | None = None
 

litellm/litellm_core_utils/hidden_params.py

@@ -63,7 +63,8 @@ def stored_attribute(obj: object, name: str) -> object:
 def _get_hidden_params_storage(obj: object) -> object | None:
     if isinstance(obj, dict):
         return obj.get(_HIDDEN_PARAMS_ATTR)
-    if isinstance(obj, BaseModel) and _HIDDEN_PARAMS_ATTR in type(obj).__private_attributes__:
+    model_type: Final = type(obj)
+    if issubclass(model_type, BaseModel) and _HIDDEN_PARAMS_ATTR in model_type.__private_attributes__:
         return getattr(obj, _HIDDEN_PARAMS_ATTR, None)
     return stored_attribute(obj, _HIDDEN_PARAMS_ATTR)
 

litellm/litellm_core_utils/streaming_chunk_builder_utils.py

@@ -1351,9 +1351,6 @@ def stream_chunk_builder_text_completion(
     return TextCompletionResponse(**response)
 
 
-_CALCULATOR_PRICED_REPORTED_COST_PROVIDERS: Final = frozenset({LlmProviders.XAI.value})
-
-
 def _logged_provider(logging_obj: Optional["Logging"]) -> object:
     if logging_obj is None:
         return None
@@ -1361,10 +1358,6 @@ def _logged_provider(logging_obj: Optional["Logging"]) -> object:
     return provider
 
 
-def _reported_cost_is_priced_by_calculator(logging_obj: Optional["Logging"]) -> bool:
-    return _logged_provider(logging_obj) in _CALCULATOR_PRICED_REPORTED_COST_PROVIDERS
-
-
 def _stream_builder_response_cost(
     response: ModelResponse, logging_obj: Optional["Logging"], chunks: Sequence[object]
 ) -> float | None:
@@ -1375,9 +1368,6 @@ def _stream_builder_response_cost(
             for chunk in reversed(chunks)
         )
         return next((cost for cost in reported if cost is not None), None)
-    usage_cost: Final = getattr(getattr(response, "usage", None), "cost", None)
-    if isinstance(usage_cost, (int, float)) and not _reported_cost_is_priced_by_calculator(logging_obj):
-        return float(usage_cost)
     if logging_obj is not None:
         return None
     provider_hint: Final = response._hidden_params.get(  # pyright: ignore[reportPrivateUsage]  # no public accessor
@@ -1421,15 +1411,12 @@ def _set_stream_builder_response_cost(
     _set_hidden_response_cost(response, response_cost)
 
 
-def _stamp_streaming_usage_cost(usage: Usage, response: ModelResponse, logging_obj: Optional["Logging"]) -> None:
+def _stamp_streaming_response_cost(response: ModelResponse, logging_obj: Optional["Logging"]) -> None:
     if logging_obj is None:
         return
-    if isinstance(getattr(usage, "cost", None), (int, float)):
-        return
     computed_cost: Final = logging_obj.response_cost_calculator(result=response)
-    if not isinstance(computed_cost, (int, float)) or computed_cost <= 0:
+    if isinstance(computed_cost, bool) or not isinstance(computed_cost, (int, float)) or computed_cost < 0:
         return
-    setattr(usage, "cost", computed_cost)
     _set_hidden_response_cost(response, float(computed_cost))
 
 
@@ -1600,7 +1587,7 @@ def stream_chunk_builder(
                 response,
                 logging_obj,
             )
-            _stamp_streaming_usage_cost(usage, response, logging_obj)
+            _stamp_streaming_response_cost(response, logging_obj)
             _set_stream_builder_response_cost(
                 response,
                 logging_obj,
@@ -1788,7 +1775,7 @@ def stream_chunk_builder(
                 )
                 break
 
-        _stamp_streaming_usage_cost(usage, response, logging_obj)
+        _stamp_streaming_response_cost(response, logging_obj)
         _set_stream_builder_response_cost(
             response,
             logging_obj,

litellm/litellm_core_utils/streaming_handler.py

@@ -92,6 +92,17 @@ def _next_sync_or_exhausted(it: Iterator[object]) -> object:
         return _SYNC_ITER_EXHAUSTED
 
 
+def _stamp_hidden_response_cost(
+    usage_chunk: ModelResponse | TextCompletionResponse | ModelResponseStream,
+    complete_response: ModelResponse | TextCompletionResponse | ModelResponseStream,
+) -> None:
+    """The bill the builder priced for the spend row rides the usage chunk, so the proxy shows the same number."""
+    bill: Final = (get_hidden_params(complete_response) or {}).get("response_cost")
+    target: Final = get_hidden_params(usage_chunk)
+    if target is not None and isinstance(bill, (int, float)) and not isinstance(bill, bool):
+        target["response_cost"] = float(bill)
+
+
 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:
@@ -2124,6 +2135,7 @@ class CustomStreamWrapper:
                         getattr(complete_streaming_response, "usage"),
                     )
                     _stamp_served_service_tier(response, complete_streaming_response)
+                    _stamp_hidden_response_cost(response, complete_streaming_response)
                     try:
                         _cache_copy = complete_streaming_response.model_copy(deep=True)
                         _log_copy = complete_streaming_response.model_copy(deep=True)
@@ -2345,6 +2357,7 @@ class CustomStreamWrapper:
                     getattr(complete_streaming_response, "usage"),
                 )
                 _stamp_served_service_tier(response, complete_streaming_response)
+                _stamp_hidden_response_cost(response, complete_streaming_response)
                 try:
                     _copy = complete_streaming_response.model_copy(deep=True)
                 except RuntimeError:

litellm/proxy/proxy_server.py

@@ -930,6 +930,7 @@ from litellm.proxy.ui_crud_endpoints.proxy_setting_endpoints import (
 from litellm.proxy.ui_crud_endpoints.user_banner_endpoints import (
     router as user_banner_endpoints_router,
 )
+from litellm.proxy.usage_cost_egress import with_client_stream_usage_cost
 from litellm.proxy.utils import (  # noqa: F401, RUF100  # legacy module exports
     PrismaClient,
     ProxyLogging,
@@ -10563,7 +10564,7 @@ async def async_data_generator(
                 error_state.observe_chunk(chunk)
             raw_passthrough = False
             if isinstance(chunk, BaseModel):
-                chunk = _serialize_streaming_chunk(chunk)
+                chunk = _serialize_streaming_chunk(with_client_stream_usage_cost(chunk))
             elif isinstance(chunk, bytes):
                 chunk = chunk.decode("utf-8", errors="replace")
                 if is_raw_sse_stream:

litellm/proxy/usage_cost_egress.py

@@ -0,0 +1,63 @@
+"""What a client reads in a streamed ``usage.cost``: the bill litellm already priced for the spend row, copied once on
+the way out of the proxy onto a copy of the event. Nothing is priced here. Inside the pipeline ``usage.cost`` stays
+what the upstream reported."""
+
+from __future__ import annotations
+
+import math
+from typing import Final, TypeVar
+
+from pydantic import BaseModel
+
+from litellm.litellm_core_utils.hidden_params import get_hidden_params
+from litellm.types.llms.openai import ResponseAPIUsage, ResponsesAPIResponse
+from litellm.types.utils import ModelResponseStream, Usage
+
+_RESPONSES_USAGE_EVENTS: Final = frozenset({"response.completed", "response.incomplete"})
+_ReplyT: Final = TypeVar("_ReplyT", bound=BaseModel)
+
+
+def _billable(cost: object) -> float | None:
+    if isinstance(cost, bool) or not isinstance(cost, (int, float)) or not math.isfinite(cost) or cost < 0:
+        return None
+    return float(cost)
+
+
+def _client_usage_cost(upstream_cost: object, bill: object) -> float | None:
+    """The bill, where the upstream put a number in usage.cost (a $0 bill included) or litellm priced the
+    stream above zero. None leaves the usage as the upstream sent it."""
+    billable: Final = _billable(bill)
+    if billable is None:
+        return None
+    reported: Final = isinstance(upstream_cost, (int, float)) and not isinstance(upstream_cost, bool)
+    return billable if reported or billable > 0 else None
+
+
+def _with_usage_cost(reply: _ReplyT, usage: Usage | ResponseAPIUsage, cost: float) -> _ReplyT:
+    return reply.model_copy(update={"usage": usage.model_copy(update={"cost": cost})})
+
+
+def _hidden_bill(reply: object) -> object:
+    hidden: Final = get_hidden_params(reply)
+    return hidden.get("response_cost") if hidden is not None else None
+
+
+def with_client_stream_usage_cost(chunk: BaseModel) -> BaseModel:
+    """A streamed event as the client gets it. Only the usage-bearing events change: the chat chunk carrying
+    ``usage`` and the Responses ``response.completed`` / ``response.incomplete`` events, each read against the
+    hidden ``response_cost`` the pipeline stamped on it."""
+    if isinstance(chunk, ModelResponseStream):
+        usage: Final = getattr(chunk, "usage", None)
+        if not isinstance(usage, Usage):
+            return chunk
+        cost: Final = _client_usage_cost(getattr(usage, "cost", None), _hidden_bill(chunk))
+        return chunk if cost is None else _with_usage_cost(chunk, usage, cost)
+    if getattr(chunk, "type", None) not in _RESPONSES_USAGE_EVENTS:
+        return chunk
+    response: Final = getattr(chunk, "response", None)
+    if not isinstance(response, ResponsesAPIResponse) or not isinstance(response.usage, ResponseAPIUsage):
+        return chunk
+    response_cost: Final = _client_usage_cost(getattr(response.usage, "cost", None), _hidden_bill(response))
+    if response_cost is None:
+        return chunk
+    return chunk.model_copy(update={"response": _with_usage_cost(response, response.usage, response_cost)})

litellm/responses/litellm_completion_transformation/streaming_iterator.py

@@ -25,7 +25,7 @@ from litellm.responses.litellm_completion_transformation.tool_search import (
 from litellm.responses.litellm_completion_transformation.transformation import (
     LiteLLMCompletionResponsesConfig,
 )
-from litellm.responses.streaming_iterator import ResponsesAPIStreamingIterator, stamp_responses_usage_cost
+from litellm.responses.streaming_iterator import ResponsesAPIStreamingIterator, normalize_responses_usage
 from litellm.responses.utils import ResponsesAPIRequestUtils
 from litellm.types.llms.openai import (
     PART_UNION_TYPES,
@@ -1481,7 +1481,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
                 }
             )
         )
-        stamp_responses_usage_cost(partial, self.litellm_logging_obj)
+        normalize_responses_usage(partial, self.litellm_logging_obj)
         return ResponseIncompleteEvent(type=ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE, response=partial)
 
     def _emit_terminal_response_event(

litellm/responses/streaming_iterator.py

@@ -41,6 +41,7 @@ from litellm.exceptions import MidStreamFallbackError, RateLimitError
 from litellm.litellm_core_utils.asyncify import run_async_function
 from litellm.litellm_core_utils.core_helpers import process_response_headers
 from litellm.litellm_core_utils.exception_mapping_utils import with_deployment_api_base
+from litellm.litellm_core_utils.hidden_params import get_hidden_params
 from litellm.litellm_core_utils.internal_call_metadata import (
     is_background_response,
     requests_background,
@@ -595,7 +596,7 @@ class BaseResponsesAPIStreamingIterator:
                         else openai_responses_api_chunk.model_copy(update={"response": _billed_response})
                     )
                     self.completed_response = _terminal_chunk
-                    stamp_responses_usage_cost(_billed_response, self.logging_obj)
+                    normalize_responses_usage(_billed_response, self.logging_obj)
 
                     if _chunk_type == openai_types.ResponsesAPIStreamEvents.RESPONSE_FAILED:
                         self._handle_logging_failed_response()
@@ -1123,7 +1124,7 @@ class BaseResponsesAPIStreamingIterator:
                 "output": _partial_output_items(self._generated_content),
             }
         )
-        stamp_responses_usage_cost(partial, self.logging_obj)
+        normalize_responses_usage(partial, self.logging_obj)
         return ResponseIncompleteEvent(type=ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE, response=partial)
 
     def _billed_without_disconnect(self) -> bool:
@@ -1822,23 +1823,24 @@ def _estimate_usage_safely(
         return None
 
 
-def stamp_responses_usage_cost(
-    response_obj: ResponsesAPIResponse | None, logging_obj: LiteLLMLoggingObj | None
-) -> None:
-    if response_obj is None or logging_obj is None:
+def normalize_responses_usage(response_obj: ResponsesAPIResponse | None, logging_obj: LiteLLMLoggingObj | None) -> None:
+    """A terminal response built off an untyped dict carries its usage as a dict, logging and the proxy read a model.
+    The bill the logging object prices for it rides its hidden ``response_cost``, so the proxy shows the same number."""
+    if response_obj is None:
         return
     usage_obj: Final[ResponseAPIUsage | None] = _usage_as_model(getattr(response_obj, "usage", None))
     if usage_obj is None:
         return
-    response_obj.usage = usage_obj  # rebind-ok: the stamped cost has to ride on the response the client receives
-    if isinstance(getattr(usage_obj, "cost", None), (int, float)):
+    response_obj.usage = usage_obj  # rebind-ok: the typed usage has to ride on the response the client receives
+    if logging_obj is None:
         return
     try:
-        cost: Final[float | None] = logging_obj.response_cost_calculator(result=response_obj)
-    except Exception:
+        bill: Final = logging_obj.response_cost_calculator(result=response_obj)
+    except Exception:  # noqa: BLE001  # a pricing failure leaves the client usage as the upstream sent it
         return
-    if isinstance(cost, (int, float)) and cost > 0:
-        setattr(usage_obj, "cost", cost)
+    hidden: Final = get_hidden_params(response_obj)
+    if hidden is not None and isinstance(bill, (int, float)) and not isinstance(bill, bool):
+        hidden["response_cost"] = float(bill)
 
 
 def opening_item_payload(payload: Mapping[str, object]) -> Mapping[str, object]:
@@ -1872,7 +1874,7 @@ def build_synthetic_response_events(
     chunk_size: int,
 ) -> list[ResponsesAPIStreamingResponse]:
     openai_types: Final = get_openai_response_types()
-    stamp_responses_usage_cost(transformed, logging_obj)
+    normalize_responses_usage(transformed, logging_obj)
 
     events: Final[list[ResponsesAPIStreamingResponse]] = [
         _build_response_status_event(openai_types.ResponsesAPIStreamEvents.RESPONSE_CREATED, transformed),

tests/integration/cost_calculation/cases/openai.py

@@ -2,6 +2,7 @@
 
 PARITY pairs the streamed case with its plain twin where both bill the same row."""
 
+from collections.abc import Mapping
 from typing import Final, cast
 
 from integration.cost_calculation.cost_tracking_case import (
@@ -11878,7 +11879,215 @@ GPT_5_6_TERRA_MESSAGES_BRIDGED_STREAM: Final = GPT_5_6_TERRA_MESSAGES_BRIDGED_JS
     }
 )
 
+
+# An OpenAI-compatible gateway that reports what it charged under usage.cost (#47533). The reported total is the bill,
+# split across input and output by the ratio of the cost map rates (100 * 1e-6 : 20 * 3e-6), on plain and streamed
+# replies alike. A deployment the operator priced keeps the operator's rates on both.
+_GATEWAY_REPORTED_USAGE: Final = {"prompt_tokens": 100, "completion_tokens": 20, "total_tokens": 120, "cost": 0.037}
+_GATEWAY_REPORTED_EXPECTED: Final = ExactExpected(
+    spend=0.037, input_cost=0.023125, output_cost=0.013875, prompt_tokens=100, completion_tokens=20
+)
+_GATEWAY_OPERATOR_PRICED: Final = Deployment(input_cost_per_token=2e-06, output_cost_per_token=6e-06)
+_GATEWAY_OPERATOR_PRICED_EXPECTED: Final = ExactExpected(
+    spend=0.00032, input_cost=0.0002, output_cost=0.00012, prompt_tokens=100, completion_tokens=20
+)
+_GATEWAY_CHUNK: Final = {
+    "id": "chatcmpl-$REQUEST_ID",
+    "object": "chat.completion.chunk",
+    "created": 1,
+    "model": "gateway-reported-cost-fixture",
+}
+
+GATEWAY_REPORTED_COST: Final = CostTrackingTestCase(
+    name="gateway-reported-cost",
+    covers="quota_management.spend_tracking.cost_matrix.logs_cost",
+    model="gateway-reported-cost-fixture",
+    endpoint="/v1/chat/completions",
+    request={"model": "$MODEL", "messages": [{"role": "user", "content": "reported cost"}], "stream": False},
+    response=JsonResponse(
+        content_type="application/json",
+        body={
+            "id": "chatcmpl-$REQUEST_ID",
+            "object": "chat.completion",
+            "created": 1,
+            "model": "gateway-reported-cost-fixture",
+            "choices": [{"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}],
+            "usage": _GATEWAY_REPORTED_USAGE,
+        },
+    ),
+    expected=_GATEWAY_REPORTED_EXPECTED,
+)
+
+GATEWAY_REPORTED_COST_STREAM: Final = CostTrackingTestCase(
+    name="gateway-reported-cost-stream",
+    covers="quota_management.spend_tracking.cost_matrix.logs_cost",
+    model="gateway-reported-cost-fixture",
+    endpoint="/v1/chat/completions",
+    request={
+        "model": "$MODEL",
+        "messages": [{"role": "user", "content": "reported cost"}],
+        "stream": True,
+        "stream_options": {"include_usage": True},
+    },
+    response=SseResponse(
+        content_type="text/event-stream",
+        frames=sse_frames(
+            {**_GATEWAY_CHUNK, "choices": [{"index": 0, "delta": {"role": "assistant", "content": "ok"}}]},
+            {**_GATEWAY_CHUNK, "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]},
+            {**_GATEWAY_CHUNK, "choices": [], "usage": _GATEWAY_REPORTED_USAGE},
+            done=True,
+        ),
+    ),
+    expected=_GATEWAY_REPORTED_EXPECTED,
+)
+
+GATEWAY_REPORTED_COST_OPERATOR_PRICED: Final = GATEWAY_REPORTED_COST.model_copy(
+    update={
+        "name": "gateway-reported-cost-operator_priced",
+        "deployment": _GATEWAY_OPERATOR_PRICED,
+        "expected": _GATEWAY_OPERATOR_PRICED_EXPECTED,
+    }
+)
+
+GATEWAY_REPORTED_COST_OPERATOR_PRICED_STREAM: Final = GATEWAY_REPORTED_COST_STREAM.model_copy(
+    update={
+        "name": "gateway-reported-cost-operator_priced-stream",
+        "deployment": _GATEWAY_OPERATOR_PRICED,
+        "expected": _GATEWAY_OPERATOR_PRICED_EXPECTED,
+    }
+)
+
+_GATEWAY_RESPONSES_USAGE: Final[Mapping[str, JsonValue]] = {"input_tokens": 100, "output_tokens": 20, "total_tokens": 120, "cost": 0.037}
+_GATEWAY_RESPONSES_ITEM: Final[Mapping[str, JsonValue]] = {
+    "id": "msg_gateway",
+    "type": "message",
+    "role": "assistant",
+    "status": "completed",
+    "content": [{"type": "output_text", "text": "ok", "annotations": []}],
+}
+_GATEWAY_RESPONSES_BODY: Final[Mapping[str, JsonValue]] = {
+    "id": "resp_$REQUEST_ID",
+    "object": "response",
+    "created_at": 1,
+    "status": "completed",
+    "model": "gateway-reported-cost-fixture",
+    "output": [_GATEWAY_RESPONSES_ITEM],
+    "usage": _GATEWAY_RESPONSES_USAGE,
+}
+
+GATEWAY_REPORTED_COST_RESPONSES: Final = CostTrackingTestCase(
+    name="gateway-reported-cost-responses",
+    covers="quota_management.spend_tracking.cost_matrix.logs_cost",
+    model="gateway-reported-cost-fixture",
+    endpoint="/v1/responses",
+    request={"model": "$MODEL", "input": "reported cost", "cache": {"no-cache": True}},
+    response=JsonResponse(content_type="application/json", body=_GATEWAY_RESPONSES_BODY),
+    expected=_GATEWAY_REPORTED_EXPECTED,
+)
+
+GATEWAY_REPORTED_COST_RESPONSES_STREAM: Final = GATEWAY_REPORTED_COST_RESPONSES.model_copy(
+    update={
+        "name": "gateway-reported-cost-responses-stream",
+        "request": {**GATEWAY_REPORTED_COST_RESPONSES.request, "stream": True},
+        "response": SseResponse(
+            content_type="text/event-stream",
+            frames=sse_frames(
+                {
+                    "type": "response.created",
+                    "sequence_number": 0,
+                    "response": {**_GATEWAY_RESPONSES_BODY, "status": "in_progress", "output": [], "usage": None},
+                },
+                {
+                    "type": "response.output_text.delta",
+                    "sequence_number": 1,
+                    "item_id": "msg_gateway",
+                    "output_index": 0,
+                    "content_index": 0,
+                    "delta": "ok",
+                },
+                {"type": "response.completed", "sequence_number": 2, "response": _GATEWAY_RESPONSES_BODY},
+            ),
+        ),
+    }
+)
+
+GATEWAY_REPORTED_COST_RESPONSES_OPERATOR_PRICED: Final = GATEWAY_REPORTED_COST_RESPONSES.model_copy(
+    update={
+        "name": "gateway-reported-cost-responses-operator_priced",
+        "deployment": _GATEWAY_OPERATOR_PRICED,
+        "expected": _GATEWAY_OPERATOR_PRICED_EXPECTED,
+    }
+)
+
+GATEWAY_REPORTED_COST_RESPONSES_OPERATOR_PRICED_STREAM: Final = GATEWAY_REPORTED_COST_RESPONSES_STREAM.model_copy(
+    update={
+        "name": "gateway-reported-cost-responses-operator_priced-stream",
+        "deployment": _GATEWAY_OPERATOR_PRICED,
+        "expected": _GATEWAY_OPERATOR_PRICED_EXPECTED,
+    }
+)
+
+
+_GATEWAY_ZERO_USAGE: Final = {**_GATEWAY_REPORTED_USAGE, "cost": 0}
+
+GATEWAY_REPORTED_ZERO_OPERATOR_PRICED: Final = GATEWAY_REPORTED_COST_OPERATOR_PRICED.model_copy(
+    update={
+        "name": "gateway-reported-zero-operator_priced",
+        "response": JsonResponse(
+            content_type="application/json",
+            body={**GATEWAY_REPORTED_COST.response.body, "usage": _GATEWAY_ZERO_USAGE},
+        ),
+    }
+)
+
+GATEWAY_REPORTED_ZERO_OPERATOR_PRICED_STREAM: Final = GATEWAY_REPORTED_COST_OPERATOR_PRICED_STREAM.model_copy(
+    update={
+        "name": "gateway-reported-zero-operator_priced-stream",
+        "response": SseResponse(
+            content_type="text/event-stream",
+            frames=sse_frames(
+                {**_GATEWAY_CHUNK, "choices": [{"index": 0, "delta": {"role": "assistant", "content": "ok"}}]},
+                {**_GATEWAY_CHUNK, "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]},
+                {**_GATEWAY_CHUNK, "choices": [], "usage": _GATEWAY_ZERO_USAGE},
+                done=True,
+            ),
+        ),
+    }
+)
+
+_GATEWAY_OPENROUTER_OPERATOR_PRICED: Final = _GATEWAY_OPERATOR_PRICED.model_copy(
+    update={"model": "openrouter/vendor/gateway-reported-cost-fixture"}
+)
+
+GATEWAY_REPORTED_COST_OPENROUTER_OPERATOR_PRICED: Final = GATEWAY_REPORTED_COST_OPERATOR_PRICED.model_copy(
+    update={
+        "name": "gateway-reported-cost-openrouter-operator_priced",
+        "deployment": _GATEWAY_OPENROUTER_OPERATOR_PRICED,
+    }
+)
+
+GATEWAY_REPORTED_COST_OPENROUTER_OPERATOR_PRICED_STREAM: Final = (
+    GATEWAY_REPORTED_COST_OPERATOR_PRICED_STREAM.model_copy(
+        update={
+            "name": "gateway-reported-cost-openrouter-operator_priced-stream",
+            "deployment": _GATEWAY_OPENROUTER_OPERATOR_PRICED,
+        }
+    )
+)
+
 CASES: Final[tuple[CostTrackingTestCase, ...]] = (
+    GATEWAY_REPORTED_COST,
+    GATEWAY_REPORTED_COST_STREAM,
+    GATEWAY_REPORTED_COST_OPERATOR_PRICED,
+    GATEWAY_REPORTED_COST_OPERATOR_PRICED_STREAM,
+    GATEWAY_REPORTED_COST_RESPONSES,
+    GATEWAY_REPORTED_COST_RESPONSES_STREAM,
+    GATEWAY_REPORTED_COST_RESPONSES_OPERATOR_PRICED,
+    GATEWAY_REPORTED_COST_RESPONSES_OPERATOR_PRICED_STREAM,
+    GATEWAY_REPORTED_ZERO_OPERATOR_PRICED,
+    GATEWAY_REPORTED_ZERO_OPERATOR_PRICED_STREAM,
+    GATEWAY_REPORTED_COST_OPENROUTER_OPERATOR_PRICED,
+    GATEWAY_REPORTED_COST_OPENROUTER_OPERATOR_PRICED_STREAM,
     GPT_5_6_SLASH_ALIAS,
     GPT_5_6_SLASH_ALIAS_STREAM,
     GPT_6_SOL_RESPONSES_DEPLOYMENT_JSON,
@@ -12120,6 +12329,22 @@ CASES: Final[tuple[CostTrackingTestCase, ...]] = (
 )
 
 PARITY: Final[tuple[StreamParityTestCase, ...]] = (
+    StreamParityTestCase(plain=GATEWAY_REPORTED_COST, streamed=GATEWAY_REPORTED_COST_STREAM),
+    StreamParityTestCase(
+        plain=GATEWAY_REPORTED_COST_OPERATOR_PRICED, streamed=GATEWAY_REPORTED_COST_OPERATOR_PRICED_STREAM
+    ),
+    StreamParityTestCase(
+        plain=GATEWAY_REPORTED_ZERO_OPERATOR_PRICED, streamed=GATEWAY_REPORTED_ZERO_OPERATOR_PRICED_STREAM
+    ),
+    StreamParityTestCase(
+        plain=GATEWAY_REPORTED_COST_OPENROUTER_OPERATOR_PRICED,
+        streamed=GATEWAY_REPORTED_COST_OPENROUTER_OPERATOR_PRICED_STREAM,
+    ),
+    StreamParityTestCase(plain=GATEWAY_REPORTED_COST_RESPONSES, streamed=GATEWAY_REPORTED_COST_RESPONSES_STREAM),
+    StreamParityTestCase(
+        plain=GATEWAY_REPORTED_COST_RESPONSES_OPERATOR_PRICED,
+        streamed=GATEWAY_REPORTED_COST_RESPONSES_OPERATOR_PRICED_STREAM,
+    ),
     StreamParityTestCase(plain=GPT_5_3_CODEX_INPUT_TEXT, streamed=GPT_5_3_CODEX_STREAM),
     StreamParityTestCase(plain=GPT_5_3_CODEX_INPUT_TEXT, streamed=GPT_5_3_CODEX_STREAM_INCOMPLETE),
     StreamParityTestCase(plain=GPT_5_3_CODEX_INPUT_TEXT, streamed=GPT_5_3_CODEX_STREAM_UNVALIDATED),

tests/integration/cost_calculation/cost_map.py

@@ -26,6 +26,12 @@ COST_MAP_ENTRIES: Final[dict[str, dict[str, JsonValue]]] = {
         "input_cost_per_token_above_128k_tokens": 2.2e-07,
         "output_cost_per_token_above_128k_tokens": 8.8e-07,
     },
+    "gateway-reported-cost-fixture": {
+        "litellm_provider": "openai",
+        "mode": "chat",
+        "input_cost_per_token": 1e-06,
+        "output_cost_per_token": 3e-06,
+    },
     "openrouter/anthropic/claude-sonnet-5": {
         "litellm_provider": "openrouter",
         "mode": "chat",

tests/unit/litellm_core_utils/test_streaming_handler.py

@@ -9,7 +9,7 @@ import time
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from types import CodeType, FrameType, MappingProxyType
-from typing import AsyncIterator, Callable, Final, Iterator, Literal, NoReturn, Optional, cast
+from typing import AsyncIterator, Callable, Final, Iterator, Literal, NoReturn, Optional, Tuple, cast
 from unittest.mock import AsyncMock, MagicMock, Mock, patch
 
 import httpx
@@ -17,8 +17,9 @@ import pytest
 from openai import AsyncOpenAI, AsyncStream, OpenAI, Stream
 from openai.types.chat import ChatCompletionChunk
 from pydantic import BaseModel, TypeAdapter
+
 import litellm
-from litellm import verbose_logger
+from litellm import completion, verbose_logger
 from litellm._logging import session_id_var, trace_id_var
 from litellm.exceptions import MidStreamFallbackError
 from litellm.integrations.custom_logger import CustomLogger
@@ -29,7 +30,6 @@ from litellm.litellm_core_utils.streaming_handler import (
     _ProviderChunkEarlyReturn,
     _ProviderChunkParsed,
 )
-from litellm.llms.custom_httpx.http_handler import HTTPHandler
 from litellm.llms.openai.chat.gpt_transformation import OpenAIChatCompletionStreamingHandler
 from litellm.responses.mcp.chat_completions_handler import _add_mcp_metadata_to_response
 from litellm.types.utils import (
@@ -38,13 +38,11 @@ from litellm.types.utils import (
     ModelResponse,
     ModelResponseStream,
     PromptTokensDetailsWrapper,
-    StreamingChoices,
     StandardLoggingPayload,
+    StreamingChoices,
     Usage,
 )
 from litellm.utils import ModelResponseListIterator
-from litellm import completion
-from typing import Tuple
 
 
 @pytest.mark.parametrize("async_stream", [False, True])
@@ -2547,14 +2545,14 @@ def test_perplexity_streaming_dict_cost_bills_through_its_own_calculator():
     ) == pytest.approx(0.00503)
 
 
-def test_openai_compatible_streaming_cost_is_priced_from_the_cost_map():
+def test_openai_compatible_streaming_cost_is_the_reported_cost():
     import litellm
     from litellm.cost_calculator import (
         get_response_cost_from_hidden_params,
         response_cost_calculator,
     )
 
-    model = "openai/streams-cost-in-nanodollars"
+    model = "openai/streams-with-a-reported-cost"
     litellm.register_model(
         {
             model: {
@@ -2569,7 +2567,7 @@ def test_openai_compatible_streaming_cost_is_priced_from_the_cost_map():
         id="chatcmpl-openai-compatible",
         model=model,
         choices=[],
-        usage=Usage(completion_tokens=5, prompt_tokens=10, total_tokens=15, cost=3_144_000),
+        usage=Usage(completion_tokens=5, prompt_tokens=10, total_tokens=15, cost=0.037),
     )
 
     CustomStreamWrapper._propagate_usage_cost_to_hidden_params(complete_response, "openai")
@@ -2581,7 +2579,7 @@ def test_openai_compatible_streaming_cost_is_priced_from_the_cost_map():
         custom_llm_provider="openai",
         call_type="completion",
         optional_params={},
-    ) == pytest.approx(2e-5)
+    ) == 0.037
 
 
 def test_xai_streaming_reported_cost_still_takes_the_margin(monkeypatch):
@@ -6204,6 +6202,25 @@ def test_chunk_creator_preserves_hidden_provider_specific_fields_from_parsed_chu
     assert assembled._hidden_params["provider_specific_fields"] == {"traffic_type": "ON_DEMAND_FLEX"}
 
 
+def test_the_usage_chunk_carries_the_bill_the_builder_priced_for_the_spend_row() -> None:
+    from litellm.litellm_core_utils.hidden_params import get_hidden_params
+    from litellm.litellm_core_utils.streaming_handler import (
+        _stamp_hidden_response_cost,  # pyright: ignore[reportPrivateUsage]  # the helper under test
+    )
+
+    usage: Final = Usage(prompt_tokens=100, completion_tokens=20, total_tokens=120)
+    complete: Final = ModelResponse(model="omni-model", usage=usage)
+    complete.hidden_params = {"response_cost": 0.00032}
+    usage_chunk: Final = ModelResponseStream(id="chunk-1", model="omni-model", choices=[], usage=usage)
+    usage_chunk.hidden_params = {"response_cost": None}
+
+    _stamp_hidden_response_cost(usage_chunk, complete)
+
+    stamped: Final = get_hidden_params(usage_chunk)
+    assert stamped is not None and stamped["response_cost"] == 0.00032
+    assert getattr(usage, "cost", None) is None
+
+
 def test_chunk_creator_keeps_provider_model_private_across_stream():
     from litellm.router_utils.add_retry_fallback_headers import (
         get_hidden_params_dict,

tests/unit/llms/azure_ai/test_azure_ai_cost_calculator.py

@@ -14,7 +14,6 @@ import litellm
 from litellm.cost_calculator import completion_cost
 from litellm.litellm_core_utils.litellm_logging import Logging
 from litellm.llms.azure_ai.cost_calculator import (
-    calculate_azure_model_router_flat_cost,
     cost_per_token,
     is_azure_model_router,
 )
@@ -263,7 +262,14 @@ class TestAzureModelRouterCostBreakdown:
         assert route.call_count == 1
         assert json.loads(route.calls.last.request.content)["model"] == "model-router"
         assert (final_usage.prompt_tokens, final_usage.completion_tokens) == (5000, 2000)
-        assert final_usage.cost == pytest.approx(routed_prompt_cost + routed_completion_cost + ROUTED_FEE, rel=1e-9)
+        assert getattr(final_usage, "cost", None) is None, "the bill reaches a client through the proxy egress only"
+        assembled: Final = litellm.stream_chunk_builder(
+            chunks=list(chunks), messages=messages, logging_obj=stream.logging_obj
+        )
+        assert assembled is not None
+        assert assembled._hidden_params["response_cost"] == pytest.approx(
+            routed_prompt_cost + routed_completion_cost + ROUTED_FEE, rel=1e-9
+        )
 
     @pytest.mark.respx(assert_all_called=True)
     def test_model_router_completion_reports_routed_cost_plus_flat_fee(self, respx_mock: respx.MockRouter) -> None:

tests/unit/proxy/test_usage_cost_egress.py

@@ -0,0 +1,126 @@
+from __future__ import annotations
+
+from typing import Final
+
+import pytest
+
+from litellm.litellm_core_utils.hidden_params import get_hidden_params
+from litellm.proxy.usage_cost_egress import with_client_stream_usage_cost
+from litellm.types.llms.openai import (
+    ResponseAPIUsage,
+    ResponseCompletedEvent,
+    ResponseIncompleteEvent,
+    ResponsesAPIResponse,
+    ResponsesAPIStreamEvents,
+)
+from litellm.types.utils import Delta, ModelResponseStream, StreamingChoices, Usage
+
+UPSTREAM_COST: Final = 0.037
+OPERATOR_BILL: Final = 100 * 2e-6 + 20 * 6e-6
+
+
+def _responses_reply(upstream_cost: float | None, bill: object) -> ResponsesAPIResponse:
+    reply: Final = ResponsesAPIResponse(
+        id="resp_egress",
+        created_at=1,
+        status="completed",
+        model="omni-model",
+        object="response",
+        output=[],
+        usage=ResponseAPIUsage(input_tokens=100, output_tokens=20, total_tokens=120, cost=upstream_cost),
+    )
+    hidden: Final = get_hidden_params(reply)
+    assert hidden is not None
+    hidden["response_cost"] = bill
+    return reply
+
+
+def _usage_chunk(upstream_cost: float | None, bill: object) -> ModelResponseStream:
+    chunk: Final = ModelResponseStream(
+        id="chatcmpl-egress",
+        model="omni-model",
+        choices=[],
+        usage=Usage(prompt_tokens=100, completion_tokens=20, total_tokens=120, cost=upstream_cost),
+    )
+    chunk.hidden_params = {"response_cost": bill}
+    return chunk
+
+
+def test_the_streamed_usage_chunk_tells_the_client_what_litellm_billed() -> None:
+    chunk: Final = _usage_chunk(UPSTREAM_COST, OPERATOR_BILL)
+
+    billed: Final = with_client_stream_usage_cost(chunk)
+
+    assert isinstance(billed, ModelResponseStream) and billed is not chunk
+    assert billed.usage is not None and billed.usage.cost == OPERATOR_BILL
+    assert chunk.usage is not None and chunk.usage.cost == UPSTREAM_COST
+
+
+def test_a_zero_bill_replaces_the_upstream_cost() -> None:
+    billed: Final = with_client_stream_usage_cost(_usage_chunk(UPSTREAM_COST, 0.0))
+
+    assert isinstance(billed, ModelResponseStream)
+    assert billed.usage is not None and billed.usage.cost == 0.0
+
+
+@pytest.mark.parametrize("bill", [None, -0.001, float("nan"), float("inf"), True])
+def test_an_unpriced_stream_keeps_the_upstream_cost(bill: object) -> None:
+    chunk: Final = _usage_chunk(UPSTREAM_COST, bill)
+
+    assert with_client_stream_usage_cost(chunk) is chunk
+    assert chunk.usage is not None and chunk.usage.cost == UPSTREAM_COST
+
+
+def test_a_priced_stream_without_an_upstream_cost_still_tells_the_client_the_bill() -> None:
+    billed: Final = with_client_stream_usage_cost(_usage_chunk(None, OPERATOR_BILL))
+
+    assert isinstance(billed, ModelResponseStream)
+    assert billed.usage is not None and billed.usage.cost == OPERATOR_BILL
+
+
+@pytest.mark.parametrize("bill", [None, 0.0])
+def test_an_unpriced_stream_without_an_upstream_cost_goes_out_as_sent(bill: float | None) -> None:
+    chunk: Final = _usage_chunk(None, bill)
+
+    assert with_client_stream_usage_cost(chunk) is chunk
+
+
+def test_a_content_chunk_is_not_touched() -> None:
+    chunk: Final = ModelResponseStream(
+        id="chatcmpl-egress",
+        model="omni-model",
+        choices=[StreamingChoices(index=0, delta=Delta(content="hi"), finish_reason=None)],
+    )
+    chunk.hidden_params = {"response_cost": OPERATOR_BILL}
+
+    assert with_client_stream_usage_cost(chunk) is chunk
+
+
+@pytest.mark.parametrize("upstream_cost", [UPSTREAM_COST, None])
+@pytest.mark.parametrize(
+    "event_type", [ResponsesAPIStreamEvents.RESPONSE_COMPLETED, ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE]
+)
+def test_a_terminal_responses_event_carries_the_bill_with_or_without_an_upstream_cost(
+    upstream_cost: float | None, event_type: ResponsesAPIStreamEvents
+) -> None:
+    reply: Final = _responses_reply(upstream_cost, OPERATOR_BILL)
+    event: Final = (
+        ResponseCompletedEvent(type=ResponsesAPIStreamEvents.RESPONSE_COMPLETED, response=reply)
+        if event_type is ResponsesAPIStreamEvents.RESPONSE_COMPLETED
+        else ResponseIncompleteEvent(type=ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE, response=reply)
+    )
+
+    billed: Final = with_client_stream_usage_cost(event)
+
+    assert billed is not event
+    assert isinstance(billed, (ResponseCompletedEvent, ResponseIncompleteEvent))
+    assert billed.response.usage is not None and billed.response.usage.cost == OPERATOR_BILL
+    assert reply.usage is not None and reply.usage.cost == upstream_cost
+
+
+def test_an_unpriced_terminal_responses_event_goes_out_as_sent() -> None:
+    event: Final = ResponseCompletedEvent(
+        type=ResponsesAPIStreamEvents.RESPONSE_COMPLETED, response=_responses_reply(UPSTREAM_COST, None)
+    )
+
+    assert with_client_stream_usage_cost(event) is event

tests/unit/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py

@@ -10,6 +10,7 @@ before response.completed, and that every event of a bridged stream carries the
 spend tracking stores, so a follow-up previous_response_id still finds the conversation.
 """
 
+import itertools
 import json
 from collections.abc import Callable
 from typing import Final
@@ -23,6 +24,7 @@ from litellm.responses.litellm_completion_transformation.streaming_iterator impo
 from litellm.responses.utils import ResponsesAPIRequestUtils
 from litellm.types.llms.openai import (
     BaseLiteLLMOpenAIResponseObject,
+    OutputTextDeltaEvent,
     ResponsesAPIStreamEvents,
 )
 from litellm.types.responses.main import build_web_search_call
@@ -1558,3 +1560,29 @@ async def test_a_compaction_block_without_its_summary_streams_no_compaction_item
 
     assert [event["item"]["type"] for event in events if "item" in event] == ["message", "message"]
     assert [item["type"] for item in events[-1]["response"]["output"]] == ["message"]
+
+
+def test_the_partial_response_billed_on_disconnect_carries_the_logging_objects_bill_as_hidden_response_cost() -> None:
+    from litellm.litellm_core_utils.hidden_params import get_hidden_params
+
+    wrapper: Final = _FakeStreamWrapper([_chunk("Hello"), _chunk(" world", finish_reason="stop")])
+    wrapper.logging_obj.response_cost_calculator.return_value = 0.00032
+    iterator: Final = LiteLLMCompletionStreamingIterator(
+        model="claude-haiku-4-5",
+        litellm_custom_stream_wrapper=wrapper,  # pyright: ignore[reportArgumentType]  # the fake stands in for CustomStreamWrapper, as in _build_iterator
+        request_input="What is the weather in San Francisco?",
+        responses_api_request={},
+        custom_llm_provider="anthropic",
+        litellm_metadata={},
+    )
+    relayed: Final = tuple(
+        itertools.takewhile(lambda event: not isinstance(event, OutputTextDeltaEvent), iterator)
+    )
+    assert relayed, "the stream must relay at least one chunk before the client goes away"
+
+    partial: Final = iterator.partial_response_on_disconnect()
+
+    assert partial is not None
+    hidden: Final = get_hidden_params(partial.response)
+    assert hidden is not None and hidden["response_cost"] == 0.00032
+    assert wrapper.logging_obj.response_cost_calculator.call_args.kwargs["result"] is partial.response

tests/unit/responses/test_streaming_iterator.py

@@ -3,15 +3,18 @@ completion_start_time on the first chunk so downstream TTFT consumers
 (Prometheus, OTEL, SpendLogs completionStartTime) do not fall back to
 completion_start_time = end_time."""
 
-import asyncio, importlib
+import asyncio
 import functools
 import gc
+import importlib
 import json
 import threading
 import weakref
 from collections.abc import Callable, Mapping
+from contextlib import suppress
 from datetime import datetime
-from typing import Final, Optional
+from types import SimpleNamespace
+from typing import Final
 from unittest.mock import AsyncMock, MagicMock, Mock, patch
 
 import httpx
@@ -22,9 +25,11 @@ from pydantic_core import PydanticSerializationError
 import litellm
 from litellm.exceptions import MidStreamFallbackError
 from litellm.integrations.custom_logger import CustomLogger
+from litellm.litellm_core_utils.hidden_params import get_hidden_params
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
 from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig
-from litellm.responses.streaming_iterator import(
+from litellm.responses import streaming_iterator as streaming_module
+from litellm.responses.streaming_iterator import (
     CachedResponsesAPIStreamingIterator,
     MockResponsesAPIStreamingIterator,
     ResponsesAPIStreamingIterator,
@@ -38,15 +43,12 @@ from litellm.types.llms.openai import (
     ResponsesAPIStreamEvents,
     ResponsesAPIStreamingResponse,
 )
-from contextlib import suppress
-from litellm.responses import streaming_iterator as streaming_module
 from litellm.types.utils import CallTypes
 from tests._vcr_conftest_common import install_live_call_probe, record_vcr_outcome
-from types import SimpleNamespace
 
 
 def _sse_event(payload: dict) -> bytes:
-    return f"data: {json.dumps(payload)}\n\n".encode("utf-8")
+    return f"data: {json.dumps(payload)}\n\n".encode()
 
 
 def _mock_config() -> Mock:
@@ -80,7 +82,7 @@ def _make_iterator(
     *,
     sse_events: list[bytes],
     logging_obj: LiteLLMLoggingObj,
-    trailing_error: Optional[Exception] = None,
+    trailing_error: Exception | None = None,
     config: Mock | None = None,
     request_data: dict | None = None,
 ) -> ResponsesAPIStreamingIterator:
@@ -109,7 +111,7 @@ def _make_sync_iterator(
     *,
     sse_events: list[bytes],
     logging_obj: LiteLLMLoggingObj,
-    trailing_error: Optional[Exception] = None,
+    trailing_error: Exception | None = None,
 ) -> SyncResponsesAPIStreamingIterator:
     def iter_bytes():
         for evt in sse_events:
@@ -491,7 +493,6 @@ class _LoopRecordingLogger(CustomLogger):
         self.hook_loop = asyncio.get_running_loop()
         await asyncio.sleep(0.05)
         self.hook_finished = True
-        return None
 
 
 class _SyncOnlyRecordingLogger(CustomLogger):
@@ -840,32 +841,6 @@ def _responses_api_response_with_usage() -> ResponsesAPIResponse:
     )
 
 
-def test_stamp_responses_usage_cost_stamps_computed_cost():
-    from litellm.responses.streaming_iterator import stamp_responses_usage_cost
-
-    response = _responses_api_response_with_usage()
-    logging_obj = Mock(spec=LiteLLMLoggingObj)
-    logging_obj.response_cost_calculator.return_value = 0.000704
-
-    stamp_responses_usage_cost(response, logging_obj)
-
-    assert getattr(response.usage, "cost", None) == pytest.approx(0.000704)
-    logging_obj.response_cost_calculator.assert_called_once_with(result=response)
-
-
-def test_stamp_responses_usage_cost_keeps_provider_reported_cost():
-    from litellm.responses.streaming_iterator import stamp_responses_usage_cost
-
-    response = _responses_api_response_with_usage()
-    setattr(response.usage, "cost", 0.5)
-    logging_obj = Mock(spec=LiteLLMLoggingObj)
-
-    stamp_responses_usage_cost(response, logging_obj)
-
-    assert getattr(response.usage, "cost", None) == pytest.approx(0.5)
-    logging_obj.response_cost_calculator.assert_not_called()
-
-
 def _unvalidated_response_with_dict_usage(usage: dict) -> ResponsesAPIResponse:
     return ResponsesAPIResponse.model_construct(
         id="resp_lit7391",
@@ -879,8 +854,9 @@ def _unvalidated_response_with_dict_usage(usage: dict) -> ResponsesAPIResponse:
     )
 
 
-def test_stamp_responses_usage_cost_keeps_provider_cost_from_dict_usage():
-    from litellm.responses.streaming_iterator import stamp_responses_usage_cost
+def test_normalize_responses_usage_types_a_dict_usage_and_keeps_the_provider_cost():
+    from litellm.responses.streaming_iterator import normalize_responses_usage
+
     response = _unvalidated_response_with_dict_usage(
         {
             "input_tokens": 29,
@@ -890,41 +866,52 @@ def test_stamp_responses_usage_cost_keeps_provider_cost_from_dict_usage():
             "cost": {"currency": "USD", "input_cost": 0, "output_cost": 3e-05, "total_cost": 3e-05},
         }
     )
-    logging_obj = Mock(spec=LiteLLMLoggingObj)
 
-    stamp_responses_usage_cost(response, logging_obj)
+    normalize_responses_usage(response, None)
 
     assert isinstance(response.usage, ResponseAPIUsage)
     assert response.usage.cost == pytest.approx(3e-05)
     assert response.usage.output_tokens_details.reasoning_tokens == 117
-    logging_obj.response_cost_calculator.assert_not_called()
 
 
-def test_stamp_responses_usage_cost_computes_cost_for_dict_usage_without_cost():
-    from litellm.responses.streaming_iterator import stamp_responses_usage_cost
-    response = _unvalidated_response_with_dict_usage({"input_tokens": 29, "output_tokens": 120, "total_tokens": 149})
-    logging_obj = Mock(spec=LiteLLMLoggingObj)
-    logging_obj.response_cost_calculator.return_value = 0.000704
-
-    stamp_responses_usage_cost(response, logging_obj)
-
-    assert isinstance(response.usage, ResponseAPIUsage)
-    assert response.usage.cost == pytest.approx(0.000704)
-    logging_obj.response_cost_calculator.assert_called_once_with(result=response)
-
-
-def test_stamp_responses_usage_cost_survives_calculator_failure():
-    from litellm.responses.streaming_iterator import stamp_responses_usage_cost
+def test_normalize_responses_usage_never_writes_a_cost_the_provider_did_not_report():
+    from litellm.responses.streaming_iterator import normalize_responses_usage
 
     response = _responses_api_response_with_usage()
-    logging_obj = Mock(spec=LiteLLMLoggingObj)
-    logging_obj.response_cost_calculator.side_effect = RuntimeError("cost map unavailable")
 
-    stamp_responses_usage_cost(response, logging_obj)
+    normalize_responses_usage(response, None)
 
     assert getattr(response.usage, "cost", None) is None
 
 
+def test_normalize_responses_usage_stamps_the_logging_objects_bill_as_hidden_response_cost():
+    from litellm.responses.streaming_iterator import normalize_responses_usage
+
+    from litellm.litellm_core_utils.litellm_logging import Logging
+
+    response: Final = _responses_api_response_with_usage()
+    logging_obj: Final = Logging(
+        model="mantle-claude",
+        messages=[{"role": "user", "content": "q"}],
+        stream=True,
+        call_type="aresponses",
+        start_time=None,
+        litellm_call_id="call",
+        function_id="fn",
+    )
+    logging_obj.update_environment_variables(
+        litellm_params={"custom_llm_provider": "openai", "input_cost_per_token": 2e-6, "output_cost_per_token": 6e-6},
+        optional_params={},
+    )
+    logging_obj.custom_cost_per_token = {"input_cost_per_token": 2e-6, "output_cost_per_token": 6e-6}
+
+    normalize_responses_usage(response, logging_obj)
+
+    hidden: Final = get_hidden_params(response)
+    assert hidden is not None and hidden["response_cost"] == pytest.approx(20 * 2e-6 + 60 * 6e-6)
+    assert getattr(response.usage, "cost", None) is None, "the client field is written at the proxy egress, not here"
+
+
 def _capture_dispatch(logged: list):
     """Record the object handed to the success handlers.
 
@@ -942,7 +929,7 @@ def _capture_dispatch(logged: list):
     return _dispatch
 
 
-def _headers_config(*, transform_hidden_params: Optional[dict] = None) -> Mock:
+def _headers_config(*, transform_hidden_params: Mapping[str, object] | None = None) -> Mock:
     """Config whose completed event carries a real ResponsesAPIResponse, so the logging copy
     performs a genuine model_dump/model_validate round trip."""
     mock_config = Mock(spec=BaseResponsesAPIConfig)
@@ -1259,10 +1246,10 @@ async def test_completed_event_without_usage_counts_tool_input_deltas(tool_delta
 
 
 @pytest.mark.asyncio
-async def test_completed_event_with_a_dict_response_is_typed_and_billed():
+async def test_completed_event_with_a_dict_response_is_typed():
     """transform_streaming_response can model_construct a terminal event whose
     response stays a plain dict; the iterator must type it so the estimated
-    usage reaches the cost stamping path."""
+    usage reaches logging and the proxy egress as a model."""
     dict_response: Final = {
         "id": "resp_dict",
         "model": "gpt-4o-mini",
@@ -1304,8 +1291,7 @@ async def test_completed_event_with_a_dict_response_is_typed_and_billed():
     assert usage is not None
     assert usage.input_tokens > 0
     assert usage.output_tokens > 0
-    assert usage.cost == pytest.approx(0.000704)
-    logging_obj.response_cost_calculator.assert_any_call(result=completed_response)
+    assert getattr(usage, "cost", None) is None, "the bill goes on at the proxy egress, never on the event logging prices"
 
 
 def test_billed_terminal_response_keeps_a_response_that_already_has_usage():
@@ -1375,6 +1361,7 @@ def _vcr_outcome_gate(request, vcr):
     yield
     record_vcr_outcome(request, vcr)
 
+
 @pytest.fixture(scope="function")
 def setup_and_teardown():
     """
@@ -1392,6 +1379,7 @@ def setup_and_teardown():
     loop.close()
     asyncio.set_event_loop(None)
 
+
 class _FakeLoggingObj:
     def __init__(self):
         self.success_calls = 0
@@ -1432,10 +1420,12 @@ class _FakeLoggingObj:
         self.completion_start_time = completion_start_time
         self.model_call_details["completion_start_time"] = completion_start_time
 
+
 async def _await_tasks_spawned_since(existing: set["asyncio.Task[object]"]) -> None:
     spawned: Final = asyncio.all_tasks() - existing
     await asyncio.wait_for(asyncio.gather(*spawned, return_exceptions=True), timeout=5)
 
+
 def _make_completed_response(response_id: str = "resp_test") -> ResponseCompletedEvent:
     return ResponseCompletedEvent(
         type=ResponsesAPIStreamEvents.RESPONSE_COMPLETED,
@@ -1463,6 +1453,7 @@ def _make_completed_response(response_id: str = "resp_test") -> ResponseComplete
         ),
     )
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
 @pytest.mark.asyncio
 async def test_log_background_task_failure_logs_task_exceptions(monkeypatch):
@@ -1485,6 +1476,7 @@ async def test_log_background_task_failure_logs_task_exceptions(monkeypatch):
         task.exception(),
     )
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
 @pytest.mark.asyncio
 async def test_log_background_task_failure_ignores_cancelled_tasks(monkeypatch):
@@ -1500,6 +1492,7 @@ async def test_log_background_task_failure_ignores_cancelled_tasks(monkeypatch):
 
     error_logger.assert_not_called()
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
 def test_content_part_done_event_supports_refusal_and_reasoning_text():
     refusal_event = streaming_module._build_content_part_done_event(
@@ -1527,6 +1520,7 @@ def test_content_part_done_event_supports_refusal_and_reasoning_text():
     assert reasoning_event.part.reasoning == "because"
     assert unsupported_event is None
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
 def test_dump_response_object_handles_model_and_unknown_values():
     response = ResponsesAPIResponse(
@@ -1542,6 +1536,7 @@ def test_dump_response_object_handles_model_and_unknown_values():
     assert streaming_module._dump_response_object({"type": "message"}) == {"type": "message"}
     assert streaming_module._dump_response_object(object()) == {}
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
 @pytest.mark.asyncio
 async def test_responses_streaming_triggers_hooks(monkeypatch):
@@ -1598,6 +1593,7 @@ async def test_responses_streaming_triggers_hooks(monkeypatch):
     assert seen["request_data"].get("litellm_params") is not None
     assert seen["call_type"] == CallTypes.responses
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
 @pytest.mark.asyncio
 async def test_responses_streaming_calls_post_streaming_deployment_hook(monkeypatch):
@@ -1638,6 +1634,7 @@ async def test_responses_streaming_calls_post_streaming_deployment_hook(monkeypa
     # reset callbacks
     litellm.callbacks = original_callbacks
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
 @pytest.mark.asyncio
 async def test_responses_streaming_failure_triggers_failure_handlers():
@@ -1667,6 +1664,7 @@ async def test_responses_streaming_failure_triggers_failure_handlers():
     assert logging_obj.failure_calls >= 1
     assert logging_obj.async_failure_calls >= 1
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
 def test_process_chunk_requires_provider_config():
     iterator = ResponsesAPIStreamingIterator(
@@ -1681,6 +1679,7 @@ def test_process_chunk_requires_provider_config():
     with pytest.raises(ValueError, match="responses_api_provider_config is required"):
         iterator._process_chunk(json.dumps({"type": "response.completed"}))
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
 def test_process_chunk_wraps_encrypted_content_with_model_id():
     openai_types = streaming_module._get_openai_response_types()
@@ -1715,8 +1714,9 @@ def test_process_chunk_wraps_encrypted_content_with_model_id():
     assert event.item.encrypted_content.startswith("litellm_enc:")
     assert event.item.encrypted_content.endswith(";ciphertext")
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
-def test_process_chunk_completed_response_updates_id_and_usage_cost(monkeypatch):
+def test_process_chunk_completed_response_updates_id_and_leaves_usage_cost_alone(monkeypatch: pytest.MonkeyPatch):
     original_include_cost = litellm.include_cost_in_streaming_usage
     litellm.include_cost_in_streaming_usage = True
     openai_types = streaming_module._get_openai_response_types()
@@ -1765,9 +1765,10 @@ def test_process_chunk_completed_response_updates_id_and_usage_cost(monkeypatch)
     assert iterator.completed_response is event
     assert event.response.id != "resp_live"
     assert event.response.id.startswith("resp_")
-    assert event.response.usage.cost == 1.23
+    assert event.response.usage.cost is None, "the bill goes on at the proxy egress, never on the event logging prices"
     completion_handler.assert_called_once()
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
 def test_process_chunk_failed_response_triggers_failure_logging(monkeypatch):
     openai_types = streaming_module._get_openai_response_types()
@@ -1803,6 +1804,7 @@ def test_process_chunk_failed_response_triggers_failure_logging(monkeypatch):
     assert iterator.completed_response is event
     failure_handler.assert_called_once()
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
 @pytest.mark.asyncio
 async def test_handle_logging_failed_response_uses_response_error_message():
@@ -1835,6 +1837,7 @@ async def test_handle_logging_failed_response_uses_response_error_message():
     assert logging_obj.failure_calls == 1
     assert logging_obj.async_failure_calls == 1
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
 def test_process_chunk_returns_none_for_invalid_json_and_non_dict_payload():
     class _NoopConfig:
@@ -1853,6 +1856,7 @@ def test_process_chunk_returns_none_for_invalid_json_and_non_dict_payload():
     assert iterator._process_chunk("not-json") is None
     assert iterator._process_chunk(json.dumps(["not", "a", "dict"])) is None
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
 def test_process_chunk_cost_annotation_failure_is_nonfatal(monkeypatch):
     original_include_cost = litellm.include_cost_in_streaming_usage
@@ -1900,6 +1904,7 @@ def test_process_chunk_cost_annotation_failure_is_nonfatal(monkeypatch):
     assert event.response.usage.cost is None
     completion_handler.assert_called_once()
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
 def test_get_completed_response_object_accepts_direct_response():
     logging_obj = _FakeLoggingObj()
@@ -1916,6 +1921,7 @@ def test_get_completed_response_object_accepts_direct_response():
 
     assert iterator._get_completed_response_object() is direct_response
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
 @pytest.mark.asyncio
 async def test_responses_streaming_completed_event_persists_async_cache():
@@ -1964,6 +1970,7 @@ async def test_responses_streaming_completed_event_persists_async_cache():
     assert json.loads(litellm.cache.async_add_cache.call_args.args[0])["id"] == iterator.completed_response.response.id
     litellm.cache = original_cache
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
 def test_responses_streaming_completed_event_persists_sync_cache():
     logging_obj = _FakeLoggingObj()
@@ -2009,6 +2016,7 @@ def test_responses_streaming_completed_event_persists_sync_cache():
     assert json.loads(litellm.cache.add_cache.call_args.args[0])["id"] == iterator.completed_response.response.id
     litellm.cache = original_cache
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
 def test_log_completed_response_sync_direct_path(monkeypatch):
     hook_calls = {"post_call": 0, "metadata": 0}
@@ -2050,6 +2058,7 @@ def test_log_completed_response_sync_direct_path(monkeypatch):
     assert hook_calls["post_call"] == 1
     assert hook_calls["metadata"] == 1
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
 def test_log_completed_response_falls_back_when_model_validate_fails(monkeypatch):
     class _BadSerializableResponse:
@@ -2079,6 +2088,7 @@ def test_log_completed_response_falls_back_when_model_validate_fails(monkeypatch
     assert logging_obj.success_calls == 1
     assert logging_obj.async_success_calls == 1
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
 @pytest.mark.parametrize(
     "scenario",
@@ -2143,6 +2153,7 @@ def test_persist_completed_response_to_cache_guard_branches(monkeypatch, scenari
     expected_cached_flag = scenario == "already_cached"
     assert iterator._completed_response_cached is expected_cached_flag
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
 def test_build_synthetic_response_events_covers_annotations_function_calls_and_refusals():
     original_include_cost = litellm.include_cost_in_streaming_usage
@@ -2201,6 +2212,7 @@ def test_build_synthetic_response_events_covers_annotations_function_calls_and_r
     assert "response.function_call_arguments.done" in event_types
     assert event_types[-1] == "response.completed"
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
 @pytest.mark.asyncio
 async def test_mock_responses_streaming_iterator_async_iteration_logs_completion(
@@ -2251,6 +2263,7 @@ async def test_mock_responses_streaming_iterator_async_iteration_logs_completion
     assert hook_calls["post_call"] == 1
     assert hook_calls["metadata"] == 1
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
 def test_mock_responses_streaming_iterator_sync_iteration_logs_completion(monkeypatch):
     hook_calls = {"post_call": 0, "metadata": 0}
@@ -2296,6 +2309,7 @@ def test_mock_responses_streaming_iterator_sync_iteration_logs_completion(monkey
     assert hook_calls["post_call"] == 1
     assert hook_calls["metadata"] == 1
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
 @pytest.mark.asyncio
 async def test_cached_responses_stream_async_hit_triggers_success_callbacks(
@@ -2355,6 +2369,7 @@ async def test_cached_responses_stream_async_hit_triggers_success_callbacks(
     litellm.cache.add_cache.assert_not_called()
     litellm.cache = original_cache
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
 def test_cached_responses_stream_sync_hit_triggers_success_callbacks(monkeypatch):
     hook_calls = {"post_call": 0, "metadata": 0}
@@ -2453,7 +2468,9 @@ def test_faked_stream_bridged_to_chat_sends_tool_arguments_once():
     tool_call_deltas: Final = tuple(
         tool_call
         for chunk in chunks
-        for tool_call in (chunk.choices[0].delta.tool_calls or ())  # comprehension-ok: flattening each chunk's tool calls
+        for tool_call in (
+            chunk.choices[0].delta.tool_calls or ()
+        )  # comprehension-ok: flattening each chunk's tool calls
     )
     joined_arguments: Final = "".join(tool_call.function.arguments or "" for tool_call in tool_call_deltas)
     joined_text: Final = "".join(chunk.choices[0].delta.content or "" for chunk in chunks)
@@ -2509,7 +2526,6 @@ def test_faked_stream_opens_items_in_progress_and_empty_like_a_real_stream():
     assert [part.text for part in added_parts] == [""]
 
 
-
 def _answer_with_reasoning_refusal_and_custom_tool_call() -> ResponsesAPIResponse:
     return ResponsesAPIResponse(
         id="resp_synthetic",
@@ -2643,7 +2659,9 @@ async def test_shutdown_drain_waits_for_a_pending_logging_task():
 
 @pytest.mark.asyncio
 async def test_shutdown_drain_gives_up_on_a_stuck_logging_task_at_its_timeout():
-    stuck: Final = streaming_module.spawn_logging_task(asyncio.get_running_loop(), asyncio.Event().wait(), task_name="stuck")
+    stuck: Final = streaming_module.spawn_logging_task(
+        asyncio.get_running_loop(), asyncio.Event().wait(), task_name="stuck"
+    )
 
     await asyncio.wait_for(streaming_module.drain_logging_tasks(timeout=0.05), timeout=1)
 

tests/unit/test_cost_calculator.py

@@ -1,46 +1,75 @@
-import json
 import asyncio
 import datetime
+import json
+import os
+import threading
 import time
 from collections.abc import Iterator, Mapping
 from types import MappingProxyType, SimpleNamespace
-from typing import Final, Literal, cast
+from typing import Final, cast
+from unittest.mock import MagicMock, patch
 
 import httpx
 import pytest
 import respx
-from openai.types.completion_usage import CompletionUsage
 from pydantic import BaseModel
 
 import litellm
 from litellm import TranscriptionResponse, model_cost
 from litellm.constants import REALTIME_METERED_USAGE_KEY
-from litellm.cost_calculator import BaseTokenUsageProcessor, RealtimeAPITokenUsageProcessor, ResponsesWebSocketTokenUsageProcessor, batch_cost_calculator, completion_cost, cost_per_token, handle_realtime_stream_cost_calculation, response_cost_calculator
-from litellm.litellm_core_utils.llm_cost_calc.utils import parse_prompt_tokens_details
+from litellm.cost_calculator import (
+    BaseTokenUsageProcessor,
+    RealtimeAPITokenUsageProcessor,
+    ResponsesWebSocketTokenUsageProcessor,
+    batch_cost_calculator,
+    completion_cost,
+    cost_per_token,
+    handle_realtime_stream_cost_calculation,
+    response_cost_calculator,
+)
 from litellm.integrations.custom_logger import CustomLogger
 from litellm.litellm_core_utils.fallback_generalizations import (
     get_fallback_generalization_rules,
     set_fallback_generalizations,
 )
-from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
 from litellm.litellm_core_utils.litellm_logging import Logging
+from litellm.litellm_core_utils.llm_cost_calc.utils import parse_prompt_tokens_details
 from litellm.litellm_core_utils.llm_response_utils.convert_dict_to_response import (
     convert_to_model_response_object,
 )
+from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
 from litellm.llms.base_llm.ocr.transformation import OCRPage, OCRResponse, OCRUsageInfo
 from litellm.llms.custom_httpx.http_handler import HTTPHandler
 from litellm.llms.fireworks_ai.cost_calculator import get_base_model_for_pricing
-from litellm.llms.together_ai.cost_calculator import get_model_params_and_category
 from litellm.types.llms.base import CachedTokensDetails
-from litellm.types.llms.openai import HttpxBinaryResponseContent, OpenAIRealtimeStreamList, ResponseAPIUsage, ResponsesAPIResponse
+from litellm.types.llms.openai import (
+    HttpxBinaryResponseContent,
+    OpenAIRealtimeStreamList,
+    ResponseAPIUsage,
+    ResponsesAPIResponse,
+)
 from litellm.types.llms.xai import RealtimeMeteredUsage
 from litellm.types.rerank import RerankResponse
-from litellm.types.utils import CacheCreationTokenDetails, CallTypes, Choices, CompletionTokensDetailsWrapper, EmbeddingResponse, ImageObject, ImageResponse, ImageUsage, ImageUsageInputTokensDetails, LiteLLMRealtimeStreamLoggingObject, Message, ModelInfo, ModelResponse, PromptTokensDetails, PromptTokensDetailsWrapper, Usage
+from litellm.types.utils import (
+    CacheCreationTokenDetails,
+    CallTypes,
+    Choices,
+    CompletionTokensDetailsWrapper,
+    EmbeddingResponse,
+    ImageObject,
+    ImageResponse,
+    ImageUsage,
+    ImageUsageInputTokensDetails,
+    LiteLLMRealtimeStreamLoggingObject,
+    Message,
+    ModelInfo,
+    ModelResponse,
+    PromptTokensDetails,
+    PromptTokensDetailsWrapper,
+    Usage,
+)
 from litellm.types.videos.main import VideoObject
 from litellm.utils import supports_prompt_caching
-import os
-import threading
-from unittest.mock import MagicMock, patch
 
 
 class ResponseCostCapture(CustomLogger):
@@ -516,9 +545,7 @@ def test_completion_cost_prices_base_model_under_its_own_provider(
     )
 
     info: Final = litellm.get_model_info("claude-sonnet-4-5", custom_llm_provider="anthropic")
-    assert cost == pytest.approx(
-        1000 * info["input_cost_per_token"] + 1000 * info["output_cost_per_token"]
-    )
+    assert cost == pytest.approx(1000 * info["input_cost_per_token"] + 1000 * info["output_cost_per_token"])
 
 
 def test_completion_cost_strips_dated_azure_snapshot_model(_local_model_cost_map: None) -> None:
@@ -1613,9 +1640,7 @@ def test_per_query_priced_rerank_deployment_completion_cost_is_nonzero():
         "vertex_ai/semantic-ranker-fast-004",
     ),
 )
-def test_vertex_semantic_ranker_rerank_completion_cost_is_nonzero(
-    model: str, _local_model_cost_map: None
-) -> None:
+def test_vertex_semantic_ranker_rerank_completion_cost_is_nonzero(model: str, _local_model_cost_map: None) -> None:
     response: Final = RerankResponse(
         id="vertex_ai_rerank_pricing_test",
         results=[],
@@ -3371,9 +3396,7 @@ def test_gemini_cache_tokens_details_no_negative_values():
     assert usage.prompt_tokens_details.text_tokens == 9402
     assert usage.prompt_tokens_details.image_tokens == 258
     assert usage.prompt_tokens_details.cached_tokens == 9651
-    assert usage.prompt_tokens_details.cached_tokens_details == CachedTokensDetails(
-        text_tokens=9393, image_tokens=258
-    )
+    assert usage.prompt_tokens_details.cached_tokens_details == CachedTokensDetails(text_tokens=9393, image_tokens=258)
     assert prompt_details["text_tokens"] == 9
     assert prompt_details["image_tokens"] == 0
     assert prompt_details["text_tokens"] >= 0
@@ -6941,23 +6964,17 @@ def test_cost_ft_gpt_35():
             usage=Usage(prompt_tokens=21, completion_tokens=17, total_tokens=38),
         )
 
-        cost = litellm.completion_cost(
-            completion_response=resp, custom_llm_provider="openai"
-        )
+        cost = litellm.completion_cost(completion_response=resp, custom_llm_provider="openai")
         print("\n Calculated Cost for ft:gpt-3.5", cost)
         input_cost = model_cost["ft:gpt-3.5-turbo"]["input_cost_per_token"]
         output_cost = model_cost["ft:gpt-3.5-turbo"]["output_cost_per_token"]
         print(input_cost, output_cost)
-        expected_cost = (input_cost * resp.usage.prompt_tokens) + (
-            output_cost * resp.usage.completion_tokens
-        )
+        expected_cost = (input_cost * resp.usage.prompt_tokens) + (output_cost * resp.usage.completion_tokens)
         print("\n Excpected cost", expected_cost)
         assert cost == expected_cost
     except Exception as e:
         print(f"Error: {e}")
-        pytest.fail(
-            f"Cost Calc failed for ft:gpt-3.5. Expected {expected_cost}, Calculated cost {cost}"
-        )
+        pytest.fail(f"Cost Calc failed for ft:gpt-3.5. Expected {expected_cost}, Calculated cost {cost}")
 
 
 def test_cost_azure_gpt_35():
@@ -6983,28 +7000,22 @@ def test_cost_azure_gpt_35():
             usage=Usage(prompt_tokens=21, completion_tokens=17, total_tokens=38),
         )
 
-        cost = litellm.completion_cost(
-            completion_response=resp, model="azure/chatgpt-deployment-2"
-        )
+        cost = litellm.completion_cost(completion_response=resp, model="azure/chatgpt-deployment-2")
         print("\n Calculated Cost for azure/gpt-3.5-turbo", cost)
         input_cost = model_cost["azure/gpt-35-turbo"]["input_cost_per_token"]
         output_cost = model_cost["azure/gpt-35-turbo"]["output_cost_per_token"]
-        expected_cost = (input_cost * resp.usage.prompt_tokens) + (
-            output_cost * resp.usage.completion_tokens
-        )
+        expected_cost = (input_cost * resp.usage.prompt_tokens) + (output_cost * resp.usage.completion_tokens)
         print("\n Excpected cost", expected_cost)
         assert cost == expected_cost
     except Exception as e:
-        pytest.fail(f"Cost Calc failed for azure/gpt-3.5-turbo. {str(e)}")
+        pytest.fail(f"Cost Calc failed for azure/gpt-3.5-turbo. {e!s}")
 
 
 def test_cost_bedrock_pricing_actual_calls():
     litellm.set_verbose = True
     model = "anthropic.claude-3-5-sonnet-20240620-v1:0"
     messages = [{"role": "user", "content": "Hey, how's it going?"}]
-    response = litellm.completion(
-        model=model, messages=messages, mock_response="hello cool one"
-    )
+    response = litellm.completion(model=model, messages=messages, mock_response="hello cool one")
 
     print("response", response)
     cost = litellm.completion_cost(
@@ -7035,8 +7046,7 @@ def test_whisper_openai():
     print(f"cost: {cost}")
     print(f"whisper dict: {litellm.model_cost['whisper-1']}")
     expected_cost = round(
-        litellm.model_cost["whisper-1"]["output_cost_per_second"]
-        * _total_time_in_seconds,
+        litellm.model_cost["whisper-1"]["output_cost_per_second"] * _total_time_in_seconds,
         5,
     )
     assert round(cost, 5) == round(expected_cost, 5)
@@ -7056,15 +7066,12 @@ def test_whisper_azure():
     _total_time_in_seconds = 3
     setattr(transcription, "duration", _total_time_in_seconds)
 
-    cost = litellm.completion_cost(
-        model="azure/azure-whisper", completion_response=transcription
-    )
+    cost = litellm.completion_cost(model="azure/azure-whisper", completion_response=transcription)
 
     print(f"cost: {cost}")
     print(f"whisper dict: {litellm.model_cost['whisper-1']}")
     expected_cost = round(
-        litellm.model_cost["whisper-1"]["output_cost_per_second"]
-        * _total_time_in_seconds,
+        litellm.model_cost["whisper-1"]["output_cost_per_second"] * _total_time_in_seconds,
         5,
     )
     assert round(cost, 5) == round(expected_cost, 5)
@@ -7111,9 +7118,7 @@ def test_transcription_cost_falls_back_to_provider_billed_seconds(
         response_type="audio_transcription",
     )
 
-    cost: Final = completion_cost(
-        completion_response=transcription, model=requested_model, call_type="atranscription"
-    )
+    cost: Final = completion_cost(completion_response=transcription, model=requested_model, call_type="atranscription")
 
     assert cost == pytest.approx(0.01 * expected_seconds)
 
@@ -7205,9 +7210,7 @@ def test_replicate_llama3_cost_tracking():
         model="replicate/meta/meta-llama-3-8b-instruct",
         object="chat.completion",
         system_fingerprint=None,
-        usage=litellm.utils.Usage(
-            prompt_tokens=48, completion_tokens=31, total_tokens=79
-        ),
+        usage=litellm.utils.Usage(prompt_tokens=48, completion_tokens=31, total_tokens=79),
     )
     cost = litellm.completion_cost(
         completion_response=response,
@@ -7217,20 +7220,14 @@ def test_replicate_llama3_cost_tracking():
     print(f"cost: {cost}")
     cost = round(cost, 5)
     expected_cost = round(
-        litellm.model_cost["replicate/meta/meta-llama-3-8b-instruct"][
-            "input_cost_per_token"
-        ]
-        * 48
-        + litellm.model_cost["replicate/meta/meta-llama-3-8b-instruct"][
-            "output_cost_per_token"
-        ]
-        * 31,
+        litellm.model_cost["replicate/meta/meta-llama-3-8b-instruct"]["input_cost_per_token"] * 48
+        + litellm.model_cost["replicate/meta/meta-llama-3-8b-instruct"]["output_cost_per_token"] * 31,
         5,
     )
     assert cost == expected_cost
 
 
-@pytest.mark.parametrize("is_streaming", [True, False])  #
+@pytest.mark.parametrize("is_streaming", [True, False])
 def test_groq_response_cost_tracking(is_streaming):
     from litellm.utils import (
         CallTypes,
@@ -7360,7 +7357,7 @@ def test_vertex_ai_completion_cost():
 
     model_info = litellm.get_model_info(model="gemini-3.8-flash")
 
-    print("\nExpected model info:\n{}\n\n".format(model_info))
+    print(f"\nExpected model info:\n{model_info}\n\n")
 
     expected_input_cost = prompt_tokens * model_info["input_cost_per_token"]
 
@@ -7373,8 +7370,8 @@ def test_vertex_ai_completion_cost():
     )
 
     assert round(expected_input_cost, 6) == round(calculated_input_cost, 6)
-    print("expected_input_cost: {}".format(expected_input_cost))
-    print("calculated_input_cost: {}".format(calculated_input_cost))
+    print(f"expected_input_cost: {expected_input_cost}")
+    print(f"calculated_input_cost: {calculated_input_cost}")
 
 
 def test_vertex_ai_medlm_completion_cost():
@@ -7385,9 +7382,7 @@ def test_vertex_ai_medlm_completion_cost():
 
     model = "vertex_ai/medlm-medium"
     messages = [{"role": "user", "content": "Test MedLM completion cost."}]
-    predictive_cost = completion_cost(
-        model=model, messages=messages, custom_llm_provider="vertex_ai"
-    )
+    predictive_cost = completion_cost(model=model, messages=messages, custom_llm_provider="vertex_ai")
     assert predictive_cost > 0
 
     model = "vertex_ai/medlm-large"
@@ -7404,13 +7399,11 @@ def test_vertex_ai_embedding_completion_cost(caplog):
     litellm.model_cost = litellm.get_model_cost_map(url="")
 
     text = "The quick brown fox jumps over the lazy dog."
-    input_tokens = litellm.token_counter(
-        model="vertex_ai/text-embedding-004", text=text
-    )
+    input_tokens = litellm.token_counter(model="vertex_ai/text-embedding-004", text=text)
 
     model_info = litellm.get_model_info(model="vertex_ai/text-embedding-004")
 
-    print("\nExpected model info:\n{}\n\n".format(model_info))
+    print(f"\nExpected model info:\n{model_info}\n\n")
 
     expected_input_cost = input_tokens * model_info["input_cost_per_token"]
 
@@ -7423,16 +7416,13 @@ def test_vertex_ai_embedding_completion_cost(caplog):
     )
 
     assert round(expected_input_cost, 6) == round(calculated_input_cost, 6)
-    print("expected_input_cost: {}".format(expected_input_cost))
-    print("calculated_input_cost: {}".format(calculated_input_cost))
+    print(f"expected_input_cost: {expected_input_cost}")
+    print(f"calculated_input_cost: {calculated_input_cost}")
 
     captured_logs = [rec.message for rec in caplog.records]
     for item in captured_logs:
-        print("\nitem:{}\n".format(item))
-        if (
-            "litellm.litellm_core_utils.llm_cost_calc.google.cost_per_character(): Exception occured "
-            in item
-        ):
+        print(f"\nitem:{item}\n")
+        if "litellm.litellm_core_utils.llm_cost_calc.google.cost_per_character(): Exception occured " in item:
             raise Exception("Error log raised for calculating embedding cost")
 
 
@@ -7474,9 +7464,7 @@ def test_vertex_ai_mistral_predict_cost(usage):
     else:
         from openai.types.completion_usage import CompletionUsage
 
-        response_usage = CompletionUsage(
-            prompt_tokens=32, completion_tokens=55, total_tokens=87
-        )
+        response_usage = CompletionUsage(prompt_tokens=32, completion_tokens=55, total_tokens=87)
     response_object = ModelResponse(
         id="26c0ef045020429d9c5c9b078c01e564",
         choices=[
@@ -7510,9 +7498,7 @@ def test_vertex_ai_mistral_predict_cost(usage):
     assert predictive_cost > 0
 
 
-@pytest.mark.parametrize(
-    "model", ["openai/tts-1", "azure/tts-1", "openai/gpt-4o-mini-tts"]
-)
+@pytest.mark.parametrize("model", ["openai/tts-1", "azure/tts-1", "openai/gpt-4o-mini-tts"])
 def test_completion_cost_tts(model):
     os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True"
     litellm.model_cost = litellm.get_model_cost_map(url="")
@@ -7611,9 +7597,7 @@ def test_completion_cost_prompt_caching(model, custom_llm_provider):
 
     cost_1 = completion_cost(model=model, completion_response=response_1)
 
-    _model_info = litellm.get_model_info(
-        model=model, custom_llm_provider=custom_llm_provider
-    )
+    _model_info = litellm.get_model_info(model=model, custom_llm_provider=custom_llm_provider)
     expected_cost = (
         (
             response_1.usage.prompt_tokens
@@ -7621,12 +7605,9 @@ def test_completion_cost_prompt_caching(model, custom_llm_provider):
             - response_1.usage.prompt_tokens_details.cache_creation_tokens
         )
         * _model_info["input_cost_per_token"]
-        + (response_1.usage.prompt_tokens_details.cached_tokens or 0)
-        * _model_info["cache_read_input_token_cost"]
-        + (response_1.usage.cache_creation_input_tokens or 0)
-        * _model_info["cache_creation_input_token_cost"]
-        + (response_1.usage.completion_tokens or 0)
-        * _model_info["output_cost_per_token"]
+        + (response_1.usage.prompt_tokens_details.cached_tokens or 0) * _model_info["cache_read_input_token_cost"]
+        + (response_1.usage.cache_creation_input_tokens or 0) * _model_info["cache_creation_input_token_cost"]
+        + (response_1.usage.completion_tokens or 0) * _model_info["output_cost_per_token"]
     )  # Cost of processing (non-cache hit + cache hit) + Cost of cache-writing (cache writing)
 
     assert round(expected_cost, 5) == round(cost_1, 5)
@@ -7718,9 +7699,7 @@ def test_completion_cost_databricks_embedding(model, monkeypatch):
     sync_handler = HTTPHandler()
 
     with patch.object(HTTPHandler, "post", return_value=mock_response):
-        resp = litellm.embedding(
-            model=model, input=["hey, how's it going?"], client=sync_handler
-        )
+        resp = litellm.embedding(model=model, input=["hey, how's it going?"], client=sync_handler)
 
         print(resp)
         cost = completion_cost(completion_response=resp)
@@ -7810,11 +7789,9 @@ def test_cost_openai_prompt_caching():
     usage = response_2.usage
 
     _expected_cost2 = (
-        (usage.prompt_tokens - usage.prompt_tokens_details.cached_tokens)
-        * model_info["input_cost_per_token"]
+        (usage.prompt_tokens - usage.prompt_tokens_details.cached_tokens) * model_info["input_cost_per_token"]
         + usage.completion_tokens * model_info["output_cost_per_token"]
-        + usage.prompt_tokens_details.cached_tokens
-        * model_info["cache_read_input_token_cost"]
+        + usage.prompt_tokens_details.cached_tokens * model_info["cache_read_input_token_cost"]
     )
 
     print("_expected_cost2", _expected_cost2)
@@ -7855,9 +7832,7 @@ def test_completion_cost_azure_ai_rerank(model):
         },
     )
     print("response", response)
-    cost = completion_cost(
-        model=model, completion_response=response, call_type="arerank"
-    )
+    cost = completion_cost(model=model, completion_response=response, call_type="arerank")
     assert cost > 0
 
 
@@ -8809,9 +8784,7 @@ async def test_test_completion_cost_gpt4o_audio_output_from_model(stream):
         completion_tokens=34,
         prompt_tokens=16,
         total_tokens=50,
-        completion_tokens_details=CompletionTokensDetailsWrapper(
-            audio_tokens=28, reasoning_tokens=0, text_tokens=6
-        ),
+        completion_tokens_details=CompletionTokensDetailsWrapper(audio_tokens=28, reasoning_tokens=0, text_tokens=6),
         prompt_tokens_details=PromptTokensDetailsWrapper(
             audio_tokens=0, cached_tokens=0, text_tokens=16, image_tokens=0
         ),
@@ -8850,27 +8823,15 @@ async def test_test_completion_cost_gpt4o_audio_output_from_model(stream):
     print(f"model_info: {model_info}")
     ## input cost
 
-    input_audio_cost = (
-        model_info["input_cost_per_audio_token"]
-        * usage_object.prompt_tokens_details.audio_tokens
-    )
-    input_text_cost = (
-        model_info["input_cost_per_token"]
-        * usage_object.prompt_tokens_details.text_tokens
-    )
+    input_audio_cost = model_info["input_cost_per_audio_token"] * usage_object.prompt_tokens_details.audio_tokens
+    input_text_cost = model_info["input_cost_per_token"] * usage_object.prompt_tokens_details.text_tokens
 
     total_input_cost = input_audio_cost + input_text_cost
 
     ## output cost
 
-    output_audio_cost = (
-        model_info["output_cost_per_audio_token"]
-        * usage_object.completion_tokens_details.audio_tokens
-    )
-    output_text_cost = (
-        model_info["output_cost_per_token"]
-        * usage_object.completion_tokens_details.text_tokens
-    )
+    output_audio_cost = model_info["output_cost_per_audio_token"] * usage_object.completion_tokens_details.audio_tokens
+    output_text_cost = model_info["output_cost_per_token"] * usage_object.completion_tokens_details.text_tokens
 
     total_output_cost = output_audio_cost + output_text_cost
 
@@ -8933,7 +8894,7 @@ def test_completion_cost_model_response_cost(response_model, custom_llm_provider
 
 def test_select_model_name_for_cost_calc():
     from litellm.cost_calculator import select_model_name_for_cost_calc
-    from litellm.types.utils import ModelResponse, Choices, Usage, Message
+    from litellm.types.utils import Choices, Message, ModelResponse, Usage
 
     args = {
         "model": "Mistral-large-nmefg",
@@ -8995,9 +8956,7 @@ def test_cost_calculator_with_base_model_with_router(base_model_arg):
     }
 
     if base_model_arg == "litellm_param":
-        model_item["litellm_params"][
-            "base_model"
-        ] = "bedrock/anthropic.claude-sonnet-5"
+        model_item["litellm_params"]["base_model"] = "bedrock/anthropic.claude-sonnet-5"
     elif base_model_arg == "model_info":
         model_item["model_info"] = {
             "base_model": "bedrock/anthropic.claude-sonnet-5",
@@ -9063,9 +9022,7 @@ def isolated_pricing_registry(local_model_cost_map: None, monkeypatch: pytest.Mo
 
 
 _CUSTOM_PRICING_MODEL: Final = "migration-custom-pricing-chat"
-_CUSTOM_PRICING_RATES: Final = MappingProxyType(
-    {"input_cost_per_token": 0.0000008, "output_cost_per_token": 0.0000032}
-)
+_CUSTOM_PRICING_RATES: Final = MappingProxyType({"input_cost_per_token": 0.0000008, "output_cost_per_token": 0.0000032})
 _CUSTOM_PRICING_EXPECTED_COST: Final = 7 * 0.0000008 + 4 * 0.0000032
 
 
@@ -9086,7 +9043,13 @@ def _custom_pricing_upstream(request: httpx.Request) -> httpx.Response:
         body: Final = "".join(
             (
                 _custom_pricing_chunk(
-                    [{"index": 0, "delta": {"role": "assistant", "content": "I am ready to help."}, "finish_reason": None}]
+                    [
+                        {
+                            "index": 0,
+                            "delta": {"role": "assistant", "content": "I am ready to help."},
+                            "finish_reason": None,
+                        }
+                    ]
                 ),
                 _custom_pricing_chunk([{"index": 0, "delta": {}, "finish_reason": "stop"}]),
                 _custom_pricing_chunk([], {"prompt_tokens": 7, "completion_tokens": 4, "total_tokens": 11}),
@@ -9102,7 +9065,11 @@ def _custom_pricing_upstream(request: httpx.Request) -> httpx.Response:
             "created": 1,
             "model": _CUSTOM_PRICING_MODEL,
             "choices": [
-                {"index": 0, "message": {"role": "assistant", "content": "I am ready to help."}, "finish_reason": "stop"}
+                {
+                    "index": 0,
+                    "message": {"role": "assistant", "content": "I am ready to help."},
+                    "finish_reason": "stop",
+                }
             ],
             "usage": {"prompt_tokens": 7, "completion_tokens": 4, "total_tokens": 11},
         },
@@ -9216,7 +9183,12 @@ def test_fireworks_completion_cost_uses_registered_rate(monkeypatch: pytest.Monk
     monkeypatch.setitem(
         litellm.model_cost,
         model,
-        {"input_cost_per_token": 0.001, "output_cost_per_token": 0.002, "litellm_provider": "fireworks_ai", "mode": "chat"},
+        {
+            "input_cost_per_token": 0.001,
+            "output_cost_per_token": 0.002,
+            "litellm_provider": "fireworks_ai",
+            "mode": "chat",
+        },
     )
     respx.post("https://api.fireworks.ai/inference/v1/chat/completions").mock(
         return_value=httpx.Response(
@@ -9350,16 +9322,13 @@ def test_speech_cost_with_custom_pricing_bills_the_reported_usage() -> None:
     )
 
     assert cost == pytest.approx(
-        usage.prompt_tokens * rates["input_cost_per_token"]
-        + usage.completion_tokens * rates["output_cost_per_token"]
+        usage.prompt_tokens * rates["input_cost_per_token"] + usage.completion_tokens * rates["output_cost_per_token"]
     )
 
 
 def test_moderation_cost_uses_test_owned_token_rate() -> None:
     prompt: Final = "moderation input"
-    rates: Final = MappingProxyType(
-        {"input_cost_per_token": 0.001, "output_cost_per_token": 0.0}
-    )
+    rates: Final = MappingProxyType({"input_cost_per_token": 0.001, "output_cost_per_token": 0.0})
 
     cost: Final = completion_cost(
         completion_response=None,
@@ -9377,9 +9346,7 @@ def test_moderation_cost_uses_test_owned_token_rate() -> None:
 def test_vertex_predictive_cost_uses_test_owned_token_rate() -> None:
     model: Final = "vertex_ai/migration-llama"
     messages: Final = [{"role": "user", "content": "price control"}]
-    rates: Final = MappingProxyType(
-        {"input_cost_per_token": 0.001, "output_cost_per_token": 0.002}
-    )
+    rates: Final = MappingProxyType({"input_cost_per_token": 0.001, "output_cost_per_token": 0.002})
 
     cost: Final = completion_cost(
         model=model,
@@ -9387,9 +9354,7 @@ def test_vertex_predictive_cost_uses_test_owned_token_rate() -> None:
         custom_llm_provider="vertex_ai",
         custom_cost_per_token=rates,
     )
-    expected_cost: Final = (
-        litellm.token_counter(model=model, messages=messages) * 0.001
-    )
+    expected_cost: Final = litellm.token_counter(model=model, messages=messages) * 0.001
 
     assert cost == pytest.approx(expected_cost)
 
@@ -9398,9 +9363,7 @@ def test_vertex_predictive_cost_uses_test_owned_token_rate() -> None:
 @pytest.mark.asyncio
 @pytest.mark.usefixtures("isolated_pricing_registry")
 @respx.mock
-async def test_custom_pricing_cost_reaches_success_callback(
-    async_mode: bool, monkeypatch: pytest.MonkeyPatch
-) -> None:
+async def test_custom_pricing_cost_reaches_success_callback(async_mode: bool, monkeypatch: pytest.MonkeyPatch) -> None:
     model: Final = "migration-callback-pricing-model"
     litellm.register_model(
         {
@@ -9442,9 +9405,7 @@ async def test_custom_pricing_cost_reaches_success_callback(
         "input_cost_per_token": 0.001,
         "output_cost_per_token": 0.002,
     }
-    response: Final = (
-        await litellm.acompletion(**request) if async_mode else litellm.completion(**request)
-    )
+    response: Final = await litellm.acompletion(**request) if async_mode else litellm.completion(**request)
     await GLOBAL_LOGGING_WORKER.flush()
 
     assert await asyncio.to_thread(callback.success_event.wait, 10.0)
@@ -9540,7 +9501,7 @@ def test_json_valid_model_cost_map():
         json_str = json.dumps(model_cost)
         json.loads(json_str)
     except json.JSONDecodeError as e:
-        pytest.fail(f"Invalid JSON format: {str(e)}")
+        pytest.fail(f"Invalid JSON format: {e!s}")
 
 
 def test_batch_cost_calculator():
@@ -9845,9 +9806,7 @@ def test_xai_realtime_bills_text_messages(_xai_voice_test_model: None) -> None:
     assert cost == pytest.approx(30 * 0.0013333333333333333 + 0.004)
 
 
-def test_xai_realtime_missing_rates_bills_nothing(
-    _local_model_cost_map: None, monkeypatch: pytest.MonkeyPatch
-) -> None:
+def test_xai_realtime_missing_rates_bills_nothing(_local_model_cost_map: None, monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setitem(
         litellm.model_cost,
         "grok-voice-unpriced",
@@ -9942,9 +9901,8 @@ def test_live_session_deduplicates_delegated_responses_and_charges_voice(_local_
     )
     live_rate: Final = cast(float, litellm.model_cost["gpt-live-1"]["input_cost_per_second"])
     backend_model: Final = litellm.model_cost["gpt-4o-mini"]
-    expected_backend_cost: Final = (
-        10 * cast(float, backend_model["input_cost_per_token"])
-        + 5 * cast(float, backend_model["output_cost_per_token"])
+    expected_backend_cost: Final = 10 * cast(float, backend_model["input_cost_per_token"]) + 5 * cast(
+        float, backend_model["output_cost_per_token"]
     )
 
     cost: Final = handle_realtime_stream_cost_calculation(
@@ -10589,3 +10547,213 @@ def test_bedrock_registered_cache_prices_preserve_dynamic_threshold_rates(
     assert entry is not None
     assert entry[1]["input_cost_per_token_above_300k_tokens"] == 0.03
     assert entry[1][tier_field] == (tier_override if tier_override is not None else 0.007)
+
+
+_REPORTED_USAGE: Final = {"prompt_tokens": 100, "completion_tokens": 20, "total_tokens": 120, "cost": 0.037}
+_REPORTED_COST_FIXTURE_MODEL: Final = "openai/reported-cost-fixture"
+_REPORTED_COST_FIXTURE_RATES: Final = {
+    "litellm_provider": "openai",
+    "mode": "chat",
+    "input_cost_per_token": 1e-6,
+    "output_cost_per_token": 3e-6,
+}
+
+
+def _reported_cost_logging(stream: bool, litellm_params: Mapping[str, object]) -> Logging:
+    logging_obj: Final = Logging(
+        model=_REPORTED_COST_FIXTURE_MODEL,
+        messages=[{"role": "user", "content": "hi"}],
+        stream=stream,
+        call_type="completion",
+        start_time=datetime.datetime(2026, 1, 1),
+        litellm_call_id="reported-cost",
+        function_id="reported-cost",
+    )
+    logging_obj.update_from_kwargs(
+        kwargs={}, optional_params={}, litellm_params={"custom_llm_provider": "openai", **litellm_params}
+    )
+    logging_obj.model_call_details["custom_llm_provider"] = "openai"
+    return logging_obj
+
+
+def _reported_cost_stream_chunk(content: str | None, finish_reason: str | None, usage: Usage | None) -> object:
+    return litellm.ModelResponseStream(
+        id="chatcmpl-reported",
+        model="reported-cost-fixture",
+        created=1,
+        object="chat.completion.chunk",
+        choices=[
+            litellm.types.utils.StreamingChoices(
+                index=0, delta=litellm.types.utils.Delta(content=content), finish_reason=finish_reason
+            )
+        ],
+        usage=usage,
+    )
+
+
+def _bill_reported_cost(
+    stream: bool, litellm_params: Mapping[str, object], hidden_params: Mapping[str, object]
+) -> float | None:
+    """The spend the proxy logs for one upstream reply carrying usage.cost, streamed or not."""
+    logging_obj: Final = _reported_cost_logging(stream, litellm_params)
+    if stream:
+        response: Final[object] = litellm.stream_chunk_builder(
+            [
+                _reported_cost_stream_chunk("hi", None, None),
+                _reported_cost_stream_chunk(None, "stop", None),
+                _reported_cost_stream_chunk(None, None, Usage(**_REPORTED_USAGE)),
+            ],
+            messages=[{"role": "user", "content": "hi"}],
+            logging_obj=logging_obj,
+        )
+    else:
+        response = ModelResponse(
+            id="chatcmpl-reported",
+            model="reported-cost-fixture",
+            choices=[Choices(index=0, message=Message(role="assistant", content="hi"), finish_reason="stop")],
+            usage=Usage(**_REPORTED_USAGE),
+        )
+    assert isinstance(response, ModelResponse)
+    response._hidden_params.update({"custom_llm_provider": "openai", **hidden_params})
+    return logging_obj.response_cost_calculator(result=response)
+
+
+@pytest.mark.parametrize("stream", [False, True])
+@pytest.mark.parametrize("mapped", [False, True])
+def test_provider_reported_usage_cost_is_billed_on_stream_and_non_stream(
+    monkeypatch: pytest.MonkeyPatch, stream: bool, mapped: bool
+) -> None:
+    """Regression for #47533: a plain openai-compatible deployment billed usage.cost on streams and $0 on non-stream."""
+    if mapped:
+        monkeypatch.setitem(litellm.model_cost, _REPORTED_COST_FIXTURE_MODEL, _REPORTED_COST_FIXTURE_RATES)
+    else:
+        monkeypatch.delitem(litellm.model_cost, _REPORTED_COST_FIXTURE_MODEL, raising=False)
+
+    assert _bill_reported_cost(stream, {}, {}) == pytest.approx(0.037)
+
+
+@pytest.mark.parametrize("stream", [False, True])
+def test_deployment_pricing_beats_the_provider_reported_usage_cost(
+    monkeypatch: pytest.MonkeyPatch, stream: bool
+) -> None:
+    """Streams used to bill usage.cost even when the operator priced the deployment; both paths bill the operator's rates."""
+    deployment_id: Final = "reported-cost-deployment-priced-by-the-operator"
+    monkeypatch.setitem(
+        litellm.model_cost,
+        deployment_id,
+        {"litellm_provider": "openai", "mode": "chat", "input_cost_per_token": 2e-6, "output_cost_per_token": 6e-6},
+    )
+    litellm_params: Final = {
+        "input_cost_per_token": 2e-6,
+        "output_cost_per_token": 6e-6,
+        "model_info": {"id": deployment_id},
+    }
+
+    assert _bill_reported_cost(stream, litellm_params, {"model_id": deployment_id}) == pytest.approx(
+        100 * 2e-6 + 20 * 6e-6
+    )
+
+
+@pytest.mark.parametrize("stream", [False, True])
+def test_base_model_pricing_beats_the_provider_reported_usage_cost(
+    monkeypatch: pytest.MonkeyPatch, stream: bool
+) -> None:
+    base_model: Final = "openai/reported-cost-base-model-fixture"
+    monkeypatch.setitem(
+        litellm.model_cost,
+        base_model,
+        {"litellm_provider": "openai", "mode": "chat", "input_cost_per_token": 5e-6, "output_cost_per_token": 7e-6},
+    )
+
+    assert _bill_reported_cost(stream, {"base_model": base_model}, {}) == pytest.approx(100 * 5e-6 + 20 * 7e-6)
+
+
+@pytest.mark.parametrize("stream", [False, True])
+def test_provider_reported_usage_cost_takes_discount_then_margin(monkeypatch: pytest.MonkeyPatch, stream: bool) -> None:
+    monkeypatch.setattr(litellm, "cost_discount_config", {"openai": 0.2})
+    monkeypatch.setattr(litellm, "cost_margin_config", {"global": {"percentage": 0.5, "fixed_amount": 0.001}})
+
+    assert _bill_reported_cost(stream, {}, {}) == pytest.approx(0.037 * 0.8 * 1.5 + 0.001)
+
+
+def test_the_stream_builder_bills_the_operator_rates_and_leaves_usage_cost_as_the_upstream_reported_it(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The bill rides in hidden response_cost. usage.cost stays the upstream's figure, so pricing the assembled
+    stream again (an MCP round, a logging copy) reads a provider report, never litellm's own bill."""
+    deployment_id: Final = "reported-cost-deployment-priced-by-the-operator"
+    monkeypatch.setitem(
+        litellm.model_cost,
+        deployment_id,
+        {"litellm_provider": "openai", "mode": "chat", "input_cost_per_token": 2e-6, "output_cost_per_token": 6e-6},
+    )
+    logging_obj: Final = _reported_cost_logging(
+        True, {"input_cost_per_token": 2e-6, "output_cost_per_token": 6e-6, "model_info": {"id": deployment_id}}
+    )
+    operator_bill: Final = 100 * 2e-6 + 20 * 6e-6
+
+    response: Final[object] = litellm.stream_chunk_builder(
+        [
+            _reported_cost_stream_chunk("hi", None, None),
+            _reported_cost_stream_chunk(None, "stop", None),
+            _reported_cost_stream_chunk(None, None, Usage(**_REPORTED_USAGE)),
+        ],
+        messages=[{"role": "user", "content": "hi"}],
+        logging_obj=logging_obj,
+    )
+
+    assert isinstance(response, ModelResponse)
+    response._hidden_params.update({"custom_llm_provider": "openai", "model_id": deployment_id})
+    assert logging_obj.response_cost_calculator(result=response, ignore_stamped_cost=True) == pytest.approx(
+        operator_bill
+    )
+    assert response.usage is not None and response.usage.cost == _REPORTED_USAGE["cost"]
+
+
+def test_provider_reported_usage_cost_splits_input_and_output_by_the_token_math_ratio(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setitem(litellm.model_cost, _REPORTED_COST_FIXTURE_MODEL, _REPORTED_COST_FIXTURE_RATES)
+    usage: Final = Usage(**_REPORTED_USAGE)
+
+    prompt_cost, completion_cost = cost_per_token(
+        model=_REPORTED_COST_FIXTURE_MODEL, custom_llm_provider="openai", usage_object=usage
+    )
+
+    priced_prompt: Final = 100 * 1e-6
+    priced_completion: Final = 20 * 3e-6
+    assert prompt_cost + completion_cost == pytest.approx(0.037)
+    assert prompt_cost == pytest.approx(0.037 * priced_prompt / (priced_prompt + priced_completion))
+
+
+def test_provider_reported_usage_cost_without_rates_is_all_output(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delitem(litellm.model_cost, _REPORTED_COST_FIXTURE_MODEL, raising=False)
+
+    assert cost_per_token(
+        model=_REPORTED_COST_FIXTURE_MODEL, custom_llm_provider="openai", usage_object=Usage(**_REPORTED_USAGE)
+    ) == (0.0, 0.037)
+
+
+@pytest.mark.parametrize("reported", [-0.037, float("nan"), float("inf"), True])
+def test_unusable_provider_reported_usage_cost_falls_back_to_the_token_math(
+    monkeypatch: pytest.MonkeyPatch, reported: object
+) -> None:
+    monkeypatch.setitem(litellm.model_cost, _REPORTED_COST_FIXTURE_MODEL, _REPORTED_COST_FIXTURE_RATES)
+    usage: Final = Usage(prompt_tokens=100, completion_tokens=20, total_tokens=120)
+    setattr(usage, "cost", reported)
+
+    assert cost_per_token(model=_REPORTED_COST_FIXTURE_MODEL, custom_llm_provider="openai", usage_object=usage) == (
+        pytest.approx(100 * 1e-6),
+        pytest.approx(20 * 3e-6),
+    )
+
+
+def test_caller_supplied_custom_cost_per_token_beats_the_provider_reported_usage_cost() -> None:
+    assert cost_per_token(
+        model=_REPORTED_COST_FIXTURE_MODEL,
+        custom_llm_provider="openai",
+        prompt_tokens=100,
+        completion_tokens=20,
+        usage_object=Usage(**_REPORTED_USAGE),
+        custom_cost_per_token={"input_cost_per_token": 0.001, "output_cost_per_token": 0.002},
+    ) == (pytest.approx(0.1), pytest.approx(0.04))

tests/unit/test_main.py

@@ -20,7 +20,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import httpx
 import pytest
 import respx
-from openai import APITimeoutError, AsyncOpenAI, OpenAI
+from openai import AsyncOpenAI, OpenAI
 from openai._legacy_response import LegacyAPIResponse
 from openai.types.chat.chat_completion import ChatCompletion
 from pydantic import TypeAdapter
@@ -291,9 +291,7 @@ def test_completion_missing_role(openai_api_response):
 
     print(f"openai_api_response: {openai_api_response}")
 
-    with patch.object(
-        client, "post", MagicMock(return_value=mock_raw_response)
-    ) as mock_create:
+    with patch.object(client, "post", MagicMock(return_value=mock_raw_response)) as mock_create:
         litellm.completion(
             model="gpt-4o-mini",
             messages=[
@@ -447,7 +445,7 @@ async def test_url_with_format_param(model, sync_mode, monkeypatch):
             else:
                 response = await acompletion(**args, client=client)
             print(response)
-        except Exception as e:
+        except Exception:
             pass
 
         mock_client.assert_called()
@@ -2860,9 +2858,7 @@ def test_parallel_function_call_stream_reassembles_each_arguments_payload():
             created=1751934860,
             model="gpt-4.1-mini",
             object="chat.completion.chunk",
-            choices=[
-                StreamingChoices(index=0, delta=Delta(tool_calls=tool_calls), finish_reason=finish_reason)
-            ],
+            choices=[StreamingChoices(index=0, delta=Delta(tool_calls=tool_calls), finish_reason=finish_reason)],
         )
 
     chunks: Final = [
@@ -2993,9 +2989,7 @@ def test_parallel_function_call_stream_over_responses_bridge_ignores_events_it_d
         },
     )
     route: Final = respx.post("https://api.openai.com/v1/responses").mock(
-        return_value=httpx.Response(
-            200, headers={"content-type": "text/event-stream"}, content=_responses_sse(events)
-        )
+        return_value=httpx.Response(200, headers={"content-type": "text/event-stream"}, content=_responses_sse(events))
     )
 
     response: Final = litellm.completion(
@@ -3609,9 +3603,7 @@ async def test_retrying() -> None:
         )
 
 
-def test_anthropic_disable_url_suffix_env_var(
-    monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter
-) -> None:
+def test_anthropic_disable_url_suffix_env_var(monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter) -> None:
     monkeypatch.setenv("ANTHROPIC_API_BASE", "https://api.example.com")
     default_route: Final = respx_mock.post("https://api.example.com/v1/messages").mock(
         return_value=httpx.Response(
@@ -3874,9 +3866,7 @@ async def test_async_mock_streaming_request():
         print(chunk)
         complete_response += chunk["choices"][0]["delta"]["content"] or ""
 
-    assert (
-        complete_response == "LiteLLM is awesome"
-    ), f"Unexpected response got {complete_response}"
+    assert complete_response == "LiteLLM is awesome", f"Unexpected response got {complete_response}"
 
 
 def test_mock_request_n_greater_than_1():
@@ -4402,7 +4392,10 @@ async def test_acompletion_forwards_store_and_prompt_cache_key_to_openai():
         assert request_body["prompt_cache_key"] == "test-cache-key"
 
 
-_TOOL: Final[Mapping[str, object]] = {"type": "function", "function": {"name": "f", "parameters": {"type": "object", "properties": {}}}}
+_TOOL: Final[Mapping[str, object]] = {
+    "type": "function",
+    "function": {"name": "f", "parameters": {"type": "object", "properties": {}}},
+}
 
 # Client settings ride in OPENAI_CHAT_COMPLETION_PARAMS but configure the HTTP client, never the body.
 _OPENAI_CLIENT_SETTINGS: Final = frozenset(
@@ -4462,7 +4455,9 @@ _OPENAI_BODY_PARAM_CASES: Final[Mapping[str, tuple[str, Mapping[str, object]]]]
 
 
 def test_every_openai_body_param_has_a_forwarding_case():
-    body_params: Final = frozenset(litellm.OPENAI_CHAT_COMPLETION_PARAMS) - _OPENAI_CLIENT_SETTINGS - _NOT_OPENAI_BODY_PARAMS
+    body_params: Final = (
+        frozenset(litellm.OPENAI_CHAT_COMPLETION_PARAMS) - _OPENAI_CLIENT_SETTINGS - _NOT_OPENAI_BODY_PARAMS
+    )
     assert frozenset(_OPENAI_BODY_PARAM_CASES) == body_params, (
         "a param added to OPENAI_CHAT_COMPLETION_PARAMS needs a case in _OPENAI_BODY_PARAM_CASES proving "
         "completion() sends it to OpenAI; prompt_cache_retention (LIT-9758) and store/prompt_cache_key "
@@ -5221,7 +5216,7 @@ def _stream_builder_logging_obj(model: str = "gpt-4o", custom_llm_provider: str
     return logging_obj
 
 
-def test_stream_chunk_builder_stamps_streaming_usage_cost_by_default(monkeypatch: pytest.MonkeyPatch):
+def test_stream_chunk_builder_bills_a_stream_without_writing_into_its_usage(monkeypatch: pytest.MonkeyPatch):
     monkeypatch.setattr(litellm, "include_cost_in_streaming_usage", False)
     chunks: Final = [
         _stream_builder_text_chunk("gpt-4o", "Hello "),
@@ -5233,10 +5228,28 @@ def test_stream_chunk_builder_stamps_streaming_usage_cost_by_default(monkeypatch
     )
 
     assert response is not None
-    usage_cost: Final = getattr(response.usage, "cost", None)
-    assert usage_cost is not None
-    assert usage_cost > 0
-    assert response._hidden_params["response_cost"] == pytest.approx(usage_cost)
+    assert getattr(response.usage, "cost", None) is None, "usage.cost is the upstream's field, the bill rides hidden"
+    assert response._hidden_params["response_cost"] > 0
+
+
+def test_stream_chunk_builder_keeps_a_zero_bill_as_hidden_response_cost() -> None:
+    logging_obj: Final = _stream_builder_logging_obj()
+    logging_obj.update_environment_variables(
+        litellm_params={"custom_llm_provider": "openai", "input_cost_per_token": 0.0, "output_cost_per_token": 0.0},
+        optional_params={},
+    )
+    logging_obj.custom_cost_per_token = {"input_cost_per_token": 0.0, "output_cost_per_token": 0.0}
+    chunks: Final = [
+        _stream_builder_text_chunk("gpt-4o", "Hello "),
+        _stream_builder_text_chunk("gpt-4o", "world.", finish_reason="stop"),
+    ]
+
+    response: Final = litellm.stream_chunk_builder(
+        chunks=chunks, messages=[{"role": "user", "content": "hi"}], logging_obj=logging_obj
+    )
+
+    assert response is not None
+    assert response._hidden_params["response_cost"] == 0.0, "a free stream still tells the egress what was billed"
 
 
 def test_stream_chunk_builder_stamped_cost_is_not_re_read_as_xai_reported_cost():
@@ -5267,7 +5280,8 @@ def test_stream_chunk_builder_stamped_cost_is_not_re_read_as_xai_reported_cost()
     )
 
     assert response is not None
-    usage_cost: Final = getattr(response.usage, "cost", None)
+    assert getattr(response.usage, "cost", None) is None
+    usage_cost: Final = response._hidden_params.get("response_cost")
     assert usage_cost is not None
     assert usage_cost > 0
 
@@ -5278,10 +5292,9 @@ def test_stream_chunk_builder_stamped_cost_is_not_re_read_as_xai_reported_cost()
     assert breakdown["output_cost"] > 0
     assert breakdown["input_cost"] + breakdown["output_cost"] == pytest.approx(usage_cost)
     assert spend_row_cost == pytest.approx(usage_cost)
-    assert response._hidden_params.get("response_cost") == pytest.approx(usage_cost)
 
 
-def test_stream_chunk_builder_skips_stamp_when_cost_is_unpriceable():
+def test_stream_chunk_builder_stamps_the_zero_bill_when_cost_is_unpriceable():
     import time as time_module
 
     from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
@@ -5310,7 +5323,7 @@ def test_stream_chunk_builder_skips_stamp_when_cost_is_unpriceable():
 
     assert response is not None
     assert getattr(response.usage, "cost", None) is None
-    assert response._hidden_params.get("response_cost") is None
+    assert response._hidden_params.get("response_cost") == 0.0, "the spend row reads 0, so the egress must too"
 
 
 def test_stream_chunk_builder_prices_alias_response_at_its_deployment_model():
@@ -5344,7 +5357,7 @@ def test_stream_chunk_builder_prices_alias_response_at_its_deployment_model():
     entry: Final = litellm.model_cost[deployment_model]
     expected_cost: Final = 40 * entry["input_cost_per_token"] + 5 * entry["output_cost_per_token"]
     assert response is not None
-    assert getattr(response.usage, "cost", None) == pytest.approx(expected_cost)
+    assert getattr(response.usage, "cost", None) is None
     assert response._hidden_params.get("response_cost") == pytest.approx(expected_cost)
 
 
@@ -5387,7 +5400,9 @@ def test_stream_chunk_builder_prices_alias_from_openai_sdk_usage_chunk():
     assert response._hidden_params["response_cost"] == pytest.approx(0.000704)
 
 
-def test_stream_chunk_builder_leaves_xai_reported_cost_to_the_calculator(monkeypatch: pytest.MonkeyPatch):
+def test_stream_chunk_builder_bills_the_margin_over_the_xai_reported_cost_it_leaves_in_usage(
+    monkeypatch: pytest.MonkeyPatch,
+):
     monkeypatch.setattr(litellm, "cost_margin_config", {"xai": 0.5})
     usage_chunk: Final = _stream_builder_text_chunk("grok-4", "")
     usage_chunk.usage = Usage(prompt_tokens=5, completion_tokens=2, total_tokens=7, cost=0.42)
@@ -5403,9 +5418,10 @@ def test_stream_chunk_builder_leaves_xai_reported_cost_to_the_calculator(monkeyp
     )
 
     assert response is not None
-    assert getattr(response.usage, "cost", None) == pytest.approx(0.42)
-    assert response._hidden_params.get("response_cost") is None
+    assert getattr(response.usage, "cost", None) == pytest.approx(0.42), "the upstream's figure stays in usage.cost"
+    assert response._hidden_params.get("response_cost") == pytest.approx(0.63)
     assert logging_obj.response_cost_calculator(result=response) == pytest.approx(0.63)
+    assert logging_obj.response_cost_calculator(result=response, ignore_stamped_cost=True) == pytest.approx(0.63)
 
 
 def test_speech_mistral_dispatches_and_decodes_audio(respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch):
@@ -5908,17 +5924,11 @@ def test_drop_params_false_still_rejects_an_invalid_stream_chunk_size() -> None:
 def test_acompletion_params():
     import inspect
 
-    from litellm.types.completion import CompletionRequest
-
     acompletion_params_odict = inspect.signature(acompletion).parameters
     completion_params_dict = inspect.signature(completion).parameters
 
-    acompletion_params = {
-        name: param.annotation for name, param in acompletion_params_odict.items()
-    }
-    completion_params = {
-        name: param.annotation for name, param in completion_params_dict.items()
-    }
+    acompletion_params = {name: param.annotation for name, param in acompletion_params_odict.items()}
+    completion_params = {name: param.annotation for name, param in completion_params_dict.items()}
 
     keys_acompletion = set(acompletion_params.keys())
     keys_completion = set(completion_params.keys())
@@ -5972,9 +5982,7 @@ def test_null_role_response():
     import openai
 
     openai_client = openai.OpenAI()
-    with patch.object(
-        openai_client, "post", side_effect=_openai_mock_response
-    ) as mock_response:
+    with patch.object(openai_client, "post", side_effect=_openai_mock_response) as mock_response:
         response = litellm.completion(
             model="gpt-3.5-turbo",
             messages=[{"role": "user", "content": "Hey! how's it going?"}],
@@ -6072,7 +6080,8 @@ def test_completion_perplexity_api():
         openai_client = OpenAI()
 
         with patch.object(
-            openai_client, "post",
+            openai_client,
+            "post",
             side_effect=_return_pydantic_obj,
         ) as mock_client:
             # litellm.set_verbose= True
@@ -6115,12 +6124,10 @@ async def test_openai_compatible_custom_api_base(provider):
 
     openai_client = OpenAI(api_key="fake-key")
 
-    with patch.object(
-        openai_client, "post", new=MagicMock()
-    ) as mock_call:
+    with patch.object(openai_client, "post", new=MagicMock()) as mock_call:
         try:
             completion(
-                model="{provider}/my-vllm-model".format(provider=provider),
+                model=f"{provider}/my-vllm-model",
                 messages=messages,
                 response_format={"type": "json_object"},
                 client=openai_client,
@@ -6132,7 +6139,7 @@ async def test_openai_compatible_custom_api_base(provider):
 
         mock_call.assert_called_once()
 
-        print("Call KWARGS - {}".format(mock_call.call_args.kwargs))
+        print(f"Call KWARGS - {mock_call.call_args.kwargs}")
 
         assert "hello" in mock_call.call_args.kwargs["options"]["extra_json"]
 
@@ -6166,12 +6173,10 @@ async def test_openai_compatible_custom_api_video(provider):
 
     openai_client = OpenAI(api_key="fake-key")
 
-    with patch.object(
-        openai_client, "post", new=MagicMock()
-    ) as mock_call:
+    with patch.object(openai_client, "post", new=MagicMock()) as mock_call:
         try:
             completion(
-                model="{provider}/my-vllm-model".format(provider=provider),
+                model=f"{provider}/my-vllm-model",
                 messages=messages,
                 response_format={"type": "json_object"},
                 client=openai_client,
@@ -6361,10 +6366,7 @@ async def test_completion_functions_param():
             mock_client.assert_called()
             print(f"mock_client.call_args.kwargs: {mock_client.call_args.kwargs}")
             assert "tools" in mock_client.call_args.kwargs["json"]
-            assert (
-                "litellm_param_is_function_call"
-                not in mock_client.call_args.kwargs["json"]
-            )
+            assert "litellm_param_is_function_call" not in mock_client.call_args.kwargs["json"]
             assert response.choices[0].message.function_call is not None
     except Exception as e:
         pytest.fail(f"Error occurred: {e}")
@@ -6427,7 +6429,7 @@ def test_bedrock_deepseek_custom_prompt_dict():
                 messages=messages,
                 client=client,
             )
-        except Exception as e:
+        except Exception:
             pass
 
         mock_post.assert_called_once()
@@ -6440,9 +6442,7 @@ def test_bedrock_deepseek_custom_prompt_dict():
 
 
 def test_bedrock_deepseek_known_tokenizer_config(monkeypatch):
-    model = (
-        "deepseek_r1/arn:aws:bedrock:us-west-2:888602223428:imported-model/bnnr6463ejgf"
-    )
+    model = "deepseek_r1/arn:aws:bedrock:us-west-2:888602223428:imported-model/bnnr6463ejgf"
     from unittest.mock import Mock
 
     import httpx
@@ -6513,9 +6513,7 @@ def test_completion_anthropic_hanging():
         {"role": "function", "name": "get_capital", "content": "Kokoko"},
     ]
 
-    converted_messages = anthropic_messages_pt(
-        messages, model="claude-3-sonnet-20240229", llm_provider="anthropic"
-    )
+    converted_messages = anthropic_messages_pt(messages, model="claude-3-sonnet-20240229", llm_provider="anthropic")
 
     assert len(converted_messages) == 3
     for i, msg in enumerate(converted_messages):
@@ -6822,9 +6820,7 @@ def test_completion_logprobs_parses_top_logprobs():
         "top_logprobs": 3,
         "reasoning_effort": "none",
     }
-    assert [
-        item.token for item in response.choices[0].logprobs.content[0].top_logprobs
-    ] == ["hello", "hi", "hey"]
+    assert [item.token for item in response.choices[0].logprobs.content[0].top_logprobs] == ["hello", "hi", "hey"]
     choice: Final = response.choices[0]
     assert "logprobs" in choice
     assert "content" in choice.logprobs
@@ -6896,9 +6892,11 @@ def test_completion_logprobs_stream_parses_top_logprobs():
         "stream_options": {"include_usage": True},
         "stream": True,
     }
-    assert [
-        item.token for item in response_chunks[0].choices[0].logprobs.content[0].top_logprobs
-    ] == ["hello", "hi", "hey"]
+    assert [item.token for item in response_chunks[0].choices[0].logprobs.content[0].top_logprobs] == [
+        "hello",
+        "hi",
+        "hey",
+    ]
 
 
 @respx.mock
@@ -6969,9 +6967,7 @@ async def test_bedrock_converse_parses_scripted_response(
         "api_key": "test-bedrock-token",
         "aws_region_name": "us-west-2",
     }
-    response: Final = (
-        await litellm.acompletion(**request) if async_mode else litellm.completion(**request)
-    )
+    response: Final = await litellm.acompletion(**request) if async_mode else litellm.completion(**request)
 
     assert route.calls[0].request.headers["authorization"] == "Bearer test-bedrock-token"
     assert json.loads(route.calls[0].request.content) == {
@@ -7058,9 +7054,7 @@ def test_bedrock_converse_parses_scripted_tool_call_arguments(
     tool_call: Final = response.choices[0].message.tool_calls[0]
     assert route.call_count == 1
     assert tool_call.function.name == "trade"
-    assert json.loads(tool_call.function.arguments) == {
-        "orders": [{"action": "buy", "asset": "BTC", "amount": 0.1}]
-    }
+    assert json.loads(tool_call.function.arguments) == {"orders": [{"action": "buy", "asset": "BTC", "amount": 0.1}]}
 
 
 @pytest.mark.parametrize(
@@ -7098,9 +7092,9 @@ def test_bedrock_invoke_signs_request_and_parses_scripted_response(
     expected_request_body: dict[str, object],
     fake_provider_credentials: None,
 ):
-    route: Final = respx.post(
-        f"https://bedrock-runtime.us-west-2.amazonaws.com/model/{invoke_model_id}/invoke"
-    ).mock(return_value=httpx.Response(200, json=response_body))
+    route: Final = respx.post(f"https://bedrock-runtime.us-west-2.amazonaws.com/model/{invoke_model_id}/invoke").mock(
+        return_value=httpx.Response(200, json=response_body)
+    )
 
     response: Final = litellm.completion(
         model=model,
@@ -7111,9 +7105,7 @@ def test_bedrock_invoke_signs_request_and_parses_scripted_response(
     )
 
     request: Final = route.calls[0].request
-    assert request.headers["authorization"].startswith(
-        "AWS4-HMAC-SHA256 Credential=unit-test/"
-    )
+    assert request.headers["authorization"].startswith("AWS4-HMAC-SHA256 Credential=unit-test/")
     assert "/us-west-2/bedrock/aws4_request" in request.headers["authorization"]
     assert request.headers["x-amz-date"].endswith("Z")
     assert json.loads(request.content) == expected_request_body
@@ -7123,9 +7115,9 @@ def test_bedrock_invoke_signs_request_and_parses_scripted_response(
 
 @respx.mock
 def test_completion_azure_ad_token_is_forwarded(openai_api_response):
-    route: Final = respx.post(
-        "https://example.openai.azure.com/openai/deployments/gpt-6.1-sol/chat/completions"
-    ).mock(return_value=httpx.Response(200, json=openai_api_response))
+    route: Final = respx.post("https://example.openai.azure.com/openai/deployments/gpt-6.1-sol/chat/completions").mock(
+        return_value=httpx.Response(200, json=openai_api_response)
+    )
 
     litellm.completion(
         model="azure/gpt-6.1-sol",
@@ -7140,9 +7132,9 @@ def test_completion_azure_ad_token_is_forwarded(openai_api_response):
 
 @respx.mock
 def test_completion_azure_extra_headers_are_forwarded(openai_api_response):
-    route: Final = respx.post(
-        "https://example.openai.azure.com/openai/deployments/gpt-6.1-sol/chat/completions"
-    ).mock(return_value=httpx.Response(200, json=openai_api_response))
+    route: Final = respx.post("https://example.openai.azure.com/openai/deployments/gpt-6.1-sol/chat/completions").mock(
+        return_value=httpx.Response(200, json=openai_api_response)
+    )
 
     litellm.completion(
         model="azure/gpt-6.1-sol",
@@ -7163,9 +7155,9 @@ def test_completion_azure_extra_headers_are_forwarded(openai_api_response):
 
 @respx.mock
 def test_completion_azure_api_key_argument_is_forwarded(openai_api_response):
-    route: Final = respx.post(
-        "https://example.openai.azure.com/openai/deployments/gpt-6.1-sol/chat/completions"
-    ).mock(return_value=httpx.Response(200, json=openai_api_response))
+    route: Final = respx.post("https://example.openai.azure.com/openai/deployments/gpt-6.1-sol/chat/completions").mock(
+        return_value=httpx.Response(200, json=openai_api_response)
+    )
 
     response: Final = litellm.completion(
         model="azure/gpt-6.1-sol",
@@ -7822,9 +7814,7 @@ def test_completion_deep_infra(drop_params):
         ],
         "usage": {"prompt_tokens": 20, "completion_tokens": 5, "total_tokens": 25},
     }
-    route: Final = respx.post(endpoint).mock(
-        return_value=httpx.Response(200, json=response_body)
-    )
+    route: Final = respx.post(endpoint).mock(return_value=httpx.Response(200, json=response_body))
 
     if not drop_params:
         with pytest.raises(litellm.exceptions.UnsupportedParamsError):
@@ -7888,9 +7878,7 @@ def test_completion_deep_infra_mistral():
         ],
         "usage": {"prompt_tokens": 20, "completion_tokens": 5, "total_tokens": 25},
     }
-    route: Final = respx.post(endpoint).mock(
-        return_value=httpx.Response(200, json=response_body)
-    )
+    route: Final = respx.post(endpoint).mock(return_value=httpx.Response(200, json=response_body))
 
     response: Final = completion(
         model=model_name,
@@ -7939,19 +7927,13 @@ def test_unified_auth_params(provider, model, project, region_name, token):
     translated_optional_params = litellm.utils.get_optional_params(**data)
 
     if provider == "azure":
-        special_auth_params = (
-            litellm.AzureOpenAIConfig().get_mapped_special_auth_params()
-        )
+        special_auth_params = litellm.AzureOpenAIConfig().get_mapped_special_auth_params()
     elif provider == "bedrock":
-        special_auth_params = (
-            litellm.AmazonBedrockGlobalConfig().get_mapped_special_auth_params()
-        )
+        special_auth_params = litellm.AmazonBedrockGlobalConfig().get_mapped_special_auth_params()
     elif provider == "vertex_ai":
         special_auth_params = litellm.VertexAIConfig().get_mapped_special_auth_params()
     elif provider == "watsonx":
-        special_auth_params = (
-            litellm.IBMWatsonXAIConfig().get_mapped_special_auth_params()
-        )
+        special_auth_params = litellm.IBMWatsonXAIConfig().get_mapped_special_auth_params()
 
     for param, value in special_auth_params.items():
         assert param in data
@@ -7983,9 +7965,7 @@ async def test_dynamic_azure_params(stream, sync_mode):
         mock_client = AsyncMock(return_value="Hello world!")
 
     ## CHECK IF CLIENT IS USED (NO PARAM CHANGE)
-    with patch.object(
-        client.chat.completions.with_raw_response, "create", new=mock_client
-    ) as mock_client:
+    with patch.object(client.chat.completions.with_raw_response, "create", new=mock_client) as mock_client:
         try:
             # client.chat.completions.with_raw_response.create = mock_client
             if sync_mode:
@@ -8014,9 +7994,7 @@ async def test_dynamic_azure_params(stream, sync_mode):
         new_mock_client = AsyncMock(return_value="Hello world!")
 
     ## CHECK IF NEW CLIENT IS USED (PARAM CHANGE)
-    with patch.object(
-        client.chat.completions.with_raw_response, "create", new=new_mock_client
-    ) as new_mock_client:
+    with patch.object(client.chat.completions.with_raw_response, "create", new=new_mock_client) as new_mock_client:
         try:
             if sync_mode:
                 _ = completion(
@@ -8227,9 +8205,7 @@ def test_completion_novita_ai():
 
     openai_client = OpenAI(api_key="fake-key")
 
-    with patch.object(
-        openai_client, "post", return_value=MagicMock(spec=LegacyAPIResponse)
-    ) as mock_call:
+    with patch.object(openai_client, "post", return_value=MagicMock(spec=LegacyAPIResponse)) as mock_call:
         mock_call.return_value.headers = {}
         mock_call.return_value.parse.return_value = litellm.ModelResponse(
             choices=[{"message": {"role": "assistant", "content": "Hello"}}]
@@ -8246,10 +8222,7 @@ def test_completion_novita_ai():
             assert response.choices[0].message.content == "Hello"
 
             # Verify model is passed correctly
-            assert (
-                mock_call.call_args.kwargs["body"]["model"]
-                == "meta-llama/llama-3.3-70b-instruct"
-            )
+            assert mock_call.call_args.kwargs["body"]["model"] == "meta-llama/llama-3.3-70b-instruct"
             # Verify messages are passed correctly
             assert mock_call.call_args.kwargs["body"]["messages"] == messages
 
@@ -8306,9 +8279,7 @@ def test_completion_openai_metadata(monkeypatch, enable_preview_features):
     litellm.set_verbose = True
 
     monkeypatch.setattr(litellm, "enable_preview_features", enable_preview_features)
-    with patch.object(
-        client, "post", return_value=MagicMock()
-    ) as mock_completion:
+    with patch.object(client, "post", return_value=MagicMock()) as mock_completion:
         try:
             resp = litellm.completion(
                 model="openai/gpt-3.5-turbo",
@@ -8321,9 +8292,7 @@ def test_completion_openai_metadata(monkeypatch, enable_preview_features):
 
         mock_completion.assert_called_once()
         if enable_preview_features:
-            assert mock_completion.call_args.kwargs["body"]["metadata"] == {
-                "my-test-key": "my-test-value"
-            }
+            assert mock_completion.call_args.kwargs["body"]["metadata"] == {"my-test-key": "my-test-value"}
         else:
             assert "metadata" not in mock_completion.call_args.kwargs["body"]
 
@@ -8544,17 +8513,30 @@ async def test_async_responses_still_retries_provider_server_errors(monkeypatch)
     monkeypatch.setattr(litellm, "num_retries", None)
     monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
     with respx.mock as upstream:
-        response = upstream.post("https://openai-test.invalid/v1/responses").mock(side_effect=[
-            httpx.Response(500, json={"error": {"message": "temporary provider failure", "type": "server_error"}}),
-            httpx.Response(200, json={
-                "id": "resp-retry", "object": "response", "created_at": 1, "status": "completed",
-                "model": "test-model", "output": [],
-                "usage": {"input_tokens": 3, "output_tokens": 2, "total_tokens": 5},
-            }),
-        ])
+        response = upstream.post("https://openai-test.invalid/v1/responses").mock(
+            side_effect=[
+                httpx.Response(500, json={"error": {"message": "temporary provider failure", "type": "server_error"}}),
+                httpx.Response(
+                    200,
+                    json={
+                        "id": "resp-retry",
+                        "object": "response",
+                        "created_at": 1,
+                        "status": "completed",
+                        "model": "test-model",
+                        "output": [],
+                        "usage": {"input_tokens": 3, "output_tokens": 2, "total_tokens": 5},
+                    },
+                ),
+            ]
+        )
         result = await litellm.aresponses(
-            model="openai/test-model", input="ping", api_key="test-key",
-            api_base="https://openai-test.invalid/v1", num_retries=1, max_retries=0,
+            model="openai/test-model",
+            input="ping",
+            api_key="test-key",
+            api_base="https://openai-test.invalid/v1",
+            num_retries=1,
+            max_retries=0,
         )
         assert result.status == "completed"
         assert response.call_count == 2
@@ -8740,9 +8722,7 @@ def test_openai_chat_stream_routes_legacy_function_calls_and_usage(
             "object": "chat.completion.chunk",
             "created": 1,
             "model": "gpt-4o-mini",
-            "choices": [
-                {"index": 0, "delta": {}, "finish_reason": "function_call"}
-            ],
+            "choices": [{"index": 0, "delta": {}, "finish_reason": "function_call"}],
         },
     )
 
@@ -8766,9 +8746,7 @@ def test_openai_chat_stream_routes_legacy_function_calls_and_usage(
             ),
         )
     )
-    route: Final = respx.post(f"{api_base}/chat/completions").mock(
-        side_effect=lambda _: next(scripted_responses)
-    )
+    route: Final = respx.post(f"{api_base}/chat/completions").mock(side_effect=lambda _: next(scripted_responses))
     stream: Final = litellm.completion(
         model="gpt-4o-mini",
         api_key="sk-test",
@@ -8779,11 +8757,14 @@ def test_openai_chat_stream_routes_legacy_function_calls_and_usage(
     )
     answer_chunks: Final = tuple(stream)
 
-    assert "".join(
-        chunk.choices[0].delta.content
-        for chunk in answer_chunks
-        if chunk.choices and chunk.choices[0].delta.content
-    ) == "The answer is 42."
+    assert (
+        "".join(
+            chunk.choices[0].delta.content
+            for chunk in answer_chunks
+            if chunk.choices and chunk.choices[0].delta.content
+        )
+        == "The answer is 42."
+    )
     assert tuple(
         chunk.choices[0].finish_reason
         for chunk in answer_chunks
@@ -8800,12 +8781,7 @@ def test_openai_chat_stream_routes_legacy_function_calls_and_usage(
     assert answer_chunks[-1].usage.completion_tokens == 2
     assert answer_chunks[-1].usage.total_tokens == 6
     assert stream._hidden_params["api_base"] == api_base
-    assert (
-        answer_chunks[0]._hidden_params["additional_headers"][
-            "llm_provider-x-ratelimit-remaining-requests"
-        ]
-        == "17"
-    )
+    assert answer_chunks[0]._hidden_params["additional_headers"]["llm_provider-x-ratelimit-remaining-requests"] == "17"
 
     function_stream: Final = litellm.completion(
         model="gpt-4o-mini",
@@ -8817,9 +8793,7 @@ def test_openai_chat_stream_routes_legacy_function_calls_and_usage(
         stream=True,
     )
     function_chunks: Final = tuple(function_stream)
-    assembled: Final = litellm.stream_chunk_builder(
-        chunks=function_chunks, messages=messages
-    )
+    assembled: Final = litellm.stream_chunk_builder(chunks=function_chunks, messages=messages)
     function_call: Final = assembled.choices[0].message.function_call
 
     assert function_call is not None
@@ -8831,12 +8805,8 @@ def test_openai_chat_stream_routes_legacy_function_calls_and_usage(
         if chunk.choices and chunk.choices[0].finish_reason is not None
     ) == ("function_call",)
     assert route.call_count == 2
-    assert json.loads(route.calls[0].request.content)["stream_options"] == {
-        "include_usage": True
-    }
-    assert json.loads(route.calls[1].request.content)["function_call"] == {
-        "name": "get_current_weather"
-    }
+    assert json.loads(route.calls[0].request.content)["stream_options"] == {"include_usage": True}
+    assert json.loads(route.calls[1].request.content)["function_call"] == {"name": "get_current_weather"}
     assert json.loads(route.calls[1].request.content)["functions"] == list(functions)
 
 
@@ -8875,15 +8845,9 @@ async def test_async_openai_stream_options_include_usage_on_the_last_chunk(
             "usage": {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5},
         },
     )
-    body: Final = "".join(
-        f"data: {json.dumps(event)}\n\n" for event in events
-    ) + "data: [DONE]\n\n"
-    route: Final = respx.post(
-        "https://api.openai.com/v1/chat/completions"
-    ).mock(
-        return_value=httpx.Response(
-            200, text=body, headers={"content-type": "text/event-stream"}
-        )
+    body: Final = "".join(f"data: {json.dumps(event)}\n\n" for event in events) + "data: [DONE]\n\n"
+    route: Final = respx.post("https://api.openai.com/v1/chat/completions").mock(
+        return_value=httpx.Response(200, text=body, headers={"content-type": "text/event-stream"})
     )
     stream: Final = await litellm.acompletion(
         model="gpt-4o-mini",
@@ -8894,11 +8858,10 @@ async def test_async_openai_stream_options_include_usage_on_the_last_chunk(
     )
     chunks: Final = tuple([chunk async for chunk in stream])
 
-    assert "".join(
-        chunk.choices[0].delta.content
-        for chunk in chunks
-        if chunk.choices and chunk.choices[0].delta.content
-    ) == "async reply"
+    assert (
+        "".join(chunk.choices[0].delta.content for chunk in chunks if chunk.choices and chunk.choices[0].delta.content)
+        == "async reply"
+    )
     assert tuple(getattr(chunk, "usage", None) for chunk in chunks[:-1]) == (
         None,
         None,
@@ -8906,9 +8869,7 @@ async def test_async_openai_stream_options_include_usage_on_the_last_chunk(
     assert chunks[-1].usage.prompt_tokens == 3
     assert chunks[-1].usage.completion_tokens == 2
     assert chunks[-1].usage.total_tokens == 5
-    assert json.loads(route.calls[0].request.content)["stream_options"] == {
-        "include_usage": True
-    }
+    assert json.loads(route.calls[0].request.content)["stream_options"] == {"include_usage": True}
 
 
 @respx.mock
@@ -8945,16 +8906,10 @@ async def test_parallel_openai_streams_keep_prompt_responses_separate(
                 "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
             },
         )
-        body: Final = "".join(
-            f"data: {json.dumps(event)}\n\n" for event in events
-        ) + "data: [DONE]\n\n"
-        return httpx.Response(
-            200, text=body, headers={"content-type": "text/event-stream"}
-        )
+        body: Final = "".join(f"data: {json.dumps(event)}\n\n" for event in events) + "data: [DONE]\n\n"
+        return httpx.Response(200, text=body, headers={"content-type": "text/event-stream"})
 
-    route: Final = respx.post(
-        "https://api.openai.com/v1/chat/completions"
-    ).mock(side_effect=respond)
+    route: Final = respx.post("https://api.openai.com/v1/chat/completions").mock(side_effect=respond)
 
     async def collect(prompt: str) -> tuple[str, tuple[str, ...]]:
         stream = await litellm.acompletion(
@@ -8964,9 +8919,7 @@ async def test_parallel_openai_streams_keep_prompt_responses_separate(
             stream=True,
         )
         chunks: Final = tuple([chunk async for chunk in stream])
-        response_text: Final = "".join(
-            chunk.choices[0].delta.content or "" for chunk in chunks if chunk.choices
-        )
+        response_text: Final = "".join(chunk.choices[0].delta.content or "" for chunk in chunks if chunk.choices)
         finish_reasons: Final = tuple(
             chunk.choices[0].finish_reason
             for chunk in chunks
@@ -8974,9 +8927,7 @@ async def test_parallel_openai_streams_keep_prompt_responses_separate(
         )
         return response_text, finish_reasons
 
-    responses: Final = tuple(
-        await asyncio.gather(collect("request one"), collect("request two"))
-    )
+    responses: Final = tuple(await asyncio.gather(collect("request one"), collect("request two")))
 
     assert tuple(response for response, _ in responses) == (
         "response one",
@@ -9149,7 +9100,7 @@ def test_openai_streaming_tool_call_arguments_are_valid_json(
 
 
 @respx.mock
-def test_streaming_openai_usage_includes_test_owned_cost(
+def test_streaming_openai_usage_carries_no_cost_the_upstream_did_not_report(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
@@ -9170,9 +9121,7 @@ def test_streaming_openai_usage_includes_test_owned_cost(
             "object": "chat.completion.chunk",
             "created": 1,
             "model": model_name,
-            "choices": [
-                {"index": 0, "delta": {"role": "assistant", "content": "priced"}, "finish_reason": None}
-            ],
+            "choices": [{"index": 0, "delta": {"role": "assistant", "content": "priced"}, "finish_reason": None}],
         },
         {
             "id": "chatcmpl-cost",
@@ -9207,12 +9156,10 @@ def test_streaming_openai_usage_includes_test_owned_cost(
     )
     stream_chunks: Final = tuple(stream)
     usage_chunks: Final = tuple(
-        getattr(chunk, "usage", None)
-        for chunk in stream_chunks
-        if getattr(chunk, "usage", None) is not None
+        getattr(chunk, "usage", None) for chunk in stream_chunks if getattr(chunk, "usage", None) is not None
     )
     assert len(usage_chunks) == 1
-    assert usage_chunks[0].cost == pytest.approx(4 * 0.001 + 2 * 0.002)
+    assert getattr(usage_chunks[0], "cost", None) is None, "the bill reaches a client through the proxy egress only"
 
 
 def test_completion_ensure_alternating_roles_sends_continue_messages(respx_mock: respx.MockRouter) -> None:
@@ -9225,9 +9172,7 @@ def test_completion_ensure_alternating_roles_sends_continue_messages(respx_mock:
                 "object": "chat.completion",
                 "created": 0,
                 "model": "databricks-meta-llama-3-1-70b-instruct",
-                "choices": [
-                    {"index": 0, "message": {"role": "assistant", "content": "Sure"}, "finish_reason": "stop"}
-                ],
+                "choices": [{"index": 0, "message": {"role": "assistant", "content": "Sure"}, "finish_reason": "stop"}],
                 "usage": {"prompt_tokens": 40, "completion_tokens": 1, "total_tokens": 41},
             },
         )
@@ -9437,9 +9382,12 @@ def test_azure_ai_forwards_reasoning_effort_once_the_deployment_declares_levels(
     assert _chat_body_sent(respx_mock, "azure_ai/my-kimi", "adaptive", _KIMI_DEPLOYMENT)["reasoning_effort"] == (
         "adaptive"
     )
-    assert _chat_body_sent(respx_mock, "azure_ai/my-kimi", "none", {"reasoning_effort_levels": ["low", "high"]})[
-        "reasoning_effort"
-    ] == "none"
+    assert (
+        _chat_body_sent(respx_mock, "azure_ai/my-kimi", "none", {"reasoning_effort_levels": ["low", "high"]})[
+            "reasoning_effort"
+        ]
+        == "none"
+    )
 
 
 def test_a_model_map_declaration_alone_leaves_reasoning_effort_to_drop_params(
@@ -9574,7 +9522,9 @@ def test_a_default_effort_beside_a_thinking_budget_sends_no_effort_on_a_declarin
         thinking={"type": "enabled", "budget_tokens": 2048, **display},
         **default_effort,
     )
-    adaptive_alone: Final = body_sent(respx_mock, _OPUS_DECLARING_LOW_AND_HIGH, thinking={"type": "adaptive", **display})
+    adaptive_alone: Final = body_sent(
+        respx_mock, _OPUS_DECLARING_LOW_AND_HIGH, thinking={"type": "adaptive", **display}
+    )
 
     assert "output_config" not in defaulted
     assert defaulted == adaptive_alone
@@ -10079,7 +10029,9 @@ async def test_aspeech_answers_a_non_string_provider_hint_with_a_bad_request(
     hint: str, respx_mock: respx.MockRouter
 ) -> None:
     with pytest.raises(litellm.BadRequestError) as raised:
-        await litellm.aspeech(model="kokoro", input="hello", voice="alloy", api_key="fake_openai_api_key", **{hint: 123})
+        await litellm.aspeech(
+            model="kokoro", input="hello", voice="alloy", api_key="fake_openai_api_key", **{hint: 123}
+        )
 
     assert raised.value.status_code == 400
     assert f"Invalid {hint}" in raised.value.message
@@ -10269,7 +10221,9 @@ async def test_converse_fallback_prices_the_current_attempt(
     }
     if asynchronous:
         with pytest.raises(litellm.BadRequestError):
-            await litellm.acompletion(model="bedrock/converse/amazon.nova-micro-v1:0", model_id=failed_profile, **kwargs)
+            await litellm.acompletion(
+                model="bedrock/converse/amazon.nova-micro-v1:0", model_id=failed_profile, **kwargs
+            )
     else:
         with pytest.raises(litellm.BadRequestError):
             litellm.completion(model="bedrock/converse/amazon.nova-micro-v1:0", model_id=failed_profile, **kwargs)
@@ -10293,9 +10247,7 @@ async def test_converse_fallback_prices_the_current_attempt(
     assert result is not None
     prices: Final = litellm.get_model_info(profile or model, custom_llm_provider="bedrock")
     expected: Final = (
-        0.001 + 0.002
-        if custom_price
-        else prices["input_cost_per_token"] + prices["output_cost_per_token"]
+        0.001 + 0.002 if custom_price else prices["input_cost_per_token"] + prices["output_cost_per_token"]
     )
     active_logging: Final = response.logging_obj if stream else logging_obj
     assert active_logging.response_cost_calculator(result) == pytest.approx(expected)