From 1100b2568bf96ba2a7d0b822c6a7a94e77f02b41 Mon Sep 17 00:00:00 2001 From: Ali Khan Date: Mon, 6 Jul 2026 22:36:50 -0400 Subject: [PATCH] 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"