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..a27a53feb5a 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 @@ -5,10 +5,13 @@ import json import traceback from collections import deque from collections.abc import AsyncIterator, Iterator, Mapping -from typing import TYPE_CHECKING, Any, Final +from typing import TYPE_CHECKING, Any, Final, cast + +from typing_extensions import TypeIs from litellm import verbose_logger from litellm._uuid import uuid +from litellm.exceptions import MidStreamFallbackError from litellm.litellm_core_utils.prompt_templates.common_utils import ( encrypted_reasoning_signature, ) @@ -26,6 +29,22 @@ from .transformation import ( if TYPE_CHECKING: from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObject +# Map upstream HTTP status codes to the Anthropic error types from +# https://docs.anthropic.com/en/api/errors — anything unmapped is an api_error. +_STATUS_TO_ANTHROPIC_ERROR_TYPE: Final[dict[int, str]] = { + 400: "invalid_request_error", + 401: "authentication_error", + 403: "permission_error", + 404: "not_found_error", + 413: "request_too_large", + 429: "rate_limit_error", + 529: "overloaded_error", +} + + +def _is_json_object(value: object) -> TypeIs[dict[str, object]]: # guard-ok: trivial isinstance; JSON keys are str + return isinstance(value, dict) + class AnthropicResponsesStreamWrapper: """ @@ -84,6 +103,18 @@ class AnthropicResponsesStreamWrapper: }, } + @staticmethod + def _make_error_event(error_type: str | None, error_message: str | None) -> dict[str, object]: + """The Anthropic streaming spec's terminal error frame.""" + event: dict[str, object] = { + "type": "error", + "error": { + "type": error_type or "api_error", + "message": error_message or "Upstream response failed", + }, + } + return event + def _next_block_index(self) -> int: self._current_block_index += 1 return self._current_block_index @@ -293,10 +324,42 @@ class AnthropicResponsesStreamWrapper: ) return + # ---- response failed -> terminal error event ---- + if event_type == "response.failed": + event_obj = cast("object", event) + failed_response: object | None + if _is_json_object(event_obj): + failed_response = event_obj.get("response") + else: + failed_response = getattr(event_obj, "response", None) + + failed_error: object | None = None + if failed_response is not None: + if _is_json_object(failed_response): + failed_error = failed_response.get("error") + else: + failed_error = getattr(failed_response, "error", None) + + error_type: str | None = None + error_message: str | None = None + if failed_error is not None: + if _is_json_object(failed_error): + raw_type: object | None = failed_error.get("type") or failed_error.get("code") + raw_message: object | None = failed_error.get("message") + else: + raw_type = getattr(failed_error, "type", None) or getattr(failed_error, "code", None) + raw_message = getattr(failed_error, "message", None) + if isinstance(raw_type, str): + error_type = raw_type + if isinstance(raw_message, str): + error_message = raw_message + + self._chunk_queue.append(self._make_error_event(error_type, error_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 ( @@ -382,6 +445,18 @@ class AnthropicResponsesStreamWrapper: return self._chunk_queue.popleft() except StopAsyncIteration: pass + except MidStreamFallbackError as e: + # Do not swallow mid-stream upstream failures: re-emit them as the + # Anthropic streaming spec's terminal error event, otherwise the + # client sees a silent, unterminated stream. + verbose_logger.error("AnthropicResponsesStreamWrapper upstream failed: %s", e) + original_message: object | None = getattr(e.original_exception, "message", None) + self._chunk_queue.append( + self._make_error_event( + _STATUS_TO_ANTHROPIC_ERROR_TYPE.get(e.status_code, "api_error"), + str(original_message) if original_message else e.message, + ) + ) except Exception as e: verbose_logger.error("AnthropicResponsesStreamWrapper error: %s\n%s", e, traceback.format_exc()) 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..d846661605c 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 @@ -452,3 +452,124 @@ 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"] + + +class TestResponseFailedSurfacesUpstreamFailure: + """Regression for https://github.com/BerriAI/litellm/issues/39703. + + A provider-side failure that arrives inside an already-200 stream + (``response.failed``, e.g. an exhausted quota) must reach the client as a + failure. It used to share the ``response.completed`` branch and was + relabelled as a successful empty turn (``stop_reason: "end_turn"``, + usage 0/0), so agent clients (e.g. Claude Code) just stopped mid-session + with no error anywhere. + """ + + @staticmethod + def _failed_response() -> SimpleNamespace: + return SimpleNamespace( + status="failed", + usage=None, + output=[], + error=SimpleNamespace( + code="insufficient_quota", + type="insufficient_quota", + message="You have no credits remaining. Add credits to continue using the API.", + ), + ) + + def test_response_failed_emits_terminal_error_event_not_end_turn(self): + chunks = _drain_async( + [ + {"type": "response.created"}, + {"type": "response.failed", "response": self._failed_response()}, + ] + ) + + assert [c["type"] for c in chunks] == ["message_start", "error"] + assert chunks[1] == { + "type": "error", + "error": { + "type": "insufficient_quota", + "message": "You have no credits remaining. Add credits to continue using the API.", + }, + } + assert not [c for c in chunks if c["type"] == "message_delta"] + assert not [c for c in chunks if c["type"] == "message_stop"] + + def test_response_failed_with_dict_error_object(self): + chunks = _process_all( + [ + { + "type": "response.failed", + "response": { + "status": "failed", + "error": {"code": "insufficient_quota", "message": "no credits"}, + }, + } + ] + ) + + assert chunks == [{"type": "error", "error": {"type": "insufficient_quota", "message": "no credits"}}] + + def test_response_failed_without_error_object_still_fails(self): + response = SimpleNamespace(status="failed", usage=None, output=[], error=None) + chunks = _process_all([{"type": "response.failed", "response": response}]) + + assert chunks == [{"type": "error", "error": {"type": "api_error", "message": "Upstream response failed"}}] + + def test_response_completed_still_emits_successful_turn(self): + response = SimpleNamespace(status="completed", usage=None, output=[]) + chunks = _process_all([{"type": "response.completed", "response": response}]) + + assert [c["type"] for c in chunks] == ["message_delta", "message_stop"] + assert chunks[0]["delta"]["stop_reason"] == "end_turn" + + +class TestMidStreamFallbackErrorNotSwallowed: + """Regression for https://github.com/BerriAI/litellm/issues/39703 (1.97+). + + The Responses stream iterator raises ``MidStreamFallbackError`` on a failed + event. ``__anext__`` used to catch it in a blanket ``except Exception``, + log, and fall through to ``StopAsyncIteration``, leaving the client with a + lone ``message_start`` and an unterminated stream. It must be surfaced as + the Anthropic streaming spec's terminal error event instead. + """ + + @staticmethod + def _stream_raising_mid_stream_fallback() -> list: + from litellm.exceptions import MidStreamFallbackError + + async def _gen(): + yield {"type": "response.created"} + raise MidStreamFallbackError( + message="litellm.APIError: API Error: Status Code 429", + model="gpt-4o-mini", + llm_provider="openai", + original_exception=SimpleNamespace(status_code=429, message="Rate limit exceeded"), + ) + + async def _run() -> list: + wrapper = AnthropicResponsesStreamWrapper(responses_stream=_gen(), model="gpt-4o-mini") + return [chunk async for chunk in wrapper] + + return asyncio.run(_run()) + + def test_mid_stream_fallback_error_becomes_terminal_error_event(self): + chunks = self._stream_raising_mid_stream_fallback() + + assert [c["type"] for c in chunks] == ["message_start", "error"] + assert chunks[1]["type"] == "error" + assert chunks[1]["error"]["type"] == "rate_limit_error" + assert chunks[1]["error"]["message"] == "Rate limit exceeded" + + def test_stream_terminates_after_the_error_event(self): + wrapper = AnthropicResponsesStreamWrapper(responses_stream=None, model="gpt-4o-mini") + wrapper._sent_message_start = True # real path: message_start precedes the failure + wrapper._chunk_queue.append({"type": "error", "error": {"type": "api_error", "message": "boom"}}) + + async def _drain(): + return [chunk async for chunk in wrapper] + + chunks = asyncio.run(_drain()) + assert [c["type"] for c in chunks] == ["error"]