fix(anthropic): report a bridged stream failure only once the client has taken the error event

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
 

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 == ()