fix(anthropic): preserve response stream failures

Co-Authored-By: Claude Code <noreply@anthropic.com>

litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py

@@ -1,17 +1,20 @@
 # What is this?
 ## Translates OpenAI call to Anthropic `/v1/messages` format
 import json
-import traceback
 from collections import deque
 from collections.abc import AsyncIterator, Mapping
 from typing import Any, Final
 
-from litellm import verbose_logger
 from litellm._uuid import uuid
-from litellm.types.llms.anthropic_messages.anthropic_response import AnthropicUsage
+from litellm.exceptions import APIError, MidStreamFallbackError
 
 from .transformation import LiteLLMAnthropicToResponsesAPIAdapter
 
+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."
+)
+
 
 class AnthropicResponsesStreamWrapper:
     """
@@ -38,10 +41,14 @@ class AnthropicResponsesStreamWrapper:
         self._current_block_index: int = -1
         # Map item_id -> content_block_index so we can stop the right block later
         self._item_id_to_block_index: dict[str, int] = {}
-        # Track open function_call items by item_id so we can emit tool_use start
-        self._pending_tool_ids: dict[str, str] = {}  # item_id -> call_id / name accumulator
+        self._output_item_types: dict[str, str] = {}
+        self._function_call_argument_deltas: dict[str, list[str]] = {}
+        self._finalized_function_call_item_ids: set[str] = set()
+        self._closed_output_item_ids: set[str] = set()
         self._sent_message_start = False
         self._sent_message_stop = False
+        self._sent_error = False
+        self._open_block_indexes: set[int] = set()
         self._chunk_queue: deque = deque()
 
     def _make_message_start(self) -> dict[str, Any]:
@@ -72,6 +79,7 @@ class AnthropicResponsesStreamWrapper:
         block_idx = self._next_block_index()
         if item_id:
             self._item_id_to_block_index[item_id] = block_idx
+        self._open_block_indexes.add(block_idx)
         self._chunk_queue.append(
             {
                 "type": "content_block_start",
@@ -81,13 +89,88 @@ class AnthropicResponsesStreamWrapper:
         )
         return block_idx
 
+    @staticmethod
+    def _event_value(event: object, name: str) -> object | None:
+        if isinstance(event, Mapping):
+            return event.get(name)
+        return getattr(event, name, None)
+
+    def _queue_error(self, message: str) -> None:
+        if self._sent_message_stop or self._sent_error:
+            return
+        self._chunk_queue.append({"type": "error", "error": {"type": "api_error", "message": message}})
+        self._sent_error = True
+
+    def _queue_completion(self, response_obj: object, stop_reason: str) -> None:
+        if self._sent_message_stop or self._sent_error:
+            return
+        anthropic_usage: Final = LiteLLMAnthropicToResponsesAPIAdapter.translate_responses_api_usage_to_anthropic_usage(
+            self._event_value(response_obj, "usage")
+        )
+        self._chunk_queue.append(
+            {
+                "type": "message_delta",
+                "delta": {"stop_reason": stop_reason, "stop_sequence": None},
+                "usage": dict(anthropic_usage),
+            }
+        )
+        self._chunk_queue.append({"type": "message_stop"})
+        self._sent_message_stop = True
+
+    def _has_replayable_incomplete_output(self, response_obj: object) -> bool:
+        output: Final = self._event_value(response_obj, "output")
+        output_items: Final = output if isinstance(output, list) else ()
+        if not output_items or self._open_block_indexes:
+            return False
+
+        for output_item in output_items:
+            item_id: Final = self._event_value(output_item, "id")
+            item_type: Final = self._event_value(output_item, "type")
+            if (
+                not isinstance(item_id, str)
+                or item_type not in {"message", "reasoning", "function_call"}
+                or self._output_item_types.get(item_id) != item_type
+                or item_id not in self._closed_output_item_ids
+            ):
+                return False
+            if item_type == "function_call" and item_id not in self._finalized_function_call_item_ids:
+                return False
+        return True
+
+    def _has_complete_terminal_state(self) -> bool:
+        return not self._open_block_indexes
+
+    def _has_complete_function_calls(self, output_items: object) -> bool:
+        if not isinstance(output_items, list):
+            return True
+        return all(
+            self._event_value(output_item, "type") != "function_call"
+            or (
+                isinstance(self._event_value(output_item, "id"), str)
+                and self._event_value(output_item, "id") in self._finalized_function_call_item_ids
+            )
+            for output_item in output_items
+        )
+
+    def _response_incomplete_reason(self, response_obj: object) -> str | None:
+        incomplete_details: Final = self._event_value(response_obj, "incomplete_details")
+        return_value: Final = self._event_value(incomplete_details, "reason")
+        return return_value if isinstance(return_value, str) else None
+
+    def _process_incomplete_response(self, response_obj: object) -> None:
+        incomplete_reason: Final = self._response_incomplete_reason(response_obj)
+        if incomplete_reason == "max_output_tokens" and self._has_replayable_incomplete_output(response_obj):
+            self._queue_completion(response_obj, "max_tokens")
+            return
+        self._queue_error("Provider returned an incomplete response that cannot be safely continued.")
+
     def _process_event(self, event: Any) -> None:
         """Convert one Responses API event into zero or more Anthropic chunks queued for emission."""
         event_type = getattr(event, "type", None)
         if event_type is None and isinstance(event, dict):
             event_type = event.get("type")
 
-        if event_type is None:
+        if event_type is None or self._sent_message_stop or self._sent_error:
             return
 
         # ---- message_start ----
@@ -105,6 +188,9 @@ class AnthropicResponsesStreamWrapper:
             item_type: Final = getattr(item, "type", None) or (item.get("type") if isinstance(item, dict) else None)
             item_id = getattr(item, "id", None) or (item.get("id") if isinstance(item, dict) else None)
 
+            if isinstance(item_id, str) and isinstance(item_type, str):
+                self._output_item_types[item_id] = item_type
+
             if item_type == "message":
                 self._open_block(item_id, {"type": "text", "text": ""})
             elif item_type == "function_call":
@@ -112,8 +198,6 @@ class AnthropicResponsesStreamWrapper:
                     getattr(item, "call_id", None) or (item.get("call_id") if isinstance(item, dict) else None) or ""
                 )
                 name = getattr(item, "name", None) or (item.get("name") if isinstance(item, dict) else None) or ""
-                if item_id:
-                    self._pending_tool_ids[item_id] = call_id
                 self._open_block(
                     item_id,
                     {
@@ -131,9 +215,8 @@ class AnthropicResponsesStreamWrapper:
             delta = getattr(event, "delta", "") or (event.get("delta", "") if isinstance(event, dict) else "")
             block_idx = self._item_id_to_block_index.get(item_id, -1) if item_id else self._current_block_index
             if block_idx < 0:
-                # Some providers (e.g. LMStudio) skip response.output_item.added,
-                # so no text block is open yet; synthesize content_block_start
-                # instead of emitting a delta with index -1
+                if isinstance(item_id, str):
+                    self._output_item_types[item_id] = "message"
                 block_idx = self._open_block(item_id, {"type": "text", "text": ""})
             self._chunk_queue.append(
                 {
@@ -169,11 +252,14 @@ class AnthropicResponsesStreamWrapper:
         if event_type == "response.function_call_arguments.delta":
             item_id = getattr(event, "item_id", None) or (event.get("item_id") if isinstance(event, dict) else None)
             delta = getattr(event, "delta", "") or (event.get("delta", "") if isinstance(event, dict) else "")
-            block_idx = (
-                self._item_id_to_block_index.get(item_id, self._current_block_index)
-                if item_id
-                else self._current_block_index
-            )
+            if not isinstance(item_id, str) or self._output_item_types.get(item_id) != "function_call":
+                return
+            if not isinstance(delta, str):
+                return
+            self._function_call_argument_deltas.setdefault(item_id, []).append(delta)
+            block_idx = self._item_id_to_block_index.get(item_id, -1)
+            if block_idx < 0:
+                return
             self._chunk_queue.append(
                 {
                     "type": "content_block_delta",
@@ -183,15 +269,41 @@ class AnthropicResponsesStreamWrapper:
             )
             return
 
+        if event_type == "response.function_call_arguments.done":
+            item_id = getattr(event, "item_id", None) or (event.get("item_id") if isinstance(event, dict) else None)
+            arguments = getattr(event, "arguments", None) or (
+                event.get("arguments") if isinstance(event, dict) else None
+            )
+            argument_deltas: Final = (
+                self._function_call_argument_deltas.pop(item_id, []) if isinstance(item_id, str) else []
+            )
+            if (
+                isinstance(item_id, str)
+                and self._output_item_types.get(item_id) == "function_call"
+                and isinstance(arguments, str)
+                and "".join(argument_deltas) == arguments
+            ):
+                try:
+                    parsed_arguments: Final = json.loads(arguments)
+                except json.JSONDecodeError:
+                    return
+                if not isinstance(parsed_arguments, Mapping):
+                    return
+                self._finalized_function_call_item_ids.add(item_id)
+            return
+
         # ---- output item done -> content_block_stop ----
         if event_type == "response.output_item.done":
             item = getattr(event, "item", None) or (event.get("item") if isinstance(event, dict) else None)
             item_id = (
                 getattr(item, "id", None) or (item.get("id") if isinstance(item, dict) else None) if item else None
             )
+            if isinstance(item_id, str):
+                self._closed_output_item_ids.add(item_id)
             block_idx = self._item_id_to_block_index.get(item_id, -1) if item_id else self._current_block_index
-            if block_idx < 0:
+            if block_idx < 0 or block_idx not in self._open_block_indexes:
                 return
+            self._open_block_indexes.remove(block_idx)
             self._chunk_queue.append(
                 {
                     "type": "content_block_stop",
@@ -200,48 +312,40 @@ class AnthropicResponsesStreamWrapper:
             )
             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 (
-                event.get("response") if isinstance(event, dict) else None
+        response_obj: Final = self._event_value(event, "response")
+        if event_type == "response.completed":
+            if response_obj is None:
+                self._queue_error("Provider completed a response without a response body.")
+                return
+            if self._event_value(response_obj, "status") == "incomplete":
+                self._process_incomplete_response(response_obj)
+                return
+            if not self._has_complete_terminal_state():
+                self._queue_error("Provider completed a response with an unclosed content block.")
+                return
+            output: Final = self._event_value(response_obj, "output")
+            output_items: Final = output if isinstance(output, list) else ()
+            if not self._has_complete_function_calls(output_items):
+                self._queue_error("Provider completed a response with an incomplete function call.")
+                return
+            stop_reason: Final = (
+                "tool_use"
+                if any(self._event_value(output_item, "type") == "function_call" for output_item in output_items)
+                else "end_turn"
             )
-            stop_reason = "end_turn"
-            anthropic_usage: AnthropicUsage = AnthropicUsage(input_tokens=0, output_tokens=0)
+            self._queue_completion(response_obj, stop_reason)
+            return
 
+        if event_type == "response.failed":
+            message: Final = self._event_value(self._event_value(response_obj, "error"), "message")
+            self._queue_error(message if isinstance(message, str) else "Provider failed to generate a response.")
+            return
+
+        if event_type == "response.incomplete":
             if response_obj is not None:
-                status: Final = getattr(response_obj, "status", None)
-                if status == "incomplete":
-                    stop_reason = "max_tokens"
-                anthropic_usage = (
-                    LiteLLMAnthropicToResponsesAPIAdapter.translate_responses_api_usage_to_anthropic_usage(
-                        getattr(response_obj, "usage", None)
-                    )
-                )
-
-            # Check if tool_use was in the output to override stop_reason
-            if response_obj is not None:
-                output: Final = getattr(response_obj, "output", []) or []
-                for out_item in output:
-                    out_type = getattr(out_item, "type", None) or (
-                        out_item.get("type") if isinstance(out_item, dict) else None
-                    )
-                    if out_type == "function_call":
-                        stop_reason = "tool_use"
-                        break
-
-            self._chunk_queue.append(
-                {
-                    "type": "message_delta",
-                    "delta": {"stop_reason": stop_reason, "stop_sequence": None},
-                    "usage": dict(anthropic_usage),
-                }
-            )
-            self._chunk_queue.append({"type": "message_stop"})
-            self._sent_message_stop = True
+                self._process_incomplete_response(response_obj)
+                return
+            self._queue_error("Provider returned an incomplete response that cannot be safely continued.")
             return
 
     def __aiter__(self) -> "AnthropicResponsesStreamWrapper":
@@ -258,21 +362,23 @@ class AnthropicResponsesStreamWrapper:
             self._chunk_queue.append(self._make_message_start())
             return self._chunk_queue.popleft()
 
-        # Consume the upstream stream
         try:
             async for event in self.responses_stream:
                 self._process_event(event)
                 if self._chunk_queue:
                     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 (APIError, MidStreamFallbackError):
+            raise
+        except Exception:
+            self._queue_error("Provider stream failed before a terminal response was emitted.")
 
-        # Drain any remaining queued chunks
         if self._chunk_queue:
             return self._chunk_queue.popleft()
 
+        if not self._sent_message_stop and not self._sent_error:
+            self._queue_error(INCOMPLETE_STREAM_ERROR_MESSAGE)
+            return self._chunk_queue.popleft()
+
         raise StopAsyncIteration
 
     async def async_anthropic_sse_wrapper(self) -> AsyncIterator[bytes]:

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

@@ -4,10 +4,14 @@ Tests for AnthropicResponsesStreamWrapper
 """
 
 import asyncio
