fix(review): collision-safe streamed index, named suppression, merge tests into mapped file

Review follow-up for #41014:

- Replace the len()-based fallback index with max(known)+1 so a missing or
  duplicate output_index can never collide with a real index and silently
  discard a streamed item.
- Replace the banned '# type: ignore' with a named pyright suppression
  (reportAssignmentType) carrying a reason, per repo directive (LIT009).
- Move the regression tests into the existing mapped test file
  (tests/test_litellm/test_responses_api_bridge_non_stream.py) instead of a
  separate file, and add the missing async coverage for
  _collect_response_from_stream_async.

litellm/completion_extras/litellm_responses_transformation/handler.py

@@ -94,7 +94,7 @@ class ResponsesToCompletionBridgeHandler:
         if len(response.output) == 0 and callable(get_streamed_items):
             streamed_items: Final = get_streamed_items()
             if streamed_items:
-                response.output = streamed_items  # type: ignore[assignment]
+                response.output = streamed_items  # pyright: ignore[reportAssignmentType]  # streamed items are the typed SSE output items
         return response
 
     async def _collect_response_from_stream_async(self, stream_iter: Any) -> "ResponsesAPIResponse":
@@ -114,7 +114,7 @@ class ResponsesToCompletionBridgeHandler:
         if len(response.output) == 0 and callable(get_streamed_items):
             streamed_items: Final = get_streamed_items()
             if streamed_items:
-                response.output = streamed_items  # type: ignore[assignment]
+                response.output = streamed_items  # pyright: ignore[reportAssignmentType]  # streamed items are the typed SSE output items
         return response
 
     def validate_input_kwargs(self, kwargs: dict) -> ResponsesToCompletionBridgeHandlerInputKwargs:

litellm/responses/streaming_iterator.py

@@ -408,10 +408,20 @@ class BaseResponsesAPIStreamingIterator:
                     _done_item: Final[object] = getattr(openai_responses_api_chunk, "item", None)
                     if _done_item is not None:
                         _output_index: Final = getattr(openai_responses_api_chunk, "output_index", None)
-                        _index: Final = (
-                            _output_index if isinstance(_output_index, int) else len(self._streamed_output_items)
-                        )
-                        self._streamed_output_items.setdefault(_index, _done_item)
+                        if (
+                            isinstance(_output_index, int)
+                            and not isinstance(_output_index, bool)
+                            and _output_index not in self._streamed_output_items
+                        ):
+                            self._streamed_output_items[_output_index] = _done_item
+                        else:
+                            # missing/invalid/duplicate index: append after the highest known
+                            # index so the fallback can never collide with a real index and
+                            # silently discard a streamed item
+                            _fallback_index: Final = (
+                                max(self._streamed_output_items.keys(), default=-1) + 1
+                            )
+                            self._streamed_output_items[_fallback_index] = _done_item
                 elif _event_type == ResponsesAPIStreamEvents.OUTPUT_TEXT_ANNOTATION_ADDED:
                     _annotation: Final[object] = getattr(openai_responses_api_chunk, "annotation", None)
                     if _annotation is not None:

tests/test_litellm/test_responses_api_bridge_non_stream.py

@@ -65,6 +65,86 @@ def test_should_collect_response_from_stream():
     assert collected.id == "resp-1"
     assert collected._hidden_params.get("headers") == {"x-test": "1"}
 
