scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-03 07:23:33 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 166

fix(responses): serialize sequence_number on every bridged streaming event (#44944)

Browse source 
* fix(responses): serialize sequence_number on every bridged streaming event

* test(responses): cover sequence_number on bridged streaming events

* fix(responses): read undeclared fields through dict accessors and number guardrail block continuations

* fix(responses): scope dict reads that see undeclared fields to Responses stream events

Moves the extras-aware __getitem__, get, __contains__ and items from the
shared response base onto a new base for the 37 stream event classes, so
usage, Gemini, Interactions and moderation objects keep reading __dict__
and Gemini responses keep _hidden_params out of .get

* fix(router): renumber Responses stream events only after a fallback takes over

A stream that never falls back now keeps its provider's own sequence numbers, repeats included, so a starting_after resume still matches them. After a fallback the wrapper continues above the highest number the client saw

* fix(guardrails): keep Responses stream numbers unique when LLM Shield flushes held text

The restorer's flush copied the last delta and kept its sequence_number, so a
native stream repeated a number and a bridged stream sent the flush unnumbered.
A flush now takes the done event's slot and later events move up by one, and a
stream the provider never closed numbers its flush after the last event.

Also pins the created_at in the streaming iterator test fixture instead of
reading the clock.

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
Kaustubh Anil Jain 2026-11-02 22:31:58 +05:30 • committed by GitHub
parent 1046716b6d
commit 6f3251c7e3
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 511 additions and 144 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
litellm/llms/openai/responses/guardrail_translation/handler.py
View file

@ -1988,7 +1988,7 @@ class OpenAIResponsesHandler(BaseTranslation):
"annotations": (),
"logprobs": None,
}
return (
events: Final = (
*_open_item_closing_events(responses_so_far),
OutputItemAddedEvent(
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED,
@ -2033,6 +2033,10 @@ class OpenAIResponsesHandler(BaseTranslation):
response=_blocked_response(exc, response_id=response_id, model=model, output_item=item),
),
)
return tuple(
event.model_copy(update={"sequence_number": number})
for number, event in enumerate(events, start=_next_stream_sequence_number(responses_so_far))
)
class _BlockedContentPart(TypedDict):

30
litellm/proxy/guardrails/guardrail_hooks/llm_shield_proxy/stream_restorers.py
View file

@ -261,7 +261,8 @@ class ResponsesStreamRestorer:
own window, keyed by the family, the item id and the part index.
- Any `*.done` event closes the stream of the same family. Whatever its window still
holds goes out first, as a copy of that stream's last delta event -- so it carries
the stream's own ids, and repeats that event's `sequence_number`. Then every text
the stream's own ids. It takes the done event's `sequence_number`, and every later
event's number moves up by one, so the numbers stay unique and in order. Then every text
field on the done event is restored in full: its string fields other than
identifiers, plus any `part` or `item` it repeats.
- `response.completed` / `response.incomplete` repeat the whole reply, and are
@ -273,6 +274,8 @@ class ResponsesStreamRestorer:
self._rehydrate: Final = rehydrate
self._carries: Final[dict[ResponsesStreamKey, str]] = {} # mutable-ok: per-stream windows advanced in place.
self._last_deltas: Final[dict[ResponsesStreamKey, object]] = {} # mutable-ok: newest delta per stream.
self._inserted: int = 0
self._last_sequence_number: int | None = None
async def restore(self, event: object) -> tuple[object, ...]:
"""The events to emit in place of `event`: any flush, then the event itself."""
@ -281,7 +284,7 @@ class ResponsesStreamRestorer:
return (event,)
if kind.endswith(".delta") and kind not in RESPONSES_BINARY_DELTAS:
await self._restore_delta(event, kind)
return (event,)
return self._numbered((), event)
slots: Final[SlotSink] = []
flushed: Final = await self._flush(responses_stream_key(event, kind)) if kind.endswith(".done") else ()
if kind.endswith(".done"):
@ -294,12 +297,29 @@ class ResponsesStreamRestorer:
for item in read_list(read_field(event, "response"), "output"):
collect_response_item(item, slots)
await rehydrate_slots(slots, self._rehydrate)
return (*flushed, event)
return self._numbered(flushed, event)
async def finish(self) -> tuple[object, ...]:
"""Flushes every stream the provider never closed, e.g. a truncated reply."""
flushed: Final = tuple([await self._flush(key) for key in tuple(self._carries)])
return tuple(itertools.chain.from_iterable(flushed))
flushed: Final = tuple(itertools.chain.from_iterable([await self._flush(key) for key in tuple(self._carries)]))
last: Final = self._last_sequence_number
if last is not None:
for offset, flush in enumerate(flushed, start=1):
write_field(flush, "sequence_number", last + offset)
return flushed
def _numbered(self, flushed: tuple[object, ...], event: object) -> tuple[object, ...]:
"""Gives each flush the slot `event` arrived in and moves `event` and later events up."""
number: Final = read_field(event, "sequence_number")
if not isinstance(number, int):
return (*flushed, event)
for offset, flush in enumerate(flushed):
write_field(flush, "sequence_number", number + self._inserted + offset)
self._inserted += len(flushed)
self._last_sequence_number = number + self._inserted
if self._inserted:
write_field(event, "sequence_number", self._last_sequence_number)
return (*flushed, event)
async def _restore_delta(self, event: object, kind: str) -> None:
text: Final = read_field(event, "delta")

125
litellm/responses/litellm_completion_transformation/streaming_iterator.py
View file

@ -27,7 +27,11 @@ from litellm.responses.litellm_completion_transformation.tool_search import (
from litellm.responses.litellm_completion_transformation.transformation import (
LiteLLMCompletionResponsesConfig,
)
from litellm.responses.streaming_iterator import ResponsesAPIStreamingIterator, stamp_responses_usage_cost
from litellm.responses.streaming_iterator import (
SEQUENCE_NUMBER_STAMPED_ON_EMISSION,
ResponsesAPIStreamingIterator,
stamp_responses_usage_cost,
)
from litellm.responses.utils import ResponsesAPIRequestUtils
from litellm.types.llms.openai import (
PART_UNION_TYPES,
@ -239,14 +243,11 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
return execution
def _next_output_item_added(self, output_index: int, item: BaseLiteLLMOpenAIResponseObject) -> OutputItemAddedEvent:
self._sequence_number += 1
added: Final = OutputItemAddedEvent(
return OutputItemAddedEvent(
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED,
output_index=output_index,
item=item,
)
added.__dict__["sequence_number"] = self._sequence_number
return added
def _queue_output_item_added(self, output_index: int, item: BaseLiteLLMOpenAIResponseObject) -> None:
self._pending_tool_events.append(self._next_output_item_added(output_index, item))
@ -268,12 +269,10 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
item: Final = build_tool_search_call(
call_id, final_args or self._tool_args_by_call_id.get(call_id, ""), "completed", execution
)
self._sequence_number += 1
self._pending_tool_events.append(
OutputItemDoneEvent(
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
output_index=output_index,
sequence_number=self._sequence_number,
item=BaseLiteLLMOpenAIResponseObject(**item.model_dump()),
)
)
@ -346,12 +345,10 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
if streamed_item is None:
return
self._sent_compaction_item_done = True
self._sequence_number += 1
self._pending_response_events.append(
OutputItemDoneEvent(
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
output_index=self._compaction_output_index,
sequence_number=self._sequence_number,
item=streamed_item,
)
)
@ -494,15 +491,12 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
chunk_size = 10 # Match typical OpenAI delta size
for i in range(0, len(fn_args_delta), chunk_size):
delta_chunk = fn_args_delta[i : i + chunk_size]
self._sequence_number += 1
delta_event: BaseLiteLLMOpenAIResponseObject = FunctionCallArgumentsDeltaEvent(
type=ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA,
item_id=self._tool_item_id_by_call_id.get(call_id, call_id),
output_index=output_index,
delta=delta_chunk,
)
# Add sequence_number as extra field (BaseLiteLLMOpenAIResponseObject allows extra fields)
delta_event.__dict__["sequence_number"] = self._sequence_number
self._pending_tool_events.append(delta_event)
def _queue_final_tool_call_done_events(self, litellm_complete_object: ModelResponse) -> None:
@ -567,33 +561,27 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
chunk_size = 10 # Match typical OpenAI delta size
for i in range(0, len(remaining_args), chunk_size):
delta_chunk = remaining_args[i : i + chunk_size]
self._sequence_number += 1
delta_event = FunctionCallArgumentsDeltaEvent(
type=ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA,
item_id=self._tool_item_id_by_call_id.get(call_id, call_id),
output_index=output_index,
delta=delta_chunk,
)
delta_event.__dict__["sequence_number"] = self._sequence_number
self._pending_tool_events.append(delta_event)
self._sequence_number += 1
done_event = FunctionCallArgumentsDoneEvent(
type=ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DONE,
item_id=self._tool_item_id_by_call_id.get(call_id, call_id),
output_index=output_index,
arguments=final_args,
)
done_event.__dict__["sequence_number"] = self._sequence_number
self._pending_tool_events.append(done_event)
self._sequence_number += 1
item_kwargs = self._tool_call_item_kwargs(call_id, fn_name, final_args, "completed")
item_kwargs["id"] = self._tool_item_id_by_call_id.setdefault(call_id, item_kwargs["id"])
item_done_event = OutputItemDoneEvent(
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
output_index=output_index,
sequence_number=self._sequence_number,
item=BaseLiteLLMOpenAIResponseObject(
**item_kwargs,
**(_tool_call_provider_specific_fields(tc) if self.tool_call_provider_fields else {}),
@ -625,16 +613,11 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
_WEB_SEARCH_PROGRESS_EVENTS[:1] if item.status == "in_progress" else _WEB_SEARCH_PROGRESS_EVENTS
)
for event_type, event_class in progress_events:
self._sequence_number += 1
event = event_class(type=event_type, output_index=output_index, item_id=item.id)
event.__dict__["sequence_number"] = self._sequence_number
self._pending_tool_events.append(event)
self._sequence_number += 1
self._pending_tool_events.append(event_class(type=event_type, output_index=output_index, item_id=item.id))
self._pending_tool_events.append(
OutputItemDoneEvent(
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
output_index=output_index,
sequence_number=self._sequence_number,
item=BaseLiteLLMOpenAIResponseObject(**item.model_dump()),
)
)
@ -770,31 +753,24 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
"""
response_created_event_data: Final = self._default_response_created_event_data()
self._sequence_number += 1
event: Final = ResponseCreatedEvent(
return ResponseCreatedEvent(
type=ResponsesAPIStreamEvents.RESPONSE_CREATED,
response=self._with_encoded_response_id(ResponsesAPIResponse(**response_created_event_data)),
)
event.__dict__["sequence_number"] = self._sequence_number
return event
def create_response_in_progress_event(self) -> ResponseInProgressEvent:
response_in_progress_event_data: Final = self._default_response_created_event_data()
response_in_progress_event_data["status"] = "in_progress"
self._sequence_number += 1
event: Final = ResponseInProgressEvent(
return ResponseInProgressEvent(
type=ResponsesAPIStreamEvents.RESPONSE_IN_PROGRESS,
response=self._with_encoded_response_id(ResponsesAPIResponse(**response_in_progress_event_data)),
)
event.__dict__["sequence_number"] = self._sequence_number
return event
def create_output_item_added_event(self) -> OutputItemAddedEvent:
if self._cached_item_id is None:
self._cached_item_id = f"msg_{uuid.uuid4()}"
self._sequence_number += 1
event: Final = OutputItemAddedEvent(
return OutputItemAddedEvent(
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED,
output_index=self._message_index(),
item=BaseLiteLLMOpenAIResponseObject(
@ -807,23 +783,18 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
}
),
)
event.__dict__["sequence_number"] = self._sequence_number
return event
def create_content_part_added_event(self) -> ContentPartAddedEvent:
if self._cached_item_id is None:
self._cached_item_id = f"msg_{uuid.uuid4()}"
self._sequence_number += 1
event: Final = ContentPartAddedEvent(
return ContentPartAddedEvent(
type=ResponsesAPIStreamEvents.CONTENT_PART_ADDED,
item_id=self._cached_item_id,
output_index=self._message_index(),
content_index=0,
part=BaseLiteLLMOpenAIResponseObject(**{"type": "output_text", "text": "", "annotations": []}),
)
event.__dict__["sequence_number"] = self._sequence_number
return event
def _queue_message_item_added_events(self) -> None:
if not self.sent_message_item_added_event:
@ -834,7 +805,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
def _queue_message_item_added_event(self) -> None:
item_id: Final = self._message_item_id()
self.sent_message_item_added_event = True
self._sequence_number += 1
event: Final = OutputItemAddedEvent(
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED,
output_index=self._message_index(),
@ -848,7 +818,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
}
),
)
event.__dict__["sequence_number"] = self._sequence_number
self._pending_response_events.append(event)
def _queue_refusal_events(self, refusal: str) -> None:
@ -885,12 +854,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
part=ContentPartDonePartRefusal(type="refusal", refusal=refusal),
),
)
first_sequence_number: Final = self._sequence_number + 1
self._sequence_number += len(events)
self._pending_response_events.extend(
event.model_copy(update={"sequence_number": sequence_number})
for sequence_number, event in enumerate(events, start=first_sequence_number)
)
self._pending_response_events.extend(events)
def _merge_provider_specific_fields(self, src: dict) -> None:
"""Merge provider_specific_fields using last-value-wins for lists.
@ -944,7 +908,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
self,
reasoning_item_id: str,
reasoning_content: str,
sequence_number: int,
) -> ReasoningSummaryTextDoneEvent:
"""
Create response.reasoning_summary_text.done event.
@ -954,7 +917,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
"type": "response.reasoning_summary_text.done",
"item_id": "rs_0c5dae30e53172980069708ba2f59c8197b71ca9820edad07c",
"output_index": 0,
"sequence_number": 97,
"summary_index": 0,
"text": "**Clarifying the first humans**\n\nThe I'm addressing the user's specific interest."
}
@ -963,7 +925,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
type=ResponsesAPIStreamEvents.REASONING_SUMMARY_TEXT_DONE,
item_id=reasoning_item_id,
output_index=self._reasoning_index(),
sequence_number=sequence_number,
sequence_number=SEQUENCE_NUMBER_STAMPED_ON_EMISSION,
summary_index=0,
text=reasoning_content,
)
@ -972,7 +934,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
self,
reasoning_item_id: str,
reasoning_content: str,
sequence_number: int,
) -> ReasoningSummaryPartDoneEvent:
"""
Create response.reasoning_summary_part.done event.
@ -986,7 +947,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
"type": "summary_text",
"text": "**Clarifying the first humans**\n\nThe earlier hominins. It feels important to ensure I'm addressing the user's specific interest."
},
"sequence_number": 98,
"summary_index": 0
}
"""
@ -994,7 +954,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
type=ResponsesAPIStreamEvents.REASONING_SUMMARY_PART_DONE,
item_id=reasoning_item_id,
output_index=self._reasoning_index(),
sequence_number=sequence_number,
sequence_number=SEQUENCE_NUMBER_STAMPED_ON_EMISSION,
summary_index=0,
part=BaseLiteLLMOpenAIResponseObject(
**{
@ -1066,7 +1026,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
return OutputItemDoneEvent(
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
output_index=self._message_index(),
sequence_number=1,
item=BaseLiteLLMOpenAIResponseObject(
**{
"id": self._cached_item_id,
@ -1082,7 +1041,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
self,
reasoning_item_id: str,
reasoning_content: str,
sequence_number: int,
) -> OutputItemDoneEvent:
"""
Create response.output_item.done event for reasoning items.
@ -1091,7 +1049,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
{
"type": "response.output_item.done",
"output_index": 0,
"sequence_number": 99,
"item": {
"id": "rs_0c5dae30e53172980069708ba2f59c8197b71ca9820edad07c",
"type": "reasoning",
@ -1107,7 +1064,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
return OutputItemDoneEvent(
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
output_index=self._reasoning_index(),
sequence_number=sequence_number,
item=BaseLiteLLMOpenAIResponseObject(
**{
"id": reasoning_item_id,
@ -1223,7 +1179,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
if chunk.choices[0].finish_reason is None and not _delta_carries_output(delta):
return
self._sequence_number += 1
self.sent_output_item_added_event = True
# Reasoning-first
@ -1233,20 +1188,20 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
self._cached_reasoning_item_id = mint_reasoning_item_id()
self._reasoning_item_id = self._cached_reasoning_item_id
event = OutputItemAddedEvent(
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED,
output_index=self._reasoning_index(),
item=BaseLiteLLMOpenAIResponseObject(
**{
"id": self._cached_reasoning_item_id,
"type": "reasoning",
"status": "in_progress",
"summary": None,
}
),
self._pending_response_events.append(
OutputItemAddedEvent(
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED,
output_index=self._reasoning_index(),
item=BaseLiteLLMOpenAIResponseObject(
**{
"id": self._cached_reasoning_item_id,
"type": "reasoning",
"status": "in_progress",
"summary": None,
}
),
)
)
event.__dict__["sequence_number"] = self._sequence_number
self._pending_response_events.append(event)
return
# Tool-first
@ -1279,30 +1234,34 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
)
reasoning_item_id: Final = self._cached_reasoning_item_id
self._queue_compaction_item_done()
self._sequence_number += 1
text_done_event: Final = self.create_reasoning_summary_text_done_event(
reasoning_item_id=reasoning_item_id,
reasoning_content=reasoning_content,
sequence_number=self._sequence_number,
)
self._sequence_number += 1
part_done_event: Final = self.create_reasoning_summary_part_done_event(
reasoning_item_id=reasoning_item_id,
reasoning_content=reasoning_content,
sequence_number=self._sequence_number,
)
self._sequence_number += 1
reasoning_output_item_done_event: Final = self.create_reasoning_output_item_done_event(
reasoning_item_id=reasoning_item_id,
reasoning_content=reasoning_content,
sequence_number=self._sequence_number,
)
self._pending_response_events.extend([text_done_event, part_done_event, reasoning_output_item_done_event])
self._reasoning_done_emitted = True
self._reasoning_active = False
def _stamp_sequence_number(self, event: BaseLiteLLMOpenAIResponseObject) -> BaseLiteLLMOpenAIResponseObject:
stamped: Final = event.model_copy(update={"sequence_number": self._sequence_number})
self._sequence_number += 1
return stamped
async def __anext__(
self,
) -> ResponsesAPIStreamingResponse | ResponseCompletedEvent | BaseLiteLLMOpenAIResponseObject:
return self._stamp_sequence_number(await self._anext_unstamped_event())
async def _anext_unstamped_event(
self,
) -> ResponsesAPIStreamingResponse | ResponseCompletedEvent | BaseLiteLLMOpenAIResponseObject:
try:
while True:
@ -1363,6 +1322,11 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
def __next__(
self,
) -> ResponsesAPIStreamingResponse | ResponseCompletedEvent | BaseLiteLLMOpenAIResponseObject:
return self._stamp_sequence_number(self._next_unstamped_event())
def _next_unstamped_event(
self,
) -> ResponsesAPIStreamingResponse | ResponseCompletedEvent | BaseLiteLLMOpenAIResponseObject:
try:
while True:
@ -1504,16 +1468,13 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
if not self.sent_content_part_added_event:
self._queue_message_item_added_events()
self._hold_tool_calls_behind_text(getattr(chunk.choices[0].delta, "tool_calls", None))
self._sequence_number += 1
text_delta_event: Final = OutputTextDeltaEvent(
return OutputTextDeltaEvent(
type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA,
item_id=item_id,
output_index=self._message_index(),
content_index=0,
delta=delta_content,
)
text_delta_event.__dict__["sequence_number"] = self._sequence_number
return text_delta_event
# Priority 3: Handle tool call deltas (if any) -> queue events and emit them
# For each tool call delta, we emit events one at a time to match OpenAI's streaming behavior

12
litellm/responses/streaming_iterator.py
View file

@ -379,6 +379,7 @@ def _mid_stream_fallback_eligible(mapped_exception: Exception) -> bool:
PRE_OUTPUT_LIFECYCLE_EVENT_TYPES: Final = frozenset({"response.created", "response.in_progress", "response.queued"})
SEQUENCE_NUMBER_STAMPED_ON_EMISSION: Final = -1
class BaseResponsesAPIStreamingIterator:
@ -1933,7 +1934,6 @@ def build_synthetic_response_events(
_build_response_status_event(openai_types.ResponsesAPIStreamEvents.RESPONSE_IN_PROGRESS, transformed),
]
sequence_number = 0
output_items: Final[Sequence[object]] = getattr(transformed, "output", []) or []
for output_index, output_item in enumerate(output_items):
output_item_payload = _dump_response_object(output_item)
@ -2043,35 +2043,31 @@ def build_synthetic_response_events(
delta=summary_text[i : i + chunk_size],
)
)
sequence_number += 1
events.append(
openai_types.ReasoningSummaryTextDoneEvent(
type=openai_types.ResponsesAPIStreamEvents.REASONING_SUMMARY_TEXT_DONE,
item_id=item_id,
output_index=output_index,
sequence_number=sequence_number,
sequence_number=SEQUENCE_NUMBER_STAMPED_ON_EMISSION,
summary_index=summary_index,
text=summary_text,
)
)
sequence_number += 1
events.append(
openai_types.ReasoningSummaryPartDoneEvent(
type=openai_types.ResponsesAPIStreamEvents.REASONING_SUMMARY_PART_DONE,
item_id=item_id,
output_index=output_index,
sequence_number=sequence_number,
sequence_number=SEQUENCE_NUMBER_STAMPED_ON_EMISSION,
summary_index=summary_index,
part=openai_types.BaseLiteLLMOpenAIResponseObject(**summary_payload),
)
)
sequence_number += 1
events.append(
openai_types.OutputItemDoneEvent(
type=openai_types.ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
output_index=output_index,
sequence_number=sequence_number,
item=openai_types.BaseLiteLLMOpenAIResponseObject(**output_item_payload),
)
)
@ -2082,7 +2078,7 @@ def build_synthetic_response_events(
response=transformed,
)
)
return events
return [event.model_copy(update={"sequence_number": number}) for number, event in enumerate(events)]
# ---------------------------------------------------------------------------

19
litellm/router_python_core.py
View file

@ -544,6 +544,7 @@ if TYPE_CHECKING:
ResponseAPIUsage,
ResponseIncompleteEvent,
ResponsesAPIResponse,
ResponsesAPIStreamingResponse,
)
Span = _Span
@ -4385,12 +4386,13 @@ class PythonCore(DeprecatedRouterMembers):
fallback_headers_adopted: bool = False
def __init__(self, async_generator: AsyncGenerator):
def __init__(self, async_generator: "AsyncGenerator[ResponsesAPIStreamingResponse, None]"):
import time
from datetime import datetime
self._async_generator = async_generator
self._billing_source: object = source_iterator
self._last_sequence_number = -1
# Mirror every attribute BaseResponsesAPIStreamingIterator.__init__
# would have set. The wrapper bypasses super().__init__ (it has no
# httpx.Response of its own and no provider config to drive), so
@ -4497,7 +4499,16 @@ class PythonCore(DeprecatedRouterMembers):
if self.completed_response is None and getattr(chunk, "type", None) in _RESPONSES_TERMINAL_EVENT_TYPES:
self.completed_response = chunk
self._yielded_first_chunk = True
return chunk
sequence_number: Final = getattr(chunk, "sequence_number", None)
if not isinstance(sequence_number, int):
return chunk
if not self.fallback_headers_adopted:
self._last_sequence_number = max(self._last_sequence_number, sequence_number)
return chunk
self._last_sequence_number = max(self._last_sequence_number + 1, sequence_number)
if self._last_sequence_number == sequence_number:
return chunk
return chunk.model_copy(update={"sequence_number": self._last_sequence_number})
async def aclose(self):
# async generators always expose aclose — no defensive check needed.
@ -4559,7 +4570,9 @@ class PythonCore(DeprecatedRouterMembers):
exc,
)
wrapper: Final = FallbackResponsesStreamWrapper(stream_with_fallbacks())
wrapper: Final = FallbackResponsesStreamWrapper(
stream_with_fallbacks() # pyright: ignore[reportArgumentType] # relays response events typed as object
)
return wrapper
def _aresponses_background_stream_can_release(

100
litellm/types/llms/openai.py
View file

@ -1,8 +1,9 @@
import builtins
import math
from collections.abc import Iterable, Mapping, Sequence
from collections.abc import ItemsView, Iterable, Mapping, Sequence
from enum import Enum
from os import PathLike
from types import MappingProxyType
from typing import IO, TYPE_CHECKING, Any, Final, Generic, Literal, Optional, TypeAlias, Union
import httpx
@ -1754,17 +1755,40 @@ class ResponsesAPIStreamEvents(str, Enum):
ERROR = "error"
class ResponseCreatedEvent(BaseLiteLLMOpenAIResponseObject):
_NO_EXTRAS: Final[Mapping[str, object]] = MappingProxyType({})
class BaseResponsesStreamEvent(BaseLiteLLMOpenAIResponseObject): # frozen-ok: guardrails rewrite events in place
def __getitem__(self, key: str) -> object:
return self._mapping_holding(key)[key]
def get(self, key: str, default: object = None) -> object:
return self._mapping_holding(key).get(key, default)
def __contains__(self, key: object) -> bool:
return key in self.__dict__ or key in self._extras()
def items(self) -> ItemsView[str, object]:
return MappingProxyType({**self.__dict__, **self._extras()}).items()
def _extras(self) -> Mapping[str, object]:
return self.__pydantic_extra__ or _NO_EXTRAS
def _mapping_holding(self, key: str) -> Mapping[str, object]:
return self.__dict__ if key in self.__dict__ else self._extras()
class ResponseCreatedEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.RESPONSE_CREATED]
response: ResponsesAPIResponse
class ResponseInProgressEvent(BaseLiteLLMOpenAIResponseObject):
class ResponseInProgressEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.RESPONSE_IN_PROGRESS]
response: ResponsesAPIResponse
class ResponseCompletedEvent(BaseLiteLLMOpenAIResponseObject):
class ResponseCompletedEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.RESPONSE_COMPLETED]
response: ResponsesAPIResponse
_hidden_params: dict = PrivateAttr(default_factory=dict)
@ -1778,12 +1802,12 @@ class ResponseCompletedEvent(BaseLiteLLMOpenAIResponseObject):
self._hidden_params = hidden_params
class ResponseFailedEvent(BaseLiteLLMOpenAIResponseObject):
class ResponseFailedEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.RESPONSE_FAILED]
response: ResponsesAPIResponse
class ResponseIncompleteEvent(BaseLiteLLMOpenAIResponseObject):
class ResponseIncompleteEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE]
response: ResponsesAPIResponse
@ -1791,7 +1815,7 @@ class ResponseIncompleteEvent(BaseLiteLLMOpenAIResponseObject):
ResponsesTerminalEvent: TypeAlias = ResponseCompletedEvent | ResponseIncompleteEvent | ResponseFailedEvent
class ResponsePartAddedEvent(BaseLiteLLMOpenAIResponseObject):
class ResponsePartAddedEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.RESPONSE_PART_ADDED]
item_id: str
output_index: int
@ -1799,7 +1823,7 @@ class ResponsePartAddedEvent(BaseLiteLLMOpenAIResponseObject):
part: dict
class ReasoningSummaryTextDeltaEvent(BaseLiteLLMOpenAIResponseObject):
class ReasoningSummaryTextDeltaEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.REASONING_SUMMARY_TEXT_DELTA]
item_id: str
output_index: int
@ -1807,7 +1831,7 @@ class ReasoningSummaryTextDeltaEvent(BaseLiteLLMOpenAIResponseObject):
delta: str
class ReasoningSummaryTextDoneEvent(BaseLiteLLMOpenAIResponseObject):
class ReasoningSummaryTextDoneEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.REASONING_SUMMARY_TEXT_DONE]
item_id: str
output_index: int
@ -1816,7 +1840,7 @@ class ReasoningSummaryTextDoneEvent(BaseLiteLLMOpenAIResponseObject):
text: str
class ReasoningSummaryPartDoneEvent(BaseLiteLLMOpenAIResponseObject):
class ReasoningSummaryPartDoneEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.REASONING_SUMMARY_PART_DONE]
item_id: str
output_index: int
@ -1825,13 +1849,13 @@ class ReasoningSummaryPartDoneEvent(BaseLiteLLMOpenAIResponseObject):
part: BaseLiteLLMOpenAIResponseObject
class OutputItemAddedEvent(BaseLiteLLMOpenAIResponseObject):
class OutputItemAddedEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED]
output_index: int
item: BaseLiteLLMOpenAIResponseObject | None
class OutputItemDoneEvent(BaseLiteLLMOpenAIResponseObject):
class OutputItemDoneEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE]
output_index: int
sequence_number: int = 1
@ -1851,7 +1875,7 @@ class OpenAIChatCompletionLogprobsContent(TypedDict, total=False):
top_logprobs: list[OpenAIChatCompletionLogprobsContentTopLogprobs]
class ContentPartAddedEvent(BaseLiteLLMOpenAIResponseObject):
class ContentPartAddedEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.CONTENT_PART_ADDED]
item_id: str
output_index: int
@ -1879,7 +1903,7 @@ class ContentPartDonePartReasoningText(BaseLiteLLMOpenAIResponseObject):
PART_UNION_TYPES = ContentPartDonePartOutputText | ContentPartDonePartRefusal | ContentPartDonePartReasoningText
class ContentPartDoneEvent(BaseLiteLLMOpenAIResponseObject):
class ContentPartDoneEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.CONTENT_PART_DONE]
item_id: str
output_index: int
@ -1887,7 +1911,7 @@ class ContentPartDoneEvent(BaseLiteLLMOpenAIResponseObject):
part: PART_UNION_TYPES
class OutputTextDeltaEvent(BaseLiteLLMOpenAIResponseObject):
class OutputTextDeltaEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA]
item_id: str
output_index: int
@ -1895,7 +1919,7 @@ class OutputTextDeltaEvent(BaseLiteLLMOpenAIResponseObject):
delta: str
class OutputTextAnnotationAddedEvent(BaseLiteLLMOpenAIResponseObject):
class OutputTextAnnotationAddedEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.OUTPUT_TEXT_ANNOTATION_ADDED]
item_id: str
output_index: int
@ -1904,7 +1928,7 @@ class OutputTextAnnotationAddedEvent(BaseLiteLLMOpenAIResponseObject):
annotation: dict
class OutputTextDoneEvent(BaseLiteLLMOpenAIResponseObject):
class OutputTextDoneEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE]
item_id: str
output_index: int
@ -1912,7 +1936,7 @@ class OutputTextDoneEvent(BaseLiteLLMOpenAIResponseObject):
text: str
class RefusalDeltaEvent(BaseLiteLLMOpenAIResponseObject):
class RefusalDeltaEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.REFUSAL_DELTA]
item_id: str
output_index: int
@ -1920,7 +1944,7 @@ class RefusalDeltaEvent(BaseLiteLLMOpenAIResponseObject):
delta: str
class RefusalDoneEvent(BaseLiteLLMOpenAIResponseObject):
class RefusalDoneEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.REFUSAL_DONE]
item_id: str
output_index: int
@ -1928,72 +1952,72 @@ class RefusalDoneEvent(BaseLiteLLMOpenAIResponseObject):
refusal: str
class FunctionCallArgumentsDeltaEvent(BaseLiteLLMOpenAIResponseObject):
class FunctionCallArgumentsDeltaEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA]
item_id: str
output_index: int
delta: str
class FunctionCallArgumentsDoneEvent(BaseLiteLLMOpenAIResponseObject):
class FunctionCallArgumentsDoneEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DONE]
item_id: str
output_index: int
arguments: str
class FileSearchCallInProgressEvent(BaseLiteLLMOpenAIResponseObject):
class FileSearchCallInProgressEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.FILE_SEARCH_CALL_IN_PROGRESS]
output_index: int
item_id: str
class FileSearchCallSearchingEvent(BaseLiteLLMOpenAIResponseObject):
class FileSearchCallSearchingEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.FILE_SEARCH_CALL_SEARCHING]
output_index: int
item_id: str
class FileSearchCallCompletedEvent(BaseLiteLLMOpenAIResponseObject):
class FileSearchCallCompletedEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.FILE_SEARCH_CALL_COMPLETED]
output_index: int
item_id: str
class WebSearchCallInProgressEvent(BaseLiteLLMOpenAIResponseObject):
class WebSearchCallInProgressEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.WEB_SEARCH_CALL_IN_PROGRESS]
output_index: int
item_id: str
class WebSearchCallSearchingEvent(BaseLiteLLMOpenAIResponseObject):
class WebSearchCallSearchingEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.WEB_SEARCH_CALL_SEARCHING]
output_index: int
item_id: str
class WebSearchCallCompletedEvent(BaseLiteLLMOpenAIResponseObject):
class WebSearchCallCompletedEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.WEB_SEARCH_CALL_COMPLETED]
output_index: int
item_id: str
# MCP List Tools Events
class MCPListToolsInProgressEvent(BaseLiteLLMOpenAIResponseObject):
class MCPListToolsInProgressEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.MCP_LIST_TOOLS_IN_PROGRESS]
sequence_number: int
output_index: int
item_id: str
class MCPListToolsCompletedEvent(BaseLiteLLMOpenAIResponseObject):
class MCPListToolsCompletedEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.MCP_LIST_TOOLS_COMPLETED]
sequence_number: int
output_index: int
item_id: str
class MCPListToolsFailedEvent(BaseLiteLLMOpenAIResponseObject):
class MCPListToolsFailedEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.MCP_LIST_TOOLS_FAILED]
sequence_number: int
output_index: int
@ -2001,14 +2025,14 @@ class MCPListToolsFailedEvent(BaseLiteLLMOpenAIResponseObject):
# MCP Call Events
class MCPCallInProgressEvent(BaseLiteLLMOpenAIResponseObject):
class MCPCallInProgressEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.MCP_CALL_IN_PROGRESS]
sequence_number: int
output_index: int
item_id: str
class MCPCallArgumentsDeltaEvent(BaseLiteLLMOpenAIResponseObject):
class MCPCallArgumentsDeltaEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.MCP_CALL_ARGUMENTS_DELTA]
output_index: int
item_id: str
@ -2016,7 +2040,7 @@ class MCPCallArgumentsDeltaEvent(BaseLiteLLMOpenAIResponseObject):
sequence_number: int
class MCPCallArgumentsDoneEvent(BaseLiteLLMOpenAIResponseObject):
class MCPCallArgumentsDoneEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.MCP_CALL_ARGUMENTS_DONE]
output_index: int
item_id: str
@ -2024,21 +2048,21 @@ class MCPCallArgumentsDoneEvent(BaseLiteLLMOpenAIResponseObject):
sequence_number: int
class MCPCallCompletedEvent(BaseLiteLLMOpenAIResponseObject):
class MCPCallCompletedEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.MCP_CALL_COMPLETED]
sequence_number: int
item_id: str
output_index: int
class MCPCallFailedEvent(BaseLiteLLMOpenAIResponseObject):
class MCPCallFailedEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.MCP_CALL_FAILED]
sequence_number: int
item_id: str
output_index: int
class ImageGenerationPartialImageEvent(BaseLiteLLMOpenAIResponseObject):
class ImageGenerationPartialImageEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.IMAGE_GENERATION_PARTIAL_IMAGE]
partial_image_index: int
b64_json: str
@ -2053,7 +2077,7 @@ class ErrorEventError(BaseLiteLLMOpenAIResponseObject):
param: str | dict[str, object] | None = None
class ErrorEvent(BaseLiteLLMOpenAIResponseObject):
class ErrorEvent(BaseResponsesStreamEvent):
type: Literal[ResponsesAPIStreamEvents.ERROR]
sequence_number: int
error: ErrorEventError | None = None
@ -2062,7 +2086,7 @@ class ErrorEvent(BaseLiteLLMOpenAIResponseObject):
param: str | Mapping[str, object] | None = None
class GenericEvent(BaseLiteLLMOpenAIResponseObject):
class GenericEvent(BaseResponsesStreamEvent):
type: str
model_config = ConfigDict(extra="allow", protected_namespaces=())

