fix(anthropic): run the proxy failure hook when a detached /v1/messages stream fails

litellm/litellm_core_utils/litellm_logging.py

@@ -10,7 +10,7 @@ import subprocess
 import sys
 import time
 import traceback
-from collections.abc import Callable, Mapping, Sequence
+from collections.abc import Awaitable, Callable, Mapping, Sequence
 from datetime import datetime as dt_object
 from functools import lru_cache
 from types import MappingProxyType, TracebackType
@@ -576,6 +576,7 @@ class Logging(LiteLLMLoggingBaseClass):
         # enqueue closure here instead of firing it immediately.
         self._defer_async_logging: bool = False
         self._enqueue_deferred_logging: Callable[[], None] | None = None
+        self._on_detached_stream_failure: Callable[[Exception], Awaitable[None]] | None = None
 
     def set_response_timing_metrics(self, timing_metrics: Mapping[str, float]) -> None:
         """Keep ``_response_ms`` / ``litellm_overhead_time_ms`` for a result that has no ``_hidden_params``."""

litellm/llms/anthropic/experimental_pass_through/messages/streaming_iterator.py

@@ -176,6 +176,11 @@ def _try_claim_detached_drain_slot() -> bool:
     return True
 
 
+def _exception_left_unconsumed(queue: "asyncio.Queue[bytes | None | BaseException]", exc: BaseException) -> bool:
+    remaining: Final = tuple(queue.get_nowait() for _ in range(queue.qsize()))
+    return any(item is exc for item in remaining)
+
+
 def _sse_event(event_type: str, payload: Mapping[str, object]) -> bytes:
     return f"event: {event_type}\ndata: {json.dumps(payload)}\n\n".encode()
 
@@ -663,18 +668,16 @@ class BaseAnthropicMessagesStreamingIterator:
         collected_chunks: Sequence[bytes],
         exc: Exception,
     ) -> None:
-        """Forward a provider error to a still-connected client and log the request as failed.
+        """Log the request as failed with its partial usage, then make sure the proxy's failure hook runs once.
 
-        The relay re-raises the forwarded exception so the proxy's failure hook
-        keeps the provider status; the logging object's failure handlers fire
-        here either way, carrying the partial usage the provider already
-        billed, so a client that left before consuming the exception still
-        gets a failure row rather than a success one.
+        A still-connected client gets the original exception through the queue,
+        the relay re-raises it, and the proxy's own failure handling records the
+        failed spend. When the client already left, or leaves before consuming
+        the queued exception, that handling never runs, so the detached-failure
+        hook the proxy armed on the logging object fires here instead.
         """
         from litellm.proxy.pass_through_endpoints.streaming_handler import PassThroughStreamingHandler
 
-        if not client_detached.is_set():
-            await self._enqueue_for_client(queue, client_detached, exc)
         PassThroughStreamingHandler.schedule_stream_failure_logging(
             litellm_logging_obj=self.litellm_logging_obj,
             endpoint_type=EndpointType.ANTHROPIC,
@@ -682,3 +685,21 @@ class BaseAnthropicMessagesStreamingIterator:
             raw_bytes=collected_chunks,
             exception=exc,
         )
+        if not client_detached.is_set() and await self._enqueue_for_client(queue, client_detached, exc):
+            await client_detached.wait()
+            if not _exception_left_unconsumed(queue, exc):
+                return
+        await self._fire_detached_failure_hook(exc)
+
+    async def _fire_detached_failure_hook(self, exc: Exception) -> None:
+        from litellm._logging import verbose_proxy_logger
+
+        on_detached_failure: Final = getattr(self.litellm_logging_obj, "_on_detached_stream_failure", None)
+        if on_detached_failure is None:
+            return
+        try:
+            await on_detached_failure(exc)
+        except Exception as hook_failure:  # noqa: BLE001  # a failing proxy hook must not crash the detached pump
+            verbose_proxy_logger.warning(
+                "async_sse_wrapper detached failure hook raised: %s(%s)", type(hook_failure).__name__, hook_failure
+            )

litellm/proxy/common_request_processing.py

@@ -2520,6 +2520,11 @@ class ProxyBaseLLMRequestProcessing:
                     # This handles cases like websearch_interception agentic loop
                     # which returns a non-streaming dict even for streaming requests
                     if self._is_streaming_response(response):
+                        self._arm_detached_stream_failure_hook(
+                            logging_obj=logging_obj,
+                            user_api_key_dict=user_api_key_dict,
+                            proxy_logging_obj=proxy_logging_obj,
+                        )
                         selected_data_generator = ProxyBaseLLMRequestProcessing.async_sse_data_generator(
                             response=response,
                             user_api_key_dict=user_api_key_dict,
@@ -2875,6 +2880,34 @@ class ProxyBaseLLMRequestProcessing:
             ),
         )
 
+    def _arm_detached_stream_failure_hook(
+        self,
+        logging_obj: LiteLLMLoggingObj,
+        user_api_key_dict: "UserAPIKeyAuth",
+        proxy_logging_obj: ProxyLogging,
+    ) -> None:
+        """Let a stream that fails after the client left still reach ``post_call_failure_hook``.
+
+        The client-facing generator reports a mid-stream failure itself, but once
+        the client disconnects that generator is gone and the detached upstream
+        drain is the only code that sees the provider error. It fires this closure
+        so the failed spend is still written and the budget reservation released;
+        a replacement error the hook raises has no client left to reach.
+        """
+        request_data: Final = self.data
+
+        async def _on_detached_stream_failure(exc: Exception) -> None:
+            try:
+                await proxy_logging_obj.post_call_failure_hook(
+                    user_api_key_dict=user_api_key_dict,
+                    original_exception=exc,
+                    request_data=request_data,
+                )
+            except HTTPException:
+                return
+
+        logging_obj._on_detached_stream_failure = _on_detached_stream_failure
+
     def _is_streaming_response(self, response: Any) -> bool:
         """
         Check if the response object is actually a streaming response by inspecting its type.

tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_streaming_iterator.py

@@ -544,6 +544,25 @@ async def test_async_sse_wrapper_dispatches_deferred_logging_when_client_disconn
     await asyncio.wait_for(deferred_fired.wait(), timeout=5)
 
 
+class _DetachedFailureRecorder:
+    """Stands in for the closure the proxy arms so a detached-stream failure still reaches its failure hook."""
+
+    def __init__(self):
+        self.exceptions = []
+
+    async def __call__(self, exc: Exception) -> None:
+        self.exceptions.append(exc)
+
+
+async def _wait_for_detached_failure(recorder: _DetachedFailureRecorder) -> Exception:
+    for _ in range(200):
+        if recorder.exceptions:
+            await asyncio.sleep(0.02)
+            return recorder.exceptions[0]
+        await asyncio.sleep(0.01)
+    raise AssertionError("the detached failure hook never fired")
+
+
 class _ProviderStreamError(Exception):
     """Stand-in for a provider-specific streaming failure carrying a status code."""
 
@@ -573,6 +592,8 @@ async def test_async_sse_wrapper_reraises_upstream_error_to_connected_client():
         litellm_logging_obj=_make_logging_obj("test_reraises_upstream_error", recorder),
         request_body={},
     )
+    detached_hook = _DetachedFailureRecorder()
+    iterator.litellm_logging_obj._on_detached_stream_failure = detached_hook
 
     received = []
 
@@ -591,6 +612,8 @@ async def test_async_sse_wrapper_reraises_upstream_error_to_connected_client():
     assert iterator.logged_chunks == []
     assert failure_kwargs["standard_logging_object"]["status"] == "failure"
     assert failure_kwargs["standard_logging_object"]["prompt_tokens"] == 52
+    await asyncio.sleep(0.05)
+    assert detached_hook.exceptions == [], "the relay re-raised the error, so the proxy failure hook already ran"
 
 
 @pytest.mark.asyncio
@@ -614,6 +637,8 @@ async def test_async_sse_wrapper_logs_failure_on_upstream_error_after_disconnect
         litellm_logging_obj=_make_logging_obj("test_failure_logged_on_late_error", recorder),
         request_body={},
     )
+    detached_hook = _DetachedFailureRecorder()
+    iterator.litellm_logging_obj._on_detached_stream_failure = detached_hook
 
     gen = iterator.async_sse_wrapper(_gated_failing_stream())
     received = [await gen.__anext__(), await gen.__anext__()]
@@ -627,6 +652,8 @@ async def test_async_sse_wrapper_logs_failure_on_upstream_error_after_disconnect
     assert failure_kwargs["standard_logging_object"]["status"] == "failure"
     assert failure_kwargs["standard_logging_object"]["prompt_tokens"] == 52
     assert isinstance(failure_kwargs["exception"], _ProviderStreamError)
+    assert await _wait_for_detached_failure(detached_hook) is failure_kwargs["exception"]
+    assert len(detached_hook.exceptions) == 1
 
 
 @pytest.mark.asyncio
@@ -651,6 +678,8 @@ async def test_async_sse_wrapper_logs_failure_when_queued_error_is_never_consume
         litellm_logging_obj=_make_logging_obj("test_failure_logged_on_unconsumed_queued_error", recorder),
         request_body={},
     )
+    detached_hook = _DetachedFailureRecorder()
+    iterator.litellm_logging_obj._on_detached_stream_failure = detached_hook
 
     gen = iterator.async_sse_wrapper(_failing_stream())
     received = [await gen.__anext__(), await gen.__anext__()]
@@ -663,6 +692,8 @@ async def test_async_sse_wrapper_logs_failure_when_queued_error_is_never_consume
     assert iterator.logging_call_count == 0
     assert failure_kwargs["standard_logging_object"]["status"] == "failure"
     assert failure_kwargs["standard_logging_object"]["prompt_tokens"] == 52
+    assert await _wait_for_detached_failure(detached_hook) is failure_kwargs["exception"]
+    assert len(detached_hook.exceptions) == 1
 
 
 @pytest.mark.asyncio

