refactor(cost): write the bill into usage.cost once at the proxy egress instead of inside the pipeline

litellm/cost_calculator.py

@@ -25,7 +25,7 @@ from litellm.constants import (
 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, provider_stated_usage_cost
+from litellm.litellm_core_utils.hidden_params import HIDDEN_PARAMS_ATTR
 from litellm.litellm_core_utils.llm_cost_calc.tool_call_cost_tracking import (
     StandardBuiltInToolCostTracking,
 )
@@ -1560,16 +1560,6 @@ def _store_cost_breakdown_in_logging_obj(
         # Don't fail the main cost calculation if breakdown storage fails
 
 
-def _with_provider_stated_cost(completion_response: object, usage: Usage | None) -> Usage | None:
-    if usage is None:
-        return usage
-    stamped_cost: Final = getattr(usage, "cost", None)
-    stated_cost: Final = provider_stated_usage_cost(completion_response, stamped_cost)
-    if stated_cost is stamped_cost:
-        return usage
-    return usage.model_copy(update=MappingProxyType({"cost": stated_cost}))
-
-
 def _without_provider_stated_cost(usage: Usage | None) -> Usage | None:
     if usage is None or getattr(usage, "cost", None) is None:
         return usage
@@ -1712,9 +1702,7 @@ def completion_cost(
         cache_creation_input_tokens: int | None = None
         cache_read_input_tokens: int | None = None
         audio_transcription_file_duration: float = 0.0
-        provider_usage_object: Final = _with_provider_stated_cost(
-            completion_response, get_usage_object(completion_response=completion_response)
-        )
+        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 or base_model is not None

litellm/litellm_core_utils/hidden_params.py

@@ -134,32 +134,3 @@ def get_or_create_hidden_params(obj: object) -> MutableMapping[str, object]:
     if hidden_params_mapping is not None:
         return hidden_params_mapping
     raise TypeError(f"unsupported hidden params storage: {type(hidden_params).__name__}")
-
-
-USAGE_COST_BEFORE_LITELLM_STAMP: Final = "usage_cost_before_litellm_stamp"
-
-
-def stamp_billed_usage_cost(obj: object, usage: object, cost: float) -> None:
-    """usage.cost becomes what litellm billed; the provider's own figure is kept so a later pricing pass bills the same."""
-    hidden_params: Final = get_hidden_params(obj)
-    if hidden_params is None or USAGE_COST_BEFORE_LITELLM_STAMP not in hidden_params:
-        set_hidden_param(obj, USAGE_COST_BEFORE_LITELLM_STAMP, getattr(usage, "cost", None))
-    setattr(usage, "cost", cost)
-
-
-def stamp_billable_usage_cost(obj: object, usage: object, cost: object) -> float | None:
-    """Stamp a priced bill, a $0 one included once the provider put a number there; an unpriceable reply keeps its usage."""
-    if isinstance(cost, bool) or not isinstance(cost, (int, float)) or cost < 0:
-        return None
-    if cost == 0 and getattr(usage, "cost", None) is None:
-        return None
-    stamp_billed_usage_cost(obj, usage, float(cost))
-    return float(cost)
-
-
-def provider_stated_usage_cost(obj: object, usage_cost: object) -> object:
-    """What the provider itself put in usage.cost, looking past a bill litellm stamped there."""
-    hidden_params: Final = get_hidden_params(obj)
-    if hidden_params is None or USAGE_COST_BEFORE_LITELLM_STAMP not in hidden_params:
-        return usage_cost
-    return hidden_params.get(USAGE_COST_BEFORE_LITELLM_STAMP)

litellm/litellm_core_utils/llm_response_utils/response_metadata.py

@@ -6,23 +6,15 @@ from typing import Final
 import httpx
 from pydantic import TypeAdapter, ValidationError
 
-import litellm
 from litellm.constants import LITELLM_DETAILED_TIMING
 from litellm.litellm_core_utils.core_helpers import get_litellm_metadata_from_kwargs, process_response_headers
-from litellm.litellm_core_utils.hidden_params import (
-    USAGE_COST_BEFORE_LITELLM_STAMP,
-    get_hidden_params,
-    stamp_billable_usage_cost,
-)
 from litellm.litellm_core_utils.llm_response_utils.get_api_base import get_api_base
 from litellm.litellm_core_utils.logging_utils import LiteLLMLoggingObject
-from litellm.types.llms.openai import ResponseAPIUsage, ResponsesAPIResponse
 from litellm.types.utils import (
     EmbeddingResponse,
     HiddenParams,
     ModelResponse,
     TranscriptionResponse,
-    Usage,
 )
 
 
@@ -167,18 +159,16 @@ class ResponseMetadata:
         ## ADD OTHER HIDDEN PARAMS
         model_info: Final = kwargs.get("model_info", {}) or {}
         model_id: Final = model_info.get("id", None)
-        response_cost: Final = logging_obj.response_cost_calculator(
-            result=self.result,
-            litellm_model_name=model,
-            router_model_id=model_id,
-            cache_hit=cache_hit,
-        )
-        self._stamp_billed_usage_cost(response_cost)
         new_params: Final = {
             "litellm_call_id": getattr(logging_obj, "litellm_call_id", None),
             "api_base": get_api_base(model=model or "", optional_params=kwargs),
             "model_id": model_id,
-            "response_cost": response_cost,
+            "response_cost": logging_obj.response_cost_calculator(
+                result=self.result,
+                litellm_model_name=model,
+                router_model_id=model_id,
+                cache_hit=cache_hit,
+            ),
             "additional_headers": process_response_headers(
                 self._get_additional_headers_from_hidden_params() or {},
                 preserve_litellm_internal_headers=True,
@@ -189,25 +179,6 @@ class ResponseMetadata:
         }
         self._update_hidden_params(new_params)
 
-    def _stamp_billed_usage_cost(self, response_cost: object) -> None:
-        """A usage.cost the upstream reported becomes what litellm billed, so a client and the spend log agree on a
-        priced deployment. A reply without one keeps its usage as the upstream sent it unless the operator asked for
-        the bill in every usage block with ``litellm.include_cost_in_streaming_usage``, the same switch that puts it
-        into streamed usage."""
-        if not isinstance(self.result, (ModelResponse, ResponsesAPIResponse)):
-            return
-        usage: Final = getattr(self.result, "usage", None)
-        if not isinstance(usage, (Usage, ResponseAPIUsage)):
-            return
-        if getattr(usage, "cost", None) is None and not litellm.include_cost_in_streaming_usage:
-            return
-        if stamp_billable_usage_cost(self.result, usage, response_cost) is None:
-            return
-        stamped: Final = get_hidden_params(self.result)
-        if stamped is None or USAGE_COST_BEFORE_LITELLM_STAMP not in stamped:
-            return
-        self._update_hidden_params({USAGE_COST_BEFORE_LITELLM_STAMP: stamped[USAGE_COST_BEFORE_LITELLM_STAMP]})
-
     def _update_hidden_params(self, new_params: Mapping[str, object]) -> None:
         """
         Update hidden params - handles when self._hidden_params is a dict or HiddenParams object

litellm/litellm_core_utils/streaming_chunk_builder_utils.py

@@ -13,7 +13,7 @@ from typing_extensions import ReadOnly, Required
 import litellm
 from litellm._logging import verbose_logger
 from litellm.litellm_core_utils.core_helpers import chunk_usage, provider_reported_usage_cost
-from litellm.litellm_core_utils.hidden_params import HIDDEN_PARAMS_ATTR, set_hidden_params, stamp_billable_usage_cost
+from litellm.litellm_core_utils.hidden_params import HIDDEN_PARAMS_ATTR, set_hidden_params
 from litellm.litellm_core_utils.llm_cost_calc.guardrail_cost import BEDROCK_TRACE_FIELDS, object_items, object_mapping
 from litellm.litellm_core_utils.prompt_templates.common_utils import (
     get_content_from_model_response,  # pyright: ignore[reportUnknownVariableType]  # response param accepts an untyped dict
@@ -1411,13 +1411,13 @@ 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
-    billed: Final = stamp_billable_usage_cost(response, usage, logging_obj.response_cost_calculator(result=response))
-    if billed is None:
+    computed_cost: Final = logging_obj.response_cost_calculator(result=response)
+    if not isinstance(computed_cost, (int, float)) or computed_cost <= 0:
         return
-    _set_hidden_response_cost(response, billed)
+    _set_hidden_response_cost(response, float(computed_cost))
 
 
 def _field_of(value: object, name: str) -> object:
@@ -1587,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,
@@ -1775,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/proxy/common_request_processing.py

@@ -130,6 +130,7 @@ from litellm.proxy.route_llm_request import (
     route_request,
 )
 from litellm.proxy.telemetry.request_context import bind_call_id as bind_telemetry_call_id
+from litellm.proxy.usage_cost_egress import with_client_usage_cost
 from litellm.proxy.utils import (  # noqa: F401  # legacy module exports
     ProxyLogging,
     _check_and_merge_model_level_guardrails,  # pyright: ignore[reportPrivateUsage,reportUnusedImport]  # backwards-compatible package export
@@ -273,6 +274,7 @@ from litellm.proxy.litellm_pre_call_utils import (
 )
 from litellm.proxy.policy_engine.response_retrieval import attach_post_call_pipelines_to_retrieval
 from litellm.types.litellm_params import STREAM_USAGE_INJECTED_FIELD
+from litellm.types.llms.openai import ResponsesAPIResponse
 from litellm.types.utils import (
     ModelResponse,
     ModelResponseStream,
@@ -3325,7 +3327,7 @@ class ProxyBaseLLMRequestProcessing:
                 user_api_key_dict=user_api_key_dict,
             )
 
-        return response
+        return with_client_usage_cost(response, response_cost_for_headers)
 
     @staticmethod
     async def _record_container_owners_from_responses_if_needed(
@@ -4710,6 +4712,7 @@ class ProxyBaseLLMRequestProcessing:
             completion_tokens_details=usage.get("completion_tokens_details"),
             prompt_tokens_details=usage.get("prompt_tokens_details"),
             server_tool_use=usage.get("server_tool_use"),
+            cost=usage.get("cost"),
         )
 
     @staticmethod
@@ -4724,7 +4727,7 @@ class ProxyBaseLLMRequestProcessing:
 
     @staticmethod
     def _completion_cost_or_none(
-        model_response: ModelResponse, model_name: str, service_tier: str | None
+        model_response: ModelResponse | ResponsesAPIResponse, model_name: str, service_tier: str | None
     ) -> float | None:
         try:
             return litellm.completion_cost(
@@ -4735,7 +4738,7 @@ class ProxyBaseLLMRequestProcessing:
 
     @staticmethod
     def _logging_obj_cost_or_none(
-        model_response: ModelResponse, litellm_logging_obj: LiteLLMLoggingObj
+        model_response: ModelResponse | ResponsesAPIResponse, litellm_logging_obj: LiteLLMLoggingObj
     ) -> float | None:
         # Pricing a frame stamps cost_breakdown and, on failure, the cost-failure debug key onto
         # the live logging object. The pass-through handlers never recompute either one, so a
@@ -4762,7 +4765,7 @@ class ProxyBaseLLMRequestProcessing:
 
     @staticmethod
     def _streamed_usage_cost(
-        model_response: ModelResponse,
+        model_response: ModelResponse | ResponsesAPIResponse,
         model_name: str,
         service_tier: str | None,
         litellm_logging_obj: LiteLLMLoggingObj | None,

litellm/proxy/proxy_server.py

@@ -922,6 +922,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 Pricer, with_client_stream_usage_cost
 from litellm.proxy.utils import (  # noqa: F401, RUF100  # legacy module exports
     PrismaClient,
     ProxyLogging,
@@ -10134,6 +10135,22 @@ def _fast_serialize_simple_model_response_stream(
     return orjson.dumps(payload)
 
 
+def _stream_usage_pricer(request_data: Mapping[str, object]) -> Pricer:
+    """Bills a usage-bearing stream event the way the request's logging object bills the spend row."""
+    logging_obj: Final = request_data.get("litellm_logging_obj")
+    model_name: Final = str(request_data.get("model", ""))
+
+    def price(reply: ModelResponse | ResponsesAPIResponse) -> float | None:
+        return ProxyBaseLLMRequestProcessing._streamed_usage_cost(  # pyright: ignore[reportPrivateUsage]  # the proxy's one stream pricer
+            reply,
+            model_name,
+            None,
+            logging_obj if isinstance(logging_obj, LiteLLMLoggingObj) else None,
+        )
+
+    return price
+
+
 def _serialize_streaming_chunk(chunk: BaseModel) -> str | bytes:
     if isinstance(chunk, ModelResponseStream):
         serialized_chunk: Final = _fast_serialize_simple_model_response_stream(chunk)
@@ -10472,6 +10489,7 @@ async def async_data_generator(
         needs_per_chunk_hook: Final = proxy_logging_obj.needs_per_chunk_streaming_hook()
         is_raw_sse_stream: Final = bool(request_data.get("_litellm_raw_sse_stream"))
         strip_stream_usage: Final = bool(request_data.get("_litellm_strip_stream_usage"))
+        price_stream_usage: Final = _stream_usage_pricer(request_data)
         raw_sse_buffer = ""
 
         # A stream can start on a deployment with keepalive off and fall back
@@ -10558,7 +10576,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, price_stream_usage))
             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,77 @@
+"""What a client reads in ``usage.cost``: litellm's bill, written once on the way out of the proxy, onto a copy.
+
+Inside the pipeline ``usage.cost`` stays what the upstream reported, so pricing a reply again (an MCP round, a
+logging copy, a stream rebuilt from its chunks) never reads litellm's own bill back as a provider report.
+"""
+
+from __future__ import annotations
+
+import math
+from collections.abc import Callable
+from typing import Final, TypeVar
+
+from pydantic import BaseModel
+
+import litellm
+from litellm.types.llms.openai import ResponseAPIUsage, ResponsesAPIResponse
+from litellm.types.utils import ModelResponse, ModelResponseStream, Usage
+
+Pricer = Callable[[ModelResponse | ResponsesAPIResponse], float | None]
+"""Prices a reply the way the request's logging object bills it (deployment rates, discount, margin)."""
+
+_RESPONSES_USAGE_EVENTS: Final = frozenset({"response.completed", "response.incomplete"})
+_ReplyT = 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, *, wanted: bool = False) -> float | None:
+    """The bill, when the upstream put a number in usage.cost or the operator wants it in every usage block
+    (``litellm.include_cost_in_streaming_usage``). None leaves the usage as the upstream sent it."""
+    reported: Final = isinstance(upstream_cost, (int, float)) and not isinstance(upstream_cost, bool)
+    if not reported and not wanted and not litellm.include_cost_in_streaming_usage:
+        return None
+    return billable(bill)
+
+
+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 with_client_usage_cost(response: object, bill: object) -> object:
+    """A plain chat or Responses reply as the client gets it: a copy with the bill in usage.cost, or the reply
+    itself when its usage stays as the upstream sent it."""
+    if not isinstance(response, (ModelResponse, ResponsesAPIResponse)):
+        return response
+    usage: Final = getattr(response, "usage", None)
+    if not isinstance(usage, (Usage, ResponseAPIUsage)):
+        return response
+    cost: Final = client_usage_cost(getattr(usage, "cost", None), bill)
+    if cost is None:
+        return response
+    return _with_usage_cost(response, usage, cost)
+
+
+def with_client_stream_usage_cost(chunk: BaseModel, price: Pricer) -> BaseModel:
+    """A streamed event as the client gets it. Only the usage-bearing events are priced: the chat chunk carrying
+    ``usage`` and the Responses ``response.completed`` / ``response.incomplete`` events, which carry the bill
+    whether or not the upstream reported one, as they did before the bill moved out here."""
+    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), price(ModelResponse(usage=usage)))
+        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), price(response), wanted=True)
+    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,
@@ -1480,7 +1480,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
                 }
             )
         )
-        stamp_responses_usage_cost(partial, self.litellm_logging_obj)
+        normalize_responses_usage(partial)
         return ResponseIncompleteEvent(type=ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE, response=partial)
 
     def _emit_terminal_response_event(

litellm/responses/streaming_iterator.py

@@ -41,7 +41,6 @@ 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 USAGE_COST_BEFORE_LITELLM_STAMP, stamp_billable_usage_cost
 from litellm.litellm_core_utils.internal_call_metadata import without_budget_reservation
 from litellm.litellm_core_utils.litellm_logging import BilledRoundsSpend
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
@@ -586,7 +585,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)
 
                     if _chunk_type == openai_types.ResponsesAPIStreamEvents.RESPONSE_FAILED:
                         self._handle_logging_failed_response()
@@ -676,17 +675,10 @@ class BaseResponsesAPIStreamingIterator:
         headers: Final[Mapping[str, object]] = processed if isinstance(processed, Mapping) else EMPTY_MAPPING
         raw_headers: Final[Mapping[str, object]] = raw if isinstance(raw, Mapping) else EMPTY_MAPPING
         # rebuild by value and let existing keys win: sharing the source dicts would alias what the proxy
-        # splats into the client's HTTP headers, and copying non-header keys would carry response_cost.
-        # The provider's own usage.cost travels along, or the stamped bill would be priced as a provider report
-        stamp: Final[Mapping[str, object]] = (
-            {USAGE_COST_BEFORE_LITELLM_STAMP: source[USAGE_COST_BEFORE_LITELLM_STAMP]}
-            if USAGE_COST_BEFORE_LITELLM_STAMP in source
-            else EMPTY_MAPPING
-        )
+        # splats into the client's HTTP headers, and copying non-header keys would carry response_cost
         target.hidden_params = {
             "additional_headers": {**headers},
             "headers": {**raw_headers},
-            **stamp,
             **existing,
         }
 
@@ -1077,7 +1069,7 @@ class BaseResponsesAPIStreamingIterator:
                 "output": _partial_output_items(self._generated_content),
             }
         )
-        stamp_responses_usage_cost(partial, self.logging_obj)
+        normalize_responses_usage(partial)
         return ResponseIncompleteEvent(type=ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE, response=partial)
 
     def _note_yielded_event(self, event: ResponsesAPIStreamingResponse) -> None:
@@ -1749,20 +1741,14 @@ 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) -> None:
+    """A terminal response built off an untyped dict carries its usage as a dict, logging and the proxy read a model."""
+    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
-    try:
-        cost: Final[float | None] = logging_obj.response_cost_calculator(result=response_obj)
-    except Exception:
-        return
-    stamp_billable_usage_cost(response_obj, usage_obj, cost)
+    response_obj.usage = usage_obj  # rebind-ok: the typed usage has to ride on the response the client receives
 
 
 def opening_item_payload(payload: Mapping[str, object]) -> Mapping[str, object]:
@@ -1796,7 +1782,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)
 
     events: Final[list[ResponsesAPIStreamingResponse]] = [
         _build_response_status_event(openai_types.ResponsesAPIStreamEvents.RESPONSE_CREATED, transformed),

tests/unit/proxy/test_usage_cost_egress.py

@@ -0,0 +1,200 @@
+from __future__ import annotations
+
+from typing import Final
+
+import pytest
+
+import litellm
+from litellm.proxy.usage_cost_egress import with_client_stream_usage_cost, with_client_usage_cost
+from litellm.types.llms.openai import (
+    ResponseAPIUsage,
+    ResponseCompletedEvent,
+    ResponseIncompleteEvent,
+    ResponsesAPIResponse,
+    ResponsesAPIStreamEvents,
+)
+from litellm.types.utils import Choices, Message, ModelResponse, ModelResponseStream, StreamingChoices, Usage
+
+UPSTREAM_COST: Final = 0.037
+OPERATOR_BILL: Final = 100 * 2e-6 + 20 * 6e-6
+
+
+def _chat_reply(upstream_cost: float | None) -> ModelResponse:
+    reply: Final = ModelResponse(
+        id="chatcmpl-egress",
+        model="omni-model",
+        choices=[Choices(index=0, message=Message(role="assistant", content="hi"), finish_reason="stop")],
+        usage=Usage(prompt_tokens=100, completion_tokens=20, total_tokens=120, cost=upstream_cost),
+    )
+    reply._hidden_params["response_cost"] = OPERATOR_BILL
+    return reply
+
+
+def _responses_reply(upstream_cost: float | None) -> ResponsesAPIResponse:
+    return 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),
+    )
+
+
+def _usage_chunk(upstream_cost: float | None) -> ModelResponseStream:
+    return ModelResponseStream(
+        id="chatcmpl-egress",
+        model="omni-model",
+        choices=[],
+        usage=Usage(prompt_tokens=100, completion_tokens=20, total_tokens=120, cost=upstream_cost),
+    )
+
+
+def _bill(_reply: object) -> float | None:
+    return OPERATOR_BILL
+
+
+def _unpriceable(_reply: object) -> float | None:
+    return None
+
+
+@pytest.mark.parametrize("reply", [_chat_reply(UPSTREAM_COST), _responses_reply(UPSTREAM_COST)])
+def test_a_reply_whose_upstream_reported_a_cost_tells_the_client_what_litellm_billed(reply: object) -> None:
+    billed: Final = with_client_usage_cost(reply, OPERATOR_BILL)
+
+    assert billed is not reply, "the pipeline's own object keeps the upstream figure"
+    assert getattr(getattr(billed, "usage", None), "cost", None) == pytest.approx(OPERATOR_BILL)
+    assert getattr(getattr(reply, "usage", None), "cost", None) == UPSTREAM_COST
+
+
+def test_the_billed_copy_of_a_chat_reply_keeps_its_hidden_params() -> None:
+    reply: Final = _chat_reply(UPSTREAM_COST)
+
+    billed: Final = with_client_usage_cost(reply, OPERATOR_BILL)
+
+    assert isinstance(billed, ModelResponse)
+    assert billed._hidden_params["response_cost"] == pytest.approx(OPERATOR_BILL)
+    assert billed.model_dump()["usage"]["cost"] == pytest.approx(OPERATOR_BILL)
+
+
+@pytest.mark.parametrize("reply", [_chat_reply(None), _responses_reply(None)])
+def test_a_reply_without_an_upstream_cost_goes_out_as_the_upstream_sent_it(
+    monkeypatch: pytest.MonkeyPatch, reply: object
+) -> None:
+    monkeypatch.setattr(litellm, "include_cost_in_streaming_usage", False)
+
+    assert with_client_usage_cost(reply, OPERATOR_BILL) is reply
+
+
+@pytest.mark.parametrize("reply", [_chat_reply(None), _responses_reply(None)])
+def test_include_cost_in_streaming_usage_puts_the_bill_on_a_reply_without_an_upstream_cost(
+    monkeypatch: pytest.MonkeyPatch, reply: object
+) -> None:
+    monkeypatch.setattr(litellm, "include_cost_in_streaming_usage", True)
+
+    billed: Final = with_client_usage_cost(reply, OPERATOR_BILL)
+
+    assert getattr(getattr(billed, "usage", None), "cost", None) == pytest.approx(OPERATOR_BILL)
+
+
+def test_a_zero_bill_replaces_the_upstream_cost() -> None:
+    billed: Final = with_client_usage_cost(_chat_reply(UPSTREAM_COST), 0.0)
+
+    assert getattr(getattr(billed, "usage", None), "cost", None) == 0.0
+
+
+@pytest.mark.parametrize("bill", [None, -0.001, float("nan"), float("inf"), True, "0.1"])
+def test_an_unpriceable_reply_keeps_the_upstream_cost(bill: object) -> None:
+    reply: Final = _chat_reply(UPSTREAM_COST)
+
+    assert with_client_usage_cost(reply, bill) is reply
+    assert reply.usage is not None and reply.usage.cost == UPSTREAM_COST
+
+
+def test_a_reply_that_is_neither_chat_nor_responses_passes_through() -> None:
+    reply: Final = {"usage": {"cost": UPSTREAM_COST}}
+
+    assert with_client_usage_cost(reply, OPERATOR_BILL) is reply
+
+
+def test_the_streamed_usage_chunk_tells_the_client_what_litellm_billed() -> None:
+    chunk: Final = _usage_chunk(UPSTREAM_COST)
+
+    billed: Final = with_client_stream_usage_cost(chunk, _bill)
+
+    assert isinstance(billed, ModelResponseStream) and billed is not chunk
+    assert billed.usage is not None and billed.usage.cost == pytest.approx(OPERATOR_BILL)
+    assert chunk.usage is not None and chunk.usage.cost == UPSTREAM_COST
+
+
+def test_a_content_chunk_is_not_priced() -> None:
+    chunk: Final = ModelResponseStream(
+        id="chatcmpl-egress", model="omni-model", choices=[StreamingChoices(index=0, delta={"content": "hi"})]
+    )
+    priced: list[object] = []
+
+    def price(reply: object) -> float | None:
+        priced.append(reply)
+        return OPERATOR_BILL
+
+    assert with_client_stream_usage_cost(chunk, price) is chunk
+    assert priced == []
+
+
+def test_a_streamed_usage_chunk_without_an_upstream_cost_goes_out_as_sent(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(litellm, "include_cost_in_streaming_usage", False)
+    chunk: Final = _usage_chunk(None)
+
+    assert with_client_stream_usage_cost(chunk, _bill) is chunk
+
+
+def test_include_cost_in_streaming_usage_puts_the_bill_on_the_streamed_usage_chunk(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(litellm, "include_cost_in_streaming_usage", True)
+
+    billed: Final = with_client_stream_usage_cost(_usage_chunk(None), _bill)
+
+    assert isinstance(billed, ModelResponseStream)
+    assert billed.usage is not None and billed.usage.cost == pytest.approx(OPERATOR_BILL)
+
+
+def test_the_stream_pricer_sees_the_upstream_cost_so_the_bill_matches_the_spend_row() -> None:
+    seen: list[object] = []
+
+    def price(reply: object) -> float | None:
+        seen.append(getattr(getattr(reply, "usage", None), "cost", None))
+        return OPERATOR_BILL
+
+    with_client_stream_usage_cost(_usage_chunk(UPSTREAM_COST), price)
+
+    assert seen == [UPSTREAM_COST]
+
+
+@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(
+    monkeypatch: pytest.MonkeyPatch, upstream_cost: float | None, event_type: ResponsesAPIStreamEvents
+) -> None:
+    monkeypatch.setattr(litellm, "include_cost_in_streaming_usage", False)
+    event_cls: Final = (
+        ResponseCompletedEvent if event_type == ResponsesAPIStreamEvents.RESPONSE_COMPLETED else ResponseIncompleteEvent
+    )
+    event: Final = event_cls(type=event_type, response=_responses_reply(upstream_cost))
+
+    billed: Final = with_client_stream_usage_cost(event, _bill)
+
+    assert billed is not event
+    assert getattr(billed, "response").usage.cost == pytest.approx(OPERATOR_BILL)
+    assert event.response.usage is not None and event.response.usage.cost == upstream_cost
+
+
+def test_an_unpriceable_terminal_responses_event_goes_out_as_sent() -> None:
+    event: Final = ResponseCompletedEvent(
+        type=ResponsesAPIStreamEvents.RESPONSE_COMPLETED, response=_responses_reply(None)
+    )
+
+    assert with_client_stream_usage_cost(event, _unpriceable) is event

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
@@ -24,7 +27,8 @@ from litellm.exceptions import MidStreamFallbackError
 from litellm.integrations.custom_logger import CustomLogger
 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 +42,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 +81,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 +110,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:
@@ -489,7 +490,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):
@@ -838,51 +838,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_puts_the_bill_over_the_provider_reported_cost():
-    from litellm.litellm_core_utils.hidden_params import USAGE_COST_BEFORE_LITELLM_STAMP, provider_stated_usage_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)
-    logging_obj.response_cost_calculator.return_value = 0.00032
-
-    stamp_responses_usage_cost(response, logging_obj)
-
-    assert getattr(response.usage, "cost", None) == pytest.approx(0.00032)
-    assert provider_stated_usage_cost(response, response.usage.cost) == pytest.approx(0.5)
-    assert response._hidden_params[USAGE_COST_BEFORE_LITELLM_STAMP] == pytest.approx(0.5)
-
-
-def test_stamp_responses_usage_cost_stamps_a_zero_bill_only_over_a_provider_reported_cost():
-    from litellm.responses.streaming_iterator import stamp_responses_usage_cost
-
-    unpriced = _responses_api_response_with_usage()
-    reported = _responses_api_response_with_usage()
-    setattr(reported.usage, "cost", 0.5)
-    logging_obj = Mock(spec=LiteLLMLoggingObj)
-    logging_obj.response_cost_calculator.return_value = 0.0
-
-    stamp_responses_usage_cost(unpriced, logging_obj)
-    stamp_responses_usage_cost(reported, logging_obj)
-
-    assert getattr(unpriced.usage, "cost", None) is None
-    assert getattr(reported.usage, "cost", None) == 0.0
-
-
 def _unvalidated_response_with_dict_usage(usage: dict) -> ResponsesAPIResponse:
     return ResponsesAPIResponse.model_construct(
         id="resp_lit7391",
@@ -896,8 +851,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,
@@ -907,37 +863,20 @@ 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)
-    logging_obj.response_cost_calculator.return_value = 3e-05
 
-    stamp_responses_usage_cost(response, logging_obj)
+    normalize_responses_usage(response)
 
     assert isinstance(response.usage, ResponseAPIUsage)
     assert response.usage.cost == pytest.approx(3e-05)
     assert response.usage.output_tokens_details.reasoning_tokens == 117
 
 
-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)
 
     assert getattr(response.usage, "cost", None) is None
 
@@ -959,7 +898,7 @@ def _capture_dispatch(logged: list):
     return _dispatch
 
 
-def _headers_config(*, transform_hidden_params: Optional[dict] = None) -> Mock:
+def _headers_config(*, transform_hidden_params: dict | 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)
@@ -1274,10 +1213,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",
@@ -1319,8 +1258,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():
@@ -1390,6 +1328,7 @@ def _vcr_outcome_gate(request, vcr):
     yield
     record_vcr_outcome(request, vcr)
 
+
 @pytest.fixture(scope="function")
 def setup_and_teardown():
     """
@@ -1407,6 +1346,7 @@ def setup_and_teardown():
     loop.close()
     asyncio.set_event_loop(None)
 
+
 class _FakeLoggingObj:
     def __init__(self):
         self.success_calls = 0
@@ -1447,10 +1387,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,
@@ -1478,6 +1420,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):
@@ -1500,6 +1443,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):
@@ -1515,6 +1459,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(
@@ -1542,6 +1487,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(
@@ -1557,6 +1503,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):
@@ -1613,6 +1560,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):
@@ -1653,6 +1601,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():
@@ -1682,6 +1631,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(
@@ -1696,6 +1646,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()
@@ -1730,8 +1681,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):
     original_include_cost = litellm.include_cost_in_streaming_usage
     litellm.include_cost_in_streaming_usage = True
     openai_types = streaming_module._get_openai_response_types()
@@ -1780,9 +1732,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()
@@ -1818,6 +1771,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():
@@ -1850,6 +1804,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:
@@ -1868,6 +1823,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
@@ -1915,6 +1871,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()
@@ -1931,6 +1888,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():
@@ -1979,6 +1937,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()
@@ -2024,6 +1983,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}
@@ -2065,6 +2025,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:
@@ -2094,6 +2055,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",
@@ -2158,6 +2120,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
@@ -2216,6 +2179,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(
@@ -2266,6 +2230,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}
@@ -2311,6 +2276,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(
@@ -2370,6 +2336,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}
@@ -2468,7 +2435,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)
@@ -2524,7 +2493,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",
@@ -2658,7 +2626,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)
 
@@ -2800,39 +2770,3 @@ async def test_websocket_turns_logging_survives_garbage_collection(terminal_even
     await _run_websocket_turn(logger, terminal_event)
 
     assert await _logging_tasks_left_after_garbage_collection(logger) == 1
-
-
-def test_logging_copy_keeps_the_provider_usage_cost_behind_a_stamped_bill() -> None:
-    from litellm.litellm_core_utils.hidden_params import USAGE_COST_BEFORE_LITELLM_STAMP
-    from litellm.responses.streaming_iterator import _logging_copy
-
-    logging_obj = LiteLLMLoggingObj(
-        model="gpt-4o-mini",
-        messages=[{"role": "user", "content": "hi"}],
-        stream=True,
-        call_type="aresponses",
-        start_time=datetime.now(),
-        litellm_call_id="stamped-copy",
-        function_id="stamped-copy",
-    )
-    iterator = _make_iterator(sse_events=[], logging_obj=logging_obj)
-    billed = ResponsesAPIResponse(
-        id="resp_stamped",
-        created_at=1,
-        status="completed",
-        model="gpt-4o-mini",
-        object="response",
-        output=[],
-        usage={"input_tokens": 100, "output_tokens": 20, "total_tokens": 120, "cost": 0.63},
-    )
-    billed.hidden_params = {USAGE_COST_BEFORE_LITELLM_STAMP: 0.42, "response_cost": 0.63}
-    iterator.completed_response = ResponseCompletedEvent(
-        type=ResponsesAPIStreamEvents.RESPONSE_COMPLETED, response=billed
-    )
-    logging_response = _logging_copy(iterator.completed_response)
-
-    iterator._restore_provider_response_headers(logging_response)
-
-    copied = logging_response.response.hidden_params
-    assert copied[USAGE_COST_BEFORE_LITELLM_STAMP] == 0.42
-    assert "response_cost" not in copied

tests/unit/test_cost_calculator.py

@@ -1,48 +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_response_utils.response_metadata import update_response_metadata
-from litellm.responses.streaming_iterator import stamp_responses_usage_cost
+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):
@@ -518,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:
@@ -1615,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=[],
@@ -3373,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
@@ -6876,23 +6897,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():
@@ -6918,28 +6933,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(
@@ -6970,8 +6979,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)
@@ -6991,15 +6999,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)
@@ -7046,9 +7051,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)
 
@@ -7140,9 +7143,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,
@@ -7152,20 +7153,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,
@@ -7295,7 +7290,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"]
 
@@ -7308,8 +7303,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():
@@ -7320,9 +7315,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"
@@ -7339,13 +7332,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"]
 
@@ -7358,16 +7349,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")
 
 
@@ -7409,9 +7397,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=[
@@ -7445,9 +7431,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="")
@@ -7546,9 +7530,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
@@ -7556,12 +7538,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)
@@ -7653,9 +7632,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)
@@ -7745,11 +7722,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)
@@ -7790,9 +7765,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
 
 
@@ -8744,9 +8717,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
         ),
@@ -8785,27 +8756,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
 
@@ -8868,7 +8827,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",
@@ -8930,9 +8889,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",
@@ -8998,9 +8955,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
 
 
@@ -9021,7 +8976,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}),
@@ -9037,7 +8998,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},
         },
@@ -9151,7 +9116,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(
@@ -9285,16 +9255,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,
@@ -9312,9 +9279,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,
@@ -9322,9 +9287,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)
 
@@ -9333,9 +9296,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(
         {
@@ -9377,9 +9338,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)
@@ -9475,7 +9434,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():
@@ -9780,9 +9739,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",
@@ -9877,9 +9834,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(
@@ -10559,12 +10515,18 @@ def _reported_cost_stream_chunk(content: str | None, finish_reason: str | None,
         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)],
+        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:
+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:
@@ -10604,7 +10566,9 @@ def test_provider_reported_usage_cost_is_billed_on_stream_and_non_stream(
 
 
 @pytest.mark.parametrize("stream", [False, True])
-def test_deployment_pricing_beats_the_provider_reported_usage_cost(monkeypatch: pytest.MonkeyPatch, stream: bool) -> None:
+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(
@@ -10624,7 +10588,9 @@ def test_deployment_pricing_beats_the_provider_reported_usage_cost(monkeypatch:
 
 
 @pytest.mark.parametrize("stream", [False, True])
-def test_base_model_pricing_beats_the_provider_reported_usage_cost(monkeypatch: pytest.MonkeyPatch, stream: bool) -> None:
+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,
@@ -10643,34 +10609,21 @@ def test_provider_reported_usage_cost_takes_discount_then_margin(monkeypatch: py
     assert _bill_reported_cost(stream, {}, {}) == pytest.approx(0.037 * 0.8 * 1.5 + 0.001)
 
 
-def test_a_usage_cost_litellm_stamped_on_a_stream_is_not_billed_again_as_a_provider_report(
+def test_the_stream_builder_bills_the_operator_rates_and_leaves_usage_cost_as_the_upstream_reported_it(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    monkeypatch.setitem(litellm.model_cost, _REPORTED_COST_FIXTURE_MODEL, _REPORTED_COST_FIXTURE_RATES)
-    monkeypatch.setattr(litellm, "cost_margin_config", {"global": {"fixed_amount": 0.002}})
-    logging_obj: Final = _reported_cost_logging(True, {})
-    token_math_plus_margin: Final = 100 * 1e-6 + 20 * 3e-6 + 0.002
-
-    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(prompt_tokens=100, completion_tokens=20, total_tokens=120)),
-        ],
-        messages=[{"role": "user", "content": "hi"}],
-        logging_obj=logging_obj,
+    """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},
     )
