From 97da25fcea54fead10737cad7ad84149ae4f08ed Mon Sep 17 00:00:00 2001 From: nuernber Date: Wed, 2 Sep 2026 12:00:46 -0700 Subject: [PATCH 1/6] fix(bedrock): flush message_start on arrival for /v1/messages streams The Bedrock invoke /v1/messages reader passed chunk_size=1024 to httpx.Response.aiter_bytes. httpx's ByteChunker is an accumulator, not a cap: below the threshold it returns nothing at all. A message_start frame is roughly 500-700 bytes, so it sat in httpx's buffer until enough further upstream bytes arrived. During a reasoning phase nothing else arrives, so the preamble was released only once the first content block showed up, tens of seconds later, past the client's first-byte watchdog. Dropping chunk_size lets each transport read through as it lands. The event-stream decoder already emits on botocore event boundaries, so framing is unaffected. The /chat/completions invoke path was never affected: it passes stream_chunk_size, which defaults to None. --- .../anthropic_claude3_transformation.py | 9 +- .../test_anthropic_claude3_transformation.py | 105 ++++++++++++++++++ 2 files changed, 110 insertions(+), 4 deletions(-) diff --git a/litellm/llms/bedrock/messages/invoke_transformations/anthropic_claude3_transformation.py b/litellm/llms/bedrock/messages/invoke_transformations/anthropic_claude3_transformation.py index 6ff9f0155f9..6b82ae4ba0d 100644 --- a/litellm/llms/bedrock/messages/invoke_transformations/anthropic_claude3_transformation.py +++ b/litellm/llms/bedrock/messages/invoke_transformations/anthropic_claude3_transformation.py @@ -777,9 +777,11 @@ class AmazonAnthropicClaudeMessagesConfig( aws_decoder: Final = AmazonAnthropicClaudeMessagesStreamDecoder( model=model, ) - completion_stream: Final = aws_decoder.aiter_bytes( - httpx_response.aiter_bytes(chunk_size=aws_decoder.DEFAULT_CHUNK_SIZE) - ) + # No ``chunk_size``: httpx's ByteChunker withholds bytes until that many + # accumulate, stranding a smaller ``message_start`` frame until the next + # upstream event, which after a reasoning phase is tens of seconds later + # (BerriAI/litellm#38689). + completion_stream: Final = aws_decoder.aiter_bytes(httpx_response.aiter_bytes()) # Convert decoded Bedrock events to Server-Sent Events expected by Anthropic clients. return self.bedrock_sse_wrapper( completion_stream=completion_stream, @@ -934,7 +936,6 @@ class AmazonAnthropicClaudeMessagesStreamDecoder(AWSEventStreamDecoder): Iterator to return Bedrock invoke response in anthropic /messages format """ super().__init__(model=model) - self.DEFAULT_CHUNK_SIZE = 1024 def _chunk_parser(self, chunk_data: dict) -> GChunk | ModelResponseStream | dict: """ diff --git a/tests/test_litellm/llms/bedrock/messages/invoke_transformations/test_anthropic_claude3_transformation.py b/tests/test_litellm/llms/bedrock/messages/invoke_transformations/test_anthropic_claude3_transformation.py index 09ebc1a3c95..1d74f355156 100644 --- a/tests/test_litellm/llms/bedrock/messages/invoke_transformations/test_anthropic_claude3_transformation.py +++ b/tests/test_litellm/llms/bedrock/messages/invoke_transformations/test_anthropic_claude3_transformation.py @@ -3250,3 +3250,108 @@ def test_bedrock_messages_strips_effort_but_keeps_format_for_sonnet_4_5(local_mo ) assert result.get("output_config") == {"format": schema_format} + + +def _bedrock_invoke_event_frame(payload: dict) -> bytes: + """Encode one Bedrock invoke event-stream frame carrying an Anthropic event.""" + import base64 + import struct + from binascii import crc32 + + encoded_event = base64.b64encode(json.dumps(payload, separators=(",", ":")).encode()).decode() + body = json.dumps({"bytes": encoded_event}).encode() + + def _str_header(name: str, value: str) -> bytes: + name_b = name.encode() + value_b = value.encode() + return ( + struct.pack("!B", len(name_b)) + name_b + struct.pack("!B", 7) + struct.pack("!H", len(value_b)) + value_b + ) + + headers = ( + _str_header(":event-type", "chunk") + + _str_header(":content-type", "application/json") + + _str_header(":message-type", "event") + ) + prelude = struct.pack("!II", 12 + len(headers) + len(body) + 4, len(headers)) + prelude_crc = crc32(prelude) & 0xFFFFFFFF + prelude_crc_b = struct.pack("!I", prelude_crc) + msg_crc_b = struct.pack("!I", crc32(prelude_crc_b + headers + body, prelude_crc) & 0xFFFFFFFF) + return prelude + prelude_crc_b + headers + body + msg_crc_b + + +_MESSAGE_START_EVENT = { + "type": "message_start", + "message": { + "id": "msg_bdrk_01WxYzAbCdEfGhIjKlMnOpQr", + "type": "message", + "role": "assistant", + "model": "claude-sonnet-4-5-20250929", + "content": [], + "stop_reason": None, + "stop_sequence": None, + "usage": {"input_tokens": 41, "output_tokens": 1}, + }, +} + +_CONTENT_BLOCK_START_EVENT = { + "type": "content_block_start", + "index": 0, + "content_block": {"type": "thinking", "thinking": "", "signature": ""}, +} + + +@pytest.mark.asyncio +async def test_streaming_flushes_message_start_before_the_next_upstream_event(): + """``message_start`` must reach the client as soon as Bedrock sends it. + + Bedrock emits ``message_start`` immediately, then goes silent for the whole + reasoning phase (tens of seconds at high effort) before the first content + block. Reading the response with an ``httpx`` ``chunk_size`` stranded the + preamble in httpx's ByteChunker until enough further bytes accumulated, so + the client's first byte landed at first-content time and tripped its + first-byte watchdog. Regression for BerriAI/litellm#38689. + """ + import httpx + + message_start_frame = _bedrock_invoke_event_frame(_MESSAGE_START_EVENT) + # The stall only happens for a preamble smaller than the read threshold, so a + # frame that grew past it would make this test pass without the fix. + assert len(message_start_frame) < 1024 + + reasoning_finished = asyncio.Event() + + class _ReasoningStall(httpx.AsyncByteStream): + async def __aiter__(self): + yield message_start_frame + await reasoning_finished.wait() + yield _bedrock_invoke_event_frame(_CONTENT_BLOCK_START_EVENT) + + cfg = AmazonAnthropicClaudeMessagesConfig() + stream = cfg.get_async_streaming_response_iterator( + model="us.anthropic.claude-sonnet-4-5-20250929-v1:0", + httpx_response=httpx.Response(200, stream=_ReasoningStall()), + request_body={}, + litellm_logging_obj=LiteLLMLoggingObj( + model="bedrock/invoke/us.anthropic.claude-sonnet-4-5-20250929-v1:0", + messages=[{"role": "user", "content": "think hard"}], + stream=True, + call_type="anthropic_messages", + start_time=datetime.now(), + litellm_call_id="test_flush_message_start", + function_id="test_flush_message_start", + ), + ) + + try: + # Fails by timing out while the upstream is still mid-reasoning if the + # preamble is being held back. + first_chunk = await asyncio.wait_for(stream.__anext__(), timeout=5) + assert b"event: message_start" in first_chunk + + reasoning_finished.set() + second_chunk = await asyncio.wait_for(stream.__anext__(), timeout=5) + assert b"event: content_block_start" in second_chunk + finally: + reasoning_finished.set() + await stream.aclose() From 1b2cbb553000991be88bad0e66234dc9d571e345 Mon Sep 17 00:00:00 2001 From: nuernber Date: Thu, 3 Sep 2026 08:21:40 -0700 Subject: [PATCH 2/6] fix(router): skip mid-stream-fallback buffering when no fallback is configured Router._aanthropic_messages_streaming_iterator buffers message_start, content_block_start, and pings until the first content_block_delta arrives, so a mid-stream provider error can retry against a fallback deployment before any lifecycle frame commits the client to the current attempt. That buffering ran unconditionally, with no check for whether the router has any fallback deployment to retry against in the first place. For adaptive-thinking Bedrock models, the first content_block_delta doesn't arrive until the model's entire thinking pass finishes, which can take well over a minute. With no fallback configured, the buffering held message_start and content_block_start hostage for that whole window even though the provider answered in about two seconds, purely because there was nothing to protect against. Add Router._has_any_configured_fallback to check whether any fallback (general, context-window, content-policy, or a catch-all default) could resolve for the requested model group, and skip the buffering entirely when it can't: forward the source stream live instead. When a fallback is configured, behavior is unchanged. --- litellm/router.py | 49 +++++++++++ ...test_router_anthropic_messages_fallback.py | 46 +++++++++++ tests/test_litellm/test_router.py | 81 ++++++++++++++++++- 3 files changed, 175 insertions(+), 1 deletion(-) diff --git a/litellm/router.py b/litellm/router.py index 303b22c9484..77dba630467 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -512,6 +512,16 @@ def _anthropic_stream_commits_now(chunk: object, has_generated_content: bool, bu return is_anthropic_content_delta_chunk(chunk) or buffered_chunk_count >= MAX_BUFFERED_PRE_CONTENT_ANTHROPIC_CHUNKS +async def _anthropic_messages_stream_without_fallback_protection( + source_iterator: AsyncIterator[bytes], +) -> AsyncGenerator[bytes, None]: + """No fallback is configured for the requested model group, so there is nothing the + buffer-until-content protection in Router._aanthropic_messages_streaming_iterator would + protect: forward the source iterator live instead of wrapping it.""" + async for chunk in source_iterator: + yield chunk + + class FallbackAwareAnthropicMessagesStream: """ Bare async generators can't carry the `_hidden_params` attribute the @@ -5141,6 +5151,15 @@ class Router: source_iterator: Final = response + model_group: Final = cast(str, initial_kwargs.get("model")) # cast-ok: kwargs always carries the model group + if not self._has_any_configured_fallback(model_group, initial_kwargs): + # Nothing to fall back to, so buffering lifecycle frames to protect a mid-stream + # fallback attempt would only add latency for no benefit: forward the source + # iterator live, exactly as it would stream without this wrapper. + return FallbackAwareAnthropicMessagesStream( + _anthropic_messages_stream_without_fallback_protection(source_iterator), source_iterator + ) + async def stream_with_fallbacks() -> AsyncGenerator[bytes, None]: from litellm.exceptions import MidStreamFallbackError @@ -8109,6 +8128,36 @@ class Router: return True return False + def _has_any_configured_fallback(self, model_group: str, kwargs: Mapping[str, Any]) -> bool: + """ + Whether any fallback deployment - general, context-window, content-policy, or a + catch-all default - could resolve for this model group. + + Gates whether _aanthropic_messages_streaming_iterator's buffer-until-content + protection is worth paying for: that protection exists so a mid-stream provider + error can retry against a fallback deployment before any lifecycle frame commits + the client to this attempt. With no fallback destination configured at all, a + retry can never happen, so holding message_start/content_block_start hostage + until real content arrives protects nothing and only adds latency (most visibly + on adaptive-thinking models, where the first content_block_delta can lag + message_start by well over a minute). + """ + lookup_groups: Final = fallback_lookup_groups(kwargs, model_group) + for fallback_key, configured_fallbacks in ( + ("fallbacks", self.fallbacks), + ("context_window_fallbacks", self.context_window_fallbacks), + ("content_policy_fallbacks", self.content_policy_fallbacks), + ): + fallbacks_value: Final = kwargs.get(fallback_key, configured_fallbacks) + if fallbacks_value is not None and ( + self._get_fallback_model_group_for_lookup_groups( + fallbacks=fallbacks_value, lookup_groups=lookup_groups + ) + is not None + ): + return True + return self._has_default_fallbacks() + def _has_content_policy_fallback(self, model_group: str, kwargs: Mapping[str, Any]) -> bool: """ Whether a content-policy fallback would resolve for this request, keyed the same way diff --git a/tests/router_unit_tests/test_router_anthropic_messages_fallback.py b/tests/router_unit_tests/test_router_anthropic_messages_fallback.py index 4812d199c06..46bfa787097 100644 --- a/tests/router_unit_tests/test_router_anthropic_messages_fallback.py +++ b/tests/router_unit_tests/test_router_anthropic_messages_fallback.py @@ -478,6 +478,52 @@ def test_has_content_policy_fallback_default_fallbacks_arm(): assert router._has_content_policy_fallback("any-group", {"content_policy_fallbacks": [{"other": ["x"]}]}) is False +def test_has_any_configured_fallback_general_fallbacks_arm(): + router = Router(model_list=[FABLE_TIER, OPUS_TARGET], fallbacks=[{"fable-tier": ["opus-target"]}]) + + assert router._has_any_configured_fallback("fable-tier", {}) is True + assert router._has_any_configured_fallback("other-group", {}) is False + + +def test_has_any_configured_fallback_context_window_fallbacks_arm(): + router = Router( + model_list=[FABLE_TIER, OPUS_TARGET], + context_window_fallbacks=[{"fable-tier": ["opus-target"]}], + ) + + assert router._has_any_configured_fallback("fable-tier", {}) is True + assert router._has_any_configured_fallback("other-group", {}) is False + + +def test_has_any_configured_fallback_content_policy_fallbacks_arm(): + router = Router( + model_list=[FABLE_TIER, OPUS_TARGET], + content_policy_fallbacks=[{"fable-tier": ["opus-target"]}], + ) + + assert router._has_any_configured_fallback("fable-tier", {}) is True + + +def test_has_any_configured_fallback_default_fallbacks_arm(): + router = Router(model_list=[OPUS_TARGET], fallbacks=[{"*": ["opus-target"]}]) + + assert router._has_any_configured_fallback("any-group", {}) is True + + +def test_has_any_configured_fallback_nothing_configured(): + router = Router(model_list=[FABLE_TIER, OPUS_TARGET]) + + assert router._has_any_configured_fallback("fable-tier", {}) is False + + +def test_has_any_configured_fallback_honors_per_request_kwargs_override(): + router = Router(model_list=[FABLE_TIER, OPUS_TARGET]) + + assert ( + router._has_any_configured_fallback("fable-tier", {"fallbacks": [{"fable-tier": ["opus-target"]}]}) is True + ) + + def test_get_fallback_model_group_for_lookup_groups_orders_tier_before_requested(): router = _router(content_policy_fallbacks=None) fallbacks = [{"tier1": ["backup-a"]}, {"smart-router": ["backup-b"]}] diff --git a/tests/test_litellm/test_router.py b/tests/test_litellm/test_router.py index c843a66a1c1..1b9ed4b1398 100644 --- a/tests/test_litellm/test_router.py +++ b/tests/test_litellm/test_router.py @@ -10089,6 +10089,13 @@ def _anthropic_messages_make_wrapper() -> FallbackAwareAnthropicMessagesStream: def _anthropic_messages_make_router() -> Router: + """A ``fallbacks`` entry mapping "primary" -> "fallback" is required, not just + a second model_list entry: _has_any_configured_fallback gates the whole + buffer-until-content mechanism on whether the router could actually resolve + a fallback for the model group, so a router with nothing configured under + ``fallbacks``/``context_window_fallbacks``/``content_policy_fallbacks`` + (matching a real deployment with no fallback set up) skips buffering + entirely and streams live - see test_anthropic_messages_streaming_iterator_skips_buffering_without_any_configured_fallback.""" return Router( model_list=[ { @@ -10104,7 +10111,8 @@ def _anthropic_messages_make_router() -> Router: "model": "bedrock/anthropic.claude-sonnet-4-5", }, }, - ] + ], + fallbacks=[{"primary": ["fallback"]}], ) @@ -10152,6 +10160,29 @@ class _AnthropicMessagesRaisingByteStream: self.closed = True +class _AnthropicMessagesHangingByteStream: + """Yields the given chunks, then hangs forever on the next `__anext__()` + instead of raising StopAsyncIteration - simulates a real upstream stuck + mid-thinking-pass, so a test can prove a chunk reached the caller without + waiting for the rest of the stream (which, here, never arrives).""" + + def __init__(self, chunks: list) -> None: + self._chunks = list(chunks) + self._hidden_params: dict = {} + + def __aiter__(self): + return self + + async def __anext__(self) -> bytes: + if self._chunks: + return self._chunks.pop(0) + await asyncio.Event().wait() + raise AssertionError("unreachable") # pragma: no cover + + async def aclose(self) -> None: + pass + + class _AnthropicMessagesFallbackByteStream: def __init__(self, chunks: list, hidden_params: dict | None = None) -> None: self._chunks = list(chunks) @@ -10264,6 +10295,54 @@ async def test_anthropic_messages_streaming_iterator_flushes_buffered_frames_on_ await wrapped.__anext__() +@pytest.mark.asyncio +async def test_anthropic_messages_streaming_iterator_skips_buffering_without_any_configured_fallback(): + """Regression: with no fallback configured for the model group (no + ``fallbacks``, ``context_window_fallbacks``, ``content_policy_fallbacks``, + or catch-all default), lifecycle frames like message_start must reach the + caller as soon as the source produces them, not be buffered until real + content arrives. Buffering exists to protect a mid-stream fallback + attempt; with nothing to fall back to, there is nothing to protect, and + the delay is pure added latency (most visible on adaptive-thinking models, + where the first content_block_delta can lag message_start by a minute or + more). The source here hangs forever after its first chunk, so this can + only pass if that chunk was forwarded live rather than buffered.""" + router = Router( + model_list=[ + { + "model_name": "primary", + "litellm_params": {"model": "anthropic/claude-sonnet-4-5", "api_key": "sk-test"}, + }, + ] + ) + source = _AnthropicMessagesHangingByteStream([_anthropic_messages_message_start_chunk()]) + + wrapped = await router._aanthropic_messages_streaming_iterator( + response=source, initial_kwargs={"model": "primary"} + ) + + first_chunk = await asyncio.wait_for(wrapped.__anext__(), timeout=1.0) + assert first_chunk == _anthropic_messages_message_start_chunk() + + +@pytest.mark.asyncio +async def test_anthropic_messages_streaming_iterator_still_buffers_lifecycle_frames_when_fallback_configured(): + """Confirms the buffer-until-content protection is still intact once a + fallback IS configured for the model group - only the no-fallback case + added by this fix skips it. The source hangs forever after message_start, + so if it were forwarded live this would resolve within the timeout instead + of raising.""" + router = _anthropic_messages_make_router() + source = _AnthropicMessagesHangingByteStream([_anthropic_messages_message_start_chunk()]) + + wrapped = await router._aanthropic_messages_streaming_iterator( + response=source, initial_kwargs={"model": "primary"} + ) + + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(wrapped.__anext__(), timeout=0.2) + + @pytest.mark.asyncio async def test_anthropic_messages_content_coalesced_with_error_in_one_physical_chunk_skips_fallback(): """Greptile review round: transport-level buffering can coalesce a real From 8f5f42faeb20f801a96b192aa85ad5bb2ee70558 Mon Sep 17 00:00:00 2001 From: nuernber Date: Thu, 3 Sep 2026 08:39:40 -0700 Subject: [PATCH 3/6] fix(router): satisfy lint on the fallback-configured gate Rewrite _has_any_configured_fallback's loop as a generator expression: basedpyright rejects a Final-annotated variable assigned inside a for-loop body (reportGeneralTypeIssues), and ruff format reflows the result. --- litellm/router.py | 23 +++++++++++------------ 1 file changed, 11 insertions(+), 12 deletions(-) diff --git a/litellm/router.py b/litellm/router.py index 77dba630467..cb51d8a085d 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -8143,19 +8143,18 @@ class Router: message_start by well over a minute). """ lookup_groups: Final = fallback_lookup_groups(kwargs, model_group) - for fallback_key, configured_fallbacks in ( - ("fallbacks", self.fallbacks), - ("context_window_fallbacks", self.context_window_fallbacks), - ("content_policy_fallbacks", self.content_policy_fallbacks), + candidate_fallback_lists: Final = ( + kwargs.get("fallbacks", self.fallbacks), + kwargs.get("context_window_fallbacks", self.context_window_fallbacks), + kwargs.get("content_policy_fallbacks", self.content_policy_fallbacks), + ) + if any( + fallbacks_value is not None + and self._get_fallback_model_group_for_lookup_groups(fallbacks=fallbacks_value, lookup_groups=lookup_groups) + is not None + for fallbacks_value in candidate_fallback_lists ): - fallbacks_value: Final = kwargs.get(fallback_key, configured_fallbacks) - if fallbacks_value is not None and ( - self._get_fallback_model_group_for_lookup_groups( - fallbacks=fallbacks_value, lookup_groups=lookup_groups - ) - is not None - ): - return True + return True return self._has_default_fallbacks() def _has_content_policy_fallback(self, model_group: str, kwargs: Mapping[str, Any]) -> bool: From 7a1b52f481415326c88e9e0d315a0b1a079955f0 Mon Sep 17 00:00:00 2001 From: nuernber Date: Thu, 3 Sep 2026 10:21:49 -0700 Subject: [PATCH 4/6] fix(router): address PR review feedback on the fallback-buffering gate Close the upstream stream on client disconnect in the no-fallback live-forward path, matching the buffered path's shielded aclose_if_supported cleanup. Fix _has_any_configured_fallback to match what a real retry would resolve: a stripped model-group key and a client-supplied non-standard fallbacks list both arm a retry but were reported as "nothing configured" before this fix. --- litellm/router.py | 36 ++++++++++++++++--- ...test_router_anthropic_messages_fallback.py | 20 +++++++++++ tests/test_litellm/test_router.py | 31 +++++++++++++++- 3 files changed, 81 insertions(+), 6 deletions(-) diff --git a/litellm/router.py b/litellm/router.py index cb51d8a085d..5ab7623aa3e 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -517,9 +517,23 @@ async def _anthropic_messages_stream_without_fallback_protection( ) -> AsyncGenerator[bytes, None]: """No fallback is configured for the requested model group, so there is nothing the buffer-until-content protection in Router._aanthropic_messages_streaming_iterator would - protect: forward the source iterator live instead of wrapping it.""" - async for chunk in source_iterator: - yield chunk + protect: forward the source iterator live instead of wrapping it. + + A client that disconnects mid-stream leaves this generator suspended at `yield` + rather than exhausted, so the `finally` below - not the `async for` running to + completion - is what closes the upstream connection; without it, a disconnect + during a long adaptive-thinking pass would leak the request to the provider. + """ + from litellm.llms.anthropic.experimental_pass_through.messages.streaming_iterator import ( + aclose_if_supported, + ) + + try: + async for chunk in source_iterator: + yield chunk + finally: + with anyio.CancelScope(shield=True), contextlib.suppress(BaseException): + await aclose_if_supported(source_iterator) class FallbackAwareAnthropicMessagesStream: @@ -8141,16 +8155,28 @@ class Router: until real content arrives protects nothing and only adds latency (most visibly on adaptive-thinking models, where the first content_block_delta can lag message_start by well over a minute). + + Matching mirrors what async_function_with_fallbacks_common_utils actually resolves + at retry time, not just an exact model-group key: get_fallback_model_group_for_lookup_groups + also checks a stripped model-group match (e.g. a fallback keyed by the bare model name + still arming a request routed with a provider prefix), and a client-supplied non-standard + ``fallbacks`` list (a plain list of model names, or of full override params) applies to + every model group unconditionally rather than being keyed by one at all - self._get_fallback_model_group_for_lookup_groups + checks neither, so using it here would report "nothing to fall back to" for a request + that a real error would in fact retry. """ + fallbacks: Final = kwargs.get("fallbacks", self.fallbacks) + if _check_non_standard_fallback_format(fallbacks=fallbacks): + return True lookup_groups: Final = fallback_lookup_groups(kwargs, model_group) candidate_fallback_lists: Final = ( - kwargs.get("fallbacks", self.fallbacks), + fallbacks, kwargs.get("context_window_fallbacks", self.context_window_fallbacks), kwargs.get("content_policy_fallbacks", self.content_policy_fallbacks), ) if any( fallbacks_value is not None - and self._get_fallback_model_group_for_lookup_groups(fallbacks=fallbacks_value, lookup_groups=lookup_groups) + and get_fallback_model_group_for_lookup_groups(fallbacks=fallbacks_value, lookup_groups=lookup_groups)[0] is not None for fallbacks_value in candidate_fallback_lists ): diff --git a/tests/router_unit_tests/test_router_anthropic_messages_fallback.py b/tests/router_unit_tests/test_router_anthropic_messages_fallback.py index 46bfa787097..4eb4b8d9260 100644 --- a/tests/router_unit_tests/test_router_anthropic_messages_fallback.py +++ b/tests/router_unit_tests/test_router_anthropic_messages_fallback.py @@ -524,6 +524,26 @@ def test_has_any_configured_fallback_honors_per_request_kwargs_override(): ) +def test_has_any_configured_fallback_matches_stripped_model_group(): + """Regression: async_function_with_fallbacks_common_utils resolves a fallback keyed by + the bare model name even when the request was routed with a provider prefix (e.g. a + fallback keyed "fable-tier" still arms "openai/fable-tier"); the gate must recognize + that same stripped match instead of requiring an exact model-group key.""" + router = Router(model_list=[FABLE_TIER, OPUS_TARGET], fallbacks=[{"fable-tier": ["opus-target"]}]) + + assert router._has_any_configured_fallback("openai/fable-tier", {}) is True + + +def test_has_any_configured_fallback_matches_non_standard_client_fallbacks(): + """Regression: a client-supplied non-standard `fallbacks` list (a plain list of model + names, not keyed by model group at all) applies unconditionally at retry time via + _check_non_standard_fallback_format, so the gate must arm for it too rather than only + recognizing the dict-keyed `{"model_group": [...]}` shape.""" + router = Router(model_list=[FABLE_TIER, OPUS_TARGET]) + + assert router._has_any_configured_fallback("fable-tier", {"fallbacks": ["opus-target"]}) is True + + def test_get_fallback_model_group_for_lookup_groups_orders_tier_before_requested(): router = _router(content_policy_fallbacks=None) fallbacks = [{"tier1": ["backup-a"]}, {"smart-router": ["backup-b"]}] diff --git a/tests/test_litellm/test_router.py b/tests/test_litellm/test_router.py index 1b9ed4b1398..7870f0b7cdc 100644 --- a/tests/test_litellm/test_router.py +++ b/tests/test_litellm/test_router.py @@ -10169,6 +10169,7 @@ class _AnthropicMessagesHangingByteStream: def __init__(self, chunks: list) -> None: self._chunks = list(chunks) self._hidden_params: dict = {} + self.closed = False def __aiter__(self): return self @@ -10180,7 +10181,7 @@ class _AnthropicMessagesHangingByteStream: raise AssertionError("unreachable") # pragma: no cover async def aclose(self) -> None: - pass + self.closed = True class _AnthropicMessagesFallbackByteStream: @@ -10325,6 +10326,34 @@ async def test_anthropic_messages_streaming_iterator_skips_buffering_without_any assert first_chunk == _anthropic_messages_message_start_chunk() +@pytest.mark.asyncio +async def test_anthropic_messages_streaming_iterator_closes_upstream_on_disconnect_without_fallback(): + """Regression: without any fallback configured, the live-forwarding generator must + still close the upstream stream when the caller disconnects mid-stream (aclose() + on the wrapper), not just when the source iterator runs to exhaustion on its own - + otherwise a client that disconnects during a long thinking pass leaks the upstream + request/connection to the provider indefinitely.""" + router = Router( + model_list=[ + { + "model_name": "primary", + "litellm_params": {"model": "anthropic/claude-sonnet-4-5", "api_key": "sk-test"}, + }, + ] + ) + source = _AnthropicMessagesHangingByteStream([_anthropic_messages_message_start_chunk()]) + + wrapped = await router._aanthropic_messages_streaming_iterator( + response=source, initial_kwargs={"model": "primary"} + ) + + await asyncio.wait_for(wrapped.__anext__(), timeout=1.0) + assert source.closed is False + + await wrapped.aclose() + assert source.closed is True + + @pytest.mark.asyncio async def test_anthropic_messages_streaming_iterator_still_buffers_lifecycle_frames_when_fallback_configured(): """Confirms the buffer-until-content protection is still intact once a From 499ecf71574a1ce276217c731500fa43e0dc820c Mon Sep 17 00:00:00 2001 From: nuernber Date: Thu, 3 Sep 2026 16:14:51 -0700 Subject: [PATCH 5/6] fix(router): recognize order-based and weighted-failover retries in the fallback gate _has_any_configured_fallback only checked fallbacks/context_window_fallbacks/ content_policy_fallbacks, missing two retry paths the same dispatcher runs unconditionally: order-based fallback across deployments at different `order` levels, and enable_weighted_failover's intra-group re-pick. Either could still retry a mid-stream error after lifecycle frames were forwarded live, appending a second message_start onto an already-committed stream. --- litellm/router.py | 17 ++++++++++ ...test_router_anthropic_messages_fallback.py | 31 +++++++++++++++++++ 2 files changed, 48 insertions(+) diff --git a/litellm/router.py b/litellm/router.py index 5ab7623aa3e..8c677fc7d63 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -8164,10 +8164,27 @@ class Router: every model group unconditionally rather than being keyed by one at all - self._get_fallback_model_group_for_lookup_groups checks neither, so using it here would report "nothing to fall back to" for a request that a real error would in fact retry. + + Two more retry paths in the same dispatcher fire without any of `fallbacks` / + `context_window_fallbacks` / `content_policy_fallbacks` configured at all: order-based + fallback (deployments in the model group at more than one `order` level) and weighted + intra-group failover (`enable_weighted_failover`), both of which pick a different + deployment for the retry, not the one that already streamed lifecycle frames live. """ fallbacks: Final = kwargs.get("fallbacks", self.fallbacks) if _check_non_standard_fallback_format(fallbacks=fallbacks): return True + if self.enable_weighted_failover: + return True + team_id: Final = (kwargs.get("metadata", {}) or {}).get("user_api_key_team_id") + all_deployments: Final = self.get_model_list(model_name=model_group, team_id=team_id) or [] + order_values: Final = { + litellm.utils._get_deployment_order(d) + for d in all_deployments + if litellm.utils._get_deployment_order(d) is not None + } + if len(order_values) > 1: + return True lookup_groups: Final = fallback_lookup_groups(kwargs, model_group) candidate_fallback_lists: Final = ( fallbacks, diff --git a/tests/router_unit_tests/test_router_anthropic_messages_fallback.py b/tests/router_unit_tests/test_router_anthropic_messages_fallback.py index 4eb4b8d9260..1e10c186571 100644 --- a/tests/router_unit_tests/test_router_anthropic_messages_fallback.py +++ b/tests/router_unit_tests/test_router_anthropic_messages_fallback.py @@ -544,6 +544,37 @@ def test_has_any_configured_fallback_matches_non_standard_client_fallbacks(): assert router._has_any_configured_fallback("fable-tier", {"fallbacks": ["opus-target"]}) is True +def test_has_any_configured_fallback_arms_on_order_based_deployments(): + """Regression: async_function_with_fallbacks_common_utils retries against a higher-order + deployment in the same model group whenever more than one `order` level is present, even + with zero `fallbacks`/`context_window_fallbacks`/`content_policy_fallbacks` configured - + the gate must recognize that retry path too, or a mid-stream error can still trigger an + order-based retry that appends a second message_start onto a stream already forwarded live.""" + router = Router( + model_list=[ + { + "model_name": "fable-tier", + "litellm_params": {"model": "anthropic/claude-fable-5", "api_key": "sk-test", "order": 1}, + }, + { + "model_name": "fable-tier", + "litellm_params": {"model": "anthropic/claude-opus-5", "api_key": "sk-test", "order": 2}, + }, + ] + ) + + assert router._has_any_configured_fallback("fable-tier", {}) is True + + +def test_has_any_configured_fallback_arms_on_weighted_failover(): + """Regression: enable_weighted_failover lets a retryable failure re-pick across the + model group's other deployments before any cross-group fallback runs, independent of + `fallbacks` config entirely - the gate must arm for it too.""" + router = Router(model_list=[FABLE_TIER, OPUS_TARGET], enable_weighted_failover=True) + + assert router._has_any_configured_fallback("fable-tier", {}) is True + + def test_get_fallback_model_group_for_lookup_groups_orders_tier_before_requested(): router = _router(content_policy_fallbacks=None) fallbacks = [{"tier1": ["backup-a"]}, {"smart-router": ["backup-b"]}] From bab7ea4c22a5f2df410e09c053b6d02e69c35eb4 Mon Sep 17 00:00:00 2001 From: nuernber Date: Thu, 3 Sep 2026 16:45:12 -0700 Subject: [PATCH 6/6] fix(router): respect disable_fallbacks and check weighted failover viability Addresses two PR review comments: 1. When disable_fallbacks=True is set on a streaming Anthropic Messages request, skip the buffer-until-content path even if fallbacks are configured. The request explicitly opted out of recovery, so withholding lifecycle frames provides no benefit and only adds latency. 2. When enable_weighted_failover is enabled, only report a recovery path if the routing strategy is simple-shuffle AND there are multiple deployments available. Weighted failover cannot select an alternative deployment for single-deployment groups or non-simple-shuffle strategies, so the unconditional return was incorrectly triggering buffering with no actual fallback protection. Both issues caused Anthropic lifecycle frames (message_start, content_block_start) to be buffered until visible content arrived, preserving the adaptive-thinking delay this PR is intended to remove. --- litellm/router.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/litellm/router.py b/litellm/router.py index 8c677fc7d63..a748d17f63e 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -5166,7 +5166,9 @@ class Router: source_iterator: Final = response model_group: Final = cast(str, initial_kwargs.get("model")) # cast-ok: kwargs always carries the model group - if not self._has_any_configured_fallback(model_group, initial_kwargs): + if fallbacks_disabled_for_request(initial_kwargs) or not self._has_any_configured_fallback( + model_group, initial_kwargs + ): # Nothing to fall back to, so buffering lifecycle frames to protect a mid-stream # fallback attempt would only add latency for no benefit: forward the source # iterator live, exactly as it would stream without this wrapper. @@ -8174,10 +8176,12 @@ class Router: fallbacks: Final = kwargs.get("fallbacks", self.fallbacks) if _check_non_standard_fallback_format(fallbacks=fallbacks): return True - if self.enable_weighted_failover: - return True team_id: Final = (kwargs.get("metadata", {}) or {}).get("user_api_key_team_id") all_deployments: Final = self.get_model_list(model_name=model_group, team_id=team_id) or [] + if self.enable_weighted_failover: + strategy, _ = self._get_routing_context(model_group, kwargs) # pyright: ignore[reportArgumentType] # Mapping is read-only, safe for dict param + if strategy == "simple-shuffle" and len(all_deployments) > 1: + return True order_values: Final = { litellm.utils._get_deployment_order(d) for d in all_deployments