diff --git a/litellm/completion_extras/litellm_responses_transformation/handler.py b/litellm/completion_extras/litellm_responses_transformation/handler.py index 642a78789b2..4984fcd1a0e 100644 --- a/litellm/completion_extras/litellm_responses_transformation/handler.py +++ b/litellm/completion_extras/litellm_responses_transformation/handler.py @@ -87,6 +87,14 @@ class ResponsesToCompletionBridgeHandler: response: Final = self._coerce_response_object(response_obj, hidden_params) if not isinstance(response, ResponsesAPIResponse): raise ValueError("Stream completed response is invalid") + # Some providers emit a terminal response.completed event with an empty output + # array even though output_item.done events streamed the full answer. Rebuild + # the output from those streamed items before failing on empty choices. + get_streamed_items: Final = getattr(stream_iter, "get_streamed_output_items", None) + if len(response.output) == 0 and callable(get_streamed_items): + streamed_items: Final = get_streamed_items() + if streamed_items: + response.output = streamed_items # pyright: ignore[reportAttributeAccessIssue] # assigning typed SSE items back to ResponsesAPIResponse.output return response async def _collect_response_from_stream_async(self, stream_iter: AsyncIterable[object]) -> "ResponsesAPIResponse": @@ -102,6 +110,11 @@ class ResponsesToCompletionBridgeHandler: response: Final = self._coerce_response_object(response_obj, hidden_params) if not isinstance(response, ResponsesAPIResponse): raise ValueError("Stream completed response is invalid") + get_streamed_items: Final = getattr(stream_iter, "get_streamed_output_items", None) + if len(response.output) == 0 and callable(get_streamed_items): + streamed_items: Final = get_streamed_items() + if streamed_items: + response.output = streamed_items # pyright: ignore[reportAttributeAccessIssue] # assigning typed SSE items back to ResponsesAPIResponse.output 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 12bc9adbac8..19b37b953d9 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -324,6 +324,10 @@ class BaseResponsesAPIStreamingIterator: self._completed_response_logged = False self._completed_response_cache_hit: bool | None = None self._persist_completed_response_before_logging = True + # output_item.done events seen during the stream, keyed by output_index. + # Some providers emit a terminal response.completed event with an empty + # output array even though items were streamed; these are used to rebuild it. + self._streamed_output_items: dict[int, object] = {} # mutable-ok: streaming accumulator keyed by output_index self._stream_created_time: float = time.time() # track request context for hooks @@ -431,6 +435,22 @@ class BaseResponsesAPIStreamingIterator: custom_llm_provider=self.custom_llm_provider, model_id=_stream_model_id, ) + if _event_type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE: + _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) + 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: @@ -533,6 +553,18 @@ class BaseResponsesAPIStreamingIterator: self._handle_failure(e) raise + def get_streamed_output_items(self) -> list[object]: # mutable-ok: response.output takes a real list + """ + Output items received via response.output_item.done events, ordered by output_index. + + Used to rebuild a terminal response.completed payload whose output array is empty. + """ + if not self._streamed_output_items: + return [] # mutable-ok: response.output takes a real list + return [ # mutable-ok: response.output takes a real list + item for _, item in sorted(self._streamed_output_items.items()) + ] + def _log_completed_response(self, *, is_async: bool) -> None: if self._completed_response_logged: return diff --git a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py index c5d7ca96a21..0ede5203066 100644 --- a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py +++ b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py @@ -335,3 +335,131 @@ async def test_acompletion_keeps_provider_native_model_id_through_responses( handed_model = responses_call.call_args.kwargs["model"] assert _upstream_model_for(handed_model, custom_llm_provider) == expected_upstream_model +# ------------------------------------------------------------------------ +# #41009/#41014: rebuild empty streamed output from output_item.done events +# ------------------------------------------------------------------------ + + +class _FakeCompletedResponse: + response = { + "id": "resp_41014", + "object": "response", + "created_at": 0, + "status": "completed", + "model": "gpt-5.4", + "output": [], + } + + +class _FakeStreamIter: + def __init__(self, streamed_items): + self.completed_response = _FakeCompletedResponse() + self._hidden_params = None + self._items = streamed_items + + def __iter__(self): + return iter([]) + + def get_streamed_output_items(self): + return self._items + + +class _FakeAsyncStreamIter(_FakeStreamIter): + def __aiter__(self): + return self + + async def __anext__(self): + raise StopAsyncIteration + + +def test_collect_response_from_stream_rebuilds_empty_output_from_streamed_items(): + from litellm.completion_extras.litellm_responses_transformation.handler import ( + ResponsesToCompletionBridgeHandler, + ) + + bridge = ResponsesToCompletionBridgeHandler() + streamed_items = [ + { + "type": "message", + "id": "msg_1", + "role": "assistant", + "content": [{"type": "output_text", "text": "Hello", "annotations": []}], + } + ] + response = bridge._collect_response_from_stream(_FakeStreamIter(streamed_items)) + assert len(response.output) == 1 + assert response.output[0]["id"] == "msg_1" + + +def test_collect_response_from_stream_keeps_nonempty_output_untouched(): + from litellm.completion_extras.litellm_responses_transformation.handler import ( + ResponsesToCompletionBridgeHandler, + ) + + bridge = ResponsesToCompletionBridgeHandler() + + class _NonEmptyCompleted: + response = { + "id": "resp_nonempty", + "object": "response", + "created_at": 0, + "status": "completed", + "model": "gpt-5.4", + "output": [{"type": "message", "id": "msg_0"}], + } + + class _StreamNoItems: + completed_response = _NonEmptyCompleted() + _hidden_params = None + + def __iter__(self): + return iter([]) + + response = bridge._collect_response_from_stream(_StreamNoItems()) + assert len(response.output) == 1 + assert response.output[0]["id"] == "msg_0" + + +def test_collect_response_from_stream_without_streamed_items_keeps_empty_output(): + from litellm.completion_extras.litellm_responses_transformation.handler import ( + ResponsesToCompletionBridgeHandler, + ) + + bridge = ResponsesToCompletionBridgeHandler() + + class _NoStreamedItems: + completed_response = _FakeCompletedResponse() + _hidden_params = None + + def __iter__(self): + return iter([]) + + response = bridge._collect_response_from_stream(_NoStreamedItems()) + assert len(response.output) == 0 + + +def test_collect_response_from_stream_async_rebuilds_empty_output(): + from litellm.completion_extras.litellm_responses_transformation.handler import ( + ResponsesToCompletionBridgeHandler, + ) + from litellm.types.utils import ResponsesAPIResponse + import asyncio + + async def run(): + bridge = ResponsesToCompletionBridgeHandler() + streamed_items = [ + { + "type": "message", + "id": "msg_async_1", + "role": "assistant", + "content": [{"type": "output_text", "text": "Hi", "annotations": []}], + } + ] + response = await bridge._collect_response_from_stream_async(_FakeAsyncStreamIter(streamed_items)) + assert isinstance(response, ResponsesAPIResponse) + assert len(response.output) == 1 + assert response.output[0]["id"] == "msg_async_1" + + asyncio.run(run()) + + diff --git a/tests/unit/responses/test_streaming_iterator.py b/tests/unit/responses/test_streaming_iterator.py index 2f6dccb37f3..19d46785aa3 100644 --- a/tests/unit/responses/test_streaming_iterator.py +++ b/tests/unit/responses/test_streaming_iterator.py @@ -1280,3 +1280,89 @@ def test_persist_completed_response_to_cache_survives_an_unserializable_response iterator._persist_completed_response_to_cache(is_async=False) cache.add_cache.assert_not_called() + + +def test_process_chunk_collects_output_item_done_in_order(): + """output_item.done events are keyed by index and returned in order.""" + logging_obj = _logging_obj_for_collector() + iterator = _make_collector_iterator(logging_obj=logging_obj) + + for index, item in enumerate([{"id": "msg_1"}, {"id": "msg_2"}]): + iterator._process_chunk( + json.dumps({"type": "response.output_item.done", "output_index": index, "item": item}) + ) + + assert iterator.get_streamed_output_items() == [{"id": "msg_1"}, {"id": "msg_2"}] + + +def test_process_chunk_collector_fallback_index_avoids_collision(): + """A missing/invalid index appends after the max key, never overwriting an entry.""" + logging_obj = _logging_obj_for_collector() + iterator = _make_collector_iterator(logging_obj=logging_obj) + + iterator._process_chunk( + json.dumps( + {"type": "response.output_item.done", "output_index": 5, "item": {"id": "msg_5"}} + ) + ) + iterator._process_chunk( + json.dumps({"type": "response.output_item.done", "item": {"id": "msg_fallback"}}) + ) + items = iterator.get_streamed_output_items() + assert items == [{"id": "msg_5"}, {"id": "msg_fallback"}] + + +def test_process_chunk_collector_duplicate_index_appends(): + """A repeated index is never dropped: the second item is appended after the max key.""" + logging_obj = _logging_obj_for_collector() + iterator = _make_collector_iterator(logging_obj=logging_obj) + + iterator._process_chunk( + json.dumps( + {"type": "response.output_item.done", "output_index": 1, "item": {"id": "msg_a"}} + ) + ) + iterator._process_chunk( + json.dumps( + {"type": "response.output_item.done", "output_index": 1, "item": {"id": "msg_b"}} + ) + ) + assert iterator.get_streamed_output_items() == [{"id": "msg_a"}, {"id": "msg_b"}] + + +def test_get_streamed_output_items_empty_when_no_done_events(): + logging_obj = _logging_obj_for_collector() + iterator = _make_collector_iterator(logging_obj=logging_obj) + assert iterator.get_streamed_output_items() == [] + + +def _streamed_item_config() -> Mock: + """transform_streaming_response returns output_item.done events with item/index preserved.""" + mock_config = Mock() + mock_config.transform_streaming_response.side_effect = ( + lambda model, parsed_chunk, logging_obj: Mock( + type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, + item=parsed_chunk.get("item"), + output_index=parsed_chunk.get("output_index"), + ) + ) + return mock_config + + +def _make_collector_iterator(*, logging_obj: Mock) -> SyncResponsesAPIStreamingIterator: + return SyncResponsesAPIStreamingIterator( + response=Mock(headers={}), + model="gpt-4o-mini", + responses_api_provider_config=_streamed_item_config(), + logging_obj=logging_obj, + litellm_metadata={}, + custom_llm_provider="openai", + call_type="responses", + ) + + +def _logging_obj_for_collector() -> Mock: + logging_obj = Mock(spec=LiteLLMLoggingObj) + logging_obj.completion_start_time = None + logging_obj.model_call_details = {"litellm_params": {}} + return logging_obj diff --git a/tests/unit/test_responses_api_bridge_non_stream.py b/tests/unit/test_responses_api_bridge_non_stream.py index 617b2cfc031..358134fb5e0 100644 --- a/tests/unit/test_responses_api_bridge_non_stream.py +++ b/tests/unit/test_responses_api_bridge_non_stream.py @@ -66,6 +66,83 @@ def test_should_collect_response_from_stream(): 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", prompt_tokens: int = 10,