From ab6d649aa05ccfa6e8043f2c31879b5f14677f5d Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Thu, 1 Oct 2026 17:23:49 -0700 Subject: [PATCH] fix(anthropic): report a bridged stream failure only once the client has taken the error event --- .../responses_adapters/streaming_iterator.py | 24 ++++----- ...t_responses_adapters_streaming_iterator.py | 52 ++++++++++++++++++- 2 files changed, 62 insertions(+), 14 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 03b3ceb6653..797f0f2a908 100644 --- a/litellm/llms/anthropic/pass_through/responses_adapters/streaming_iterator.py +++ b/litellm/llms/anthropic/pass_through/responses_adapters/streaming_iterator.py @@ -442,14 +442,11 @@ class AnthropicResponsesStreamWrapper: ) 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: + async def _report_failure_once_the_error_event_was_consumed(self) -> None: + failure: Final = self._unreported_failure + if failure is None: + return + self._unreported_failure = None on_failure: Final[Callable[[Exception], Awaitable[None]] | None] = getattr( self.litellm_logging_obj, "_on_detached_stream_failure", None ) @@ -469,21 +466,22 @@ class AnthropicResponsesStreamWrapper: async def __anext__(self) -> dict[str, object]: if self._chunk_queue: - return await self._pop_chunk() + return self._chunk_queue.popleft() if self._stream_failed: + await self._report_failure_once_the_error_event_was_consumed() raise StopAsyncIteration if not self._sent_message_start: self._sent_message_start = True self._chunk_queue.append(self._make_message_start()) - return await self._pop_chunk() + return self._chunk_queue.popleft() try: if hasattr(self.responses_stream, "__aiter__"): async for event in self.responses_stream: self._process_event(event) if self._chunk_queue: - return await self._pop_chunk() + return self._chunk_queue.popleft() else: if self._sync_responses_iterator is None: self._sync_responses_iterator = iter(self.responses_stream) @@ -492,7 +490,7 @@ class AnthropicResponsesStreamWrapper: while (event := await asyncio.to_thread(next, sync_iterator, missing)) is not missing: self._process_event(event) if self._chunk_queue: - return await self._pop_chunk() + return self._chunk_queue.popleft() except StopAsyncIteration: pass except Exception as e: # noqa: BLE001 # every upstream failure becomes a client error event @@ -509,7 +507,7 @@ class AnthropicResponsesStreamWrapper: self._fail_stream(500, INCOMPLETE_STREAM_ERROR_MESSAGE) if self._chunk_queue: - return await self._pop_chunk() + return self._chunk_queue.popleft() raise StopAsyncIteration 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 47b9dca0dac..1619c71990b 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 @@ -657,8 +657,18 @@ def _collect_with_logging(stream, logging_obj: LitellmLogging) -> list: return asyncio.run(_run()) +def _failing_after_one_delta(): + 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 litellm.RateLimitError(message="You have no credits remaining.", llm_provider="openai", model="m") + + return _gen() + + class TestUpstreamFailureReachesTheProxyFailureHook: - def test_raised_upstream_failure_reaches_the_hook_unwrapped_before_the_error_event(self): + def test_raised_upstream_failure_reaches_the_hook_unwrapped_once_the_error_event_was_consumed(self): rate_limit = litellm.RateLimitError(message="You have no credits remaining.", llm_provider="openai", model="m") wrapped = MidStreamFallbackError( message=str(rate_limit), @@ -716,3 +726,43 @@ class TestUpstreamFailureReachesTheProxyFailureHook: 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"} + + def test_the_hook_fires_only_after_the_consumer_took_the_error_event(self): + delivered: list[dict] = [] + delivered_when_hook_fired: list[int] = [] + + async def _hook(failure: Exception) -> None: + delivered_when_hook_fired.append(len(delivered)) + + async def _run() -> None: + wrapper = AnthropicResponsesStreamWrapper( + responses_stream=_failing_after_one_delta(), model="m", litellm_logging_obj=_logging_obj_with_failure_hook(_hook) + ) + async for chunk in wrapper: + delivered.append(chunk) + with pytest.raises(StopAsyncIteration): + await wrapper.__anext__() + + asyncio.run(_run()) + assert [chunk["type"] for chunk in delivered] == ["message_start", "content_block_start", "content_block_delta", "error"] + assert delivered_when_hook_fired == [len(delivered)] + + def test_a_consumer_that_closes_the_stream_on_the_error_event_recovers_it_without_the_hook(self): + recorder = _FailureHookRecorder() + + async def _run() -> list[bytes]: + wrapper = AnthropicResponsesStreamWrapper( + responses_stream=_failing_after_one_delta(), model="m", litellm_logging_obj=_logging_obj_with_failure_hook(recorder) + ) + sse = wrapper.async_anthropic_sse_wrapper() + frames: list[bytes] = [] + async for frame in sse: + frames.append(frame) + if frame.startswith(b"event: error\n"): + await sse.aclose() + break + return frames + + frames = asyncio.run(_run()) + assert frames[-1].startswith(b"event: error\n") + assert recorder.received == ()