From 755a8b1e26224ae13cbf244bfe1b4b6077b98b61 Mon Sep 17 00:00:00 2001 From: Colin Rognlie Date: Wed, 19 Aug 2026 07:30:57 +0000 Subject: [PATCH] fix(streaming): backfill response.completed output from output_item.done SSE events When streaming ChatGPT subscription responses, the terminal response.completed event carries an empty output: [] even though the model produced text. The streaming iterator now accumulates output_item.done and output_text.done events as they arrive and backfills them into the completed response object at the terminal event, so logging, spend tracking, and post-stream hooks see the real content instead of an empty list. --- .../responses/mcp/mcp_streaming_iterator.py | 2 + litellm/responses/sse_output_recovery.py | 1 + litellm/responses/streaming_iterator.py | 112 ++++- litellm/router.py | 2 + .../test_chatgpt_responses_transformation.py | 36 +- ...test_streaming_iterator_output_recovery.py | 458 ++++++++++++++++++ 6 files changed, 582 insertions(+), 29 deletions(-) create mode 100644 tests/test_litellm/responses/test_streaming_iterator_output_recovery.py diff --git a/litellm/responses/mcp/mcp_streaming_iterator.py b/litellm/responses/mcp/mcp_streaming_iterator.py index c7471518398..6579569ff92 100644 --- a/litellm/responses/mcp/mcp_streaming_iterator.py +++ b/litellm/responses/mcp/mcp_streaming_iterator.py @@ -274,6 +274,8 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): # Streaming state management self.phase = "initial_response" # initial_response -> mcp_discovery -> (continue_initial_response <-> tool_execution) -> finished self.finished = False + self._streamed_output_items: dict[int, BaseLiteLLMOpenAIResponseObject] = {} # mutable-ok: SSE accumulator + self._streamed_text_only_items: dict[int, BaseLiteLLMOpenAIResponseObject] = {} # mutable-ok: SSE fallback # Event queues and generation flags self.mcp_discovery_events: list[ResponsesAPIStreamingResponse] = ( diff --git a/litellm/responses/sse_output_recovery.py b/litellm/responses/sse_output_recovery.py index 208dec10c62..41218b27c29 100644 --- a/litellm/responses/sse_output_recovery.py +++ b/litellm/responses/sse_output_recovery.py @@ -13,6 +13,7 @@ from typing import Any, Final from litellm.constants import STREAM_SSE_DONE_STRING _MAX_CONTENT_INDEX: Final = 1024 +MAX_CONTENT_INDEX: Final = _MAX_CONTENT_INDEX def parse_sse_json_chunk(chunk: str) -> dict[str, Any] | None: diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index e678fba2852..44a5f13c574 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -16,6 +16,7 @@ from openai._streaming import SSEDecoder from typing_extensions import TypeIs import litellm +from litellm._logging import verbose_logger from litellm.constants import ( LITELLM_MAX_STREAMING_DURATION_SECONDS, STREAM_SSE_DONE_STRING, @@ -30,7 +31,9 @@ from litellm.litellm_core_utils.llm_response_utils.response_metadata import ( ) 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 MAX_CONTENT_INDEX from litellm.responses.utils import ResponseAPILoggingUtils, ResponsesAPIRequestUtils +from litellm.types.llms.base import BaseLiteLLMOpenAIResponseObject from litellm.types.llms.openai import ( PART_UNION_TYPES, ResponseAPIUsage, @@ -175,6 +178,12 @@ class BaseResponsesAPIStreamingIterator: self.finished = False self.responses_api_provider_config = responses_api_provider_config self.completed_response: ResponsesAPIStreamingResponse | None = None + self._streamed_output_items: dict[ # mutable-ok: SSE accumulator + int, BaseLiteLLMOpenAIResponseObject + ] = {} # mutable-ok: initialized empty; filled incrementally per SSE event + self._streamed_text_only_items: dict[ # mutable-ok: SSE fallback accumulator + int, BaseLiteLLMOpenAIResponseObject + ] = {} # mutable-ok: initialized empty; filled incrementally per SSE event self.start_time = getattr(logging_obj, "start_time", datetime.now()) self._failure_handled = False # Track if failure handler has been called self._yielded_first_chunk = False @@ -338,6 +347,32 @@ class BaseResponsesAPIStreamingIterator: openai_types.ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE, openai_types.ResponsesAPIStreamEvents.RESPONSE_FAILED, ): + if _chunk_type in ( + openai_types.ResponsesAPIStreamEvents.RESPONSE_COMPLETED, + openai_types.ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE, + ): + _response_obj: Final = getattr(openai_responses_api_chunk, "response", None) + if ( + _response_obj is not None + and not getattr(_response_obj, "output", None) + and (self._streamed_output_items or self._streamed_text_only_items) + ): + try: + _merged_items: Final = { # mutable-ok: transient merge for backfill sort; not retained + **self._streamed_text_only_items, + **self._streamed_output_items, + } + _backfill: Final = [ # mutable-ok: assigned to response obj output field which expects list + item.model_dump() if hasattr(item, "model_dump") else item + for _, item in sorted(_merged_items.items()) + ] + _response_obj.output = _backfill # mutable-ok: patching response obj from provider before it's stored; no immutable path here + except Exception: # noqa: BLE001 # best-effort backfill; any failure must not crash the stream + verbose_logger.warning( + "streaming_iterator: failed to backfill %s output", + _chunk_type, + exc_info=True, + ) 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: @@ -575,6 +610,73 @@ class BaseResponsesAPIStreamingIterator: self._completed_response_cached = True + def _accumulate_streamed_output_item(self, chunk: ResponsesAPIStreamingResponse) -> None: + """ + Accumulate OUTPUT_ITEM_DONE / OUTPUT_TEXT_DONE payloads from a post-hook chunk + so they can backfill response.completed.output when the provider sends it empty. + Called after async_post_call_streaming_deployment_hook so only the final, + hook-transformed item is retained (not the raw pre-hook version). + """ + _chunk_type: Final = getattr(chunk, "type", None) + if _chunk_type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE: + _item: Final = getattr(chunk, "item", None) + _output_index: Final = getattr( + chunk, + "output_index", + max(self._streamed_output_items, default=-1) + 1, + ) + if _item is not None and isinstance(_output_index, int): + self._streamed_output_items[_output_index] = ( + _item # mutable-ok: incremental index-keyed accumulation across SSE events; no immutable equivalent + ) + + elif _chunk_type == ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE: + _text: Final = getattr(chunk, "text", None) + _text_output_index: Final = getattr(chunk, "output_index", None) + if ( + isinstance(_text, str) + and isinstance(_text_output_index, int) + and _text_output_index not in self._streamed_output_items + ): + _content_index: Final = getattr(chunk, "content_index", 0) or 0 + if 0 <= _content_index <= MAX_CONTENT_INDEX: + _item_id: Final = getattr(chunk, "item_id", None) or f"msg_{_text_output_index}" + _existing: Final = self._streamed_text_only_items.get(_text_output_index) + _existing_content: Final = list( # mutable-ok: copy existing content for slot replacement + getattr(_existing, "content", None) or [] # mutable-ok: empty fallback for missing content + ) + _annotations: Final = getattr(chunk, "annotations", None) + _slot: Final = { # mutable-ok: content dict matches provider schema + "type": "output_text", + "text": _text, + "annotations": _annotations or [], # mutable-ok: empty fallback for missing annotations + } + _content: Final = ( # mutable-ok: list concat building content array; computed once + _existing_content[:_content_index] + + [_slot] # mutable-ok: list concat for slot replacement + + _existing_content[_content_index + 1 :] + if _content_index < len(_existing_content) + else _existing_content + + [ # mutable-ok: list concat for gap padding + { # mutable-ok: placeholder content dict for gap padding + "type": "output_text", + "text": "", + "annotations": [], # mutable-ok: empty annotations placeholder + } + for _ in range(_content_index - len(_existing_content)) + ] + + [_slot] # mutable-ok: list concat appending final slot + ) + self._streamed_text_only_items[_text_output_index] = ( + BaseLiteLLMOpenAIResponseObject( # mutable-ok: incremental index-keyed fallback accumulation; no immutable equivalent + type="message", + id=getattr(_existing, "id", _item_id), + role="assistant", + status="completed", + content=_content, + ) + ) + async def _call_post_streaming_deployment_hook( self, chunk: ResponsesAPIStreamingResponse ) -> ResponsesAPIStreamingResponse: @@ -787,6 +889,7 @@ class ResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): chunk=result, ) self._yielded_first_chunk = True + self._accumulate_streamed_output_item(result) return result # If result is None, continue the loop to get the next chunk @@ -869,6 +972,7 @@ class SyncResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): chunk=result, ) self._yielded_first_chunk = True + self._accumulate_streamed_output_item(result) return result # If result is None, continue the loop to get the next chunk @@ -945,6 +1049,10 @@ class MockResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): chunk_size=self.CHUNK_SIZE, ) self._idx = 0 + # completed_response is set directly here and in __anext__/__next__ because + # these iterators replay pre-built events from _build_synthetic_response_events, + # which always populates output. They bypass _process_chunk intentionally, so the + # output backfill logic there does not apply. self.completed_response = self._events[-1] def __aiter__(self): @@ -1012,6 +1120,8 @@ class CachedResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): chunk_size=MockResponsesAPIStreamingIterator.CHUNK_SIZE, ) self._idx = 0 + # See MockResponsesAPIStreamingIterator._set_events_from_response for why + # completed_response is set directly rather than via _process_chunk. self.completed_response = self._events[-1] def __aiter__(self): @@ -1334,8 +1444,6 @@ def _build_synthetic_response_events( # WebSocket mode streaming (bidirectional forwarding) # --------------------------------------------------------------------------- -from litellm._logging import verbose_logger - # Conservative per-frame output-token floor used when a response.create # frame omits max_output_tokens, so a project OTPM quota can't be bypassed # by simply never declaring an output cap. diff --git a/litellm/router.py b/litellm/router.py index efd3b5a527e..3c1ec8add8a 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -2631,6 +2631,8 @@ class Router: self.finished = False self.responses_api_provider_config = getattr(source_iterator, "responses_api_provider_config", None) self.completed_response = None + self._streamed_output_items: dict = {} # mutable-ok: mirrors base class attr; keyed accumulator reset per stream + self._streamed_text_only_items: dict = {} # mutable-ok: mirrors base class attr; keyed fallback accumulator reset per stream self.start_time = getattr(source_iterator, "start_time", datetime.now()) self._failure_handled = False self._yielded_first_chunk = False diff --git a/tests/test_litellm/llms/chatgpt/responses/test_chatgpt_responses_transformation.py b/tests/test_litellm/llms/chatgpt/responses/test_chatgpt_responses_transformation.py index 90a1c24bada..422d4a16c3c 100644 --- a/tests/test_litellm/llms/chatgpt/responses/test_chatgpt_responses_transformation.py +++ b/tests/test_litellm/llms/chatgpt/responses/test_chatgpt_responses_transformation.py @@ -54,14 +54,10 @@ class TestChatGPTResponsesAPITransformation: url = config.get_complete_url(api_base=None, litellm_params={}) assert url == "https://chatgpt.example.com/responses" - custom_url = config.get_complete_url( - api_base="https://custom.chatgpt.com", litellm_params={} - ) + custom_url = config.get_complete_url(api_base="https://custom.chatgpt.com", litellm_params={}) assert custom_url == "https://custom.chatgpt.com/responses" - url_with_slash = config.get_complete_url( - api_base="https://chatgpt.example.com/", litellm_params={} - ) + url_with_slash = config.get_complete_url(api_base="https://chatgpt.example.com/", litellm_params={}) assert url_with_slash == "https://chatgpt.example.com/responses" @patch("litellm.llms.chatgpt.responses.transformation.Authenticator") @@ -124,9 +120,7 @@ class TestChatGPTResponsesAPITransformation: "user": "user_123", "temperature": 0.2, "top_p": 0.9, - "context_management": [ - {"type": "compaction", "compact_threshold": 200000} - ], + "context_management": [{"type": "compaction", "compact_threshold": 200000}], "metadata": {"foo": "bar"}, "max_output_tokens": 123, "stream_options": {"include_usage": True}, @@ -165,9 +159,7 @@ class TestChatGPTResponsesAPITransformation: ("chatgpt/gpt-5.3-codex", "gpt-5.3-codex"), ], ) - def test_chatgpt_non_stream_sse_response_parsing( - self, model_name: str, response_model: str - ): + def test_chatgpt_non_stream_sse_response_parsing(self, model_name: str, response_model: str): config = ChatGPTResponsesAPIConfig() response_payload = { "id": "resp_test", @@ -190,9 +182,7 @@ class TestChatGPTResponsesAPITransformation: "", ] ) - raw_response = httpx.Response( - 200, headers={"content-type": "text/event-stream"}, text=sse_body - ) + raw_response = httpx.Response(200, headers={"content-type": "text/event-stream"}, text=sse_body) logging_obj = MagicMock() parsed = config.transform_response_api_response( @@ -210,9 +200,7 @@ class TestChatGPTResponsesAPITransformation: ("chatgpt/gpt-5.3-codex", "gpt-5.3-codex"), ], ) - def test_chatgpt_non_stream_sse_response_recovers_output_items( - self, model_name: str, response_model: str - ): + def test_chatgpt_non_stream_sse_response_recovers_output_items(self, model_name: str, response_model: str): config = ChatGPTResponsesAPIConfig() response_payload = { "id": "resp_test", @@ -235,9 +223,7 @@ class TestChatGPTResponsesAPITransformation: "", ] ) - raw_response = httpx.Response( - 200, headers={"content-type": "text/event-stream"}, text=sse_body - ) + raw_response = httpx.Response(200, headers={"content-type": "text/event-stream"}, text=sse_body) logging_obj = MagicMock() parsed = config.transform_response_api_response( @@ -277,9 +263,7 @@ class TestChatGPTResponsesAPITransformation: "", ] ) - raw_response = httpx.Response( - 200, headers={"content-type": "text/event-stream"}, text=sse_body - ) + raw_response = httpx.Response(200, headers={"content-type": "text/event-stream"}, text=sse_body) logging_obj = MagicMock() parsed = config.transform_response_api_response( @@ -312,9 +296,7 @@ class TestChatGPTResponsesAPITransformation: "", ] ) - raw_response = httpx.Response( - 502, headers={"content-type": "text/event-stream"}, text=sse_body - ) + raw_response = httpx.Response(502, headers={"content-type": "text/event-stream"}, text=sse_body) logging_obj = MagicMock() with pytest.raises(OpenAIError) as exc_info: diff --git a/tests/test_litellm/responses/test_streaming_iterator_output_recovery.py b/tests/test_litellm/responses/test_streaming_iterator_output_recovery.py new file mode 100644 index 00000000000..9f1451b066b --- /dev/null +++ b/tests/test_litellm/responses/test_streaming_iterator_output_recovery.py @@ -0,0 +1,458 @@ +""" +Regression for #25429. + +chatgpt.com's Codex backend sends response.completed with an empty output +array. The actual assistant content arrives via preceding +response.output_item.done events. Without accumulation in the streaming +iterator, completed_response.response.output is [] and the +chat-completions bridge raises "Unknown items in responses API response: []". + +These tests verify that BaseResponsesAPIStreamingIterator accumulates +output_item.done payloads (via _accumulate_streamed_output_item, called +after post-call hooks in __anext__/__next__) and backfills them into the +response.completed chunk before storing it as completed_response. +""" + +import json +import os +import sys +from unittest.mock import MagicMock, patch + +import httpx + +sys.path.insert(0, os.path.abspath("../../..")) + + +def _make_iterator(): + from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj + from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig + from litellm.responses import streaming_iterator as _si_mod + from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator + + logging_obj = MagicMock(spec=LiteLLMLoggingObj) + logging_obj.model_call_details = {"litellm_params": {}} + logging_obj.start_time = None + logging_obj.completion_start_time = None + + response = httpx.Response(200, headers={"content-type": "text/event-stream"}, text="") + + # Patch get_api_base to avoid triggering chatgpt device-auth during construction. + with patch.object(_si_mod, "get_api_base", return_value=None): + return BaseResponsesAPIStreamingIterator( + response=response, + model="chatgpt/gpt-5.4", + responses_api_provider_config=OpenAIResponsesAPIConfig(), + logging_obj=logging_obj, + custom_llm_provider="chatgpt", + request_data={}, + ) + + +_OUTPUT_ITEM = { + "type": "message", + "id": "msg_0", + "role": "assistant", + "content": [{"type": "output_text", "text": "OK my lord", "annotations": []}], + "status": "completed", +} + +_RESPONSE_BASE = { + "id": "resp_test", + "object": "response", + "created_at": 1700000000, + "status": "completed", + "model": "gpt-5.4", +} + + +def _process_and_accumulate(iterator, raw_json: str) -> None: + """ + Simulate what __anext__ does: _process_chunk then _accumulate_streamed_output_item. + Accumulation now happens on the post-hook chunk; in tests there are no hooks, + so the chunk is unchanged and we call both methods in sequence. + """ + result = iterator._process_chunk(raw_json) + if result is not None: + iterator._accumulate_streamed_output_item(result) + + +def _text_from_output_item(item) -> str: + content = item["content"] if isinstance(item, dict) else item.content + part = content[0] + return part["text"] if isinstance(part, dict) else part.text + + +class TestStreamingIteratorOutputRecovery: + def test_completed_response_output_stays_empty_without_preceding_items(self): + """ + Baseline: response.completed with output:[] and no preceding + output_item.done leaves completed_response.response.output empty. + Confirms the backfill only activates when items were actually streamed. + """ + iterator = _make_iterator() + + with patch.object(iterator, "_handle_logging_completed_response"): + iterator._process_chunk( + json.dumps({"type": "response.created", "response": {**_RESPONSE_BASE, "output": []}}) + ) + iterator._process_chunk( + json.dumps({"type": "response.completed", "response": {**_RESPONSE_BASE, "output": []}}) + ) + + assert iterator.completed_response is not None + output = getattr(iterator.completed_response.response, "output", None) + assert output == [] or output is None + + def test_completed_response_output_backfilled_from_output_item_done(self): + """ + Core regression: when response.completed.output is [] but + response.output_item.done events preceded it, the iterator + backfills output so completed_response.response.output is non-empty. + + Before the fix this assertion would fail because output stayed []. + """ + iterator = _make_iterator() + + with patch.object(iterator, "_handle_logging_completed_response"): + _process_and_accumulate( + iterator, json.dumps({"type": "response.output_item.done", "output_index": 0, "item": _OUTPUT_ITEM}) + ) + iterator._process_chunk( + json.dumps({"type": "response.completed", "response": {**_RESPONSE_BASE, "output": []}}) + ) + + assert iterator.completed_response is not None + output = iterator.completed_response.response.output + assert len(output) == 1 + assert _text_from_output_item(output[0]) == "OK my lord" + assert isinstance(output[0], dict), f"expected dict for transformation compat, got {type(output[0])}" + assert output[0].get("type") == "message" + + def test_authoritative_output_is_not_overwritten(self): + """ + When the provider sends a non-empty output in response.completed, + the backfill must not overwrite it. + """ + authoritative_item = { + **_OUTPUT_ITEM, + "id": "msg_auth", + "content": [{"type": "output_text", "text": "authoritative", "annotations": []}], + } + iterator = _make_iterator() + + with patch.object(iterator, "_handle_logging_completed_response"): + _process_and_accumulate( + iterator, json.dumps({"type": "response.output_item.done", "output_index": 0, "item": _OUTPUT_ITEM}) + ) + iterator._process_chunk( + json.dumps( + { + "type": "response.completed", + "response": {**_RESPONSE_BASE, "output": [authoritative_item]}, + } + ) + ) + + output = iterator.completed_response.response.output + assert len(output) == 1 + assert _text_from_output_item(output[0]) == "authoritative" + + def test_multiple_output_items_ordered_by_index(self): + """ + Multiple output_item.done events are ordered by output_index, + not by the order they arrived. + """ + item_a = { + **_OUTPUT_ITEM, + "id": "msg_a", + "content": [{"type": "output_text", "text": "A", "annotations": []}], + } + item_b = { + **_OUTPUT_ITEM, + "id": "msg_b", + "content": [{"type": "output_text", "text": "B", "annotations": []}], + } + iterator = _make_iterator() + + with patch.object(iterator, "_handle_logging_completed_response"): + _process_and_accumulate( + iterator, json.dumps({"type": "response.output_item.done", "output_index": 1, "item": item_b}) + ) + _process_and_accumulate( + iterator, json.dumps({"type": "response.output_item.done", "output_index": 0, "item": item_a}) + ) + iterator._process_chunk( + json.dumps({"type": "response.completed", "response": {**_RESPONSE_BASE, "output": []}}) + ) + + output = iterator.completed_response.response.output + assert len(output) == 2 + assert _text_from_output_item(output[0]) == "A" + assert _text_from_output_item(output[1]) == "B" + + def test_output_text_done_backfilled_when_no_output_item_done(self): + """ + Fallback for providers that emit OUTPUT_TEXT_DONE without OUTPUT_ITEM_DONE: + the text-only item is used to backfill output when response.completed.output + is empty and no OUTPUT_ITEM_DONE events were received. + """ + iterator = _make_iterator() + + with patch.object(iterator, "_handle_logging_completed_response"): + _process_and_accumulate( + iterator, + json.dumps( + { + "type": "response.output_text.done", + "output_index": 0, + "content_index": 0, + "item_id": "msg_text_only", + "text": "text only content", + } + ), + ) + iterator._process_chunk( + json.dumps({"type": "response.completed", "response": {**_RESPONSE_BASE, "output": []}}) + ) + + assert iterator.completed_response is not None + output = iterator.completed_response.response.output + assert len(output) == 1 + assert _text_from_output_item(output[0]) == "text only content" + + def test_output_item_done_takes_precedence_over_output_text_done(self): + """ + When both OUTPUT_TEXT_DONE and OUTPUT_ITEM_DONE arrive for the same + output_index, the real OUTPUT_ITEM_DONE item must win. + """ + iterator = _make_iterator() + + with patch.object(iterator, "_handle_logging_completed_response"): + _process_and_accumulate( + iterator, + json.dumps( + { + "type": "response.output_text.done", + "output_index": 0, + "content_index": 0, + "item_id": "msg_text_only", + "text": "text only content", + } + ), + ) + _process_and_accumulate( + iterator, json.dumps({"type": "response.output_item.done", "output_index": 0, "item": _OUTPUT_ITEM}) + ) + iterator._process_chunk( + json.dumps({"type": "response.completed", "response": {**_RESPONSE_BASE, "output": []}}) + ) + + output = iterator.completed_response.response.output + assert len(output) == 1 + assert _text_from_output_item(output[0]) == "OK my lord" + + def test_incomplete_response_output_backfilled_from_output_item_done(self): + """ + response.incomplete (max-tokens truncation) with output:[] must be + backfilled the same way as response.completed — partial content that + arrived via output_item.done must not be silently dropped. + """ + iterator = _make_iterator() + + with patch.object(iterator, "_handle_logging_completed_response"): + _process_and_accumulate( + iterator, json.dumps({"type": "response.output_item.done", "output_index": 0, "item": _OUTPUT_ITEM}) + ) + iterator._process_chunk( + json.dumps( + { + "type": "response.incomplete", + "response": {**_RESPONSE_BASE, "status": "incomplete", "output": []}, + } + ) + ) + + assert iterator.completed_response is not None + output = iterator.completed_response.response.output + assert len(output) == 1 + assert _text_from_output_item(output[0]) == "OK my lord" + + def test_response_failed_sets_completed_response_without_backfill(self): + """ + response.failed must set completed_response so logging still fires, + but must NOT backfill output (no content to recover from a failed response). + """ + iterator = _make_iterator() + + with patch.object(iterator, "_handle_logging_completed_response"): + _process_and_accumulate( + iterator, json.dumps({"type": "response.output_item.done", "output_index": 0, "item": _OUTPUT_ITEM}) + ) + iterator._process_chunk( + json.dumps( + { + "type": "response.failed", + "response": {**_RESPONSE_BASE, "status": "failed", "output": []}, + } + ) + ) + + assert iterator.completed_response is not None + output = getattr(iterator.completed_response.response, "output", None) + assert output == [] or output is None + + def test_output_text_done_replace_in_place(self): + """ + When a second OUTPUT_TEXT_DONE arrives for the same output_index and + content_index as an existing slot, it must replace that slot in-place + rather than appending. + """ + iterator = _make_iterator() + + with patch.object(iterator, "_handle_logging_completed_response"): + _process_and_accumulate( + iterator, + json.dumps( + { + "type": "response.output_text.done", + "output_index": 0, + "content_index": 0, + "item_id": "msg_replace", + "text": "first", + } + ), + ) + _process_and_accumulate( + iterator, + json.dumps( + { + "type": "response.output_text.done", + "output_index": 0, + "content_index": 0, + "item_id": "msg_replace", + "text": "replaced", + } + ), + ) + iterator._process_chunk( + json.dumps({"type": "response.completed", "response": {**_RESPONSE_BASE, "output": []}}) + ) + + output = iterator.completed_response.response.output + assert len(output) == 1 + content = output[0]["content"] + assert len(content) == 1 + assert content[0]["text"] == "replaced" + + def test_output_text_done_gap_padding(self): + """ + When OUTPUT_TEXT_DONE arrives with content_index=2 and no prior slots, + the iterator must insert two empty placeholder slots before it. + """ + iterator = _make_iterator() + + with patch.object(iterator, "_handle_logging_completed_response"): + _process_and_accumulate( + iterator, + json.dumps( + { + "type": "response.output_text.done", + "output_index": 0, + "content_index": 2, + "item_id": "msg_gap", + "text": "late content", + } + ), + ) + iterator._process_chunk( + json.dumps({"type": "response.completed", "response": {**_RESPONSE_BASE, "output": []}}) + ) + + output = iterator.completed_response.response.output + assert len(output) == 1 + content = output[0]["content"] + assert len(content) == 3 + assert content[0]["text"] == "" + assert content[1]["text"] == "" + assert content[2]["text"] == "late content" + + def test_output_index_absent_uses_sequential_fallback(self): + """ + When an OUTPUT_ITEM_DONE chunk lacks the output_index field entirely, + the iterator assigns a synthetic index via max(existing)+1 so items + are not silently dropped. + """ + item_a = {**_OUTPUT_ITEM, "id": "msg_a", "content": [{"type": "output_text", "text": "A", "annotations": []}]} + item_b = {**_OUTPUT_ITEM, "id": "msg_b", "content": [{"type": "output_text", "text": "B", "annotations": []}]} + iterator = _make_iterator() + + with patch.object(iterator, "_handle_logging_completed_response"): + _process_and_accumulate( + iterator, json.dumps({"type": "response.output_item.done", "output_index": 0, "item": item_a}) + ) + _process_and_accumulate(iterator, json.dumps({"type": "response.output_item.done", "item": item_b})) + iterator._process_chunk( + json.dumps({"type": "response.completed", "response": {**_RESPONSE_BASE, "output": []}}) + ) + + output = iterator.completed_response.response.output + assert len(output) == 2 + assert _text_from_output_item(output[0]) == "A" + assert _text_from_output_item(output[1]) == "B" + + def test_backfill_exception_is_swallowed(self): + """ + If model_dump() raises during backfill, the exception must be swallowed + and logged as a warning rather than crashing the stream. completed_response + is still set so logging fires. + """ + iterator = _make_iterator() + + with patch.object(iterator, "_handle_logging_completed_response"): + _process_and_accumulate( + iterator, json.dumps({"type": "response.output_item.done", "output_index": 0, "item": _OUTPUT_ITEM}) + ) + with patch.object( + iterator._streamed_output_items[0], # pyright: ignore[reportPrivateUsage] # test-only: verify accumulator state to simulate serialization failure + "model_dump", + side_effect=RuntimeError("serialization failure"), + ): + iterator._process_chunk( + json.dumps({"type": "response.completed", "response": {**_RESPONSE_BASE, "output": []}}) + ) + + assert iterator.completed_response is not None + + def test_post_hook_item_is_accumulated_not_pre_hook(self): + """ + Accumulation must happen after async_post_call_streaming_deployment_hook runs, + not before. If a hook replaces the item on the chunk, the replaced (post-hook) + item must be what ends up in completed_response.response.output. + """ + from litellm.types.llms.base import BaseLiteLLMOpenAIResponseObject + + redacted_item = BaseLiteLLMOpenAIResponseObject( + type="message", + id="msg_redacted", + role="assistant", + status="completed", + content=[{"type": "output_text", "text": "[REDACTED]", "annotations": []}], + ) + + iterator = _make_iterator() + + with patch.object(iterator, "_handle_logging_completed_response"): + raw = json.dumps({"type": "response.output_item.done", "output_index": 0, "item": _OUTPUT_ITEM}) + pre_hook_chunk = iterator._process_chunk(raw) + assert pre_hook_chunk is not None + # Simulate a hook that replaced the item on the chunk. + pre_hook_chunk.item = redacted_item + iterator._accumulate_streamed_output_item(pre_hook_chunk) + + iterator._process_chunk( + json.dumps({"type": "response.completed", "response": {**_RESPONSE_BASE, "output": []}}) + ) + + output = iterator.completed_response.response.output + assert len(output) == 1 + assert output[0]["content"][0]["text"] == "[REDACTED]"