diff --git a/.circleci/config.yml b/.circleci/config.yml index 32983c92e2..3139bd3cb2 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -158,6 +158,8 @@ jobs: CHOCOLATEY_CONFIRM_ALL: "true" - run: name: Install Dependencies + environment: + UV_HTTP_TIMEOUT: "300" command: | $installer = Join-Path $env:TEMP "uv-install.ps1" Invoke-WebRequest -Uri https://astral.sh/uv/0.10.9/install.ps1 -OutFile $installer diff --git a/litellm/interactions/litellm_responses_transformation/streaming_iterator.py b/litellm/interactions/litellm_responses_transformation/streaming_iterator.py index 90f9517e8b..4a3eb63084 100644 --- a/litellm/interactions/litellm_responses_transformation/streaming_iterator.py +++ b/litellm/interactions/litellm_responses_transformation/streaming_iterator.py @@ -2,7 +2,17 @@ Streaming iterator for transforming Responses API stream to Interactions API stream. """ -from typing import Any, AsyncIterator, Dict, Iterator, Optional, cast +from collections import deque +from typing import ( + Any, + AsyncIterator, + Deque, + Dict, + Iterator, + List, + Optional, + cast, +) from litellm.responses.streaming_iterator import ( BaseResponsesAPIStreamingIterator, @@ -33,9 +43,9 @@ class LiteLLMResponsesInteractionsStreamingIterator: Schema selection: - New schema (default, use_legacy_interactions_schema=False): - interaction.created → step.start → step.delta … → step.stop → interaction.completed + interaction.created -> step.start -> step.delta ... -> step.stop -> interaction.completed - Legacy schema (use_legacy_interactions_schema=True, remove after June 8 2026): - interaction.start → content.start → content.delta … → content.stop → interaction.complete + interaction.start -> content.start -> content.delta ... -> content.stop -> interaction.complete """ def __init__( @@ -63,86 +73,152 @@ class LiteLLMResponsesInteractionsStreamingIterator: # emitted by this stream use a consistent schema, even if the global # flag is mutated mid-stream (e.g. by a config reload). self._use_legacy: bool = litellm.use_legacy_interactions_schema + # Buffer of events that have been derived from upstream chunks but not + # yet returned to the caller. A single Responses API chunk may expand + # into multiple Interactions API events (e.g. the first text delta + # produces interaction.created + step.start + step.delta), and the + # terminal sequence on stream end may also span multiple events + # (step.stop + interaction.completed). + self._pending_events: Deque[InteractionsAPIStreamingResponse] = deque() + # Tracks whether we've already emitted a terminal completion event so + # the StopIteration fallback path doesn't double-emit. + self._sent_completion_event = False + # ID resolved from the first upstream chunk (item_id on a text delta or + # response.id on response.created). Persisted so the EOF terminal + # events stay correlated with the start events delivered earlier. + self._interaction_id: Optional[str] = None - def _transform_responses_chunk_to_interactions_chunk( - self, - responses_chunk: ResponsesAPIStreamingResponse, - ) -> Optional[InteractionsAPIStreamingResponse]: + # ------------------------------------------------------------------ + # Event builders + # ------------------------------------------------------------------ + + def _build_interaction_start_event( + self, interaction_id: str + ) -> InteractionsAPIStreamingResponse: + event_type = "interaction.start" if self._use_legacy else "interaction.created" + return InteractionsAPIStreamingResponse( + event_type=event_type, + id=interaction_id, + object="interaction", + status="in_progress", + model=self.model, + ) + + def _build_content_start_event( + self, interaction_id: str + ) -> InteractionsAPIStreamingResponse: + if self._use_legacy: + return InteractionsAPIStreamingResponse( + event_type="content.start", + id=interaction_id, + object="content", + delta={"type": "text", "text": ""}, + ) + return InteractionsAPIStreamingResponse( + event_type="step.start", + index=0, + step={"type": "model_output", "content": []}, + ) + + def _build_text_delta_event( + self, interaction_id: str, delta_text: str + ) -> InteractionsAPIStreamingResponse: + if self._use_legacy: + return InteractionsAPIStreamingResponse( + event_type="content.delta", + id=interaction_id, + object="content", + delta={"type": "text", "text": delta_text}, + ) + return InteractionsAPIStreamingResponse( + event_type="step.delta", + index=0, + delta={"type": "text", "text": delta_text}, + ) + + def _build_content_stop_event( + self, interaction_id: Optional[str] + ) -> InteractionsAPIStreamingResponse: + if self._use_legacy: + return InteractionsAPIStreamingResponse( + event_type="content.stop", + id=interaction_id, + object="content", + delta={"type": "text", "text": self.collected_text}, + ) + return InteractionsAPIStreamingResponse( + event_type="step.stop", + index=0, + ) + + def _build_completion_event( + self, response_id: str + ) -> InteractionsAPIStreamingResponse: + if self._use_legacy: + return InteractionsAPIStreamingResponse( + event_type="interaction.complete", + id=response_id, + object="interaction", + status="completed", + model=self.model, + outputs=[{"type": "text", "text": self.collected_text}], + ) + return InteractionsAPIStreamingResponse( + event_type="interaction.completed", + id=response_id, + object="interaction", + status="completed", + model=self.model, + steps=[ + { + "type": "model_output", + "content": [{"type": "text", "text": self.collected_text}], + } + ], + ) + + # ------------------------------------------------------------------ + # Per-chunk transform (returns a list of events to enqueue) + # ------------------------------------------------------------------ + + def _events_for_chunk( + self, responses_chunk: ResponsesAPIStreamingResponse + ) -> List[InteractionsAPIStreamingResponse]: """ - Transform a Responses API streaming chunk to an Interactions API streaming chunk. + Translate a single upstream Responses API chunk into the list of + Interactions API events it should produce. - Emits new-schema events by default; falls back to legacy events when - ``litellm.use_legacy_interactions_schema`` is True. - Remove legacy branch after June 8, 2026. + Returning a list (rather than a single event) lets a chunk emit any + synthetic start events that haven't been sent yet *together with* the + actual delta event, so we never silently drop the chunk's payload. """ if not responses_chunk: - return None + return [] - use_legacy = self._use_legacy - - # Handle OutputTextDeltaEvent + # Text delta: emit any missing start events, then the delta itself. if isinstance(responses_chunk, OutputTextDeltaEvent): delta_text = ( responses_chunk.delta if isinstance(responses_chunk.delta, str) else "" ) self.collected_text += delta_text - item_id = ( + interaction_id = ( getattr(responses_chunk, "item_id", None) or f"interaction_{id(self)}" ) + if self._interaction_id is None: + self._interaction_id = interaction_id - # Send the "interaction started" event on the first delta + events: List[InteractionsAPIStreamingResponse] = [] if not self.sent_interaction_start: self.sent_interaction_start = True - if use_legacy: - return InteractionsAPIStreamingResponse( - event_type="interaction.start", - id=item_id, - object="interaction", - status="in_progress", - model=self.model, - ) - else: - return InteractionsAPIStreamingResponse( - event_type="interaction.created", - id=item_id, - object="interaction", - status="in_progress", - model=self.model, - ) - - # Send the "content/step started" event on the second delta + events.append(self._build_interaction_start_event(interaction_id)) if not self.sent_content_start: self.sent_content_start = True - if use_legacy: - return InteractionsAPIStreamingResponse( - event_type="content.start", - id=item_id, - object="content", - delta={"type": "text", "text": ""}, - ) - else: - return InteractionsAPIStreamingResponse( - event_type="step.start", - index=0, - step={"type": "model_output", "content": []}, - ) + events.append(self._build_content_start_event(interaction_id)) + events.append(self._build_text_delta_event(interaction_id, delta_text)) + return events - # Emit the delta itself - if use_legacy: - return InteractionsAPIStreamingResponse( - event_type="content.delta", - id=item_id, - object="content", - delta={"type": "text", "text": delta_text}, - ) - else: - return InteractionsAPIStreamingResponse( - event_type="step.delta", - index=0, - delta={"type": "text", "text": delta_text}, - ) - - # Handle ResponseCreatedEvent or ResponseInProgressEvent + # Response created / in-progress: synthesize interaction start if we + # haven't already sent one. if isinstance(responses_chunk, (ResponseCreatedEvent, ResponseInProgressEvent)): if not self.sent_interaction_start: self.sent_interaction_start = True @@ -151,224 +227,135 @@ class LiteLLMResponsesInteractionsStreamingIterator: if hasattr(responses_chunk, "response") else None ) or f"interaction_{id(self)}" - event_type = ( - "interaction.start" if use_legacy else "interaction.created" - ) - return InteractionsAPIStreamingResponse( - event_type=event_type, - id=response_id, - object="interaction", - status="in_progress", - model=self.model, - ) + if self._interaction_id is None: + self._interaction_id = response_id + return [self._build_interaction_start_event(response_id)] + return [] - # Handle ResponseCompletedEvent + # Response completed: emit step.stop (if content was started) followed + # by the terminal completion event. Prefer the interaction id already + # established by earlier events so consumers can correlate the start + # and completion events by id (response.id may differ from the item_id + # used to derive the initial id when the stream starts directly with a + # text delta). if isinstance(responses_chunk, ResponseCompletedEvent): self.finished = True response = responses_chunk.response - response_id = getattr(response, "id", None) or f"interaction_{id(self)}" + response_id = ( + self._interaction_id + or getattr(response, "id", None) + or f"interaction_{id(self)}" + ) - if use_legacy: - return InteractionsAPIStreamingResponse( - event_type="interaction.complete", - id=response_id, - object="interaction", - status="completed", - model=self.model, - outputs=[{"type": "text", "text": self.collected_text}], - ) - else: - return InteractionsAPIStreamingResponse( - event_type="interaction.completed", - id=response_id, - object="interaction", - status="completed", - model=self.model, - steps=[ - { - "type": "model_output", - "content": [{"type": "text", "text": self.collected_text}], - } - ], - ) + terminal: List[InteractionsAPIStreamingResponse] = [] + if self.sent_content_start: + terminal.append(self._build_content_stop_event(response_id)) + terminal.append(self._build_completion_event(response_id)) + self._sent_completion_event = True + return terminal - # For other event types, return None (skip) - return None + return [] + + def _build_terminal_events_on_eof( + self, + ) -> List[InteractionsAPIStreamingResponse]: + """ + Build the events to flush when the upstream stream ends without a + ResponseCompletedEvent. Ensures consumers always observe a terminal + interaction.completed/interaction.complete carrying the full text. + """ + if self._sent_completion_event: + return [] + + fallback_id = self._interaction_id or f"interaction_{id(self)}" + terminal: List[InteractionsAPIStreamingResponse] = [] + if self.sent_content_start: + terminal.append(self._build_content_stop_event(fallback_id)) + if self.sent_interaction_start or self.collected_text: + terminal.append(self._build_completion_event(fallback_id)) + self._sent_completion_event = True + return terminal + + # ------------------------------------------------------------------ + # Iteration + # ------------------------------------------------------------------ def __iter__(self) -> Iterator[InteractionsAPIStreamingResponse]: - """Sync iterator implementation.""" return self def __next__(self) -> InteractionsAPIStreamingResponse: - """Get next chunk in sync mode.""" - # Check for a pending interaction.complete/completed event BEFORE the - # finished check — otherwise the buffered completion event (which - # carries the full text) would be dropped after `self.finished` is set. - if hasattr(self, "_pending_interaction_complete"): - pending: InteractionsAPIStreamingResponse = getattr( - self, "_pending_interaction_complete" - ) - delattr(self, "_pending_interaction_complete") - return pending + if self._pending_events: + return self._pending_events.popleft() if self.finished: raise StopIteration - # Use a loop instead of recursion to avoid stack overflow sync_iterator = cast( SyncResponsesAPIStreamingIterator, self.responses_stream_iterator ) while True: try: - # Get next chunk from responses API stream chunk = next(sync_iterator) - - # Transform chunk (chunk is already a ResponsesAPIStreamingResponse) - transformed = self._transform_responses_chunk_to_interactions_chunk( - chunk - ) - - if transformed: - completion_event_type = ( - "interaction.complete" - if self._use_legacy - else "interaction.completed" - ) - stop_event_type = ( - "content.stop" if self._use_legacy else "step.stop" - ) - # If content was started, send the stop event before the completion event. - if ( - self.finished - and self.sent_content_start - and transformed.event_type == completion_event_type - ): - stop_kwargs: Dict[str, Any] = { - "event_type": stop_event_type, - "index": 0, - } - if self._use_legacy: - stop_kwargs["id"] = transformed.id - stop_kwargs["object"] = "content" - stop_kwargs["delta"] = { - "type": "text", - "text": self.collected_text, - } - stop_chunk = InteractionsAPIStreamingResponse(**stop_kwargs) - self._pending_interaction_complete = transformed - return stop_chunk - return transformed - - # If no transformation, continue to next chunk (loop continues) - except StopIteration: self.finished = True + self._pending_events.extend(self._build_terminal_events_on_eof()) + if self._pending_events: + return self._pending_events.popleft() + raise - # Send final stop event if content was started - if self.sent_content_start: - stop_event_type = ( - "content.stop" if self._use_legacy else "step.stop" - ) - stop_kwargs = { - "event_type": stop_event_type, - "index": 0, - } - if self._use_legacy: - stop_kwargs["object"] = "content" - stop_kwargs["delta"] = { - "type": "text", - "text": self.collected_text, - } - return InteractionsAPIStreamingResponse(**stop_kwargs) - - raise StopIteration + events = self._events_for_chunk(chunk) + if events: + self._pending_events.extend(events) + return self._pending_events.popleft() def __aiter__(self) -> AsyncIterator[InteractionsAPIStreamingResponse]: - """Async iterator implementation.""" return self async def __anext__(self) -> InteractionsAPIStreamingResponse: - """Get next chunk in async mode.""" - # Check for a pending interaction.complete/completed event BEFORE the - # finished check — otherwise the buffered completion event (which - # carries the full text) would be dropped after `self.finished` is set. - if hasattr(self, "_pending_interaction_complete"): - pending: InteractionsAPIStreamingResponse = getattr( - self, "_pending_interaction_complete" - ) - delattr(self, "_pending_interaction_complete") - return pending + if self._pending_events: + return self._pending_events.popleft() if self.finished: raise StopAsyncIteration - # Use a loop instead of recursion to avoid stack overflow async_iterator = cast( ResponsesAPIStreamingIterator, self.responses_stream_iterator ) while True: try: - # Get next chunk from responses API stream chunk = await async_iterator.__anext__() - - # Transform chunk (chunk is already a ResponsesAPIStreamingResponse) - transformed = self._transform_responses_chunk_to_interactions_chunk( - chunk - ) - - if transformed: - completion_event_type = ( - "interaction.complete" - if self._use_legacy - else "interaction.completed" - ) - stop_event_type = ( - "content.stop" if self._use_legacy else "step.stop" - ) - # If content was started, send the stop event before the completion event. - if ( - self.finished - and self.sent_content_start - and transformed.event_type == completion_event_type - ): - stop_kwargs_async: Dict[str, Any] = { - "event_type": stop_event_type, - "index": 0, - } - if self._use_legacy: - stop_kwargs_async["id"] = transformed.id - stop_kwargs_async["object"] = "content" - stop_kwargs_async["delta"] = { - "type": "text", - "text": self.collected_text, - } - stop_chunk = InteractionsAPIStreamingResponse( - **stop_kwargs_async - ) - self._pending_interaction_complete = transformed - return stop_chunk - return transformed - - # If no transformation, continue to next chunk (loop continues) - except StopAsyncIteration: self.finished = True + self._pending_events.extend(self._build_terminal_events_on_eof()) + if self._pending_events: + return self._pending_events.popleft() + raise - # Send final stop event if content was started - if self.sent_content_start: - stop_event_type = ( - "content.stop" if self._use_legacy else "step.stop" - ) - stop_kwargs_async = { - "event_type": stop_event_type, - "index": 0, - } - if self._use_legacy: - stop_kwargs_async["object"] = "content" - stop_kwargs_async["delta"] = { - "type": "text", - "text": self.collected_text, - } - return InteractionsAPIStreamingResponse(**stop_kwargs_async) + events = self._events_for_chunk(chunk) + if events: + self._pending_events.extend(events) + return self._pending_events.popleft() - raise StopAsyncIteration + # ------------------------------------------------------------------ + # Backwards-compatible single-chunk transform (used by tests and any + # external callers that drove the iterator chunk-by-chunk pre-fix). + # ------------------------------------------------------------------ + + def _transform_responses_chunk_to_interactions_chunk( + self, + responses_chunk: ResponsesAPIStreamingResponse, + ) -> Optional[InteractionsAPIStreamingResponse]: + """ + Compatibility shim: returns the *first* event produced for this chunk + and queues any remaining events on ``self._pending_events`` so they + are surfaced on subsequent calls/iterations. + + Prefer ``_events_for_chunk`` in new code. + """ + events = self._events_for_chunk(responses_chunk) + if not events: + return None + first = events[0] + if len(events) > 1: + self._pending_events.extend(events[1:]) + return first diff --git a/tests/test_litellm/interactions/test_gemini_interactions_transformation.py b/tests/test_litellm/interactions/test_gemini_interactions_transformation.py index b4575ceb9b..524589abf5 100644 --- a/tests/test_litellm/interactions/test_gemini_interactions_transformation.py +++ b/tests/test_litellm/interactions/test_gemini_interactions_transformation.py @@ -25,6 +25,7 @@ from litellm.llms.gemini.interactions.transformation import ( ) from litellm.types.llms.openai import ( OutputTextDeltaEvent, + ResponseCompletedEvent, ResponseCreatedEvent, ) from litellm.types.router import GenericLiteLLMParams @@ -151,7 +152,12 @@ class TestTransformRequest: request_body = config.transform_request( model=None, agent="my-custom-slides-agent", - input=[{"type": "text", "text": "Create a 5-slide presentation about AI trends."}], + input=[ + { + "type": "text", + "text": "Create a 5-slide presentation about AI trends.", + } + ], optional_params={ "environment": "remote", "stream": False, @@ -306,7 +312,55 @@ class TestStreamingIterator: assert chunk.id == "resp_123" def test_text_delta_sequence_new_schema(self): - """First two OutputTextDeltaEvents emit created + step.start; third emits step.delta.""" + """First chunk yields created + step.start + step.delta; later chunks yield step.delta.""" + it = self._make_iterator(use_legacy=False) + + first_events = it._events_for_chunk(self._make_text_delta("Hello")) + assert [e.event_type for e in first_events] == [ + "interaction.created", + "step.start", + "step.delta", + ] + assert first_events[-1].delta == {"type": "text", "text": "Hello"} + assert it.sent_interaction_start is True + assert it.sent_content_start is True + + second_events = it._events_for_chunk(self._make_text_delta(" World")) + assert [e.event_type for e in second_events] == ["step.delta"] + assert second_events[0].delta == {"type": "text", "text": " World"} + + third_events = it._events_for_chunk(self._make_text_delta("!")) + assert [e.event_type for e in third_events] == ["step.delta"] + assert third_events[0].delta == {"type": "text", "text": "!"} + + def test_text_delta_sequence_legacy_schema(self): + """Legacy: first chunk yields interaction.start + content.start + content.delta.""" + it = self._make_iterator(use_legacy=True) + + first_events = it._events_for_chunk(self._make_text_delta("Hello")) + assert [e.event_type for e in first_events] == [ + "interaction.start", + "content.start", + "content.delta", + ] + assert first_events[-1].delta == {"type": "text", "text": "Hello"} + + second_events = it._events_for_chunk(self._make_text_delta(" World")) + assert [e.event_type for e in second_events] == ["content.delta"] + assert second_events[0].delta == {"type": "text", "text": " World"} + + def test_first_text_delta_without_item_id_uses_fallback_id(self): + it = self._make_iterator(use_legacy=False) + event = self._make_text_delta("Hi") + event.item_id = None + + events = it._events_for_chunk(event) + + assert events[0].event_type == "interaction.created" + assert events[0].id == f"interaction_{id(it)}" + + def test_first_text_delta_emits_text_via_compat_shim(self): + """The legacy single-chunk shim must surface the synthetic events AND the delta.""" it = self._make_iterator(use_legacy=False) first = it._transform_responses_chunk_to_interactions_chunk( @@ -314,57 +368,134 @@ class TestStreamingIterator: ) assert first is not None assert first.event_type == "interaction.created" - assert it.sent_interaction_start is True - assert it.sent_content_start is False - second = it._transform_responses_chunk_to_interactions_chunk( - self._make_text_delta(" World") - ) + second = it.__next__() if it._pending_events else None assert second is not None assert second.event_type == "step.start" - assert it.sent_content_start is True - third = it._transform_responses_chunk_to_interactions_chunk( - self._make_text_delta("!") - ) + third = it.__next__() if it._pending_events else None assert third is not None assert third.event_type == "step.delta" - assert third.delta == {"type": "text", "text": "!"} + assert third.delta == {"type": "text", "text": "Hello"} - def test_text_delta_sequence_legacy_schema(self): - """Legacy: interaction.start → content.start → content.delta.""" - it = self._make_iterator(use_legacy=True) - - first = it._transform_responses_chunk_to_interactions_chunk( - self._make_text_delta("Hello") - ) - assert first is not None - assert first.event_type == "interaction.start" - - second = it._transform_responses_chunk_to_interactions_chunk( - self._make_text_delta(" World") - ) - assert second is not None - assert second.event_type == "content.start" - assert second.delta == {"type": "text", "text": ""} - - third = it._transform_responses_chunk_to_interactions_chunk( - self._make_text_delta("!") - ) - assert third is not None - assert third.event_type == "content.delta" - assert third.delta == {"type": "text", "text": "!"} - - def test_first_text_delta_without_item_id_uses_fallback_id(self): + def test_response_created_then_text_delta_emits_step_start_and_delta(self): + """Realistic flow: response.created arrives first, then text delta.""" it = self._make_iterator(use_legacy=False) - event = self._make_text_delta("Hi") - event.item_id = None - chunk = it._transform_responses_chunk_to_interactions_chunk(event) + first = it._events_for_chunk(self._make_response_created()) + assert [e.event_type for e in first] == ["interaction.created"] - assert chunk is not None - assert chunk.event_type == "interaction.created" - assert chunk.id == f"interaction_{id(it)}" + second = it._events_for_chunk(self._make_text_delta("Hello")) + assert [e.event_type for e in second] == ["step.start", "step.delta"] + assert second[-1].delta == {"type": "text", "text": "Hello"} + + def test_no_text_token_is_dropped_during_streaming(self): + """Concatenated step.delta payloads must equal the upstream text.""" + it = self._make_iterator(use_legacy=False) + + chunks = ["Hello", " ", "world", "!"] + emitted_text = "" + for c in chunks: + for ev in it._events_for_chunk(self._make_text_delta(c)): + if ev.event_type == "step.delta": + assert ev.delta is not None + emitted_text += ev.delta["text"] + + assert emitted_text == "Hello world!" + + def test_stop_iteration_fallback_emits_completion_event(self): + """If upstream ends without ResponseCompletedEvent, terminal events still flow.""" + from unittest.mock import MagicMock + + text_event = self._make_text_delta("hi") + sync_iter = MagicMock() + sync_iter.__iter__ = lambda self: self + sync_iter.__next__ = MagicMock(side_effect=[text_event, StopIteration]) + + original = litellm.use_legacy_interactions_schema + litellm.use_legacy_interactions_schema = False + try: + it = LiteLLMResponsesInteractionsStreamingIterator( + model="gpt-5.4", + litellm_custom_stream_wrapper=sync_iter, + request_input="hi", + optional_params={}, + ) + finally: + litellm.use_legacy_interactions_schema = original + + emitted: list = [] + try: + while True: + emitted.append(next(it)) + except StopIteration: + pass + + event_types = [e.event_type for e in emitted] + assert event_types == [ + "interaction.created", + "step.start", + "step.delta", + "step.stop", + "interaction.completed", + ] + terminal = emitted[-1] + assert terminal.steps == [ + { + "type": "model_output", + "content": [{"type": "text", "text": "hi"}], + } + ] + # EOF-flushed terminal event must carry the same id as interaction.created. + assert terminal.id == emitted[0].id == "item_1" + + def test_response_completed_emits_stop_then_completion(self): + """ResponseCompletedEvent expands into step.stop + interaction.completed.""" + from unittest.mock import MagicMock + + text_event = self._make_text_delta("hi") + completed = MagicMock(spec=ResponseCompletedEvent) + completed.response = MagicMock(id="resp_999") + + sync_iter = MagicMock() + sync_iter.__iter__ = lambda self: self + sync_iter.__next__ = MagicMock(side_effect=[text_event, completed]) + + original = litellm.use_legacy_interactions_schema + litellm.use_legacy_interactions_schema = False + try: + it = LiteLLMResponsesInteractionsStreamingIterator( + model="gpt-5.4", + litellm_custom_stream_wrapper=sync_iter, + request_input="hi", + optional_params={}, + ) + finally: + litellm.use_legacy_interactions_schema = original + + emitted: list = [] + try: + while True: + emitted.append(next(it)) + except StopIteration: + pass + + event_types = [e.event_type for e in emitted] + assert event_types == [ + "interaction.created", + "step.start", + "step.delta", + "step.stop", + "interaction.completed", + ] + # StopIteration fallback path must NOT add a duplicate completion event. + assert event_types.count("interaction.completed") == 1 + # When the stream starts directly with a text delta (no preceding + # response.created), the terminal events must reuse the id derived from + # the first chunk's item_id rather than switching to response.id, so + # consumers can correlate the start and completion events by id. + assert emitted[0].id == "item_1" + assert emitted[-1].id == "item_1" class TestInteractionOperationUrls: