From 9f290d8b99b13d3920fcd015a89cc4c374c9ebfd Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Thu, 27 Aug 2026 15:29:50 -0700 Subject: [PATCH 1/3] fix(router): fall over on raised mid-stream errors in /v1/messages streams --- litellm/router.py | 101 +++++++++++++++++++++++--- tests/test_litellm/test_router.py | 114 ++++++++++++++++++++++++++++++ 2 files changed, 207 insertions(+), 8 deletions(-) diff --git a/litellm/router.py b/litellm/router.py index 021dafa9791..8175a5c5c40 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -420,6 +420,46 @@ def _anthropic_stream_should_decline_fallback(has_generated_content: bool, error return has_generated_content or not error.is_pre_first_chunk +def _anthropic_stream_raised_error_status(error: Exception) -> int | None: + raw_status: Final = getattr(error, "status_code", None) + if isinstance(raw_status, int): + return raw_status + if isinstance(raw_status, str) and raw_status.isdigit(): + return int(raw_status) + response_status: Final = getattr(getattr(error, "response", None), "status_code", None) + return response_status if isinstance(response_status, int) else None + + +def _anthropic_stream_fallback_error_for_raised( + error: Exception, model: str, llm_provider: str, has_generated_content: bool +) -> "MidStreamFallbackError | None": + """ + A provider iterator that fails mid-stream by raising (Bedrock surfaces + its event-stream exception frames as a BedrockError, a transport drop + raises httpx's error) never produces the Anthropic SSE `event: error` + frame the wrapper detects, so the raise is converted into the same + MidStreamFallbackError a detected error event gets, under the same gate: + only before real content reached the caller and only for a retriable + status (429, 5xx, or none at all for a transport failure), mirroring + CustomStreamWrapper._handle_stream_fallback_error on /chat/completions. + None means the exception propagates to the caller unchanged. + """ + from litellm.exceptions import MidStreamFallbackError + + if has_generated_content: + return None + status_code: Final = _anthropic_stream_raised_error_status(error) + if status_code is not None and not _is_retriable_anthropic_status(status_code): + return None + return MidStreamFallbackError( + message=str(error), + model=model, + llm_provider=llm_provider, + original_exception=error, + is_pre_first_chunk=True, + ) + + def _anthropic_stream_commits_now(chunk: object, has_generated_content: bool, buffered_chunk_count: int) -> bool: """ Whether `chunk` should make Router._aanthropic_messages_streaming_iterator @@ -5019,6 +5059,8 @@ class Router: has_generated_content = False # rebind-ok: set once real content is seen, or the buffer cap is hit buffered_lifecycle_chunks: tuple[bytes, ...] = () # rebind-ok: flushed once committed or on decline model: Final = cast(str, initial_kwargs.get("model")) # cast-ok: kwargs always carries the model group + custom_llm_provider: Final = initial_kwargs.get("custom_llm_provider") + llm_provider: Final = custom_llm_provider if isinstance(custom_llm_provider, str) else "anthropic" try: async for chunk in source_iterator: if _anthropic_stream_forwards_ping_live( @@ -5061,14 +5103,16 @@ class Router: yield chunk for buffered_chunk in buffered_lifecycle_chunks: yield buffered_chunk - except MidStreamFallbackError as e: - if _anthropic_stream_should_decline_fallback(has_generated_content, e): - for buffered_chunk in buffered_lifecycle_chunks: - yield buffered_chunk - if e.original_exception is not None: - raise e.original_exception from e - raise - async for item in self._aanthropic_messages_fallback_attempt(e, initial_kwargs, wrapper): + except Exception as stream_error: # noqa: BLE001 # any raised provider error must reach the fallback gate, like CustomStreamWrapper + async for item in self._aanthropic_messages_recover_stream_error( + stream_error, + has_generated_content, + buffered_lifecycle_chunks, + model, + llm_provider, + initial_kwargs, + wrapper, + ): yield item finally: with anyio.CancelScope(shield=True), contextlib.suppress(BaseException): @@ -5080,6 +5124,47 @@ class Router: wrapper: Final = FallbackAwareAnthropicMessagesStream(stream_with_fallbacks(), source_iterator) return wrapper + async def _aanthropic_messages_recover_stream_error( + self, + stream_error: Exception, + has_generated_content: bool, + buffered_lifecycle_chunks: tuple[bytes, ...], + model: str, + llm_provider: str, + initial_kwargs: dict[str, Any], # mutable-ok: handed to _aanthropic_messages_fallback_attempt, which mutates it + wrapper: "FallbackAwareAnthropicMessagesStream", + ) -> AsyncGenerator[bytes, None]: + """ + Decides what a source-iterator failure in + Router._aanthropic_messages_streaming_iterator turns into: a fallback + attempt, or the error reaching the caller. A MidStreamFallbackError + (completion-bridge path, or the wrapper's own SSE error-event + detection) is declined per _anthropic_stream_should_decline_fallback + with the held-back lifecycle frames flushed first; any other raise is + converted per _anthropic_stream_fallback_error_for_raised and, when + not convertible, propagates untouched so the caller still gets a + clean error response. + """ + from litellm.exceptions import MidStreamFallbackError + + if isinstance(stream_error, MidStreamFallbackError) and _anthropic_stream_should_decline_fallback( + has_generated_content, stream_error + ): + for buffered_chunk in buffered_lifecycle_chunks: + yield buffered_chunk + if stream_error.original_exception is not None: + raise stream_error.original_exception from stream_error + raise stream_error + fallback_error: Final = ( + stream_error + if isinstance(stream_error, MidStreamFallbackError) + else _anthropic_stream_fallback_error_for_raised(stream_error, model, llm_provider, has_generated_content) + ) + if fallback_error is None: + raise stream_error + async for item in self._aanthropic_messages_fallback_attempt(fallback_error, initial_kwargs, wrapper): + yield item + async def _aanthropic_messages_fallback_attempt( self, e: "MidStreamFallbackError", diff --git a/tests/test_litellm/test_router.py b/tests/test_litellm/test_router.py index 8716e6d6b25..bb0ab43ead9 100644 --- a/tests/test_litellm/test_router.py +++ b/tests/test_litellm/test_router.py @@ -15,6 +15,7 @@ import litellm from litellm import Router from litellm.exceptions import MidStreamFallbackError from litellm.integrations.custom_logger import CustomLogger +from litellm.llms.bedrock.common_utils import BedrockError from litellm.llms.anthropic.experimental_pass_through.messages.agentic_streaming_iterator import ( SERVER_FULFILLED_TOOL_LEAK_ERROR_SSE_BYTES, ) @@ -10222,6 +10223,119 @@ async def test_anthropic_messages_fallback_also_catches_raised_midstream_error() assert mock_fallback.await_args.kwargs["e"] is raised_error +@pytest.mark.asyncio +@pytest.mark.parametrize( + "raised_error", + [ + BedrockError(status_code=503, message='serviceUnavailableException {"message": "Service unavailable"}'), + BedrockError(status_code=500, message='internalServerException {"message": "Internal error"}'), + BedrockError(status_code=429, message='throttlingException {"message": "Too many requests"}'), + httpx.ReadError("connection reset by upstream"), + ], + ids=["503", "500", "429", "transport-drop"], +) +async def test_anthropic_messages_raised_provider_error_before_content_triggers_fallback(raised_error): + """Bedrock surfaces a mid-stream exception frame by raising BedrockError + out of its iterator rather than yielding an Anthropic SSE error event, so + the wrapper must convert a retriable pre-content raise into a fallback + attempt exactly like a detected error event (parity with + CustomStreamWrapper._handle_stream_fallback_error on /chat/completions).""" + router = _anthropic_messages_make_router() + source = _AnthropicMessagesRaisingByteStream([_anthropic_messages_message_start_chunk()], raised_error) + fallback_stream = _AnthropicMessagesFallbackByteStream([_anthropic_messages_content_chunk("fallback answer")]) + + with patch.object( + router, + "async_function_with_fallbacks_common_utils", + new=AsyncMock(return_value=fallback_stream), + ) as mock_fallback: + wrapped = await router._aanthropic_messages_streaming_iterator( + response=source, + initial_kwargs={"model": "primary"}, + ) + collected = [chunk async for chunk in wrapped] + + assert collected == [_anthropic_messages_content_chunk("fallback answer")] + mock_fallback.assert_awaited_once() + converted = mock_fallback.await_args.kwargs["e"] + assert isinstance(converted, MidStreamFallbackError) + assert converted.original_exception is raised_error + assert converted.is_pre_first_chunk is True + assert source.closed is True + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "raised_error", + [ + BedrockError(status_code=400, message='validationException {"message": "Malformed input"}'), + BedrockError(status_code=424, message='modelStreamErrorException {"message": "Model stream error"}'), + ], + ids=["400", "424"], +) +async def test_anthropic_messages_raised_non_retriable_provider_error_propagates_unchanged(raised_error): + """A raised 4xx (other than 429) is a client error no other deployment can + fix: it reaches the caller as the very same exception, with no fallback + attempt and nothing flushed, so the proxy still answers a clean 4xx.""" + router = _anthropic_messages_make_router() + source = _AnthropicMessagesRaisingByteStream([_anthropic_messages_message_start_chunk()], raised_error) + + with patch.object( + router, + "async_function_with_fallbacks_common_utils", + new=AsyncMock(), + ) as mock_fallback: + wrapped = await router._aanthropic_messages_streaming_iterator( + response=source, + initial_kwargs={"model": "primary"}, + ) + collected = [] + + async def _consume(): + async for chunk in wrapped: + collected.append(chunk) + + with pytest.raises(BedrockError) as exc_info: + await _consume() + + assert collected == [] + assert exc_info.value is raised_error + mock_fallback.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_anthropic_messages_raised_provider_error_after_content_propagates_unchanged(): + """Once real content reached the caller a fallback would append a second + message lifecycle to the same SSE stream, so a raised provider error after + content propagates as-is even when its status is retriable.""" + router = _anthropic_messages_make_router() + content = _anthropic_messages_content_chunk("partial answer") + raised_error = BedrockError(status_code=503, message='serviceUnavailableException {"message": "Service unavailable"}') + source = _AnthropicMessagesRaisingByteStream([content], raised_error) + + with patch.object( + router, + "async_function_with_fallbacks_common_utils", + new=AsyncMock(), + ) as mock_fallback: + wrapped = await router._aanthropic_messages_streaming_iterator( + response=source, + initial_kwargs={"model": "primary"}, + ) + collected = [] + + async def _consume(): + async for chunk in wrapped: + collected.append(chunk) + + with pytest.raises(BedrockError) as exc_info: + await _consume() + + assert collected == [content] + assert exc_info.value is raised_error + mock_fallback.assert_not_awaited() + + @pytest.mark.asyncio async def test_anthropic_messages_non_retriable_client_error_skips_fallback(): """A 4xx (non-429) error type (e.g. invalid_request_error) is a client From 406db3fccf1abfab8d8f084891641dc8ebb53227 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Thu, 27 Aug 2026 15:57:04 -0700 Subject: [PATCH 2/3] refactor(router): drop dead provider derivation in raised-stream fallback --- litellm/router.py | 10 +++------- tests/test_litellm/test_router.py | 19 +++++++++++++++++-- 2 files changed, 20 insertions(+), 9 deletions(-) diff --git a/litellm/router.py b/litellm/router.py index 8175a5c5c40..c47eea31f81 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -431,7 +431,7 @@ def _anthropic_stream_raised_error_status(error: Exception) -> int | None: def _anthropic_stream_fallback_error_for_raised( - error: Exception, model: str, llm_provider: str, has_generated_content: bool + error: Exception, model: str, has_generated_content: bool ) -> "MidStreamFallbackError | None": """ A provider iterator that fails mid-stream by raising (Bedrock surfaces @@ -454,7 +454,7 @@ def _anthropic_stream_fallback_error_for_raised( return MidStreamFallbackError( message=str(error), model=model, - llm_provider=llm_provider, + llm_provider="anthropic", original_exception=error, is_pre_first_chunk=True, ) @@ -5059,8 +5059,6 @@ class Router: has_generated_content = False # rebind-ok: set once real content is seen, or the buffer cap is hit buffered_lifecycle_chunks: tuple[bytes, ...] = () # rebind-ok: flushed once committed or on decline model: Final = cast(str, initial_kwargs.get("model")) # cast-ok: kwargs always carries the model group - custom_llm_provider: Final = initial_kwargs.get("custom_llm_provider") - llm_provider: Final = custom_llm_provider if isinstance(custom_llm_provider, str) else "anthropic" try: async for chunk in source_iterator: if _anthropic_stream_forwards_ping_live( @@ -5109,7 +5107,6 @@ class Router: has_generated_content, buffered_lifecycle_chunks, model, - llm_provider, initial_kwargs, wrapper, ): @@ -5130,7 +5127,6 @@ class Router: has_generated_content: bool, buffered_lifecycle_chunks: tuple[bytes, ...], model: str, - llm_provider: str, initial_kwargs: dict[str, Any], # mutable-ok: handed to _aanthropic_messages_fallback_attempt, which mutates it wrapper: "FallbackAwareAnthropicMessagesStream", ) -> AsyncGenerator[bytes, None]: @@ -5158,7 +5154,7 @@ class Router: fallback_error: Final = ( stream_error if isinstance(stream_error, MidStreamFallbackError) - else _anthropic_stream_fallback_error_for_raised(stream_error, model, llm_provider, has_generated_content) + else _anthropic_stream_fallback_error_for_raised(stream_error, model, has_generated_content) ) if fallback_error is None: raise stream_error diff --git a/tests/test_litellm/test_router.py b/tests/test_litellm/test_router.py index bb0ab43ead9..752bbf942d1 100644 --- a/tests/test_litellm/test_router.py +++ b/tests/test_litellm/test_router.py @@ -4,6 +4,7 @@ import json import logging import os import threading +from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch import httpx @@ -10264,14 +10265,28 @@ async def test_anthropic_messages_raised_provider_error_before_content_triggers_ assert source.closed is True +class _AnthropicMessagesStringStatusError(Exception): + def __init__(self): + super().__init__("bad request") + self.status_code = "400" + + +class _AnthropicMessagesResponseOnlyStatusError(Exception): + def __init__(self): + super().__init__("bad request") + self.response = SimpleNamespace(status_code=400) + + @pytest.mark.asyncio @pytest.mark.parametrize( "raised_error", [ BedrockError(status_code=400, message='validationException {"message": "Malformed input"}'), BedrockError(status_code=424, message='modelStreamErrorException {"message": "Model stream error"}'), + _AnthropicMessagesStringStatusError(), + _AnthropicMessagesResponseOnlyStatusError(), ], - ids=["400", "424"], + ids=["400", "424", "str-400", "response-only-400"], ) async def test_anthropic_messages_raised_non_retriable_provider_error_propagates_unchanged(raised_error): """A raised 4xx (other than 429) is a client error no other deployment can @@ -10295,7 +10310,7 @@ async def test_anthropic_messages_raised_non_retriable_provider_error_propagates async for chunk in wrapped: collected.append(chunk) - with pytest.raises(BedrockError) as exc_info: + with pytest.raises(type(raised_error)) as exc_info: await _consume() assert collected == [] From e6a568d99bdd612664df20306928c598216e86fd Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Thu, 27 Aug 2026 18:42:59 -0700 Subject: [PATCH 3/3] test(router): cover the raised-stream fallback helpers by name and trim their docstrings --- litellm/router.py | 26 +------- tests/test_litellm/test_router.py | 107 ++++++++++++++++++++++++++---- 2 files changed, 98 insertions(+), 35 deletions(-) diff --git a/litellm/router.py b/litellm/router.py index c47eea31f81..07f545231de 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -433,17 +433,7 @@ def _anthropic_stream_raised_error_status(error: Exception) -> int | None: def _anthropic_stream_fallback_error_for_raised( error: Exception, model: str, has_generated_content: bool ) -> "MidStreamFallbackError | None": - """ - A provider iterator that fails mid-stream by raising (Bedrock surfaces - its event-stream exception frames as a BedrockError, a transport drop - raises httpx's error) never produces the Anthropic SSE `event: error` - frame the wrapper detects, so the raise is converted into the same - MidStreamFallbackError a detected error event gets, under the same gate: - only before real content reached the caller and only for a retriable - status (429, 5xx, or none at all for a transport failure), mirroring - CustomStreamWrapper._handle_stream_fallback_error on /chat/completions. - None means the exception propagates to the caller unchanged. - """ + """Same gate as a detected SSE error event; None means the raise propagates unchanged.""" from litellm.exceptions import MidStreamFallbackError if has_generated_content: @@ -5101,7 +5091,7 @@ class Router: yield chunk for buffered_chunk in buffered_lifecycle_chunks: yield buffered_chunk - except Exception as stream_error: # noqa: BLE001 # any raised provider error must reach the fallback gate, like CustomStreamWrapper + except Exception as stream_error: # noqa: BLE001 # any raised provider error must reach the fallback gate async for item in self._aanthropic_messages_recover_stream_error( stream_error, has_generated_content, @@ -5130,17 +5120,7 @@ class Router: initial_kwargs: dict[str, Any], # mutable-ok: handed to _aanthropic_messages_fallback_attempt, which mutates it wrapper: "FallbackAwareAnthropicMessagesStream", ) -> AsyncGenerator[bytes, None]: - """ - Decides what a source-iterator failure in - Router._aanthropic_messages_streaming_iterator turns into: a fallback - attempt, or the error reaching the caller. A MidStreamFallbackError - (completion-bridge path, or the wrapper's own SSE error-event - detection) is declined per _anthropic_stream_should_decline_fallback - with the held-back lifecycle frames flushed first; any other raise is - converted per _anthropic_stream_fallback_error_for_raised and, when - not convertible, propagates untouched so the caller still gets a - clean error response. - """ + """Turns a source-iterator failure into a fallback attempt or the error reaching the caller.""" from litellm.exceptions import MidStreamFallbackError if isinstance(stream_error, MidStreamFallbackError) and _anthropic_stream_should_decline_fallback( diff --git a/tests/test_litellm/test_router.py b/tests/test_litellm/test_router.py index 752bbf942d1..0115438ae80 100644 --- a/tests/test_litellm/test_router.py +++ b/tests/test_litellm/test_router.py @@ -24,6 +24,8 @@ from litellm.router import ( MAX_BUFFERED_PRE_CONTENT_ANTHROPIC_CHUNKS, FallbackAwareAnthropicMessagesStream, _anthropic_stream_commits_now, + _anthropic_stream_fallback_error_for_raised, + _anthropic_stream_raised_error_status, _anthropic_stream_should_decline_fallback, _anthropic_stream_error_is_gateway_verdict, _anthropic_stream_forwards_ping_live, @@ -10236,11 +10238,7 @@ async def test_anthropic_messages_fallback_also_catches_raised_midstream_error() ids=["503", "500", "429", "transport-drop"], ) async def test_anthropic_messages_raised_provider_error_before_content_triggers_fallback(raised_error): - """Bedrock surfaces a mid-stream exception frame by raising BedrockError - out of its iterator rather than yielding an Anthropic SSE error event, so - the wrapper must convert a retriable pre-content raise into a fallback - attempt exactly like a detected error event (parity with - CustomStreamWrapper._handle_stream_fallback_error on /chat/completions).""" + """A retriable raise before content falls over exactly like a detected SSE error event.""" router = _anthropic_messages_make_router() source = _AnthropicMessagesRaisingByteStream([_anthropic_messages_message_start_chunk()], raised_error) fallback_stream = _AnthropicMessagesFallbackByteStream([_anthropic_messages_content_chunk("fallback answer")]) @@ -10289,9 +10287,7 @@ class _AnthropicMessagesResponseOnlyStatusError(Exception): ids=["400", "424", "str-400", "response-only-400"], ) async def test_anthropic_messages_raised_non_retriable_provider_error_propagates_unchanged(raised_error): - """A raised 4xx (other than 429) is a client error no other deployment can - fix: it reaches the caller as the very same exception, with no fallback - attempt and nothing flushed, so the proxy still answers a clean 4xx.""" + """A raised client error reaches the caller as the same exception, nothing flushed, no fallback.""" router = _anthropic_messages_make_router() source = _AnthropicMessagesRaisingByteStream([_anthropic_messages_message_start_chunk()], raised_error) @@ -10320,12 +10316,12 @@ async def test_anthropic_messages_raised_non_retriable_provider_error_propagates @pytest.mark.asyncio async def test_anthropic_messages_raised_provider_error_after_content_propagates_unchanged(): - """Once real content reached the caller a fallback would append a second - message lifecycle to the same SSE stream, so a raised provider error after - content propagates as-is even when its status is retriable.""" + """A raise after content propagates unchanged even when its status is retriable.""" router = _anthropic_messages_make_router() content = _anthropic_messages_content_chunk("partial answer") - raised_error = BedrockError(status_code=503, message='serviceUnavailableException {"message": "Service unavailable"}') + raised_error = BedrockError( + status_code=503, message='serviceUnavailableException {"message": "Service unavailable"}' + ) source = _AnthropicMessagesRaisingByteStream([content], raised_error) with patch.object( @@ -10351,6 +10347,93 @@ async def test_anthropic_messages_raised_provider_error_after_content_propagates mock_fallback.assert_not_awaited() +@pytest.mark.parametrize( + "error, expected_status", + [ + (BedrockError(status_code=503, message="unavailable"), 503), + (_AnthropicMessagesStringStatusError(), 400), + (_AnthropicMessagesResponseOnlyStatusError(), 400), + (httpx.ReadError("connection reset by upstream"), None), + ], + ids=["int", "digit-str", "response-only", "none"], +) +def test_anthropic_stream_raised_error_status_reads_every_status_shape(error, expected_status): + assert _anthropic_stream_raised_error_status(error) == expected_status + + +@pytest.mark.parametrize( + "error, has_generated_content, converts", + [ + (BedrockError(status_code=503, message="unavailable"), False, True), + (httpx.ReadError("connection reset by upstream"), False, True), + (BedrockError(status_code=400, message="malformed"), False, False), + (BedrockError(status_code=503, message="unavailable"), True, False), + ], + ids=["retriable", "no-status", "client-error", "after-content"], +) +def test_anthropic_stream_fallback_error_for_raised_gates_like_a_detected_error_event( + error, has_generated_content, converts +): + converted = _anthropic_stream_fallback_error_for_raised(error, "primary", has_generated_content) + if not converts: + assert converted is None + return + assert isinstance(converted, MidStreamFallbackError) + assert converted.original_exception is error + assert converted.is_pre_first_chunk is True + assert converted.llm_provider == "anthropic" + + +@pytest.mark.asyncio +async def test_aanthropic_messages_recover_stream_error_flushes_buffered_frames_before_declining(): + router = _anthropic_messages_make_router() + original = BedrockError(status_code=503, message="unavailable") + declined = MidStreamFallbackError( + message="unavailable", + model="primary", + llm_provider="anthropic", + original_exception=original, + is_pre_first_chunk=False, + ) + buffered = (_anthropic_messages_message_start_chunk(),) + flushed = [] + + async def drain(recovery) -> None: + async for chunk in recovery: + flushed.append(chunk) + + with patch.object(router, "_aanthropic_messages_fallback_attempt") as mock_attempt: + recovery = router._aanthropic_messages_recover_stream_error( + declined, True, buffered, "primary", {"model": "primary"}, _anthropic_messages_make_wrapper() + ) + with pytest.raises(BedrockError) as exc_info: + await drain(recovery) + assert flushed == list(buffered) + assert exc_info.value is original + mock_attempt.assert_not_called() + + +@pytest.mark.asyncio +async def test_aanthropic_messages_recover_stream_error_hands_converted_raise_to_fallback_attempt(): + router = _anthropic_messages_make_router() + raised = BedrockError(status_code=503, message="unavailable") + handed_over = [] + + async def fake_attempt(fallback_error, initial_kwargs, wrapper): + handed_over.append(fallback_error) + yield b"fallback" + + with patch.object(router, "_aanthropic_messages_fallback_attempt", new=fake_attempt): + recovery = router._aanthropic_messages_recover_stream_error( + raised, False, (), "primary", {"model": "primary"}, _anthropic_messages_make_wrapper() + ) + collected = [chunk async for chunk in recovery] + assert collected == [b"fallback"] + assert len(handed_over) == 1 + assert isinstance(handed_over[0], MidStreamFallbackError) + assert handed_over[0].original_exception is raised + + @pytest.mark.asyncio async def test_anthropic_messages_non_retriable_client_error_skips_fallback(): """A 4xx (non-429) error type (e.g. invalid_request_error) is a client