From 1100b2568bf96ba2a7d0b822c6a7a94e77f02b41 Mon Sep 17 00:00:00 2001 From: Ali Khan Date: Mon, 6 Jul 2026 22:36:50 -0400 Subject: [PATCH 1/6] fix(responses): synthesize missing streaming lifecycle events for native providers (#20975) Native /responses providers whose upstream truncates the streaming lifecycle (emitting only response.output_text.delta frames followed by response.completed) left strict clients like the OpenAI Codex CLI with no active item, failing hard with "OutputTextDelta without active item" The live async/sync streaming iterators did a strict one-chunk-to-one-event passthrough with no memory of which lifecycle events had been seen, so when the upstream omitted the response.created / response.in_progress / response.output_item.added / response.content_part.added openers and the matching output_text.done / content_part.done / output_item.done teardown, those events were never produced. The chat-completions bridge and the fake-stream/Mock/Cached paths already synthesize the full sequence; only the native live passthrough did not Add an idempotent, seen-tracking gap filler that the live iterators drain before pulling the next SSE frame. It synthesizes the missing openers and teardown, anchoring them to the same item_id / output_index / content_index as the deltas and backfilling done text from the accumulated deltas, and it is a no-op for providers that already emit the full spec sequence so compliant OpenAI / Azure / vLLM streams pass through byte-for-byte. Mock and Cached iterators override the loop and stay untouched The post-call streaming deployment hook runs on each real provider chunk before the gap filler accumulates it, so the synthesized done events carry post-hook (for example guardrail-redacted) text rather than the raw provider delta; a hook that redacts response.output_text.delta content is therefore not bypassed on the teardown Claude-Session: https://claude.ai/code/session_01HWegvoX1BdLDD34VD8H3mg --- litellm/responses/streaming_iterator.py | 453 ++++++++++++++- ...t_base_responses_api_streaming_iterator.py | 18 +- .../responses/test_streaming_iterator.py | 531 +++++++++++++++++- .../test_streaming_iterator_error_events.py | 9 +- 4 files changed, 986 insertions(+), 25 deletions(-) diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index 9f9016c5a7f..1d71c4d24b6 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -6,13 +6,15 @@ import time import traceback import uuid from collections.abc import Awaitable, Callable, Iterable, Mapping, Sequence +from dataclasses import dataclass from datetime import datetime from functools import lru_cache from types import MappingProxyType -from typing import TYPE_CHECKING, Any, Final, Literal, Protocol, overload, runtime_checkable +from typing import TYPE_CHECKING, Any, Final, Literal, Protocol, TypeVar, overload, runtime_checkable import httpx from openai._streaming import SSEDecoder +from pydantic import BaseModel from typing_extensions import TypeIs import litellm @@ -34,7 +36,19 @@ from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfi from litellm.responses.utils import ResponseAPILoggingUtils, ResponsesAPIRequestUtils from litellm.types.llms.openai import ( PART_UNION_TYPES, + BaseLiteLLMOpenAIResponseObject, + ContentPartAddedEvent, + ContentPartDoneEvent, + ContentPartDonePartOutputText, + ContentPartDonePartRefusal, + FunctionCallArgumentsDoneEvent, + OutputItemAddedEvent, + OutputItemDoneEvent, + OutputTextDoneEvent, + RefusalDoneEvent, ResponseAPIUsage, + ResponseCreatedEvent, + ResponseInProgressEvent, ResponsesAPIResponse, ResponsesAPIStreamEvents, ResponsesAPIStreamingResponse, @@ -220,6 +234,393 @@ def _status_code_for_error_fields(error_type: str | None, error_code: str | None ) +def _obj_get(obj: object, key: str, default: object | None = None) -> object: + """Read ``key`` from a dict or a pydantic/attr object uniformly.""" + if obj is None: + return default + if isinstance(obj, dict): + source: Mapping[object, object] = obj + return source.get(key, default) + return getattr(obj, key, default) + + +def _safe_int(value: object, default: int) -> int: + """Narrow a dynamically-read value to int, falling back for missing/malformed input.""" + if isinstance(value, bool): + return default + if isinstance(value, int): + return value + if isinstance(value, str): + try: + return int(value) + except ValueError: + return default + return default + + +def _safe_str(value: object, default: str) -> str: + return value if isinstance(value, str) else default + + +_ResponseModelT = TypeVar("_ResponseModelT", bound=BaseModel) + + +def _build_bag( + model_cls: type[_ResponseModelT], + **fields: object, # kwargs-ok: generic forwarder for extra-allow Responses payload models +) -> _ResponseModelT: + """ + Construct a Responses API pydantic payload from keyword fields. + + ``BaseLiteLLMOpenAIResponseObject`` (and the loosely-typed content-part models) + accept extra fields but declare none, so direct ``Cls(id=..., type=...)`` calls + trip the type checker. Funnelling construction through this generic keeps callers + strongly typed while the ``**fields`` splat keeps the field kwargs valid. + """ + return model_cls(**fields) + + +@dataclass(slots=True) +class _ResponsesStreamItemState: + """Per-``output_index`` lifecycle bookkeeping for one streamed output item.""" + + item_id: str + output_index: int + content_index: int = 0 + has_content_part: bool = True # message/refusal have a content part; function_call does not + part_kind: str = "output_text" # "output_text" | "refusal" + accumulated_text: str = "" + output_item_added_seen: bool = False + content_part_added_seen: bool = False + leaf_done_seen: bool = False # output_text.done / refusal.done / function_call_arguments.done + content_part_done_seen: bool = False + output_item_done_seen: bool = False + + +_ItemStateMap = dict[int, _ResponsesStreamItemState] + + +class _ResponsesLifecycleGapFiller: + """ + Guarantee the Responses API streaming lifecycle wrapper events are present. + + Native providers whose upstream emits only ``response.output_text.delta`` + + ``response.completed`` (e.g. github_copilot, ollama cloud, Azure gpt-5) leave + strict clients (OpenAI Codex CLI) without an "active item", which they reject + with ``OutputTextDelta without active item``. Given the one event a provider + just produced, ``expand`` prepends any missing openers + (``response.created``/``response.in_progress`` before the first event; + ``output_item.added``/``content_part.added`` before the first delta of an + item) and, right before ``response.completed``, any missing teardown + (``*.done``). Every injection is gated on a not-already-seen flag, so a + provider that already emits the full sequence passes through unchanged and + is never double-wrapped. + """ + + def __init__(self, *, model: str, response_id: str) -> None: + self._model = model + self._response_id = response_id + self._created_seen = False + self._in_progress_seen = False + # Per-output-index lifecycle state accumulated across streamed SSE chunks. + self._items: _ItemStateMap = {} # mutable-ok: per-chunk streaming state + + def expand(self, event: ResponsesAPIStreamingResponse) -> tuple[ResponsesAPIStreamingResponse, ...]: + """ + Given the one event a provider just produced, return the ordered events to + emit: any missing openers, then the event itself (and, for a terminal + event, any missing teardown before it). Response-level openers are tied to + the first item/content event, so a stream with no output (e.g. a lone + ``response.completed``) passes through untouched. + """ + ev = ResponsesAPIStreamEvents + etype = _obj_get(event, "type") + + if etype == ev.RESPONSE_CREATED: + self._created_seen = True + return (event,) + if etype == ev.RESPONSE_IN_PROGRESS: + self._created_seen = True + self._in_progress_seen = True + return (event,) + if etype == ev.OUTPUT_ITEM_ADDED: + openers = self._response_openers() + self._observe_output_item_added(event) + return (*openers, event) + if etype == ev.CONTENT_PART_ADDED: + openers = self._response_openers() + self._observe_content_part_added(event) + return (*openers, event) + if etype in (ev.OUTPUT_TEXT_DELTA, ev.REFUSAL_DELTA): + openers = ( + *self._response_openers(), + *self._ensure_message_item(event, is_refusal=(etype == ev.REFUSAL_DELTA)), + ) + self._accumulate(event, _safe_str(_obj_get(event, "delta", ""), "")) + return (*openers, event) + if etype == ev.FUNCTION_CALL_ARGUMENTS_DELTA: + openers = ( + *self._response_openers(), + *self._ensure_function_call_item(event), + ) + self._accumulate(event, _safe_str(_obj_get(event, "delta", ""), "")) + return (*openers, event) + if etype in ( + ev.OUTPUT_TEXT_DONE, + ev.REFUSAL_DONE, + ev.FUNCTION_CALL_ARGUMENTS_DONE, + ): + self._mark_seen(event, "leaf_done_seen") + return (event,) + if etype == ev.CONTENT_PART_DONE: + self._mark_seen(event, "content_part_done_seen") + return (event,) + if etype == ev.OUTPUT_ITEM_DONE: + self._observe_output_item_done(event) + return (event,) + if etype in (ev.RESPONSE_COMPLETED, ev.RESPONSE_INCOMPLETE, ev.RESPONSE_FAILED): + openers = self._response_openers() if (self._items or self._created_seen) else () + return (*openers, *self._teardown(), event) + return (event,) + + def _response_openers(self) -> tuple[BaseLiteLLMOpenAIResponseObject, ...]: + need_created = not self._created_seen + need_in_progress = not self._in_progress_seen + self._created_seen = True + self._in_progress_seen = True + return ( + *((self._status_event(is_created=True),) if need_created else ()), + *((self._status_event(is_created=False),) if need_in_progress else ()), + ) + + def _status_event(self, *, is_created: bool) -> BaseLiteLLMOpenAIResponseObject: + # Known caveat: when these openers are synthesized (truncated upstream), the + # real response id only arrives on response.completed, so response.created / + # response.in_progress carry the placeholder _response_id and will not match + # completed's id. Clients must correlate synthesized events by output_index, + # not response.id. We do not rewrite completed's real id (clients store it for + # follow-up GETs). Providers that emit their own response.created are passed + # through untouched and keep their real id. + response = _build_bag( + ResponsesAPIResponse, + id=self._response_id, + created_at=int(time.time()), + model=self._model, + object="response", + status="in_progress", + output=(), + ) + if is_created: + return ResponseCreatedEvent(type=ResponsesAPIStreamEvents.RESPONSE_CREATED, response=response) + return ResponseInProgressEvent(type=ResponsesAPIStreamEvents.RESPONSE_IN_PROGRESS, response=response) + + def _item_for(self, event: object) -> _ResponsesStreamItemState: + output_index = _safe_int(_obj_get(event, "output_index", 0), 0) + existing = self._items.get(output_index) + if existing is not None: + return existing + state = _ResponsesStreamItemState( + item_id=_safe_str(_obj_get(event, "item_id", ""), "") or self._response_id, + output_index=output_index, + content_index=_safe_int(_obj_get(event, "content_index", 0), 0), + ) + self._items[output_index] = state + return state + + def _ensure_message_item(self, event: object, *, is_refusal: bool) -> tuple[BaseLiteLLMOpenAIResponseObject, ...]: + state = self._item_for(event) + state.has_content_part = True + state.part_kind = "refusal" if is_refusal else "output_text" + need_item = not state.output_item_added_seen + need_part = not state.content_part_added_seen + state.output_item_added_seen = True + state.content_part_added_seen = True + return ( + *((self._build_output_item_added(state),) if need_item else ()), + *((self._build_content_part_added(state),) if need_part else ()), + ) + + def _ensure_function_call_item(self, event: object) -> tuple[BaseLiteLLMOpenAIResponseObject, ...]: + state = self._item_for(event) + state.has_content_part = False + if state.output_item_added_seen: + return () + state.output_item_added_seen = True + return (self._build_output_item_added(state),) + + def _accumulate(self, event: object, delta: str) -> None: + self._item_for(event).accumulated_text += delta + + def _mark_seen(self, event: object, flag: str) -> None: + setattr(self._item_for(event), flag, True) + + def _observe_output_item_added(self, event: object) -> None: + output_index = _safe_int(_obj_get(event, "output_index", 0), 0) + item = _obj_get(event, "item") + item_id = ( + _safe_str(_obj_get(item, "id", ""), "") + or _safe_str(_obj_get(event, "item_id", ""), "") + or self._response_id + ) + state = self._items.get(output_index) or _ResponsesStreamItemState(item_id=item_id, output_index=output_index) + state.output_item_added_seen = True + item_type = _obj_get(item, "type") + if item_type is not None: + state.has_content_part = item_type in ("message", "refusal") + self._items[output_index] = state + + def _observe_content_part_added(self, event: object) -> None: + self._item_for(event).content_part_added_seen = True + + def _observe_output_item_done(self, event: object) -> None: + output_index = _safe_int(_obj_get(event, "output_index", 0), 0) + state = self._items.get(output_index) + if state is not None: + state.output_item_done_seen = True + + def _teardown(self) -> tuple[BaseLiteLLMOpenAIResponseObject, ...]: + return tuple(event for _, state in sorted(self._items.items()) for event in self._item_teardown(state)) + + def _item_teardown(self, state: _ResponsesStreamItemState) -> tuple[BaseLiteLLMOpenAIResponseObject, ...]: + if state.output_item_done_seen: + return () + need_leaf = not state.leaf_done_seen + need_content_part = state.has_content_part and not state.content_part_done_seen + state.leaf_done_seen = True + state.content_part_done_seen = True + state.output_item_done_seen = True + return ( + *((self._build_leaf_done(state),) if need_leaf else ()), + *((self._build_content_part_done(state),) if need_content_part else ()), + self._build_output_item_done(state), + ) + + def _build_output_item_added(self, state: _ResponsesStreamItemState) -> OutputItemAddedEvent: + if state.has_content_part: + item = _build_bag( + BaseLiteLLMOpenAIResponseObject, + id=state.item_id, + type="message", + status="in_progress", + role="assistant", + content=(), + ) + else: + item = _build_bag( + BaseLiteLLMOpenAIResponseObject, + id=state.item_id, + type="function_call", + status="in_progress", + ) + return OutputItemAddedEvent( + type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, + output_index=state.output_index, + item=item, + ) + + def _build_content_part_added(self, state: _ResponsesStreamItemState) -> ContentPartAddedEvent: + if state.part_kind == "refusal": + part = _build_bag(BaseLiteLLMOpenAIResponseObject, type="refusal", refusal="") + else: + part = _build_bag( + BaseLiteLLMOpenAIResponseObject, + type="output_text", + text="", + annotations=(), + ) + return ContentPartAddedEvent( + type=ResponsesAPIStreamEvents.CONTENT_PART_ADDED, + item_id=state.item_id, + output_index=state.output_index, + content_index=state.content_index, + part=part, + ) + + def _build_leaf_done(self, state: _ResponsesStreamItemState) -> BaseLiteLLMOpenAIResponseObject: + if not state.has_content_part: + return FunctionCallArgumentsDoneEvent( + type=ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DONE, + item_id=state.item_id, + output_index=state.output_index, + arguments=state.accumulated_text, + ) + if state.part_kind == "refusal": + return RefusalDoneEvent( + type=ResponsesAPIStreamEvents.REFUSAL_DONE, + item_id=state.item_id, + output_index=state.output_index, + content_index=state.content_index, + refusal=state.accumulated_text, + ) + return OutputTextDoneEvent( + type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE, + item_id=state.item_id, + output_index=state.output_index, + content_index=state.content_index, + text=state.accumulated_text, + ) + + def _build_content_part_done(self, state: _ResponsesStreamItemState) -> ContentPartDoneEvent: + if state.part_kind == "refusal": + part: BaseLiteLLMOpenAIResponseObject = _build_bag( + ContentPartDonePartRefusal, + type="refusal", + refusal=state.accumulated_text, + ) + else: + part = _build_bag( + ContentPartDonePartOutputText, + type="output_text", + text=state.accumulated_text, + annotations=(), + logprobs=None, + ) + return ContentPartDoneEvent( + type=ResponsesAPIStreamEvents.CONTENT_PART_DONE, + item_id=state.item_id, + output_index=state.output_index, + content_index=state.content_index, + part=part, + ) + + def _build_output_item_done(self, state: _ResponsesStreamItemState) -> OutputItemDoneEvent: + if not state.has_content_part: + item = _build_bag( + BaseLiteLLMOpenAIResponseObject, + id=state.item_id, + type="function_call", + status="completed", + arguments=state.accumulated_text, + ) + else: + if state.part_kind == "refusal": + content_part = _build_bag( + BaseLiteLLMOpenAIResponseObject, + type="refusal", + refusal=state.accumulated_text, + ) + else: + content_part = _build_bag( + BaseLiteLLMOpenAIResponseObject, + type="output_text", + text=state.accumulated_text, + annotations=(), + ) + item = _build_bag( + BaseLiteLLMOpenAIResponseObject, + id=state.item_id, + type="message", + status="completed", + role="assistant", + content=(content_part,), + ) + return OutputItemDoneEvent( + type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, + output_index=state.output_index, + item=item, + ) + + class BaseResponsesAPIStreamingIterator: """ Base class for streaming iterators that process responses from the Responses API. @@ -254,6 +655,16 @@ class BaseResponsesAPIStreamingIterator: self._persist_completed_response_before_logging = True self._stream_created_time: float = time.time() + # Guarantee the Responses API streaming lifecycle wrapper events are present + # even when the upstream provider truncates them (issue #20975). Only the live + # __anext__/__next__ loops drain this; Mock/Cached iterators override the loop + # and build their own event list, so they never invoke it. + self._pending_events: tuple[ResponsesAPIStreamingResponse, ...] = () + self._lifecycle_gap_filler = _ResponsesLifecycleGapFiller( + model=model or "", + response_id=f"resp_{uuid.uuid4().hex}", + ) + # track request context for hooks self.litellm_metadata = litellm_metadata self.custom_llm_provider = custom_llm_provider @@ -870,6 +1281,13 @@ class ResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): try: self._check_max_streaming_duration() while True: + # Drain events the gap-filler already expanded (openers, the hooked + # provider chunk, teardown) before pulling the next SSE line. + if self._pending_events: + pending_event, self._pending_events = self._pending_events[0], self._pending_events[1:] + self._yielded_first_chunk = True + return pending_event + # Get the next chunk from the stream try: sse = await self.stream_iterator.__anext__() @@ -884,14 +1302,13 @@ class ResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): raise StopAsyncIteration elif result is not None: self._maybe_raise_for_error_event(result) - # Await hook directly instead of run_async_function - # (which spawns a thread + event loop per call) - result = await self._call_post_streaming_deployment_hook( - chunk=result, + # Run the deployment hook on the real chunk before the gap-filler + # accumulates it, so synthesized *.done events carry post-hook + # (e.g. guardrail-redacted) text, not the raw provider delta. + self._pending_events = self._lifecycle_gap_filler.expand( + await self._call_post_streaming_deployment_hook(chunk=result) ) - self._yielded_first_chunk = True - return result - # If result is None, continue the loop to get the next chunk + # Loop back to drain pending (or read the next chunk if none). except StopAsyncIteration: # Normal end of stream - don't log as failure @@ -952,6 +1369,13 @@ class SyncResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): try: self._check_max_streaming_duration() while True: + # Drain events the gap-filler already expanded before pulling the next + # SSE line (see the async path for the hook-ordering rationale). + if self._pending_events: + pending_event, self._pending_events = self._pending_events[0], self._pending_events[1:] + self._yielded_first_chunk = True + return pending_event + # Get the next chunk from the stream try: sse = next(self.stream_iterator) @@ -966,14 +1390,13 @@ class SyncResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): raise StopIteration elif result is not None: self._maybe_raise_for_error_event(result) - # Sync path: use run_async_function for the hook - result = run_async_function( - async_function=self._call_post_streaming_deployment_hook, - chunk=result, + self._pending_events = self._lifecycle_gap_filler.expand( + run_async_function( + async_function=self._call_post_streaming_deployment_hook, + chunk=result, + ) ) - self._yielded_first_chunk = True - return result - # If result is None, continue the loop to get the next chunk + # Loop back to drain pending (or read the next chunk if none). except StopIteration: # Normal end of stream - don't log as failure diff --git a/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py b/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py index bd617587cf3..f19b696b841 100644 --- a/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py +++ b/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py @@ -2,12 +2,12 @@ Unit tests for BaseResponsesAPIStreamingIterator Tests core functionality including: -1. Processing chunks and handling ResponseCompletedEvent +1. Processing chunks and handling ResponseCompletedEvent 2. Ensuring _update_responses_api_response_id_with_model_id is called for final chunk 3. Verifying ID update is NOT called for non-final chunks (delta events) 4. Edge case handling for invalid JSON, empty chunks, and [DONE] markers -These tests ensure the streaming iterator correctly processes response chunks +These tests ensure the streaming iterator correctly processes response chunks and applies model ID updates only to completed responses, as required for proper response tracking and logging. """ @@ -429,8 +429,11 @@ class TestBaseResponsesAPIStreamingIterator: except StopAsyncIteration: pass # This is expected - # Verify we got the chunk - assert len(chunks_received) == 1 + # The provider delta is delivered as the final event. Since #20975 the live + # iterator also synthesizes the missing lifecycle wrapper events ahead of a + # bare delta, so it is no longer necessarily the only chunk. + assert mock_delta_event in chunks_received + assert chunks_received[-1] is mock_delta_event # CRITICAL: Verify that failure handlers were NOT called # StopAsyncIteration is a normal end of stream, not a failure @@ -490,8 +493,11 @@ class TestBaseResponsesAPIStreamingIterator: except StopIteration: pass # This is expected - # Verify we got the chunk - assert len(chunks_received) == 1 + # The provider delta is delivered as the final event. Since #20975 the live + # iterator also synthesizes the missing lifecycle wrapper events ahead of a + # bare delta, so it is no longer necessarily the only chunk. + assert mock_delta_event in chunks_received + assert chunks_received[-1] is mock_delta_event # CRITICAL: Verify that failure handlers were NOT called # StopIteration is a normal end of stream, not a failure diff --git a/tests/test_litellm/responses/test_streaming_iterator.py b/tests/test_litellm/responses/test_streaming_iterator.py index c226c0b4d09..e03ac4db1a4 100644 --- a/tests/test_litellm/responses/test_streaming_iterator.py +++ b/tests/test_litellm/responses/test_streaming_iterator.py @@ -1,21 +1,47 @@ -"""Regression tests for LIT-4185 — /v1/responses streaming must stamp +"""Regression tests for litellm/responses/streaming_iterator.py. + +Two concerns live here: + +TTFT stamping (LIT-4185): /v1/responses streaming must stamp completion_start_time on the first chunk so downstream TTFT consumers (Prometheus, OTEL, SpendLogs completionStartTime) do not fall back to -completion_start_time = end_time.""" +completion_start_time = end_time. + +Lifecycle-event synthesis (issue #20975): native /responses providers whose +upstream truncates the streaming lifecycle (emitting only +response.output_text.delta ... response.completed) left strict clients like +OpenAI Codex CLI without an "active item" ("OutputTextDelta without active +item"). The live iterators must synthesize the missing setup (response.created, +response.in_progress, response.output_item.added, response.content_part.added) +and teardown (output_text.done, content_part.done, output_item.done) events, +pass an already-complete sequence through unchanged (idempotency), and run the +post-call streaming deployment hook BEFORE the gap filler accumulates deltas so +a hook that redacts delta text is not bypassed on the synthesized *.done events. + +The #20975 tests drive the REAL ResponsesAPIStreamingIterator / +SyncResponsesAPIStreamingIterator with the REAL OpenAIResponsesAPIConfig, +feeding a dependency-injected fake SSE byte stream (no monkeypatching of the +code under test). +""" import json from datetime import datetime -from typing import Optional +from typing import Any, Dict, List, Optional from unittest.mock import Mock, patch import httpx import pytest +import litellm from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig +from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig from litellm.responses.streaming_iterator import ( ResponsesAPIStreamingIterator, SyncResponsesAPIStreamingIterator, + _obj_get, + _ResponsesLifecycleGapFiller, + _safe_int, ) from litellm.types.llms.openai import ( ResponseCompletedEvent, @@ -23,6 +49,14 @@ from litellm.types.llms.openai import ( ResponsesAPIStreamEvents, ) +EV = ResponsesAPIStreamEvents +E = EV # shorthand + + +# --------------------------------------------------------------------------- +# TTFT stamping (LIT-4185) +# --------------------------------------------------------------------------- + def _sse_event(payload: dict) -> bytes: return f"data: {json.dumps(payload)}\n\n".encode("utf-8") @@ -537,3 +571,494 @@ async def test_streaming_logging_copy_fallback_leaves_caller_event_untouched(): assert logged == [iterator.completed_response] assert iterator.completed_response.response._hidden_params == {} + + +# --------------------------------------------------------------------------- +# Lifecycle-event synthesis (issue #20975) +# --------------------------------------------------------------------------- + + +def _response_body(status: str) -> Dict[str, Any]: + return { + "id": "resp_real_upstream", + "object": "response", + "created_at": 1_700_000_000, + "status": status, + "model": "gpt-5", + "output": [ + { + "id": "msg_1", + "type": "message", + "status": "completed", + "role": "assistant", + "content": [{"type": "output_text", "text": "Hello world", "annotations": []}], + } + ], + "parallel_tool_calls": True, + "tool_choice": "auto", + "tools": [], + } + + +def _sse_frames(events: List[Dict[str, Any]]) -> List[bytes]: + """One `data: {...}\\n\\n` SSE frame per event, plus a terminating [DONE].""" + frames = [f"data: {json.dumps(evt)}\n\n".encode("utf-8") for evt in events] + frames.append(b"data: [DONE]\n\n") + return frames + + +class _FakeStreamResponse: + """Minimal stand-in for httpx.Response exposing (a)iter_bytes over fixed frames.""" + + def __init__(self, frames: List[bytes]): + self.headers: Dict[str, str] = {} + self._frames = frames + + async def aiter_bytes(self): + for frame in self._frames: + yield frame + + def iter_bytes(self): + for frame in self._frames: + yield frame + + +def _make_logging_obj() -> Any: + logging_obj = Mock(spec=LiteLLMLoggingObj) + logging_obj.model_call_details = {"litellm_params": {}} + logging_obj.completion_start_time = None + return logging_obj + + +def _iterator(events: List[Dict[str, Any]], *, sync: bool, model: str = "gpt-5") -> Any: + response = _FakeStreamResponse(_sse_frames(events)) + cls = SyncResponsesAPIStreamingIterator if sync else ResponsesAPIStreamingIterator + return cls( + response=response, + model=model, + responses_api_provider_config=OpenAIResponsesAPIConfig(), + logging_obj=_make_logging_obj(), + litellm_metadata={"model_info": {"id": "model_123"}}, + custom_llm_provider="openai", + ) + + +async def _drive(events: List[Dict[str, Any]], *, sync: bool, model: str = "gpt-5") -> List[Any]: + iterator = _iterator(events, sync=sync, model=model) + collected: List[Any] = [] + if sync: + for chunk in iterator: + collected.append(chunk) + else: + async for chunk in iterator: + collected.append(chunk) + return collected + + +def _types(events: List[Any]) -> List[Any]: + return [getattr(e, "type", None) for e in events] + + +# ----- truncated upstream (the copilot / ollama / Azure case) ----- + +_TRUNCATED_TEXT_EVENTS: List[Dict[str, Any]] = [ + { + "type": "response.output_text.delta", + "item_id": "msg_1", + "output_index": 0, + "content_index": 0, + "delta": "Hello", + }, + { + "type": "response.output_text.delta", + "item_id": "msg_1", + "output_index": 0, + "content_index": 0, + "delta": " world", + }, + {"type": "response.completed", "response": _response_body("completed")}, +] + +_FULL_TEXT_EVENTS: List[Dict[str, Any]] = [ + {"type": "response.created", "response": _response_body("in_progress")}, + {"type": "response.in_progress", "response": _response_body("in_progress")}, + { + "type": "response.output_item.added", + "output_index": 0, + "item": { + "id": "msg_1", + "type": "message", + "status": "in_progress", + "role": "assistant", + "content": [], + }, + }, + { + "type": "response.content_part.added", + "item_id": "msg_1", + "output_index": 0, + "content_index": 0, + "part": {"type": "output_text", "text": "", "annotations": []}, + }, + { + "type": "response.output_text.delta", + "item_id": "msg_1", + "output_index": 0, + "content_index": 0, + "delta": "Hello world", + }, + { + "type": "response.output_text.done", + "item_id": "msg_1", + "output_index": 0, + "content_index": 0, + "text": "Hello world", + }, + { + "type": "response.content_part.done", + "item_id": "msg_1", + "output_index": 0, + "content_index": 0, + "part": {"type": "output_text", "text": "Hello world", "annotations": []}, + }, + { + "type": "response.output_item.done", + "output_index": 0, + "item": { + "id": "msg_1", + "type": "message", + "status": "completed", + "role": "assistant", + "content": [{"type": "output_text", "text": "Hello world", "annotations": []}], + }, + }, + {"type": "response.completed", "response": _response_body("completed")}, +] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("sync", [False, True], ids=["async", "sync"]) +async def test_truncated_text_stream_synthesizes_full_lifecycle(sync): + collected = await _drive(_TRUNCATED_TEXT_EVENTS, sync=sync) + types = _types(collected) + + assert types == [ + E.RESPONSE_CREATED, + E.RESPONSE_IN_PROGRESS, + E.OUTPUT_ITEM_ADDED, + E.CONTENT_PART_ADDED, + E.OUTPUT_TEXT_DELTA, + E.OUTPUT_TEXT_DELTA, + E.OUTPUT_TEXT_DONE, + E.CONTENT_PART_DONE, + E.OUTPUT_ITEM_DONE, + E.RESPONSE_COMPLETED, + ], types + + # openers must anchor to the same item_id / indices as the deltas + output_item_added = collected[2] + content_part_added = collected[3] + assert output_item_added.item.id == "msg_1" + assert content_part_added.item_id == "msg_1" + assert content_part_added.output_index == 0 + assert content_part_added.content_index == 0 + + # teardown text must equal the concatenation of streamed deltas + output_text_done = collected[6] + assert output_text_done.text == "Hello world" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("sync", [False, True], ids=["async", "sync"]) +async def test_complete_stream_passes_through_without_duplication(sync): + collected = await _drive(_FULL_TEXT_EVENTS, sync=sync) + types = _types(collected) + + # byte-for-byte: same event types, same count, nothing injected + assert types == [evt["type"] for evt in _FULL_TEXT_EVENTS], types + assert len(collected) == len(_FULL_TEXT_EVENTS) + # no duplicated openers + assert types.count(E.RESPONSE_CREATED) == 1 + assert types.count(E.OUTPUT_ITEM_ADDED) == 1 + assert types.count(E.CONTENT_PART_ADDED) == 1 + assert types.count(E.OUTPUT_ITEM_DONE) == 1 + + +@pytest.mark.asyncio +async def test_truncated_function_call_stream_synthesizes_item_lifecycle(): + events = [ + { + "type": "response.function_call_arguments.delta", + "item_id": "fc_1", + "output_index": 0, + "delta": '{"city":', + }, + { + "type": "response.function_call_arguments.delta", + "item_id": "fc_1", + "output_index": 0, + "delta": '"NYC"}', + }, + {"type": "response.completed", "response": _response_body("completed")}, + ] + collected = await _drive(events, sync=False) + types = _types(collected) + + assert types == [ + E.RESPONSE_CREATED, + E.RESPONSE_IN_PROGRESS, + E.OUTPUT_ITEM_ADDED, + E.FUNCTION_CALL_ARGUMENTS_DELTA, + E.FUNCTION_CALL_ARGUMENTS_DELTA, + E.FUNCTION_CALL_ARGUMENTS_DONE, + E.OUTPUT_ITEM_DONE, + E.RESPONSE_COMPLETED, + ], types + + # function_call items have NO content part + assert E.CONTENT_PART_ADDED not in types + assert E.CONTENT_PART_DONE not in types + + output_item_added = collected[2] + assert output_item_added.item.type == "function_call" + assert output_item_added.item.id == "fc_1" + + args_done = collected[5] + assert args_done.arguments == '{"city":"NYC"}' + + +@pytest.mark.asyncio +@pytest.mark.parametrize("sync", [False, True], ids=["async", "sync"]) +async def test_complete_gpt_5_6_reasoning_stream_preserves_item_lifecycle(sync: bool) -> None: + reasoning_events = [ + { + "type": "response.output_item.added", + "output_index": 0, + "item": {"id": "rs_1", "type": "reasoning", "summary": []}, + }, + { + "type": "response.reasoning_summary_text.delta", + "item_id": "rs_1", + "output_index": 0, + "summary_index": 0, + "delta": "Thinking", + }, + { + "type": "response.reasoning_summary_text.done", + "item_id": "rs_1", + "output_index": 0, + "summary_index": 0, + "sequence_number": 4, + "text": "Thinking", + }, + { + "type": "response.output_item.done", + "output_index": 0, + "item": { + "id": "rs_1", + "type": "reasoning", + "summary": [{"type": "summary_text", "text": "Thinking"}], + }, + }, + ] + message_events = [ + {**event, **({"output_index": 1} if "output_index" in event else {})} + for event in _FULL_TEXT_EVENTS[2:] + ] + events = [ + { + **event, + **({"response": {**event["response"], "model": "gpt-5.6"}} if "response" in event else {}), + } + for event in [*_FULL_TEXT_EVENTS[:2], *reasoning_events, *message_events] + ] + collected = await _drive(events, sync=sync, model="gpt-5.6") + + assert _types(collected) == [event["type"] for event in events] + assert collected[2].item.id == "rs_1" + assert collected[2].item.type == "reasoning" + assert collected[3].delta == "Thinking" + assert collected[4].text == "Thinking" + assert collected[5].item.type == "reasoning" + assert collected[6].output_index == 1 + assert collected[6].item.id == "msg_1" + assert collected[8].delta == "Hello world" + assert collected[9].text == "Hello world" + assert collected[-1].response.model == "gpt-5.6" + assert E.FUNCTION_CALL_ARGUMENTS_DONE not in _types(collected) + + +@pytest.mark.asyncio +async def test_synthesized_events_survive_proxy_serialization(): + """ + The proxy serializes each event with model_dump_json(exclude_none=True, + exclude_unset=True). Synthesized events must set their required fields + explicitly so nothing load-bearing is stripped off the wire. + """ + collected = await _drive(_TRUNCATED_TEXT_EVENTS, sync=False) + + required_by_type = { + E.OUTPUT_ITEM_ADDED: ["type", "output_index", "item"], + E.CONTENT_PART_ADDED: [ + "type", + "item_id", + "output_index", + "content_index", + "part", + ], + E.OUTPUT_TEXT_DONE: [ + "type", + "item_id", + "output_index", + "content_index", + "text", + ], + E.CONTENT_PART_DONE: [ + "type", + "item_id", + "output_index", + "content_index", + "part", + ], + E.OUTPUT_ITEM_DONE: ["type", "output_index", "item"], + } + + seen_types = set() + for event in collected: + etype = getattr(event, "type", None) + if etype not in required_by_type: + continue + seen_types.add(etype) + wire = json.loads(event.model_dump_json(exclude_none=True, exclude_unset=True)) + for field in required_by_type[etype]: + assert field in wire, f"{etype} lost required field {field}: {wire}" + + # all synthesized wrapper events were exercised + assert seen_types == set(required_by_type.keys()) + + +class _RedactingDeploymentHook: + """A streaming deployment hook that redacts output_text delta content.""" + + REDACTION = "[REDACTED]" + + async def async_post_call_streaming_deployment_hook(self, *, request_data, response_chunk, call_type): + if getattr(response_chunk, "type", None) == E.OUTPUT_TEXT_DELTA: + response_chunk.delta = self.REDACTION + return response_chunk + + +@pytest.fixture +def redacting_deployment_hook(): + hook = _RedactingDeploymentHook() + litellm.callbacks.append(hook) + try: + yield hook + finally: + litellm.callbacks.remove(hook) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("sync", [False, True], ids=["async", "sync"]) +async def test_streaming_hook_governs_synthesized_teardown(sync, redacting_deployment_hook): + """ + A post-call streaming deployment hook that redacts response.output_text.delta + must also govern the SYNTHESIZED teardown. The gap filler accumulates the + post-hook delta, so output_text.done / content_part.done / output_item.done + carry the redacted text, never the raw provider text (issue #20975 review: + the pre-hook accumulation leaked redacted content through the done events). + """ + redacted = _RedactingDeploymentHook.REDACTION * 2 # two deltas + collected = await _drive(_TRUNCATED_TEXT_EVENTS, sync=sync) + + by_type: Dict[Any, List[Any]] = {} + for event in collected: + by_type.setdefault(getattr(event, "type", None), []).append(event) + + # client-visible deltas are redacted + assert [d.delta for d in by_type[E.OUTPUT_TEXT_DELTA]] == [ + _RedactingDeploymentHook.REDACTION, + _RedactingDeploymentHook.REDACTION, + ] + + # synthesized teardown reflects the post-hook (redacted) accumulation + assert by_type[E.OUTPUT_TEXT_DONE][0].text == redacted + assert by_type[E.CONTENT_PART_DONE][0].part.text == redacted + assert by_type[E.OUTPUT_ITEM_DONE][0].item.content[0].text == redacted + + # the raw provider text never leaks anywhere in the stream + assert all(getattr(e, "text", None) != "Hello world" for e in collected) + + +# ----- truncated refusal stream ----- + +_TRUNCATED_REFUSAL_EVENTS: List[Dict[str, Any]] = [ + { + "type": "response.refusal.delta", + "item_id": "msg_r", + "output_index": 0, + "content_index": 0, + "delta": "I can", + }, + { + "type": "response.refusal.delta", + "item_id": "msg_r", + "output_index": 0, + "content_index": 0, + "delta": "not help", + }, + {"type": "response.completed", "response": _response_body("completed")}, +] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("sync", [False, True], ids=["async", "sync"]) +async def test_truncated_refusal_stream_synthesizes_lifecycle(sync): + collected = await _drive(_TRUNCATED_REFUSAL_EVENTS, sync=sync) + types = _types(collected) + + assert types == [ + E.RESPONSE_CREATED, + E.RESPONSE_IN_PROGRESS, + E.OUTPUT_ITEM_ADDED, + E.CONTENT_PART_ADDED, + E.REFUSAL_DELTA, + E.REFUSAL_DELTA, + E.REFUSAL_DONE, + E.CONTENT_PART_DONE, + E.OUTPUT_ITEM_DONE, + E.RESPONSE_COMPLETED, + ], types + + # the synthesized content part is a refusal part, not output_text + assert collected[3].part.type == "refusal" + # teardown carries the accumulated refusal text at every level + assert collected[6].refusal == "I cannot help" + assert collected[7].part.refusal == "I cannot help" + assert collected[8].item.content[0].refusal == "I cannot help" + + +def test_obj_get_handles_dict_object_and_none(): + assert _obj_get({"a": 1}, "a") == 1 + assert _obj_get({"a": 1}, "missing", "d") == "d" + assert _obj_get(None, "a", "d") == "d" + + class _Obj: + x = 5 + + assert _obj_get(_Obj(), "x") == 5 + assert _obj_get(_Obj(), "y", "fallback") == "fallback" + + +def test_safe_int_narrows_dynamic_values(): + assert _safe_int(3, 0) == 3 + assert _safe_int(True, 9) == 9 # bool is not an accepted int + assert _safe_int("5", 0) == 5 + assert _safe_int("nope", 7) == 7 + assert _safe_int(1.5, 4) == 4 + + +def test_gap_filler_passes_unknown_event_through(): + gap_filler = _ResponsesLifecycleGapFiller(model="m", response_id="resp_x") + event = {"type": "response.some_unhandled_event"} + assert gap_filler.expand(event) == (event,) diff --git a/tests/test_litellm/responses/test_streaming_iterator_error_events.py b/tests/test_litellm/responses/test_streaming_iterator_error_events.py index ad74861c096..aa459584830 100644 --- a/tests/test_litellm/responses/test_streaming_iterator_error_events.py +++ b/tests/test_litellm/responses/test_streaming_iterator_error_events.py @@ -212,7 +212,14 @@ async def test_async_iterator_error_after_first_chunk_carries_generated_content( with pytest.raises(MidStreamFallbackError) as exc_info: await _drain() - assert len(chunks) == 2 + assert [chunk.type for chunk in chunks] == [ + "response.created", + "response.in_progress", + "response.output_item.added", + "response.content_part.added", + "response.output_text.delta", + "response.output_text.delta", + ] assert exc_info.value.status_code == 500 assert exc_info.value.is_pre_first_chunk is False assert exc_info.value.generated_content == "hello world" From c82c53b69a3ee460c634a8cd09061d2239905098 Mon Sep 17 00:00:00 2001 From: Ali Khan Date: Tue, 8 Sep 2026 12:54:16 -0400 Subject: [PATCH 2/6] fix(responses): preserve native stream items and immutable lifecycle state --- litellm/responses/streaming_iterator.py | 121 ++++++++++-------- ...t_fireworks_ai_responses_transformation.py | 26 +++- .../responses/test_streaming_iterator.py | 36 ++++-- 3 files changed, 122 insertions(+), 61 deletions(-) diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index 1d71c4d24b6..114f2f16d6e 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -6,7 +6,7 @@ import time import traceback import uuid from collections.abc import Awaitable, Callable, Iterable, Mapping, Sequence -from dataclasses import dataclass +from dataclasses import dataclass, replace from datetime import datetime from functools import lru_cache from types import MappingProxyType @@ -280,14 +280,14 @@ def _build_bag( return model_cls(**fields) -@dataclass(slots=True) +@dataclass(frozen=True, slots=True) class _ResponsesStreamItemState: """Per-``output_index`` lifecycle bookkeeping for one streamed output item.""" item_id: str output_index: int content_index: int = 0 - has_content_part: bool = True # message/refusal have a content part; function_call does not + item_type: Literal["message", "function_call"] = "message" part_kind: str = "output_text" # "output_text" | "refusal" accumulated_text: str = "" output_item_added_seen: bool = False @@ -296,8 +296,12 @@ class _ResponsesStreamItemState: content_part_done_seen: bool = False output_item_done_seen: bool = False + @property + def has_content_part(self) -> bool: + return self.item_type == "message" -_ItemStateMap = dict[int, _ResponsesStreamItemState] + +_ItemStateMap = Mapping[int, _ResponsesStreamItemState] class _ResponsesLifecycleGapFiller: @@ -322,8 +326,7 @@ class _ResponsesLifecycleGapFiller: self._response_id = response_id self._created_seen = False self._in_progress_seen = False - # Per-output-index lifecycle state accumulated across streamed SSE chunks. - self._items: _ItemStateMap = {} # mutable-ok: per-chunk streaming state + self._items: _ItemStateMap = MappingProxyType({}) def expand(self, event: ResponsesAPIStreamingResponse) -> tuple[ResponsesAPIStreamingResponse, ...]: """ @@ -336,12 +339,10 @@ class _ResponsesLifecycleGapFiller: ev = ResponsesAPIStreamEvents etype = _obj_get(event, "type") - if etype == ev.RESPONSE_CREATED: + if etype in (ev.RESPONSE_CREATED, ev.RESPONSE_IN_PROGRESS): + self._response_id = _safe_str(_obj_get(_obj_get(event, "response"), "id"), self._response_id) self._created_seen = True - return (event,) - if etype == ev.RESPONSE_IN_PROGRESS: - self._created_seen = True - self._in_progress_seen = True + self._in_progress_seen = self._in_progress_seen or etype == ev.RESPONSE_IN_PROGRESS return (event,) if etype == ev.OUTPUT_ITEM_ADDED: openers = self._response_openers() @@ -379,7 +380,7 @@ class _ResponsesLifecycleGapFiller: self._observe_output_item_done(event) return (event,) if etype in (ev.RESPONSE_COMPLETED, ev.RESPONSE_INCOMPLETE, ev.RESPONSE_FAILED): - openers = self._response_openers() if (self._items or self._created_seen) else () + openers = self._response_openers() if self._items else () return (*openers, *self._teardown(), event) return (event,) @@ -415,68 +416,90 @@ class _ResponsesLifecycleGapFiller: return ResponseInProgressEvent(type=ResponsesAPIStreamEvents.RESPONSE_IN_PROGRESS, response=response) def _item_for(self, event: object) -> _ResponsesStreamItemState: - output_index = _safe_int(_obj_get(event, "output_index", 0), 0) - existing = self._items.get(output_index) + output_index: Final = _safe_int(_obj_get(event, "output_index", 0), 0) + existing: Final = self._items.get(output_index) if existing is not None: return existing - state = _ResponsesStreamItemState( + state: Final = _ResponsesStreamItemState( item_id=_safe_str(_obj_get(event, "item_id", ""), "") or self._response_id, output_index=output_index, content_index=_safe_int(_obj_get(event, "content_index", 0), 0), ) - self._items[output_index] = state + return self._store_item(state) + + def _store_item(self, state: _ResponsesStreamItemState) -> _ResponsesStreamItemState: + self._items = MappingProxyType({**self._items, state.output_index: state}) return state def _ensure_message_item(self, event: object, *, is_refusal: bool) -> tuple[BaseLiteLLMOpenAIResponseObject, ...]: - state = self._item_for(event) - state.has_content_part = True - state.part_kind = "refusal" if is_refusal else "output_text" - need_item = not state.output_item_added_seen - need_part = not state.content_part_added_seen - state.output_item_added_seen = True - state.content_part_added_seen = True + previous: Final = self._item_for(event) + need_item: Final = not previous.output_item_added_seen + need_part: Final = not previous.content_part_added_seen + state: Final = self._store_item( + replace( + previous, + item_type="message", + part_kind="refusal" if is_refusal else "output_text", + output_item_added_seen=True, + content_part_added_seen=True, + ) + ) return ( *((self._build_output_item_added(state),) if need_item else ()), *((self._build_content_part_added(state),) if need_part else ()), ) def _ensure_function_call_item(self, event: object) -> tuple[BaseLiteLLMOpenAIResponseObject, ...]: - state = self._item_for(event) - state.has_content_part = False - if state.output_item_added_seen: - return () - state.output_item_added_seen = True - return (self._build_output_item_added(state),) + previous: Final = self._item_for(event) + state: Final = self._store_item(replace(previous, item_type="function_call", output_item_added_seen=True)) + return () if previous.output_item_added_seen else (self._build_output_item_added(state),) def _accumulate(self, event: object, delta: str) -> None: - self._item_for(event).accumulated_text += delta + state: Final = self._item_for(event) + self._store_item(replace(state, accumulated_text=state.accumulated_text + delta)) - def _mark_seen(self, event: object, flag: str) -> None: - setattr(self._item_for(event), flag, True) + def _mark_seen(self, event: object, flag: Literal["leaf_done_seen", "content_part_done_seen"]) -> None: + state: Final = self._item_for(event) + self._store_item( + replace( + state, + leaf_done_seen=state.leaf_done_seen or flag == "leaf_done_seen", + content_part_done_seen=state.content_part_done_seen or flag == "content_part_done_seen", + ) + ) def _observe_output_item_added(self, event: object) -> None: - output_index = _safe_int(_obj_get(event, "output_index", 0), 0) - item = _obj_get(event, "item") - item_id = ( + output_index: Final = _safe_int(_obj_get(event, "output_index", 0), 0) + item: Final = _obj_get(event, "item") + item_type: Final = _obj_get(item, "type") + # Reasoning and server-side tools have separate lifecycles; never synthesize function calls for them. + if item_type not in ("message", "refusal", "function_call"): + return + item_id: Final = ( _safe_str(_obj_get(item, "id", ""), "") or _safe_str(_obj_get(event, "item_id", ""), "") or self._response_id ) - state = self._items.get(output_index) or _ResponsesStreamItemState(item_id=item_id, output_index=output_index) - state.output_item_added_seen = True - item_type = _obj_get(item, "type") - if item_type is not None: - state.has_content_part = item_type in ("message", "refusal") - self._items[output_index] = state + state: Final = self._items.get(output_index) or _ResponsesStreamItemState( + item_id=item_id, output_index=output_index + ) + self._store_item( + replace( + state, + item_id=item_id, + item_type="function_call" if item_type == "function_call" else "message", + output_item_added_seen=True, + ) + ) def _observe_content_part_added(self, event: object) -> None: - self._item_for(event).content_part_added_seen = True + self._store_item(replace(self._item_for(event), content_part_added_seen=True)) def _observe_output_item_done(self, event: object) -> None: - output_index = _safe_int(_obj_get(event, "output_index", 0), 0) - state = self._items.get(output_index) + output_index: Final = _safe_int(_obj_get(event, "output_index", 0), 0) + state: Final = self._items.get(output_index) if state is not None: - state.output_item_done_seen = True + self._store_item(replace(state, output_item_done_seen=True)) def _teardown(self) -> tuple[BaseLiteLLMOpenAIResponseObject, ...]: return tuple(event for _, state in sorted(self._items.items()) for event in self._item_teardown(state)) @@ -484,11 +507,9 @@ class _ResponsesLifecycleGapFiller: def _item_teardown(self, state: _ResponsesStreamItemState) -> tuple[BaseLiteLLMOpenAIResponseObject, ...]: if state.output_item_done_seen: return () - need_leaf = not state.leaf_done_seen - need_content_part = state.has_content_part and not state.content_part_done_seen - state.leaf_done_seen = True - state.content_part_done_seen = True - state.output_item_done_seen = True + need_leaf: Final = not state.leaf_done_seen + need_content_part: Final = state.has_content_part and not state.content_part_done_seen + self._store_item(replace(state, leaf_done_seen=True, content_part_done_seen=True, output_item_done_seen=True)) return ( *((self._build_leaf_done(state),) if need_leaf else ()), *((self._build_content_part_done(state),) if need_content_part else ()), diff --git a/tests/test_litellm/llms/fireworks_ai/responses/test_fireworks_ai_responses_transformation.py b/tests/test_litellm/llms/fireworks_ai/responses/test_fireworks_ai_responses_transformation.py index b9408d44e9a..219e5892209 100644 --- a/tests/test_litellm/llms/fireworks_ai/responses/test_fireworks_ai_responses_transformation.py +++ b/tests/test_litellm/llms/fireworks_ai/responses/test_fireworks_ai_responses_transformation.py @@ -450,6 +450,30 @@ def test_streaming_responses_call_hits_native_endpoint_and_yields_every_firework True, True, ) - assert tuple(event.type for event in received) == tuple(event["type"] for event in FIREWORKS_SSE_EVENTS) + assert tuple(event.type for event in received) == ( + "response.created", + "response.in_progress", + "response.output_item.added", + "response.reasoning_summary_text.delta", + "response.output_item.added", + "response.content_part.added", + "response.output_text.delta", + "response.output_text.delta", + "response.output_text.done", + "response.content_part.done", + "response.output_item.done", + "response.completed", + ) + upstream_events: Final = tuple(received[index] for index in (0, 2, 3, 4, 6, 7, 11)) + assert tuple(event.type for event in upstream_events) == tuple(event["type"] for event in FIREWORKS_SSE_EVENTS) + assert tuple(event.sequence_number for event in upstream_events) == tuple(range(len(FIREWORKS_SSE_EVENTS))) + assert received[0].response.id == received[1].response.id == received[-1].response.id + assert received[2].item.type == "reasoning" + assert received[3].delta == "pong" + assert received[8].text == "pong" + assert received[9].part.text == "pong" + assert received[10].item.type == "message" + assert received[10].item.id == "msg_1" + assert received[10].output_index == 1 assert "".join(event.delta for event in received if event.type == "response.output_text.delta") == "pong" assert received[-1].response.usage.output_tokens == 89 diff --git a/tests/test_litellm/responses/test_streaming_iterator.py b/tests/test_litellm/responses/test_streaming_iterator.py index e03ac4db1a4..c75cb8cba10 100644 --- a/tests/test_litellm/responses/test_streaming_iterator.py +++ b/tests/test_litellm/responses/test_streaming_iterator.py @@ -829,7 +829,8 @@ async def test_truncated_function_call_stream_synthesizes_item_lifecycle(): @pytest.mark.asyncio @pytest.mark.parametrize("sync", [False, True], ids=["async", "sync"]) -async def test_complete_gpt_5_6_reasoning_stream_preserves_item_lifecycle(sync: bool) -> None: +@pytest.mark.parametrize("complete_reasoning", [False, True], ids=["truncated-reasoning", "complete-reasoning"]) +async def test_gpt_5_6_reasoning_stream_preserves_item_lifecycle(sync: bool, complete_reasoning: bool) -> None: reasoning_events = [ { "type": "response.output_item.added", @@ -862,15 +863,18 @@ async def test_complete_gpt_5_6_reasoning_stream_preserves_item_lifecycle(sync: }, ] message_events = [ - {**event, **({"output_index": 1} if "output_index" in event else {})} - for event in _FULL_TEXT_EVENTS[2:] + {**event, **({"output_index": 1} if "output_index" in event else {})} for event in _FULL_TEXT_EVENTS[2:] ] events = [ { **event, **({"response": {**event["response"], "model": "gpt-5.6"}} if "response" in event else {}), } - for event in [*_FULL_TEXT_EVENTS[:2], *reasoning_events, *message_events] + for event in [ + *_FULL_TEXT_EVENTS[:2], + *(reasoning_events if complete_reasoning else reasoning_events[:2]), + *message_events, + ] ] collected = await _drive(events, sync=sync, model="gpt-5.6") @@ -878,16 +882,28 @@ async def test_complete_gpt_5_6_reasoning_stream_preserves_item_lifecycle(sync: assert collected[2].item.id == "rs_1" assert collected[2].item.type == "reasoning" assert collected[3].delta == "Thinking" - assert collected[4].text == "Thinking" - assert collected[5].item.type == "reasoning" - assert collected[6].output_index == 1 - assert collected[6].item.id == "msg_1" - assert collected[8].delta == "Hello world" - assert collected[9].text == "Hello world" + if complete_reasoning: + assert collected[4].text == "Thinking" + assert collected[5].item.type == "reasoning" + message_start = 6 if complete_reasoning else 4 + assert collected[message_start].output_index == 1 + assert collected[message_start].item.id == "msg_1" + assert collected[message_start + 2].delta == "Hello world" + assert collected[message_start + 3].text == "Hello world" assert collected[-1].response.model == "gpt-5.6" assert E.FUNCTION_CALL_ARGUMENTS_DONE not in _types(collected) +@pytest.mark.asyncio +@pytest.mark.parametrize("sync", [False, True], ids=["async", "sync"]) +async def test_stream_without_item_events_preserves_response_status_events(sync: bool) -> None: + events = [_FULL_TEXT_EVENTS[0], _FULL_TEXT_EVENTS[-1]] + collected = await _drive(events, sync=sync) + + assert _types(collected) == [E.RESPONSE_CREATED, E.RESPONSE_COMPLETED] + assert collected[0].response.id == collected[1].response.id + + @pytest.mark.asyncio async def test_synthesized_events_survive_proxy_serialization(): """ From 4fc45377e178cd4c2166771995576c05f847b497 Mon Sep 17 00:00:00 2001 From: Ali Khan Date: Tue, 8 Sep 2026 13:07:30 -0400 Subject: [PATCH 3/6] fix(responses): close truncated reasoning items without losing payloads --- litellm/responses/streaming_iterator.py | 93 ++- ...t_fireworks_ai_responses_transformation.py | 16 +- .../responses/test_streaming_iterator.py | 688 ++++++++++-------- 3 files changed, 461 insertions(+), 336 deletions(-) diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index 114f2f16d6e..1eade16a520 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -238,7 +238,7 @@ def _obj_get(obj: object, key: str, default: object | None = None) -> object: """Read ``key`` from a dict or a pydantic/attr object uniformly.""" if obj is None: return default - if isinstance(obj, dict): + if isinstance(obj, Mapping): source: Mapping[object, object] = obj return source.get(key, default) return getattr(obj, key, default) @@ -287,7 +287,9 @@ class _ResponsesStreamItemState: item_id: str output_index: int content_index: int = 0 - item_type: Literal["message", "function_call"] = "message" + item_type: str = "message" + item_snapshot: str = "{}" + reasoning_summary: tuple[tuple[int, str], ...] = () part_kind: str = "output_text" # "output_text" | "refusal" accumulated_text: str = "" output_item_added_seen: bool = False @@ -366,6 +368,14 @@ class _ResponsesLifecycleGapFiller: ) self._accumulate(event, _safe_str(_obj_get(event, "delta", ""), "")) return (*openers, event) + if etype in ( + "response.reasoning_summary_text.delta", + "response.reasoning_summary_text.done", + "response.reasoning_summary_part.added", + "response.reasoning_summary_part.done", + ): + self._observe_reasoning_summary(event, is_delta=etype == "response.reasoning_summary_text.delta") + return (event,) if etype in ( ev.OUTPUT_TEXT_DONE, ev.REFUSAL_DONE, @@ -381,7 +391,7 @@ class _ResponsesLifecycleGapFiller: return (event,) if etype in (ev.RESPONSE_COMPLETED, ev.RESPONSE_INCOMPLETE, ev.RESPONSE_FAILED): openers = self._response_openers() if self._items else () - return (*openers, *self._teardown(), event) + return (*openers, *self._teardown(event), event) return (event,) def _response_openers(self) -> tuple[BaseLiteLLMOpenAIResponseObject, ...]: @@ -472,9 +482,6 @@ class _ResponsesLifecycleGapFiller: output_index: Final = _safe_int(_obj_get(event, "output_index", 0), 0) item: Final = _obj_get(event, "item") item_type: Final = _obj_get(item, "type") - # Reasoning and server-side tools have separate lifecycles; never synthesize function calls for them. - if item_type not in ("message", "refusal", "function_call"): - return item_id: Final = ( _safe_str(_obj_get(item, "id", ""), "") or _safe_str(_obj_get(event, "item_id", ""), "") @@ -487,7 +494,8 @@ class _ResponsesLifecycleGapFiller: replace( state, item_id=item_id, - item_type="function_call" if item_type == "function_call" else "message", + item_type="message" if item_type == "refusal" else _safe_str(item_type, "message"), + item_snapshot=item.model_dump_json() if isinstance(item, BaseModel) else json.dumps(item), output_item_added_seen=True, ) ) @@ -501,12 +509,38 @@ class _ResponsesLifecycleGapFiller: if state is not None: self._store_item(replace(state, output_item_done_seen=True)) - def _teardown(self) -> tuple[BaseLiteLLMOpenAIResponseObject, ...]: - return tuple(event for _, state in sorted(self._items.items()) for event in self._item_teardown(state)) + def _observe_reasoning_summary(self, event: object, *, is_delta: bool) -> None: + state: Final = self._items.get(_safe_int(_obj_get(event, "output_index"), 0)) + if state is None or state.item_type != "reasoning": + return + index: Final = _safe_int(_obj_get(event, "summary_index"), 0) + previous: Final = next((text for part_index, text in state.reasoning_summary if part_index == index), "") + text: Final = _safe_str( + _obj_get(event, "delta", _obj_get(event, "text", _obj_get(_obj_get(event, "part"), "text"))), "" + ) + self._store_item( + replace( + state, + reasoning_summary=( + *((part_index, value) for part_index, value in state.reasoning_summary if part_index != index), + (index, previous + text if is_delta else text), + ), + ) + ) - def _item_teardown(self, state: _ResponsesStreamItemState) -> tuple[BaseLiteLLMOpenAIResponseObject, ...]: + def _teardown(self, terminal_event: object) -> tuple[BaseLiteLLMOpenAIResponseObject, ...]: + return tuple( + event for _, state in sorted(self._items.items()) for event in self._item_teardown(state, terminal_event) + ) + + def _item_teardown( + self, state: _ResponsesStreamItemState, terminal_event: object + ) -> tuple[BaseLiteLLMOpenAIResponseObject, ...]: if state.output_item_done_seen: return () + if state.item_type not in ("message", "function_call"): + self._store_item(replace(state, output_item_done_seen=True)) + return (self._build_other_item_done(state, terminal_event),) need_leaf: Final = not state.leaf_done_seen need_content_part: Final = state.has_content_part and not state.content_part_done_seen self._store_item(replace(state, leaf_done_seen=True, content_part_done_seen=True, output_item_done_seen=True)) @@ -516,6 +550,45 @@ class _ResponsesLifecycleGapFiller: self._build_output_item_done(state), ) + def _build_other_item_done(self, state: _ResponsesStreamItemState, terminal_event: object) -> OutputItemDoneEvent: + response: Final = _obj_get(terminal_event, "response") + terminal_item: Final = next( + ( + item + for item in _json_array_or_empty(_obj_get(response, "output")) + if _obj_get(item, "id") == state.item_id + ), + None, + ) + payload: Final = _load_json_value( + state.item_snapshot + if terminal_item is None + else terminal_item.model_dump_json() + if isinstance(terminal_item, BaseModel) + else json.dumps(terminal_item) + ) + fields: Final = MappingProxyType(payload) if _is_json_object(payload) else EMPTY_MAPPING + summary: Final = MappingProxyType( + { + "summary": tuple( + _build_bag(BaseLiteLLMOpenAIResponseObject, type="summary_text", text=text) + for _, text in sorted(state.reasoning_summary) + ) + } + if state.reasoning_summary + else {} + ) + return OutputItemDoneEvent( + type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, + output_index=state.output_index, + item=_build_bag( + BaseLiteLLMOpenAIResponseObject, + **MappingProxyType( + {**fields, "id": state.item_id, "type": state.item_type, "status": "completed", **summary} + ), + ), + ) + def _build_output_item_added(self, state: _ResponsesStreamItemState) -> OutputItemAddedEvent: if state.has_content_part: item = _build_bag( diff --git a/tests/test_litellm/llms/fireworks_ai/responses/test_fireworks_ai_responses_transformation.py b/tests/test_litellm/llms/fireworks_ai/responses/test_fireworks_ai_responses_transformation.py index 219e5892209..59d47a77597 100644 --- a/tests/test_litellm/llms/fireworks_ai/responses/test_fireworks_ai_responses_transformation.py +++ b/tests/test_litellm/llms/fireworks_ai/responses/test_fireworks_ai_responses_transformation.py @@ -459,21 +459,25 @@ def test_streaming_responses_call_hits_native_endpoint_and_yields_every_firework "response.content_part.added", "response.output_text.delta", "response.output_text.delta", + "response.output_item.done", "response.output_text.done", "response.content_part.done", "response.output_item.done", "response.completed", ) - upstream_events: Final = tuple(received[index] for index in (0, 2, 3, 4, 6, 7, 11)) + upstream_events: Final = tuple(received[index] for index in (0, 2, 3, 4, 6, 7, 12)) assert tuple(event.type for event in upstream_events) == tuple(event["type"] for event in FIREWORKS_SSE_EVENTS) assert tuple(event.sequence_number for event in upstream_events) == tuple(range(len(FIREWORKS_SSE_EVENTS))) assert received[0].response.id == received[1].response.id == received[-1].response.id assert received[2].item.type == "reasoning" assert received[3].delta == "pong" - assert received[8].text == "pong" - assert received[9].part.text == "pong" - assert received[10].item.type == "message" - assert received[10].item.id == "msg_1" - assert received[10].output_index == 1 + assert received[8].item.type == "reasoning" + assert received[8].item.id == "rs_1" + assert json.loads(received[8].model_dump_json())["item"]["summary"][0]["text"] == "pong" + assert received[9].text == "pong" + assert received[10].part.text == "pong" + assert received[11].item.type == "message" + assert received[11].item.id == "msg_1" + assert received[11].output_index == 1 assert "".join(event.delta for event in received if event.type == "response.output_text.delta") == "pong" assert received[-1].response.usage.output_tokens == 89 diff --git a/tests/test_litellm/responses/test_streaming_iterator.py b/tests/test_litellm/responses/test_streaming_iterator.py index c75cb8cba10..e9d2dbd0d64 100644 --- a/tests/test_litellm/responses/test_streaming_iterator.py +++ b/tests/test_litellm/responses/test_streaming_iterator.py @@ -25,8 +25,10 @@ code under test). """ import json +from collections.abc import Mapping, Sequence from datetime import datetime -from typing import Any, Dict, List, Optional +from types import MappingProxyType +from typing import Any, Dict, Final, List, Optional from unittest.mock import Mock, patch import httpx @@ -578,40 +580,44 @@ async def test_streaming_logging_copy_fallback_leaves_caller_event_untouched(): # --------------------------------------------------------------------------- -def _response_body(status: str) -> Dict[str, Any]: - return { - "id": "resp_real_upstream", - "object": "response", - "created_at": 1_700_000_000, - "status": status, - "model": "gpt-5", - "output": [ - { - "id": "msg_1", - "type": "message", - "status": "completed", - "role": "assistant", - "content": [{"type": "output_text", "text": "Hello world", "annotations": []}], - } - ], - "parallel_tool_calls": True, - "tool_choice": "auto", - "tools": [], - } +def _response_body(status: str) -> Mapping[str, Any]: + return MappingProxyType( + { + "id": "resp_real_upstream", + "object": "response", + "created_at": 1700000000, + "status": status, + "model": "gpt-5", + "output": ( + MappingProxyType( + { + "id": "msg_1", + "type": "message", + "status": "completed", + "role": "assistant", + "content": ( + MappingProxyType({"type": "output_text", "text": "Hello world", "annotations": ()}), + ), + } + ), + ), + "parallel_tool_calls": True, + "tool_choice": "auto", + "tools": (), + } + ) -def _sse_frames(events: List[Dict[str, Any]]) -> List[bytes]: +def _sse_frames(events: Sequence[Mapping[str, object]]) -> tuple[bytes, ...]: """One `data: {...}\\n\\n` SSE frame per event, plus a terminating [DONE].""" - frames = [f"data: {json.dumps(evt)}\n\n".encode("utf-8") for evt in events] - frames.append(b"data: [DONE]\n\n") - return frames + return (*(f"data: {json.dumps(evt, default=dict)}\n\n".encode("utf-8") for evt in events), b"data: [DONE]\n\n") class _FakeStreamResponse: """Minimal stand-in for httpx.Response exposing (a)iter_bytes over fixed frames.""" - def __init__(self, frames: List[bytes]): - self.headers: Dict[str, str] = {} + def __init__(self, frames: tuple[bytes, ...]): + self.headers: Mapping[str, str] = MappingProxyType({}) self._frames = frames async def aiter_bytes(self): @@ -624,125 +630,130 @@ class _FakeStreamResponse: def _make_logging_obj() -> Any: - logging_obj = Mock(spec=LiteLLMLoggingObj) - logging_obj.model_call_details = {"litellm_params": {}} + logging_obj: Final = Mock(spec=LiteLLMLoggingObj) + logging_obj.model_call_details = MappingProxyType({"litellm_params": MappingProxyType({})}) logging_obj.completion_start_time = None return logging_obj -def _iterator(events: List[Dict[str, Any]], *, sync: bool, model: str = "gpt-5") -> Any: - response = _FakeStreamResponse(_sse_frames(events)) - cls = SyncResponsesAPIStreamingIterator if sync else ResponsesAPIStreamingIterator +def _iterator(events: tuple[Mapping[str, Any], ...], *, sync: bool, model: str = "gpt-5") -> Any: + response: Final = _FakeStreamResponse(_sse_frames(events)) + cls: Final = SyncResponsesAPIStreamingIterator if sync else ResponsesAPIStreamingIterator return cls( response=response, model=model, responses_api_provider_config=OpenAIResponsesAPIConfig(), logging_obj=_make_logging_obj(), - litellm_metadata={"model_info": {"id": "model_123"}}, + litellm_metadata=MappingProxyType({"model_info": MappingProxyType({"id": "model_123"})}), custom_llm_provider="openai", ) -async def _drive(events: List[Dict[str, Any]], *, sync: bool, model: str = "gpt-5") -> List[Any]: - iterator = _iterator(events, sync=sync, model=model) - collected: List[Any] = [] +async def _drive(events: Sequence[Mapping[str, object]], *, sync: bool, model: str = "gpt-5") -> tuple[Any, ...]: + iterator: Final = _iterator(events, sync=sync, model=model) if sync: - for chunk in iterator: - collected.append(chunk) - else: - async for chunk in iterator: - collected.append(chunk) - return collected + return tuple(iterator) + return tuple([chunk async for chunk in iterator]) -def _types(events: List[Any]) -> List[Any]: - return [getattr(e, "type", None) for e in events] +def _types(events: tuple[Any, ...]) -> tuple[Any, ...]: + return tuple((getattr(e, "type", None) for e in events)) -# ----- truncated upstream (the copilot / ollama / Azure case) ----- - -_TRUNCATED_TEXT_EVENTS: List[Dict[str, Any]] = [ - { - "type": "response.output_text.delta", - "item_id": "msg_1", - "output_index": 0, - "content_index": 0, - "delta": "Hello", - }, - { - "type": "response.output_text.delta", - "item_id": "msg_1", - "output_index": 0, - "content_index": 0, - "delta": " world", - }, - {"type": "response.completed", "response": _response_body("completed")}, -] - -_FULL_TEXT_EVENTS: List[Dict[str, Any]] = [ - {"type": "response.created", "response": _response_body("in_progress")}, - {"type": "response.in_progress", "response": _response_body("in_progress")}, - { - "type": "response.output_item.added", - "output_index": 0, - "item": { - "id": "msg_1", - "type": "message", - "status": "in_progress", - "role": "assistant", - "content": [], - }, - }, - { - "type": "response.content_part.added", - "item_id": "msg_1", - "output_index": 0, - "content_index": 0, - "part": {"type": "output_text", "text": "", "annotations": []}, - }, - { - "type": "response.output_text.delta", - "item_id": "msg_1", - "output_index": 0, - "content_index": 0, - "delta": "Hello world", - }, - { - "type": "response.output_text.done", - "item_id": "msg_1", - "output_index": 0, - "content_index": 0, - "text": "Hello world", - }, - { - "type": "response.content_part.done", - "item_id": "msg_1", - "output_index": 0, - "content_index": 0, - "part": {"type": "output_text", "text": "Hello world", "annotations": []}, - }, - { - "type": "response.output_item.done", - "output_index": 0, - "item": { - "id": "msg_1", - "type": "message", - "status": "completed", - "role": "assistant", - "content": [{"type": "output_text", "text": "Hello world", "annotations": []}], - }, - }, - {"type": "response.completed", "response": _response_body("completed")}, -] +_TRUNCATED_TEXT_EVENTS: Final[tuple[Mapping[str, Any], ...]] = ( + MappingProxyType( + { + "type": "response.output_text.delta", + "item_id": "msg_1", + "output_index": 0, + "content_index": 0, + "delta": "Hello", + } + ), + MappingProxyType( + { + "type": "response.output_text.delta", + "item_id": "msg_1", + "output_index": 0, + "content_index": 0, + "delta": " world", + } + ), + MappingProxyType({"type": "response.completed", "response": _response_body("completed")}), +) +_FULL_TEXT_EVENTS: Final[tuple[Mapping[str, Any], ...]] = ( + MappingProxyType({"type": "response.created", "response": _response_body("in_progress")}), + MappingProxyType({"type": "response.in_progress", "response": _response_body("in_progress")}), + MappingProxyType( + { + "type": "response.output_item.added", + "output_index": 0, + "item": MappingProxyType( + {"id": "msg_1", "type": "message", "status": "in_progress", "role": "assistant", "content": ()} + ), + } + ), + MappingProxyType( + { + "type": "response.content_part.added", + "item_id": "msg_1", + "output_index": 0, + "content_index": 0, + "part": MappingProxyType({"type": "output_text", "text": "", "annotations": ()}), + } + ), + MappingProxyType( + { + "type": "response.output_text.delta", + "item_id": "msg_1", + "output_index": 0, + "content_index": 0, + "delta": "Hello world", + } + ), + MappingProxyType( + { + "type": "response.output_text.done", + "item_id": "msg_1", + "output_index": 0, + "content_index": 0, + "text": "Hello world", + } + ), + MappingProxyType( + { + "type": "response.content_part.done", + "item_id": "msg_1", + "output_index": 0, + "content_index": 0, + "part": MappingProxyType({"type": "output_text", "text": "Hello world", "annotations": ()}), + } + ), + MappingProxyType( + { + "type": "response.output_item.done", + "output_index": 0, + "item": MappingProxyType( + { + "id": "msg_1", + "type": "message", + "status": "completed", + "role": "assistant", + "content": (MappingProxyType({"type": "output_text", "text": "Hello world", "annotations": ()}),), + } + ), + } + ), + MappingProxyType({"type": "response.completed", "response": _response_body("completed")}), +) @pytest.mark.asyncio -@pytest.mark.parametrize("sync", [False, True], ids=["async", "sync"]) +@pytest.mark.parametrize("sync", (False, True), ids=("async", "sync")) async def test_truncated_text_stream_synthesizes_full_lifecycle(sync): - collected = await _drive(_TRUNCATED_TEXT_EVENTS, sync=sync) - types = _types(collected) - - assert types == [ + collected: Final = await _drive(_TRUNCATED_TEXT_EVENTS, sync=sync) + types: Final = _types(collected) + assert types == ( E.RESPONSE_CREATED, E.RESPONSE_IN_PROGRESS, E.OUTPUT_ITEM_ADDED, @@ -753,31 +764,24 @@ async def test_truncated_text_stream_synthesizes_full_lifecycle(sync): E.CONTENT_PART_DONE, E.OUTPUT_ITEM_DONE, E.RESPONSE_COMPLETED, - ], types - - # openers must anchor to the same item_id / indices as the deltas - output_item_added = collected[2] - content_part_added = collected[3] + ), types + output_item_added: Final = collected[2] + content_part_added: Final = collected[3] assert output_item_added.item.id == "msg_1" assert content_part_added.item_id == "msg_1" assert content_part_added.output_index == 0 assert content_part_added.content_index == 0 - - # teardown text must equal the concatenation of streamed deltas - output_text_done = collected[6] + output_text_done: Final = collected[6] assert output_text_done.text == "Hello world" @pytest.mark.asyncio -@pytest.mark.parametrize("sync", [False, True], ids=["async", "sync"]) +@pytest.mark.parametrize("sync", (False, True), ids=("async", "sync")) async def test_complete_stream_passes_through_without_duplication(sync): - collected = await _drive(_FULL_TEXT_EVENTS, sync=sync) - types = _types(collected) - - # byte-for-byte: same event types, same count, nothing injected - assert types == [evt["type"] for evt in _FULL_TEXT_EVENTS], types + collected: Final = await _drive(_FULL_TEXT_EVENTS, sync=sync) + types: Final = _types(collected) + assert types == tuple((evt["type"] for evt in _FULL_TEXT_EVENTS)), types assert len(collected) == len(_FULL_TEXT_EVENTS) - # no duplicated openers assert types.count(E.RESPONSE_CREATED) == 1 assert types.count(E.OUTPUT_ITEM_ADDED) == 1 assert types.count(E.CONTENT_PART_ADDED) == 1 @@ -786,25 +790,23 @@ async def test_complete_stream_passes_through_without_duplication(sync): @pytest.mark.asyncio async def test_truncated_function_call_stream_synthesizes_item_lifecycle(): - events = [ - { - "type": "response.function_call_arguments.delta", - "item_id": "fc_1", - "output_index": 0, - "delta": '{"city":', - }, - { - "type": "response.function_call_arguments.delta", - "item_id": "fc_1", - "output_index": 0, - "delta": '"NYC"}', - }, - {"type": "response.completed", "response": _response_body("completed")}, - ] - collected = await _drive(events, sync=False) - types = _types(collected) - - assert types == [ + events: Final = ( + MappingProxyType( + { + "type": "response.function_call_arguments.delta", + "item_id": "fc_1", + "output_index": 0, + "delta": '{"city":', + } + ), + MappingProxyType( + {"type": "response.function_call_arguments.delta", "item_id": "fc_1", "output_index": 0, "delta": '"NYC"}'} + ), + MappingProxyType({"type": "response.completed", "response": _response_body("completed")}), + ) + collected: Final = await _drive(events, sync=False) + types: Final = _types(collected) + assert types == ( E.RESPONSE_CREATED, E.RESPONSE_IN_PROGRESS, E.OUTPUT_ITEM_ADDED, @@ -813,94 +815,177 @@ async def test_truncated_function_call_stream_synthesizes_item_lifecycle(): E.FUNCTION_CALL_ARGUMENTS_DONE, E.OUTPUT_ITEM_DONE, E.RESPONSE_COMPLETED, - ], types - - # function_call items have NO content part + ), types assert E.CONTENT_PART_ADDED not in types assert E.CONTENT_PART_DONE not in types - - output_item_added = collected[2] + output_item_added: Final = collected[2] assert output_item_added.item.type == "function_call" assert output_item_added.item.id == "fc_1" - - args_done = collected[5] + args_done: Final = collected[5] assert args_done.arguments == '{"city":"NYC"}' @pytest.mark.asyncio -@pytest.mark.parametrize("sync", [False, True], ids=["async", "sync"]) -@pytest.mark.parametrize("complete_reasoning", [False, True], ids=["truncated-reasoning", "complete-reasoning"]) +@pytest.mark.parametrize("sync", (False, True), ids=("async", "sync")) +@pytest.mark.parametrize("complete_reasoning", (False, True), ids=("truncated-reasoning", "complete-reasoning")) async def test_gpt_5_6_reasoning_stream_preserves_item_lifecycle(sync: bool, complete_reasoning: bool) -> None: - reasoning_events = [ - { - "type": "response.output_item.added", - "output_index": 0, - "item": {"id": "rs_1", "type": "reasoning", "summary": []}, - }, - { - "type": "response.reasoning_summary_text.delta", - "item_id": "rs_1", - "output_index": 0, - "summary_index": 0, - "delta": "Thinking", - }, - { - "type": "response.reasoning_summary_text.done", - "item_id": "rs_1", - "output_index": 0, - "summary_index": 0, - "sequence_number": 4, - "text": "Thinking", - }, - { - "type": "response.output_item.done", - "output_index": 0, - "item": { - "id": "rs_1", - "type": "reasoning", - "summary": [{"type": "summary_text", "text": "Thinking"}], - }, - }, - ] - message_events = [ - {**event, **({"output_index": 1} if "output_index" in event else {})} for event in _FULL_TEXT_EVENTS[2:] - ] - events = [ - { - **event, - **({"response": {**event["response"], "model": "gpt-5.6"}} if "response" in event else {}), - } - for event in [ - *_FULL_TEXT_EVENTS[:2], - *(reasoning_events if complete_reasoning else reasoning_events[:2]), - *message_events, - ] - ] - collected = await _drive(events, sync=sync, model="gpt-5.6") - - assert _types(collected) == [event["type"] for event in events] + reasoning_events: Final = ( + MappingProxyType( + { + "type": "response.output_item.added", + "output_index": 0, + "item": MappingProxyType({"id": "rs_1", "type": "reasoning", "summary": ()}), + } + ), + MappingProxyType( + { + "type": "response.reasoning_summary_text.delta", + "item_id": "rs_1", + "output_index": 0, + "summary_index": 0, + "delta": "Thinking", + } + ), + MappingProxyType( + { + "type": "response.reasoning_summary_text.done", + "item_id": "rs_1", + "output_index": 0, + "summary_index": 0, + "sequence_number": 4, + "text": "Thinking", + } + ), + MappingProxyType( + { + "type": "response.output_item.done", + "output_index": 0, + "item": MappingProxyType( + { + "id": "rs_1", + "type": "reasoning", + "summary": (MappingProxyType({"type": "summary_text", "text": "Thinking"}),), + } + ), + } + ), + ) + message_events: Final = tuple( + ( + MappingProxyType( + { + **event, + **(MappingProxyType({"output_index": 1}) if "output_index" in event else MappingProxyType({})), + } + ) + for event in _FULL_TEXT_EVENTS[2:] + ) + ) + events: Final = tuple( + ( + MappingProxyType( + { + **event, + **( + MappingProxyType({"response": MappingProxyType({**event["response"], "model": "gpt-5.6"})}) + if "response" in event + else MappingProxyType({}) + ), + } + ) + for event in ( + *_FULL_TEXT_EVENTS[:2], + *(reasoning_events if complete_reasoning else reasoning_events[:2]), + *message_events, + ) + ) + ) + collected: Final = await _drive(events, sync=sync, model="gpt-5.6") + expected_types: Final = tuple((event["type"] for event in events)) + assert _types(collected) == ( + expected_types if complete_reasoning else (*expected_types[:-1], E.OUTPUT_ITEM_DONE, expected_types[-1]) + ) assert collected[2].item.id == "rs_1" assert collected[2].item.type == "reasoning" assert collected[3].delta == "Thinking" if complete_reasoning: assert collected[4].text == "Thinking" assert collected[5].item.type == "reasoning" - message_start = 6 if complete_reasoning else 4 + message_start: Final = 6 if complete_reasoning else 4 assert collected[message_start].output_index == 1 assert collected[message_start].item.id == "msg_1" assert collected[message_start + 2].delta == "Hello world" assert collected[message_start + 3].text == "Hello world" assert collected[-1].response.model == "gpt-5.6" assert E.FUNCTION_CALL_ARGUMENTS_DONE not in _types(collected) + reasoning_done: Final = next( + (event for event in collected if event.type == E.OUTPUT_ITEM_DONE and event.item.id == "rs_1") + ) + assert reasoning_done.item.type == "reasoning" + assert json.loads(reasoning_done.model_dump_json())["item"]["summary"][0]["text"] == "Thinking" @pytest.mark.asyncio -@pytest.mark.parametrize("sync", [False, True], ids=["async", "sync"]) -async def test_stream_without_item_events_preserves_response_status_events(sync: bool) -> None: - events = [_FULL_TEXT_EVENTS[0], _FULL_TEXT_EVENTS[-1]] - collected = await _drive(events, sync=sync) +@pytest.mark.parametrize("sync", (False, True), ids=("async", "sync")) +@pytest.mark.parametrize("terminal_has_item", (False, True)) +@pytest.mark.parametrize("has_summary_deltas", (False, True)) +async def test_reasoning_teardown_preserves_summary_indices_and_encrypted_content( + sync: bool, terminal_has_item: bool, has_summary_deltas: bool +): + opening_item: Final = MappingProxyType( + {"id": "rs_1", "type": "reasoning", "summary": (), "encrypted_content": "opening-encrypted"} + ) + terminal_item: Final = MappingProxyType( + {**opening_item, "status": "completed", "encrypted_content": "final-encrypted"} + ) + events: Final = ( + *_FULL_TEXT_EVENTS[:2], + MappingProxyType({"type": E.OUTPUT_ITEM_ADDED, "output_index": 0, "item": opening_item}), + *( + MappingProxyType( + { + "type": "response.reasoning_summary_text.delta", + "item_id": "rs_1", + "output_index": 0, + "summary_index": index, + "delta": text, + } + ) + for index, text in ((1, "Second"), (0, "Think"), (0, "ing")) + if has_summary_deltas + ), + MappingProxyType( + { + "type": E.RESPONSE_COMPLETED, + "response": MappingProxyType( + {**_response_body("completed"), "output": (terminal_item,) if terminal_has_item else ()} + ), + } + ), + ) + collected: Final = await _drive(events, sync=sync) + done: Final = collected[-2] + wire: Final = json.loads(done.model_dump_json(exclude_none=True, exclude_unset=True)) - assert _types(collected) == [E.RESPONSE_CREATED, E.RESPONSE_COMPLETED] + assert done.type == E.OUTPUT_ITEM_DONE + assert done.output_index == 0 + assert wire["item"]["id"] == "rs_1" + assert wire["item"]["type"] == "reasoning" + assert wire["item"]["status"] == "completed" + assert wire["item"]["encrypted_content"] == ("final-encrypted" if terminal_has_item else "opening-encrypted") + assert tuple(part["text"] for part in wire["item"]["summary"]) == ( + ("Thinking", "Second") if has_summary_deltas else () + ) + assert _types(collected).count(E.OUTPUT_ITEM_DONE) == 1 + assert E.FUNCTION_CALL_ARGUMENTS_DONE not in _types(collected) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("sync", (False, True), ids=("async", "sync")) +async def test_stream_without_item_events_preserves_response_status_events(sync: bool) -> None: + events: Final = (_FULL_TEXT_EVENTS[0], _FULL_TEXT_EVENTS[-1]) + collected: Final = await _drive(events, sync=sync) + assert _types(collected) == (E.RESPONSE_CREATED, E.RESPONSE_COMPLETED) assert collected[0].response.id == collected[1].response.id @@ -911,71 +996,46 @@ async def test_synthesized_events_survive_proxy_serialization(): exclude_unset=True). Synthesized events must set their required fields explicitly so nothing load-bearing is stripped off the wire. """ - collected = await _drive(_TRUNCATED_TEXT_EVENTS, sync=False) - - required_by_type = { - E.OUTPUT_ITEM_ADDED: ["type", "output_index", "item"], - E.CONTENT_PART_ADDED: [ - "type", - "item_id", - "output_index", - "content_index", - "part", - ], - E.OUTPUT_TEXT_DONE: [ - "type", - "item_id", - "output_index", - "content_index", - "text", - ], - E.CONTENT_PART_DONE: [ - "type", - "item_id", - "output_index", - "content_index", - "part", - ], - E.OUTPUT_ITEM_DONE: ["type", "output_index", "item"], - } - - seen_types = set() - for event in collected: - etype = getattr(event, "type", None) - if etype not in required_by_type: - continue - seen_types.add(etype) - wire = json.loads(event.model_dump_json(exclude_none=True, exclude_unset=True)) - for field in required_by_type[etype]: - assert field in wire, f"{etype} lost required field {field}: {wire}" - - # all synthesized wrapper events were exercised - assert seen_types == set(required_by_type.keys()) + collected: Final = await _drive(_TRUNCATED_TEXT_EVENTS, sync=False) + required_by_type: Final = MappingProxyType( + { + E.OUTPUT_ITEM_ADDED: ("type", "output_index", "item"), + E.CONTENT_PART_ADDED: ("type", "item_id", "output_index", "content_index", "part"), + E.OUTPUT_TEXT_DONE: ("type", "item_id", "output_index", "content_index", "text"), + E.CONTENT_PART_DONE: ("type", "item_id", "output_index", "content_index", "part"), + E.OUTPUT_ITEM_DONE: ("type", "output_index", "item"), + } + ) + seen_types: Final = frozenset((event.type for event in collected if event.type in required_by_type)) + serialized: Final = tuple( + (event.type, json.loads(event.model_dump_json(exclude_none=True, exclude_unset=True))) + for event in collected + if event.type in required_by_type + ) + assert all(field in wire for event_type, wire in serialized for field in required_by_type[event_type]) + assert seen_types == frozenset(required_by_type) class _RedactingDeploymentHook: """A streaming deployment hook that redacts output_text delta content.""" - REDACTION = "[REDACTED]" + REDACTION: Final = "[REDACTED]" async def async_post_call_streaming_deployment_hook(self, *, request_data, response_chunk, call_type): if getattr(response_chunk, "type", None) == E.OUTPUT_TEXT_DELTA: - response_chunk.delta = self.REDACTION + return response_chunk.model_copy(update=MappingProxyType({"delta": self.REDACTION})) return response_chunk @pytest.fixture -def redacting_deployment_hook(): - hook = _RedactingDeploymentHook() - litellm.callbacks.append(hook) - try: - yield hook - finally: - litellm.callbacks.remove(hook) +def redacting_deployment_hook(monkeypatch): + hook: Final = _RedactingDeploymentHook() + monkeypatch.setattr(litellm, "callbacks", (*litellm.callbacks, hook)) + return hook @pytest.mark.asyncio -@pytest.mark.parametrize("sync", [False, True], ids=["async", "sync"]) +@pytest.mark.parametrize("sync", (False, True), ids=("async", "sync")) async def test_streaming_hook_governs_synthesized_teardown(sync, redacting_deployment_hook): """ A post-call streaming deployment hook that redacts response.output_text.delta @@ -984,56 +1044,47 @@ async def test_streaming_hook_governs_synthesized_teardown(sync, redacting_deplo carry the redacted text, never the raw provider text (issue #20975 review: the pre-hook accumulation leaked redacted content through the done events). """ - redacted = _RedactingDeploymentHook.REDACTION * 2 # two deltas - collected = await _drive(_TRUNCATED_TEXT_EVENTS, sync=sync) - - by_type: Dict[Any, List[Any]] = {} - for event in collected: - by_type.setdefault(getattr(event, "type", None), []).append(event) - - # client-visible deltas are redacted - assert [d.delta for d in by_type[E.OUTPUT_TEXT_DELTA]] == [ + redacted: Final = _RedactingDeploymentHook.REDACTION * 2 + collected: Final = await _drive(_TRUNCATED_TEXT_EVENTS, sync=sync) + by_type: Final = MappingProxyType( + { + event_type: tuple((event for event in collected if event.type == event_type)) + for event_type in _types(collected) + } + ) + assert tuple((d.delta for d in by_type[E.OUTPUT_TEXT_DELTA])) == ( _RedactingDeploymentHook.REDACTION, _RedactingDeploymentHook.REDACTION, - ] - - # synthesized teardown reflects the post-hook (redacted) accumulation + ) assert by_type[E.OUTPUT_TEXT_DONE][0].text == redacted assert by_type[E.CONTENT_PART_DONE][0].part.text == redacted assert by_type[E.OUTPUT_ITEM_DONE][0].item.content[0].text == redacted - - # the raw provider text never leaks anywhere in the stream - assert all(getattr(e, "text", None) != "Hello world" for e in collected) + assert all((getattr(e, "text", None) != "Hello world" for e in collected)) -# ----- truncated refusal stream ----- - -_TRUNCATED_REFUSAL_EVENTS: List[Dict[str, Any]] = [ - { - "type": "response.refusal.delta", - "item_id": "msg_r", - "output_index": 0, - "content_index": 0, - "delta": "I can", - }, - { - "type": "response.refusal.delta", - "item_id": "msg_r", - "output_index": 0, - "content_index": 0, - "delta": "not help", - }, - {"type": "response.completed", "response": _response_body("completed")}, -] +_TRUNCATED_REFUSAL_EVENTS: Final[tuple[Mapping[str, Any], ...]] = ( + MappingProxyType( + {"type": "response.refusal.delta", "item_id": "msg_r", "output_index": 0, "content_index": 0, "delta": "I can"} + ), + MappingProxyType( + { + "type": "response.refusal.delta", + "item_id": "msg_r", + "output_index": 0, + "content_index": 0, + "delta": "not help", + } + ), + MappingProxyType({"type": "response.completed", "response": _response_body("completed")}), +) @pytest.mark.asyncio -@pytest.mark.parametrize("sync", [False, True], ids=["async", "sync"]) +@pytest.mark.parametrize("sync", (False, True), ids=("async", "sync")) async def test_truncated_refusal_stream_synthesizes_lifecycle(sync): - collected = await _drive(_TRUNCATED_REFUSAL_EVENTS, sync=sync) - types = _types(collected) - - assert types == [ + collected: Final = await _drive(_TRUNCATED_REFUSAL_EVENTS, sync=sync) + types: Final = _types(collected) + assert types == ( E.RESPONSE_CREATED, E.RESPONSE_IN_PROGRESS, E.OUTPUT_ITEM_ADDED, @@ -1044,23 +1095,20 @@ async def test_truncated_refusal_stream_synthesizes_lifecycle(sync): E.CONTENT_PART_DONE, E.OUTPUT_ITEM_DONE, E.RESPONSE_COMPLETED, - ], types - - # the synthesized content part is a refusal part, not output_text + ), types assert collected[3].part.type == "refusal" - # teardown carries the accumulated refusal text at every level assert collected[6].refusal == "I cannot help" assert collected[7].part.refusal == "I cannot help" assert collected[8].item.content[0].refusal == "I cannot help" def test_obj_get_handles_dict_object_and_none(): - assert _obj_get({"a": 1}, "a") == 1 - assert _obj_get({"a": 1}, "missing", "d") == "d" + assert _obj_get(MappingProxyType({"a": 1}), "a") == 1 + assert _obj_get(MappingProxyType({"a": 1}), "missing", "d") == "d" assert _obj_get(None, "a", "d") == "d" class _Obj: - x = 5 + x: Final = 5 assert _obj_get(_Obj(), "x") == 5 assert _obj_get(_Obj(), "y", "fallback") == "fallback" @@ -1068,13 +1116,13 @@ def test_obj_get_handles_dict_object_and_none(): def test_safe_int_narrows_dynamic_values(): assert _safe_int(3, 0) == 3 - assert _safe_int(True, 9) == 9 # bool is not an accepted int + assert _safe_int(True, 9) == 9 assert _safe_int("5", 0) == 5 assert _safe_int("nope", 7) == 7 assert _safe_int(1.5, 4) == 4 def test_gap_filler_passes_unknown_event_through(): - gap_filler = _ResponsesLifecycleGapFiller(model="m", response_id="resp_x") - event = {"type": "response.some_unhandled_event"} + gap_filler: Final = _ResponsesLifecycleGapFiller(model="m", response_id="resp_x") + event: Final = MappingProxyType({"type": "response.some_unhandled_event"}) assert gap_filler.expand(event) == (event,) From e0b8c0d06644a6c30899043cab459f9a51cf88af Mon Sep 17 00:00:00 2001 From: Ali Khan Date: Tue, 8 Sep 2026 13:15:00 -0400 Subject: [PATCH 4/6] style(responses): trim lifecycle implementation commentary --- litellm/responses/streaming_iterator.py | 62 ++----------------- ...t_base_responses_api_streaming_iterator.py | 6 -- .../responses/test_streaming_iterator.py | 54 +--------------- 3 files changed, 8 insertions(+), 114 deletions(-) diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index 1eade16a520..cac503b3ee5 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -235,7 +235,6 @@ def _status_code_for_error_fields(error_type: str | None, error_code: str | None def _obj_get(obj: object, key: str, default: object | None = None) -> object: - """Read ``key`` from a dict or a pydantic/attr object uniformly.""" if obj is None: return default if isinstance(obj, Mapping): @@ -245,7 +244,6 @@ def _obj_get(obj: object, key: str, default: object | None = None) -> object: def _safe_int(value: object, default: int) -> int: - """Narrow a dynamically-read value to int, falling back for missing/malformed input.""" if isinstance(value, bool): return default if isinstance(value, int): @@ -269,32 +267,22 @@ def _build_bag( model_cls: type[_ResponseModelT], **fields: object, # kwargs-ok: generic forwarder for extra-allow Responses payload models ) -> _ResponseModelT: - """ - Construct a Responses API pydantic payload from keyword fields. - - ``BaseLiteLLMOpenAIResponseObject`` (and the loosely-typed content-part models) - accept extra fields but declare none, so direct ``Cls(id=..., type=...)`` calls - trip the type checker. Funnelling construction through this generic keeps callers - strongly typed while the ``**fields`` splat keeps the field kwargs valid. - """ return model_cls(**fields) @dataclass(frozen=True, slots=True) class _ResponsesStreamItemState: - """Per-``output_index`` lifecycle bookkeeping for one streamed output item.""" - item_id: str output_index: int content_index: int = 0 item_type: str = "message" item_snapshot: str = "{}" reasoning_summary: tuple[tuple[int, str], ...] = () - part_kind: str = "output_text" # "output_text" | "refusal" + part_kind: str = "output_text" accumulated_text: str = "" output_item_added_seen: bool = False content_part_added_seen: bool = False - leaf_done_seen: bool = False # output_text.done / refusal.done / function_call_arguments.done + leaf_done_seen: bool = False content_part_done_seen: bool = False output_item_done_seen: bool = False @@ -307,22 +295,6 @@ _ItemStateMap = Mapping[int, _ResponsesStreamItemState] class _ResponsesLifecycleGapFiller: - """ - Guarantee the Responses API streaming lifecycle wrapper events are present. - - Native providers whose upstream emits only ``response.output_text.delta`` + - ``response.completed`` (e.g. github_copilot, ollama cloud, Azure gpt-5) leave - strict clients (OpenAI Codex CLI) without an "active item", which they reject - with ``OutputTextDelta without active item``. Given the one event a provider - just produced, ``expand`` prepends any missing openers - (``response.created``/``response.in_progress`` before the first event; - ``output_item.added``/``content_part.added`` before the first delta of an - item) and, right before ``response.completed``, any missing teardown - (``*.done``). Every injection is gated on a not-already-seen flag, so a - provider that already emits the full sequence passes through unchanged and - is never double-wrapped. - """ - def __init__(self, *, model: str, response_id: str) -> None: self._model = model self._response_id = response_id @@ -331,13 +303,6 @@ class _ResponsesLifecycleGapFiller: self._items: _ItemStateMap = MappingProxyType({}) def expand(self, event: ResponsesAPIStreamingResponse) -> tuple[ResponsesAPIStreamingResponse, ...]: - """ - Given the one event a provider just produced, return the ordered events to - emit: any missing openers, then the event itself (and, for a terminal - event, any missing teardown before it). Response-level openers are tied to - the first item/content event, so a stream with no output (e.g. a lone - ``response.completed``) passes through untouched. - """ ev = ResponsesAPIStreamEvents etype = _obj_get(event, "type") @@ -405,13 +370,7 @@ class _ResponsesLifecycleGapFiller: ) def _status_event(self, *, is_created: bool) -> BaseLiteLLMOpenAIResponseObject: - # Known caveat: when these openers are synthesized (truncated upstream), the - # real response id only arrives on response.completed, so response.created / - # response.in_progress carry the placeholder _response_id and will not match - # completed's id. Clients must correlate synthesized events by output_index, - # not response.id. We do not rewrite completed's real id (clients store it for - # follow-up GETs). Providers that emit their own response.created are passed - # through untouched and keep their real id. + # Without an upstream opener, this ID is temporary; preserve the terminal ID for follow-up requests. response = _build_bag( ResponsesAPIResponse, id=self._response_id, @@ -749,10 +708,6 @@ class BaseResponsesAPIStreamingIterator: self._persist_completed_response_before_logging = True self._stream_created_time: float = time.time() - # Guarantee the Responses API streaming lifecycle wrapper events are present - # even when the upstream provider truncates them (issue #20975). Only the live - # __anext__/__next__ loops drain this; Mock/Cached iterators override the loop - # and build their own event list, so they never invoke it. self._pending_events: tuple[ResponsesAPIStreamingResponse, ...] = () self._lifecycle_gap_filler = _ResponsesLifecycleGapFiller( model=model or "", @@ -1375,8 +1330,6 @@ class ResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): try: self._check_max_streaming_duration() while True: - # Drain events the gap-filler already expanded (openers, the hooked - # provider chunk, teardown) before pulling the next SSE line. if self._pending_events: pending_event, self._pending_events = self._pending_events[0], self._pending_events[1:] self._yielded_first_chunk = True @@ -1396,13 +1349,10 @@ class ResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): raise StopAsyncIteration elif result is not None: self._maybe_raise_for_error_event(result) - # Run the deployment hook on the real chunk before the gap-filler - # accumulates it, so synthesized *.done events carry post-hook - # (e.g. guardrail-redacted) text, not the raw provider delta. + # Accumulate post-hook deltas so teardown cannot restore redacted text. self._pending_events = self._lifecycle_gap_filler.expand( await self._call_post_streaming_deployment_hook(chunk=result) ) - # Loop back to drain pending (or read the next chunk if none). except StopAsyncIteration: # Normal end of stream - don't log as failure @@ -1463,8 +1413,6 @@ class SyncResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): try: self._check_max_streaming_duration() while True: - # Drain events the gap-filler already expanded before pulling the next - # SSE line (see the async path for the hook-ordering rationale). if self._pending_events: pending_event, self._pending_events = self._pending_events[0], self._pending_events[1:] self._yielded_first_chunk = True @@ -1484,13 +1432,13 @@ class SyncResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): raise StopIteration elif result is not None: self._maybe_raise_for_error_event(result) + # Accumulate post-hook deltas so teardown cannot restore redacted text. self._pending_events = self._lifecycle_gap_filler.expand( run_async_function( async_function=self._call_post_streaming_deployment_hook, chunk=result, ) ) - # Loop back to drain pending (or read the next chunk if none). except StopIteration: # Normal end of stream - don't log as failure diff --git a/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py b/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py index f19b696b841..f3645347c91 100644 --- a/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py +++ b/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py @@ -429,9 +429,6 @@ class TestBaseResponsesAPIStreamingIterator: except StopAsyncIteration: pass # This is expected - # The provider delta is delivered as the final event. Since #20975 the live - # iterator also synthesizes the missing lifecycle wrapper events ahead of a - # bare delta, so it is no longer necessarily the only chunk. assert mock_delta_event in chunks_received assert chunks_received[-1] is mock_delta_event @@ -493,9 +490,6 @@ class TestBaseResponsesAPIStreamingIterator: except StopIteration: pass # This is expected - # The provider delta is delivered as the final event. Since #20975 the live - # iterator also synthesizes the missing lifecycle wrapper events ahead of a - # bare delta, so it is no longer necessarily the only chunk. assert mock_delta_event in chunks_received assert chunks_received[-1] is mock_delta_event diff --git a/tests/test_litellm/responses/test_streaming_iterator.py b/tests/test_litellm/responses/test_streaming_iterator.py index e9d2dbd0d64..ea162bae518 100644 --- a/tests/test_litellm/responses/test_streaming_iterator.py +++ b/tests/test_litellm/responses/test_streaming_iterator.py @@ -1,28 +1,7 @@ -"""Regression tests for litellm/responses/streaming_iterator.py. - -Two concerns live here: - -TTFT stamping (LIT-4185): /v1/responses streaming must stamp +"""Regression tests for LIT-4185 — /v1/responses streaming must stamp completion_start_time on the first chunk so downstream TTFT consumers (Prometheus, OTEL, SpendLogs completionStartTime) do not fall back to -completion_start_time = end_time. - -Lifecycle-event synthesis (issue #20975): native /responses providers whose -upstream truncates the streaming lifecycle (emitting only -response.output_text.delta ... response.completed) left strict clients like -OpenAI Codex CLI without an "active item" ("OutputTextDelta without active -item"). The live iterators must synthesize the missing setup (response.created, -response.in_progress, response.output_item.added, response.content_part.added) -and teardown (output_text.done, content_part.done, output_item.done) events, -pass an already-complete sequence through unchanged (idempotency), and run the -post-call streaming deployment hook BEFORE the gap filler accumulates deltas so -a hook that redacts delta text is not bypassed on the synthesized *.done events. - -The #20975 tests drive the REAL ResponsesAPIStreamingIterator / -SyncResponsesAPIStreamingIterator with the REAL OpenAIResponsesAPIConfig, -feeding a dependency-injected fake SSE byte stream (no monkeypatching of the -code under test). -""" +completion_start_time = end_time.""" import json from collections.abc import Mapping, Sequence @@ -52,12 +31,7 @@ from litellm.types.llms.openai import ( ) EV = ResponsesAPIStreamEvents -E = EV # shorthand - - -# --------------------------------------------------------------------------- -# TTFT stamping (LIT-4185) -# --------------------------------------------------------------------------- +E = EV def _sse_event(payload: dict) -> bytes: @@ -575,11 +549,6 @@ async def test_streaming_logging_copy_fallback_leaves_caller_event_untouched(): assert iterator.completed_response.response._hidden_params == {} -# --------------------------------------------------------------------------- -# Lifecycle-event synthesis (issue #20975) -# --------------------------------------------------------------------------- - - def _response_body(status: str) -> Mapping[str, Any]: return MappingProxyType( { @@ -609,13 +578,10 @@ def _response_body(status: str) -> Mapping[str, Any]: def _sse_frames(events: Sequence[Mapping[str, object]]) -> tuple[bytes, ...]: - """One `data: {...}\\n\\n` SSE frame per event, plus a terminating [DONE].""" return (*(f"data: {json.dumps(evt, default=dict)}\n\n".encode("utf-8") for evt in events), b"data: [DONE]\n\n") class _FakeStreamResponse: - """Minimal stand-in for httpx.Response exposing (a)iter_bytes over fixed frames.""" - def __init__(self, frames: tuple[bytes, ...]): self.headers: Mapping[str, str] = MappingProxyType({}) self._frames = frames @@ -991,11 +957,6 @@ async def test_stream_without_item_events_preserves_response_status_events(sync: @pytest.mark.asyncio async def test_synthesized_events_survive_proxy_serialization(): - """ - The proxy serializes each event with model_dump_json(exclude_none=True, - exclude_unset=True). Synthesized events must set their required fields - explicitly so nothing load-bearing is stripped off the wire. - """ collected: Final = await _drive(_TRUNCATED_TEXT_EVENTS, sync=False) required_by_type: Final = MappingProxyType( { @@ -1017,8 +978,6 @@ async def test_synthesized_events_survive_proxy_serialization(): class _RedactingDeploymentHook: - """A streaming deployment hook that redacts output_text delta content.""" - REDACTION: Final = "[REDACTED]" async def async_post_call_streaming_deployment_hook(self, *, request_data, response_chunk, call_type): @@ -1037,13 +996,6 @@ def redacting_deployment_hook(monkeypatch): @pytest.mark.asyncio @pytest.mark.parametrize("sync", (False, True), ids=("async", "sync")) async def test_streaming_hook_governs_synthesized_teardown(sync, redacting_deployment_hook): - """ - A post-call streaming deployment hook that redacts response.output_text.delta - must also govern the SYNTHESIZED teardown. The gap filler accumulates the - post-hook delta, so output_text.done / content_part.done / output_item.done - carry the redacted text, never the raw provider text (issue #20975 review: - the pre-hook accumulation leaked redacted content through the done events). - """ redacted: Final = _RedactingDeploymentHook.REDACTION * 2 collected: Final = await _drive(_TRUNCATED_TEXT_EVENTS, sync=sync) by_type: Final = MappingProxyType( From d0811cf0dffbb60bb700befe7a01eeebf7c3e6eb Mon Sep 17 00:00:00 2001 From: Ali Khan Date: Tue, 8 Sep 2026 13:22:58 -0400 Subject: [PATCH 5/6] test(responses): use concrete streaming helper types --- .../responses/test_streaming_iterator.py | 89 ++++++++++--------- 1 file changed, 48 insertions(+), 41 deletions(-) diff --git a/tests/test_litellm/responses/test_streaming_iterator.py b/tests/test_litellm/responses/test_streaming_iterator.py index ea162bae518..267339a330e 100644 --- a/tests/test_litellm/responses/test_streaming_iterator.py +++ b/tests/test_litellm/responses/test_streaming_iterator.py @@ -7,7 +7,7 @@ import json from collections.abc import Mapping, Sequence from datetime import datetime from types import MappingProxyType -from typing import Any, Dict, Final, List, Optional +from typing import Final, Optional from unittest.mock import Mock, patch import httpx @@ -23,12 +23,15 @@ from litellm.responses.streaming_iterator import ( _obj_get, _ResponsesLifecycleGapFiller, _safe_int, + _safe_str, ) from litellm.types.llms.openai import ( ResponseCompletedEvent, ResponsesAPIResponse, ResponsesAPIStreamEvents, + ResponsesAPIStreamingResponse, ) +from litellm.types.utils import CallTypes EV = ResponsesAPIStreamEvents E = EV @@ -549,14 +552,14 @@ async def test_streaming_logging_copy_fallback_leaves_caller_event_untouched(): assert iterator.completed_response.response._hidden_params == {} -def _response_body(status: str) -> Mapping[str, Any]: +def _response_body(status: str, *, model: str = "gpt-5") -> Mapping[str, object]: return MappingProxyType( { "id": "resp_real_upstream", "object": "response", "created_at": 1700000000, "status": status, - "model": "gpt-5", + "model": model, "output": ( MappingProxyType( { @@ -581,52 +584,41 @@ def _sse_frames(events: Sequence[Mapping[str, object]]) -> tuple[bytes, ...]: return (*(f"data: {json.dumps(evt, default=dict)}\n\n".encode("utf-8") for evt in events), b"data: [DONE]\n\n") -class _FakeStreamResponse: - def __init__(self, frames: tuple[bytes, ...]): - self.headers: Mapping[str, str] = MappingProxyType({}) - self._frames = frames - - async def aiter_bytes(self): - for frame in self._frames: - yield frame - - def iter_bytes(self): - for frame in self._frames: - yield frame - - -def _make_logging_obj() -> Any: +def _make_logging_obj() -> Mock: logging_obj: Final = Mock(spec=LiteLLMLoggingObj) logging_obj.model_call_details = MappingProxyType({"litellm_params": MappingProxyType({})}) logging_obj.completion_start_time = None return logging_obj -def _iterator(events: tuple[Mapping[str, Any], ...], *, sync: bool, model: str = "gpt-5") -> Any: - response: Final = _FakeStreamResponse(_sse_frames(events)) +def _iterator( + events: Sequence[Mapping[str, object]], *, sync: bool, model: str = "gpt-5" +) -> ResponsesAPIStreamingIterator | SyncResponsesAPIStreamingIterator: + response: Final = httpx.Response(200, content=b"".join(_sse_frames(events))) cls: Final = SyncResponsesAPIStreamingIterator if sync else ResponsesAPIStreamingIterator return cls( response=response, model=model, responses_api_provider_config=OpenAIResponsesAPIConfig(), logging_obj=_make_logging_obj(), - litellm_metadata=MappingProxyType({"model_info": MappingProxyType({"id": "model_123"})}), custom_llm_provider="openai", ) -async def _drive(events: Sequence[Mapping[str, object]], *, sync: bool, model: str = "gpt-5") -> tuple[Any, ...]: +async def _drive( + events: Sequence[Mapping[str, object]], *, sync: bool, model: str = "gpt-5" +) -> tuple[ResponsesAPIStreamingResponse, ...]: iterator: Final = _iterator(events, sync=sync, model=model) - if sync: + if isinstance(iterator, SyncResponsesAPIStreamingIterator): return tuple(iterator) return tuple([chunk async for chunk in iterator]) -def _types(events: tuple[Any, ...]) -> tuple[Any, ...]: - return tuple((getattr(e, "type", None) for e in events)) +def _types(events: Sequence[ResponsesAPIStreamingResponse]) -> tuple[str, ...]: + return tuple(_safe_str(_obj_get(event, "type"), "") for event in events) -_TRUNCATED_TEXT_EVENTS: Final[tuple[Mapping[str, Any], ...]] = ( +_TRUNCATED_TEXT_EVENTS: Final[tuple[Mapping[str, object], ...]] = ( MappingProxyType( { "type": "response.output_text.delta", @@ -647,7 +639,7 @@ _TRUNCATED_TEXT_EVENTS: Final[tuple[Mapping[str, Any], ...]] = ( ), MappingProxyType({"type": "response.completed", "response": _response_body("completed")}), ) -_FULL_TEXT_EVENTS: Final[tuple[Mapping[str, Any], ...]] = ( +_FULL_TEXT_EVENTS: Final[tuple[Mapping[str, object], ...]] = ( MappingProxyType({"type": "response.created", "response": _response_body("in_progress")}), MappingProxyType({"type": "response.in_progress", "response": _response_body("in_progress")}), MappingProxyType( @@ -716,7 +708,7 @@ _FULL_TEXT_EVENTS: Final[tuple[Mapping[str, Any], ...]] = ( @pytest.mark.asyncio @pytest.mark.parametrize("sync", (False, True), ids=("async", "sync")) -async def test_truncated_text_stream_synthesizes_full_lifecycle(sync): +async def test_truncated_text_stream_synthesizes_full_lifecycle(sync: bool) -> None: collected: Final = await _drive(_TRUNCATED_TEXT_EVENTS, sync=sync) types: Final = _types(collected) assert types == ( @@ -743,7 +735,7 @@ async def test_truncated_text_stream_synthesizes_full_lifecycle(sync): @pytest.mark.asyncio @pytest.mark.parametrize("sync", (False, True), ids=("async", "sync")) -async def test_complete_stream_passes_through_without_duplication(sync): +async def test_complete_stream_passes_through_without_duplication(sync: bool) -> None: collected: Final = await _drive(_FULL_TEXT_EVENTS, sync=sync) types: Final = _types(collected) assert types == tuple((evt["type"] for evt in _FULL_TEXT_EVENTS)), types @@ -755,7 +747,7 @@ async def test_complete_stream_passes_through_without_duplication(sync): @pytest.mark.asyncio -async def test_truncated_function_call_stream_synthesizes_item_lifecycle(): +async def test_truncated_function_call_stream_synthesizes_item_lifecycle() -> None: events: Final = ( MappingProxyType( { @@ -853,7 +845,14 @@ async def test_gpt_5_6_reasoning_stream_preserves_item_lifecycle(sync: bool, com { **event, **( - MappingProxyType({"response": MappingProxyType({**event["response"], "model": "gpt-5.6"})}) + MappingProxyType( + { + "response": _response_body( + "completed" if event["type"] == E.RESPONSE_COMPLETED else "in_progress", + model="gpt-5.6", + ) + } + ) if "response" in event else MappingProxyType({}) ), @@ -897,7 +896,7 @@ async def test_gpt_5_6_reasoning_stream_preserves_item_lifecycle(sync: bool, com @pytest.mark.parametrize("has_summary_deltas", (False, True)) async def test_reasoning_teardown_preserves_summary_indices_and_encrypted_content( sync: bool, terminal_has_item: bool, has_summary_deltas: bool -): +) -> None: opening_item: Final = MappingProxyType( {"id": "rs_1", "type": "reasoning", "summary": (), "encrypted_content": "opening-encrypted"} ) @@ -956,7 +955,7 @@ async def test_stream_without_item_events_preserves_response_status_events(sync: @pytest.mark.asyncio -async def test_synthesized_events_survive_proxy_serialization(): +async def test_synthesized_events_survive_proxy_serialization() -> None: collected: Final = await _drive(_TRUNCATED_TEXT_EVENTS, sync=False) required_by_type: Final = MappingProxyType( { @@ -980,14 +979,20 @@ async def test_synthesized_events_survive_proxy_serialization(): class _RedactingDeploymentHook: REDACTION: Final = "[REDACTED]" - async def async_post_call_streaming_deployment_hook(self, *, request_data, response_chunk, call_type): + async def async_post_call_streaming_deployment_hook( + self, + *, + request_data: Mapping[str, object], + response_chunk: ResponsesAPIStreamingResponse, + call_type: CallTypes | None, + ) -> ResponsesAPIStreamingResponse: if getattr(response_chunk, "type", None) == E.OUTPUT_TEXT_DELTA: return response_chunk.model_copy(update=MappingProxyType({"delta": self.REDACTION})) return response_chunk @pytest.fixture -def redacting_deployment_hook(monkeypatch): +def redacting_deployment_hook(monkeypatch: pytest.MonkeyPatch) -> _RedactingDeploymentHook: hook: Final = _RedactingDeploymentHook() monkeypatch.setattr(litellm, "callbacks", (*litellm.callbacks, hook)) return hook @@ -995,7 +1000,9 @@ def redacting_deployment_hook(monkeypatch): @pytest.mark.asyncio @pytest.mark.parametrize("sync", (False, True), ids=("async", "sync")) -async def test_streaming_hook_governs_synthesized_teardown(sync, redacting_deployment_hook): +async def test_streaming_hook_governs_synthesized_teardown( + sync: bool, redacting_deployment_hook: _RedactingDeploymentHook +) -> None: redacted: Final = _RedactingDeploymentHook.REDACTION * 2 collected: Final = await _drive(_TRUNCATED_TEXT_EVENTS, sync=sync) by_type: Final = MappingProxyType( @@ -1014,7 +1021,7 @@ async def test_streaming_hook_governs_synthesized_teardown(sync, redacting_deplo assert all((getattr(e, "text", None) != "Hello world" for e in collected)) -_TRUNCATED_REFUSAL_EVENTS: Final[tuple[Mapping[str, Any], ...]] = ( +_TRUNCATED_REFUSAL_EVENTS: Final[tuple[Mapping[str, object], ...]] = ( MappingProxyType( {"type": "response.refusal.delta", "item_id": "msg_r", "output_index": 0, "content_index": 0, "delta": "I can"} ), @@ -1033,7 +1040,7 @@ _TRUNCATED_REFUSAL_EVENTS: Final[tuple[Mapping[str, Any], ...]] = ( @pytest.mark.asyncio @pytest.mark.parametrize("sync", (False, True), ids=("async", "sync")) -async def test_truncated_refusal_stream_synthesizes_lifecycle(sync): +async def test_truncated_refusal_stream_synthesizes_lifecycle(sync: bool) -> None: collected: Final = await _drive(_TRUNCATED_REFUSAL_EVENTS, sync=sync) types: Final = _types(collected) assert types == ( @@ -1054,7 +1061,7 @@ async def test_truncated_refusal_stream_synthesizes_lifecycle(sync): assert collected[8].item.content[0].refusal == "I cannot help" -def test_obj_get_handles_dict_object_and_none(): +def test_obj_get_handles_dict_object_and_none() -> None: assert _obj_get(MappingProxyType({"a": 1}), "a") == 1 assert _obj_get(MappingProxyType({"a": 1}), "missing", "d") == "d" assert _obj_get(None, "a", "d") == "d" @@ -1066,7 +1073,7 @@ def test_obj_get_handles_dict_object_and_none(): assert _obj_get(_Obj(), "y", "fallback") == "fallback" -def test_safe_int_narrows_dynamic_values(): +def test_safe_int_narrows_dynamic_values() -> None: assert _safe_int(3, 0) == 3 assert _safe_int(True, 9) == 9 assert _safe_int("5", 0) == 5 @@ -1074,7 +1081,7 @@ def test_safe_int_narrows_dynamic_values(): assert _safe_int(1.5, 4) == 4 -def test_gap_filler_passes_unknown_event_through(): +def test_gap_filler_passes_unknown_event_through() -> None: gap_filler: Final = _ResponsesLifecycleGapFiller(model="m", response_id="resp_x") event: Final = MappingProxyType({"type": "response.some_unhandled_event"}) assert gap_filler.expand(event) == (event,) From 40fef3cedb5efb03a9bd6fea874feb0a56bf2702 Mon Sep 17 00:00:00 2001 From: Ali Khan Date: Tue, 8 Sep 2026 13:33:55 -0400 Subject: [PATCH 6/6] test(responses): cover reasoning events without tracked items --- .../responses/test_streaming_iterator.py | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/tests/test_litellm/responses/test_streaming_iterator.py b/tests/test_litellm/responses/test_streaming_iterator.py index 267339a330e..d248d65480e 100644 --- a/tests/test_litellm/responses/test_streaming_iterator.py +++ b/tests/test_litellm/responses/test_streaming_iterator.py @@ -26,6 +26,7 @@ from litellm.responses.streaming_iterator import ( _safe_str, ) from litellm.types.llms.openai import ( + BaseLiteLLMOpenAIResponseObject, ResponseCompletedEvent, ResponsesAPIResponse, ResponsesAPIStreamEvents, @@ -1081,7 +1082,18 @@ def test_safe_int_narrows_dynamic_values() -> None: assert _safe_int(1.5, 4) == 4 -def test_gap_filler_passes_unknown_event_through() -> None: +@pytest.mark.parametrize("event_type", ("response.some_unhandled_event", "response.reasoning_summary_text.delta")) +def test_gap_filler_passes_events_without_items_through(event_type: str) -> None: gap_filler: Final = _ResponsesLifecycleGapFiller(model="m", response_id="resp_x") - event: Final = MappingProxyType({"type": "response.some_unhandled_event"}) + event: Final = BaseLiteLLMOpenAIResponseObject.model_validate( + MappingProxyType( + { + "type": event_type, + "item_id": "rs_orphan", + "output_index": 0, + "summary_index": 0, + "delta": "Thinking", + } + ) + ) assert gap_filler.expand(event) == (event,)