-
-    assert isinstance(response, ModelResponse)
-    assert response.usage is not None and response.usage.cost == pytest.approx(token_math_plus_margin)
-    assert logging_obj.response_cost_calculator(result=response, ignore_stamped_cost=True) == pytest.approx(
-        token_math_plus_margin
+    logging_obj: Final = _reported_cost_logging(
+        True, {"input_cost_per_token": 2e-6, "output_cost_per_token": 6e-6, "model_info": {"id": deployment_id}}
     )
-
-
-def test_a_zero_bill_is_stamped_over_the_provider_reported_usage_cost_on_a_stream(monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setattr(litellm, "cost_discount_config", {"openai": 1.0})
-    logging_obj: Final = _reported_cost_logging(True, {})
+    operator_bill: Final = 100 * 2e-6 + 20 * 6e-6
 
     response: Final[object] = litellm.stream_chunk_builder(
         [
@@ -10683,195 +10636,11 @@ def test_a_zero_bill_is_stamped_over_the_provider_reported_usage_cost_on_a_strea
     )
 
     assert isinstance(response, ModelResponse)
-    assert response.usage is not None and response.usage.cost == 0.0
-    assert logging_obj.response_cost_calculator(result=response) == 0.0
-
-
-_PRICED_DEPLOYMENT_ID: Final = "reported-cost-deployment-priced-by-the-operator"
-_PRICED_LITELLM_PARAMS: Final[Mapping[str, object]] = MappingProxyType(
-    {
-        "input_cost_per_token": 2e-6,
-        "output_cost_per_token": 6e-6,
-        "metadata": {"model_info": {"id": _PRICED_DEPLOYMENT_ID}},
-    }
-)
-
-
-def _reported_cost_usage() -> Usage:
-    return Usage(prompt_tokens=100, completion_tokens=20, total_tokens=120, cost=0.037)
-
-
-def _priced_logging(stream: bool, call_type: str) -> Logging:
-    """The logging object the proxy hands a priced deployment's call, the deployment under metadata.model_info."""
-    logging_obj: Final = _reported_cost_logging(stream, _PRICED_LITELLM_PARAMS)
-    logging_obj.call_type = call_type
-    return logging_obj
-
-
-def _reported_cost_responses_response() -> ResponsesAPIResponse:
-    return ResponsesAPIResponse(
-        id="resp_reported",
-        created_at=1,
-        status="completed",
-        model="reported-cost-fixture",
-        object="response",
-        output=[],
-        usage=ResponseAPIUsage(input_tokens=100, output_tokens=20, total_tokens=120, cost=0.037),
+    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
     )
-
-
-def _client_visible_reply(route: str) -> tuple[object, Logging]:
-    """The object a client of the given route reads usage.cost from, after litellm priced it."""
-    match route:
-        case "chat":
-            logging_obj: Logging = _priced_logging(False, "completion")
-            response: object = ModelResponse(
-                id="chatcmpl-reported",
-                model="reported-cost-fixture",
-                choices=[Choices(index=0, message=Message(role="assistant", content="hi"), finish_reason="stop")],
-                usage=_reported_cost_usage(),
-            )
-            update_response_metadata(
-                result=response,
-                logging_obj=logging_obj,
-                model=_REPORTED_COST_FIXTURE_MODEL,
-                kwargs={},
-                start_time=datetime.datetime(2026, 1, 1),
-                end_time=datetime.datetime(2026, 1, 1, 0, 0, 1),
-            )
-        case "chat-stream":
-            logging_obj = _priced_logging(True, "completion")
-            response = litellm.stream_chunk_builder(
-                [
-                    _reported_cost_stream_chunk("hi", None, None),
-                    _reported_cost_stream_chunk(None, "stop", None),
-                    _reported_cost_stream_chunk(None, None, _reported_cost_usage()),
-                ],
-                messages=[{"role": "user", "content": "hi"}],
-                logging_obj=logging_obj,
-            )
-        case "responses":
-            logging_obj = _priced_logging(False, "aresponses")
-            response = _reported_cost_responses_response()
-            update_response_metadata(
-                result=response,
-                logging_obj=logging_obj,
-                model=_REPORTED_COST_FIXTURE_MODEL,
-                kwargs={},
-                start_time=datetime.datetime(2026, 1, 1),
-                end_time=datetime.datetime(2026, 1, 1, 0, 0, 1),
-            )
-        case _:
-            logging_obj = _priced_logging(True, "aresponses")
-            response = _reported_cost_responses_response()
-            stamp_responses_usage_cost(response, logging_obj)
-    return response, logging_obj
-
-
-@pytest.mark.parametrize("route", ["chat", "chat-stream", "responses", "responses-stream"])
-def test_the_client_reads_what_litellm_billed_from_usage_cost_on_every_route(
-    monkeypatch: pytest.MonkeyPatch, route: str
-) -> None:
-    """A priced deployment showed its bill on chat streams only; the other three routes still showed the upstream's number."""
-    monkeypatch.setitem(
-        litellm.model_cost,
-        _PRICED_DEPLOYMENT_ID,
-        {"litellm_provider": "openai", "mode": "chat", "input_cost_per_token": 2e-6, "output_cost_per_token": 6e-6},
-    )
-    bill: Final = 100 * 2e-6 + 20 * 6e-6
-
-    response, logging_obj = _client_visible_reply(route)
-
-    usage: Final = getattr(response, "usage", None)
-    assert usage is not None and usage.cost == pytest.approx(bill)
-    assert logging_obj.response_cost_calculator(result=response, ignore_stamped_cost=True) == pytest.approx(bill)
-
-
-def test_a_plain_reply_without_an_upstream_usage_cost_keeps_none_after_pricing(monkeypatch: pytest.MonkeyPatch) -> None:
-    """Clients comparing the usage dict to the upstream's byte for byte must not find a cost litellm added."""
-    monkeypatch.setitem(
-        litellm.model_cost,
-        _PRICED_DEPLOYMENT_ID,
-        {"litellm_provider": "openai", "mode": "chat", "input_cost_per_token": 2e-6, "output_cost_per_token": 6e-6},
-    )
-    response: Final = ModelResponse(
-        id="chatcmpl-unreported",
-        model="reported-cost-fixture",
-        choices=[Choices(index=0, message=Message(role="assistant", content="hi"), finish_reason="stop")],
-        usage=Usage(prompt_tokens=100, completion_tokens=20, total_tokens=120),
-    )
-
-    update_response_metadata(
-        result=response,
-        logging_obj=_priced_logging(False, "completion"),
-        model=_REPORTED_COST_FIXTURE_MODEL,
-        kwargs={},
-        start_time=datetime.datetime(2026, 1, 1),
-        end_time=datetime.datetime(2026, 1, 1, 0, 0, 1),
-    )
-
-    assert response.usage is not None and getattr(response.usage, "cost", None) is None
-    assert response._hidden_params["response_cost"] == pytest.approx(100 * 2e-6 + 20 * 6e-6)
-
-
-def test_include_cost_in_streaming_usage_puts_the_bill_on_a_plain_reply_without_an_upstream_usage_cost(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    """The switch that injects the bill into streamed usage gives plain replies the same key."""
-    monkeypatch.setattr(litellm, "include_cost_in_streaming_usage", True)
-    monkeypatch.setitem(
-        litellm.model_cost,
-        _PRICED_DEPLOYMENT_ID,
-        {"litellm_provider": "openai", "mode": "chat", "input_cost_per_token": 2e-6, "output_cost_per_token": 6e-6},
-    )
-    response: Final = ModelResponse(
-        id="chatcmpl-unreported",
-        model="reported-cost-fixture",
-        choices=[Choices(index=0, message=Message(role="assistant", content="hi"), finish_reason="stop")],
-        usage=Usage(prompt_tokens=100, completion_tokens=20, total_tokens=120),
-    )
-    logging_obj: Final = _priced_logging(False, "completion")
-
-    update_response_metadata(
-        result=response,
-        logging_obj=logging_obj,
-        model=_REPORTED_COST_FIXTURE_MODEL,
-        kwargs={},
-        start_time=datetime.datetime(2026, 1, 1),
-        end_time=datetime.datetime(2026, 1, 1, 0, 0, 1),
-    )
-
-    bill: Final = 100 * 2e-6 + 20 * 6e-6
-    assert response.usage is not None and response.usage.cost == pytest.approx(bill)
-    assert logging_obj.response_cost_calculator(result=response, ignore_stamped_cost=True) == pytest.approx(bill)
-
-
-def test_a_bill_stamped_on_a_plain_reply_survives_the_metadata_pass_for_a_second_pricing(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    """The metadata pass installs a fresh hidden-params dict on a reply that arrived with none, which dropped the
-    provider's figure the stamp keeps and let a second pricing pass apply the margin twice."""
-    monkeypatch.setattr(litellm, "cost_margin_config", {"global": {"percentage": 0.5}})
-    monkeypatch.delitem(litellm.model_cost, _REPORTED_COST_FIXTURE_MODEL, raising=False)
-    logging_obj: Final = _reported_cost_logging(False, {})
-    response: Final = ModelResponse(
-        id="chatcmpl-reported",
-        model="reported-cost-fixture",
-        choices=[Choices(index=0, message=Message(role="assistant", content="hi"), finish_reason="stop")],
-        usage=_reported_cost_usage(),
-    )
-
-    update_response_metadata(
-        result=response,
-        logging_obj=logging_obj,
-        model=_REPORTED_COST_FIXTURE_MODEL,
-        kwargs={},
-        start_time=datetime.datetime(2026, 1, 1),
-        end_time=datetime.datetime(2026, 1, 1, 0, 0, 1),
-    )
-
-    assert response.usage is not None and response.usage.cost == pytest.approx(0.037 * 1.5)
-    assert logging_obj.response_cost_calculator(result=response, ignore_stamped_cost=True) == pytest.approx(0.037 * 1.5)
+    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(