diff --git a/litellm/llms/anthropic/common_utils.py b/litellm/llms/anthropic/common_utils.py index 9461e40cf2e..53cc3464761 100644 --- a/litellm/llms/anthropic/common_utils.py +++ b/litellm/llms/anthropic/common_utils.py @@ -38,6 +38,21 @@ DROP_DISABLED_THINKING_WARNING: Final = ( "thinking blocks, and those thinking tokens are billed as output tokens." ) +# Anthropic error `type` (both the JSON error body and SSE `event: error` +# payloads use this field) mapped to the HTTP status code it corresponds to. +ANTHROPIC_ERROR_STATUS_CODE_MAP: Final = MappingProxyType( + { + "invalid_request_error": 400, + "authentication_error": 401, + "permission_error": 403, + "not_found_error": 404, + "rate_limit_error": 429, + "api_error": 500, + "overloaded_error": 503, + "timeout_error": 504, + } +) + _BEDROCK_VERSION_SUFFIX_RE: Final = re.compile(r"-v\d+(?::\d+)?$") _INFERENCE_PROFILE_MINOR_RE: Final = re.compile(r":\d+$") _DATED_RELEASE_SUFFIX_RE: Final = re.compile(r"-\d{8}$") diff --git a/litellm/llms/anthropic/experimental_pass_through/messages/streaming_iterator.py b/litellm/llms/anthropic/experimental_pass_through/messages/streaming_iterator.py index 922769dbbfd..0a12bc3135f 100644 --- a/litellm/llms/anthropic/experimental_pass_through/messages/streaming_iterator.py +++ b/litellm/llms/anthropic/experimental_pass_through/messages/streaming_iterator.py @@ -1,6 +1,6 @@ import asyncio import json -from collections.abc import AsyncIterator +from collections.abc import AsyncIterator, Mapping from datetime import datetime from typing import Any, Final, Protocol, runtime_checkable @@ -11,9 +11,11 @@ from typing_extensions import TypedDict from litellm.litellm_core_utils.core_helpers import process_response_headers from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER +from litellm.llms.anthropic.common_utils import ANTHROPIC_ERROR_STATUS_CODE_MAP from litellm.proxy.pass_through_endpoints.success_handler import ( PassThroughEndpointLogging, ) +from litellm.types.llms.anthropic_messages.anthropic_response import AnthropicMessagesResponse from litellm.types.passthrough_endpoints.pass_through_endpoints import EndpointType from litellm.types.utils import GenericStreamingChunk, ModelResponseStream @@ -33,26 +35,239 @@ def _is_message_stop_chunk(chunk: object) -> bool: return False -def _is_provider_error_chunk(chunk: object) -> bool: +def is_anthropic_ping_chunk(chunk: object) -> bool: + """ + Whether a chunk is a pure ``ping`` keepalive frame. It carries no content + and can recur indefinitely on a slow-starting or idle connection, so a + mid-stream fallback wrapper drops it outright while still deciding + whether to commit to the primary stream, rather than buffering it. + + A physical transport chunk that coalesces a ping with any other SSE + event (``message_start``, ``content_block_delta``, ``event: error``, ...) + is NOT a pure ping - dropping it whole would discard those events - so + only a chunk whose every ``event:`` line is ``event: ping`` qualifies. + """ if isinstance(chunk, dict): - return chunk.get("type") == "error" + return chunk.get("type") == "ping" if isinstance(chunk, (bytes, bytearray)): - return any(line == b"event: error" for line in chunk.splitlines()) + event_lines: Final = tuple(line for line in chunk.splitlines() if line.startswith(b"event:")) + return bool(event_lines) and all(line == b"event: ping" for line in event_lines) return False +def is_anthropic_content_delta_chunk(chunk: object) -> bool: + """ + Whether a chunk carries actual assistant-generated output (a + ``content_block_delta`` frame), as opposed to a lifecycle/bookkeeping + frame (``message_start``, ``content_block_start``/``stop``, + ``message_delta``, ``message_stop``, ``ping``) that carries nothing + worth preserving before an invisible mid-stream fallback retry. + """ + if isinstance(chunk, dict): + return chunk.get("type") == "content_block_delta" + if isinstance(chunk, (bytes, bytearray)): + return any(line == b"event: content_block_delta" for line in chunk.splitlines()) + return False + + +def _decoded_sse_data_line(line: bytes) -> object | None: + if not line.startswith(b"data:"): + return None + try: + return json.loads(line[len(b"data:") :].strip()) + except (ValueError, TypeError): + return None + + +def _anthropic_error_event_payload(chunk: object) -> Mapping[str, object] | None: + if isinstance(chunk, dict): + return chunk if chunk.get("type") == "error" else None + if isinstance(chunk, (bytes, bytearray)): + decoded_lines: Final = (_decoded_sse_data_line(line) for line in chunk.splitlines()) + return next( + ( + candidate + for candidate in decoded_lines + if isinstance(candidate, dict) and candidate.get("type") == "error" + ), + None, + ) + return None + + +def _anthropic_error_body(chunk: object) -> Mapping[str, object] | None: + """Return the ``error`` object of an Anthropic SSE ``event: error`` chunk, or None.""" + payload: Final = _anthropic_error_event_payload(chunk) + error_body: Final = payload.get("error") if payload is not None else None + return error_body if isinstance(error_body, dict) else None + + +def _is_provider_error_chunk(chunk: object) -> bool: + return _anthropic_error_body(chunk) is not None + + +def parse_anthropic_error_event(chunk: object) -> tuple[str, str, int] | None: + """ + Extract ``(error_type, message, http_status_code)`` from an Anthropic SSE + ``event: error`` chunk (raw bytes or an already-decoded dict), or None if + ``chunk`` is not an error event. + + The status code is looked up via ANTHROPIC_ERROR_STATUS_CODE_MAP, + defaulting to 500 for an error ``type`` Anthropic hasn't documented yet. + """ + error_body: Final = _anthropic_error_body(chunk) + if error_body is None: + return None + error_type: Final = error_body.get("type") + if not isinstance(error_type, str): + return None + message: Final = error_body.get("message") + return ( + error_type, + message if isinstance(message, str) else error_type, + ANTHROPIC_ERROR_STATUS_CODE_MAP.get(error_type, 500), + ) + + def _is_terminal_stream_chunk(chunk: object) -> bool: return _is_message_stop_chunk(chunk) or _is_provider_error_chunk(chunk) +def _sse_event(event_type: str, payload: Mapping[str, object]) -> bytes: + return f"event: {event_type}\ndata: {json.dumps(payload)}\n\n".encode() + + def _incomplete_stream_error_sse_event() -> bytes: - payload: Final = json.dumps( - { - "type": "error", - "error": {"type": "api_error", "message": INCOMPLETE_STREAM_ERROR_MESSAGE}, - } + return _sse_event( # mutable-ok: one-shot JSON payload, never mutated after construction + "error", + {"type": "error", "error": {"type": "api_error", "message": INCOMPLETE_STREAM_ERROR_MESSAGE}}, + ) + + +def _anthropic_content_block_start_and_deltas( + block: Mapping[str, object], +) -> tuple[Mapping[str, object], tuple[Mapping[str, object], ...]]: + """ + ``(content_block_start.content_block, content_block_delta.delta events)`` + for one Anthropic response content block. A thinking block emits both a + thinking_delta and a trailing signature_delta - a real Anthropic stream + does the same, and dropping the signature makes any replay of that + assistant message (a follow-up turn, a tool-use continuation) fail + Anthropic's thinking-signature verification. redacted_thinking has no + delta at all - it is sent complete in content_block_start. + """ + match block.get("type"): + case "tool_use": + return ( + { # mutable-ok: one-shot payload + "id": block.get("id"), + "name": block.get("name"), + "input": {}, # mutable-ok: one-shot payload + "type": "tool_use", + }, + ( + { # mutable-ok: one-shot payload + "partial_json": json.dumps(block.get("input") or {}), # mutable-ok: one-shot payload + "type": "input_json_delta", + }, + ), + ) + case "thinking": + signature: Final = block.get("signature") + signature_deltas: Final = ( + ({"signature": signature, "type": "signature_delta"},) # mutable-ok: one-shot payload + if isinstance(signature, str) and signature + else () + ) + return ( + {"thinking": "", "signature": "", "type": "thinking"}, # mutable-ok: one-shot payload + ( + {"thinking": block.get("thinking") or "", "type": "thinking_delta"}, # mutable-ok: one-shot payload + *signature_deltas, + ), + ) + case "redacted_thinking": + return ({"type": "redacted_thinking", "data": block.get("data")}, ()) # mutable-ok: one-shot JSON payload + case _: + return ( + {"type": "text", "text": ""}, # mutable-ok: one-shot JSON payload + ({"type": "text_delta", "text": block.get("text") or ""},), # mutable-ok: one-shot JSON payload + ) + + +def anthropic_messages_response_as_sse_events(response: AnthropicMessagesResponse) -> tuple[bytes, ...]: + """ + Render a complete (non-streaming) AnthropicMessagesResponse as the SSE + event sequence a real streaming request would have produced. + + A mid-stream fallback can resolve to a non-streaming response even + though the client asked to stream (e.g. an agentic tool-use loop that + intercepts and returns a complete message) - yielding that dict directly + into a `/v1/messages` SSE byte stream would produce a malformed + response, so it's synthesized into the message_start/content_block_*/ + message_delta/message_stop lifecycle a real stream would have sent. + """ + content_blocks: Final = response.get("content") or () + content_events: Final = ( + event for index, block in enumerate(content_blocks) for event in _anthropic_content_block_events(index, block) + ) + # A real message_start always carries a null stop_reason/stop_sequence and + # a zero output_tokens - those are only known once generation finishes, so + # copying the completed response's final values here would let a client + # treat the message as already finished, or double-count output tokens. + message_start_usage: Final = { # mutable-ok: one-shot JSON payload + **(response.get("usage") or {}), + "output_tokens": 0, + } + message_start_payload: Final = { # mutable-ok: one-shot JSON payload, never mutated after construction + "type": "message_start", + "message": { # mutable-ok: one-shot JSON payload + **response, + "content": [], # mutable-ok: one-shot JSON payload + "stop_reason": None, + "stop_sequence": None, + "usage": message_start_usage, + }, + } + message_delta_payload: Final = { # mutable-ok: one-shot JSON payload, never mutated after construction + "type": "message_delta", + "delta": { # mutable-ok: one-shot JSON payload + "stop_reason": response.get("stop_reason"), + "stop_sequence": response.get("stop_sequence"), + }, + "usage": response.get("usage") or {}, # mutable-ok: one-shot JSON payload + } + return ( + _sse_event("message_start", message_start_payload), + *content_events, + _sse_event("message_delta", message_delta_payload), + _sse_event("message_stop", {"type": "message_stop"}), # mutable-ok: one-shot JSON payload + ) + + +def _anthropic_content_block_events(index: int, block: Mapping[str, object]) -> tuple[bytes, ...]: + start_block, deltas = _anthropic_content_block_start_and_deltas(block) + start_payload: Final = { # mutable-ok: one-shot payload + "type": "content_block_start", + "index": index, + "content_block": start_block, + } + stop_payload: Final = { # mutable-ok: one-shot payload + "type": "content_block_stop", + "index": index, + } + delta_events: Final = tuple( + _sse_event( + "content_block_delta", + {"type": "content_block_delta", "index": index, "delta": delta}, # mutable-ok: one-shot payload + ) + for delta in deltas + ) + return ( + _sse_event("content_block_start", start_payload), + *delta_events, + _sse_event("content_block_stop", stop_payload), ) - return f"event: error\ndata: {payload}\n\n".encode() class AnthropicMessagesStreamHiddenParams(TypedDict): diff --git a/litellm/llms/anthropic/files/handler.py b/litellm/llms/anthropic/files/handler.py index 0c62418708f..5fdf2ceff7f 100644 --- a/litellm/llms/anthropic/files/handler.py +++ b/litellm/llms/anthropic/files/handler.py @@ -22,19 +22,7 @@ from litellm.types.llms.openai import ( from litellm.types.utils import CallTypes, LlmProviders, ModelResponse from ..chat.transformation import AnthropicConfig -from ..common_utils import AnthropicModelInfo - -# Map Anthropic error types to HTTP status codes -ANTHROPIC_ERROR_STATUS_CODE_MAP: Final = { - "invalid_request_error": 400, - "authentication_error": 401, - "permission_error": 403, - "not_found_error": 404, - "rate_limit_error": 429, - "api_error": 500, - "overloaded_error": 503, - "timeout_error": 504, -} +from ..common_utils import ANTHROPIC_ERROR_STATUS_CODE_MAP, AnthropicModelInfo class AnthropicFilesHandler: diff --git a/litellm/router.py b/litellm/router.py index c658bc441ba..d07effd0d90 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -8,6 +8,7 @@ # Thank you ! We ❤️ you! - Krrish & Ishaan import asyncio +import contextlib import copy import enum import hashlib @@ -20,7 +21,7 @@ import time import traceback import weakref from collections import defaultdict -from collections.abc import AsyncGenerator, Callable, Generator, Mapping, Sequence +from collections.abc import AsyncGenerator, AsyncIterator, Callable, Generator, Mapping, Sequence from functools import lru_cache from types import MappingProxyType from typing import TYPE_CHECKING, Any, Final, Literal, Optional, TypeAlias, TypeVar, Union, cast @@ -248,6 +249,7 @@ from .router_utils.pattern_match_deployments import PatternMatchRouter if TYPE_CHECKING: from opentelemetry.trace import Span as _Span + from litellm.exceptions import MidStreamFallbackError from litellm.responses.streaming_iterator import ( BaseResponsesAPIStreamingIterator, ) @@ -264,6 +266,9 @@ if TYPE_CHECKING: from litellm.router_strategy.quality_router.quality_router import ( QualityRouter, ) + from litellm.types.llms.anthropic_messages.anthropic_response import ( + AnthropicMessagesResponse, + ) from litellm.types.llms.base import BaseLiteLLMOpenAIResponseObject from litellm.types.llms.openai import ( ResponseAPIUsage, @@ -361,6 +366,101 @@ def _stream_chunks_have_generated_content(chunks: Sequence[ModelResponseStream]) return False +# Router._aanthropic_messages_streaming_iterator buffers lifecycle chunks +# until real content commits the primary stream; a hostile or slow-starting +# upstream that never emits content or an error could otherwise grow that +# buffer without bound, so hitting this cap forces an early commit instead. +MAX_BUFFERED_PRE_CONTENT_ANTHROPIC_CHUNKS: Final = 200 + + +def _anthropic_stream_should_drop_pre_content_ping(chunk: object, has_generated_content: bool) -> bool: + """A `ping` keepalive seen before any real content is dropped outright - it recurs indefinitely on a + slow-starting connection and carries nothing worth buffering toward a possible fallback.""" + from litellm.llms.anthropic.experimental_pass_through.messages.streaming_iterator import is_anthropic_ping_chunk + + if has_generated_content: + return False + return is_anthropic_ping_chunk(chunk) + + +def _is_retriable_anthropic_status(status_code: int) -> bool: + return status_code == 429 or status_code >= 500 + + +def _anthropic_stream_should_decline_fallback(has_generated_content: bool, error: "MidStreamFallbackError") -> bool: + """ + A MidStreamFallbackError raised directly by the source iterator (the + completion-bridge path's CustomStreamWrapper, e.g. on a transport drop) + carries its own pre_first_chunk bookkeeping - gated the same way a + detected SSE error event is, so a fallback is never appended after real + content already reached the client on either path. + """ + return has_generated_content or not error.is_pre_first_chunk + + +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 + commit to the primary Anthropic stream (real content arrived, or the + pre-content buffer cap was hit) rather than keep buffering lifecycle + frames toward a possible fallback. + """ + from litellm.llms.anthropic.experimental_pass_through.messages.streaming_iterator import ( + is_anthropic_content_delta_chunk, + ) + + if has_generated_content: + return False + return is_anthropic_content_delta_chunk(chunk) or buffered_chunk_count >= MAX_BUFFERED_PRE_CONTENT_ANTHROPIC_CHUNKS + + +class FallbackAwareAnthropicMessagesStream: + """ + Bare async generators can't carry the `_hidden_params` attribute the + proxy reads response headers off of (see + router_utils.add_retry_fallback_headers.get_hidden_params_dict), so this + thin wrapper carries it through from the source iterator - mirrors + AnthropicMessagesStreamingResponse. Used by + Router._aanthropic_messages_streaming_iterator. + """ + + def __init__(self, async_generator: AsyncGenerator[bytes, None], source_iterator: object) -> None: + self._async_generator = async_generator + self._hidden_params = dict( # mutable-ok: mutated in place by merge_fallback_hidden_params + getattr(source_iterator, "_hidden_params", None) or {} + ) + + def __aiter__(self) -> "FallbackAwareAnthropicMessagesStream": + return self + + async def __anext__(self) -> bytes: + return await self._async_generator.__anext__() + + async def aclose(self) -> None: + await self._async_generator.aclose() + + def merge_fallback_hidden_params( + self, + fallback_hidden_params: Mapping[str, object], + fallback_headers: Mapping[str, object], + ) -> None: + """ + Raw bytes can't carry their own _hidden_params the way a + ModelResponseStream/ResponsesAPI event can, so a mid-stream + fallback's provider headers (e.g. Bedrock's x-amzn-requestid) are + merged onto the wrapper itself instead - mirrors + Router._apply_fallback_hidden_params_to_item's merge shape. + """ + existing_headers: Final = cast( # cast-ok: additional_headers is always a dict[str, object] when present + "dict[str, object]", self._hidden_params.get("additional_headers") or {} + ) + self._hidden_params = { # mutable-ok: matches _hidden_params' existing dict[str, object] shape + **self._hidden_params, + **fallback_hidden_params, + "additional_headers": {**existing_headers, **fallback_headers}, # mutable-ok: same shape + } + + class RoutingArgs(enum.Enum): ttl = 60 # 1min (RPM/TPM expire key) @@ -4806,6 +4906,264 @@ class Router: ) return response + async def _aanthropic_messages_streaming_iterator( + self, + response: AsyncIterator[bytes], + initial_kwargs: dict[str, Any], # mutable-ok: mutated in-place before re-entering the fallback chain + ) -> AsyncIterator[bytes]: + """ + Wrap an anthropic_messages (/v1/messages) streaming response so a + mid-stream provider error triggers the Router's fallback chain + (parity with _acompletion_streaming_iterator for the + chat-completions path). See #24004. + + anthropic_messages goes through _ageneric_api_call_with_fallbacks + rather than _acompletion, so the returned byte iterator is never + wrapped by the chat-completions fallback handler. Two failure + shapes land here: + - the completion-bridge path (deployments with no native + /v1/messages endpoint, via + LiteLLMMessagesToCompletionTransformationHandler) already + raises MidStreamFallbackError out of its underlying + CustomStreamWrapper; this wrapper only needs to catch it. + - a native Anthropic/Bedrock passthrough never raises anything + for a provider SSE `event: error` frame (e.g. `overloaded_error`, + `internal_server_error`) - it is forwarded to the client as-is - + so this wrapper detects it via parse_anthropic_error_event and + raises MidStreamFallbackError itself. + + Only an error before any real content (a content_block_delta frame) + has reached the caller triggers a fallback attempt, mirroring the + restriction _acompletion_streaming_iterator applies: once generated + output has already reached the caller, retrying would start a + second, overlapping Anthropic message lifecycle on the same SSE + stream, so the error is left to propagate instead of being retried + invisibly. A non-retriable client error (4xx other than 429) is + never worth a fallback attempt either, so it is also left to + propagate. + + Lifecycle/bookkeeping frames (message_start, content_block_start, + ping, ...) do not by themselves disqualify a fallback attempt - + Anthropic routinely sends message_start before an overload error - + but they are BUFFERED rather than forwarded immediately, since + forwarding one and then appending a fallback attempt's own + message_start would produce two overlapping message lifecycles on + one SSE stream. Buffered frames are flushed, in order, the moment + real content arrives (the primary attempt has committed by then + anyway) or once the stream ends without ever producing content or + an error. + """ + from litellm.llms.anthropic.experimental_pass_through.messages.streaming_iterator import ( + aclose_if_supported, + parse_anthropic_error_event, + ) + + source_iterator: Final = response + + async def stream_with_fallbacks() -> AsyncGenerator[bytes, None]: + from litellm.exceptions import MidStreamFallbackError + + # Lifecycle/bookkeeping frames (message_start, content_block_start, + # ping, ...) are held back rather than forwarded immediately: + # Anthropic routinely sends message_start before an overload + # error, and once a byte reaches the client a fallback attempt + # can only append its OWN message_start, producing two + # overlapping message lifecycles on one SSE stream. Buffered + # frames are flushed the moment real content (content_block_delta) + # arrives - at that point the primary attempt has committed and a + # clean retry is no longer possible anyway - or once the primary + # stream ends without ever producing content. A `ping` keepalive + # is dropped outright rather than buffered, since it can recur + # indefinitely on a slow-starting connection and carries nothing + # worth preserving; hitting MAX_BUFFERED_PRE_CONTENT_ANTHROPIC_CHUNKS + # forces the same early commit as real content arriving, so a + # hostile or pathological upstream can't grow the buffer forever. + 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 + try: + async for chunk in source_iterator: + if _anthropic_stream_should_drop_pre_content_ping(chunk, has_generated_content): + continue + if _anthropic_stream_commits_now(chunk, has_generated_content, len(buffered_lifecycle_chunks)): + has_generated_content = True # rebind-ok: real content seen, or the buffer cap was hit + error_event = parse_anthropic_error_event(chunk) + retriable_pending_error = ( # rebind-ok: freshly computed each iteration, never carried over + not has_generated_content + and error_event is not None + and _is_retriable_anthropic_status(error_event[2]) + ) + if not has_generated_content and not retriable_pending_error and error_event is None: + buffered_lifecycle_chunks = (*buffered_lifecycle_chunks, chunk) + continue + if retriable_pending_error: + assert error_event is not None # guard-ok: retriable_pending_error implies this + _error_type, message, status_code = error_event + raise MidStreamFallbackError( + message=message, + model=model, + llm_provider="anthropic", + original_exception=litellm.exceptions.APIError( + status_code=status_code, + message=message, + llm_provider="anthropic", + model=model, + ), + is_pre_first_chunk=True, + ) + for buffered_chunk in buffered_lifecycle_chunks: + yield buffered_chunk + buffered_lifecycle_chunks = () + 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): + yield item + finally: + with anyio.CancelScope(shield=True), contextlib.suppress(BaseException): + await aclose_if_supported(source_iterator) + + # Referenced by stream_with_fallbacks via closure - assigned here, before + # the generator body ever runs, so the reference resolves fine despite + # being defined textually after the function that captures it. + wrapper: Final = FallbackAwareAnthropicMessagesStream(stream_with_fallbacks(), source_iterator) + return wrapper + + async def _aanthropic_messages_fallback_attempt( + self, + e: "MidStreamFallbackError", + initial_kwargs: dict[str, Any], # mutable-ok: mutated in-place before re-entering the fallback chain + wrapper: "FallbackAwareAnthropicMessagesStream", + ) -> AsyncGenerator[bytes, None]: + """ + Re-enters the Router's fallback chain for a mid-stream + anthropic_messages error and yields whatever the fallback attempt + produces. Split out of _aanthropic_messages_streaming_iterator to + keep each function's cyclomatic complexity within the repo's C901 + budget. + """ + from litellm.exceptions import MidStreamFallbackError + from litellm.llms.anthropic.experimental_pass_through.messages.streaming_iterator import ( + aclose_if_supported, + anthropic_messages_response_as_sse_events, + ) + + fallback_response = None # rebind-ok: pre-init so finally can close it if a fallback was actually attempted + try: + model_group: Final = cast(str, initial_kwargs.get("model")) # cast-ok: model group + fallbacks: Final[list | None] = initial_kwargs.get( # mutable-ok: matches the common_utils list|None param + "fallbacks", self.fallbacks + ) + context_window_fallbacks: Final[list | None] = initial_kwargs.get( # mutable-ok: matches the param below + "context_window_fallbacks", self.context_window_fallbacks + ) + content_policy_fallbacks: Final[list | None] = initial_kwargs.get( # mutable-ok: matches the param below + "content_policy_fallbacks", self.content_policy_fallbacks + ) + initial_kwargs["original_function"] = self._ageneric_api_call_with_fallbacks_helper + self._update_kwargs_before_fallbacks( + model=model_group, + kwargs=initial_kwargs, + metadata_variable_name="litellm_metadata", + ) + fallback_response = await self.async_function_with_fallbacks_common_utils( # rebind-ok: set on success + e=e, + disable_fallbacks=False, + fallbacks=fallbacks, + context_window_fallbacks=context_window_fallbacks, + content_policy_fallbacks=content_policy_fallbacks, + model_group=model_group, + args=(), + kwargs=initial_kwargs, + include_fallback_errors=initial_kwargs.get("include_fallback_errors", False) is True, + ) + fallback_hidden_params, fallback_headers = Router._prepare_fallback_hidden_params(fallback_response) + wrapper.merge_fallback_hidden_params(fallback_hidden_params, fallback_headers) + if hasattr(fallback_response, "__aiter__"): + async for fallback_item in fallback_response: + yield fallback_item + else: + # A fallback can resolve to a complete AnthropicMessagesResponse + # dict even for a streaming request (e.g. an agentic tool-use + # interception loop) - yielding it as-is would put a raw dict + # into a byte stream, so it's synthesized into the SSE + # lifecycle a real stream would have sent instead. + for event in anthropic_messages_response_as_sse_events( + cast("AnthropicMessagesResponse", fallback_response) # cast-ok: non-streaming shape by elimination + ): + yield event + except Exception as fallback_error: + verbose_router_logger.error("Anthropic messages streaming fallback also failed: %s", fallback_error) + if isinstance(fallback_error, MidStreamFallbackError) and fallback_error.original_exception is not None: + raise fallback_error.original_exception from fallback_error + raise + finally: + if fallback_response is not None: + with anyio.CancelScope(shield=True), contextlib.suppress(BaseException): + await aclose_if_supported(fallback_response) + + async def _aanthropic_messages_with_streaming_fallbacks( + self, + original_function: Callable, + **kwargs: object, # kwargs-ok: forwarded verbatim to original_function, shape varies per call site + ) -> Union["AnthropicMessagesResponse", AsyncIterator[bytes]]: + """ + _ageneric_api_call_with_fallbacks for anthropic_messages, with the + addition of mid-stream fallback handling (see + _aanthropic_messages_streaming_iterator). Parity with + _aresponses_with_streaming_fallbacks for the Responses API. + """ + from litellm.litellm_core_utils.core_helpers import safe_deep_copy + + # Snapshot the request kwargs before the primary attempt mutates them + # in place: _update_kwargs_with_deployment writes deployment-specific + # fields (deployment, model_info, api_base, tags, ...) into the + # SAME litellm_metadata/metadata dicts a shallow .copy() would still + # share, leaking primary-deployment metadata into the mid-stream + # fallback request. safe_deep_copy avoids deep-copying the full + # kwargs (which can hold non-deepcopyable logging handles/clients). + fallback_kwargs: Final[dict[str, object]] = kwargs.copy() # mutable-ok: mutated below before re-entry + if isinstance(fallback_kwargs.get("litellm_metadata"), dict): + fallback_kwargs["litellm_metadata"] = safe_deep_copy(fallback_kwargs["litellm_metadata"]) + if isinstance(fallback_kwargs.get("metadata"), dict): + fallback_kwargs["metadata"] = safe_deep_copy(fallback_kwargs["metadata"]) + fallback_kwargs["original_generic_function"] = original_function + + response: Final = await self._ageneric_api_call_with_fallbacks(original_function=original_function, **kwargs) + + if kwargs.get("stream") and hasattr(response, "__aiter__"): + return await self._aanthropic_messages_streaming_iterator( + response=cast("AsyncIterator[bytes]", response), # cast-ok: stream=True always returns a byte iterator + initial_kwargs=fallback_kwargs, + ) + return response + + async def _dispatch_generic_call_type( + self, + call_type: str, + original_function: Callable, + **kwargs: object, # kwargs-ok: forwarded verbatim to the per-call-type helper, shape varies per call site + ): + """ + factory_function's shared dispatch for call types with no + call-specific handling, except anthropic_messages: kept out of + factory_function's own async_wrapper (already at the repo's C901 + complexity ceiling) so routing its mid-stream fallback handling + (#24004) doesn't add another branch there. + """ + if call_type == "anthropic_messages": + return await self._aanthropic_messages_with_streaming_fallbacks( + original_function=original_function, **kwargs + ) + return await self._ageneric_api_call_with_fallbacks(original_function=original_function, **kwargs) + def _generic_api_call_with_fallbacks(self, model: str, original_function: Callable, **kwargs): """ Make a generic LLM API call through the router, this allows you to use retries/fallbacks with litellm router @@ -5992,7 +6350,8 @@ class Router: "aget_skill", "adelete_skill", ): - return await self._ageneric_api_call_with_fallbacks( + return await self._dispatch_generic_call_type( + call_type=call_type, original_function=original_function, **kwargs, ) diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_streaming_iterator.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_streaming_iterator.py index f33bb3dda8b..652c1f077a9 100644 --- a/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_streaming_iterator.py +++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_streaming_iterator.py @@ -11,6 +11,10 @@ from litellm.llms.anthropic.experimental_pass_through.messages.streaming_iterato BaseAnthropicMessagesStreamingIterator, _incomplete_stream_error_sse_event, _is_message_stop_chunk, + _is_provider_error_chunk, + anthropic_messages_response_as_sse_events, + is_anthropic_content_delta_chunk, + parse_anthropic_error_event, ) @@ -157,6 +161,96 @@ def test_is_message_stop_chunk_ignores_substring_in_payload(): assert _is_message_stop_chunk(delta_frame_with_substring) is False +def test_parse_anthropic_error_event_from_dict_chunk(): + """Regression for #24004: dict-shaped error chunks parse to + (type, message, status) so the Router can decide whether to fall back.""" + chunk = {"type": "error", "error": {"type": "overloaded_error", "message": "Overloaded"}} + assert parse_anthropic_error_event(chunk) == ("overloaded_error", "Overloaded", 503) + assert _is_provider_error_chunk(chunk) is True + + +def test_parse_anthropic_error_event_from_sse_bytes(): + """Regression for #24004: a raw `event: error` SSE frame (what a native + Anthropic/Bedrock passthrough forwards verbatim today) must parse + identically to the dict shape so the Router can raise a fallback.""" + sse_chunk = ( + b"event: error\n" + b'data: {"type": "error", "error": {"type": "internal_server_error", "message": "boom"}}\n\n' + ) + assert parse_anthropic_error_event(sse_chunk) == ("internal_server_error", "boom", 500) + assert _is_provider_error_chunk(sse_chunk) is True + + +def test_parse_anthropic_error_event_defaults_status_for_unknown_type(): + chunk = {"type": "error", "error": {"type": "some_future_error_type", "message": "?"}} + assert parse_anthropic_error_event(chunk) == ("some_future_error_type", "?", 500) + + +def test_parse_anthropic_error_event_missing_message_falls_back_to_type(): + chunk = {"type": "error", "error": {"type": "overloaded_error"}} + assert parse_anthropic_error_event(chunk) == ("overloaded_error", "overloaded_error", 503) + + +def test_parse_anthropic_error_event_non_string_error_type_returns_none(): + """A malformed error body whose `type` field isn't a string (e.g. an + upstream bug sends null or a number) must not be treated as an error + event rather than crashing or forwarding a garbage error_type.""" + chunk = {"type": "error", "error": {"type": None, "message": "boom"}} + assert parse_anthropic_error_event(chunk) is None + + +def test_decoded_sse_data_line_swallows_invalid_json(): + """A `data:` line that isn't valid JSON (a malformed/truncated frame) + must not be treated as an error event or raise, just be ignored.""" + malformed_frame = b"event: error\ndata: {not valid json\n\n" + assert parse_anthropic_error_event(malformed_frame) is None + assert _is_provider_error_chunk(malformed_frame) is False + + +class TestIsAnthropicContentDeltaChunk: + def test_dict_content_block_delta(self): + assert is_anthropic_content_delta_chunk({"type": "content_block_delta"}) is True + + def test_dict_other_type(self): + assert is_anthropic_content_delta_chunk({"type": "message_start"}) is False + + def test_bytes_content_block_delta(self): + assert is_anthropic_content_delta_chunk(b"event: content_block_delta\ndata: {}\n\n") is True + + def test_bytes_other_event(self): + assert is_anthropic_content_delta_chunk(b"event: message_start\ndata: {}\n\n") is False + + def test_neither_dict_nor_bytes(self): + assert is_anthropic_content_delta_chunk("content_block_delta") is False + assert is_anthropic_content_delta_chunk(None) is False + + +@pytest.mark.parametrize( + "chunk", + [ + {"type": "content_block_delta", "delta": {"type": "text_delta", "text": "hi"}}, + b'event: content_block_delta\ndata: {"type": "content_block_delta"}\n\n', + b"raw-bytes", + "error", + None, + ], +) +def test_parse_anthropic_error_event_non_error_chunks_return_none(chunk): + assert parse_anthropic_error_event(chunk) is None + assert _is_provider_error_chunk(chunk) is False + + +def test_parse_anthropic_error_event_ignores_substring_in_payload(): + """A content_block_delta whose partial_json happens to contain the + literal string `"type": "error"` must not be misread as an error event.""" + delta_frame_with_substring = ( + b"event: content_block_delta\n" + b'data: {"type": "content_block_delta", "delta": ' + b'{"type": "input_json_delta", "partial_json": "\\"type\\": \\"error\\""}}\n\n' + ) + assert parse_anthropic_error_event(delta_frame_with_substring) is None + + @pytest.mark.asyncio async def test_async_sse_wrapper_emits_error_when_bytes_stream_only_mentions_message_stop_in_payload(): """ @@ -307,3 +401,117 @@ def test_incomplete_stream_error_sse_event_is_valid_anthropic_error(): "error": {"type": "api_error", "message": INCOMPLETE_STREAM_ERROR_MESSAGE}, } assert event.endswith("\n\n") + + +def _decode_sse_events(events: tuple[bytes, ...]) -> list[tuple[str, dict]]: + decoded = [] + for event in events: + assert isinstance(event, bytes) + lines = event.decode().split("\n") + assert lines[0].startswith("event: ") + decoded.append((lines[0].removeprefix("event: "), json.loads(lines[1].removeprefix("data: ")))) + return decoded + + +def test_anthropic_messages_response_as_sse_events_text_block(): + response = { + "id": "msg_1", + "model": "claude-haiku", + "role": "assistant", + "type": "message", + "stop_reason": "end_turn", + "stop_sequence": None, + "content": [{"type": "text", "text": "hello"}], + "usage": {"input_tokens": 3, "output_tokens": 2}, + } + decoded = _decode_sse_events(anthropic_messages_response_as_sse_events(response)) + types = [event_type for event_type, _ in decoded] + assert types == [ + "message_start", + "content_block_start", + "content_block_delta", + "content_block_stop", + "message_delta", + "message_stop", + ] + # message_start must not carry generated content itself, matching a real + # streaming response - it arrives via the content_block_delta that follows. + assert decoded[0][1]["message"]["content"] == [] + assert decoded[0][1]["message"]["id"] == "msg_1" + # Bugbot regression: message_start must not carry the completed response's + # final stop_reason/stop_sequence/output_tokens - a real stream keeps those + # null/zero until message_delta, so a client could otherwise treat the + # message as already finished, or double-count output tokens. + assert decoded[0][1]["message"]["stop_reason"] is None + assert decoded[0][1]["message"]["stop_sequence"] is None + assert decoded[0][1]["message"]["usage"] == {"input_tokens": 3, "output_tokens": 0} + assert decoded[1][1]["content_block"] == {"type": "text", "text": ""} + assert decoded[2][1]["delta"] == {"type": "text_delta", "text": "hello"} + assert decoded[4][1]["delta"]["stop_reason"] == "end_turn" + assert decoded[4][1]["usage"] == {"input_tokens": 3, "output_tokens": 2} + + +def test_anthropic_messages_response_as_sse_events_tool_use_block(): + response = { + "id": "msg_2", + "content": [{"type": "tool_use", "id": "toolu_1", "name": "get_weather", "input": {"city": "NYC"}}], + "stop_reason": "tool_use", + } + decoded = _decode_sse_events(anthropic_messages_response_as_sse_events(response)) + content_block_start = dict(decoded)["content_block_start"] + assert content_block_start["content_block"] == { + "type": "tool_use", + "id": "toolu_1", + "name": "get_weather", + "input": {}, + } + content_block_delta = dict(decoded)["content_block_delta"] + assert json.loads(content_block_delta["delta"]["partial_json"]) == {"city": "NYC"} + assert content_block_delta["delta"]["type"] == "input_json_delta" + + +def test_anthropic_messages_response_as_sse_events_thinking_block_emits_signature_delta(): + """Bugbot regression: a thinking block's real `signature` must reach the + client via a trailing signature_delta, not be silently dropped - Anthropic + rejects a replayed assistant message (a follow-up turn, a tool-use + continuation) whose thinking block lacks its original signature.""" + response = { + "id": "msg_5", + "content": [{"type": "thinking", "thinking": "let me think", "signature": "sig-abc123"}], + "stop_reason": "end_turn", + } + decoded = _decode_sse_events(anthropic_messages_response_as_sse_events(response)) + deltas = [payload["delta"] for event_type, payload in decoded if event_type == "content_block_delta"] + assert deltas == [ + {"type": "thinking_delta", "thinking": "let me think"}, + {"type": "signature_delta", "signature": "sig-abc123"}, + ] + + +def test_anthropic_messages_response_as_sse_events_thinking_block_without_signature_omits_delta(): + response = { + "id": "msg_6", + "content": [{"type": "thinking", "thinking": "let me think", "signature": None}], + "stop_reason": "end_turn", + } + decoded = _decode_sse_events(anthropic_messages_response_as_sse_events(response)) + deltas = [payload["delta"] for event_type, payload in decoded if event_type == "content_block_delta"] + assert deltas == [{"type": "thinking_delta", "thinking": "let me think"}] + + +def test_anthropic_messages_response_as_sse_events_multiple_blocks_are_indexed(): + response = { + "id": "msg_3", + "content": [{"type": "text", "text": "a"}, {"type": "text", "text": "b"}], + } + decoded = _decode_sse_events(anthropic_messages_response_as_sse_events(response)) + starts = [payload for event_type, payload in decoded if event_type == "content_block_start"] + assert [s["index"] for s in starts] == [0, 1] + deltas = [payload for event_type, payload in decoded if event_type == "content_block_delta"] + assert [d["delta"]["text"] for d in deltas] == ["a", "b"] + + +def test_anthropic_messages_response_as_sse_events_no_content_blocks(): + response = {"id": "msg_4", "content": [], "stop_reason": "end_turn"} + decoded = _decode_sse_events(anthropic_messages_response_as_sse_events(response)) + assert [event_type for event_type, _ in decoded] == ["message_start", "message_delta", "message_stop"] diff --git a/tests/test_litellm/test_router.py b/tests/test_litellm/test_router.py index 0b82ab971bf..56fd6df446f 100644 --- a/tests/test_litellm/test_router.py +++ b/tests/test_litellm/test_router.py @@ -12,8 +12,17 @@ import pytest import litellm +from litellm import Router from litellm.exceptions import MidStreamFallbackError from litellm.integrations.custom_logger import CustomLogger +from litellm.router import ( + MAX_BUFFERED_PRE_CONTENT_ANTHROPIC_CHUNKS, + FallbackAwareAnthropicMessagesStream, + _anthropic_stream_commits_now, + _anthropic_stream_should_decline_fallback, + _anthropic_stream_should_drop_pre_content_ping, + _is_retriable_anthropic_status, +) def test_update_kwargs_does_not_mutate_defaults_and_merges_metadata(): @@ -9224,3 +9233,1095 @@ class TestAddDeploymentApiBaseProviderResolution: deployment = router.get_deployment_by_model_group_name("openai-via-gateway") assert deployment is not None assert deployment.litellm_params.custom_llm_provider == "openai" + +# ===================================================================== +# anthropic_messages mid-stream-fallback helpers, added for #24004 +# (mid-stream fallback not supported for anthropic_messages route type). +# +# anthropic_messages goes through _ageneric_api_call_with_fallbacks rather +# than _acompletion, so its returned iterator was never wrapped by the chat +# completions fallback handler: an SSE `event: error` frame from a native +# Anthropic/Bedrock passthrough passed through to the client silently, and a +# MidStreamFallbackError raised by the completion-bridge path's +# CustomStreamWrapper (e.g. a Vertex AI transport drop) propagated +# unhandled. +# +# Targets the helpers introduced on Router: +# - _aanthropic_messages_streaming_iterator +# - _aanthropic_messages_fallback_attempt +# - _aanthropic_messages_with_streaming_fallbacks +# - _dispatch_generic_call_type +# ===================================================================== + + +async def _anthropic_messages_empty_generator(): + return + yield # pragma: no cover - makes this an async generator + + +def _anthropic_messages_make_wrapper() -> FallbackAwareAnthropicMessagesStream: + """A minimal wrapper for tests that call _aanthropic_messages_fallback_attempt + directly, bypassing _aanthropic_messages_streaming_iterator.""" + return FallbackAwareAnthropicMessagesStream(_anthropic_messages_empty_generator(), object()) + + +def _anthropic_messages_make_router() -> Router: + return Router( + model_list=[ + { + "model_name": "primary", + "litellm_params": { + "model": "anthropic/claude-sonnet-4-5", + "api_key": "sk-test", + }, + }, + { + "model_name": "fallback", + "litellm_params": { + "model": "bedrock/anthropic.claude-sonnet-4-5", + }, + }, + ] + ) + + +class _AnthropicMessagesFakeByteStream: + """Minimal AsyncIterator[bytes], carrying _hidden_params like + AnthropicMessagesStreamingResponse does.""" + + def __init__(self, chunks: list) -> None: + self._chunks = list(chunks) + self._hidden_params = {"additional_headers": {"x-amzn-requestid": "req-1"}} + self.closed = False + + def __aiter__(self): + return self + + async def __anext__(self) -> bytes: + if not self._chunks: + raise StopAsyncIteration + return self._chunks.pop(0) + + async def aclose(self) -> None: + self.closed = True + + +class _AnthropicMessagesRaisingByteStream: + """Simulates the completion-bridge path: no error SSE chunk is ever + yielded, the underlying CustomStreamWrapper raises MidStreamFallbackError + directly out of the iterator instead (a Vertex AI transport drop).""" + + def __init__(self, chunks: list, error: Exception) -> None: + self._chunks = list(chunks) + self._error = error + 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) + raise self._error + + 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) + self._hidden_params = hidden_params if hidden_params is not None else {} + + def __aiter__(self): + return self + + async def __anext__(self) -> bytes: + if not self._chunks: + raise StopAsyncIteration + return self._chunks.pop(0) + + +def _anthropic_messages_overloaded_error_chunk() -> bytes: + return ( + b"event: error\n" + b'data: {"type": "error", "error": {"type": "overloaded_error", "message": "Overloaded"}}\n\n' + ) + + +def _anthropic_messages_invalid_request_error_chunk() -> bytes: + return ( + b"event: error\n" + b'data: {"type": "error", "error": {"type": "invalid_request_error", "message": "bad request"}}\n\n' + ) + + +def _anthropic_messages_rate_limit_error_chunk() -> bytes: + return ( + b"event: error\n" + b'data: {"type": "error", "error": {"type": "rate_limit_error", "message": "Too many requests"}}\n\n' + ) + + +def _anthropic_messages_content_chunk(text: str = "hi") -> bytes: + payload = f'{{"type": "content_block_delta", "delta": {{"type": "text_delta", "text": "{text}"}}}}' + return f"event: content_block_delta\ndata: {payload}\n\n".encode() + + +def _anthropic_messages_message_start_chunk() -> bytes: + """A lifecycle/bookkeeping frame Anthropic sends before any real content - + routinely the very first event before an overload error.""" + return b'event: message_start\ndata: {"type": "message_start", "message": {"id": "msg_1"}}\n\n' + + +def _anthropic_messages_ping_chunk() -> bytes: + return b'event: ping\ndata: {"type": "ping"}\n\n' + + +# -------- _aanthropic_messages_streaming_iterator (passthrough) -------- + + +@pytest.mark.asyncio +async def test_anthropic_messages_streaming_iterator_passthrough(): + """Without any error chunk, the wrapper forwards every chunk unchanged + and carries the source iterator's _hidden_params through (so response + headers like Bedrock's request-id keep flowing to the client).""" + router = _anthropic_messages_make_router() + source = _AnthropicMessagesFakeByteStream( + [_anthropic_messages_content_chunk("hi"), _anthropic_messages_content_chunk(" there")] + ) + + 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("hi"), _anthropic_messages_content_chunk(" there")] + assert wrapped._hidden_params["additional_headers"]["x-amzn-requestid"] == "req-1" + + +@pytest.mark.asyncio +async def test_anthropic_messages_streaming_iterator_flushes_buffered_lifecycle_frames_in_order(): + """Regression: lifecycle frames held back to guard against a mid-stream + fallback must still reach the client, in order, once real content + arrives - buffering them for the fallback-safety check must not silently + drop them on the happy path.""" + router = _anthropic_messages_make_router() + message_stop = b'event: message_stop\ndata: {"type": "message_stop"}\n\n' + source = _AnthropicMessagesFakeByteStream( + [_anthropic_messages_message_start_chunk(), _anthropic_messages_content_chunk("hi"), message_stop] + ) + + wrapped = await router._aanthropic_messages_streaming_iterator( + response=source, initial_kwargs={"model": "primary"} + ) + + collected = [chunk async for chunk in wrapped] + assert collected == [_anthropic_messages_message_start_chunk(), _anthropic_messages_content_chunk("hi"), message_stop] + + +@pytest.mark.asyncio +async def test_anthropic_messages_streaming_iterator_flushes_buffered_frames_on_stream_end(): + """Regression: if the primary stream ends with only lifecycle frames and + no content and no error, the buffered frames must still reach the + client rather than being silently swallowed.""" + router = _anthropic_messages_make_router() + message_stop = b'event: message_stop\ndata: {"type": "message_stop"}\n\n' + source = _AnthropicMessagesFakeByteStream([_anthropic_messages_message_start_chunk(), message_stop]) + + wrapped = await router._aanthropic_messages_streaming_iterator( + response=source, initial_kwargs={"model": "primary"} + ) + + collected = [chunk async for chunk in wrapped] + assert collected == [_anthropic_messages_message_start_chunk(), message_stop] + + with pytest.raises(StopAsyncIteration): + await wrapped.__anext__() + + +@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 + content_block_delta and a following retriable error into ONE physical + read from the source iterator. Since the whole chunk (content and error + together) is forwarded to the client atomically, the client genuinely + receives the content - so no fallback must be attempted, exactly as if + the two events had arrived as separate reads.""" + router = _anthropic_messages_make_router() + coalesced_chunk = _anthropic_messages_content_chunk("partial") + _anthropic_messages_overloaded_error_chunk() + source = _AnthropicMessagesFakeByteStream([coalesced_chunk]) + + with patch.object( + router, + "async_function_with_fallbacks_common_utils", + new=AsyncMock(return_value=_AnthropicMessagesFallbackByteStream([])), + ) 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 == [coalesced_chunk] + mock_fallback.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_anthropic_messages_ping_keepalive_never_buffered_or_forwarded(): + """Bugbot regression: a `ping` keepalive carries no content and must be + dropped outright before any real content arrives, rather than buffered - + otherwise a slow-starting connection sending many pings could grow the + pre-content buffer without bound.""" + router = _anthropic_messages_make_router() + source = _AnthropicMessagesFakeByteStream( + [_anthropic_messages_ping_chunk(), _anthropic_messages_content_chunk("hi")] + ) + + wrapped = await router._aanthropic_messages_streaming_iterator(response=source, initial_kwargs={"model": "primary"}) + collected = [chunk async for chunk in wrapped] + + assert _anthropic_messages_ping_chunk() not in collected + assert collected == [_anthropic_messages_content_chunk("hi")] + + +@pytest.mark.asyncio +async def test_anthropic_messages_pre_content_buffer_cap_forces_commit(): + """Bugbot regression: a hostile or pathological upstream that never emits + real content or an error must not grow the pre-content lifecycle buffer + without bound - hitting MAX_BUFFERED_PRE_CONTENT_ANTHROPIC_CHUNKS commits + to the primary stream early, exactly as real content arriving would.""" + router = _anthropic_messages_make_router() + lifecycle_chunk = _anthropic_messages_message_start_chunk() + error_chunk = _anthropic_messages_overloaded_error_chunk() + chunks = [lifecycle_chunk] * (MAX_BUFFERED_PRE_CONTENT_ANTHROPIC_CHUNKS + 5) + [error_chunk] + source = _AnthropicMessagesFakeByteStream(chunks) + + with patch.object( + router, + "async_function_with_fallbacks_common_utils", + new=AsyncMock(return_value=_AnthropicMessagesFallbackByteStream([])), + ) as mock_fallback: + wrapped = await router._aanthropic_messages_streaming_iterator( + response=source, initial_kwargs={"model": "primary"} + ) + collected = [chunk async for chunk in wrapped] + + mock_fallback.assert_not_awaited() + assert collected.count(lifecycle_chunk) == MAX_BUFFERED_PRE_CONTENT_ANTHROPIC_CHUNKS + 5 + assert collected[-1] == error_chunk + + +@pytest.mark.asyncio +async def test_anthropic_messages_ping_coalesced_with_content_in_one_physical_chunk_is_forwarded(): + """Greptile/Bugbot regression: transport-level buffering can coalesce a + `ping` keepalive and a real content_block_delta into ONE physical read. + The pre-content ping-drop must only discard PURE ping frames - dropping + the whole coalesced chunk would silently lose generated content.""" + router = _anthropic_messages_make_router() + coalesced_chunk = _anthropic_messages_ping_chunk() + _anthropic_messages_content_chunk("hi") + source = _AnthropicMessagesFakeByteStream([coalesced_chunk]) + + wrapped = await router._aanthropic_messages_streaming_iterator(response=source, initial_kwargs={"model": "primary"}) + collected = [chunk async for chunk in wrapped] + + assert collected == [coalesced_chunk] + + +@pytest.mark.asyncio +async def test_anthropic_messages_ping_coalesced_with_retriable_error_still_falls_back(): + """Greptile/Bugbot regression: a physical chunk coalescing a `ping` with a + retriable `event: error` must not be discarded as a keepalive - the error + inside it must still trigger the mid-stream fallback.""" + router = _anthropic_messages_make_router() + coalesced_chunk = _anthropic_messages_ping_chunk() + _anthropic_messages_overloaded_error_chunk() + source = _AnthropicMessagesFakeByteStream([coalesced_chunk]) + 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] + + mock_fallback.assert_awaited_once() + assert collected == [_anthropic_messages_content_chunk("fallback answer")] + + +# -------- _aanthropic_messages_fallback_attempt -------- + + +@pytest.mark.asyncio +async def test_aanthropic_messages_fallback_attempt_yields_fallback_stream(): + """Direct-call regression: the fallback-attempt helper re-enters the + Router's fallback chain and forwards whatever the fallback produces.""" + router = _anthropic_messages_make_router() + fallback_stream = _AnthropicMessagesFallbackByteStream([_anthropic_messages_content_chunk("fallback answer")]) + error = MidStreamFallbackError(message="overloaded", model="primary", llm_provider="anthropic") + + with patch.object( + router, + "async_function_with_fallbacks_common_utils", + new=AsyncMock(return_value=fallback_stream), + ) as mock_fallback: + collected = [ + chunk + async for chunk in router._aanthropic_messages_fallback_attempt( + error, + {"model": "primary", "messages": [{"role": "user", "content": "hi"}]}, + _anthropic_messages_make_wrapper(), + ) + ] + + assert collected == [_anthropic_messages_content_chunk("fallback answer")] + mock_fallback.assert_awaited_once() + assert mock_fallback.await_args.kwargs["e"] is error + + +@pytest.mark.asyncio +async def test_aanthropic_messages_fallback_attempt_raises_original_exception_on_double_failure(): + """Direct-call regression: when the fallback attempt itself fails with a + MidStreamFallbackError wrapping a real provider exception, that real + exception must surface rather than the internal wrapper exception.""" + router = _anthropic_messages_make_router() + error = MidStreamFallbackError(message="overloaded", model="primary", llm_provider="anthropic") + original_exception = litellm.APIError( + status_code=503, message="fallback also overloaded", llm_provider="bedrock", model="fallback" + ) + fallback_failure = MidStreamFallbackError( + message="fallback failed", model="fallback", llm_provider="bedrock", original_exception=original_exception + ) + + with patch.object( + router, + "async_function_with_fallbacks_common_utils", + new=AsyncMock(side_effect=fallback_failure), + ): + with pytest.raises(litellm.APIError) as exc_info: + async for _ in router._aanthropic_messages_fallback_attempt( + error, {"model": "primary"}, _anthropic_messages_make_wrapper() + ): + pass + + assert exc_info.value is original_exception + + +@pytest.mark.asyncio +async def test_aanthropic_messages_fallback_attempt_yields_non_streaming_fallback_response(): + """Bugbot regression: a fallback that resolves to a non-streaming + response (no __aiter__, e.g. an agentic tool-use interception loop) must + be synthesized into a valid SSE byte sequence, not yielded as a raw dict + into a byte stream - the generator is typed AsyncGenerator[bytes, None] + and every item reaching the client must be a real SSE frame.""" + router = _anthropic_messages_make_router() + error = MidStreamFallbackError(message="overloaded", model="primary", llm_provider="anthropic") + non_streaming_response = {"id": "msg_1", "type": "message", "content": [{"type": "text", "text": "hi"}]} + + with patch.object( + router, + "async_function_with_fallbacks_common_utils", + new=AsyncMock(return_value=non_streaming_response), + ): + collected = [ + item + async for item in router._aanthropic_messages_fallback_attempt( + error, {"model": "primary"}, _anthropic_messages_make_wrapper() + ) + ] + + assert all(isinstance(item, bytes) for item in collected) + event_types = [item.split(b"\n")[0].removeprefix(b"event: ") for item in collected] + assert event_types == [ + b"message_start", + b"content_block_start", + b"content_block_delta", + b"content_block_stop", + b"message_delta", + b"message_stop", + ] + assert b'"text": "hi"' in collected[2] + + +@pytest.mark.asyncio +async def test_aanthropic_messages_fallback_attempt_reraises_plain_exception_on_double_failure(): + """Direct-call regression: when the fallback attempt fails with a plain + exception (not a MidStreamFallbackError), that exception itself must + propagate unchanged.""" + router = _anthropic_messages_make_router() + error = MidStreamFallbackError(message="overloaded", model="primary", llm_provider="anthropic") + fallback_failure = ValueError("no healthy deployments") + + with patch.object( + router, + "async_function_with_fallbacks_common_utils", + new=AsyncMock(side_effect=fallback_failure), + ): + with pytest.raises(ValueError, match="no healthy deployments") as exc_info: + async for _ in router._aanthropic_messages_fallback_attempt( + error, {"model": "primary"}, _anthropic_messages_make_wrapper() + ): + pass + + assert exc_info.value is fallback_failure + + +# -------- _aanthropic_messages_with_streaming_fallbacks -------- + + +@pytest.mark.asyncio +async def test_aanthropic_messages_with_streaming_fallbacks_non_streaming_passthrough(): + """A non-streaming response (plain dict) is returned unchanged, never wrapped.""" + router = _anthropic_messages_make_router() + plain_response = {"id": "msg_1", "type": "message"} + + async def fake_original(**_kwargs): + return plain_response + + with patch.object( + router, + "_ageneric_api_call_with_fallbacks", + new=AsyncMock(return_value=plain_response), + ): + out = await router._aanthropic_messages_with_streaming_fallbacks( + original_function=fake_original, + model="primary", + stream=False, + ) + assert out is plain_response + + +@pytest.mark.asyncio +async def test_aanthropic_messages_with_streaming_fallbacks_wraps_streaming_iterator(): + """A streaming response is wrapped via _aanthropic_messages_streaming_iterator.""" + router = _anthropic_messages_make_router() + streaming_iter = _AnthropicMessagesFakeByteStream([_anthropic_messages_content_chunk()]) + wrapped_marker = object() + + async def fake_original(**_kwargs): + return streaming_iter + + with ( + patch.object( + router, + "_ageneric_api_call_with_fallbacks", + new=AsyncMock(return_value=streaming_iter), + ), + patch.object( + router, + "_aanthropic_messages_streaming_iterator", + new=AsyncMock(return_value=wrapped_marker), + ) as mock_wrap, + ): + out = await router._aanthropic_messages_with_streaming_fallbacks( + original_function=fake_original, + model="primary", + stream=True, + ) + assert out is wrapped_marker + mock_wrap.assert_awaited_once() + + +# -------- mid-stream error handling -------- + + +@pytest.mark.asyncio +async def test_anthropic_messages_fallback_on_pre_first_chunk_error_event(): + """Regression for #24004: a retriable SSE `event: error` frame + (overloaded_error/internal_server_error) that arrives before any real + content must trigger the router's fallback chain instead of passing + through to the client silently.""" + router = _anthropic_messages_make_router() + source = _AnthropicMessagesFakeByteStream([_anthropic_messages_overloaded_error_chunk()]) + 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", "messages": [{"role": "user", "content": "hi"}]}, + ) + collected = [chunk async for chunk in wrapped] + + assert collected == [_anthropic_messages_content_chunk("fallback answer")] + mock_fallback.assert_awaited_once() + raised = mock_fallback.await_args.kwargs["e"] + assert isinstance(raised, MidStreamFallbackError) + assert raised.status_code == 503 + assert raised.is_pre_first_chunk is True + assert source.closed is True + + +@pytest.mark.asyncio +async def test_anthropic_messages_mid_stream_error_preserves_real_status_code(): + """Bugbot regression: the MidStreamFallbackError raised for a detected SSE + `event: error` frame must carry the error's REAL parsed status code + (via original_exception), not silently default to 503 for every error + type - a rate_limit_error (429) must surface as 429, not 503.""" + router = _anthropic_messages_make_router() + source = _AnthropicMessagesFakeByteStream([_anthropic_messages_rate_limit_error_chunk()]) + 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", "messages": [{"role": "user", "content": "hi"}]}, + ) + [chunk async for chunk in wrapped] + + raised = mock_fallback.await_args.kwargs["e"] + assert isinstance(raised, MidStreamFallbackError) + assert raised.status_code == 429 + assert raised.original_exception is not None + assert raised.original_exception.status_code == 429 + assert raised.original_exception.llm_provider == "anthropic" + + +def test_merge_fallback_hidden_params_direct_call(): + """Direct-call regression: merge_fallback_hidden_params combines the + fallback's hidden params/headers with whatever was already present, + with the fallback's values winning on key collisions.""" + wrapper = FallbackAwareAnthropicMessagesStream( + _anthropic_messages_empty_generator(), + _AnthropicMessagesFakeByteStream([]), # carries {"additional_headers": {"x-amzn-requestid": "req-1"}} + ) + wrapper.merge_fallback_hidden_params( + {"model_id": "fallback-deployment"}, + {"x-amzn-requestid": "req-2", "x-fallback-only": "yes"}, + ) + assert wrapper._hidden_params["model_id"] == "fallback-deployment" + assert wrapper._hidden_params["additional_headers"] == { + "x-amzn-requestid": "req-2", + "x-fallback-only": "yes", + } + + +def test_anthropic_stream_should_drop_pre_content_ping_direct_call(): + ping = _anthropic_messages_ping_chunk() + content = _anthropic_messages_content_chunk("hi") + assert _anthropic_stream_should_drop_pre_content_ping(ping, has_generated_content=False) is True + assert _anthropic_stream_should_drop_pre_content_ping(ping, has_generated_content=True) is False + assert _anthropic_stream_should_drop_pre_content_ping(content, has_generated_content=False) is False + + +def test_is_retriable_anthropic_status_direct_call(): + assert _is_retriable_anthropic_status(429) is True + assert _is_retriable_anthropic_status(503) is True + assert _is_retriable_anthropic_status(500) is True + assert _is_retriable_anthropic_status(400) is False + assert _is_retriable_anthropic_status(404) is False + + +def test_anthropic_stream_should_decline_fallback_direct_call(): + pre_first_chunk_error = MidStreamFallbackError( + message="overloaded", model="primary", llm_provider="anthropic", is_pre_first_chunk=True + ) + post_first_chunk_error = MidStreamFallbackError( + message="overloaded", model="primary", llm_provider="anthropic", is_pre_first_chunk=False + ) + assert _anthropic_stream_should_decline_fallback(False, pre_first_chunk_error) is False + assert _anthropic_stream_should_decline_fallback(True, pre_first_chunk_error) is True + assert _anthropic_stream_should_decline_fallback(False, post_first_chunk_error) is True + + +def test_anthropic_stream_commits_now_direct_call(): + content = _anthropic_messages_content_chunk("hi") + lifecycle_chunk = _anthropic_messages_message_start_chunk() + assert _anthropic_stream_commits_now(content, has_generated_content=False, buffered_chunk_count=0) is True + assert _anthropic_stream_commits_now(content, has_generated_content=True, buffered_chunk_count=0) is False + assert ( + _anthropic_stream_commits_now( + lifecycle_chunk, + has_generated_content=False, + buffered_chunk_count=MAX_BUFFERED_PRE_CONTENT_ANTHROPIC_CHUNKS, + ) + is True + ) + assert ( + _anthropic_stream_commits_now( + lifecycle_chunk, + has_generated_content=False, + buffered_chunk_count=MAX_BUFFERED_PRE_CONTENT_ANTHROPIC_CHUNKS - 1, + ) + is False + ) + + +@pytest.mark.asyncio +async def test_anthropic_messages_fallback_merges_fallback_hidden_params(): + """Bugbot regression: after a successful mid-stream fallback, the + wrapper's _hidden_params must reflect the FALLBACK deployment's own + provider headers (e.g. a different Bedrock request-id), not stay + frozen on the primary's - raw bytes can't carry per-item _hidden_params + the way a ModelResponseStream/ResponsesAPI event can, so the wrapper + itself is the only place left to expose them.""" + router = _anthropic_messages_make_router() + source = _AnthropicMessagesFakeByteStream( + [_anthropic_messages_overloaded_error_chunk()] + ) # carries x-amzn-requestid: req-1 + fallback_stream = _AnthropicMessagesFallbackByteStream( + [_anthropic_messages_content_chunk("fallback answer")], + hidden_params={"additional_headers": {"x-amzn-requestid": "req-2", "x-fallback-only": "yes"}}, + ) + + with patch.object( + router, + "async_function_with_fallbacks_common_utils", + new=AsyncMock(return_value=fallback_stream), + ): + wrapped = await router._aanthropic_messages_streaming_iterator( + response=source, + initial_kwargs={"model": "primary"}, + ) + _ = [chunk async for chunk in wrapped] + + headers = wrapped._hidden_params["additional_headers"] + assert headers["x-amzn-requestid"] == "req-2" + assert headers["x-fallback-only"] == "yes" + + +@pytest.mark.asyncio +async def test_aanthropic_messages_with_streaming_fallbacks_deep_copies_nested_metadata(): + """Bugbot regression: a shallow .copy() of kwargs still shares the + nested litellm_metadata/metadata dict objects with the primary attempt. + _update_kwargs_with_deployment mutates that dict in place with + deployment-specific fields, which must not leak into the fallback + request's metadata.""" + router = _anthropic_messages_make_router() + primary_metadata = {"model_group": "primary"} + streaming_iter_kwargs = {} + + async def fake_original(**_kwargs): + # Simulate _update_kwargs_with_deployment mutating the primary's + # litellm_metadata in place, as the real helper does. + primary_metadata["deployment"] = "primary-deployment-object" + return _AnthropicMessagesFakeByteStream([_anthropic_messages_content_chunk("hi")]) + + with patch.object( + router, + "_aanthropic_messages_streaming_iterator", + new=AsyncMock(side_effect=lambda **kwargs: streaming_iter_kwargs.update(kwargs) or "wrapped"), + ): + with patch.object( + router, + "_ageneric_api_call_with_fallbacks", + new=AsyncMock(side_effect=fake_original), + ): + await router._aanthropic_messages_with_streaming_fallbacks( + original_function=fake_original, + model="primary", + stream=True, + litellm_metadata=primary_metadata, + ) + + fallback_kwargs = streaming_iter_kwargs["initial_kwargs"] + assert fallback_kwargs["litellm_metadata"] is not primary_metadata + assert "deployment" not in fallback_kwargs["litellm_metadata"] + + +@pytest.mark.asyncio +async def test_aanthropic_messages_with_streaming_fallbacks_deep_copies_metadata_field(): + """Same regression as above for the (separate) `metadata` kwarg some + call sites use instead of `litellm_metadata`.""" + router = _anthropic_messages_make_router() + primary_metadata = {"tag": "primary"} + streaming_iter_kwargs = {} + + async def fake_original(**_kwargs): + primary_metadata["deployment"] = "primary-deployment-object" + return _AnthropicMessagesFakeByteStream([_anthropic_messages_content_chunk("hi")]) + + with patch.object( + router, + "_aanthropic_messages_streaming_iterator", + new=AsyncMock(side_effect=lambda **kwargs: streaming_iter_kwargs.update(kwargs) or "wrapped"), + ): + with patch.object( + router, + "_ageneric_api_call_with_fallbacks", + new=AsyncMock(side_effect=fake_original), + ): + await router._aanthropic_messages_with_streaming_fallbacks( + original_function=fake_original, + model="primary", + stream=True, + metadata=primary_metadata, + ) + + fallback_kwargs = streaming_iter_kwargs["initial_kwargs"] + assert fallback_kwargs["metadata"] is not primary_metadata + assert "deployment" not in fallback_kwargs["metadata"] + + +@pytest.mark.asyncio +async def test_anthropic_messages_fallback_triggers_after_lifecycle_only_frame(): + """Regression: Anthropic routinely sends a message_start lifecycle frame + before an overload error even fires. A lifecycle-only frame (no real + content) must not disqualify the fallback attempt, and must not reach + the client either - forwarding it and then appending the fallback's own + message_start would produce two overlapping message lifecycles on one + SSE stream. The primary's buffered lifecycle frame is discarded and the + client sees only the fallback's own, single, clean lifecycle.""" + router = _anthropic_messages_make_router() + source = _AnthropicMessagesFakeByteStream( + [_anthropic_messages_message_start_chunk(), _anthropic_messages_overloaded_error_chunk()] + ) + fallback_message_start = b'event: message_start\ndata: {"type": "message_start", "message": {"id": "msg_2"}}\n\n' + fallback_stream = _AnthropicMessagesFallbackByteStream( + [fallback_message_start, _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 == [fallback_message_start, _anthropic_messages_content_chunk("fallback answer")] + assert collected.count(_anthropic_messages_message_start_chunk()) == 0, ( + "the primary's message_start must never reach the client" + ) + assert sum(1 for c in collected if c.startswith(b"event: message_start")) == 1, ( + "exactly one message_start must reach the client" + ) + mock_fallback.assert_awaited_once() + raised = mock_fallback.await_args.kwargs["e"] + assert raised.is_pre_first_chunk is True + + +@pytest.mark.asyncio +async def test_anthropic_messages_raised_error_after_real_content_does_not_restart_stream(): + """Regression: a MidStreamFallbackError raised directly by the source + iterator (the completion-bridge path's CustomStreamWrapper, e.g. a + transport drop) must not trigger a fallback once real content already + reached the client - that would append a second, overlapping message + lifecycle onto the same SSE stream. The original exception must + propagate to the caller instead.""" + router = _anthropic_messages_make_router() + content = _anthropic_messages_content_chunk("partial answer") + original_exception = litellm.APIError( + status_code=503, + message="stream reset", + llm_provider="vertex_ai", + model="primary", + ) + raised_error = MidStreamFallbackError( + message="stream reset", + model="primary", + llm_provider="vertex_ai", + original_exception=original_exception, + is_pre_first_chunk=False, + ) + 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(litellm.APIError) as exc_info: + await _consume() + + assert collected == [content] + assert exc_info.value is original_exception + mock_fallback.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_anthropic_messages_fallback_also_catches_raised_midstream_error(): + """Regression for the completion-bridge path (deployments with no native + /v1/messages endpoint): its CustomStreamWrapper raises + MidStreamFallbackError directly (e.g. on a Vertex AI transport drop) + instead of yielding an SSE error chunk - the wrapper must catch that too.""" + router = _anthropic_messages_make_router() + raised_error = MidStreamFallbackError( + message="stream reset", + model="primary", + llm_provider="vertex_ai", + is_pre_first_chunk=True, + ) + source = _AnthropicMessagesRaisingByteStream([], 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() + assert mock_fallback.await_args.kwargs["e"] is raised_error + + +@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 + error a fallback attempt cannot fix, so it must be forwarded to the + client as-is rather than burning a fallback attempt.""" + router = _anthropic_messages_make_router() + error_chunk = _anthropic_messages_invalid_request_error_chunk() + source = _AnthropicMessagesFakeByteStream([error_chunk]) + + 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 = [chunk async for chunk in wrapped] + + assert collected == [error_chunk] + mock_fallback.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_anthropic_messages_post_first_chunk_error_skips_fallback(): + """Once content has already reached the caller, retrying would start a + second, overlapping Anthropic message lifecycle on the same SSE stream - + the error must be forwarded instead of triggering an invisible retry.""" + router = _anthropic_messages_make_router() + content = _anthropic_messages_content_chunk("partial answer") + error_chunk = _anthropic_messages_overloaded_error_chunk() + source = _AnthropicMessagesFakeByteStream([content, error_chunk]) + + 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 = [chunk async for chunk in wrapped] + + assert collected == [content, error_chunk] + mock_fallback.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_anthropic_messages_non_retriable_error_flushes_buffered_lifecycle_frames(): + """A non-retriable error arriving while lifecycle frames are still + buffered (no content seen yet) must flush those buffered frames before + forwarding the error, so the client still sees the whole primary + attempt rather than losing the buffered message_start silently.""" + router = _anthropic_messages_make_router() + error_chunk = _anthropic_messages_invalid_request_error_chunk() + source = _AnthropicMessagesFakeByteStream([_anthropic_messages_message_start_chunk(), error_chunk]) + + 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 = [chunk async for chunk in wrapped] + + assert collected == [_anthropic_messages_message_start_chunk(), error_chunk] + mock_fallback.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_anthropic_messages_raised_error_declined_flushes_buffered_lifecycle_frames(): + """When a raised MidStreamFallbackError is declined (source says content + was not pre-first-chunk) while lifecycle frames are still buffered, they + must be flushed to the client before the exception propagates.""" + router = _anthropic_messages_make_router() + raised_error = MidStreamFallbackError( + message="stream reset", + model="primary", + llm_provider="vertex_ai", + is_pre_first_chunk=False, + ) + 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(MidStreamFallbackError) as exc_info: + await _consume() + + assert collected == [_anthropic_messages_message_start_chunk()] + assert exc_info.value is raised_error + mock_fallback.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_anthropic_messages_raised_error_without_original_exception_reraises_itself(): + """When a declined MidStreamFallbackError carries no original_exception, + the bare exception itself must propagate rather than being swallowed.""" + router = _anthropic_messages_make_router() + content = _anthropic_messages_content_chunk("partial answer") + raised_error = MidStreamFallbackError( + message="stream reset", + model="primary", + llm_provider="vertex_ai", + is_pre_first_chunk=False, + ) + source = _AnthropicMessagesRaisingByteStream([content], raised_error) + + 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(MidStreamFallbackError) as exc_info: + await _consume() + + assert collected == [content] + assert exc_info.value is raised_error + + +@pytest.mark.asyncio +async def test_anthropic_messages_fallback_also_failing_raises_original_exception(): + """If the fallback attempt itself fails with a MidStreamFallbackError + wrapping a real provider exception, the client must see that real + exception, not the internal MidStreamFallbackError.""" + router = _anthropic_messages_make_router() + source = _AnthropicMessagesFakeByteStream([_anthropic_messages_overloaded_error_chunk()]) + original_exception = litellm.APIError( + status_code=503, + message="fallback also overloaded", + llm_provider="bedrock", + model="fallback", + ) + fallback_failure = MidStreamFallbackError( + message="fallback failed", + model="fallback", + llm_provider="bedrock", + original_exception=original_exception, + ) + + with patch.object( + router, + "async_function_with_fallbacks_common_utils", + new=AsyncMock(side_effect=fallback_failure), + ): + wrapped = await router._aanthropic_messages_streaming_iterator( + response=source, + initial_kwargs={"model": "primary"}, + ) + with pytest.raises(litellm.APIError) as exc_info: + async for _ in wrapped: + pass + + assert exc_info.value is original_exception + + +# -------- _dispatch_generic_call_type -------- + + +@pytest.mark.asyncio +async def test_dispatch_generic_call_type_routes_anthropic_messages_through_streaming_fallbacks(): + router = _anthropic_messages_make_router() + + async def fake_original(**_kwargs): + return {"id": "msg_1"} + + with patch.object( + router, + "_aanthropic_messages_with_streaming_fallbacks", + new=AsyncMock(return_value="anthropic-result"), + ) as mock_anthropic: + out = await router._dispatch_generic_call_type( + call_type="anthropic_messages", + original_function=fake_original, + model="primary", + ) + assert out == "anthropic-result" + mock_anthropic.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_dispatch_generic_call_type_other_call_types_use_generic_fallback(): + router = _anthropic_messages_make_router() + + async def fake_original(**_kwargs): + return {"id": "file_1"} + + with patch.object( + router, + "_ageneric_api_call_with_fallbacks", + new=AsyncMock(return_value="generic-result"), + ) as mock_generic: + out = await router._dispatch_generic_call_type( + call_type="afile_delete", + original_function=fake_original, + model="primary", + ) + assert out == "generic-result" + mock_generic.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_factory_function_anthropic_messages_uses_streaming_fallback_dispatch(): + """anthropic_messages must be wired through the mid-stream-fallback-aware + path rather than the bare generic dispatch every other call type without + special handling uses.""" + router = _anthropic_messages_make_router() + wrapped = router.factory_function(litellm.anthropic_messages, call_type="anthropic_messages") + assert callable(wrapped) + + with patch.object( + router, + "_aanthropic_messages_with_streaming_fallbacks", + new=AsyncMock(return_value="ok"), + ) as mock_anthropic: + result = await wrapped(model="primary") + assert result == "ok" + mock_anthropic.assert_awaited_once()