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

@@ -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)

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