28
tests/unit/llms/openai/responses/test_openai_responses_guardrail_handler.py
View file

@ -4547,6 +4547,34 @@ class TestBuildBlockSseChunks:
assert completed["output"][0]["content"][0]["text"] == "Blocked by policy."
assert completed["usage"] == {"input_tokens": 7, "output_tokens": 21, "total_tokens": 28}
def test_continuation_numbers_every_event_after_the_last_one_relayed(self):
yielded: Final = [
{"type": "response.created", "sequence_number": 0, "response": {"id": "resp_live", "output": []}},
{
"type": "response.output_item.added",
"sequence_number": 1,
"output_index": 0,
"item": {"id": "msg_orig", "type": "message", "role": "assistant"},
},
{
"type": "response.output_text.delta",
"sequence_number": 2,
"item_id": "msg_orig",
"output_index": 0,
"content_index": 0,
"delta": "partial",
},
]
payloads: Final = self._payloads(
OpenAIResponsesHandler().build_block_sse_chunks(
self._exc(original_response=yielded), stream_started=True, responses_so_far=yielded
)
)
assert payloads[-1]["type"] == "response.completed"
assert [payload.get("sequence_number") for payload in payloads] == list(range(3, 3 + len(payloads)))
def test_continuation_reads_usage_from_typed_completed_event(self):
from litellm.types.llms.openai import (
ResponseCompletedEvent,

42
tests/unit/proxy/guardrails/guardrail_hooks/test_llm_shield_proxy.py
View file

@ -1694,6 +1694,48 @@ class TestResponsesStreamRestoration:
assert "".join(e.delta for e in out) == "see a@example.com a["
@pytest.mark.asyncio
async def test_a_flushed_delta_takes_the_done_events_number_and_later_events_move_up(self):
guardrail, _ = _shielded(self.VAULT)
done = OutputTextDoneEvent(
type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE,
item_id="msg_1",
output_index=0,
content_index=0,
text="Mail [EMAIL_1] x[",
sequence_number=3,
)
out = await _restore_stream(
guardrail,
[
self._text_delta("Mail [EMA", 1),
self._text_delta("IL_1] x[", 2),
done,
self._text_delta("next", 4, content_index=1),
],
)
assert [(e.type, e.sequence_number) for e in out] == [
(ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA, 1),
(ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA, 2),
(ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA, 3),
(ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE, 4),
(ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA, 5),
]
assert out[2].delta == "["
@pytest.mark.asyncio
async def test_a_truncated_stream_flush_continues_after_the_last_number(self):
guardrail, _ = _shielded(self.VAULT)
out = await _restore_stream(
guardrail, [self._text_delta("see [EMAIL_1] a[", 7), self._text_delta(" b", 8, content_index=1)]
)
assert [e.sequence_number for e in out] == [7, 8, 9]
assert out[-1].delta == "["
@pytest.mark.asyncio
async def test_completed_response_is_restored(self):
"""The terminal event repeats the whole reply, and clients read it as the answer."""

66
tests/unit/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py
View file

@ -477,7 +477,6 @@ def test_tool_call_arguments_are_chunked_to_match_openai_behavior():
assert evt is not None
assert evt.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED
assert evt.output_index == 0
assert hasattr(evt, "__dict__") and "sequence_number" in evt.__dict__
# Collect all remaining delta events from the pending queue by creating empty chunks
delta_events = []
@ -511,17 +510,11 @@ def test_tool_call_arguments_are_chunked_to_match_openai_behavior():
assert len(evt.delta) <= 10
assert evt.item_id == "fc_call_test"
assert evt.output_index == 0
assert hasattr(evt, "__dict__") and "sequence_number" in evt.__dict__
# Verify all deltas concatenated equal the original arguments
concatenated = "".join(evt.delta for evt in delta_events)
assert concatenated == large_arguments
# Verify sequence numbers are increasing
sequence_numbers = [evt.__dict__["sequence_number"] for evt in delta_events]
assert sequence_numbers == sorted(sequence_numbers)
assert len(set(sequence_numbers)) == len(sequence_numbers) # All unique
def test_tool_call_delta_without_id_uses_index_mapping():
iterator = LiteLLMCompletionStreamingIterator(
@ -1990,6 +1983,65 @@ def test_the_partial_response_billed_on_disconnect_carries_the_logging_objects_b
), "the bill is priced on the partial's own usage"
def _serialized_sequence_numbers(events: Iterable[BaseLiteLLMOpenAIResponseObject]) -> list[int]:
payloads: Final = tuple(json.loads(event.model_dump_json()) for event in events)
assert all("sequence_number" in payload for payload in payloads), [
payload["type"] for payload in payloads if "sequence_number" not in payload
]
return [payload["sequence_number"] for payload in payloads]
def _long_text_chunks() -> tuple[ModelResponseStream, ...]:
return (*(_chunk(f"word{i} ") for i in range(300)), _chunk("end", finish_reason="stop"))
def _text_then_tool_call_chunks() -> tuple[ModelResponseStream, ...]:
return (_text_opening_a_tool_call_chunk("Checking. "), _tool_arguments_chunk('{"x": 1}', finish_reason="tool_calls"))
@pytest.mark.parametrize(
"stream",
[
lambda: (_chunk("Hello"), _chunk("!", finish_reason="stop")),
_long_text_chunks,
_text_then_tool_call_chunks,
_reasoning_then_text_chunks,
_reasoning_and_text_in_one_chunk,
_reasoning_citation_then_tool_call_finishing_chunks,
lambda: (_tool_call_chunk(), _chunk("", finish_reason="tool_calls")),
lambda: (_refusal_chunk("I can't "), _refusal_chunk("help."), _chunk("", finish_reason="stop")),
lambda: (_chunk("Sure. "), _refusal_chunk("I can't help."), _chunk("", finish_reason="stop")),
],
ids=[
"text",
"long_text",
"text_then_tool",
"reasoning_then_text",
"reasoning_and_text",
"citation_tool",
"tool",
"refusal",
"text_then_refusal",
],
)
@pytest.mark.parametrize("sync_mode", [True, False])
@pytest.mark.asyncio
async def test_every_event_serializes_its_position_in_the_stream_from_zero(stream: StreamChunks, sync_mode: bool):
iterator: Final = _build_iterator(stream())
events: Final = list(iterator) if sync_mode else [event async for event in iterator]
assert _serialized_sequence_numbers(events) == list(range(len(events)))
def test_bridged_events_answer_dict_reads_of_their_number():
events: Final = list(_build_iterator((_chunk("Hello"), _chunk("!", finish_reason="stop"))))
positions: Final = list(range(len(events)))
assert [event["sequence_number"] for event in events] == positions
assert [event.get("sequence_number") for event in events] == positions
assert all("sequence_number" in event for event in events)
def _refusal_chunk(refusal: str, finish_reason: str | None = None) -> ModelResponseStream:
return ModelResponseStream(
id=CHAT_COMPLETION_ID,

51
tests/unit/responses/test_streaming_iterator.py
View file

@ -2867,3 +2867,54 @@ async def test_websocket_turns_logging_survives_garbage_collection(terminal_even
await _run_websocket_turn(logger, terminal_event)
assert await _logging_tasks_left_after_garbage_collection(logger) == 1
@pytest.mark.parametrize(
"output",
[
[
{
"type": "message",
"id": "msg_seq",
"status": "completed",
"role": "assistant",
"content": [{"type": "output_text", "text": "hello world", "annotations": []}],
}
],
[
{
"type": "reasoning",
"id": "rs_seq",
"summary": [{"type": "summary_text", "text": "thinking about it"}],
},
{
"type": "function_call",
"id": "fc_seq",
"call_id": "call_seq",
"name": "lookup",
"arguments": '{"id":1}',
},
],
],
ids=["message", "reasoning_and_tool"],
)
def test_synthetic_stream_serializes_every_event_position_from_zero(output: list[dict[str, object]]):
transformed: Final = ResponsesAPIResponse.model_validate(
{
"id": "resp_seq",
"created_at": 1741476542,
"status": "completed",
"model": "claude-sonnet-5-5",
"object": "response",
"output": output,
}
)
events: Final = streaming_module.build_synthetic_response_events(
transformed=transformed,
logging_obj=None,
chunk_size=5,
)
payloads: Final = tuple(json.loads(event.model_dump_json()) for event in events)
assert [payload.get("sequence_number") for payload in payloads] == list(range(len(payloads)))

145
tests/unit/test_router/test_router.py
View file

@ -6776,6 +6776,151 @@ async def test_aresponses_streaming_content_policy_error_event_routes_to_content
assert fallback_call.await_args.kwargs["model"] == "gemini/gemini-2.5-flash"
@pytest.mark.asyncio
async def test_aresponses_stream_keeps_numbering_after_a_mid_stream_fallback():
from litellm.types.llms.openai import OutputTextDeltaEvent, ResponsesAPIStreamEvents
def delta(sequence_number: int, text: str) -> OutputTextDeltaEvent:
return OutputTextDeltaEvent(
type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA,
item_id="msg_1",
output_index=0,
content_index=0,
delta=text,
sequence_number=sequence_number,
)
from litellm.exceptions import MidStreamFallbackError
router: Final = _InjectedFallbackRouter(_AsyncList([delta(0, "c"), delta(1, "d")]))
primary: Final = _make_responses_iterator(
chunks=[delta(0, "a"), delta(1, "b")],
error=MidStreamFallbackError(
message="primary died",
model="primary",
llm_provider="anthropic",
is_pre_first_chunk=False,
generated_content="ab",
),
model="primary",
)
wrapped: Final = await router._aresponses_streaming_iterator(
response=primary,
initial_kwargs={"model": "primary", "stream": True, "input": "Hi", "original_generic_function": litellm.aresponses},
)
events: Final = [event async for event in wrapped]
assert [(event.delta, event.sequence_number) for event in events] == [("a", 0), ("b", 1), ("c", 2), ("d", 3)]
@pytest.mark.asyncio
async def test_aresponses_stream_fallback_keeps_higher_numbers_and_passes_unnumbered_events():
from litellm.exceptions import MidStreamFallbackError
from litellm.types.llms.openai import OutputTextDeltaEvent, ResponsesAPIStreamEvents
def delta(text: str) -> OutputTextDeltaEvent:
return OutputTextDeltaEvent(
type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA, item_id="msg_1", output_index=0, content_index=0, delta=text
)
def numbered(sequence_number: int, text: str) -> OutputTextDeltaEvent:
return delta(text).model_copy(update={"sequence_number": sequence_number})
router: Final = _InjectedFallbackRouter(_AsyncList([numbered(7, "c"), delta("d"), numbered(8, "e")]))
primary: Final = _make_responses_iterator(
chunks=[numbered(0, "a"), numbered(1, "b")],
error=MidStreamFallbackError(
message="primary died",
model="primary",
llm_provider="anthropic",
is_pre_first_chunk=False,
generated_content="ab",
),
model="primary",
)
wrapped: Final = await router._aresponses_streaming_iterator(
response=primary,
initial_kwargs={"model": "primary", "stream": True, "input": "Hi", "original_generic_function": litellm.aresponses},
)
events: Final = [event async for event in wrapped]
assert [(event.delta, event.get("sequence_number")) for event in events] == [
("a", 0),
("b", 1),
("c", 7),
("d", None),
("e", 8),
]
@pytest.mark.asyncio
async def test_aresponses_stream_without_fallback_passes_upstream_numbers_unchanged():
from litellm.types.llms.openai import OutputTextDeltaEvent, ResponsesAPIStreamEvents
def delta(sequence_number: int, text: str) -> OutputTextDeltaEvent:
return OutputTextDeltaEvent(
type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA,
item_id="msg_1",
output_index=0,
content_index=0,
delta=text,
sequence_number=sequence_number,
)
router: Final = _InjectedFallbackRouter(_AsyncList([]))
primary: Final = _make_responses_iterator(
chunks=[delta(0, "a"), delta(1, "b"), delta(1, "c"), delta(0, "d")], model="primary"
)
wrapped: Final = await router._aresponses_streaming_iterator(
response=primary,
initial_kwargs={"model": "primary", "stream": True, "input": "Hi", "original_generic_function": litellm.aresponses},
)
events: Final = [event async for event in wrapped]
assert [(event.delta, event.sequence_number) for event in events] == [("a", 0), ("b", 1), ("c", 1), ("d", 0)]
@pytest.mark.asyncio
async def test_aresponses_stream_fallback_continues_above_the_highest_number_the_primary_sent():
from litellm.exceptions import MidStreamFallbackError
from litellm.types.llms.openai import OutputTextDeltaEvent, ResponsesAPIStreamEvents
def delta(sequence_number: int, text: str) -> OutputTextDeltaEvent:
return OutputTextDeltaEvent(
type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA,
item_id="msg_1",
output_index=0,
content_index=0,
delta=text,
sequence_number=sequence_number,
)
router: Final = _InjectedFallbackRouter(_AsyncList([delta(0, "d")]))
primary: Final = _make_responses_iterator(
chunks=[delta(0, "a"), delta(3, "b"), delta(1, "c")],
error=MidStreamFallbackError(
message="primary died",
model="primary",
llm_provider="anthropic",
is_pre_first_chunk=False,
generated_content="abc",
),
model="primary",
)
wrapped: Final = await router._aresponses_streaming_iterator(
response=primary,
initial_kwargs={"model": "primary", "stream": True, "input": "Hi", "original_generic_function": litellm.aresponses},
)
events: Final = [event async for event in wrapped]
assert [(event.delta, event.sequence_number) for event in events] == [("a", 0), ("b", 3), ("c", 1), ("d", 4)]
@pytest.mark.asyncio
async def test_aresponses_streaming_iterator_unwraps_content_policy_trigger_for_fallback_dispatch():
"""The fallback dispatch matches on the trigger's own type, so the wrapper must hand it the

31
tests/unit/types/llms/test_types_llms_openai.py
View file

@ -16,8 +16,10 @@ from litellm.types.llms.openai import (
OpenAIChatCompletionLogprobsContent,
OpenAIModerationResponse,
OpenAIVideoObject,
OutputTextDeltaEvent,
ResponseCompletedEvent,
ResponsesAPIResponse,
ResponsesAPIStreamEvents,
)
@ -816,3 +818,32 @@ def test_rebuilding_a_dumped_responses_answer_keeps_the_tool_search_call_keys(ex
rebuilt: Final = ResponsesAPIResponse(**answer.model_dump())
assert rebuilt.model_dump()["output"] == [search_call]
def _text_delta() -> OutputTextDeltaEvent:
return OutputTextDeltaEvent(
type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA, item_id="msg_1", output_index=0, content_index=0, delta="hi"
)
def test_stream_event_dict_reads_see_a_field_set_after_construction():
event: Final = _text_delta().model_copy(update={"sequence_number": 3})
assert event["sequence_number"] == 3
assert event.get("sequence_number") == 3
assert "sequence_number" in event
assert dict(event.items())["sequence_number"] == 3
assert event["delta"] == "hi"
def test_stream_event_dict_reads_see_an_undeclared_field_named_keys():
assert _text_delta().model_copy(update={"keys": "value"})["keys"] == "value"
def test_stream_event_dict_reads_still_miss_absent_keys():
event: Final = _text_delta()
assert event.get("sequence_number", "fallback") == "fallback"
assert "sequence_number" not in event
with pytest.raises(KeyError):
event["sequence_number"]