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

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

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

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,)

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"