fix(anthropic): report bridged /v1/messages stream failures to the proxy failure hook

The Anthropic-to-Responses bridge turns an upstream failure into an
error event, so the proxy's streaming generator never sees an exception
and the post-call failure hook never runs: no failure spend row, no
budget release, no failure hook on custom loggers. The wrapper now hands
the unwrapped failure to the hook the proxy arms on the logging object
once per failed stream, right before the error event is popped, and a
hook that raises is logged instead of hiding the error from the client.

The partial-usage estimate on the Responses failure path keeps the same
private transform the success path already uses.

litellm/llms/anthropic/pass_through/responses_adapters/streaming_iterator.py

@@ -3,7 +3,7 @@
 import asyncio
 import json
 from collections import deque
-from collections.abc import AsyncIterator, Iterator, Mapping
+from collections.abc import AsyncIterator, Awaitable, Callable, Iterator, Mapping
 from typing import TYPE_CHECKING, Any, Final
 
 from pydantic import BaseModel, ConfigDict, field_validator
@@ -11,7 +11,7 @@ from pydantic import BaseModel, ConfigDict, field_validator
 from litellm import verbose_logger
 from litellm._logging import redact_internal_details_from_client_message
 from litellm._uuid import uuid
-from litellm.exceptions import MidStreamFallbackError
+from litellm.exceptions import APIError, MidStreamFallbackError
 from litellm.litellm_core_utils.prompt_templates.common_utils import (
     encrypted_reasoning_signature,
 )
@@ -122,9 +122,11 @@ class AnthropicResponsesStreamWrapper:
     ) -> None:
         self.responses_stream = responses_stream
         self.model = model
+        self.litellm_logging_obj = litellm_logging_obj
         self._message_id: str = f"msg_{uuid.uuid4()}"
         if litellm_logging_obj is not None:
             litellm_logging_obj.record_streamed_anthropic_message_id(self._message_id)
+        self._unreported_failure: Exception | None = None
         self._current_block_index: int = -1
         # Map item_id -> content_block_index so we can stop the right block later
         self._item_id_to_block_index: dict[str, int] = {}
@@ -430,30 +432,58 @@ class AnthropicResponsesStreamWrapper:
             self._sent_message_stop = True
             return
 
-    def _fail_stream(self, status_code: int, message: str) -> None:
+    def _fail_stream(self, status_code: int, message: str, failure: Exception | None = None) -> None:
         self._stream_failed = True
+        self._unreported_failure = failure or APIError(
+            status_code=status_code,
+            message=message,
+            llm_provider=getattr(self.responses_stream, "custom_llm_provider", None),
+            model=self.model,
+        )
         self._chunk_queue.append(_anthropic_error_chunk(status_code, message))
 
+    async def _pop_chunk(self) -> dict[str, object]:
+        if self._unreported_failure is not None:
+            failure: Final = self._unreported_failure
+            self._unreported_failure = None
+            await self._report_failure(failure)
+        return self._chunk_queue.popleft()
+
+    async def _report_failure(self, failure: Exception) -> None:
+        on_failure: Final[Callable[[Exception], Awaitable[None]] | None] = getattr(
+            self.litellm_logging_obj, "_on_detached_stream_failure", None
+        )
+        if on_failure is None:
+            return
+        try:
+            await on_failure(failure)
+        except Exception as hook_failure:  # noqa: BLE001  # a failing proxy hook must not hide the error event from the client
+            verbose_logger.warning(
+                "AnthropicResponsesStreamWrapper: stream failure hook raised: %s(%s)",
+                type(hook_failure).__name__,
+                hook_failure,
+            )
+
     def __aiter__(self) -> "AnthropicResponsesStreamWrapper":
         return self
 
     async def __anext__(self) -> dict[str, object]:
         if self._chunk_queue:
-            return self._chunk_queue.popleft()
+            return await self._pop_chunk()
         if self._stream_failed:
             raise StopAsyncIteration
 
         if not self._sent_message_start:
             self._sent_message_start = True
             self._chunk_queue.append(self._make_message_start())
