diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 463cbf7cdbe..d6f2387ac71 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/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``.""" diff --git a/litellm/llms/anthropic/experimental_pass_through/messages/streaming_iterator.py b/litellm/llms/anthropic/experimental_pass_through/messages/streaming_iterator.py index 275608fcccc..7d01aee5d98 100644 --- a/litellm/llms/anthropic/experimental_pass_through/messages/streaming_iterator.py +++ b/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 + ) diff --git a/litellm/proxy/common_request_processing.py b/litellm/proxy/common_request_processing.py index 6542842f5e4..f25fa46197e 100644 --- a/litellm/proxy/common_request_processing.py +++ b/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. diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_streaming_iterator.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_streaming_iterator.py index 3d41d0942e5..be33b2ee3b1 100644 --- a/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_streaming_iterator.py +++ b/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 diff --git a/tests/test_litellm/proxy/test_common_request_processing.py b/tests/test_litellm/proxy/test_common_request_processing.py index ea665b60b19..f7fe6ad9d39 100644 --- a/tests/test_litellm/proxy/test_common_request_processing.py +++ b/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]