From 9b7f2c8dead2a2ea5a0b6b4afbcf065d33feb8bd Mon Sep 17 00:00:00 2001 From: Kerry Lu Date: Sat, 24 Oct 2026 16:44:29 -0700 Subject: [PATCH] fix(opencode): bill OpenCode Go requests from the reported list_cost (#48254) * fix(opencode): bill OpenCode Go requests from the reported list_cost OpenCode returns usage.opencode.list_cost on every response. Read it into the provider-reported cost the cost calculator checks before the price map, for chat and Responses, streamed and not. Keep the static rows as the fallback, and stop the generic usage.cost path from billing Go requests at 0 (usage.cost is what hit the prepaid balance, 0 inside the plan). Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(core): resolve reported cost breakdowns without recursion Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(opencode): cover reported list_cost spend through the proxy Replaces the recursive _reported_cost_value with a flat _cost_number helper so the code-quality gate passes, and adds an integration test that runs the proxy against a scripted OpenCode upstream for a non-stream and a stream chat completion, asserting the response cost header and the spend log row Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(opencode): bill native /v1/messages from the reported list_cost The native Anthropic Messages route priced OpenCode Go bodies from the price rows twice over: the non-stream body is a TypedDict the response-cost header path priced as-is, and the stream assembled its ModelResponse with the rows cost already stamped. Both now read usage.opencode.list_cost, the non-stream body through billable_result and the stream from the final message_delta. The integration test now covers chat, responses and messages, stream and non-stream, through the proxy against a scripted OpenCode upstream Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(opencode): keep the stream cost stamp inside the lint gate Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(opencode): read list_cost off dict chunks and trust it only from OpenCode on /v1/messages Stream builders receive chunks both as mappings and as ModelResponseStream objects, so the list_cost scan now reads usage through chunk_usage. The native Messages route only trusts a reported cost from opencode_go, every other upstream keeps the price map, which also keeps a zero usage.cost from being billed as free. Served bodies without stop_sequence still price, and the new code stays inside the basedpyright gate Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/litellm_core_utils/core_helpers.py | 65 +++ litellm/litellm_core_utils/litellm_logging.py | 27 +- .../streaming_chunk_builder_utils.py | 40 +- .../litellm_core_utils/streaming_handler.py | 23 +- litellm/llms/opencode/chat/transformation.py | 57 ++- litellm/llms/opencode/common_utils.py | 11 + .../llms/opencode/responses/transformation.py | 40 +- .../anthropic_passthrough_logging_handler.py | 56 ++- .../test_opencode_go_reported_cost_wire.py | 198 ++++++++ .../test_litellm_logging.py | 276 +++++++----- .../llms/opencode/chat/test_transformation.py | 180 ++++++++ ...t_anthropic_passthrough_logging_handler.py | 422 ++++++++---------- 12 files changed, 1024 insertions(+), 371 deletions(-) create mode 100644 tests/integration/pricing/test_opencode_go_reported_cost_wire.py diff --git a/litellm/litellm_core_utils/core_helpers.py b/litellm/litellm_core_utils/core_helpers.py index 4be5cabac04..00c6ce97e78 100644 --- a/litellm/litellm_core_utils/core_helpers.py +++ b/litellm/litellm_core_utils/core_helpers.py @@ -928,6 +928,71 @@ class _CarriesHiddenParams(Protocol): def hidden_params(self, hidden_params: dict[str, object]) -> None: ... # mutable-ok: API requires mutation +_USAGE_BLOCK_ADAPTER: Final = TypeAdapter(Mapping[str, object]) + + +def _usage_field(usage: object, name: str) -> object: + if isinstance(usage, Mapping): + return _USAGE_BLOCK_ADAPTER.validate_python(usage).get(name) + return getattr(usage, name, None) + + +def _cost_number(value: object) -> float | None: + if isinstance(value, bool) or not isinstance(value, int | float): + return None + return float(value) + + +def _reported_cost_value(value: object) -> float | None: + """A reported cost is a number, or a breakdown whose total lives under ``total_cost``.""" + raw: Final[object] = value + if isinstance(value, Mapping): + return _cost_number(_usage_field(raw, "total_cost")) + return _cost_number(value) + + +def response_cost_in_hidden_params(response: _CarriesHiddenParams) -> float | None: + additional_headers: Final[object] = response.hidden_params.get("additional_headers") + if not isinstance(additional_headers, Mapping): + return None + return _cost_number(_USAGE_BLOCK_ADAPTER.validate_python(additional_headers).get(RESPONSE_COST_HEADER)) + + +def override_stamped_response_cost(response: _CarriesHiddenParams, cost: float) -> None: + """Replace the cost an earlier builder stamped so every reader of the hidden params agrees with the header.""" + set_response_cost_in_hidden_params(response, cost) + hidden_params: Final = response.hidden_params + hidden_params["response_cost"] = cost + + +def chunk_usage(chunk: object) -> object: + """Stream chunks reach the builders both as mappings and as ModelResponseStream objects.""" + if isinstance(chunk, Mapping): + return _USAGE_BLOCK_ADAPTER.validate_python(chunk).get("usage") + return getattr(chunk, "usage", None) + + +_MESSAGES_REPORTED_COST_PROVIDERS: Final = frozenset({"opencode_go"}) + + +def native_messages_reported_cost(usage: object, custom_llm_provider: object) -> float | None: + """Only a Messages upstream known to price its own responses is trusted, every other one keeps the price map.""" + if not isinstance(custom_llm_provider, str) or custom_llm_provider not in _MESSAGES_REPORTED_COST_PROVIDERS: + return None + return provider_reported_usage_cost(usage, custom_llm_provider) + + +def provider_reported_usage_cost(usage: object, custom_llm_provider: str | None) -> float | None: + """The request cost a provider states on its usage block, as a number or a breakdown with ``total_cost``. + + OpenCode Go reports the model's list price (peak hours and long-context tiers included) under + ``usage.opencode.list_cost``, while its ``usage.cost`` is what hit the prepaid balance, 0 inside the plan. + """ + if custom_llm_provider == "opencode_go": + return _reported_cost_value(_usage_field(_usage_field(usage, "opencode"), "list_cost")) + return _reported_cost_value(_usage_field(usage, "cost")) + + def set_response_cost_in_hidden_params(response: _CarriesHiddenParams, cost: float | None) -> None: """Record a provider-reported cost where the cost calculator looks before the price map.""" if cost is None: diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index fc8d1ecc8c1..9e0461e951d 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -77,6 +77,8 @@ from litellm.litellm_core_utils.classifier_logging import ( from litellm.litellm_core_utils.core_helpers import ( get_provider_response_headers_from_hidden_params, is_expected_client_error, + native_messages_reported_cost, + provider_reported_usage_cost, proxy_stamped_kubernetes_pod_routing, proxy_stamped_used_client_oauth_token, reconstruct_model_name, @@ -4756,7 +4758,12 @@ class Logging(LiteLLMLoggingBaseClass): ): ## return unified Usage object if isinstance(result.response.usage, ResponseAPIUsage): - set_response_cost_in_hidden_params(result.response, result.response.usage.cost) + set_response_cost_in_hidden_params( + result.response, + provider_reported_usage_cost( + result.response.usage, self.model_call_details.get("custom_llm_provider") + ), + ) transformed_usage: Final = ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage( result.response.usage ) @@ -4822,8 +4829,11 @@ class Logging(LiteLLMLoggingBaseClass): bridge is replaced by the ModelResponse its spend row is built from, anything else is priced as is.""" served: Final[object] = result - if self.bridged_model_response is None or not isinstance(result, dict): - return result + call_type: Final[object] = getattr(self, "call_type", None) + if not isinstance(result, dict): + return served + if self.bridged_model_response is None and call_type != CallTypes.anthropic_messages.value: + return served return self._anthropic_messages_logged_response(result=served) def _anthropic_messages_logged_response(self, result: object) -> ModelResponse: @@ -4866,7 +4876,13 @@ class Logging(LiteLLMLoggingBaseClass): if isinstance(result, ResponsesAPIResponse): return self._translate_responses_api_response_to_model_response(result) - provider_response_id: Final = _provider_response_id(result) + raw_body: Final[object] = result + served_body: Final = _LITELLM_PARAMS.validate_python(raw_body) if isinstance(result, dict) else None + reported_cost: Final = native_messages_reported_cost( + served_body.get("usage") if served_body is not None else None, + self.model_call_details.get("custom_llm_provider"), + ) + provider_response_id: Final = _provider_response_id(raw_body) bridged: Final = self.bridged_model_response if bridged is not None and isinstance(result, dict): return bridged if provider_response_id is None else bridged.model_copy(update={"id": provider_response_id}) @@ -4891,7 +4907,7 @@ class Logging(LiteLLMLoggingBaseClass): else: from litellm.types.llms.anthropic import AnthropicResponse - pydantic_result: Final = AnthropicResponse.model_validate(result) + pydantic_result: Final = AnthropicResponse.model_validate({"stop_sequence": None, **(served_body or {})}) import httpx result = litellm.AnthropicConfig().transform_parsed_response( @@ -4904,6 +4920,7 @@ class Logging(LiteLLMLoggingBaseClass): json_mode=None, speed=self.optional_params.get("speed") if self.optional_params else None, ) + set_response_cost_in_hidden_params(result, reported_cost) return result def _translate_responses_api_response_to_model_response(self, result: ResponsesAPIResponse) -> ModelResponse: diff --git a/litellm/litellm_core_utils/streaming_chunk_builder_utils.py b/litellm/litellm_core_utils/streaming_chunk_builder_utils.py index 50789406055..68cdc204af4 100644 --- a/litellm/litellm_core_utils/streaming_chunk_builder_utils.py +++ b/litellm/litellm_core_utils/streaming_chunk_builder_utils.py @@ -12,6 +12,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 from litellm.litellm_core_utils.prompt_templates.common_utils import ( get_content_from_model_response, # pyright: ignore[reportUnknownVariableType] # response param accepts an untyped dict @@ -1352,14 +1353,27 @@ def stream_chunk_builder_text_completion( _CALCULATOR_PRICED_REPORTED_COST_PROVIDERS: Final = frozenset({LlmProviders.XAI.value}) -def _reported_cost_is_priced_by_calculator(logging_obj: Optional["Logging"]) -> bool: +def _logged_provider(logging_obj: Optional["Logging"]) -> object: if logging_obj is None: - return False + return None provider: Final[object] = logging_obj.model_call_details.get("custom_llm_provider") - return provider in _CALCULATOR_PRICED_REPORTED_COST_PROVIDERS + return provider -def _stream_builder_response_cost(response: ModelResponse, logging_obj: Optional["Logging"]) -> float | None: +def _reported_cost_is_priced_by_calculator(logging_obj: Optional["Logging"]) -> bool: + return _logged_provider(logging_obj) in _CALCULATOR_PRICED_REPORTED_COST_PROVIDERS + + +def _stream_builder_response_cost( + response: ModelResponse, logging_obj: Optional["Logging"], chunks: Sequence[object] +) -> float | None: + if _logged_provider(logging_obj) == LlmProviders.OPENCODE_GO.value: + # usage.opencode only survives on the chunks, the assembled Usage keeps declared fields + reported: Final = ( + provider_reported_usage_cost(chunk_usage(chunk), LlmProviders.OPENCODE_GO.value) + for chunk in reversed(chunks) + ) + return next((cost for cost in reported if cost is not None), None) usage_cost: Final = getattr(getattr(response, "usage", None), "cost", None) if isinstance(usage_cost, (int, float)) and not _reported_cost_is_priced_by_calculator(logging_obj): return float(usage_cost) @@ -1397,8 +1411,10 @@ def _set_hidden_response_cost(response: ModelResponse, response_cost: float) -> hidden_params["response_cost"] = response_cost -def _set_stream_builder_response_cost(response: ModelResponse, logging_obj: Optional["Logging"]) -> None: - response_cost: Final = _stream_builder_response_cost(response, logging_obj) +def _set_stream_builder_response_cost( + response: ModelResponse, logging_obj: Optional["Logging"], chunks: Sequence[object] +) -> None: + response_cost: Final = _stream_builder_response_cost(response, logging_obj, chunks) if response_cost is None: return _set_hidden_response_cost(response, response_cost) @@ -1550,7 +1566,11 @@ def stream_chunk_builder( break _stamp_streaming_usage_cost(usage, response, logging_obj) - _set_stream_builder_response_cost(response, logging_obj) + _set_stream_builder_response_cost( + response, + logging_obj, + chunks, # pyright: ignore[reportUnknownArgumentType] # legacy chunk list is untyped + ) processor.apply_provider_assembled_streaming_metadata(response, chunks, logging_obj) return response @@ -1734,7 +1754,11 @@ def stream_chunk_builder( break _stamp_streaming_usage_cost(usage, response, logging_obj) - _set_stream_builder_response_cost(response, logging_obj) + _set_stream_builder_response_cost( + response, + logging_obj, + chunks, # pyright: ignore[reportUnknownArgumentType] # legacy chunk list is untyped + ) processor.apply_provider_assembled_streaming_metadata(response, chunks, logging_obj) return response diff --git a/litellm/litellm_core_utils/streaming_handler.py b/litellm/litellm_core_utils/streaming_handler.py index fdcdca151be..2df470edc59 100644 --- a/litellm/litellm_core_utils/streaming_handler.py +++ b/litellm/litellm_core_utils/streaming_handler.py @@ -24,6 +24,7 @@ import litellm from litellm import verbose_logger from litellm._uuid import uuid from litellm.litellm_core_utils.asyncify import asyncify +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, get_hidden_params, @@ -72,7 +73,9 @@ _SYNC_ITER_EXHAUSTED: Final = object() _GCHUNK_FIELDS: Final[frozenset] = frozenset(GChunk.__annotations__) _CompleteStreamingResponse = ModelResponse | TextCompletionResponse | ModelResponseStream | None -_USAGE_COST_HEADER_PROVIDERS: Final[frozenset[str]] = frozenset({LlmProviders.OPENROUTER.value}) +_USAGE_COST_HEADER_PROVIDERS: Final[frozenset[str]] = frozenset( + {LlmProviders.OPENROUTER.value, LlmProviders.OPENCODE_GO.value} +) def _next_sync_or_exhausted(it: Iterator[object]) -> object: @@ -1953,11 +1956,17 @@ class CustomStreamWrapper: def _propagate_usage_cost_to_hidden_params( response: "ModelResponse", custom_llm_provider: str | None, + chunks: Sequence[object] = (), ) -> None: + """The builder's usage only keeps declared fields, so provider blocks like usage.opencode are read off the chunks.""" if custom_llm_provider not in _USAGE_COST_HEADER_PROVIDERS: return - _usage: Final[Usage | None] = getattr(response, "usage", None) - _cost: Final = CustomStreamWrapper._resolve_provider_reported_cost(getattr(_usage, "cost", None)) + usages: Final = ( + getattr(response, "usage", None), + *(chunk_usage(chunk) for chunk in reversed(chunks)), + ) + costs: Final = (provider_reported_usage_cost(usage, custom_llm_provider) for usage in usages) + _cost: Final = next((cost for cost in costs if cost is not None), None) if _cost is not None: if "additional_headers" not in response.hidden_params: response.hidden_params["additional_headers"] = {} @@ -2099,7 +2108,9 @@ class CustomStreamWrapper: complete_streaming_response: Final = self._build_complete_streaming_response() response = self._new_usage_chunk() if complete_streaming_response is not None: - self._propagate_usage_cost_to_hidden_params(complete_streaming_response, self.custom_llm_provider) + self._propagate_usage_cost_to_hidden_params( + complete_streaming_response, self.custom_llm_provider, self.chunks + ) setattr( response, @@ -2318,7 +2329,9 @@ class CustomStreamWrapper: complete_streaming_response: Final = await self._async_build_complete_streaming_response() response: Final = self._new_usage_chunk() if complete_streaming_response is not None: - self._propagate_usage_cost_to_hidden_params(complete_streaming_response, self.custom_llm_provider) + self._propagate_usage_cost_to_hidden_params( + complete_streaming_response, self.custom_llm_provider, self.chunks + ) setattr( response, diff --git a/litellm/llms/opencode/chat/transformation.py b/litellm/llms/opencode/chat/transformation.py index 40b99dbdd76..dcb30b5835c 100644 --- a/litellm/llms/opencode/chat/transformation.py +++ b/litellm/llms/opencode/chat/transformation.py @@ -1,6 +1,7 @@ from collections.abc import AsyncIterator, Iterator, Mapping, Sequence from typing import TYPE_CHECKING, Final +import httpx from pydantic import TypeAdapter import litellm @@ -9,7 +10,14 @@ from litellm.llms.anthropic.chat.transformation import AnthropicConfig from litellm.llms.base_llm.chat.transformation import BaseConfig from litellm.llms.gemini.chat.transformation import GoogleAIStudioGeminiConfig from litellm.llms.openai_like.chat.transformation import OpenAILikeChatConfig -from litellm.llms.opencode.common_utils import go_api_base, go_api_key, go_endpoint, go_headers, go_model_info +from litellm.llms.opencode.common_utils import ( + attach_go_reported_cost, + go_api_base, + go_api_key, + go_endpoint, + go_headers, + go_model_info, +) from litellm.secret_managers.main import get_secret_str from litellm.types.llms.openai import AllMessageValues from litellm.types.utils import ModelResponse @@ -18,7 +26,46 @@ if TYPE_CHECKING: from litellm.litellm_core_utils.litellm_logging import Logging -class OpenCodeGoChatConfig(OpenAILikeChatConfig): +class _GoReportedCostMixin: + """Record usage.opencode.list_cost so response_cost reflects OpenCode's price, not the static rows.""" + + go_usage_key: str = "usage" + + def transform_response( + self, + model: str, + raw_response: httpx.Response, + model_response: ModelResponse, + logging_obj: "Logging", + request_data: Mapping[str, object], + messages: Sequence[AllMessageValues], + optional_params: Mapping[str, object], + litellm_params: Mapping[str, object], + encoding: object, + api_key: str | None = None, + json_mode: bool | None = None, + ) -> ModelResponse: + response: Final = super().transform_response( # pyright: ignore[reportAttributeAccessIssue, reportUnknownMemberType, reportUnknownVariableType] # mixin precedes a BaseConfig subclass + model=model, + raw_response=raw_response, + model_response=model_response, + logging_obj=logging_obj, + request_data=dict(request_data), + messages=list(messages), + optional_params=dict(optional_params), + litellm_params=dict(litellm_params), + encoding=encoding, + api_key=api_key, + json_mode=json_mode, + ) + priced: Final = TypeAdapter(ModelResponse).validate_python(response) + attach_go_reported_cost( + priced, TypeAdapter(dict[str, object]).validate_json(raw_response.content).get(self.go_usage_key) + ) + return priced + + +class OpenCodeGoChatConfig(_GoReportedCostMixin, OpenAILikeChatConfig): @property def custom_llm_provider(self) -> str: return "opencode_go" @@ -60,7 +107,7 @@ class OpenCodeGoChatConfig(OpenAILikeChatConfig): return [f"opencode_go/{entry['id']}" for entry in models if isinstance(entry.get("id"), str)] -class OpenCodeGoMessagesChatConfig(AnthropicConfig): +class OpenCodeGoMessagesChatConfig(_GoReportedCostMixin, AnthropicConfig): @property def custom_llm_provider(self) -> str: return "opencode_go" @@ -110,7 +157,9 @@ class OpenCodeGoMessagesChatConfig(AnthropicConfig): return TypeAdapter(dict[str, object]).validate_python(raw) -class OpenCodeGoGeminiChatConfig(GoogleAIStudioGeminiConfig): +class OpenCodeGoGeminiChatConfig(_GoReportedCostMixin, GoogleAIStudioGeminiConfig): + go_usage_key: str = "usageMetadata" + @property def supports_stream_param_in_request_body(self) -> bool: return False diff --git a/litellm/llms/opencode/common_utils.py b/litellm/llms/opencode/common_utils.py index 46e382d8e65..82cd957d23c 100644 --- a/litellm/llms/opencode/common_utils.py +++ b/litellm/llms/opencode/common_utils.py @@ -4,8 +4,11 @@ from typing import Final from pydantic import TypeAdapter +from litellm.litellm_core_utils.core_helpers import provider_reported_usage_cost, set_response_cost_in_hidden_params from litellm.litellm_core_utils.provider_affinity import add_provider_affinity_header from litellm.secret_managers.main import get_secret_str +from litellm.types.llms.openai import ResponsesAPIResponse +from litellm.types.utils import ModelResponse def is_opencode_api_base(api_base: str | None) -> bool: @@ -67,3 +70,11 @@ def go_headers( params, "x-opencode-session", ) + + +def go_reported_cost(usage: object) -> float | None: + return provider_reported_usage_cost(usage, "opencode_go") + + +def attach_go_reported_cost(response: ModelResponse | ResponsesAPIResponse, usage: object) -> None: + set_response_cost_in_hidden_params(response, go_reported_cost(usage)) diff --git a/litellm/llms/opencode/responses/transformation.py b/litellm/llms/opencode/responses/transformation.py index ca75a319c7e..615ca60b45b 100644 --- a/litellm/llms/opencode/responses/transformation.py +++ b/litellm/llms/opencode/responses/transformation.py @@ -1,11 +1,24 @@ from collections.abc import Mapping +from typing import TYPE_CHECKING, Final + +import httpx from litellm.litellm_core_utils.url_utils import ensure_api_route from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig -from litellm.llms.opencode.common_utils import go_api_base, go_headers +from litellm.llms.opencode.common_utils import attach_go_reported_cost, go_api_base, go_headers +from litellm.types.llms.openai import ( + ResponseCompletedEvent, + ResponseFailedEvent, + ResponseIncompleteEvent, + ResponsesAPIResponse, + ResponsesAPIStreamingResponse, +) from litellm.types.router import GenericLiteLLMParams from litellm.types.utils import LlmProviders +if TYPE_CHECKING: + from litellm.litellm_core_utils.litellm_logging import Logging + class OpenCodeGoResponsesConfig(OpenAIResponsesAPIConfig): @property @@ -25,3 +38,28 @@ class OpenCodeGoResponsesConfig(OpenAIResponsesAPIConfig): litellm_params: GenericLiteLLMParams | None, ) -> dict[str, object]: # mutable-ok: protocol handlers mutate the returned request and headers return go_headers(headers, litellm_params, litellm_params.api_key if litellm_params else None, model) + + def transform_response_api_response( + self, + model: str, + raw_response: httpx.Response, + logging_obj: "Logging", + ) -> ResponsesAPIResponse: + response: Final = super().transform_response_api_response( + model=model, raw_response=raw_response, logging_obj=logging_obj + ) + attach_go_reported_cost(response, response.usage) + return response + + def transform_streaming_response( + self, + model: str, + parsed_chunk: Mapping[str, object], + logging_obj: "Logging", + ) -> ResponsesAPIStreamingResponse: + event: Final = super().transform_streaming_response( # pyright: ignore[reportUnknownMemberType] # base takes an untyped dict + model=model, parsed_chunk=dict(parsed_chunk), logging_obj=logging_obj + ) + if isinstance(event, ResponseCompletedEvent | ResponseFailedEvent | ResponseIncompleteEvent): + attach_go_reported_cost(event.response, event.response.usage) + return event diff --git a/litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py b/litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py index 23ef89b3c5c..15f70073c7a 100644 --- a/litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py +++ b/litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py @@ -1,15 +1,20 @@ import asyncio import json -from collections.abc import Mapping, Sequence +from collections.abc import Iterator, Mapping, Sequence from datetime import datetime from typing import TYPE_CHECKING, Any, Final, cast import httpx -from pydantic import ConfigDict, TypeAdapter +from pydantic import ConfigDict, TypeAdapter, ValidationError import litellm from litellm._logging import verbose_proxy_logger from litellm.constants import ANTHROPIC_BATCHES_ROUTE +from litellm.litellm_core_utils.core_helpers import ( + native_messages_reported_cost, + override_stamped_response_cost, + response_cost_in_hidden_params, +) from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.litellm_core_utils.litellm_logging import use_custom_pricing_for_model from litellm.litellm_core_utils.prompt_templates.common_utils import ( @@ -53,6 +58,38 @@ else: _JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True)) +def _event_payload(event: str) -> Mapping[str, object] | None: + data_lines: Final = tuple( + line.strip()[len("data:") :].strip() for line in event.splitlines() if line.strip().startswith("data:") + ) + if not data_lines or data_lines[0] in ("", "[DONE]"): + return None + try: + return _JSON_OBJECT.validate_json(data_lines[0]) + except ValidationError: + return None + + +def _message_delta_usages(all_chunks: Sequence[str | bytes]) -> Iterator[object]: + for chunk in reversed(all_chunks): + yield from ( + payload.get("usage") + for payload in map(_event_payload, reversed(split_sse_chunk_into_events(chunk))) + if payload is not None and payload.get("type") == "message_delta" + ) + + +def _stamp_reported_stream_cost( + response: ModelResponse, all_chunks: Sequence[str | bytes], logging_obj: LiteLLMLoggingObj +) -> None: + """The final ``message_delta`` is where a provider that prices its own responses reports the cost.""" + provider: Final = logging_obj.model_call_details.get("custom_llm_provider") + costs: Final = (native_messages_reported_cost(usage, provider) for usage in _message_delta_usages(all_chunks)) + reported: Final = next((cost for cost in costs if cost is not None), None) + if reported is not None: + override_stamped_response_cost(response, reported) + + class AnthropicPassthroughLoggingHandler: @staticmethod def anthropic_passthrough_handler( @@ -222,6 +259,7 @@ class AnthropicPassthroughLoggingHandler: ) if not isinstance(response, ModelResponse): return response + _stamp_reported_stream_cost(response, all_chunks, litellm_logging_obj) recovered_usage: Final = AnthropicPassthroughLoggingHandler._recover_interrupted_stream_output_tokens( response=response, all_chunks=all_chunks, model=model ) @@ -286,13 +324,21 @@ class AnthropicPassthroughLoggingHandler: if custom_llm_provider and not model.startswith(f"{custom_llm_provider}/") else model ) + custom_pricing: Final = use_custom_pricing_for_model( + litellm_params=(logging_obj.litellm_params if hasattr(logging_obj, "litellm_params") else None) + ) + reported_cost: Final = ( + response_cost_in_hidden_params(litellm_model_response) + if isinstance(litellm_model_response, ModelResponse) + else None + ) + if reported_cost is not None and not custom_pricing: + return reported_cost return litellm.completion_cost( completion_response=litellm_model_response, model=model_for_cost, custom_llm_provider=custom_llm_provider, - custom_pricing=use_custom_pricing_for_model( - litellm_params=(logging_obj.litellm_params if hasattr(logging_obj, "litellm_params") else None) - ), + custom_pricing=custom_pricing, router_model_id=logging_obj.get_router_model_id(), ) diff --git a/tests/integration/pricing/test_opencode_go_reported_cost_wire.py b/tests/integration/pricing/test_opencode_go_reported_cost_wire.py new file mode 100644 index 00000000000..6ac3bdccf27 --- /dev/null +++ b/tests/integration/pricing/test_opencode_go_reported_cost_wire.py @@ -0,0 +1,198 @@ +import json +from collections.abc import Mapping +from typing import Final + +import pytest +from integration._support.client import Gateway, eventually +from integration._support.database import read_rows +from integration._support.wire import Reply, Request, wire_server +from pydantic import JsonValue, TypeAdapter + +_LIST_COST: Final = 0.00412 +_JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue]) + +# OpenCode's documented usage block (opencode.ai/v2/docs/console/api/inference, "Cost", read 2026-10-24): +# usage.cost is what hit the prepaid balance (0 inside the Go plan), usage.opencode.list_cost is the model price +_GO_BLOCK: Final[Mapping[str, JsonValue]] = { + "cost": 0, + "opencode": {"billing_source": "go", "list_cost": _LIST_COST, "allowance": _LIST_COST}, +} +_CHAT_USAGE: Final[Mapping[str, JsonValue]] = { + "prompt_tokens": 1200, + "completion_tokens": 300, + "total_tokens": 1500, + **_GO_BLOCK, +} +_RESPONSES_USAGE: Final[Mapping[str, JsonValue]] = { + "input_tokens": 1200, + "output_tokens": 300, + "total_tokens": 1500, + **_GO_BLOCK, +} +_MESSAGES_USAGE: Final[Mapping[str, JsonValue]] = {"input_tokens": 1200, "output_tokens": 300, **_GO_BLOCK} + + +def _approx(value: float) -> object: + return pytest.approx(value, rel=1e-6) # pyright: ignore[reportUnknownMemberType] # pytest lacks typed approx stubs + + +def _data_frame(event: Mapping[str, JsonValue]) -> bytes: + return f"data: {json.dumps(event)}\n\n".encode() + + +def _event_frame(event: Mapping[str, JsonValue]) -> bytes: + return f"event: {event['type']}\ndata: {json.dumps(event)}\n\n".encode() + + +def _chat_completion() -> bytes: + return json.dumps( + { + "id": "chatcmpl-go", + "object": "chat.completion", + "created": 1, + "model": "glm-5.3", + "choices": [{"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": "ok"}}], + "usage": dict(_CHAT_USAGE), + } + ).encode() + + +def _chat_stream() -> tuple[bytes, ...]: + text: Final = { + "id": "chatcmpl-go", + "object": "chat.completion.chunk", + "created": 1, + "model": "glm-5.3", + "choices": [{"index": 0, "delta": {"role": "assistant", "content": "ok"}, "finish_reason": None}], + } + finish: Final = {**text, "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]} + final_usage: Final = {**text, "choices": [], "usage": dict(_CHAT_USAGE)} + return (*(_data_frame(event) for event in (text, finish, final_usage)), b"data: [DONE]\n\n") + + +def _responses_response() -> Mapping[str, JsonValue]: + return { + "id": "resp_go", + "object": "response", + "created_at": 1, + "status": "completed", + "model": "gpt-6-luna", + "output": [ + { + "id": "msg_go", + "type": "message", + "status": "completed", + "role": "assistant", + "content": [{"type": "output_text", "text": "ok", "annotations": []}], + } + ], + "usage": dict(_RESPONSES_USAGE), + } + + +def _responses_stream() -> tuple[bytes, ...]: + response: Final = _responses_response() + return ( + _event_frame( + {"type": "response.created", "sequence_number": 0, "response": {**response, "status": "in_progress"}} + ), + _event_frame( + { + "type": "response.output_text.delta", + "sequence_number": 1, + "item_id": "msg_go", + "output_index": 0, + "content_index": 0, + "delta": "ok", + } + ), + _event_frame({"type": "response.completed", "sequence_number": 2, "response": response}), + ) + + +def _anthropic_message() -> Mapping[str, JsonValue]: + return { + "id": "msg_go", + "type": "message", + "role": "assistant", + "model": "claude-haiku-5-5", + "content": [{"type": "text", "text": "ok"}], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": dict(_MESSAGES_USAGE), + } + + +def _anthropic_stream() -> tuple[bytes, ...]: + start: Final = {**_anthropic_message(), "content": [], "stop_reason": None} + return ( + _event_frame( + {"type": "message_start", "message": {**start, "usage": {"input_tokens": 1200, "output_tokens": 0}}} + ), + _event_frame({"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}), + _event_frame({"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "ok"}}), + _event_frame({"type": "content_block_stop", "index": 0}), + _event_frame( + { + "type": "message_delta", + "delta": {"stop_reason": "end_turn", "stop_sequence": None}, + "usage": dict(_MESSAGES_USAGE), + } + ), + _event_frame({"type": "message_stop"}), + ) + + +def _respond(request: Request) -> Reply: + assert request.method == "POST", request.method + streaming: Final = _JSON_OBJECT.validate_json(request.body).get("stream") is True + if request.target.endswith("/chat/completions"): + return ( + Reply(content_type="text/event-stream", chunks=_chat_stream()) + if streaming + else Reply(body=_chat_completion()) + ) + if request.target.endswith("/responses"): + if streaming: + return Reply(content_type="text/event-stream", chunks=_responses_stream()) + return Reply(body=json.dumps(_responses_response()).encode()) + assert request.target.endswith("/v1/messages"), request.target + if streaming: + return Reply(content_type="text/event-stream", chunks=_anthropic_stream()) + return Reply(body=json.dumps(_anthropic_message()).encode()) + + +def _logged_spend(call_id: str) -> float: + rows: Final = eventually( + lambda: read_rows('SELECT spend FROM "LiteLLM_SpendLogs" WHERE litellm_call_id = %s', (call_id,)), + lambda values: len(values) == 1, + seconds=70, + ) + return float(rows[0]["spend"]) + + +def _body(path: str, model: str, stream: bool) -> Mapping[str, JsonValue]: + if path == "/v1/responses": + return {"model": model, "input": "hi", "stream": stream} + return {"model": model, "max_tokens": 16, "messages": [{"role": "user", "content": "hi"}], "stream": stream} + + +@pytest.mark.parametrize("stream", (False, True), ids=("non-stream", "stream")) +@pytest.mark.parametrize( + ("path", "model"), + ( + ("/v1/chat/completions", "opencode_go/glm-5.3"), + ("/v1/responses", "opencode_go/gpt-6-luna"), + ("/v1/messages", "opencode_go/claude-haiku-5-5"), + ), + ids=("chat", "responses", "messages"), +) +def test_request_is_billed_at_the_reported_list_cost(gateway: Gateway, path: str, model: str, stream: bool) -> None: + with wire_server(_respond) as wire, gateway.scenario() as scenario: + deployment: Final = scenario.model(model=model, api_base=wire.url, api_key="synthetic-go-key") + response: Final = gateway.request("POST", path, _body(path, deployment, stream)) + assert response.status_code == 200, response.text + assert "ok" in response.text + if not stream: + assert float(response.headers["x-litellm-response-cost"]) == _approx(_LIST_COST) + assert _logged_spend(response.headers["x-litellm-call-id"]) == _approx(_LIST_COST) diff --git a/tests/unit/litellm_core_utils/test_litellm_logging.py b/tests/unit/litellm_core_utils/test_litellm_logging.py index 8509b1d6f7d..561ac9b2b33 100644 --- a/tests/unit/litellm_core_utils/test_litellm_logging.py +++ b/tests/unit/litellm_core_utils/test_litellm_logging.py @@ -76,6 +76,7 @@ import inspect from typing import List, Optional from litellm.types.utils import LiteLLMCommonStrings + @pytest.fixture def logging_obj(): return LitellmLogging( @@ -346,8 +347,10 @@ async def test_mcp_native_structured_replacement_must_match_returned_content( ) logging_obj.dynamic_success_callbacks = [NativeReplacement()] returned = await logging_obj.async_post_mcp_tool_call_hook( - kwargs={"original_response": result}, response_obj=result, - start_time=datetime.datetime.now(), end_time=datetime.datetime.now(), + kwargs={"original_response": result}, + response_obj=result, + start_time=datetime.datetime.now(), + end_time=datetime.datetime.now(), ) assert returned is result assert result.content == [TextContent(type="text", text="native-safe" if same_content else "final-safe")] @@ -356,7 +359,6 @@ async def test_mcp_native_structured_replacement_must_match_returned_content( assert "SECRET-1234" not in result.model_dump_json() - @pytest.mark.asyncio @pytest.mark.parametrize("mode", ["none", "wrapper", "exception"]) @pytest.mark.parametrize("structured", [False, True]) @@ -374,8 +376,10 @@ async def test_mcp_direct_content_edit_invalidates_stale_structured_data(logging ) logging_obj.dynamic_success_callbacks = [DirectRedactor()] returned = await logging_obj.async_post_mcp_tool_call_hook( - kwargs={"original_response": result}, response_obj=result, - start_time=datetime.datetime.now(), end_time=datetime.datetime.now(), + kwargs={"original_response": result}, + response_obj=result, + start_time=datetime.datetime.now(), + end_time=datetime.datetime.now(), ) assert returned is result assert result.content == [TextContent(type="text", text="[REDACTED]")] @@ -2620,7 +2624,6 @@ async def test_completed_success_selects_one_implemented_hook( assert (thread_id == loop_thread) is (kind != "sync") - @pytest.mark.parametrize("for_async_request,split_registries", [(False, False), (True, False), (True, True)]) @pytest.mark.parametrize("private_first", [False, True]) @pytest.mark.parametrize("response_kind", ["responses", "completion"]) @@ -3312,8 +3315,13 @@ async def test_shadow_snapshot_stays_private_and_is_invalidated_before_logging_g class RecordingShadowLogger(ShadowEvalLogger): async def async_log_success_event( - self, kwargs: Mapping[str, object], response_obj: object, start_time: object, - end_time: object, *, guardrail_snapshot: GuardrailRequestSnapshot | None = None, + self, + kwargs: Mapping[str, object], + response_obj: object, + start_time: object, + end_time: object, + *, + guardrail_snapshot: GuardrailRequestSnapshot | None = None, ) -> None: shadow_snapshots.append(guardrail_snapshot) await super().async_log_success_event( @@ -3322,13 +3330,20 @@ async def test_shadow_snapshot_stays_private_and_is_invalidated_before_logging_g class RecordingLogger(CustomLogger): async def async_log_success_event( - self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object, + self, + kwargs: Mapping[str, object], + response_obj: object, + start_time: object, + end_time: object, ) -> None: other_payloads.append(kwargs) class LoggingGuardrail(CustomGuardrail): async def async_logging_hook( - self, kwargs: dict[str, object], result: object, call_type: str, + self, + kwargs: dict[str, object], + result: object, + call_type: str, ) -> tuple[dict[str, object], object]: hook_snapshots.append(logging_obj.shadow_eval_request_snapshot) if hook_mode == "raises": @@ -3340,26 +3355,36 @@ async def test_shadow_snapshot_stays_private_and_is_invalidated_before_logging_g "user_api_key_hash": "test-key", } snapshot: Final = GuardrailRequestSnapshot.capture( - {"messages": [{"role": "user", "content": "snapshot-only"}]}, metadata, + {"messages": [{"role": "user", "content": "snapshot-only"}]}, + metadata, ) assert snapshot is not None shadow: Final = RecordingShadowLogger(prisma_provider=no_prisma, jobs_cache=InMemoryCache()) guardrail: Final = LoggingGuardrail( - guardrail_name="late-mask", default_on=True, + guardrail_name="late-mask", + default_on=True, event_hook=GuardrailEventHooks.pre_call if hook_mode == "disabled" else GuardrailEventHooks.logging_only, ) monkeypatch.setattr(litellm, "_async_success_callback", []) logging_obj: Final = LitellmLogging( - model="test-model", messages=[], stream=stream, call_type="anthropic_messages", - start_time=datetime.datetime.now(), litellm_call_id="private-snapshot", function_id="private-snapshot", + model="test-model", + messages=[], + stream=stream, + call_type="anthropic_messages", + start_time=datetime.datetime.now(), + litellm_call_id="private-snapshot", + function_id="private-snapshot", dynamic_async_success_callbacks=[shadow, RecordingLogger(), guardrail], ) logging_obj.update_messages([{"role": "user", "content": "logged input"}]) logging_obj.update_environment_variables(litellm_params={"metadata": metadata}, optional_params={}) logging_obj.shadow_eval_request_snapshot = snapshot payload: Final = { - "id": "private-snapshot", "call_type": "anthropic_messages", "metadata": metadata, - "model_group": "test-model", "model_parameters": {}, + "id": "private-snapshot", + "call_type": "anthropic_messages", + "metadata": metadata, + "model_group": "test-model", + "model_parameters": {}, } await logging_obj.async_success_handler(result=ModelResponse(), standard_logging_object=payload) @@ -3816,7 +3841,9 @@ def test_sentry_event_scrubber_initialization(monkeypatch): call_args = mock_init.call_args[1] assert call_args["send_default_pii"] is False assert call_args["event_scrubber"].recursive is True - assert {name.lower() for name in SENTRY_PII_DENYLIST} <= {name.lower() for name in call_args["event_scrubber"].denylist} + assert {name.lower() for name in SENTRY_PII_DENYLIST} <= { + name.lower() for name in call_args["event_scrubber"].denylist + } assert call_args["before_send"] is call_args["before_send_transaction"] @@ -3833,7 +3860,9 @@ def test_sentry_send_default_pii_opt_in(monkeypatch): call_args = mock_init.call_args[1] assert call_args["send_default_pii"] is True - assert not {name.lower() for name in SENTRY_PII_DENYLIST} & {name.lower() for name in call_args["event_scrubber"].denylist} + assert not {name.lower() for name in SENTRY_PII_DENYLIST} & { + name.lower() for name in call_args["event_scrubber"].denylist + } def test_get_masked_values(): @@ -5103,12 +5132,10 @@ def test_reported_cost_respects_configured_pricing_and_additional_charges( else {} ), } - } - } - ) - logging_obj.model_call_details.update( - {"additional_response_cost": additional_cost, "cache_hit": cache_hit} + }, + }, ) + logging_obj.model_call_details.update({"additional_response_cost": additional_cost, "cache_hit": cache_hit}) logging_obj._provider_reported_cost = reported_cost response: Final = ModelResponse( model=pricing_model, @@ -7055,8 +7082,11 @@ async def test_background_interaction_completion_rebills_after_in_progress_succe class SyncRecorder(CustomLogger): def log_success_event( - self, kwargs: dict[str, object], response_obj: object, - start_time: dt.datetime, end_time: dt.datetime, + self, + kwargs: dict[str, object], + response_obj: object, + start_time: dt.datetime, + end_time: dt.datetime, ) -> None: events.put((response_obj, kwargs.get("response_cost"))) logged.set() @@ -7089,11 +7119,11 @@ async def test_background_interaction_completion_rebills_after_in_progress_succe assert logging_obj.model_call_details["response_cost"] > 0 assert logging_obj.model_call_details["standard_logging_object"]["total_tokens"] == 175 - assert events.qsize() == 2 assert events.get_nowait() == (in_progress, None) assert events.get_nowait() == (completed, logging_obj.model_call_details["response_cost"]) + @pytest.mark.asyncio async def test_background_interaction_completion_prices_the_settled_body_itself(): """ @@ -8127,7 +8157,9 @@ def test_debugging_log_with_json_logs_tolerates_missing_headers(logging_obj, mon logging_obj.litellm_request_debug = True with patch("litellm.litellm_core_utils.litellm_logging.verbose_logger.warning") as warning: - logging_obj._print_llm_call_debugging_log(api_base="https://api.openai.com/v1", headers=None, additional_args={}) + logging_obj._print_llm_call_debugging_log( + api_base="https://api.openai.com/v1", headers=None, additional_args={} + ) assert "https://api.openai.com/v1" in warning.call_args.kwargs["extra"]["api_base"] @@ -10378,8 +10410,13 @@ class TestAzurePTUSpilloverCost: assert payload is not None model_map_value: Final = payload["model_map_information"]["model_map_value"] assert model_map_value is not None - assert model_map_value["input_cost_per_token"] == litellm.model_cost[self.SERVED_MODEL]["input_cost_per_token"] - assert model_map_value["output_cost_per_token"] == litellm.model_cost[self.SERVED_MODEL]["output_cost_per_token"] + assert ( + model_map_value["input_cost_per_token"] == litellm.model_cost[self.SERVED_MODEL]["input_cost_per_token"] + ) + assert ( + model_map_value["output_cost_per_token"] + == litellm.model_cost[self.SERVED_MODEL]["output_cost_per_token"] + ) assert response_cost == pytest.approx( 100 * model_map_value["input_cost_per_token"] + 50 * model_map_value["output_cost_per_token"] ) @@ -10494,8 +10531,6 @@ def test_get_assembled_streaming_response_bills_a_provider_reported_usage_cost() assert logging_obj.response_cost_calculator(result=assembled) == 0.0042 - - def test_response_cost_calculator_prices_terminal_responses_event_from_its_response(): logging_obj: Final = _responses_stream_logging_obj() inner_response: Final = ResponsesAPIResponse( @@ -10725,9 +10760,7 @@ def test_standard_logging_retries(): ] ) - with patch.object( - customHandler, "log_failure_event", new=MagicMock() - ) as mock_client: + with patch.object(customHandler, "log_failure_event", new=MagicMock()) as mock_client: try: router.completion( model="gpt-3.5-turbo", @@ -10739,19 +10772,10 @@ def test_standard_logging_retries(): pass assert mock_client.call_count == 2 + assert mock_client.call_args_list[0].kwargs["kwargs"]["standard_logging_object"]["trace_id"] is not None assert ( - mock_client.call_args_list[0].kwargs["kwargs"]["standard_logging_object"][ - "trace_id" - ] - is not None - ) - assert ( - mock_client.call_args_list[0].kwargs["kwargs"]["standard_logging_object"][ - "trace_id" - ] - == mock_client.call_args_list[1].kwargs["kwargs"][ - "standard_logging_object" - ]["trace_id"] + mock_client.call_args_list[0].kwargs["kwargs"]["standard_logging_object"]["trace_id"] + == mock_client.call_args_list[1].kwargs["kwargs"]["standard_logging_object"]["trace_id"] ) @@ -11018,9 +11042,7 @@ def test_signoz_dispatch_requires_an_endpoint(monkeypatch): custom_logger_init_args={}, ) assert created is None - assert not [ - cb for cb in logging_module._in_memory_loggers if getattr(cb, "callback_name", None) == "signoz" - ] + assert not [cb for cb in logging_module._in_memory_loggers if getattr(cb, "callback_name", None) == "signoz"] finally: logging_module._in_memory_loggers.clear() monkeypatch.delenv("LITELLM_OTEL_V2", raising=False) @@ -11162,7 +11184,11 @@ def test_standard_logging_payload_reports_the_deployment_entry_the_request_was_b } } logging_obj.update_environment_variables( - model="gpt-5.4-nano", user="", optional_params={}, litellm_params=litellm_params, custom_llm_provider="openai" + model="gpt-5.4-nano", + user="", + optional_params={}, + litellm_params=litellm_params, + custom_llm_provider="openai", ) response: Final = ModelResponse( model="gpt-5.4-nano", @@ -11474,10 +11500,7 @@ def test_enterprise_alerting_loggers_falls_back_without_disabling_callback_contr ) logging_module._enterprise_alerting_loggers.cache_clear() for module_name in tuple(sys.modules): - if any( - module_name == prefix or module_name.startswith(f"{prefix}.") - for prefix in module_prefixes - ): + if any(module_name == prefix or module_name.startswith(f"{prefix}.") for prefix in module_prefixes): monkeypatch.delitem(sys.modules, module_name, raising=False) monkeypatch.setattr(sys, "meta_path", [BlockPagerDutyFinder(), *sys.meta_path]) @@ -11597,13 +11620,16 @@ def _vcr_outcome_gate(request, vcr): yield record_vcr_outcome(request, vcr) + @pytest_asyncio.fixture(loop_scope="function") async def drain_logging_worker(isolate_litellm_state: None) -> AsyncIterator[None]: yield await asyncio.wait_for(GLOBAL_LOGGING_WORKER.flush(), timeout=LOGGING_WORKER_DRAIN_TIMEOUT_SECONDS) + LOGGING_WORKER_DRAIN_TIMEOUT_SECONDS: Final = LOGGING_WORKER_MAX_TIME_PER_COROUTINE + 5.0 + @pytest.fixture(scope="function") def isolate_litellm_state(): """ @@ -11641,6 +11667,7 @@ def isolate_litellm_state(): if attr in _DEFAULTS: setattr(litellm, attr, _DEFAULTS[attr]) + _LIST_ATTRS = ( "callbacks", "success_callback", @@ -11668,6 +11695,7 @@ _SCALAR_ATTRS = ( _DEFAULTS: dict = {} + @pytest.fixture(scope="module") def setup_and_teardown(): """ @@ -11690,6 +11718,7 @@ def setup_and_teardown(): litellm.in_memory_llm_clients_cache.flush_cache() yield + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") @pytest.mark.parametrize( "response_obj,expected_values", @@ -11737,6 +11766,7 @@ def test_get_usage(response_obj, expected_values): assert usage.completion_tokens == expected_values[1] assert usage.total_tokens == expected_values[2] + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_get_usage_from_image_generation_response(): """ @@ -11784,6 +11814,7 @@ def test_get_usage_from_image_generation_response(): assert usage.completion_tokens_details.image_tokens == 272 assert usage.completion_tokens_details.text_tokens == 100 + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_get_additional_headers(): additional_headers = { @@ -11824,10 +11855,12 @@ def test_get_additional_headers(): assert additional_logging_headers.get("llm_provider-request-id") == "req_01F6CycZZPSHKRCCctcS1Vto" assert additional_logging_headers.get("llm_provider-anthropic-ratelimit-requests-reset") == "2024-10-29T23:57:40Z" + def all_fields_present(standard_logging_metadata: StandardLoggingMetadata): for field in StandardLoggingMetadata.__annotations__.keys(): assert field in standard_logging_metadata + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") @pytest.mark.parametrize( "metadata_key, metadata_value", @@ -11859,6 +11892,7 @@ def test_get_standard_logging_metadata(metadata_key, metadata_value): # Assert that the specific metadata field is set correctly assert standard_logging_metadata[metadata_key] == metadata_value + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_get_standard_logging_metadata_user_api_key_hash(): valid_hash = "a" * 64 # 64 character string @@ -11866,6 +11900,7 @@ def test_get_standard_logging_metadata_user_api_key_hash(): result = StandardLoggingPayloadSetup.get_standard_logging_metadata(metadata) assert result["user_api_key_hash"] == valid_hash + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_get_standard_logging_metadata_invalid_user_api_key(): invalid_hash = "not_a_valid_hash" @@ -11874,6 +11909,7 @@ def test_get_standard_logging_metadata_invalid_user_api_key(): all_fields_present(result) assert result["user_api_key_hash"] is None + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_get_standard_logging_metadata_non_string_user_api_key(): """Non-string user_api_key should not be set as user_api_key_hash.""" @@ -11882,6 +11918,7 @@ def test_get_standard_logging_metadata_non_string_user_api_key(): all_fields_present(result) assert result["user_api_key_hash"] is None + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_get_standard_logging_metadata_none_user_api_key(): """None user_api_key should not be set as user_api_key_hash.""" @@ -11890,6 +11927,7 @@ def test_get_standard_logging_metadata_none_user_api_key(): all_fields_present(result) assert result["user_api_key_hash"] is None + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_get_standard_logging_metadata_invalid_keys(): metadata = { @@ -11903,6 +11941,7 @@ def test_get_standard_logging_metadata_invalid_keys(): assert "invalid_key" not in result assert "another_invalid_key" not in result + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_cleanup_timestamps(): """Test cleanup_timestamps with different input types""" @@ -11937,6 +11976,7 @@ def test_cleanup_timestamps(): with pytest.raises(ValueError, match="start_time is required, got=invalid of type "): StandardLoggingPayloadSetup.cleanup_timestamps("invalid", end_float, completion_float) + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_get_model_cost_information(): """Test get_model_cost_information with different inputs""" @@ -11969,6 +12009,7 @@ def test_get_model_cost_information(): # assert all fields in StandardLoggingModelInformation are present assert all(field in result for field in StandardLoggingModelInformation.__annotations__) + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_get_model_cost_information_custom_pricing_uses_base_model(): result = StandardLoggingPayloadSetup.get_model_cost_information( @@ -11980,6 +12021,7 @@ def test_get_model_cost_information_custom_pricing_uses_base_model(): assert result["model_map_value"] is not None assert result["model_map_key"] != "invoke_test_claude" + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_standard_logging_payload_uses_deployment_when_no_base_model(): """metadata["deployment"] is used for cost-map lookup when base_model is not set.""" @@ -12037,6 +12079,7 @@ def test_standard_logging_payload_uses_deployment_when_no_base_model(): assert payload["model_map_information"]["model_map_value"] is not None assert payload["model_map_information"]["model_map_key"] != "invoke_test_claude" + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_get_hidden_params(): """Test get_hidden_params with different inputs""" @@ -12072,6 +12115,7 @@ def test_get_hidden_params(): # assert all fields in StandardLoggingHiddenParams are present assert all(field in result for field in StandardLoggingHiddenParams.__annotations__) + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_get_final_response_obj(): """Test get_final_response_obj with different input types and redaction scenarios""" @@ -12103,6 +12147,7 @@ def test_get_final_response_obj(): # Reset litellm.turn_off_message_logging to its original value litellm.turn_off_message_logging = False + def testget_standard_logging_payload_trace_id(): """Test get_standard_logging_payload_trace_id with different input scenarios""" # Test case 1: When litellm_trace_id is provided in litellm_params @@ -12140,6 +12185,7 @@ def testget_standard_logging_payload_trace_id(): assert result == "12345" assert isinstance(result, str) + def testget_standard_logging_payload_trace_id_prioritizes_trace_id_when_flag_on(monkeypatch): """With request_correlation_in_logs on, an explicit litellm_trace_id wins over litellm_session_id.""" from unittest.mock import MagicMock @@ -12154,6 +12200,7 @@ def testget_standard_logging_payload_trace_id_prioritizes_trace_id_when_flag_on( ) assert result == "the-trace-id" + def testget_standard_logging_payload_trace_id_prioritizes_session_id_when_flag_off(monkeypatch): """With request_correlation_in_logs off (default), legacy behavior is preserved: litellm_session_id still wins over litellm_trace_id.""" @@ -12169,6 +12216,7 @@ def testget_standard_logging_payload_trace_id_prioritizes_session_id_when_flag_o ) assert result == "the-session-id" + def testget_standard_logging_payload_session_id_when_flag_on(monkeypatch): """Test get_standard_logging_payload_session_id with different input scenarios, flag enabled""" from unittest.mock import MagicMock @@ -12221,6 +12269,7 @@ def testget_standard_logging_payload_session_id_when_flag_on(monkeypatch): ) assert result == "" + def testget_standard_logging_payload_session_id_empty_when_flag_off(monkeypatch): """When request_correlation_in_logs is off (default), session_id is always empty, even if litellm_session_id was explicitly supplied - preserves the pre-existing @@ -12237,6 +12286,7 @@ def testget_standard_logging_payload_session_id_empty_when_flag_off(monkeypatch) ) assert result == "" + @pytest.fixture def restore_verbose_logger_level() -> Iterator[None]: original_level: Final = verbose_logger.level @@ -12279,6 +12329,7 @@ def test_truncate_standard_logging_payload(): assert len(str(truncated["response"])) < 10_500 assert len(str(truncated["error_str"])) < 10_500 + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_truncate_standard_logging_payload_keeps_a_partial_payload_intact(): """A payload built with only some of its fields comes back with exactly those keys and values""" @@ -12287,12 +12338,14 @@ def test_truncate_standard_logging_payload_keeps_a_partial_payload_intact(): assert _custom_logger.truncate_standard_logging_payload_content(partial_payload) == partial_payload + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_strip_trailing_slash(): common_api_base = "https://api.test.com" assert StandardLoggingPayloadSetup.strip_trailing_slash(common_api_base + "/") == common_api_base assert StandardLoggingPayloadSetup.strip_trailing_slash(common_api_base) == common_api_base + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_get_error_information(): """Test get_error_information with different types of exceptions""" @@ -12329,6 +12382,7 @@ def test_get_error_information(): assert result["llm_provider"] == "openai" assert result["error_message"] == "litellm.RateLimitError: Test error" + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_get_response_time(): """Test get_response_time with different streaming scenarios""" @@ -12364,6 +12418,7 @@ def test_get_response_time(): # For streaming, should return completion_start_time - start_time assert response_time == 2.0 + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") @pytest.mark.parametrize( "metadata, expected_requester_metadata", @@ -12386,6 +12441,7 @@ def test_standard_logging_metadata_requester_metadata(metadata, expected_request result = StandardLoggingPayloadSetup.get_standard_logging_metadata(metadata) assert result["requester_metadata"] == expected_requester_metadata + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_cost_breakdown_in_standard_logging_payload(): """ @@ -12475,6 +12531,7 @@ def test_cost_breakdown_in_standard_logging_payload(): print("✅ Cost breakdown test passed!") + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_cost_breakdown_missing_in_standard_logging_payload(): """ @@ -12535,6 +12592,7 @@ def test_cost_breakdown_missing_in_standard_logging_payload(): print("✅ Cost breakdown missing test passed!") + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") @pytest.mark.parametrize( "use_combined_usage_object", @@ -12622,6 +12680,7 @@ def test_usage_dict_roundtrip_in_payload(use_combined_usage_object): assert usage_obj["completion_tokens"] == 58 assert usage_obj["total_tokens"] == 100 + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_standard_logging_payload_uses_actual_model_for_azure_router(): from litellm.litellm_core_utils.litellm_logging import ( @@ -12670,6 +12729,7 @@ def test_standard_logging_payload_uses_actual_model_for_azure_router(): assert payload is not None assert payload["model"] == "azure_ai/gpt-5-nano-2025-08-07" + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_standard_logging_payload_uses_actual_model_for_azure_router_with_underscore(): from litellm.litellm_core_utils.litellm_logging import ( @@ -12718,6 +12778,7 @@ def test_standard_logging_payload_uses_actual_model_for_azure_router_with_unders assert payload is not None assert payload["model"] == "azure_ai/gpt-5-nano-2025-08-07" + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_merge_litellm_metadata_basic(): """ @@ -12749,6 +12810,7 @@ def test_merge_litellm_metadata_basic(): assert result["model_info"] == {"id": "model-123"} assert result["tags"] == ["tag1", "tag2"] + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_merge_litellm_metadata_precedence(): """ @@ -12775,6 +12837,7 @@ def test_merge_litellm_metadata_precedence(): # litellm_metadata values should only be included if not in metadata assert result["model_group"] == "gpt-4-group" + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_merge_litellm_metadata_skip_non_serializable(): """ @@ -12807,6 +12870,7 @@ def test_merge_litellm_metadata_skip_non_serializable(): assert result["safe_field"] == "safe_value" assert result["model_group"] == "gpt-4-group" + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_merge_litellm_metadata_empty_params(): """ @@ -12842,6 +12906,7 @@ def test_merge_litellm_metadata_empty_params(): result = StandardLoggingPayloadSetup.merge_litellm_metadata(litellm_params) assert result == {} + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_merge_litellm_metadata_bedrock_passthrough_scenario(): """ @@ -12901,8 +12966,10 @@ def test_merge_litellm_metadata_bedrock_passthrough_scenario(): # Verify total number of fields (9 user fields + 4 model fields = 13) assert len(result) == 13 + service_logger = ServiceLogging() + def setup_logging(): return Logging( model="gpt-5.5", @@ -12992,6 +13059,7 @@ def test_get_callback_name(): # Test string callback assert logging._get_callback_name("callback_string") == "callback_string" + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_is_internal_litellm_proxy_callback(): """ @@ -13012,6 +13080,7 @@ def test_is_internal_litellm_proxy_callback(): # Test string callback assert logging._is_internal_litellm_proxy_callback("callback_string") == False + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_should_run_sync_callbacks_for_async_calls(): """ @@ -13037,6 +13106,7 @@ def test_should_run_sync_callbacks_for_async_calls(): litellm.success_callback = [PROXY_MaxIterationsHandler] assert logging._should_run_sync_callbacks_for_async_calls() == False + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_remove_internal_litellm_callbacks(): logging = setup_logging() @@ -13058,6 +13128,7 @@ def test_remove_internal_litellm_callbacks(): assert PROXY_MaxIterationsHandler not in filtered assert PROXY_CacheControlCheck not in filtered + @pytest.mark.asyncio async def test_background_interaction_completion_logs_while_in_progress_handler_is_parked(monkeypatch): """ @@ -13160,14 +13231,9 @@ class CompletionCustomHandler( assert metadata_value is None or isinstance(metadata_value, dict) if metadata_value is not None: if litellm.turn_off_message_logging is True: - assert ( - metadata_value["raw_request"] - is LiteLLMCommonStrings.redacted_by_litellm.value - ) + assert metadata_value["raw_request"] is LiteLLMCommonStrings.redacted_by_litellm.value else: - assert "raw_request" not in metadata_value or isinstance( - metadata_value["raw_request"], str - ) + assert "raw_request" not in metadata_value or isinstance(metadata_value["raw_request"], str) except Exception: print(f"Assertion Error: {traceback.format_exc()}") self.errors.append(traceback.format_exc()) @@ -13216,23 +13282,18 @@ class CompletionCustomHandler( assert isinstance(response_obj, litellm.ModelResponseStream) ## KWARGS assert isinstance(kwargs["model"], str) - assert isinstance(kwargs["messages"], list) and isinstance( - kwargs["messages"][0], dict - ) + assert isinstance(kwargs["messages"], list) and isinstance(kwargs["messages"][0], dict) assert isinstance(kwargs["optional_params"], dict) assert isinstance(kwargs["litellm_params"], dict) assert isinstance(kwargs["start_time"], (datetime, type(None))) assert isinstance(kwargs["stream"], bool) assert isinstance(kwargs["user"], (str, type(None))) - assert ( - isinstance(kwargs["input"], list) - and isinstance(kwargs["input"][0], dict) - ) or isinstance(kwargs["input"], (dict, str)) + assert (isinstance(kwargs["input"], list) and isinstance(kwargs["input"][0], dict)) or isinstance( + kwargs["input"], (dict, str) + ) assert isinstance(kwargs["api_key"], (str, type(None))) assert ( - isinstance( - kwargs["original_response"], (str, litellm.CustomStreamWrapper) - ) + isinstance(kwargs["original_response"], (str, litellm.CustomStreamWrapper)) or inspect.isasyncgen(kwargs["original_response"]) or inspect.iscoroutine(kwargs["original_response"]) ) @@ -13265,9 +13326,7 @@ class CompletionCustomHandler( ) ## KWARGS assert isinstance(kwargs["model"], str) - assert isinstance(kwargs["messages"], list) and isinstance( - kwargs["messages"][0], dict - ) + assert isinstance(kwargs["messages"], list) and isinstance(kwargs["messages"][0], dict) assert isinstance(kwargs["optional_params"], dict) assert isinstance(kwargs["litellm_params"], dict) assert isinstance(kwargs["litellm_params"]["api_base"], str) @@ -13277,10 +13336,7 @@ class CompletionCustomHandler( assert isinstance(kwargs["user"], (str, type(None))) assert ( isinstance(kwargs["input"], list) - and ( - isinstance(kwargs["input"][0], dict) - or isinstance(kwargs["input"][0], str) - ) + and (isinstance(kwargs["input"][0], dict) or isinstance(kwargs["input"][0], str)) ) or isinstance(kwargs["input"], (dict, str)) assert isinstance(kwargs["api_key"], (str, type(None))) assert isinstance( @@ -13308,9 +13364,7 @@ class CompletionCustomHandler( assert response_obj == None ## KWARGS assert isinstance(kwargs["model"], str) - assert isinstance(kwargs["messages"], list) and isinstance( - kwargs["messages"][0], dict - ) + assert isinstance(kwargs["messages"], list) and isinstance(kwargs["messages"][0], dict) assert isinstance(kwargs["optional_params"], dict) assert isinstance(kwargs["litellm_params"], dict) @@ -13318,15 +13372,12 @@ class CompletionCustomHandler( assert isinstance(kwargs["start_time"], (datetime, type(None))) assert isinstance(kwargs["stream"], bool) assert isinstance(kwargs["user"], (str, type(None))) - assert ( - isinstance(kwargs["input"], list) - and isinstance(kwargs["input"][0], dict) - ) or isinstance(kwargs["input"], (dict, str)) + assert (isinstance(kwargs["input"], list) and isinstance(kwargs["input"][0], dict)) or isinstance( + kwargs["input"], (dict, str) + ) assert isinstance(kwargs["api_key"], (str, type(None))) assert ( - isinstance( - kwargs["original_response"], (str, litellm.CustomStreamWrapper) - ) + isinstance(kwargs["original_response"], (str, litellm.CustomStreamWrapper)) or kwargs["original_response"] == None ) assert isinstance(kwargs["additional_args"], (dict, type(None))) @@ -13344,9 +13395,7 @@ class CompletionCustomHandler( assert isinstance(messages, list) and isinstance(messages[0], dict) ## KWARGS assert isinstance(kwargs["model"], str) - assert isinstance(kwargs["messages"], list) and isinstance( - kwargs["messages"][0], dict - ) + assert isinstance(kwargs["messages"], list) and isinstance(kwargs["messages"][0], dict) assert isinstance(kwargs["optional_params"], dict) assert isinstance(kwargs["litellm_params"], dict) assert isinstance(kwargs["start_time"], (datetime, type(None))) @@ -13358,9 +13407,7 @@ class CompletionCustomHandler( async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): try: - print( - "in async_log_success_event", kwargs, response_obj, start_time, end_time - ) + print("in async_log_success_event", kwargs, response_obj, start_time, end_time) self.states.append("async_success") ## START TIME assert isinstance(start_time, datetime) @@ -13389,9 +13436,7 @@ class CompletionCustomHandler( assert isinstance(kwargs["input"], (list, dict, str)) assert isinstance(kwargs["api_key"], (str, type(None))) assert ( - isinstance( - kwargs["original_response"], (str, litellm.CustomStreamWrapper) - ) + isinstance(kwargs["original_response"], (str, litellm.CustomStreamWrapper)) or inspect.isasyncgen(kwargs["original_response"]) or inspect.iscoroutine(kwargs["original_response"]) ) @@ -13423,9 +13468,7 @@ class CompletionCustomHandler( assert isinstance(kwargs["input"], (list, str, dict)) assert isinstance(kwargs["api_key"], (str, type(None))) assert ( - isinstance( - kwargs["original_response"], (str, litellm.CustomStreamWrapper) - ) + isinstance(kwargs["original_response"], (str, litellm.CustomStreamWrapper)) or inspect.isasyncgen(kwargs["original_response"]) or inspect.iscoroutine(kwargs["original_response"]) or kwargs["original_response"] == None @@ -13566,3 +13609,36 @@ def test_legacy_s3_logs_a_streamed_anthropic_messages_call_handed_off_from_the_a assert s3_writer.log_event.call_count == expected_s3_writes if expected_s3_writes: assert s3_writer.log_event.call_args.kwargs["response_obj"] is complete_response + + +_GO_PRICED_MESSAGE: Final = { + "id": "msg_go", + "type": "message", + "role": "assistant", + "model": "claude-haiku-5-5", + "content": [{"type": "text", "text": "answer"}], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": {"input_tokens": 1200, "output_tokens": 300, "cost": 0, "opencode": {"list_cost": 0.00412}}, +} + + +@pytest.mark.parametrize( + ("usage", "expected_header"), + ( + ({"input_tokens": 1200, "output_tokens": 300, "cost": 0, "opencode": {"list_cost": 0.00412}}, 0.00412), + ({"input_tokens": 1200, "output_tokens": 300}, None), + ), + ids=("list-cost-reported", "no-list-cost-keeps-the-price-map"), +) +def test_native_messages_body_carries_the_reported_list_cost_into_the_priced_response(usage, expected_header): + logging_obj = _anthropic_messages_logging_obj() + logging_obj.stream = False + logging_obj.optional_params = {} + logging_obj.model_call_details["custom_llm_provider"] = "opencode_go" + + priced = logging_obj.billable_result({**_GO_PRICED_MESSAGE, "usage": usage}) + + assert isinstance(priced, ModelResponse) + headers = priced._hidden_params.get("additional_headers") or {} + assert headers.get("llm_provider-x-litellm-response-cost") == expected_header diff --git a/tests/unit/llms/opencode/chat/test_transformation.py b/tests/unit/llms/opencode/chat/test_transformation.py index ce8f46f3da0..0a5e49ca1b2 100644 --- a/tests/unit/llms/opencode/chat/test_transformation.py +++ b/tests/unit/llms/opencode/chat/test_transformation.py @@ -1,3 +1,4 @@ +import asyncio import json from collections.abc import Mapping from datetime import datetime @@ -6,8 +7,11 @@ from typing import Final import httpx import pytest import respx +from pydantic import TypeAdapter import litellm +from litellm.integrations.custom_logger import CustomLogger +from litellm.litellm_core_utils.core_helpers import RESPONSE_COST_HEADER from litellm.litellm_core_utils.litellm_logging import Logging from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler from litellm.llms.opencode.common_utils import go_headers @@ -400,3 +404,179 @@ def test_gemini_sync_stream_preserves_tool_calls_and_usage(go_catalog: None) -> assert json.loads(tool.function.arguments) == {"city": "Paris"} assert any(c.choices[0].finish_reason == "tool_calls" for c in chunks if c.choices) assert chunks[-1].usage.total_tokens == 15 + + +GO_REPORTED_USAGE: Final = { + "cost": 0, + "opencode": {"billing_source": "go", "list_cost": 0.00412, "allowance": 0.00412}, +} +ROW_PRICE_FOR_FIXTURE_USAGE: Final = 10 * 0.000002 + 5 * 0.000005 + + +def go_priced_reply(route: str, model: str, usage_extra: Mapping[str, object]) -> Mapping[str, object]: + reply: Final = upstream_reply(route, model) + key: Final = "usageMetadata" if route == "gemini" else "usage" + usage: Final = TypeAdapter(Mapping[str, object]).validate_python(reply[key]) + return {**reply, key: {**usage, **usage_extra}} + + +class CostRecorder(CustomLogger): + def __init__(self) -> None: + super().__init__() + self.costs: list[float | None] = [] # mutable-ok: test probe accumulates logged costs + + async def async_log_success_event(self, kwargs, response_obj, start_time, end_time) -> None: # noqa: ANN001 + self.costs.append(kwargs.get("response_cost")) + + +async def logged_cost(recorder: CostRecorder) -> float | None: + for _ in range(50): + if recorder.costs: + return recorder.costs[-1] + await asyncio.sleep(0.05) + raise AssertionError("success callback never ran") + + +@pytest.fixture +def cost_recorder(monkeypatch: pytest.MonkeyPatch) -> CostRecorder: + recorder: Final = CostRecorder() + monkeypatch.setattr(litellm, "callbacks", [recorder]) + return recorder + + +@pytest.mark.asyncio +@pytest.mark.parametrize("upstream", ["chat/completions", "messages", "responses", "gemini"]) +@pytest.mark.parametrize("ingress", ["chat", "responses"]) +async def test_reported_list_cost_is_billed_instead_of_rows( + go_catalog: None, cost_recorder: CostRecorder, upstream: str, ingress: str +) -> None: + model: Final = f"fixture-{upstream.split('/')[0]}" + with respx.mock(using="httpx") as transport: + transport.post().mock( + return_value=httpx.Response(200, json=go_priced_reply(upstream, model, GO_REPORTED_USAGE)) + ) + kwargs: Final = {"model": f"opencode_go/{model}", "api_key": "fake-go-key"} + if ingress == "chat": + response: Final = await litellm.acompletion(messages=[{"role": "user", "content": "hello"}], **kwargs) + else: + response = await litellm.aresponses(input="hello", **kwargs) + assert response._hidden_params["additional_headers"][RESPONSE_COST_HEADER] == pytest.approx(0.00412) + assert await logged_cost(cost_recorder) == pytest.approx(0.00412) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("usage_extra", [{}, {"cost": 0}, {"opencode": {"billing_source": "go"}}]) +async def test_rows_price_the_request_when_no_list_cost_is_reported( + go_catalog: None, cost_recorder: CostRecorder, usage_extra: Mapping[str, object] +) -> None: + with respx.mock(using="httpx") as transport: + transport.post().mock( + return_value=httpx.Response(200, json=go_priced_reply("chat/completions", "fixture-chat", usage_extra)) + ) + response: Final = await litellm.acompletion( + model="opencode_go/fixture-chat", api_key="fake-go-key", messages=[{"role": "user", "content": "hello"}] + ) + assert RESPONSE_COST_HEADER not in response._hidden_params.get("additional_headers", {}) + assert await logged_cost(cost_recorder) == pytest.approx(ROW_PRICE_FOR_FIXTURE_USAGE) + + +def sse(*events: Mapping[str, object]) -> str: + return "".join(f"data: {json.dumps(event)}\n\n" for event in events) + "data: [DONE]\n\n" + + +@pytest.mark.asyncio +async def test_chat_stream_bills_the_final_usage_chunks_list_cost( + go_catalog: None, cost_recorder: CostRecorder +) -> None: + chunk: Final = { + "id": "chatcmpl-test", + "object": "chat.completion.chunk", + "created": 1700000000, + "model": "fixture-chat", + } + with respx.mock(using="httpx") as transport: + transport.post().mock( + return_value=httpx.Response( + 200, + headers={"content-type": "text/event-stream"}, + text=sse( + {**chunk, "choices": [{"index": 0, "delta": {"role": "assistant", "content": "hello"}}]}, + {**chunk, "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]}, + { + **chunk, + "choices": [], + "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15, **GO_REPORTED_USAGE}, + }, + ), + ) + ) + stream: Final = await litellm.acompletion( + model="opencode_go/fixture-chat", + api_key="fake-go-key", + messages=[{"role": "user", "content": "hello"}], + stream=True, + stream_options={"include_usage": True}, + ) + chunks: Final = [piece async for piece in stream] + assert "".join(piece.choices[0].delta.content or "" for piece in chunks if piece.choices) == "hello" + assert await logged_cost(cost_recorder) == pytest.approx(0.00412) + + +@pytest.mark.asyncio +async def test_responses_stream_bills_the_completed_events_list_cost( + go_catalog: None, cost_recorder: CostRecorder +) -> None: + completed: Final = go_priced_reply("responses", "fixture-responses", GO_REPORTED_USAGE) + with respx.mock(using="httpx") as transport: + transport.post().mock( + return_value=httpx.Response( + 200, + headers={"content-type": "text/event-stream"}, + text=sse( + { + "type": "response.output_text.delta", + "delta": "hello", + "item_id": "msg_test", + "output_index": 0, + "content_index": 0, + "sequence_number": 1, + }, + {"type": "response.completed", "response": completed, "sequence_number": 2}, + ), + ) + ) + stream: Final = await litellm.aresponses( + model="opencode_go/fixture-responses", api_key="fake-go-key", input="hello", stream=True + ) + events: Final = [event async for event in stream] + assert events[-1].type == "response.completed" + assert events[-1].response._hidden_params["additional_headers"][RESPONSE_COST_HEADER] == pytest.approx(0.00412) + assert await logged_cost(cost_recorder) == pytest.approx(0.00412) + + +def test_stream_chunk_builder_reads_list_cost_from_dict_chunks(): + from litellm.litellm_core_utils.litellm_logging import Logging + + logging_obj = Logging( + model="opencode_go/glm-5.3", + messages=[{"role": "user", "content": "hi"}], + stream=True, + call_type="completion", + start_time=datetime.now(), + litellm_call_id="dict-chunks", + function_id="1", + ) + logging_obj.model_call_details["custom_llm_provider"] = "opencode_go" + chunk = { + "id": "chatcmpl-go", + "object": "chat.completion.chunk", + "created": 1, + "model": "glm-5.3", + "choices": [{"index": 0, "delta": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}], + } + final_usage = {**chunk, "choices": [], "usage": dict(GO_REPORTED_USAGE)} + + assembled = litellm.stream_chunk_builder(chunks=[chunk, final_usage], logging_obj=logging_obj) + + assert assembled is not None + assert assembled._hidden_params["response_cost"] == GO_REPORTED_USAGE["opencode"]["list_cost"] diff --git a/tests/unit/proxy/pass_through_endpoints/llm_provider_handlers/test_anthropic_passthrough_logging_handler.py b/tests/unit/proxy/pass_through_endpoints/llm_provider_handlers/test_anthropic_passthrough_logging_handler.py index cc701e1ebd4..7530703a05a 100644 --- a/tests/unit/proxy/pass_through_endpoints/llm_provider_handlers/test_anthropic_passthrough_logging_handler.py +++ b/tests/unit/proxy/pass_through_endpoints/llm_provider_handlers/test_anthropic_passthrough_logging_handler.py @@ -145,9 +145,7 @@ class TestAnthropicLoggingHandlerModelFallback: '{"type": "message_stop"}', ] - def _create_mock_logging_obj( - self, model_in_details: str = None - ) -> LiteLLMLoggingObj: + def _create_mock_logging_obj(self, model_in_details: str = None) -> LiteLLMLoggingObj: """Create a mock logging object with optional model in model_call_details""" mock_logging_obj = MagicMock() @@ -166,21 +164,13 @@ class TestAnthropicLoggingHandlerModelFallback: mock_handler = MagicMock() return mock_handler - @patch.object( - AnthropicPassthroughLoggingHandler, "build_complete_streaming_response" - ) - @patch.object( - AnthropicPassthroughLoggingHandler, "_create_anthropic_response_logging_payload" - ) - def test_model_from_request_body_used_when_present( - self, mock_create_payload, mock_build_response - ): + @patch.object(AnthropicPassthroughLoggingHandler, "build_complete_streaming_response") + @patch.object(AnthropicPassthroughLoggingHandler, "_create_anthropic_response_logging_payload") + def test_model_from_request_body_used_when_present(self, mock_create_payload, mock_build_response): """Test that model from request_body is used when present""" # Arrange request_body = {"model": "claude-3-sonnet-20240229"} - logging_obj = self._create_mock_logging_obj( - model_in_details="claude-3-haiku-20240307" - ) + logging_obj = self._create_mock_logging_obj(model_in_details="claude-3-haiku-20240307") passthrough_handler = self._create_mock_passthrough_handler() # Mock successful response building @@ -204,57 +194,37 @@ class TestAnthropicLoggingHandlerModelFallback: # Verify that _build_complete_streaming_response was called with the request_body model mock_build_response.assert_called_once() call_args = mock_build_response.call_args - assert ( - call_args[1]["model"] == "claude-3-sonnet-20240229" - ) # Should use request_body model + assert call_args[1]["model"] == "claude-3-sonnet-20240229" # Should use request_body model def test_model_fallback_logic_isolated(self): """Test just the model fallback logic in isolation""" # Test case 1: Model from request body request_body = {"model": "claude-3-sonnet-20240229"} - logging_obj = self._create_mock_logging_obj( - model_in_details="claude-3-haiku-20240307" - ) + logging_obj = self._create_mock_logging_obj(model_in_details="claude-3-haiku-20240307") # Extract the logic directly from the function model = request_body.get("model", "") - if ( - not model - and hasattr(logging_obj, "model_call_details") - and logging_obj.model_call_details.get("model") - ): + if not model and hasattr(logging_obj, "model_call_details") and logging_obj.model_call_details.get("model"): model = logging_obj.model_call_details.get("model") assert model == "claude-3-sonnet-20240229" # Should use request_body model # Test case 2: Fallback to logging obj request_body = {} - logging_obj = self._create_mock_logging_obj( - model_in_details="claude-3-haiku-20240307" - ) + logging_obj = self._create_mock_logging_obj(model_in_details="claude-3-haiku-20240307") model = request_body.get("model", "") - if ( - not model - and hasattr(logging_obj, "model_call_details") - and logging_obj.model_call_details.get("model") - ): + if not model and hasattr(logging_obj, "model_call_details") and logging_obj.model_call_details.get("model"): model = logging_obj.model_call_details.get("model") assert model == "claude-3-haiku-20240307" # Should use fallback model # Test case 3: Empty string in request body, fallback to logging obj request_body = {"model": ""} - logging_obj = self._create_mock_logging_obj( - model_in_details="claude-3-opus-20240229" - ) + logging_obj = self._create_mock_logging_obj(model_in_details="claude-3-opus-20240229") model = request_body.get("model", "") - if ( - not model - and hasattr(logging_obj, "model_call_details") - and logging_obj.model_call_details.get("model") - ): + if not model and hasattr(logging_obj, "model_call_details") and logging_obj.model_call_details.get("model"): model = logging_obj.model_call_details.get("model") assert model == "claude-3-opus-20240229" # Should use fallback model @@ -264,11 +234,7 @@ class TestAnthropicLoggingHandlerModelFallback: logging_obj = self._create_mock_logging_obj() model = request_body.get("model", "") - if ( - not model - and hasattr(logging_obj, "model_call_details") - and logging_obj.model_call_details.get("model") - ): + if not model and hasattr(logging_obj, "model_call_details") and logging_obj.model_call_details.get("model"): model = logging_obj.model_call_details.get("model") assert model == "" # Should be empty @@ -284,11 +250,7 @@ class TestAnthropicLoggingHandlerModelFallback: # Extract the logic directly from the function model = request_body.get("model", "") - if ( - not model - and hasattr(logging_obj, "model_call_details") - and logging_obj.model_call_details.get("model") - ): + if not model and hasattr(logging_obj, "model_call_details") and logging_obj.model_call_details.get("model"): model = logging_obj.model_call_details.get("model") assert model == "" # Should remain empty since no fallback available @@ -298,11 +260,7 @@ class TestAnthropicLoggingHandlerModelFallback: logging_obj = self._create_mock_logging_obj() # Empty dict model = request_body.get("model", "") - if ( - not model - and hasattr(logging_obj, "model_call_details") - and logging_obj.model_call_details.get("model") - ): + if not model and hasattr(logging_obj, "model_call_details") and logging_obj.model_call_details.get("model"): model = logging_obj.model_call_details.get("model") assert model == "" # Should remain empty @@ -311,9 +269,7 @@ class TestAnthropicLoggingHandlerModelFallback: class TestAzureAnthropicCostCalculation: """Test the custom_llm_provider cost calculation logic for Azure AI Anthropic.""" - def _create_mock_logging_obj( - self, model: str = None, custom_llm_provider: str = None - ) -> LiteLLMLoggingObj: + def _create_mock_logging_obj(self, model: str = None, custom_llm_provider: str = None) -> LiteLLMLoggingObj: """Create a mock logging object with optional model and custom_llm_provider""" mock_logging_obj = MagicMock() mock_model_call_details = {} @@ -326,9 +282,7 @@ class TestAzureAnthropicCostCalculation: return mock_logging_obj @patch("litellm.completion_cost") - def test_cost_calculation_with_azure_ai_custom_llm_provider( - self, mock_completion_cost - ): + def test_cost_calculation_with_azure_ai_custom_llm_provider(self, mock_completion_cost): """Test that custom_llm_provider is passed to completion_cost for Azure AI Anthropic""" from litellm.types.utils import ModelResponse from datetime import datetime @@ -397,9 +351,7 @@ class TestAzureAnthropicCostCalculation: assert call_kwargs["custom_llm_provider"] is None @patch("litellm.completion_cost") - def test_cost_calculation_does_not_duplicate_provider_prefix( - self, mock_completion_cost - ): + def test_cost_calculation_does_not_duplicate_provider_prefix(self, mock_completion_cost): """Test that provider prefix is not duplicated if already present in model name""" from litellm.types.utils import ModelResponse from datetime import datetime @@ -436,9 +388,7 @@ class TestAzureAnthropicCostCalculation: assert call_kwargs["custom_llm_provider"] == "azure_ai" @patch("litellm.completion_cost") - def test_cost_calculation_resolves_unknown_model_from_litellm_params( - self, mock_completion_cost - ): + def test_cost_calculation_resolves_unknown_model_from_litellm_params(self, mock_completion_cost): """When the body model is the "unknown" sentinel, the deployment model from litellm_params must be used for costing, not "unknown" (which makes completion_cost raise and the cost silently fall back to $0).""" @@ -451,9 +401,7 @@ class TestAzureAnthropicCostCalculation: logging_obj = self._create_mock_logging_obj(model="unknown") logging_obj.model_call_details["litellm_params"] = { "model": "anthropic/claude-3-5-haiku-20241022", - "metadata": { - "model_group": "passthrough/anthropic/claude-3-5-haiku-20241022" - }, + "metadata": {"model_group": "passthrough/anthropic/claude-3-5-haiku-20241022"}, } logging_obj.litellm_params = logging_obj.model_call_details["litellm_params"] @@ -471,17 +419,12 @@ class TestAzureAnthropicCostCalculation: ) mock_completion_cost.assert_called_once() - assert ( - mock_completion_cost.call_args[1]["model"] - == "anthropic/claude-3-5-haiku-20241022" - ) + assert mock_completion_cost.call_args[1]["model"] == "anthropic/claude-3-5-haiku-20241022" assert kwargs["response_cost"] == 0.001 assert kwargs["model"] == "anthropic/claude-3-5-haiku-20241022" @patch("litellm.completion_cost") - def test_cost_calculation_resolves_unknown_model_from_model_group( - self, mock_completion_cost - ): + def test_cost_calculation_resolves_unknown_model_from_model_group(self, mock_completion_cost): """With only model_group available (no deployment litellm_params.model), the leading passthrough/ prefix must be stripped so the cost map can resolve the model.""" @@ -493,9 +436,7 @@ class TestAzureAnthropicCostCalculation: logging_obj = self._create_mock_logging_obj(model="unknown") logging_obj.model_call_details["litellm_params"] = { - "metadata": { - "model_group": "passthrough/anthropic/claude-3-5-haiku-20241022" - } + "metadata": {"model_group": "passthrough/anthropic/claude-3-5-haiku-20241022"} } logging_obj.litellm_params = logging_obj.model_call_details["litellm_params"] @@ -513,16 +454,11 @@ class TestAzureAnthropicCostCalculation: ) mock_completion_cost.assert_called_once() - assert ( - mock_completion_cost.call_args[1]["model"] - == "anthropic/claude-3-5-haiku-20241022" - ) + assert mock_completion_cost.call_args[1]["model"] == "anthropic/claude-3-5-haiku-20241022" assert kwargs["response_cost"] == 0.002 @patch("litellm.completion_cost") - def test_cost_calculation_skips_unknown_litellm_params_model_for_model_group( - self, mock_completion_cost - ): + def test_cost_calculation_skips_unknown_litellm_params_model_for_model_group(self, mock_completion_cost): """When litellm_params.model is itself the "unknown" sentinel, the deployment-model branch must not short-circuit; resolution falls through to model_group so costing still prices the real model instead of "unknown".""" @@ -535,9 +471,7 @@ class TestAzureAnthropicCostCalculation: logging_obj = self._create_mock_logging_obj(model="unknown") logging_obj.model_call_details["litellm_params"] = { "model": "unknown", - "metadata": { - "model_group": "passthrough/anthropic/claude-3-5-haiku-20241022" - }, + "metadata": {"model_group": "passthrough/anthropic/claude-3-5-haiku-20241022"}, } logging_obj.litellm_params = logging_obj.model_call_details["litellm_params"] @@ -555,17 +489,12 @@ class TestAzureAnthropicCostCalculation: ) mock_completion_cost.assert_called_once() - assert ( - mock_completion_cost.call_args[1]["model"] - == "anthropic/claude-3-5-haiku-20241022" - ) + assert mock_completion_cost.call_args[1]["model"] == "anthropic/claude-3-5-haiku-20241022" assert kwargs["response_cost"] == 0.003 assert kwargs["model"] == "anthropic/claude-3-5-haiku-20241022" @patch("litellm.completion_cost") - def test_streaming_cost_calculation_resolves_model_from_message_start_chunk( - self, mock_completion_cost - ): + def test_streaming_cost_calculation_resolves_model_from_message_start_chunk(self, mock_completion_cost): """On the bare /anthropic passthrough path litellm_params carries no model or model_group and the body model is the "unknown" sentinel; the model must be recovered from the message_start SSE event so completion_cost @@ -628,9 +557,7 @@ class TestAzureAnthropicCostCalculation: ), _sse("message_stop", {"type": "message_stop"}), ] - all_chunks = list( - PassThroughStreamingHandler._convert_raw_bytes_to_str_lines(frames) - ) + all_chunks = list(PassThroughStreamingHandler._convert_raw_bytes_to_str_lines(frames)) logging_obj = RealLoggingObj( model="unknown", @@ -674,9 +601,7 @@ class TestAzureAnthropicCostCalculation: ] assert ( - AnthropicPassthroughLoggingHandler._extract_model_from_anthropic_chunks( - chunks - ) + AnthropicPassthroughLoggingHandler._extract_model_from_anthropic_chunks(chunks) == "claude-3-5-haiku-20241022" ) @@ -692,14 +617,11 @@ class TestAzureAnthropicCostCalculation: ) assert ( - AnthropicPassthroughLoggingHandler._extract_model_from_anthropic_chunks( - [raw_event] - ) + AnthropicPassthroughLoggingHandler._extract_model_from_anthropic_chunks([raw_event]) == "claude-3-5-haiku-20241022" ) - class TestAnthropicBatchPassthroughCostTracking: """Test cases for Anthropic batch passthrough cost tracking functionality""" @@ -799,9 +721,7 @@ class TestAnthropicBatchPassthroughCostTracking: ) mock_batches_config_instance = MagicMock() - mock_batches_config_instance.transform_retrieve_batch_response.return_value = ( - mock_batch_response - ) + mock_batches_config_instance.transform_retrieve_batch_response.return_value = mock_batch_response mock_batches_config.return_value = mock_batches_config_instance # Test the handler @@ -929,9 +849,7 @@ class TestAnthropicBatchPassthroughCostTracking: "transform_retrieve_batch_response", return_value=mock_batch_response, ): - with patch.object( - AnthropicPassthroughLoggingHandler, "_store_batch_managed_object" - ): + with patch.object(AnthropicPassthroughLoggingHandler, "_store_batch_managed_object"): result = AnthropicPassthroughLoggingHandler.batch_creation_handler( httpx_response=mock_httpx_response, logging_obj=mock_logging_obj, @@ -946,10 +864,7 @@ class TestAnthropicBatchPassthroughCostTracking: # Verify unified_object_id contains anthropic/ prefix unified_object_id = result["kwargs"]["unified_object_id"] decoded = base64.urlsafe_b64decode(unified_object_id + "==").decode() - assert ( - "anthropic/claude-sonnet-4-5-20250929" in decoded - or "claude-sonnet-4-5-20250929" in decoded - ) + assert "anthropic/claude-sonnet-4-5-20250929" in decoded or "claude-sonnet-4-5-20250929" in decoded @pytest.mark.parametrize( "kwargs,expected_user_id,expected_team_id", @@ -1007,9 +922,7 @@ class TestAnthropicBatchPassthroughCostTracking: mock_managed_files_hook.store_unified_object_id = AsyncMock() with ( patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_pl, - patch( - "litellm.proxy.pass_through_endpoints.llm_provider_handlers.batch_attribution.verbose_proxy_logger" - ), + patch("litellm.proxy.pass_through_endpoints.llm_provider_handlers.batch_attribution.verbose_proxy_logger"), ): mock_pl.get_proxy_hook.return_value = mock_managed_files_hook AnthropicPassthroughLoggingHandler._store_batch_managed_object( @@ -1060,9 +973,7 @@ class TestAnthropicBatchPassthroughCostTracking: is never tracked. The failure has to reach the log instead of being reported as a success.""" mock_managed_files_hook = MagicMock() - mock_managed_files_hook.store_unified_object_id = AsyncMock( - side_effect=RuntimeError("db down") - ) + mock_managed_files_hook.store_unified_object_id = AsyncMock(side_effect=RuntimeError("db down")) with ( patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_pl, patch( @@ -1104,9 +1015,7 @@ class TestAnthropicBatchPassthroughCostTracking: this gate an id-scoped route reached the store with a mismatched id, where it could only either claim a row it did not create or fail the model_object_id unique constraint.""" - with patch.object( - AnthropicPassthroughLoggingHandler, "_store_batch_managed_object" - ) as mock_store: + with patch.object(AnthropicPassthroughLoggingHandler, "_store_batch_managed_object") as mock_store: AnthropicPassthroughLoggingHandler.batch_creation_handler( httpx_response=mock_httpx_response, logging_obj=mock_logging_obj, @@ -1120,9 +1029,7 @@ class TestAnthropicBatchPassthroughCostTracking: assert mock_store.call_count == (1 if registers else 0) - def test_batch_creation_handler_failure_status_code( - self, mock_logging_obj, mock_request_body - ): + def test_batch_creation_handler_failure_status_code(self, mock_logging_obj, mock_request_body): """Test batch creation handler with non-200 status code""" mock_response = MagicMock() mock_response.status_code = 400 @@ -1145,9 +1052,7 @@ class TestAnthropicBatchPassthroughCostTracking: assert result["kwargs"]["response_cost"] == 0.0 @patch("litellm.proxy.proxy_server.proxy_logging_obj") - def test_store_batch_managed_object_success( - self, mock_proxy_logging_obj, mock_logging_obj - ): + def test_store_batch_managed_object_success(self, mock_proxy_logging_obj, mock_logging_obj): """Test storing batch managed object""" from litellm.types.utils import LiteLLMBatch @@ -1178,9 +1083,7 @@ class TestAnthropicBatchPassthroughCostTracking: ) # Verify managed files hook was called - mock_proxy_logging_obj.get_proxy_hook.assert_called_once_with( - "managed_files" - ) + mock_proxy_logging_obj.get_proxy_hook.assert_called_once_with("managed_files") class TestBuildCompleteStreamingResponseRobustness: @@ -1343,11 +1246,7 @@ class TestPureTextFastPathParity: }, ) ) - frames.append( - self._sse( - "content_block_stop", {"type": "content_block_stop", "index": b} - ) - ) + frames.append(self._sse("content_block_stop", {"type": "content_block_stop", "index": b})) frames.append( self._sse( "message_delta", @@ -1405,9 +1304,7 @@ class TestPureTextFastPathParity: # And the full logged payload (sans non-deterministic response id). def _scrub(p): d = dict(p) - r = d.get("complete_streaming_response_in_db") or d.get( - "complete_streaming_response" - ) + r = d.get("complete_streaming_response_in_db") or d.get("complete_streaming_response") return d, getattr(r, "usage", None) assert _scrub(k_fast)[1] == _scrub(k_legacy)[1] @@ -1419,11 +1316,7 @@ class TestPureTextFastPathParity: self._assert_parity(self._text_stream(["Just one piece of text."])) def test_parity_cache_tokens(self): - self._assert_parity( - self._text_stream( - ["a", "b", "c"], input_tokens=20, cache_creation=5, cache_read=7 - ) - ) + self._assert_parity(self._text_stream(["a", "b", "c"], input_tokens=20, cache_creation=5, cache_read=7)) def test_parity_max_tokens_stop(self): self._assert_parity(self._text_stream(["tok"] * 8, stop_reason="max_tokens")) @@ -1484,12 +1377,7 @@ class TestPureTextFastPathParity: self._sse("message_stop", {"type": "message_stop"}), ] all_chunks = self._to_all_chunks(frames) - assert ( - AnthropicPassthroughLoggingHandler._collapse_pure_text_chunks( - list(all_chunks) - ) - is None - ) + assert AnthropicPassthroughLoggingHandler._collapse_pure_text_chunks(list(all_chunks)) is None def test_collapse_returns_none_for_thinking(self): frames = [ @@ -1526,19 +1414,12 @@ class TestPureTextFastPathParity: self._sse("message_stop", {"type": "message_stop"}), ] all_chunks = self._to_all_chunks(frames) - assert ( - AnthropicPassthroughLoggingHandler._collapse_pure_text_chunks( - list(all_chunks) - ) - is None - ) + assert AnthropicPassthroughLoggingHandler._collapse_pure_text_chunks(list(all_chunks)) is None def test_collapse_actually_shrinks_chunk_count(self): frames = self._text_stream(["a"] * 50) all_chunks = list(self._to_all_chunks(frames)) - collapsed = AnthropicPassthroughLoggingHandler._collapse_pure_text_chunks( - all_chunks - ) + collapsed = AnthropicPassthroughLoggingHandler._collapse_pure_text_chunks(all_chunks) assert collapsed is not None # 50 text deltas + 50 event markers + 1 ping collapse to far fewer. assert len(collapsed) < len(all_chunks) / 2 @@ -1604,10 +1485,7 @@ class TestPureTextFastPathParity: self._sse("message_stop", {"type": "message_stop"}), ] all_chunks = list(self._to_all_chunks(frames)) - assert ( - AnthropicPassthroughLoggingHandler._collapse_pure_text_chunks(all_chunks) - is None - ) + assert AnthropicPassthroughLoggingHandler._collapse_pure_text_chunks(all_chunks) is None class TestInterruptedStreamOutputTokenRecovery: @@ -1722,9 +1600,7 @@ class TestInterruptedStreamOutputTokenRecovery: import litellm placeholder = 2 - result = self._run( - self._interrupted_chunks(placeholder_output_tokens=placeholder) - ) + result = self._run(self._interrupted_chunks(placeholder_output_tokens=placeholder)) usage = result["result"].usage expected = litellm.token_counter( @@ -1897,9 +1773,7 @@ class TestStreamFalseDeduplication: "delta": {"type": "text_delta", "text": "Hello"}, }, ), - TestStreamFalseDeduplication._sse( - "content_block_stop", {"type": "content_block_stop", "index": 0} - ), + TestStreamFalseDeduplication._sse("content_block_stop", {"type": "content_block_stop", "index": 0}), TestStreamFalseDeduplication._sse( "message_delta", { @@ -1949,15 +1823,12 @@ class TestStreamFalseDeduplication: # The assembled response must be stored on model_call_details so callbacks # can identify this as a completed streaming call, not an in-progress one. - assert ( - logging_obj.model_call_details.get("complete_streaming_response") - is not None - ), "complete_streaming_response must be set on model_call_details after assembly" + assert logging_obj.model_call_details.get("complete_streaming_response") is not None, ( + "complete_streaming_response must be set on model_call_details after assembly" + ) # The returned result must match what was stored - assert result["result"] is logging_obj.model_call_details.get( - "complete_streaming_response" - ) + assert result["result"] is logging_obj.model_call_details.get("complete_streaming_response") def test_dedup_guard_fires_when_stream_true_on_logging_obj(self): """ @@ -1986,21 +1857,16 @@ class TestStreamFalseDeduplication: assert logging_obj._is_assembled_stream_success(result=mock_response) is True # First dispatch sets the flag - assert not logging_obj.model_call_details.get( - "has_dispatched_final_stream_success" - ) + assert not logging_obj.model_call_details.get("has_dispatched_final_stream_success") logging_obj.model_call_details["has_dispatched_final_stream_success"] = True # Second dispatch would be blocked — simulate the guard check would_skip = bool( logging_obj._is_assembled_stream_success(result=mock_response) - and logging_obj.model_call_details.get( - "has_dispatched_final_stream_success" - ) + and logging_obj.model_call_details.get("has_dispatched_final_stream_success") ) assert would_skip is True, ( - "Dedup guard must block a second dispatch_success_handlers call for the " - "same assembled streaming response" + "Dedup guard must block a second dispatch_success_handlers call for the same assembled streaming response" ) def test_sse_fallback_path_sets_stream_true_for_dedup(self): @@ -2034,9 +1900,7 @@ class TestStreamFalseDeduplication: would_skip = bool( logging_obj._is_assembled_stream_success(result=mock_response) - and logging_obj.model_call_details.get( - "has_dispatched_final_stream_success" - ) + and logging_obj.model_call_details.get("has_dispatched_final_stream_success") ) assert would_skip is True @@ -2100,9 +1964,9 @@ class TestNonStreamingResponseRedaction: logging_obj=logging_obj, ) - assert ( - "complete_streaming_response" not in logging_obj.model_call_details - ), "non-streaming responses must not populate complete_streaming_response" + assert "complete_streaming_response" not in logging_obj.model_call_details, ( + "non-streaming responses must not populate complete_streaming_response" + ) def test_streaming_sets_complete_streaming_response(self): from litellm.types.utils import ModelResponse @@ -2119,10 +1983,7 @@ class TestNonStreamingResponseRedaction: logging_obj=logging_obj, ) - assert ( - logging_obj.model_call_details.get("complete_streaming_response") - is response - ) + assert logging_obj.model_call_details.get("complete_streaming_response") is response def test_non_streaming_response_is_redacted_when_message_logging_off(self): from litellm.litellm_core_utils.redact_messages import ( @@ -2196,10 +2057,8 @@ class TestAnthropicUsageOnlyFallback: ] def test_build_usage_only_recovers_cache_inclusive_usage(self): - response = ( - AnthropicPassthroughLoggingHandler._build_usage_only_response_from_chunks( - all_chunks=self._CHUNKS, model="claude-3-5-haiku-20241022" - ) + response = AnthropicPassthroughLoggingHandler._build_usage_only_response_from_chunks( + all_chunks=self._CHUNKS, model="claude-3-5-haiku-20241022" ) assert response is not None usage = response.usage @@ -2259,10 +2118,8 @@ class TestAnthropicUsageOnlyFallback: } ), ] - response = ( - AnthropicPassthroughLoggingHandler._build_usage_only_response_from_chunks( - all_chunks=chunks, model="unknown" - ) + response = AnthropicPassthroughLoggingHandler._build_usage_only_response_from_chunks( + all_chunks=chunks, model="unknown" ) assert response is not None assert response.model == "claude-opus-4-6" @@ -2287,9 +2144,7 @@ class TestAnthropicUsageOnlyFallback: ('data: {"a": 1}', {"a": 1}), ], ) - def test_extract_sse_data_handles_malformed_and_sentinel_lines( - self, event_str, expected - ): + def test_extract_sse_data_handles_malformed_and_sentinel_lines(self, event_str, expected): assert extract_sse_data(event_str) == expected def _real_logging_obj(self): @@ -2309,12 +2164,8 @@ class TestAnthropicUsageOnlyFallback: return logging_obj @patch("litellm.completion_cost") - @patch.object( - AnthropicPassthroughLoggingHandler, "build_complete_streaming_response" - ) - def test_handler_falls_back_when_assembly_returns_none( - self, mock_assemble, mock_cost - ): + @patch.object(AnthropicPassthroughLoggingHandler, "build_complete_streaming_response") + def test_handler_falls_back_when_assembly_returns_none(self, mock_assemble, mock_cost): mock_assemble.return_value = None mock_cost.return_value = 0.0021 logging_obj = self._real_logging_obj() @@ -2335,9 +2186,7 @@ class TestAnthropicUsageOnlyFallback: assert result["kwargs"]["response_cost"] == 0.0021 @patch("litellm.completion_cost") - @patch.object( - AnthropicPassthroughLoggingHandler, "build_complete_streaming_response" - ) + @patch.object(AnthropicPassthroughLoggingHandler, "build_complete_streaming_response") def test_handler_falls_back_when_assembly_raises(self, mock_assemble, mock_cost): import litellm @@ -2367,9 +2216,7 @@ class TestAnthropicUsageOnlyFallback: assert result["result"].usage.completion_tokens == 55 assert result["kwargs"]["response_cost"] == 0.0021 - @patch.object( - AnthropicPassthroughLoggingHandler, "build_complete_streaming_response" - ) + @patch.object(AnthropicPassthroughLoggingHandler, "build_complete_streaming_response") def test_handler_returns_none_when_no_usage_recoverable(self, mock_assemble): # assembly fails AND the chunks carry no usage event, so there is nothing # to price; the handler must return None rather than fabricate a response @@ -2391,15 +2238,9 @@ class TestAnthropicUsageOnlyFallback: assert result["result"] is None assert result["kwargs"] == {} - @patch.object( - AnthropicPassthroughLoggingHandler, "build_usage_only_response_from_chunks" - ) - @patch.object( - AnthropicPassthroughLoggingHandler, "build_complete_streaming_response" - ) - def test_handler_does_not_crash_when_usage_only_fallback_raises( - self, mock_assemble, mock_fallback - ): + @patch.object(AnthropicPassthroughLoggingHandler, "build_usage_only_response_from_chunks") + @patch.object(AnthropicPassthroughLoggingHandler, "build_complete_streaming_response") + def test_handler_does_not_crash_when_usage_only_fallback_raises(self, mock_assemble, mock_fallback): # if the usage-only fallback itself raises, it must be treated as None and # drop gracefully, not propagate out and crash the success handler mock_assemble.return_value = None @@ -2427,7 +2268,6 @@ class TestAnthropicResponseCostRecordedOnModelCallDetails: builder must record it there or streaming pass-through logs $0.""" - class TestAnthropicPassthroughFastMode: """Anthropic charges a provider-specific multiplier for ``speed=fast``, applied off ``usage.speed`` and covering every token type, cache included. The response usage @@ -2436,20 +2276,20 @@ class TestAnthropicPassthroughFastMode: MODEL = "claude-opus-4-8" STREAM_CHUNKS = [ - 'event: message_start', + "event: message_start", 'data: {"type": "message_start", "message": {"id": "msg_1", "type": "message", "role": "assistant",' ' "model": "claude-opus-4-8", "content": [], "stop_reason": null,' ' "usage": {"input_tokens": 1000, "cache_read_input_tokens": 200, "output_tokens": 0}}}', - 'event: content_block_start', + "event: content_block_start", 'data: {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}', - 'event: content_block_delta', + "event: content_block_delta", 'data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "ok"}}', - 'event: content_block_stop', + "event: content_block_stop", 'data: {"type": "content_block_stop", "index": 0}', - 'event: message_delta', + "event: message_delta", 'data: {"type": "message_delta", "delta": {"stop_reason": "end_turn"},' ' "usage": {"input_tokens": 1000, "cache_read_input_tokens": 200, "output_tokens": 100}}', - 'event: message_stop', + "event: message_stop", 'data: {"type": "message_stop"}', ] @@ -2643,7 +2483,6 @@ class TestRecordPartialUsageForFailure: assert "response_cost" not in logging_obj.model_call_details - @pytest.mark.parametrize( ("all_chunks", "interrupted"), [ @@ -2664,9 +2503,7 @@ def test_stream_was_interrupted_skips_data_lines_that_are_not_json_objects( @pytest.mark.asyncio -async def test_anthropic_passthrough_handler( - mock_httpx_response, mock_response, mock_logging_obj -): +async def test_anthropic_passthrough_handler(mock_httpx_response, mock_response, mock_logging_obj): """ Unit test - Assert that the anthropic passthrough handler calls the litellm logging object's async_success_handler """ @@ -3007,3 +2844,102 @@ def test_streamed_messages_end_turn_logs_no_stop_details(mock_logging_obj): assert logged.choices[0].finish_reason == "stop" provider_fields: Final = logged.choices[0].message.provider_specific_fields or {} assert "stop_details" not in provider_fields + + +def _go_sse(event: str, data: dict) -> str: + return f"event: {event}\ndata: {json.dumps(data)}\n\n" + + +_GO_MESSAGE_DELTA_USAGE: Final = { + "input_tokens": 1200, + "output_tokens": 300, + "cost": 0, + "opencode": {"billing_source": "go", "list_cost": 0.00412}, +} + + +def _go_stream_chunks() -> list[str]: + message = { + "id": "msg_go", + "type": "message", + "role": "assistant", + "model": "claude-haiku-5-5", + "content": [], + "stop_reason": None, + "stop_sequence": None, + "usage": {"input_tokens": 1200, "output_tokens": 0}, + } + return [ + _go_sse("message_start", {"type": "message_start", "message": message}), + _go_sse( + "content_block_start", + {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}, + ), + _go_sse( + "content_block_delta", + {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "ok"}}, + ), + _go_sse("content_block_stop", {"type": "content_block_stop", "index": 0}), + _go_sse( + "message_delta", + { + "type": "message_delta", + "delta": {"stop_reason": "end_turn", "stop_sequence": None}, + "usage": _GO_MESSAGE_DELTA_USAGE, + }, + ), + _go_sse("message_stop", {"type": "message_stop"}), + ] + + +def _go_logging_obj() -> LiteLLMLoggingObj: + logging_obj = LiteLLMLoggingObj( + model="opencode_go/claude-haiku-5-5", + messages=[{"role": "user", "content": "hi"}], + stream=True, + call_type="anthropic_messages", + start_time=datetime.now(), + litellm_call_id="go-stream", + function_id="1", + ) + logging_obj.model_call_details["custom_llm_provider"] = "opencode_go" + logging_obj.model_call_details["stream"] = True + logging_obj.model_call_details["litellm_params"] = {} + logging_obj.litellm_params = {} + return logging_obj + + +def test_streamed_messages_bill_the_list_cost_reported_in_the_final_message_delta(): + result = AnthropicPassthroughLoggingHandler._handle_logging_anthropic_collected_chunks( + litellm_logging_obj=_go_logging_obj(), + passthrough_success_handler_obj=MagicMock(), + url_route="/v1/messages", + request_body={"model": "opencode_go/claude-haiku-5-5", "stream": True}, + endpoint_type="messages", + start_time=datetime.now(), + all_chunks=_go_stream_chunks(), + end_time=datetime.now(), + ) + + assert result["kwargs"]["response_cost"] == 0.00412 + assert result["result"]._hidden_params["response_cost"] == 0.00412 + + +def test_streamed_messages_from_other_providers_ignore_a_zero_usage_cost_field(): + logging_obj = _go_logging_obj() + logging_obj.model_call_details["custom_llm_provider"] = "anthropic" + + result = AnthropicPassthroughLoggingHandler._handle_logging_anthropic_collected_chunks( + litellm_logging_obj=logging_obj, + passthrough_success_handler_obj=MagicMock(), + url_route="/v1/messages", + request_body={"model": "claude-haiku-5-5", "stream": True}, + endpoint_type="messages", + start_time=datetime.now(), + all_chunks=_go_stream_chunks(), + end_time=datetime.now(), + ) + + headers = result["result"]._hidden_params.get("additional_headers") or {} + assert "llm_provider-x-litellm-response-cost" not in headers + assert result["kwargs"].get("response_cost") != 0.0