-            return self._chunk_queue.popleft()
+            return await self._pop_chunk()
 
         try:
             if hasattr(self.responses_stream, "__aiter__"):
                 async for event in self.responses_stream:
                     self._process_event(event)
                     if self._chunk_queue:
-                        return self._chunk_queue.popleft()
+                        return await self._pop_chunk()
             else:
                 if self._sync_responses_iterator is None:
                     self._sync_responses_iterator = iter(self.responses_stream)
@@ -462,14 +492,14 @@ class AnthropicResponsesStreamWrapper:
                 while (event := await asyncio.to_thread(next, sync_iterator, missing)) is not missing:
                     self._process_event(event)
                     if self._chunk_queue:
-                        return self._chunk_queue.popleft()
+                        return await self._pop_chunk()
         except StopAsyncIteration:
             pass
         except Exception as e:  # noqa: BLE001  # every upstream failure becomes a client error event
             verbose_logger.exception(
                 "AnthropicResponsesStreamWrapper: upstream Responses stream for %s failed", self.model
             )
-            self._fail_stream(*_failure_status_and_message(e))
+            self._fail_stream(*_failure_status_and_message(e), failure=_original_failure(e))
 
         if not self._chunk_queue and not self._sent_message_stop and not self._stream_failed:
             verbose_logger.error(
@@ -479,7 +509,7 @@ class AnthropicResponsesStreamWrapper:
             self._fail_stream(500, INCOMPLETE_STREAM_ERROR_MESSAGE)
 
         if self._chunk_queue:
-            return self._chunk_queue.popleft()
+            return await self._pop_chunk()
 
         raise StopAsyncIteration
 

litellm/responses/streaming_iterator.py

@@ -892,7 +892,9 @@ class BaseResponsesAPIStreamingIterator:
         )
         if estimate is None:
             return
-        usage: Final = ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(estimate)
+        usage: Final = ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(  # pyright: ignore[reportPrivateUsage]  # same shared transform the success path above uses
+            estimate
+        )
         self.logging_obj.record_assembled_response_for_failure(ModelResponse(model=self.model, usage=usage))
 
     def _handle_failure(self, exception: Exception):

tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py

@@ -7,6 +7,8 @@ import asyncio
 import json
 import os
 import sys
+import time
+from collections.abc import Awaitable, Callable
 from types import SimpleNamespace
 
 import pytest
@@ -15,6 +17,7 @@ sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../.
 
 import litellm
 from litellm.exceptions import MidStreamFallbackError
+from litellm.litellm_core_utils.litellm_logging import Logging as LitellmLogging
 from litellm.litellm_core_utils.prompt_templates.common_utils import (
     encrypted_reasoning_signature,
 )
@@ -622,3 +625,98 @@ class TestUpstreamFailureEndsStreamWithErrorEvent:
         chunks = _collect(iter(()))
         assert [chunk["type"] for chunk in chunks] == ["message_start", "error"]
         assert chunks[1]["error"] == {"type": "api_error", "message": INCOMPLETE_STREAM_ERROR_MESSAGE}
