From e1ace00ca0b58bfd6a901c0ff48aadda7fec2607 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 15 Jul 2026 05:12:29 +0000 Subject: [PATCH 1/3] fix(anthropic-adapter): surface mid-stream provider errors as Anthropic error events Co-Authored-By: Ishaan Jaffer <155045088+ishaan-berri@users.noreply.github.com> --- .../adapters/streaming_iterator.py | 45 +++++- ...est_streaming_iterator_mid_stream_error.py | 142 ++++++++++++++++++ 2 files changed, 179 insertions(+), 8 deletions(-) create mode 100644 tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_mid_stream_error.py diff --git a/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py b/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py index f02333c34c8..6eb9ebfd702 100644 --- a/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py +++ b/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py @@ -16,10 +16,12 @@ from typing import ( get_args, ) +import openai from typing_extensions import assert_never from litellm._logging import verbose_logger from litellm._uuid import uuid +from litellm.llms.base_llm.chat.transformation import BaseLLMException from litellm.types.llms.anthropic import ( AppliedEdit, CompactionBlock, @@ -37,6 +39,25 @@ if TYPE_CHECKING: _STREAMING_DELTA_TYPES = frozenset(get_args(StreamingContentBlockDeltaType)) +def _error_status_and_message(exc: Exception) -> tuple[int, str]: + if isinstance(exc, (BaseLLMException, openai.APIStatusError)): + return exc.status_code, exc.message + return 500, str(exc) or "Upstream stream ended before completion" + + +def _mid_stream_error_sse_event(exc: Exception) -> bytes: + from litellm.anthropic_interface.exceptions.exception_mapping_utils import ( + AnthropicExceptionMapping, + ) + + status_code, message = _error_status_and_message(exc) + error_response = AnthropicExceptionMapping.transform_to_anthropic_error( + status_code=status_code, + raw_message=message, + ) + return f"event: error\ndata: {json.dumps(error_response)}\n\n".encode() + + def _delta_payload_field(delta_type: StreamingContentBlockDeltaType) -> str: match delta_type: case "text_delta": @@ -823,15 +844,23 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): """ Async version of anthropic_sse_wrapper. Convert AnthropicStreamWrapper dict chunks to Server-Sent Events format. + + A failure raised while iterating the upstream stream is surfaced as an + Anthropic ``error`` event so the SSE stream stays well-formed instead of + the connection being torn down. """ - async for chunk in self: - if isinstance(chunk, dict): - event_type: str = str(chunk.get("type", "message")) - payload = f"event: {event_type}\ndata: {json.dumps(chunk)}\n\n" - yield payload.encode() - else: - # For non-dict chunks, forward the original value unchanged - yield chunk + try: + async for chunk in self: + if isinstance(chunk, dict): + event_type: str = str(chunk.get("type", "message")) + payload = f"event: {event_type}\ndata: {json.dumps(chunk)}\n\n" + yield payload.encode() + else: + # For non-dict chunks, forward the original value unchanged + yield chunk + except Exception as e: # noqa: BLE001 # boundary before the socket: any upstream failure becomes an Anthropic error event + verbose_logger.exception("Anthropic Adapter - mid-stream error, emitting Anthropic error event: %s", e) + yield _mid_stream_error_sse_event(e) def _increment_content_block_index(self): self.current_content_block_index += 1 diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_mid_stream_error.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_mid_stream_error.py new file mode 100644 index 00000000000..45ec18733f7 --- /dev/null +++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_mid_stream_error.py @@ -0,0 +1,142 @@ +""" +Regression tests for the ``/v1/messages`` async adapter dropping the socket on a +mid-stream provider error. + +When a non-Anthropic model (e.g. Bedrock Converse) is served through +``/v1/messages``, the proxy hands Starlette the async SSE iterator directly. If +the upstream provider stream raises while being pulled (Bedrock raises +``BedrockError`` when a ConverseStream ends without a terminal ``messageStop`` +event, common on cross-region inference profiles), the exception escaped the +request handler's try/except and tore down the connection. Clients like Claude +Code then showed a bare "Connection closed mid-response". + +The async SSE wrapper must instead surface the failure as a well-formed +Anthropic ``error`` event so the stream stays valid and the client can retry. +""" + +import json +import os +import sys +from typing import List, Optional +from unittest.mock import MagicMock + +import pytest + +sys.path.insert(0, os.path.abspath("../../../../..")) + +from litellm.exceptions import MidStreamFallbackError +from litellm.llms.anthropic.experimental_pass_through.adapters.streaming_iterator import ( + AnthropicStreamWrapper, + _mid_stream_error_sse_event, +) +from litellm.llms.bedrock.common_utils import BedrockError +from litellm.types.utils import Delta, StreamingChoices + + +def _make_chunk(delta: Delta, finish_reason: Optional[str] = None) -> MagicMock: + chunk = MagicMock() + chunk.choices = [ + StreamingChoices(finish_reason=finish_reason, index=0, delta=delta, logprobs=None) + ] + chunk.usage = None + chunk._hidden_params = {} + return chunk + + +class _AsyncStreamThenRaise: + """Yields the given chunks, then raises ``exc`` (mimics a provider stream + that terminates mid-response).""" + + def __init__(self, items: List[MagicMock], exc: BaseException): + self._it = iter(items) + self._exc = exc + + def __aiter__(self): + return self + + async def __anext__(self): + try: + return next(self._it) + except StopIteration: + raise self._exc + + +def _parse_sse(raw: bytes) -> tuple[str, dict]: + text = raw.decode() + event_line, data_line = text.strip().split("\n", 1) + return event_line.removeprefix("event: "), json.loads(data_line.removeprefix("data: ")) + + +async def _drain_sse(wrapper: AnthropicStreamWrapper) -> List[bytes]: + return [event async for event in wrapper.async_anthropic_sse_wrapper()] + + +@pytest.mark.asyncio +async def test_mid_stream_bedrock_error_becomes_anthropic_error_event(): + """A ``BedrockError`` raised after partial content must be surfaced as a + terminal Anthropic ``error`` event, not propagated (which drops the socket + and yields "Connection closed mid-response").""" + chunks = [_make_chunk(Delta(content="Creating a file"))] + bedrock_err = BedrockError( + status_code=500, + message="Bedrock ConverseStream ended without a terminal 'messageStop' event", + ) + wrapper = AnthropicStreamWrapper( + completion_stream=_AsyncStreamThenRaise(chunks, bedrock_err), + model="bedrock-converse-sonnet-4-6", + ) + + events = await _drain_sse(wrapper) + + parsed = [_parse_sse(e) for e in events] + event_types = [name for name, _ in parsed] + assert "message_start" in event_types + assert event_types[-1] == "error" + _, error_payload = parsed[-1] + assert error_payload["type"] == "error" + assert error_payload["error"]["type"] == "api_error" + assert "messageStop" in error_payload["error"]["message"] + + +@pytest.mark.asyncio +async def test_mid_stream_error_does_not_raise_out_of_wrapper(): + """The async wrapper must fully drain without letting the upstream exception + escape — escaping is exactly what tore down the connection before the fix.""" + wrapper = AnthropicStreamWrapper( + completion_stream=_AsyncStreamThenRaise([], BedrockError(status_code=500, message="boom")), + model="claude-x", + ) + events = await _drain_sse(wrapper) + assert _parse_sse(events[-1])[0] == "error" + + +@pytest.mark.parametrize( + "status_code, expected_type", + [(500, "api_error"), (529, "overloaded_error"), (429, "rate_limit_error")], +) +def test_error_event_maps_status_code_to_anthropic_type(status_code, expected_type): + raw = _mid_stream_error_sse_event(BedrockError(status_code=status_code, message="upstream failed")) + name, payload = _parse_sse(raw) + assert name == "error" + assert payload["error"]["type"] == expected_type + assert payload["error"]["message"] == "upstream failed" + + +def test_error_event_defaults_to_500_when_status_missing(): + raw = _mid_stream_error_sse_event(ValueError("no status here")) + _, payload = _parse_sse(raw) + assert payload["error"]["type"] == "api_error" + assert payload["error"]["message"] == "no status here" + + +def test_error_event_preserves_midstream_fallback_error(): + exc = MidStreamFallbackError( + message="BedrockException - internalServerException", + model="bedrock-converse-sonnet-4-6", + llm_provider="bedrock", + original_exception=BedrockError(status_code=500, message="internalServerException"), + ) + name, payload = _parse_sse(_mid_stream_error_sse_event(exc)) + assert name == "error" + assert payload["error"]["type"] == "api_error" + assert "internalServerException" in payload["error"]["message"] From 9a5987b055cf1c69ff02a436c1daf663b96d39ba Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 15 Jul 2026 05:25:54 +0000 Subject: [PATCH 2/3] style(anthropic-adapter): drop added comments per repo convention Co-Authored-By: Ishaan Jaffer <155045088+ishaan-berri@users.noreply.github.com> --- .../experimental_pass_through/adapters/streaming_iterator.py | 5 ----- 1 file changed, 5 deletions(-) diff --git a/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py b/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py index 6eb9ebfd702..2eaa19c0f0c 100644 --- a/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py +++ b/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py @@ -844,10 +844,6 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): """ Async version of anthropic_sse_wrapper. Convert AnthropicStreamWrapper dict chunks to Server-Sent Events format. - - A failure raised while iterating the upstream stream is surfaced as an - Anthropic ``error`` event so the SSE stream stays well-formed instead of - the connection being torn down. """ try: async for chunk in self: @@ -856,7 +852,6 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): payload = f"event: {event_type}\ndata: {json.dumps(chunk)}\n\n" yield payload.encode() else: - # For non-dict chunks, forward the original value unchanged yield chunk except Exception as e: # noqa: BLE001 # boundary before the socket: any upstream failure becomes an Anthropic error event verbose_logger.exception("Anthropic Adapter - mid-stream error, emitting Anthropic error event: %s", e) From 3a9fb724eb1d6e005d7fcf48c369e0076cfaf4d7 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 15 Jul 2026 14:57:17 +0000 Subject: [PATCH 3/3] refactor(anthropic-adapter): use MidStreamFallbackError instead of importing openai Co-Authored-By: Ishaan Jaffer <155045088+ishaan-berri@users.noreply.github.com> --- .../experimental_pass_through/adapters/streaming_iterator.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py b/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py index 2eaa19c0f0c..dbc4703b51b 100644 --- a/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py +++ b/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py @@ -16,11 +16,11 @@ from typing import ( get_args, ) -import openai from typing_extensions import assert_never from litellm._logging import verbose_logger from litellm._uuid import uuid +from litellm.exceptions import MidStreamFallbackError from litellm.llms.base_llm.chat.transformation import BaseLLMException from litellm.types.llms.anthropic import ( AppliedEdit, @@ -40,7 +40,7 @@ _STREAMING_DELTA_TYPES = frozenset(get_args(StreamingContentBlockDeltaType)) def _error_status_and_message(exc: Exception) -> tuple[int, str]: - if isinstance(exc, (BaseLLMException, openai.APIStatusError)): + if isinstance(exc, (BaseLLMException, MidStreamFallbackError)): return exc.status_code, exc.message return 500, str(exc) or "Upstream stream ended before completion"