Merge bab7ea4c22 into 4104868458

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:
         """

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

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"]}]

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

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