+class _RecoverableResponsesStream(_FakeResponsesStream):
+    """Terminal response.completed carries an empty output array even though
+    output_item.done events streamed the full answer (issue #41009)."""
+
+    def __init__(self, response, streamed_items):
+        super().__init__(response)
+        self._streamed_items = streamed_items
+
+    def get_streamed_output_items(self):
+        return self._streamed_items
+
+
+class _RecoverableAsyncResponsesStream(_RecoverableResponsesStream):
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        if not self._emitted:
+            self._emitted = True
+            self.completed_response = _CompletedEvent(self._response)
+            return {"type": "response.completed"}
+        raise StopAsyncIteration
+
+
+def test_collect_response_from_stream_rebuilds_empty_output_from_streamed_items():
+    handler = ResponsesToCompletionBridgeHandler()
+    response = ResponsesAPIResponse.model_construct(
+        id="resp-recover",
+        created_at=0,
+        output=[],
+        object="response",
+        model="gpt-5.6-sol",
+    )
+    streamed_items = [
+        {
+            "id": "msg_1",
+            "type": "message",
+            "status": "completed",
+            "role": "assistant",
+            "content": [
+                {"type": "output_text", "text": "Hi! How can I help?", "annotations": []}
+            ],
+        }
+    ]
+    stream = _RecoverableResponsesStream(response, streamed_items)
+
+    collected = handler._collect_response_from_stream(stream)
+
+    assert len(collected.output) == 1
+    assert "Hi! How can I help?" in str(collected.output[0])
+
+
+@pytest.mark.asyncio
+async def test_collect_response_from_stream_async_rebuilds_empty_output_from_streamed_items():
+    handler = ResponsesToCompletionBridgeHandler()
+    response = ResponsesAPIResponse.model_construct(
+        id="resp-recover-async",
+        created_at=0,
+        output=[],
+        object="response",
+        model="gpt-5.6-sol",
+    )
+    streamed_items = [
+        {
+            "id": "msg_1",
+            "type": "message",
+            "status": "completed",
+            "role": "assistant",
+            "content": [
+                {"type": "output_text", "text": "Hi! How can I help?", "annotations": []}
+            ],
+        }
+    ]
+    stream = _RecoverableAsyncResponsesStream(response, streamed_items)
+
+    collected = await handler._collect_response_from_stream_async(stream)
+
+    assert len(collected.output) == 1
+    assert "Hi! How can I help?" in str(collected.output[0])
+
 
 def create_mock_completion_response(
     model: str = "gpt-4",

tests/test_litellm/test_responses_api_bridge_recover_output.py

@@ -1,91 +0,0 @@
-from litellm.completion_extras.litellm_responses_transformation.handler import (
-    ResponsesToCompletionBridgeHandler,
-)
-from litellm.types.llms.openai import ResponsesAPIResponse
-
-
-class _CompletedEvent:
-    def __init__(self, response):
-        self.response = response
-
-
-class _FakeResponsesStream:
-    """
-    Simulates a Responses API stream whose terminal response.completed event carries
-    an empty output array even though output_item.done events streamed the answer
-    (see issue #41009).
-    """
-
-    def __init__(self, response, streamed_items):
-        self._emitted = False
-        self._response = response
-        self._streamed_items = streamed_items
-        self.completed_response = None
-        self._hidden_params = {"headers": {"x-test": "1"}}
-
-    def __iter__(self):
-        return self
-
-    def __next__(self):
-        if not self._emitted:
-            self._emitted = True
-            self.completed_response = _CompletedEvent(self._response)
-            return {"type": "response.completed"}
-        raise StopIteration
-
-    def get_streamed_output_items(self):
-        return self._streamed_items
-
-
-def test_collect_response_from_stream_rebuilds_empty_output_from_streamed_items():
-    handler = ResponsesToCompletionBridgeHandler()
-    response = ResponsesAPIResponse.model_construct(
-        id="resp-1",
-        created_at=0,
-        output=[],
-        object="response",
-        model="gpt-5.6-sol",
-    )
-    streamed_items = [
-        {
-            "id": "msg_1",
-            "type": "message",
-            "status": "completed",
-            "role": "assistant",
-            "content": [
-                {"type": "output_text", "text": "Hi! How can I help?", "annotations": []}
-            ],
-        }
-    ]
-    stream = _FakeResponsesStream(response, streamed_items)
-
-    collected = handler._collect_response_from_stream(stream)
-
-    assert len(collected.output) == 1
-    assert "Hi! How can I help?" in str(collected.output[0])
-
-
-def test_collect_response_from_stream_keeps_nonempty_output_untouched():
-    handler = ResponsesToCompletionBridgeHandler()
-    output = [
-        {
-            "id": "msg_0",
-            "type": "message",
-            "status": "completed",
-            "role": "assistant",
-            "content": [{"type": "output_text", "text": "from terminal event", "annotations": []}],
-        }
-    ]
-    response = ResponsesAPIResponse.model_construct(
-        id="resp-2",
-        created_at=0,
-        output=output,
-        object="response",
-        model="gpt-5.2",
-    )
-    stream = _FakeResponsesStream(response, [{"id": "msg_1", "type": "message"}])
-
-    collected = handler._collect_response_from_stream(stream)
-
-    assert len(collected.output) == 1
-    assert "from terminal event" in str(collected.output[0])