diff --git a/litellm/completion_extras/litellm_responses_transformation/handler.py b/litellm/completion_extras/litellm_responses_transformation/handler.py index 0d61914c2ba..b2567b558b1 100644 --- a/litellm/completion_extras/litellm_responses_transformation/handler.py +++ b/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: diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index eafe0ae5ffd..8c7ba9350d8 100644 --- a/litellm/responses/streaming_iterator.py +++ b/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: diff --git a/tests/test_litellm/test_responses_api_bridge_non_stream.py b/tests/test_litellm/test_responses_api_bridge_non_stream.py index 617b2cfc031..c438a409f0a 100644 --- a/tests/test_litellm/test_responses_api_bridge_non_stream.py +++ b/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", diff --git a/tests/test_litellm/test_responses_api_bridge_recover_output.py b/tests/test_litellm/test_responses_api_bridge_recover_output.py deleted file mode 100644 index 7b38e6a6ce5..00000000000 --- a/tests/test_litellm/test_responses_api_bridge_recover_output.py +++ /dev/null @@ -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])