From daa7346693ce97020de32261a23c3fa1fcaa459f Mon Sep 17 00:00:00 2001 From: jourdant Date: Fri, 10 Jul 2026 13:16:04 +1000 Subject: [PATCH 1/7] fix(responses): recover empty streamed output --- litellm/responses/streaming_iterator.py | 19 ++++++++ ...t_base_responses_api_streaming_iterator.py | 44 +++++++++++++++++++ 2 files changed, 63 insertions(+) diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index 6df544dee3e5..3daa8c4ed7ac 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -27,6 +27,7 @@ from litellm.litellm_core_utils.thread_pool_executor import executor from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig from litellm.responses.utils import ResponsesAPIRequestUtils +from litellm.types.llms.base import BaseLiteLLMOpenAIResponseObject from litellm.types.llms.openai import ResponsesAPIStreamEvents from litellm.types.utils import CallTypes from litellm.utils import async_post_call_success_deployment_hook @@ -71,6 +72,7 @@ def __init__( self.finished = False self.responses_api_provider_config = responses_api_provider_config self.completed_response: Optional[Any] = None + self._streamed_output_items: Dict[int, BaseLiteLLMOpenAIResponseObject] = {} self.start_time = getattr(logging_obj, "start_time", datetime.now()) self._failure_handled = False # Track if failure handler has been called self._completed_response_cached = False @@ -220,12 +222,29 @@ def _process_chunk(self, chunk) -> Optional[Any]: # Store the completed response (also for incomplete/failed so logging still fires) _chunk_type = getattr(openai_responses_api_chunk, "type", None) + if _chunk_type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE: + output_item = getattr(openai_responses_api_chunk, "item", None) + output_index = getattr(openai_responses_api_chunk, "output_index", None) + if output_item is not None and isinstance(output_index, int): + self._streamed_output_items[output_index] = output_item + openai_types = _get_openai_response_types() if openai_responses_api_chunk and _chunk_type in ( openai_types.ResponsesAPIStreamEvents.RESPONSE_COMPLETED, openai_types.ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE, openai_types.ResponsesAPIStreamEvents.RESPONSE_FAILED, ): + response_obj = getattr(openai_responses_api_chunk, "response", None) + if ( + _chunk_type != openai_types.ResponsesAPIStreamEvents.RESPONSE_FAILED + and response_obj is not None + and not getattr(response_obj, "output", None) + and self._streamed_output_items + ): + response_obj.output = [ + item.model_dump() if hasattr(item, "model_dump") else item + for _, item in sorted(self._streamed_output_items.items()) + ] self.completed_response = openai_responses_api_chunk # Add cost to usage object if include_cost_in_streaming_usage is True if litellm.include_cost_in_streaming_usage and self.logging_obj is not None: diff --git a/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py b/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py index 37fcc602d376..f1652081a7b2 100644 --- a/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py +++ b/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py @@ -28,7 +28,9 @@ from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator from litellm.responses.utils import ResponsesAPIRequestUtils +from litellm.types.llms.base import BaseLiteLLMOpenAIResponseObject from litellm.types.llms.openai import ( + OutputItemDoneEvent, ResponseCompletedEvent, ResponseFailedEvent, ResponseIncompleteEvent, @@ -169,6 +171,48 @@ def test_process_chunk_with_response_completed_event(self): # Verify the response was updated on the event assert result.response == updated_response + def test_process_chunk_backfills_empty_completed_output_from_output_item_done(self): + mock_response = Mock() + mock_response.headers = {} + mock_logging_obj = Mock(spec=LiteLLMLoggingObj) + mock_logging_obj.model_call_details = {"litellm_params": {}} + mock_config = Mock(spec=BaseResponsesAPIConfig) + output_item = BaseLiteLLMOpenAIResponseObject( + type="message", + role="assistant", + content=[{"type": "output_text", "text": "ok"}], + ) + output_item_done = OutputItemDoneEvent( + type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, + output_index=0, + item=output_item, + ) + completed_response = Mock(spec=ResponsesAPIResponse) + completed_response.output = [] + completed_event = Mock(spec=ResponseCompletedEvent) + completed_event.type = ResponsesAPIStreamEvents.RESPONSE_COMPLETED + completed_event.response = completed_response + mock_config.transform_streaming_response.side_effect = [ + output_item_done, + completed_event, + ] + + iterator = BaseResponsesAPIStreamingIterator( + response=mock_response, + model="gpt-5.6-sol", + responses_api_provider_config=mock_config, + logging_obj=mock_logging_obj, + custom_llm_provider="chatgpt", + ) + + iterator._process_chunk(json.dumps({"type": "response.output_item.done"})) + iterator._process_chunk( + json.dumps({"type": "response.completed", "response": {"output": []}}) + ) + + assert completed_response.output == [output_item.model_dump()] + assert iterator.completed_response is completed_event + def test_process_chunk_with_delta_event_no_id_update(self): """ Test that _process_chunk correctly processes a delta event From 778b9cd233105a7ae8400127110d1db121494d63 Mon Sep 17 00:00:00 2001 From: jourdant Date: Fri, 10 Jul 2026 13:46:10 +1000 Subject: [PATCH 2/7] test(responses): harden streamed output recovery --- litellm/responses/streaming_iterator.py | 40 ++-- ...t_base_responses_api_streaming_iterator.py | 179 +++++++++++++++++- 2 files changed, 205 insertions(+), 14 deletions(-) diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index 3daa8c4ed7ac..2ce4b860b75a 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -26,8 +26,11 @@ ) from litellm.litellm_core_utils.thread_pool_executor import executor from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig +from litellm.responses.sse_output_recovery import ( + record_output_item_chunk, + record_output_text_chunk, +) from litellm.responses.utils import ResponsesAPIRequestUtils -from litellm.types.llms.base import BaseLiteLLMOpenAIResponseObject from litellm.types.llms.openai import ResponsesAPIStreamEvents from litellm.types.utils import CallTypes from litellm.utils import async_post_call_success_deployment_hook @@ -72,7 +75,8 @@ def __init__( self.finished = False self.responses_api_provider_config = responses_api_provider_config self.completed_response: Optional[Any] = None - self._streamed_output_items: Dict[int, BaseLiteLLMOpenAIResponseObject] = {} + self._streamed_output_items: dict[int, dict[str, Any]] = {} + self._streamed_text_only_items: dict[int, dict[str, Any]] = {} self.start_time = getattr(logging_obj, "start_time", datetime.now()) self._failure_handled = False # Track if failure handler has been called self._completed_response_cached = False @@ -223,10 +227,13 @@ def _process_chunk(self, chunk) -> Optional[Any]: # Store the completed response (also for incomplete/failed so logging still fires) _chunk_type = getattr(openai_responses_api_chunk, "type", None) if _chunk_type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE: - output_item = getattr(openai_responses_api_chunk, "item", None) - output_index = getattr(openai_responses_api_chunk, "output_index", None) - if output_item is not None and isinstance(output_index, int): - self._streamed_output_items[output_index] = output_item + record_output_item_chunk(parsed_chunk, self._streamed_output_items) + elif _chunk_type == ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE: + record_output_text_chunk( + parsed_chunk, + self._streamed_output_items, + self._streamed_text_only_items, + ) openai_types = _get_openai_response_types() if openai_responses_api_chunk and _chunk_type in ( @@ -235,16 +242,25 @@ def _process_chunk(self, chunk) -> Optional[Any]: openai_types.ResponsesAPIStreamEvents.RESPONSE_FAILED, ): response_obj = getattr(openai_responses_api_chunk, "response", None) + response_output = ( + response_obj.get("output") + if isinstance(response_obj, dict) + else getattr(response_obj, "output", None) + ) + recovered_output = [ + item + for _, item in sorted({**self._streamed_text_only_items, **self._streamed_output_items}.items()) + ] if ( _chunk_type != openai_types.ResponsesAPIStreamEvents.RESPONSE_FAILED and response_obj is not None - and not getattr(response_obj, "output", None) - and self._streamed_output_items + and not response_output + and recovered_output ): - response_obj.output = [ - item.model_dump() if hasattr(item, "model_dump") else item - for _, item in sorted(self._streamed_output_items.items()) - ] + if isinstance(response_obj, dict): + response_obj["output"] = recovered_output + else: + response_obj.output = recovered_output self.completed_response = openai_responses_api_chunk # Add cost to usage object if include_cost_in_streaming_usage is True if litellm.include_cost_in_streaming_usage and self.logging_obj is not None: diff --git a/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py b/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py index f1652081a7b2..df5141a7ea64 100644 --- a/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py +++ b/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py @@ -205,14 +205,189 @@ def test_process_chunk_backfills_empty_completed_output_from_output_item_done(se custom_llm_provider="chatgpt", ) - iterator._process_chunk(json.dumps({"type": "response.output_item.done"})) iterator._process_chunk( - json.dumps({"type": "response.completed", "response": {"output": []}}) + json.dumps( + { + "type": "response.output_item.done", + "output_index": 0, + "item": output_item.model_dump(), + } + ) ) + iterator._process_chunk(json.dumps({"type": "response.completed", "response": {"output": []}})) assert completed_response.output == [output_item.model_dump()] assert iterator.completed_response is completed_event + @pytest.mark.parametrize( + ("terminal_type", "response_type"), + [ + (ResponsesAPIStreamEvents.RESPONSE_COMPLETED, ResponseCompletedEvent), + (ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE, ResponseIncompleteEvent), + ], + ) + def test_process_chunk_recovers_text_done_for_empty_terminal_output(self, terminal_type, response_type): + mock_response = Mock() + mock_response.headers = {} + mock_logging_obj = Mock(spec=LiteLLMLoggingObj) + mock_logging_obj.model_call_details = {"litellm_params": {}} + mock_config = Mock(spec=BaseResponsesAPIConfig) + text_done = Mock() + text_done.type = ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE + terminal_response = Mock(spec=ResponsesAPIResponse) + terminal_response.output = [] + terminal_event = Mock(spec=response_type) + terminal_event.type = terminal_type + terminal_event.response = terminal_response + mock_config.transform_streaming_response.side_effect = [text_done, terminal_event] + + iterator = BaseResponsesAPIStreamingIterator( + response=mock_response, + model="gpt-5.6-sol", + responses_api_provider_config=mock_config, + logging_obj=mock_logging_obj, + custom_llm_provider="chatgpt", + ) + + iterator._process_chunk( + json.dumps( + { + "type": "response.output_text.done", + "output_index": "1", + "content_index": 0, + "item_id": "msg_text", + "text": "ok", + } + ) + ) + iterator._process_chunk(json.dumps({"type": terminal_type, "response": {"output": []}})) + + assert terminal_response.output == [ + { + "type": "message", + "id": "msg_text", + "role": "assistant", + "status": "completed", + "content": [{"type": "output_text", "text": "ok", "annotations": []}], + } + ] + + def test_process_chunk_preserves_authoritative_terminal_output(self): + mock_response = Mock() + mock_response.headers = {} + mock_logging_obj = Mock(spec=LiteLLMLoggingObj) + mock_logging_obj.model_call_details = {"litellm_params": {}} + mock_config = Mock(spec=BaseResponsesAPIConfig) + output_item_done = OutputItemDoneEvent( + type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, + output_index=0, + item=BaseLiteLLMOpenAIResponseObject(type="message", content=[]), + ) + terminal_response = Mock(spec=ResponsesAPIResponse) + terminal_response.output = [{"type": "message", "content": [{"type": "output_text", "text": "final"}]}] + terminal_event = Mock(spec=ResponseCompletedEvent) + terminal_event.type = ResponsesAPIStreamEvents.RESPONSE_COMPLETED + terminal_event.response = terminal_response + mock_config.transform_streaming_response.side_effect = [output_item_done, terminal_event] + + iterator = BaseResponsesAPIStreamingIterator( + response=mock_response, + model="gpt-5.6-sol", + responses_api_provider_config=mock_config, + logging_obj=mock_logging_obj, + custom_llm_provider="chatgpt", + ) + + iterator._process_chunk( + json.dumps( + { + "type": "response.output_item.done", + "output_index": 0, + "item": {"type": "message", "content": [{"type": "output_text", "text": "streamed"}]}, + } + ) + ) + iterator._process_chunk( + json.dumps({"type": "response.completed", "response": {"output": terminal_response.output}}) + ) + + assert terminal_response.output[0]["content"][0]["text"] == "final" + + def test_process_chunk_backfills_dictionary_terminal_response_in_index_order(self): + mock_response = Mock() + mock_response.headers = {} + mock_logging_obj = Mock(spec=LiteLLMLoggingObj) + mock_logging_obj.model_call_details = {"litellm_params": {}} + mock_config = Mock(spec=BaseResponsesAPIConfig) + terminal_response = {"output": []} + terminal_event = Mock(spec=ResponseCompletedEvent) + terminal_event.type = ResponsesAPIStreamEvents.RESPONSE_COMPLETED + terminal_event.response = terminal_response + item_events = [] + for index in (2, 0): + event = Mock(spec=OutputItemDoneEvent) + event.type = ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE + item_events.append(event) + mock_config.transform_streaming_response.side_effect = [*item_events, terminal_event] + + iterator = BaseResponsesAPIStreamingIterator( + response=mock_response, + model="gpt-5.6-sol", + responses_api_provider_config=mock_config, + logging_obj=mock_logging_obj, + custom_llm_provider="chatgpt", + ) + + for index in (2, 0): + iterator._process_chunk( + json.dumps( + { + "type": "response.output_item.done", + "output_index": index, + "item": {"type": "message", "id": f"msg_{index}", "content": []}, + } + ) + ) + iterator._process_chunk(json.dumps({"type": "response.completed", "response": {"output": []}})) + + assert [item["id"] for item in terminal_response["output"]] == ["msg_0", "msg_2"] + + def test_process_chunk_does_not_backfill_failed_response(self): + mock_response = Mock() + mock_response.headers = {} + mock_logging_obj = Mock(spec=LiteLLMLoggingObj) + mock_logging_obj.model_call_details = {"litellm_params": {}} + mock_config = Mock(spec=BaseResponsesAPIConfig) + output_item_done = Mock(spec=OutputItemDoneEvent) + output_item_done.type = ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE + failed_response = Mock(spec=ResponsesAPIResponse) + failed_response.output = [] + failed_event = Mock(spec=ResponseFailedEvent) + failed_event.type = ResponsesAPIStreamEvents.RESPONSE_FAILED + failed_event.response = failed_response + mock_config.transform_streaming_response.side_effect = [output_item_done, failed_event] + + iterator = BaseResponsesAPIStreamingIterator( + response=mock_response, + model="gpt-5.6-sol", + responses_api_provider_config=mock_config, + logging_obj=mock_logging_obj, + custom_llm_provider="chatgpt", + ) + + iterator._process_chunk( + json.dumps( + { + "type": "response.output_item.done", + "output_index": 0, + "item": {"type": "message", "content": [{"type": "output_text", "text": "partial"}]}, + } + ) + ) + iterator._process_chunk(json.dumps({"type": "response.failed", "response": {"output": []}})) + + assert failed_response.output == [] + def test_process_chunk_with_delta_event_no_id_update(self): """ Test that _process_chunk correctly processes a delta event From d2b98593d755bd88de3581451a1b1c218a70472c Mon Sep 17 00:00:00 2001 From: jourdant Date: Fri, 10 Jul 2026 13:56:22 +1000 Subject: [PATCH 3/7] fix(responses): validate recovered stream items --- litellm/responses/streaming_iterator.py | 14 +- ...t_base_responses_api_streaming_iterator.py | 124 +++++++++++++----- 2 files changed, 105 insertions(+), 33 deletions(-) diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index 2ce4b860b75a..62b6057a5d68 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -31,6 +31,7 @@ record_output_text_chunk, ) from litellm.responses.utils import ResponsesAPIRequestUtils +from litellm.types.llms.base import BaseLiteLLMOpenAIResponseObject from litellm.types.llms.openai import ResponsesAPIStreamEvents from litellm.types.utils import CallTypes from litellm.utils import async_post_call_success_deployment_hook @@ -227,7 +228,16 @@ def _process_chunk(self, chunk) -> Optional[Any]: # Store the completed response (also for incomplete/failed so logging still fires) _chunk_type = getattr(openai_responses_api_chunk, "type", None) if _chunk_type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE: - record_output_item_chunk(parsed_chunk, self._streamed_output_items) + output_item = getattr(openai_responses_api_chunk, "item", None) + output_index = getattr(openai_responses_api_chunk, "output_index", None) + if isinstance(output_index, int) and output_index >= 0: + if isinstance(output_item, BaseLiteLLMOpenAIResponseObject): + output_item = output_item.model_dump() + if isinstance(output_item, dict): + record_output_item_chunk( + {"item": output_item, "output_index": output_index}, + self._streamed_output_items, + ) elif _chunk_type == ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE: record_output_text_chunk( parsed_chunk, @@ -252,7 +262,7 @@ def _process_chunk(self, chunk) -> Optional[Any]: for _, item in sorted({**self._streamed_text_only_items, **self._streamed_output_items}.items()) ] if ( - _chunk_type != openai_types.ResponsesAPIStreamEvents.RESPONSE_FAILED + _chunk_type == openai_types.ResponsesAPIStreamEvents.RESPONSE_COMPLETED and response_obj is not None and not response_output and recovered_output diff --git a/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py b/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py index df5141a7ea64..21f5486cb4fe 100644 --- a/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py +++ b/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py @@ -2,12 +2,12 @@ Unit tests for BaseResponsesAPIStreamingIterator Tests core functionality including: -1. Processing chunks and handling ResponseCompletedEvent +1. Processing chunks and handling ResponseCompletedEvent 2. Ensuring _update_responses_api_response_id_with_model_id is called for final chunk 3. Verifying ID update is NOT called for non-final chunks (delta events) 4. Edge case handling for invalid JSON, empty chunks, and [DONE] markers -These tests ensure the streaming iterator correctly processes response chunks +These tests ensure the streaming iterator correctly processes response chunks and applies model ID updates only to completed responses, as required for proper response tracking and logging. """ @@ -219,14 +219,7 @@ def test_process_chunk_backfills_empty_completed_output_from_output_item_done(se assert completed_response.output == [output_item.model_dump()] assert iterator.completed_response is completed_event - @pytest.mark.parametrize( - ("terminal_type", "response_type"), - [ - (ResponsesAPIStreamEvents.RESPONSE_COMPLETED, ResponseCompletedEvent), - (ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE, ResponseIncompleteEvent), - ], - ) - def test_process_chunk_recovers_text_done_for_empty_terminal_output(self, terminal_type, response_type): + def test_process_chunk_recovers_text_done_for_empty_completed_output(self): mock_response = Mock() mock_response.headers = {} mock_logging_obj = Mock(spec=LiteLLMLoggingObj) @@ -236,8 +229,8 @@ def test_process_chunk_recovers_text_done_for_empty_terminal_output(self, termin text_done.type = ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE terminal_response = Mock(spec=ResponsesAPIResponse) terminal_response.output = [] - terminal_event = Mock(spec=response_type) - terminal_event.type = terminal_type + terminal_event = Mock(spec=ResponseCompletedEvent) + terminal_event.type = ResponsesAPIStreamEvents.RESPONSE_COMPLETED terminal_event.response = terminal_response mock_config.transform_streaming_response.side_effect = [text_done, terminal_event] @@ -260,7 +253,9 @@ def test_process_chunk_recovers_text_done_for_empty_terminal_output(self, termin } ) ) - iterator._process_chunk(json.dumps({"type": terminal_type, "response": {"output": []}})) + iterator._process_chunk( + json.dumps({"type": ResponsesAPIStreamEvents.RESPONSE_COMPLETED, "response": {"output": []}}) + ) assert terminal_response.output == [ { @@ -325,9 +320,13 @@ def test_process_chunk_backfills_dictionary_terminal_response_in_index_order(sel terminal_event.response = terminal_response item_events = [] for index in (2, 0): - event = Mock(spec=OutputItemDoneEvent) - event.type = ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE - item_events.append(event) + item_events.append( + OutputItemDoneEvent( + type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, + output_index=index, + item=BaseLiteLLMOpenAIResponseObject(type="message", id=f"msg_{index}", content=[]), + ) + ) mock_config.transform_streaming_response.side_effect = [*item_events, terminal_event] iterator = BaseResponsesAPIStreamingIterator( @@ -352,6 +351,80 @@ def test_process_chunk_backfills_dictionary_terminal_response_in_index_order(sel assert [item["id"] for item in terminal_response["output"]] == ["msg_0", "msg_2"] + def test_process_chunk_ignores_malformed_output_index(self): + mock_response = Mock() + mock_response.headers = {} + mock_logging_obj = Mock(spec=LiteLLMLoggingObj) + mock_logging_obj.model_call_details = {"litellm_params": {}} + mock_config = Mock(spec=BaseResponsesAPIConfig) + valid_event = OutputItemDoneEvent( + type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, + output_index=1, + item=BaseLiteLLMOpenAIResponseObject(type="message", id="valid", content=[]), + ) + malformed_event = OutputItemDoneEvent.model_construct( + type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, + output_index="invalid", + item=BaseLiteLLMOpenAIResponseObject(type="message", id="invalid", content=[]), + ) + terminal_response = Mock(spec=ResponsesAPIResponse) + terminal_response.output = [] + terminal_event = Mock(spec=ResponseCompletedEvent) + terminal_event.type = ResponsesAPIStreamEvents.RESPONSE_COMPLETED + terminal_event.response = terminal_response + mock_config.transform_streaming_response.side_effect = [valid_event, malformed_event, terminal_event] + + iterator = BaseResponsesAPIStreamingIterator( + response=mock_response, + model="gpt-5.6-sol", + responses_api_provider_config=mock_config, + logging_obj=mock_logging_obj, + custom_llm_provider="chatgpt", + ) + + iterator._process_chunk(json.dumps({"type": "response.output_item.done"})) + iterator._process_chunk(json.dumps({"type": "response.output_item.done"})) + iterator._process_chunk(json.dumps({"type": "response.completed", "response": {"output": []}})) + + assert [item["id"] for item in terminal_response.output] == ["valid"] + + def test_process_chunk_does_not_backfill_incomplete_response(self): + mock_response = Mock() + mock_response.headers = {} + mock_logging_obj = Mock(spec=LiteLLMLoggingObj) + mock_logging_obj.model_call_details = {"litellm_params": {}} + mock_config = Mock(spec=BaseResponsesAPIConfig) + text_done = Mock() + text_done.type = ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE + incomplete_response = Mock(spec=ResponsesAPIResponse) + incomplete_response.output = [] + incomplete_event = Mock(spec=ResponseIncompleteEvent) + incomplete_event.type = ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE + incomplete_event.response = incomplete_response + mock_config.transform_streaming_response.side_effect = [text_done, incomplete_event] + + iterator = BaseResponsesAPIStreamingIterator( + response=mock_response, + model="gpt-5.6-sol", + responses_api_provider_config=mock_config, + logging_obj=mock_logging_obj, + custom_llm_provider="chatgpt", + ) + + iterator._process_chunk( + json.dumps( + { + "type": "response.output_text.done", + "output_index": 0, + "content_index": 0, + "text": "partial", + } + ) + ) + iterator._process_chunk(json.dumps({"type": "response.incomplete", "response": {"output": []}})) + + assert incomplete_response.output == [] + def test_process_chunk_does_not_backfill_failed_response(self): mock_response = Mock() mock_response.headers = {} @@ -405,11 +478,7 @@ def test_process_chunk_with_delta_event_no_id_update(self): mock_delta_event.type = ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA mock_delta_event.delta = "Hello" # Delta events don't have a response attribute - ( - delattr(mock_delta_event, "response") - if hasattr(mock_delta_event, "response") - else None - ) + (delattr(mock_delta_event, "response") if hasattr(mock_delta_event, "response") else None) # Set up the mock transform method to return our delta event mock_config.transform_streaming_response.return_value = mock_delta_event @@ -588,9 +657,7 @@ def test_handle_logging_completed_response_with_unpickleable_objects(self): iterator._handle_logging_completed_response() except TypeError as e: if "pickle" in str(e): - pytest.fail( - f"_handle_logging_completed_response failed with pickle error: {e}" - ) + pytest.fail(f"_handle_logging_completed_response failed with pickle error: {e}") raise @pytest.mark.asyncio @@ -770,9 +837,7 @@ def test_process_chunk_response_failed_calls_failure_handler(self): "_update_responses_api_response_id_with_model_id", return_value=mock_responses_api_response, ), - patch( - "litellm.responses.streaming_iterator.run_async_function" - ) as mock_run_async, + patch("litellm.responses.streaming_iterator.run_async_function") as mock_run_async, patch("litellm.responses.streaming_iterator.executor") as mock_executor, ): result = iterator._process_chunk(json.dumps(test_chunk_data)) @@ -784,10 +849,7 @@ def test_process_chunk_response_failed_calls_failure_handler(self): # Failure handler should have been called via _handle_failure mock_run_async.assert_called_once() call_kwargs = mock_run_async.call_args - assert ( - call_kwargs[1]["async_function"] - == mock_logging_obj.async_failure_handler - ) + assert call_kwargs[1]["async_function"] == mock_logging_obj.async_failure_handler mock_executor.submit.assert_called_once() submit_args = mock_executor.submit.call_args From ff911350739e689a898169de225d169132670171 Mon Sep 17 00:00:00 2001 From: jourdant Date: Fri, 10 Jul 2026 14:10:05 +1000 Subject: [PATCH 4/7] fix(responses): normalize recovered stream events --- litellm/responses/streaming_iterator.py | 28 +++++++++++++++---- ...t_base_responses_api_streaming_iterator.py | 14 +++++++--- 2 files changed, 32 insertions(+), 10 deletions(-) diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index 62b6057a5d68..e1391d8c4556 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -230,7 +230,7 @@ def _process_chunk(self, chunk) -> Optional[Any]: if _chunk_type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE: output_item = getattr(openai_responses_api_chunk, "item", None) output_index = getattr(openai_responses_api_chunk, "output_index", None) - if isinstance(output_index, int) and output_index >= 0: + if type(output_index) is int and output_index >= 0: if isinstance(output_item, BaseLiteLLMOpenAIResponseObject): output_item = output_item.model_dump() if isinstance(output_item, dict): @@ -239,11 +239,27 @@ def _process_chunk(self, chunk) -> Optional[Any]: self._streamed_output_items, ) elif _chunk_type == ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE: - record_output_text_chunk( - parsed_chunk, - self._streamed_output_items, - self._streamed_text_only_items, - ) + output_index = getattr(openai_responses_api_chunk, "output_index", None) + content_index = getattr(openai_responses_api_chunk, "content_index", None) + output_text = getattr(openai_responses_api_chunk, "text", None) + if ( + type(output_index) is int + and output_index >= 0 + and type(content_index) is int + and content_index >= 0 + and isinstance(output_text, str) + ): + record_output_text_chunk( + { + "output_index": output_index, + "content_index": content_index, + "item_id": getattr(openai_responses_api_chunk, "item_id", None), + "text": output_text, + "annotations": getattr(openai_responses_api_chunk, "annotations", None), + }, + self._streamed_output_items, + self._streamed_text_only_items, + ) openai_types = _get_openai_response_types() if openai_responses_api_chunk and _chunk_type in ( diff --git a/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py b/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py index 21f5486cb4fe..331d5b85599d 100644 --- a/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py +++ b/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py @@ -31,6 +31,7 @@ from litellm.types.llms.base import BaseLiteLLMOpenAIResponseObject from litellm.types.llms.openai import ( OutputItemDoneEvent, + OutputTextDoneEvent, ResponseCompletedEvent, ResponseFailedEvent, ResponseIncompleteEvent, @@ -225,8 +226,13 @@ def test_process_chunk_recovers_text_done_for_empty_completed_output(self): mock_logging_obj = Mock(spec=LiteLLMLoggingObj) mock_logging_obj.model_call_details = {"litellm_params": {}} mock_config = Mock(spec=BaseResponsesAPIConfig) - text_done = Mock() - text_done.type = ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE + text_done = OutputTextDoneEvent( + type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE, + output_index=1, + content_index=0, + item_id="msg_transformed", + text="ok", + ) terminal_response = Mock(spec=ResponsesAPIResponse) terminal_response.output = [] terminal_event = Mock(spec=ResponseCompletedEvent) @@ -246,7 +252,7 @@ def test_process_chunk_recovers_text_done_for_empty_completed_output(self): json.dumps( { "type": "response.output_text.done", - "output_index": "1", + "output_index": 1, "content_index": 0, "item_id": "msg_text", "text": "ok", @@ -260,7 +266,7 @@ def test_process_chunk_recovers_text_done_for_empty_completed_output(self): assert terminal_response.output == [ { "type": "message", - "id": "msg_text", + "id": "msg_transformed", "role": "assistant", "status": "completed", "content": [{"type": "output_text", "text": "ok", "annotations": []}], From cfb0e23945871e3f0fc9ae96e925ce0a52d19a03 Mon Sep 17 00:00:00 2001 From: jourdant Date: Fri, 10 Jul 2026 14:15:25 +1000 Subject: [PATCH 5/7] fix(responses): reject coerced stream indexes --- litellm/responses/streaming_iterator.py | 15 ++++++++++----- .../test_base_responses_api_streaming_iterator.py | 2 +- 2 files changed, 11 insertions(+), 6 deletions(-) diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index e1391d8c4556..f1d7b87b25f6 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -230,7 +230,8 @@ def _process_chunk(self, chunk) -> Optional[Any]: if _chunk_type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE: output_item = getattr(openai_responses_api_chunk, "item", None) output_index = getattr(openai_responses_api_chunk, "output_index", None) - if type(output_index) is int and output_index >= 0: + raw_output_index = parsed_chunk.get("output_index") + if type(raw_output_index) is int and raw_output_index >= 0 and output_index == raw_output_index: if isinstance(output_item, BaseLiteLLMOpenAIResponseObject): output_item = output_item.model_dump() if isinstance(output_item, dict): @@ -242,11 +243,15 @@ def _process_chunk(self, chunk) -> Optional[Any]: output_index = getattr(openai_responses_api_chunk, "output_index", None) content_index = getattr(openai_responses_api_chunk, "content_index", None) output_text = getattr(openai_responses_api_chunk, "text", None) + raw_output_index = parsed_chunk.get("output_index") + raw_content_index = parsed_chunk.get("content_index") if ( - type(output_index) is int - and output_index >= 0 - and type(content_index) is int - and content_index >= 0 + type(raw_output_index) is int + and raw_output_index >= 0 + and output_index == raw_output_index + and type(raw_content_index) is int + and raw_content_index >= 0 + and content_index == raw_content_index and isinstance(output_text, str) ): record_output_text_chunk( diff --git a/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py b/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py index 331d5b85599d..6d313e369fa5 100644 --- a/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py +++ b/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py @@ -388,7 +388,7 @@ def test_process_chunk_ignores_malformed_output_index(self): custom_llm_provider="chatgpt", ) - iterator._process_chunk(json.dumps({"type": "response.output_item.done"})) + iterator._process_chunk(json.dumps({"type": "response.output_item.done", "output_index": 1})) iterator._process_chunk(json.dumps({"type": "response.output_item.done"})) iterator._process_chunk(json.dumps({"type": "response.completed", "response": {"output": []}})) From a5407cf01410b0684d58b2fa681be59639f0fc88 Mon Sep 17 00:00:00 2001 From: jourdant Date: Fri, 10 Jul 2026 15:30:44 +1000 Subject: [PATCH 6/7] test(responses): exercise terminal non-backfill --- litellm/responses/streaming_iterator.py | 8 ++++---- ...t_base_responses_api_streaming_iterator.py | 19 +++++++++++++++---- 2 files changed, 19 insertions(+), 8 deletions(-) diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index f1d7b87b25f6..f85eb1c9d9f3 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -295,13 +295,13 @@ def _process_chunk(self, chunk) -> Optional[Any]: self.completed_response = openai_responses_api_chunk # Add cost to usage object if include_cost_in_streaming_usage is True if litellm.include_cost_in_streaming_usage and self.logging_obj is not None: - response_obj: Optional[Any] = getattr(openai_responses_api_chunk, "response", None) - if response_obj: - usage_obj: Optional[Any] = getattr(response_obj, "usage", None) + cost_response_obj: Optional[Any] = getattr(openai_responses_api_chunk, "response", None) + if cost_response_obj: + usage_obj: Optional[Any] = getattr(cost_response_obj, "usage", None) if usage_obj is not None: try: cost: Optional[float] = self.logging_obj._response_cost_calculator( - result=response_obj + result=cost_response_obj ) if cost is not None: setattr(usage_obj, "cost", cost) diff --git a/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py b/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py index 6d313e369fa5..a02319f59923 100644 --- a/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py +++ b/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py @@ -400,8 +400,13 @@ def test_process_chunk_does_not_backfill_incomplete_response(self): mock_logging_obj = Mock(spec=LiteLLMLoggingObj) mock_logging_obj.model_call_details = {"litellm_params": {}} mock_config = Mock(spec=BaseResponsesAPIConfig) - text_done = Mock() - text_done.type = ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE + text_done = OutputTextDoneEvent( + type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE, + output_index=0, + content_index=0, + item_id="msg_partial", + text="partial", + ) incomplete_response = Mock(spec=ResponsesAPIResponse) incomplete_response.output = [] incomplete_event = Mock(spec=ResponseIncompleteEvent) @@ -437,8 +442,14 @@ def test_process_chunk_does_not_backfill_failed_response(self): mock_logging_obj = Mock(spec=LiteLLMLoggingObj) mock_logging_obj.model_call_details = {"litellm_params": {}} mock_config = Mock(spec=BaseResponsesAPIConfig) - output_item_done = Mock(spec=OutputItemDoneEvent) - output_item_done.type = ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE + output_item_done = OutputItemDoneEvent( + type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, + output_index=0, + item=BaseLiteLLMOpenAIResponseObject( + type="message", + content=[{"type": "output_text", "text": "partial"}], + ), + ) failed_response = Mock(spec=ResponsesAPIResponse) failed_response.output = [] failed_event = Mock(spec=ResponseFailedEvent) From b6851dde2c581beecc74b6641b96c4fdd44fe853 Mon Sep 17 00:00:00 2001 From: jourdant Date: Fri, 10 Jul 2026 15:31:18 +1000 Subject: [PATCH 7/7] revert(responses): keep cost tracking unchanged --- litellm/responses/streaming_iterator.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index f85eb1c9d9f3..f1d7b87b25f6 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -295,13 +295,13 @@ def _process_chunk(self, chunk) -> Optional[Any]: self.completed_response = openai_responses_api_chunk # Add cost to usage object if include_cost_in_streaming_usage is True if litellm.include_cost_in_streaming_usage and self.logging_obj is not None: - cost_response_obj: Optional[Any] = getattr(openai_responses_api_chunk, "response", None) - if cost_response_obj: - usage_obj: Optional[Any] = getattr(cost_response_obj, "usage", None) + response_obj: Optional[Any] = getattr(openai_responses_api_chunk, "response", None) + if response_obj: + usage_obj: Optional[Any] = getattr(response_obj, "usage", None) if usage_obj is not None: try: cost: Optional[float] = self.logging_obj._response_cost_calculator( - result=cost_response_obj + result=response_obj ) if cost is not None: setattr(usage_obj, "cost", cost)