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 0bd46382fef..5550590d0c0 100644 --- a/litellm/llms/anthropic/experimental_pass_through/messages/streaming_iterator.py +++ b/litellm/llms/anthropic/experimental_pass_through/messages/streaming_iterator.py @@ -16,6 +16,7 @@ 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.llms.anthropic.experimental_pass_through.messages.utils import INCOMPLETE_STREAM_ERROR_MESSAGE from litellm.proxy.pass_through_endpoints.success_handler import ( PassThroughEndpointLogging, ) @@ -28,11 +29,6 @@ GLOBAL_PASS_THROUGH_SUCCESS_HANDLER_OBJ: Final = PassThroughEndpointLogging() _UPSTREAM_PUMP_TASKS: Final[set[asyncio.Task[None]]] = set() # mutable-ok: stdlib strong-ref set for pump tasks _DETACHED_STREAM_DRAINS: Final[set[asyncio.Task[None]]] = set() # mutable-ok: bounded strong-ref set, detached drains -INCOMPLETE_STREAM_ERROR_MESSAGE: Final = ( - "Provider stream ended before emitting a message_stop event; " - "the response is incomplete and any partial content (e.g. tool_use input JSON) may be truncated." -) - def _is_message_stop_chunk(chunk: object) -> bool: if isinstance(chunk, dict): diff --git a/litellm/llms/anthropic/experimental_pass_through/messages/utils.py b/litellm/llms/anthropic/experimental_pass_through/messages/utils.py index 89105c00428..fe8ac2cd7a2 100644 --- a/litellm/llms/anthropic/experimental_pass_through/messages/utils.py +++ b/litellm/llms/anthropic/experimental_pass_through/messages/utils.py @@ -15,6 +15,12 @@ if TYPE_CHECKING: from litellm.exceptions import ContentPolicyViolationError +INCOMPLETE_STREAM_ERROR_MESSAGE: Final = ( + "Provider stream ended before emitting a message_stop event; " + "the response is incomplete and any partial content (e.g. tool_use input JSON) may be truncated." +) + + def get_safeguard_refusal_stop_details(response: object) -> Mapping[str, Any] | None: """ Return the ``stop_details`` of an Anthropic Messages response refused by a diff --git a/litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py b/litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py index f753e87fee3..59ccde872fc 100644 --- a/litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py +++ b/litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py @@ -2,20 +2,25 @@ ## Translates OpenAI call to Anthropic `/v1/messages` format import asyncio import json -import traceback from collections import deque from collections.abc import AsyncIterator, Iterator, Mapping from typing import TYPE_CHECKING, Any, Final +from pydantic import BaseModel, ConfigDict, field_validator + from litellm import verbose_logger +from litellm._logging import redact_internal_details_from_client_message from litellm._uuid import uuid +from litellm.exceptions import MidStreamFallbackError from litellm.litellm_core_utils.prompt_templates.common_utils import ( encrypted_reasoning_signature, ) from litellm.llms.anthropic.experimental_pass_through.messages.utils import ( + INCOMPLETE_STREAM_ERROR_MESSAGE, refusal_stop_details, responses_output_refusal_text, ) +from litellm.responses.streaming_iterator import stream_error_status_and_message from litellm.types.llms.anthropic_messages.anthropic_response import AnthropicUsage from .transformation import ( @@ -27,6 +32,72 @@ if TYPE_CHECKING: from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObject +class _UpstreamFailure(BaseModel): + model_config = ConfigDict(frozen=True) + + status_code: int | None = None + message: str | None = None + + @field_validator("status_code", mode="before") + @classmethod + def http_error_status_or_none(cls, value: object) -> int | None: + candidate: Final = ( + value + if isinstance(value, int) and not isinstance(value, bool) + else int(value) + if isinstance(value, str) and value.isdecimal() + else None + ) + return candidate if candidate is not None and 400 <= candidate <= 599 else None + + @field_validator("message", mode="before") + @classmethod + def str_or_none(cls, value: object) -> str | None: + return value if isinstance(value, str) else None + + +class _FailedResponse(BaseModel): + model_config = ConfigDict(frozen=True, from_attributes=True) + + error: object | None = None + + +class _FailedResponseEvent(BaseModel): + model_config = ConfigDict(frozen=True, from_attributes=True) + + response: _FailedResponse | None = None + + +def _original_failure(exception: Exception) -> Exception: + failure = exception # rebind-ok: walks the MidStreamFallbackError chain down to the provider failure + while isinstance(failure, MidStreamFallbackError) and failure.original_exception is not None: + failure = failure.original_exception + return failure + + +def _failure_status_and_message(exception: Exception) -> tuple[int, str]: + original: Final = _original_failure(exception) + failure: Final = _UpstreamFailure.model_validate( + {"status_code": getattr(original, "status_code", None), "message": getattr(original, "message", None)} + ) + status_code: Final = failure.status_code if failure.status_code is not None else 500 + message: Final = failure.message or str(original) or INCOMPLETE_STREAM_ERROR_MESSAGE + return status_code, message + + +def _anthropic_error_chunk(status_code: int, message: str) -> dict[str, object]: + from litellm.anthropic_interface.exceptions.exception_mapping_utils import ( + AnthropicExceptionMapping, + ) + + return dict( + AnthropicExceptionMapping.transform_to_anthropic_error( + status_code=status_code, + raw_message=redact_internal_details_from_client_message(message), + ) + ) + + class AnthropicResponsesStreamWrapper: """ Wraps a Responses API streaming iterator and re-emits events in Anthropic SSE format. @@ -40,6 +111,7 @@ class AnthropicResponsesStreamWrapper: response.function_call_arguments.delta -> content_block_delta (input_json_delta) response.output_item.done -> content_block_delta (signature_delta) + content_block_stop response.completed -> message_delta + message_stop + response.failed -> error (the stream ends without message_stop) """ def __init__( @@ -60,6 +132,7 @@ class AnthropicResponsesStreamWrapper: self._pending_tool_ids: dict[str, str] = {} # item_id -> call_id / name accumulator self._sent_message_start = False self._sent_message_stop = False + self._stream_failed = False self._chunk_queue: deque[dict[str, object]] = deque() self._refusal_text: str = "" self._sync_responses_iterator: Iterator[object] | None = None @@ -293,10 +366,23 @@ class AnthropicResponsesStreamWrapper: ) return + if event_type == "response.failed": + failed: Final = _FailedResponseEvent.model_validate(event) + status_code, message = stream_error_status_and_message( + failed.response.error if failed.response is not None else None + ) + verbose_logger.error( + "AnthropicResponsesStreamWrapper: upstream Responses stream for %s failed (%s): %s", + self.model, + status_code, + message, + ) + self._fail_stream(status_code, message) + return + # ---- response completed -> message_delta + message_stop ---- if event_type in ( "response.completed", - "response.failed", "response.incomplete", ): response_obj: Final = getattr(event, "response", None) or ( @@ -350,21 +436,24 @@ class AnthropicResponsesStreamWrapper: self._sent_message_stop = True return + def _fail_stream(self, status_code: int, message: str) -> None: + self._stream_failed = True + self._chunk_queue.append(_anthropic_error_chunk(status_code, message)) + def __aiter__(self) -> "AnthropicResponsesStreamWrapper": return self async def __anext__(self) -> dict[str, object]: - # Return any queued chunks first if self._chunk_queue: return self._chunk_queue.popleft() + if self._stream_failed: + raise StopAsyncIteration - # Emit message_start if not yet done (fallback if response.created wasn't fired) if not self._sent_message_start: self._sent_message_start = True self._chunk_queue.append(self._make_message_start()) return self._chunk_queue.popleft() - # Consume the upstream stream try: if hasattr(self.responses_stream, "__aiter__"): async for event in self.responses_stream: @@ -382,10 +471,19 @@ class AnthropicResponsesStreamWrapper: return self._chunk_queue.popleft() except StopAsyncIteration: pass - except Exception as e: - verbose_logger.error("AnthropicResponsesStreamWrapper error: %s\n%s", e, traceback.format_exc()) + except Exception as e: # noqa: BLE001 # every upstream failure becomes a client error event + verbose_logger.exception( + "AnthropicResponsesStreamWrapper: upstream Responses stream for %s failed", self.model + ) + self._fail_stream(*_failure_status_and_message(e)) + + if not self._chunk_queue and not self._sent_message_stop and not self._stream_failed: + verbose_logger.error( + "AnthropicResponsesStreamWrapper: upstream Responses stream for %s ended without a terminal event", + self.model, + ) + self._fail_stream(500, INCOMPLETE_STREAM_ERROR_MESSAGE) - # Drain any remaining queued chunks if self._chunk_queue: return self._chunk_queue.popleft() diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index 70f2a7db6da..fdc702af005 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -230,6 +230,11 @@ def _status_code_for_error_fields(error_type: str | None, error_code: str | None return next((status for status in map(_status_code_for_error_field, fields) if status is not None), 500) +def stream_error_status_and_message(error_obj: object) -> tuple[int, str]: + message, error_type, error_code = _error_event_fields(error_obj) + return _status_code_for_error_fields(error_type, error_code), message + + def _map_stream_error_to_exception(error_obj: object, model: str, custom_llm_provider: str) -> Exception: from litellm.llms.base_llm.chat.transformation import BaseLLMException diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py index bfe2d6b7cea..392ecc2bcdd 100644 --- a/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py +++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py @@ -4,18 +4,25 @@ Tests for AnthropicResponsesStreamWrapper """ import asyncio +import json import os import sys from types import SimpleNamespace +import pytest + sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../../../.."))) +import litellm +from litellm.exceptions import MidStreamFallbackError from litellm.litellm_core_utils.prompt_templates.common_utils import ( encrypted_reasoning_signature, ) +from litellm.llms.anthropic.experimental_pass_through.messages.utils import INCOMPLETE_STREAM_ERROR_MESSAGE from litellm.llms.anthropic.experimental_pass_through.responses_adapters.streaming_iterator import ( AnthropicResponsesStreamWrapper, ) +from litellm.types.llms.openai import ResponseFailedEvent, ResponsesAPIResponse def _process_all(events: list) -> list: @@ -132,6 +139,7 @@ class TestReasoningItemWithoutSummaryText: {"type": "response.output_item.added", "item": {"type": "message", "id": "msg_1"}}, {"type": "response.output_text.delta", "item_id": "msg_1", "delta": "Hello"}, {"type": "response.output_item.done", "item": {"type": "message", "id": "msg_1"}}, + {"type": "response.completed"}, ] def test_reasoning_without_summary_emits_no_thinking_block(self): @@ -144,6 +152,8 @@ class TestReasoningItemWithoutSummaryText: ("content_block_start", 0), ("content_block_delta", 0), ("content_block_stop", 0), + ("message_delta", None), + ("message_stop", None), ] assert chunks[1]["content_block"] == {"type": "text", "text": ""} @@ -166,6 +176,8 @@ class TestReasoningItemWithoutSummaryText: ("content_block_start", 1), ("content_block_delta", 1), ("content_block_stop", 1), + ("message_delta", None), + ("message_stop", None), ] assert chunks[1]["content_block"] == {"type": "thinking", "thinking": "", "signature": ""} assert "".join(c["delta"]["thinking"] for c in chunks[2:4]) == "Weighing options" @@ -215,6 +227,8 @@ class TestEncryptedReasoningIsStreamedForReplay: ("content_block_start", 1), ("content_block_delta", 1), ("content_block_stop", 1), + ("message_delta", None), + ("message_stop", None), ] assert chunks[1]["content_block"] == { "type": "redacted_thinking", @@ -234,9 +248,7 @@ class TestEncryptedReasoningIsStreamedForReplay: ] chunks = _process_all(events) - thinking = "".join( - c["delta"]["thinking"] for c in chunks if c.get("delta", {}).get("type") == "thinking_delta" - ) + thinking = "".join(c["delta"]["thinking"] for c in chunks if c.get("delta", {}).get("type") == "thinking_delta") assert thinking == "First.\n\nSecond." assert [c["type"] for c in chunks].count("content_block_start") == 1 @@ -283,6 +295,7 @@ class TestToolUseBlockClosedExactlyOnce: "type": "response.output_item.done", "item": {"type": "message", "id": "chatcmpl-123", "status": "completed"}, }, + {"type": "response.completed"}, ] def test_one_content_block_stop_per_content_block_start(self): @@ -302,6 +315,8 @@ class TestToolUseBlockClosedExactlyOnce: ("content_block_delta", 0), ("content_block_delta", 0), ("content_block_stop", 0), + ("message_delta", None), + ("message_stop", None), ] assert chunks[1]["content_block"] == { "type": "tool_use", @@ -452,3 +467,158 @@ class TestRefusalStreamEvents: message_delta = next(c for c in chunks if c["type"] == "message_delta") assert message_delta["delta"]["stop_reason"] == "max_tokens" assert "stop_details" not in message_delta["delta"] + + +def _collect(stream) -> list: + async def _run() -> list: + wrapper = AnthropicResponsesStreamWrapper(responses_stream=stream, model="m") + return [chunk async for chunk in wrapper] + + return asyncio.run(_run()) + + +class TestUpstreamFailureEndsStreamWithErrorEvent: + """A provider failure must reach the Anthropic client as an ``error`` event that + ends the stream, never as a fabricated ``end_turn`` or a silent close.""" + + def test_response_failed_event_emits_error_event_and_stops_pulling_upstream(self): + failed = SimpleNamespace( + status="failed", + output=[], + usage=None, + error={"code": "rate_limit_exceeded", "message": "Rate limit reached for gpt-5.5, try again in 20s."}, + ) + + async def _gen(): + yield {"type": "response.created"} + yield {"type": "response.failed", "response": failed} + raise AssertionError("upstream was pulled again after the failure") + + async def _run() -> list: + wrapper = AnthropicResponsesStreamWrapper(responses_stream=_gen(), model="m") + return [frame async for frame in wrapper.async_anthropic_sse_wrapper()] + + frames = asyncio.run(_run()) + assert [frame.split(b"\n", 1)[0] for frame in frames] == [b"event: message_start", b"event: error"] + error_payload = json.loads(frames[1].split(b"data: ", 1)[1]) + assert error_payload["type"] == "error" + assert error_payload["error"] == { + "type": "rate_limit_error", + "message": "Rate limit reached for gpt-5.5, try again in 20s.", + } + + def test_raised_mid_stream_fallback_error_is_unwrapped_to_the_provider_failure(self): + rate_limit = litellm.RateLimitError(message="You have no credits remaining.", llm_provider="openai", model="m") + wrapped = MidStreamFallbackError( + message=str(rate_limit), + model="m", + llm_provider="openai", + original_exception=rate_limit, + is_pre_first_chunk=True, + ) + + async def _gen(): + yield {"type": "response.created"} + raise wrapped + + chunks = _collect(_gen()) + assert [chunk["type"] for chunk in chunks] == ["message_start", "error"] + assert chunks[1]["error"] == {"type": "rate_limit_error", "message": rate_limit.message} + + def test_sync_upstream_transport_error_after_content_becomes_api_error_event(self): + def _events(): + yield {"type": "response.created"} + yield {"type": "response.output_item.added", "item": {"type": "message", "id": "msg_1"}} + yield {"type": "response.output_text.delta", "item_id": "msg_1", "delta": "Hi"} + raise ConnectionResetError("Response payload is not completed") + + chunks = _collect(_events()) + assert [chunk["type"] for chunk in chunks] == [ + "message_start", + "content_block_start", + "content_block_delta", + "error", + ] + assert chunks[-1]["error"] == {"type": "api_error", "message": "Response payload is not completed"} + + def test_error_event_message_is_redacted_before_it_reaches_the_client(self): + async def _gen(): + yield {"type": "response.created"} + raise RuntimeError("upstream failed with key sk-proj-abcdefghijklmnopqrstuvwxyz0123456789ABCDEFGHIJ") + + chunks = _collect(_gen()) + assert chunks[-1]["type"] == "error" + assert "sk-proj-" not in chunks[-1]["error"]["message"] + assert chunks[-1]["error"]["message"].startswith("upstream failed with key") + + @pytest.mark.parametrize( + ("raised", "expected_error"), + [ + ( + MidStreamFallbackError(message="boom", model="m", llm_provider="openai"), + {"type": "api_error", "message": "litellm.MidStreamFallbackError: boom"}, + ), + ( + type("StringStatusError", (Exception,), {"status_code": "429"})("throttled"), + {"type": "rate_limit_error", "message": "throttled"}, + ), + ( + type("NonErrorStatusError", (Exception,), {"status_code": 200})("odd status"), + {"type": "api_error", "message": "odd status"}, + ), + ], + ids=["mid-stream-fallback-without-original", "digit-string-status", "status-outside-4xx-5xx"], + ) + def test_raised_failure_status_is_normalized_into_the_error_type(self, raised, expected_error): + async def _gen(): + yield {"type": "response.created"} + raise raised + + chunks = _collect(_gen()) + assert [chunk["type"] for chunk in chunks] == ["message_start", "error"] + assert chunks[1]["error"] == expected_error + + def test_pydantic_response_failed_event_is_mapped_like_a_dict_event(self): + failed = ResponsesAPIResponse( + id="resp_1", + created_at=1, + error={"code": "server_error", "message": "The server had an error while processing your request."}, + status="failed", + output=[], + model="m", + object="response", + parallel_tool_calls=False, + tool_choice="auto", + tools=[], + ) + + async def _gen(): + yield {"type": "response.created"} + yield ResponseFailedEvent(type="response.failed", response=failed) + + chunks = _collect(_gen()) + assert [chunk["type"] for chunk in chunks] == ["message_start", "error"] + assert chunks[1]["error"] == { + "type": "api_error", + "message": "The server had an error while processing your request.", + } + + def test_upstream_ending_without_a_terminal_event_is_an_error_not_a_silent_close(self): + async def _gen(): + yield {"type": "response.created"} + yield {"type": "response.output_item.added", "item": {"type": "message", "id": "msg_1"}} + yield {"type": "response.output_text.delta", "item_id": "msg_1", "delta": "Hi"} + + chunks = _collect(_gen()) + assert [chunk["type"] for chunk in chunks] == [ + "message_start", + "content_block_start", + "content_block_delta", + "error", + ] + assert chunks[-1]["error"] == {"type": "api_error", "message": INCOMPLETE_STREAM_ERROR_MESSAGE} + + def test_sync_upstream_ending_before_any_event_is_an_error_not_a_silent_close(self): + chunks = _collect(iter(())) + assert [chunk["type"] for chunk in chunks] == ["message_start", "error"] + assert chunks[1]["error"] == {"type": "api_error", "message": INCOMPLETE_STREAM_ERROR_MESSAGE}