tests/test_litellm/proxy/test_common_request_processing.py

@@ -7836,3 +7836,112 @@ def test_log_llm_api_exception_traceback_only_for_unexpected_errors(exc, expect_
     records = [r for r in caplog.records if "_handle_llm_api_exception(): Exception occured" in r.getMessage()]
     assert len(records) == 1
     assert (records[0].exc_info is not None) is expect_traceback
+
+
+class _FailureHookRecorder:
+    """Stands in for ProxyLogging.post_call_failure_hook, recording what the detached-failure closure hands it."""
+
+    def __init__(self, raises: Optional[Exception] = None):
+        self.calls = []
+        self._raises = raises
+
+    async def post_call_failure_hook(self, **kwargs):
+        self.calls.append(kwargs)
+        if self._raises is not None:
+            raise self._raises
+
+
+class TestDetachedStreamFailureHook:
+    """
+    Regression for LIT-3798. A streaming /v1/messages request whose client disconnected
+    before the provider failed mid-stream never reached the proxy's failure hook: the
+    client-facing generator was gone, and the detached upstream drain only fired the
+    logging object's callbacks, so no failure spend row was written and the budget
+    reservation stayed held. base_process_llm_request now arms a closure on the logging
+    object that the detached drain awaits, and that closure runs post_call_failure_hook
+    with the request's key and data.
+    """
+
+    @staticmethod
+    def _logging_obj():
+        logging_obj = MagicMock()
+        logging_obj.litellm_call_id = "call-lit3798"
+        logging_obj.model_call_details = {}
+        logging_obj._enqueue_deferred_logging = None
+        logging_obj._on_deferred_stream_complete = None
+        logging_obj._on_detached_stream_failure = None
+        return logging_obj
+
+    @staticmethod
+    def _proxy_logging_obj(recorder: _FailureHookRecorder):
+        proxy_logging_obj = MagicMock(spec=ProxyLogging)
+        proxy_logging_obj.during_call_hook = AsyncMock(return_value=None)
+        proxy_logging_obj.update_request_status = AsyncMock(return_value=None)
+        proxy_logging_obj.post_call_response_headers_hook = AsyncMock(return_value={})
+        proxy_logging_obj.post_call_failure_hook = recorder.post_call_failure_hook
+        return proxy_logging_obj
+
+    @pytest.mark.asyncio
+    async def test_streaming_messages_arms_the_detached_failure_hook(self, monkeypatch):
+        import litellm.proxy.common_request_processing as crp
+        from litellm.proxy._types import UserAPIKeyAuth as RealUserAPIKeyAuth
+
+        async def _stream():
+            yield b"event: message_start\n\n"
+
+        async def fake_route_request(**kwargs):
+            async def _llm_call():
+                return _stream()
+
+            return _llm_call()
+
+        monkeypatch.setattr(crp, "route_request", fake_route_request)
+        monkeypatch.setattr(litellm, "callbacks", [])
+        recorder = _FailureHookRecorder()
+        logging_obj = self._logging_obj()
+        user_api_key_dict = RealUserAPIKeyAuth(api_key="sk-test")
+        processing_obj = ProxyBaseLLMRequestProcessing(
+            data={"litellm_logging_obj": logging_obj, "model": "claude-sonnet-4-5"}
+        )
+
+        await processing_obj.base_process_llm_request(
+            request=MagicMock(spec=Request, headers={}),
+            fastapi_response=Response(),
+            user_api_key_dict=user_api_key_dict,
+            route_type="anthropic_messages",
+            proxy_logging_obj=self._proxy_logging_obj(recorder),
+            general_settings={},
+            proxy_config=MagicMock(spec=ProxyConfig),
+            select_data_generator=None,
+            llm_router=None,
+            skip_pre_call_logic=True,
+        )
+
+        failure = RuntimeError("upstream died after the client left")
+        await logging_obj._on_detached_stream_failure(failure)
+
+        assert recorder.calls == [
+            {
+                "user_api_key_dict": user_api_key_dict,
+                "original_exception": failure,
+                "request_data": processing_obj.data,
+            }
+        ]
+
+    @pytest.mark.asyncio
+    async def test_detached_failure_hook_drops_the_replacement_error_it_cannot_deliver(self):
+        from litellm.proxy._types import UserAPIKeyAuth as RealUserAPIKeyAuth
+
+        recorder = _FailureHookRecorder(raises=HTTPException(status_code=429, detail="budget exceeded"))
+        logging_obj = self._logging_obj()
+        processing_obj = ProxyBaseLLMRequestProcessing(data={"litellm_logging_obj": logging_obj})
+        processing_obj._arm_detached_stream_failure_hook(
+            logging_obj=logging_obj,
+            user_api_key_dict=RealUserAPIKeyAuth(api_key="sk-test"),
+            proxy_logging_obj=self._proxy_logging_obj(recorder),
+        )
+        failure = RuntimeError("upstream died after the client left")
+
+        await logging_obj._on_detached_stream_failure(failure)
+
+        assert [call["original_exception"] for call in recorder.calls] == [failure]