+import json
 import os
 import sys
 from types import SimpleNamespace
 
+import litellm
+from litellm.exceptions import MidStreamFallbackError
+
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../../../..")))
 
 from litellm.llms.anthropic.experimental_pass_through.responses_adapters.streaming_iterator import (
@@ -34,6 +38,43 @@ def _drain_async(events: list) -> list:
     return asyncio.run(_run())
 
 
+def _drain_sse(events: list) -> list[tuple[str, dict]]:
+    async def _gen():
+        for event in events:
+            yield event
+
+    async def _run() -> list[bytes]:
+        wrapper = AnthropicResponsesStreamWrapper(responses_stream=_gen(), model="m")
+        return [chunk async for chunk in wrapper.async_anthropic_sse_wrapper()]
+
+    frames = asyncio.run(_run())
+    return [
+        (
+            frame.decode().split("\n", maxsplit=1)[0].removeprefix("event: "),
+            json.loads(frame.decode().split("\n", maxsplit=2)[1].removeprefix("data: ")),
+        )
+        for frame in frames
+    ]
+
+
+def _response(
+    *,
+    status: str,
+    output: list[object] | None = None,
+    incomplete_reason: str | None = None,
+    error_message: str | None = None,
+) -> SimpleNamespace:
+    incomplete_details = SimpleNamespace(reason=incomplete_reason) if incomplete_reason is not None else None
+    error = SimpleNamespace(message=error_message) if error_message is not None else None
+    return SimpleNamespace(
+        status=status,
+        output=output or [],
+        incomplete_details=incomplete_details,
+        error=error,
+        usage=None,
+    )
+
+
 class TestMessageStartEmittedExactlyOnce:
     """The ``__anext__`` fallback emits ``message_start`` before consuming the
     stream, so ``_process_event`` must not emit a second one when
@@ -112,7 +153,7 @@ class TestReasoningItemWithoutSummaryText:
         ]
 
     def test_reasoning_without_summary_emits_no_thinking_block(self):
-        chunks = _drain_async(self._gpt_turn(reasoning_summary_deltas=[]))
+        chunks = _process_all(self._gpt_turn(reasoning_summary_deltas=[]))
 
         assert not [
             c for c in chunks if c["type"] == "content_block_start" and c["content_block"]["type"] == "thinking"
@@ -133,7 +174,7 @@ class TestReasoningItemWithoutSummaryText:
         ]
 
     def test_reasoning_with_summary_text_still_emits_a_thinking_block(self):
-        chunks = _drain_async(self._gpt_turn(reasoning_summary_deltas=["Weigh", "ing options"]))
+        chunks = _process_all(self._gpt_turn(reasoning_summary_deltas=["Weigh", "ing options"]))
 
         assert [(c["type"], c.get("index")) for c in chunks[1:]] == [
             ("content_block_start", 0),
@@ -207,7 +248,7 @@ class TestToolUseBlockClosedExactlyOnce:
         assert stops == [0]
 
     def test_tool_turn_event_order(self):
-        chunks = _drain_async(self._chat_completions_bridge_tool_turn())
+        chunks = _process_all(self._chat_completions_bridge_tool_turn())
 
         assert [(c["type"], c.get("index")) for c in chunks] == [
             ("message_start", None),
@@ -223,6 +264,17 @@ class TestToolUseBlockClosedExactlyOnce:
             "input": {},
         }
 
+    def test_duplicate_known_output_item_done_emits_one_content_block_stop(self):
+        chunks = _process_all(
+            [
+                {"type": "response.output_item.added", "item": {"type": "message", "id": "m1"}},
+                {"type": "response.output_item.done", "item": {"type": "message", "id": "m1"}},
+                {"type": "response.output_item.done", "item": {"type": "message", "id": "m1"}},
+            ]
+        )
+
+        assert [chunk["type"] for chunk in chunks] == ["content_block_start", "content_block_stop"]
+
 
 class TestProcessEventTextDeltaWithoutOutputItemAdded:
     """Streams that skip response.output_item.added (e.g. LMStudio) must still
@@ -308,3 +360,372 @@ class TestResponseCompletedUsage:
             "cache_creation_input_tokens": 10,
             "cache_read_input_tokens": 4004,
         }
