Skip to content

fix(responses): recover empty completed output from stream events #32724

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jourdant wants to merge 7 commits into
base: litellm_oss_staging
Choose a base branch
from
+381 −17
Open

fix(responses): recover empty completed output from stream events #32724

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
daa7346
fix(responses): recover empty streamed output
jourdant Jul 10, 2026
778b9cd
test(responses): harden streamed output recovery
jourdant Jul 10, 2026
d2b9859
fix(responses): validate recovered stream items
jourdant Jul 10, 2026
ff91135
fix(responses): normalize recovered stream events
jourdant Jul 10, 2026
cfb0e23
fix(responses): reject coerced stream indexes
jourdant Jul 10, 2026
a5407cf
test(responses): exercise terminal non-backfill
jourdant Jul 10, 2026
b6851dd
revert(responses): keep cost tracking unchanged
jourdant Jul 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 66 additions & 0 deletions litellm/responses/streaming_iterator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,12 @@
)
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
Expand Down Expand Up @@ -71,6 +76,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, 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
Expand Down Expand Up @@ -220,12 +227,71 @@ 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)
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):
record_output_item_chunk(
{"item": output_item, "output_index": output_index},
self._streamed_output_items,
)
elif _chunk_type == ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE:
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(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(
{
"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 (
openai_types.ResponsesAPIStreamEvents.RESPONSE_COMPLETED,
openai_types.ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE,
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_COMPLETED
and response_obj is not None
and not response_output
and recovered_output
):
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:
Expand Down
Loading
Loading