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/litellm/router.py b/litellm/router.py index 3f450661946..a16d26d6825 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -520,6 +520,30 @@ 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. + + 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: """ Bare async generators can't carry the `_hidden_params` attribute the @@ -5244,6 +5268,17 @@ class Router: source_iterator: Final = response + model_group: Final = cast(str, initial_kwargs.get("model")) # cast-ok: kwargs always carries the model group + 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. + 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 @@ -8208,6 +8243,66 @@ 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). + + 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. + + 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 + 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 + 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, + 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 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 + ): + 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..1e10c186571 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,103 @@ 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_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_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"]}] 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() diff --git a/tests/test_litellm/test_router.py b/tests/test_litellm/test_router.py index 7c044310e14..d731ef0b924 100644 --- a/tests/test_litellm/test_router.py +++ b/tests/test_litellm/test_router.py @@ -10642,6 +10642,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=[ { @@ -10657,7 +10664,8 @@ def _anthropic_messages_make_router() -> Router: "model": "bedrock/anthropic.claude-sonnet-4-5", }, }, - ] + ], + fallbacks=[{"primary": ["fallback"]}], ) @@ -10705,6 +10713,30 @@ 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 = {} + self.closed = False + + 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: + self.closed = True + + class _AnthropicMessagesFallbackByteStream: def __init__(self, chunks: list, hidden_params: dict | None = None) -> None: self._chunks = list(chunks) @@ -10817,6 +10849,82 @@ 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_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 + 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