+
+
+class _FailureHookRecorder:
+    def __init__(self) -> None:
+        self.received: tuple[Exception, ...] = ()
+
+    async def __call__(self, failure: Exception) -> None:
+        self.received = (*self.received, failure)
+
+
+def _logging_obj_with_failure_hook(on_failure: Callable[[Exception], Awaitable[None]]) -> LitellmLogging:
+    logging_obj = LitellmLogging(
+        model="m",
+        messages=[{"role": "user", "content": "hi"}],
+        stream=True,
+        call_type="anthropic_messages",
+        start_time=time.time(),
+        litellm_call_id="lit-6871-test",
+        function_id="lit-6871-test",
+    )
+    logging_obj._on_detached_stream_failure = on_failure
+    return logging_obj
+
+
+def _collect_with_logging(stream, logging_obj: LitellmLogging) -> list:
+    async def _run() -> list:
+        wrapper = AnthropicResponsesStreamWrapper(responses_stream=stream, model="m", litellm_logging_obj=logging_obj)
+        return [chunk async for chunk in wrapper]
+
+    return asyncio.run(_run())
+
+
+class TestUpstreamFailureReachesTheProxyFailureHook:
+    """The ``error`` event alone leaves the proxy unaware the stream failed, so the wrapper also hands
+    the failure to the hook the proxy arms on the logging object, the one that writes the failure
+    spend row and releases the budget reservation."""
+
+    def test_raised_upstream_failure_reaches_the_hook_unwrapped_before_the_error_event(self):
+        rate_limit = litellm.RateLimitError(message="You have no credits remaining.", llm_provider="openai", model="m")
+        wrapped = MidStreamFallbackError(
+            message=str(rate_limit),
+            model="m",
+            llm_provider="openai",
+            original_exception=rate_limit,
+            is_pre_first_chunk=False,
+        )
+        recorder = _FailureHookRecorder()
+
+        async def _gen():
+            yield {"type": "response.created"}
+            yield {"type": "response.output_item.added", "item": {"type": "message", "id": "msg_1"}}
+            yield {"type": "response.output_text.delta", "item_id": "msg_1", "delta": "Hi"}
+            raise wrapped
+
+        chunks = _collect_with_logging(_gen(), _logging_obj_with_failure_hook(recorder))
+        assert [chunk["type"] for chunk in chunks] == [
+            "message_start",
+            "content_block_start",
+            "content_block_delta",
+            "error",
+        ]
+        assert recorder.received == (rate_limit,)
+
+    def test_response_failed_event_reaches_the_hook_as_an_exception_carrying_the_upstream_status(self):
+        recorder = _FailureHookRecorder()
+        failed = SimpleNamespace(
+            status="failed",
+            output=[],
+            usage=None,
+            error={"code": "rate_limit_exceeded", "message": "Rate limit reached for gpt-5.5, try again in 20s."},
+        )
+
+        async def _gen():
+            yield {"type": "response.created"}
+            yield {"type": "response.failed", "response": failed}
+
+        chunks = _collect_with_logging(_gen(), _logging_obj_with_failure_hook(recorder))
+        assert chunks[-1]["type"] == "error"
+        (failure,) = recorder.received
+        assert failure.status_code == 429
+        assert "Rate limit reached for gpt-5.5, try again in 20s." in failure.message
+
+    def test_a_raising_hook_never_hides_the_error_event_from_the_client(self):
+        async def _exploding_hook(failure: Exception) -> None:
+            raise RuntimeError("spend writer is down")
+
+        logging_obj = _logging_obj_with_failure_hook(_exploding_hook)
+
+        async def _gen():
+            yield {"type": "response.created"}
+            raise ConnectionResetError("Response payload is not completed")
+
+        chunks = _collect_with_logging(_gen(), logging_obj)
+        assert [chunk["type"] for chunk in chunks] == ["message_start", "error"]
+        assert chunks[-1]["error"] == {"type": "api_error", "message": "Response payload is not completed"}

tests/unit/responses/test_responses_streaming_iterator.py

@@ -164,14 +164,14 @@ async def test_sync_callbacks_run_only_after_async_handler_completes(recording_e
 class FailureRecorder(CustomLogger):
     def __init__(self):
         super().__init__()
-        self.failure_payloads: list = []
-        self.success_payloads: list = []
+        self.failure_payloads: tuple = ()
+        self.success_payloads: tuple = ()
 
     async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
-        self.failure_payloads.append(kwargs["standard_logging_object"])
+        self.failure_payloads = (*self.failure_payloads, kwargs["standard_logging_object"])
 
     async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
-        self.success_payloads.append(kwargs["standard_logging_object"])
+        self.success_payloads = (*self.success_payloads, kwargs["standard_logging_object"])
 
 
 class _UpstreamThatTimesOutAfterThreeDeltas:
@@ -230,4 +230,4 @@ async def test_mid_stream_read_timeout_logs_failure_with_partial_usage_and_no_su
     assert payload["prompt_tokens"] > 0
     assert payload["completion_tokens"] > 0
     assert payload["response_cost"] > 0
-    assert recorder.success_payloads == []
+    assert recorder.success_payloads == ()