Merge 7dbd808bd1 into f2e0a5db1e

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  # noqa: TID251  # untyped upstream event payloads
+
+from typing_extensions import TypeIs  # noqa: TID251  # runtime isinstance-narrowing of attr-or-dict event payloads
 
 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]] = {  # mutable-ok: constant, never mutated
+    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:
     """
@@ -60,7 +79,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._chunk_queue: deque[dict[str, object]] = deque()
+        self._chunk_queue: deque[dict[str, object]] = deque()  # mutable-ok: SSE frame queue, drained via popleft
         self._refusal_text: str = ""
         self._sync_responses_iterator: Iterator[object] | None = None
 
@@ -84,6 +103,20 @@ class AnthropicResponsesStreamWrapper:
             },
         }
 
+    @staticmethod
+    def _make_error_event(
+        error_type: str | None, error_message: str | None
+    ) -> dict[str, object]:  # mutable-ok: SSE frame payload shape
+        """The Anthropic streaming spec's terminal error frame."""
+        event: dict[str, object] = {  # mutable-ok: terminal error frame payload
+            "type": "error",
+            "error": {  # mutable-ok: nested Anthropic error object
+                "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 +326,42 @@ class AnthropicResponsesStreamWrapper:
             )
             return
 
+        # ---- response failed -> terminal error event ----
+        if event_type == "response.failed":
+            event_obj = cast("object", event)  # cast-ok: pins Any event to object
+            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 +447,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())
 

tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py

@@ -234,9 +234,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
 
@@ -452,3 +450,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"]