From c35fef27e82f718e412d5a375b0ecd0be90c68f4 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Thu, 1 Oct 2026 16:27:21 -0700 Subject: [PATCH] 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. --- .../responses_adapters/streaming_iterator.py | 48 +++++++-- litellm/responses/streaming_iterator.py | 4 +- ...t_responses_adapters_streaming_iterator.py | 98 +++++++++++++++++++ .../test_responses_streaming_iterator.py | 10 +- 4 files changed, 145 insertions(+), 15 deletions(-) diff --git a/litellm/llms/anthropic/pass_through/responses_adapters/streaming_iterator.py b/litellm/llms/anthropic/pass_through/responses_adapters/streaming_iterator.py index cc6f4da3403..03b3ceb6653 100644 --- a/litellm/llms/anthropic/pass_through/responses_adapters/streaming_iterator.py +++ b/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 diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index 3522a43e0de..321bfd546e7 100644 --- a/litellm/responses/streaming_iterator.py +++ b/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): diff --git a/tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py b/tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py index e1dded214bf..227d2966e53 100644 --- a/tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py +++ b/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"} diff --git a/tests/unit/responses/test_responses_streaming_iterator.py b/tests/unit/responses/test_responses_streaming_iterator.py index 4aace64aaaa..16b125e56d4 100644 --- a/tests/unit/responses/test_responses_streaming_iterator.py +++ b/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 == ()