+
+
+class TestTerminalResponses:
+    def test_response_failed_emits_one_anthropic_error_not_a_completion(self):
+        frames = _drain_sse(
+            [
+                {
+                    "type": "response.failed",
+                    "response": _response(status="failed", error_message="upstream failed"),
+                }
+            ]
+        )
+
+        assert [event_type for event_type, _ in frames] == ["message_start", "error"]
+        assert frames[-1][1]["error"]["type"] == "api_error"
+        assert frames[-1][1]["error"]["message"] == "upstream failed"
+
+    def test_completed_response_with_an_open_block_emits_error_not_message_stop(self):
+        frames = _drain_sse(
+            [
+                {"type": "response.created"},
+                {"type": "response.output_item.added", "item": {"type": "message", "id": "m1"}},
+                {"type": "response.output_text.delta", "item_id": "m1", "delta": "partial"},
+                {
+                    "type": "response.completed",
+                    "response": _response(status="completed", output=[SimpleNamespace(type="message")]),
+                },
+            ]
+        )
+
+        assert [event_type for event_type, _ in frames][-1] == "error"
+        assert not [event_type for event_type, _ in frames if event_type == "message_stop"]
+
+    def test_events_after_a_terminal_response_are_ignored(self):
+        frames = _drain_sse(
+            [
+                {
+                    "type": "response.completed",
+                    "response": _response(status="completed"),
+                },
+                {"type": "response.output_text.delta", "item_id": "m1", "delta": "late"},
+            ]
+        )
+
+        assert [event_type for event_type, _ in frames] == ["message_start", "message_delta", "message_stop"]
+
+    def test_completed_response_with_unfinalized_function_call_emits_error(self):
+        frames = _drain_sse(
+            [
+                {"type": "response.created"},
+                {
+                    "type": "response.output_item.added",
+                    "item": {"type": "function_call", "id": "call_1", "call_id": "call_1", "name": "lookup"},
+                },
+                {"type": "response.function_call_arguments.delta", "item_id": "call_1", "delta": '{"id":'},
+                {"type": "response.output_item.done", "item": {"type": "function_call", "id": "call_1"}},
+                {
+                    "type": "response.completed",
+                    "response": _response(
+                        status="completed", output=[SimpleNamespace(type="function_call", id="call_1")]
+                    ),
+                },
+            ]
+        )
+
+        assert [event_type for event_type, _ in frames][-1] == "error"
+        assert not [event_type for event_type, _ in frames if event_type == "message_stop"]
+
+    def test_completed_event_with_incomplete_status_requires_safe_max_tokens_completion(self):
+        response = _response(
+            status="incomplete",
+            incomplete_reason="max_output_tokens",
+            output=[SimpleNamespace(type="message", id="m1")],
+        )
+        chunks = _drain_async(
+            [
+                {"type": "response.created"},
+                {"type": "response.output_item.added", "item": {"type": "message", "id": "m1"}},
+                {"type": "response.output_text.delta", "item_id": "m1", "delta": "partial"},
+                {"type": "response.output_item.done", "item": {"type": "message", "id": "m1"}},
+                {"type": "response.completed", "response": response},
+            ]
+        )
+
+        assert [chunk["type"] for chunk in chunks][-2:] == ["message_delta", "message_stop"]
+        assert chunks[-2]["delta"]["stop_reason"] == "max_tokens"
+
+    def test_completed_event_with_unsafe_incomplete_status_emits_error(self):
+        frames = _drain_sse(
+            [
+                {
+                    "type": "response.completed",
+                    "response": _response(status="incomplete", incomplete_reason="max_output_tokens"),
+                }
+            ]
+        )
+
+        assert [event_type for event_type, _ in frames] == ["message_start", "error"]
+        assert not [event_type for event_type, _ in frames if event_type == "message_stop"]
+
+    def test_empty_incomplete_response_emits_error_not_max_tokens_completion(self):
+        frames = _drain_sse(
+            [
+                {
+                    "type": "response.incomplete",
+                    "response": _response(status="incomplete", incomplete_reason="max_output_tokens"),
+                }
+            ]
+        )
+
+        assert [event_type for event_type, _ in frames] == ["message_start", "error"]
+        assert not [event_type for event_type, _ in frames if event_type in {"message_delta", "message_stop"}]
+
+    def test_completed_text_before_max_output_tokens_is_a_safe_max_tokens_completion(self):
+        response = _response(
+            status="incomplete",
+            incomplete_reason="max_output_tokens",
+            output=[SimpleNamespace(type="message", id="m1")],
+        )
+        chunks = _drain_async(
+            [
+                {"type": "response.created"},
+                {"type": "response.output_item.added", "item": {"type": "message", "id": "m1"}},
+                {"type": "response.output_text.delta", "item_id": "m1", "delta": "partial"},
+                {"type": "response.output_item.done", "item": {"type": "message", "id": "m1"}},
+                {"type": "response.incomplete", "response": response},
+            ]
+        )
+
+        assert [chunk["type"] for chunk in chunks] == [
+            "message_start",
+            "content_block_start",
+            "content_block_delta",
+            "content_block_stop",
+            "message_delta",
+            "message_stop",
+        ]
+        assert chunks[-2]["delta"]["stop_reason"] == "max_tokens"
+
+    def test_text_without_added_item_before_max_output_tokens_is_a_safe_max_tokens_completion(self):
+        response = _response(
+            status="incomplete",
+            incomplete_reason="max_output_tokens",
+            output=[SimpleNamespace(type="message", id="m1")],
+        )
+        chunks = _drain_async(
+            [
+                {"type": "response.created"},
+                {"type": "response.output_text.delta", "item_id": "m1", "delta": "partial"},
+                {"type": "response.output_item.done", "item": {"type": "message", "id": "m1"}},
+                {"type": "response.incomplete", "response": response},
+            ]
+        )
+
+        assert [chunk["type"] for chunk in chunks][-2:] == ["message_delta", "message_stop"]
+        assert chunks[-2]["delta"]["stop_reason"] == "max_tokens"
+
+    def test_closed_reasoning_without_a_summary_is_a_safe_max_tokens_completion(self):
+        response = _response(
+            status="incomplete",
+            incomplete_reason="max_output_tokens",
+            output=[SimpleNamespace(type="reasoning", id="rs_1")],
+        )
+        chunks = _drain_async(
+            [
+                {"type": "response.created"},
+                {"type": "response.output_item.added", "item": {"type": "reasoning", "id": "rs_1"}},
+                {"type": "response.output_item.done", "item": {"type": "reasoning", "id": "rs_1"}},
+                {"type": "response.incomplete", "response": response},
+            ]
+        )
+
+        assert [chunk["type"] for chunk in chunks] == ["message_start", "message_delta", "message_stop"]
+        assert chunks[-2]["delta"]["stop_reason"] == "max_tokens"
+
+    def test_closed_function_call_with_finalized_json_is_a_safe_max_tokens_completion(self):
+        response = _response(
+            status="incomplete",
+            incomplete_reason="max_output_tokens",
+            output=[SimpleNamespace(type="function_call", id="call_1")],
+        )
+        chunks = _drain_async(
+            [
+                {"type": "response.created"},
+                {
+                    "type": "response.output_item.added",
+                    "item": {"type": "function_call", "id": "call_1", "call_id": "call_1", "name": "lookup"},
+                },
+                {"type": "response.function_call_arguments.delta", "item_id": "call_1", "delta": '{"id": 1}'},
+                {
+                    "type": "response.function_call_arguments.done",
+                    "item_id": "call_1",
+                    "arguments": '{"id": 1}',
+                },
+                {"type": "response.output_item.done", "item": {"type": "function_call", "id": "call_1"}},
+                {"type": "response.incomplete", "response": response},
+            ]
+        )
+
+        assert [chunk["type"] for chunk in chunks] == [
+            "message_start",
+            "content_block_start",
+            "content_block_delta",
+            "content_block_stop",
+            "message_delta",
+            "message_stop",
+        ]
+        assert chunks[-2]["delta"]["stop_reason"] == "max_tokens"
+
+    def test_closed_function_call_with_scalar_json_emits_error(self):
+        frames = _drain_sse(
+            [
+                {"type": "response.created"},
+                {
+                    "type": "response.output_item.added",
+                    "item": {"type": "function_call", "id": "call_1", "call_id": "call_1", "name": "lookup"},
+                },
+                {"type": "response.function_call_arguments.delta", "item_id": "call_1", "delta": "1"},
+                {"type": "response.function_call_arguments.done", "item_id": "call_1", "arguments": "1"},
+                {"type": "response.output_item.done", "item": {"type": "function_call", "id": "call_1"}},
+                {
+                    "type": "response.incomplete",
+                    "response": _response(
+                        status="incomplete",
+                        incomplete_reason="max_output_tokens",
+                        output=[SimpleNamespace(type="function_call", id="call_1")],
+                    ),
+                },
+            ]
+        )
+
+        assert [event_type for event_type, _ in frames][-1] == "error"
+        assert not [event_type for event_type, _ in frames if event_type == "message_stop"]
+
+    def test_closed_function_call_with_mismatched_finalized_json_emits_error(self):
+        frames = _drain_sse(
+            [
+                {"type": "response.created"},
+                {
+                    "type": "response.output_item.added",
+                    "item": {"type": "function_call", "id": "call_1", "call_id": "call_1", "name": "lookup"},
+                },
+                {"type": "response.function_call_arguments.delta", "item_id": "call_1", "delta": '{"id": 1}'},
+                {
+                    "type": "response.function_call_arguments.done",
+                    "item_id": "call_1",
+                    "arguments": '{"id": 2}',
+                },
+                {"type": "response.output_item.done", "item": {"type": "function_call", "id": "call_1"}},
+                {
+                    "type": "response.incomplete",
+                    "response": _response(
+                        status="incomplete",
+                        incomplete_reason="max_output_tokens",
+                        output=[SimpleNamespace(type="function_call", id="call_1")],
+                    ),
+                },
+            ]
+        )
+
+        assert [event_type for event_type, _ in frames][-1] == "error"
+        assert not [event_type for event_type, _ in frames if event_type == "message_stop"]
+
+    def test_closed_function_call_without_finalized_json_emits_error(self):
+        frames = _drain_sse(
+            [
+                {"type": "response.created"},
+                {
+                    "type": "response.output_item.added",
+                    "item": {"type": "function_call", "id": "call_1", "call_id": "call_1", "name": "lookup"},
+                },
+                {"type": "response.function_call_arguments.delta", "item_id": "call_1", "delta": '{"id":'},
+                {"type": "response.output_item.done", "item": {"type": "function_call", "id": "call_1"}},
+                {
+                    "type": "response.incomplete",
+                    "response": _response(
+                        status="incomplete",
+                        incomplete_reason="max_output_tokens",
+                        output=[SimpleNamespace(type="function_call", id="call_1")],
+                    ),
+                },
+            ]
+        )
+
+        assert [event_type for event_type, _ in frames][-1] == "error"
+        assert not [event_type for event_type, _ in frames if event_type == "message_stop"]
+
+    def test_incomplete_tool_call_emits_error_not_tool_use_completion(self):
+        frames = _drain_sse(
+            [
+                {"type": "response.created"},
+                {
+                    "type": "response.output_item.added",
+                    "item": {"type": "function_call", "id": "call_1", "call_id": "call_1", "name": "lookup"},
+                },
+                {"type": "response.function_call_arguments.delta", "item_id": "call_1", "delta": '{"id":'},
+                {
+                    "type": "response.incomplete",
+                    "response": _response(
+                        status="incomplete",
+                        incomplete_reason="max_output_tokens",
+                        output=[SimpleNamespace(type="function_call")],
+                    ),
+                },
+            ]
+        )
+
+        assert [event_type for event_type, _ in frames][-1] == "error"
+        assert not [
+            payload
+            for event_type, payload in frames
+            if event_type == "message_delta" and payload["delta"]["stop_reason"] == "tool_use"
+        ]
+
+    def test_silent_eof_emits_one_terminal_error(self):
+        frames = _drain_sse([{"type": "response.created"}])
+
+        assert [event_type for event_type, _ in frames] == ["message_start", "error"]
+        assert frames[-1][1]["error"]["type"] == "api_error"
+
+    def test_upstream_api_error_propagates_to_router(self):
+        upstream_error = litellm.APIError(
+            status_code=500,
+            message="upstream failed",
+            llm_provider="openai",
+            model="m",
+        )
+
+        async def _gen():
+            raise upstream_error
+            yield None
+
+        async def _run() -> None:
+            wrapper = AnthropicResponsesStreamWrapper(responses_stream=_gen(), model="m")
+            async for _ in wrapper:
+                pass
+
+        try:
+            asyncio.run(_run())
+        except litellm.APIError as error:
+            assert error is upstream_error
+        else:
+            raise AssertionError("expected APIError to propagate")
+
+    def test_midstream_fallback_error_propagates_to_router(self):
+        upstream_error = MidStreamFallbackError(
+            message="upstream failed",
+            model="m",
+            llm_provider="openai",
+            generated_content="",
+            is_pre_first_chunk=True,
+        )
+
+        async def _gen():
+            raise upstream_error
+            yield None
+
+        async def _run() -> None:
+            wrapper = AnthropicResponsesStreamWrapper(responses_stream=_gen(), model="m")
+            async for _ in wrapper:
+                pass
+
+        try:
+            asyncio.run(_run())
+        except MidStreamFallbackError as error:
+            assert error is upstream_error
+            assert error.is_pre_first_chunk
+        else:
+            raise AssertionError("expected MidStreamFallbackError to propagate")