scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-09 22:31:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 86

Merge 755a8b1e26 into 77bfeb5708

Browse source
This commit is contained in:
Colin Rognlie 2026-09-08 20:00:07 +00:00 • committed by GitHub
commit df9905b27e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 582 additions and 29 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
litellm/responses/mcp/mcp_streaming_iterator.py
View file

@ -276,6 +276,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] = (

1
litellm/responses/sse_output_recovery.py
View file

@ -14,6 +14,7 @@ from typing import Final, SupportsInt, TypeAlias, cast # noqa: TID251 # int()
from litellm.constants import STREAM_SSE_DONE_STRING
_MAX_CONTENT_INDEX: Final = 1024
MAX_CONTENT_INDEX: Final = _MAX_CONTENT_INDEX
_ConvertibleToInt: TypeAlias = SupportsInt | str

112
litellm/responses/streaming_iterator.py
View file

@ -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 (
EMPTY_MAPPING,
LITELLM_MAX_STREAMING_DURATION_SECONDS,
@ -31,7 +32,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,
@ -244,6 +247,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
@ -410,6 +419,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
_stamp_responses_usage_cost(getattr(openai_responses_api_chunk, "response", None), self.logging_obj)
@ -673,6 +708,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:
@ -890,6 +992,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
@ -972,6 +1075,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
@ -1048,6 +1152,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):
@ -1115,6 +1223,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):
@ -1449,8 +1559,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.

2
litellm/router.py
View file

@ -3069,6 +3069,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

36
tests/test_litellm/llms/chatgpt/responses/test_chatgpt_responses_transformation.py
View file

@ -51,14 +51,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")
@ -121,9 +117,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},
@ -162,9 +156,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",
@ -187,9 +179,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(
@ -207,9 +197,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",
@ -232,9 +220,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(
@ -274,9 +260,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(
@ -309,9 +293,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:

458
tests/test_litellm/responses/test_streaming_iterator_output_recovery.py Normal file
View file

@ -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]"