From d2fbaff2c948e5beb0fe02623640e1fc9c21cf69 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Tue, 18 Aug 2026 14:05:28 -0700 Subject: [PATCH] fix(proxy): record estimated input tokens in spend logs for dispatched failed requests Failure rows in the spend log only carried token counts when a broken stream stashed recovered partial usage; non-stream requests that reached the provider and then failed (timeouts, provider 4xx/5xx) logged 0/0/0 even though the provider billed the input tokens. Estimate the input side in post_call_failure_hook with the same tokenizer fallback interrupted streams use, gated to requests that were actually dispatched (first_api_call_start_time set and no litellm_no_upstream_llm_call marker), and pin response_cost to 0.0 so failed requests never bill spend. Recovered partial-stream usage still wins over the estimate. --- litellm/proxy/utils.py | 67 +++++++-- tests/test_litellm/proxy/test_proxy_utils.py | 139 +++++++++++++++++++ 2 files changed, 197 insertions(+), 9 deletions(-) diff --git a/litellm/proxy/utils.py b/litellm/proxy/utils.py index 2ad7180bd5f..3457ae0f352 100644 --- a/litellm/proxy/utils.py +++ b/litellm/proxy/utils.py @@ -40,7 +40,7 @@ from litellm.proxy._types import ( from litellm.proxy.spend_tracking.spend_log_error_logger import spend_log_error from litellm.types.guardrails import GuardrailEventHooks from litellm.types.proxy.model_listing import ModelInfoResponse -from litellm.types.utils import CallTypes, CallTypesLiteral, ModelInfo +from litellm.types.utils import CallTypes, CallTypesLiteral, ModelInfo, Usage try: from litellm_enterprise.enterprise_callbacks.send_emails.base_email import ( @@ -403,6 +403,52 @@ def _exception_changes_request_flow(exc: BaseException) -> bool: return isinstance(exc, (SensitiveDataRouteException, ModifyResponseException)) +def _count_request_input_tokens(model: str, request_input: object) -> int: + if isinstance(request_input, str): + return litellm.token_counter(model=model, text=request_input) + if not isinstance(request_input, list) or not request_input: + return 0 + text_entries: Final = tuple(entry for entry in request_input if isinstance(entry, str)) + if len(text_entries) == len(request_input): + return litellm.token_counter(model=model, text="".join(text_entries)) + return litellm.token_counter(model=model, messages=request_input) + + +def _estimate_dispatched_failure_usage(model: str, request_input: object) -> Usage | None: + """A request that failed after dispatch consumed provider-billed input + tokens, but no provider usage ever came back. Estimate the input side with + the same tokenizer fallback interrupted streams use, so the spend log's + failure row records what was sent instead of zero.""" + try: + input_tokens: Final = _count_request_input_tokens(model=model, request_input=request_input) + except Exception: + return None + if input_tokens <= 0: + return None + return Usage(prompt_tokens=input_tokens, completion_tokens=0, total_tokens=input_tokens) + + +def _failure_usage_to_lift(model_call_details: Mapping[str, object], dispatched: bool) -> tuple[object, object] | None: + """A stream that broke mid-flight still billed the provider for the chunks + already delivered; the streaming handler stashes that recovered usage and + cost in model_call_details, so prefer it. Otherwise a request that was + dispatched to a provider and failed without upstream usage gets an + estimated input-side Usage with zero cost. Returns the + (combined_usage_object, response_cost) pair to lift, or None.""" + recovered_usage: Final = model_call_details.get("combined_usage_object") + if recovered_usage is not None: + return recovered_usage, model_call_details.get("response_cost") + if not dispatched or model_call_details.get(LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL): + return None + estimated_usage: Final = _estimate_dispatched_failure_usage( + model=str(model_call_details.get("model") or ""), + request_input=model_call_details.get("messages"), + ) + if estimated_usage is None: + return None + return estimated_usage, 0.0 + + @dataclass(frozen=True) class _CallbackCapabilities: """Cached per-hook capability flags derived from ``litellm.callbacks``. @@ -2190,15 +2236,18 @@ class ProxyLogging: if _first_handoff is not None: request_data["first_api_call_start_time"] = _first_handoff - # A stream that broke mid-flight still billed the provider for the - # chunks already delivered; the streaming handler stashes that - # recovered usage and cost here. Lift them onto request_data so the + # Lift recovered partial-stream usage, or an estimated input-side + # usage for a dispatched failure, onto request_data so the # failure-path spend callbacks (which run after the logging object - # is popped) record the real partial spend instead of zero. - _recovered_usage: Final = _model_call_details.get("combined_usage_object") - if _recovered_usage is not None: - request_data["combined_usage_object"] = _recovered_usage - request_data["response_cost"] = _model_call_details.get("response_cost") + # is popped) record real token counts instead of zero. + _usage_to_lift: Final = _failure_usage_to_lift( + model_call_details=_model_call_details, + dispatched=_first_handoff is not None, + ) + if _usage_to_lift is not None: + _lifted_usage, _lifted_cost = _usage_to_lift + request_data["combined_usage_object"] = _lifted_usage + request_data["response_cost"] = _lifted_cost # Remove before callbacks iterate — not serialisable request_data.pop("litellm_logging_obj", None) diff --git a/tests/test_litellm/proxy/test_proxy_utils.py b/tests/test_litellm/proxy/test_proxy_utils.py index 1504c3c3103..70baf157ab9 100644 --- a/tests/test_litellm/proxy/test_proxy_utils.py +++ b/tests/test_litellm/proxy/test_proxy_utils.py @@ -478,6 +478,145 @@ class TestPostCallFailureHookLiftsRecoveredPartialSpend: assert "response_cost" not in request_data +class TestPostCallFailureHookEstimatesDispatchedInputTokens: + """A non-stream request that failed after dispatch (timeout, provider + error) consumed provider-billed input tokens but recovered no usage. + post_call_failure_hook must estimate the input side onto request_data so + the spend log's failure row records what was sent instead of zero, while + never charging spend for the failure (LIT-5690). + """ + + async def _run(self, request_data): + from unittest.mock import AsyncMock, patch + + from litellm.proxy._types import UserAPIKeyAuth + + proxy_logging_obj = ProxyLogging(user_api_key_cache=DualCache()) + proxy_logging_obj.alert_types = [] + with patch.object(proxy_logging_obj, "update_request_status", new=AsyncMock()): + await proxy_logging_obj.post_call_failure_hook( + request_data=request_data, + original_exception=Exception("boom"), + user_api_key_dict=UserAPIKeyAuth(), + ) + + def _logging_obj(self, model_call_details): + logging_obj = MagicMock() + logging_obj.model_call_details = model_call_details + return logging_obj + + @pytest.mark.asyncio + async def test_dispatched_failure_estimates_input_tokens_with_zero_cost(self): + from datetime import datetime + + from litellm.types.utils import Usage + + request_data = { + "litellm_logging_obj": self._logging_obj( + { + "first_api_call_start_time": datetime.now(), + "model": "gpt-3.5-turbo", + "messages": [{"role": "user", "content": "count these input tokens please"}], + } + ), + "metadata": {}, + "response_cost": 123.0, + } + await self._run(request_data) + + estimated = request_data["combined_usage_object"] + assert isinstance(estimated, Usage) + assert estimated.prompt_tokens > 0 + assert estimated.completion_tokens == 0 + assert estimated.total_tokens == estimated.prompt_tokens + assert request_data["response_cost"] == 0.0 + + @pytest.mark.asyncio + async def test_failure_before_dispatch_stays_zero(self): + request_data = { + "litellm_logging_obj": self._logging_obj( + { + "model": "gpt-3.5-turbo", + "messages": [{"role": "user", "content": "never dispatched"}], + } + ), + "metadata": {}, + } + await self._run(request_data) + + assert "combined_usage_object" not in request_data + assert "response_cost" not in request_data + + @pytest.mark.asyncio + async def test_proxy_only_error_never_dispatched_stays_zero(self): + from datetime import datetime + + from litellm.constants import LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL + + request_data = { + "litellm_logging_obj": self._logging_obj( + { + "first_api_call_start_time": datetime.now(), + "model": "no-such-model", + "messages": [{"role": "user", "content": "hi"}], + LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL: True, + } + ), + "metadata": {}, + } + await self._run(request_data) + + assert "combined_usage_object" not in request_data + assert "response_cost" not in request_data + + @pytest.mark.asyncio + async def test_recovered_partial_usage_wins_over_estimate(self): + from datetime import datetime + + from litellm.types.utils import Usage + + recovered_usage = Usage(prompt_tokens=30, completion_tokens=7, total_tokens=37) + request_data = { + "litellm_logging_obj": self._logging_obj( + { + "first_api_call_start_time": datetime.now(), + "model": "gpt-3.5-turbo", + "messages": [{"role": "user", "content": "mid-stream failure"}], + "combined_usage_object": recovered_usage, + "response_cost": 3.5e-05, + } + ), + "metadata": {}, + } + await self._run(request_data) + + assert request_data["combined_usage_object"] is recovered_usage + assert request_data["response_cost"] == 3.5e-05 + + @pytest.mark.asyncio + async def test_dispatched_failure_with_text_completion_prompt(self): + from datetime import datetime + + from litellm.types.utils import Usage + + request_data = { + "litellm_logging_obj": self._logging_obj( + { + "first_api_call_start_time": datetime.now(), + "model": "gpt-3.5-turbo", + "messages": "a plain text-completion prompt string", + } + ), + "metadata": {}, + } + await self._run(request_data) + + estimated = request_data["combined_usage_object"] + assert isinstance(estimated, Usage) + assert estimated.prompt_tokens > 0 + assert estimated.completion_tokens == 0 + + from typing import cast import litellm