From 9091c66795b4aa1b915a0a29666bfe56c6aa2def Mon Sep 17 00:00:00 2001 From: Daniel Phang Date: Sat, 26 Sep 2026 20:28:31 -0700 Subject: [PATCH 1/7] fix(chatgpt): preserve streamed output in chat bridge --- .../handler.py | 52 +++++++++++---- .../transformation.py | 21 ++++--- ...itellm_responses_transformation_handler.py | 63 ++++++++++++++++++- 3 files changed, 115 insertions(+), 21 deletions(-) diff --git a/litellm/completion_extras/litellm_responses_transformation/handler.py b/litellm/completion_extras/litellm_responses_transformation/handler.py index 642a78789b2..ebbaec8c1bb 100644 --- a/litellm/completion_extras/litellm_responses_transformation/handler.py +++ b/litellm/completion_extras/litellm_responses_transformation/handler.py @@ -2,12 +2,13 @@ Handler for transforming /chat/completions api requests to litellm.responses requests """ -from collections.abc import AsyncIterable, Coroutine, Iterable +from collections.abc import AsyncIterable, Coroutine, Iterable, Mapping from typing import TYPE_CHECKING, Any, Final, Union +from pydantic import BaseModel from typing_extensions import TypedDict -from litellm.types.llms.openai import ResponsesAPIResponse +from litellm.types.llms.openai import ResponsesAPIResponse, ResponsesAPIStreamEvents if TYPE_CHECKING: from litellm import CustomStreamWrapper, LiteLLMLoggingObj, ModelResponse @@ -54,17 +55,32 @@ class ResponsesToCompletionBridgeHandler: def _coerce_response_object( response_obj: object, hidden_params: dict | None, + stream_events: Iterable[object] = (), ) -> "ResponsesAPIResponse": if isinstance(response_obj, ResponsesAPIResponse): - response = response_obj + base_response = response_obj elif isinstance(response_obj, dict): try: - response = ResponsesAPIResponse(**response_obj) + base_response = ResponsesAPIResponse(**response_obj) except Exception: - response = ResponsesAPIResponse.model_construct(**response_obj) + base_response = ResponsesAPIResponse.model_construct(**response_obj) else: raise ValueError("Unexpected responses stream payload") + from .transformation import LiteLLMResponsesTransformationHandler + + parsed_chunks: Final = tuple( + payload + for event in stream_events + if (payload := ResponsesToCompletionBridgeHandler._stream_event_payload(event)) is not None + ) + recovered_output: Final = LiteLLMResponsesTransformationHandler.recover_output_items_from_chunks(parsed_chunks) + response: Final = ( + base_response + if base_response.output or not recovered_output + else base_response.model_copy(update={"output": recovered_output}) + ) + if hidden_params: existing: Final = getattr(response, "_hidden_params", None) if not isinstance(existing, dict) or not existing: @@ -74,9 +90,24 @@ class ResponsesToCompletionBridgeHandler: existing.setdefault(key, value) return response + @staticmethod + def _stream_event_payload(event: object) -> Mapping[str, object] | None: + if isinstance(event, Mapping): + return event + if isinstance(event, BaseModel): + return event.model_dump() + return None + + @staticmethod + def _is_recoverable_stream_event(event: object) -> bool: + event_type: Final = event.get("type") if isinstance(event, Mapping) else getattr(event, "type", None) + return event_type in ( + ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, + ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE, + ) + def _collect_response_from_stream(self, stream_iter: Iterable[object]) -> "ResponsesAPIResponse": - for _ in stream_iter: - pass + stream_events: Final = tuple(event for event in stream_iter if self._is_recoverable_stream_event(event)) completed: Final[object] = getattr(stream_iter, "completed_response", None) response_obj: Final[object] = getattr(completed, "response", None) if completed else None @@ -84,14 +115,13 @@ class ResponsesToCompletionBridgeHandler: raise ValueError("Stream ended without a completed response") hidden_params: Final = getattr(stream_iter, "_hidden_params", None) - response: Final = self._coerce_response_object(response_obj, hidden_params) + response: Final = self._coerce_response_object(response_obj, hidden_params, stream_events) if not isinstance(response, ResponsesAPIResponse): raise ValueError("Stream completed response is invalid") return response async def _collect_response_from_stream_async(self, stream_iter: AsyncIterable[object]) -> "ResponsesAPIResponse": - async for _ in stream_iter: - pass + stream_events: Final = tuple([event async for event in stream_iter if self._is_recoverable_stream_event(event)]) completed: Final[object] = getattr(stream_iter, "completed_response", None) response_obj: Final[object] = getattr(completed, "response", None) if completed else None @@ -99,7 +129,7 @@ class ResponsesToCompletionBridgeHandler: raise ValueError("Stream ended without a completed response") hidden_params: Final = getattr(stream_iter, "_hidden_params", None) - response: Final = self._coerce_response_object(response_obj, hidden_params) + response: Final = self._coerce_response_object(response_obj, hidden_params, stream_events) if not isinstance(response, ResponsesAPIResponse): raise ValueError("Stream completed response is invalid") return response diff --git a/litellm/completion_extras/litellm_responses_transformation/transformation.py b/litellm/completion_extras/litellm_responses_transformation/transformation.py index 31af5a144eb..844da405a98 100644 --- a/litellm/completion_extras/litellm_responses_transformation/transformation.py +++ b/litellm/completion_extras/litellm_responses_transformation/transformation.py @@ -833,18 +833,11 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): return cast(list[dict[str, object]], response_output) @classmethod - def _recover_output_items_from_raw_sse(cls, raw_sse: str | None) -> list[dict[str, object]]: - if not raw_sse or not isinstance(raw_sse, str): - return [] - + def recover_output_items_from_chunks(cls, parsed_chunks: Iterable[Mapping[str, object]]) -> list[dict[str, object]]: recovered_output_items: Final[dict[int, dict[str, object]]] = {} recovered_text_only_items: Final[dict[int, dict[str, object]]] = {} - for chunk in raw_sse.splitlines(): - parsed_chunk = parse_sse_json_chunk(chunk) - if parsed_chunk is None: - continue - + for parsed_chunk in parsed_chunks: event_type = parsed_chunk.get("type") if event_type == ResponsesAPIStreamEvents.RESPONSE_COMPLETED: @@ -881,6 +874,16 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): return [] + @classmethod + def _recover_output_items_from_raw_sse(cls, raw_sse: str | None) -> list[dict[str, object]]: + if not raw_sse or not isinstance(raw_sse, str): + return [] + + parsed_chunks: Final = tuple( + parsed_chunk for chunk in raw_sse.splitlines() if (parsed_chunk := parse_sse_json_chunk(chunk)) is not None + ) + return cls.recover_output_items_from_chunks(parsed_chunks) + @classmethod def _recover_output_items_from_logging(cls, logging_obj: "LiteLLMLoggingObj") -> list[dict[str, object]]: model_call_details: Final = getattr(logging_obj, "model_call_details", {}) or {} diff --git a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py index c5d7ca96a21..76dd5193187 100644 --- a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py +++ b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py @@ -1,15 +1,16 @@ from datetime import datetime +from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch import pytest - import litellm from litellm.completion_extras.litellm_responses_transformation.handler import ( ResponsesToCompletionBridgeHandler, ) from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper +from litellm.types.llms.openai import ResponsesAPIResponse from litellm.types.router import GenericLiteLLMParams from litellm.types.utils import ModelResponse @@ -218,6 +219,66 @@ def _completed_chat_response() -> ModelResponse: ) +def _empty_responses_response() -> ResponsesAPIResponse: + return ResponsesAPIResponse.model_construct(output=[], error=None) + + +def _output_item_done_event() -> dict: + return { + "type": "response.output_item.done", + "output_index": 0, + "item": { + "type": "message", + "id": "msg_from_stream", + "role": "assistant", + "status": "completed", + "content": [ + { + "type": "output_text", + "text": "Recovered from stream", + "annotations": [], + } + ], + }, + } + + +class _SyncResponsesStream: + def __init__(self, response: ResponsesAPIResponse): + self.completed_response = SimpleNamespace(response=response) + self._hidden_params = {} + + def __iter__(self): + return iter((_output_item_done_event(), self.completed_response)) + + +class _AsyncResponsesStream: + def __init__(self, response: ResponsesAPIResponse): + self.completed_response = SimpleNamespace(response=response) + self._hidden_params = {} + + async def __aiter__(self): + for event in (_output_item_done_event(), self.completed_response): + yield event + + +def test_collect_response_from_stream_recovers_output_items(): + bridge = ResponsesToCompletionBridgeHandler() + + response = bridge._collect_response_from_stream(_SyncResponsesStream(_empty_responses_response())) + + assert response.output == [_output_item_done_event()["item"]] + + +@pytest.mark.asyncio +async def test_collect_response_from_async_stream_recovers_output_items(): + bridge = ResponsesToCompletionBridgeHandler() + + response = await bridge._collect_response_from_stream_async(_AsyncResponsesStream(_empty_responses_response())) + + assert response.output == [_output_item_done_event()["item"]] + + @pytest.mark.asyncio async def test_acompletion_streams_completed_model_response(): """A streaming request whose bridge call comes back already completed must still be From d0b0247e3c7ad11cb502474d438063cb3effcb1c Mon Sep 17 00:00:00 2001 From: Daniel Phang Date: Sat, 26 Sep 2026 21:19:44 -0700 Subject: [PATCH 2/7] fix(responses): preserve multipart recovered text --- .../handler.py | 25 ++++++------ .../transformation.py | 39 +++++++++++-------- ...responses_transformation_transformation.py | 36 +++++++++++++++++ 3 files changed, 72 insertions(+), 28 deletions(-) diff --git a/litellm/completion_extras/litellm_responses_transformation/handler.py b/litellm/completion_extras/litellm_responses_transformation/handler.py index ebbaec8c1bb..1f5b95e8c26 100644 --- a/litellm/completion_extras/litellm_responses_transformation/handler.py +++ b/litellm/completion_extras/litellm_responses_transformation/handler.py @@ -67,19 +67,20 @@ class ResponsesToCompletionBridgeHandler: else: raise ValueError("Unexpected responses stream payload") - from .transformation import LiteLLMResponsesTransformationHandler + response = base_response + if not base_response.output: + from .transformation import LiteLLMResponsesTransformationHandler - parsed_chunks: Final = tuple( - payload - for event in stream_events - if (payload := ResponsesToCompletionBridgeHandler._stream_event_payload(event)) is not None - ) - recovered_output: Final = LiteLLMResponsesTransformationHandler.recover_output_items_from_chunks(parsed_chunks) - response: Final = ( - base_response - if base_response.output or not recovered_output - else base_response.model_copy(update={"output": recovered_output}) - ) + parsed_chunks: Final = ( + payload + for event in stream_events + if (payload := ResponsesToCompletionBridgeHandler._stream_event_payload(event)) is not None + ) + recovered_output: Final = LiteLLMResponsesTransformationHandler.recover_output_items_from_chunks( + parsed_chunks + ) + if recovered_output: + response = base_response.model_copy(update={"output": recovered_output}) if hidden_params: existing: Final = getattr(response, "_hidden_params", None) diff --git a/litellm/completion_extras/litellm_responses_transformation/transformation.py b/litellm/completion_extras/litellm_responses_transformation/transformation.py index 844da405a98..e260e5188bb 100644 --- a/litellm/completion_extras/litellm_responses_transformation/transformation.py +++ b/litellm/completion_extras/litellm_responses_transformation/transformation.py @@ -334,22 +334,29 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): # Handle message items with output_text content if item_type == "message": content_list: Final = item.get("content", []) + response_text_parts: Final[list[str]] = [] + message_annotations: Final[list[ChatCompletionAnnotation]] = [] + has_output_text = False for content_item in content_list: - if isinstance(content_item, dict): - content_type = content_item.get("type") - if content_type == "output_text": - response_text = content_item.get("text", "") - # Extract annotations from content if present - annotations = LiteLLMResponsesTransformationHandler._convert_annotations_to_chat_format( - content_item.get("annotations", None) - ) - msg = Message( - role=item.get("role", "assistant"), - content=response_text if response_text else "", - annotations=annotations, - ) - choice = Choices(message=msg, finish_reason="stop", index=index) - return choice, index + 1 + if not isinstance(content_item, dict) or content_item.get("type") != "output_text": + continue + has_output_text = True + response_text = content_item.get("text", "") + response_text_parts.append(response_text if isinstance(response_text, str) else "") + annotations = LiteLLMResponsesTransformationHandler._convert_annotations_to_chat_format( + content_item.get("annotations", None) + ) + if annotations: + message_annotations.extend(annotations) + + if has_output_text: + msg = Message( + role=item.get("role", "assistant"), + content="".join(response_text_parts), + annotations=message_annotations or None, + ) + choice = Choices(message=msg, finish_reason="stop", index=index) + return choice, index + 1 # function_call / custom_tool_call dicts are intercepted and accumulated by # _convert_response_output_to_choices before this callback is reached @@ -879,7 +886,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): if not raw_sse or not isinstance(raw_sse, str): return [] - parsed_chunks: Final = tuple( + parsed_chunks: Final = ( parsed_chunk for chunk in raw_sse.splitlines() if (parsed_chunk := parse_sse_json_chunk(chunk)) is not None ) return cls.recover_output_items_from_chunks(parsed_chunks) diff --git a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py index d0f9bad795d..b09ac17d3ea 100644 --- a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py +++ b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py @@ -610,6 +610,42 @@ def test_transform_response_recovers_empty_output_from_raw_sse(): assert result.choices[0].message.content == "Recovered from SSE" +def test_transform_response_recovers_all_text_parts_from_raw_sse(): + from litellm.completion_extras.litellm_responses_transformation.transformation import ( + LiteLLMResponsesTransformationHandler, + ) + + handler = LiteLLMResponsesTransformationHandler() + raw_sse = "\n".join( + [ + 'data: {"type":"response.output_text.done","output_index":0,"content_index":0,"item_id":"msg_from_stream","text":"Hello, "}', + 'data: {"type":"response.output_text.done","output_index":0,"content_index":1,"item_id":"msg_from_stream","text":"world!"}', + 'data: {"type":"response.completed","response":{"id":"resp_from_stream","object":"response","created_at":1760144904,"status":"completed","model":"gpt-5.4","output":[]}}', + "data: [DONE]", + "", + ] + ) + raw_response = _make_empty_responses_api_response() + model_response = _make_empty_model_response() + logging_obj = Mock() + logging_obj.model_call_details = {"original_response": raw_sse} + + result = handler.transform_response( + model="gpt-5.4", + raw_response=raw_response, + model_response=model_response, + logging_obj=logging_obj, + request_data={"model": "gpt-5.4"}, + messages=[{"role": "user", "content": "Reply with exactly: Hello, world!"}], + optional_params={}, + litellm_params={}, + encoding=Mock(), + ) + + assert len(result.choices) == 1 + assert result.choices[0].message.content == "Hello, world!" + + def test_transform_response_recovers_output_item_done_from_raw_sse(): from litellm.completion_extras.litellm_responses_transformation.transformation import ( LiteLLMResponsesTransformationHandler, From 76e5741084d5a8703e2dd8949b4c069a559f9ad3 Mon Sep 17 00:00:00 2001 From: Daniel Phang Date: Sat, 26 Sep 2026 21:50:39 -0700 Subject: [PATCH 3/7] fix(responses): preserve citation offsets in recovered text --- .../handler.py | 29 +++++----- .../transformation.py | 44 +++++++++------ ...responses_transformation_transformation.py | 56 +++++++++++++++---- 3 files changed, 88 insertions(+), 41 deletions(-) diff --git a/litellm/completion_extras/litellm_responses_transformation/handler.py b/litellm/completion_extras/litellm_responses_transformation/handler.py index 1f5b95e8c26..8dc5e30dd89 100644 --- a/litellm/completion_extras/litellm_responses_transformation/handler.py +++ b/litellm/completion_extras/litellm_responses_transformation/handler.py @@ -67,20 +67,7 @@ class ResponsesToCompletionBridgeHandler: else: raise ValueError("Unexpected responses stream payload") - response = base_response - if not base_response.output: - from .transformation import LiteLLMResponsesTransformationHandler - - parsed_chunks: Final = ( - payload - for event in stream_events - if (payload := ResponsesToCompletionBridgeHandler._stream_event_payload(event)) is not None - ) - recovered_output: Final = LiteLLMResponsesTransformationHandler.recover_output_items_from_chunks( - parsed_chunks - ) - if recovered_output: - response = base_response.model_copy(update={"output": recovered_output}) + response: Final = ResponsesToCompletionBridgeHandler._recover_stream_output(base_response, stream_events) if hidden_params: existing: Final = getattr(response, "_hidden_params", None) @@ -91,6 +78,20 @@ class ResponsesToCompletionBridgeHandler: existing.setdefault(key, value) return response + @staticmethod + def _recover_stream_output(response: ResponsesAPIResponse, stream_events: Iterable[object]) -> ResponsesAPIResponse: + if response.output: + return response + from .transformation import LiteLLMResponsesTransformationHandler + + parsed_chunks: Final = ( + payload + for event in stream_events + if (payload := ResponsesToCompletionBridgeHandler._stream_event_payload(event)) is not None + ) + recovered_output: Final = LiteLLMResponsesTransformationHandler.recover_output_items_from_chunks(parsed_chunks) + return response.model_copy(update={"output": recovered_output}) if recovered_output else response + @staticmethod def _stream_event_payload(event: object) -> Mapping[str, object] | None: if isinstance(event, Mapping): diff --git a/litellm/completion_extras/litellm_responses_transformation/transformation.py b/litellm/completion_extras/litellm_responses_transformation/transformation.py index e260e5188bb..ca4dbcfa163 100644 --- a/litellm/completion_extras/litellm_responses_transformation/transformation.py +++ b/litellm/completion_extras/litellm_responses_transformation/transformation.py @@ -5,6 +5,7 @@ Handler for transforming /chat/completions api requests to litellm.responses req import json import os from collections.abc import AsyncIterator, Callable, Iterable, Iterator, Mapping, Sequence +from itertools import accumulate, chain from types import MappingProxyType from typing import TYPE_CHECKING, Any, Final, Literal, TypedDict, TypeVar, Union, cast @@ -78,6 +79,16 @@ _RESPONSES_API_ONLY_FIELDS: Final = frozenset((*Response.model_fields, *Response ) +def _offset_annotation(annotation: Mapping[str, object], offset: int) -> dict[str, object]: + citation: Final = annotation.get("url_citation") + if isinstance(citation, dict): + return {**annotation, "url_citation": _offset_annotation(citation, offset)} + return { + key: value + offset if key in ("start_index", "end_index") and isinstance(value, int) else value + for key, value in annotation.items() + } + + def _provider_metadata(response_fields: Mapping[str, object] | None) -> Mapping[str, object]: return MappingProxyType( { @@ -334,28 +345,29 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): # Handle message items with output_text content if item_type == "message": content_list: Final = item.get("content", []) - response_text_parts: Final[list[str]] = [] - message_annotations: Final[list[ChatCompletionAnnotation]] = [] - has_output_text = False - for content_item in content_list: - if not isinstance(content_item, dict) or content_item.get("type") != "output_text": - continue - has_output_text = True - response_text = content_item.get("text", "") - response_text_parts.append(response_text if isinstance(response_text, str) else "") - annotations = LiteLLMResponsesTransformationHandler._convert_annotations_to_chat_format( - content_item.get("annotations", None) + text_parts: Final = tuple( + part for part in content_list if isinstance(part, dict) and part.get("type") == "output_text" + ) + response_text_parts: Final = tuple( + text if isinstance(text := part.get("text"), str) else "" for part in text_parts + ) + offsets: Final = accumulate(map(len, response_text_parts), initial=0) + annotation_groups: Final = ( + tuple( + _offset_annotation(annotation, offset) + for annotation in self._convert_annotations_to_chat_format(part.get("annotations")) or () ) - if annotations: - message_annotations.extend(annotations) + for part, offset in zip(text_parts, offsets) + ) + message_annotations: Final = list(chain.from_iterable(annotation_groups)) - if has_output_text: - msg = Message( + if text_parts: + msg: Final = Message( role=item.get("role", "assistant"), content="".join(response_text_parts), annotations=message_annotations or None, ) - choice = Choices(message=msg, finish_reason="stop", index=index) + choice: Final = Choices(message=msg, finish_reason="stop", index=index) return choice, index + 1 # function_call / custom_tool_call dicts are intercepted and accumulated by diff --git a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py index b09ac17d3ea..85364d1fc69 100644 --- a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py +++ b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py @@ -610,25 +610,51 @@ def test_transform_response_recovers_empty_output_from_raw_sse(): assert result.choices[0].message.content == "Recovered from SSE" -def test_transform_response_recovers_all_text_parts_from_raw_sse(): +@pytest.mark.parametrize("nested_citation", [False, True]) +@pytest.mark.parametrize("recovery_path", ["raw_sse", "bridge"]) +def test_transform_response_recovers_text_and_citation_offsets(nested_citation, recovery_path): + from copy import deepcopy + + from litellm.completion_extras.litellm_responses_transformation.handler import ( + ResponsesToCompletionBridgeHandler, + ) from litellm.completion_extras.litellm_responses_transformation.transformation import ( LiteLLMResponsesTransformationHandler, ) handler = LiteLLMResponsesTransformationHandler() - raw_sse = "\n".join( - [ - 'data: {"type":"response.output_text.done","output_index":0,"content_index":0,"item_id":"msg_from_stream","text":"Hello, "}', - 'data: {"type":"response.output_text.done","output_index":0,"content_index":1,"item_id":"msg_from_stream","text":"world!"}', - 'data: {"type":"response.completed","response":{"id":"resp_from_stream","object":"response","created_at":1760144904,"status":"completed","model":"gpt-5.4","output":[]}}', - "data: [DONE]", - "", - ] + citation = {"start_index": 0, "end_index": 5, "title": "Source", "url": "https://example.com"} + annotation = ( + {"type": "url_citation", "url_citation": citation} if nested_citation else {"type": "url_citation", **citation} + ) + events = ( + { + "type": "response.output_text.done", + "output_index": 0, + "content_index": 1, + "item_id": "msg_from_stream", + "text": "world!", + "annotations": [annotation], + }, + { + "type": "response.output_text.done", + "output_index": 0, + "content_index": 0, + "item_id": "msg_from_stream", + "text": "Hello, ", + "annotations": [annotation], + }, + ) + original_events = deepcopy(events) + raw_sse = "\n".join(f"data: {json.dumps(event)}" for event in events) + raw_response = ( + ResponsesToCompletionBridgeHandler._coerce_response_object(_make_empty_responses_api_response(), None, events) + if recovery_path == "bridge" + else _make_empty_responses_api_response() ) - raw_response = _make_empty_responses_api_response() model_response = _make_empty_model_response() logging_obj = Mock() - logging_obj.model_call_details = {"original_response": raw_sse} + logging_obj.model_call_details = {"original_response": raw_sse if recovery_path == "raw_sse" else None} result = handler.transform_response( model="gpt-5.4", @@ -644,6 +670,14 @@ def test_transform_response_recovers_all_text_parts_from_raw_sse(): assert len(result.choices) == 1 assert result.choices[0].message.content == "Hello, world!" + shifted_citation = {**citation, "start_index": 7, "end_index": 12} + shifted_annotation = ( + {"type": "url_citation", "url_citation": shifted_citation} + if nested_citation + else {"type": "url_citation", **shifted_citation} + ) + assert result.choices[0].message.annotations == [annotation, shifted_annotation] + assert events == original_events def test_transform_response_recovers_output_item_done_from_raw_sse(): From 4f3973c39bddf3a34e6009e5134b36692043104c Mon Sep 17 00:00:00 2001 From: Daniel Phang Date: Sat, 26 Sep 2026 21:56:22 -0700 Subject: [PATCH 4/7] test(responses): cover stream recovery compatibility paths --- ...itellm_responses_transformation_handler.py | 45 +++++++++++++++++++ 1 file changed, 45 insertions(+) diff --git a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py index 76dd5193187..6f88a4b5bcf 100644 --- a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py +++ b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py @@ -3,6 +3,7 @@ from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch import pytest +from openai.types.responses import ResponseOutputItemDoneEvent import litellm from litellm.completion_extras.litellm_responses_transformation.handler import ( @@ -279,6 +280,50 @@ async def test_collect_response_from_async_stream_recovers_output_items(): assert response.output == [_output_item_done_event()["item"]] +@pytest.mark.parametrize( + "terminal_payload", + [ + {"id": "resp_test", "created_at": 0, "output": []}, + {"output": []}, + ], + ids=["validated-terminal", "partial-terminal"], +) +def test_recovery_accepts_sdk_events_and_dictionary_terminals(terminal_payload): + event = ResponseOutputItemDoneEvent(**_output_item_done_event(), sequence_number=1) + + response = ResponsesToCompletionBridgeHandler._coerce_response_object( + terminal_payload, + {"headers": {"x-request-id": "req_test"}}, + (object(), event), + ) + + assert response.output == [event.item.model_dump()] + assert response._hidden_params["headers"] == {"x-request-id": "req_test"} + assert terminal_payload["output"] == [] + + +def test_recovery_preserves_complete_response_without_consuming_events(): + terminal = ResponsesAPIResponse.model_construct(output=[_output_item_done_event()["item"]]) + + def unavailable_events(): + raise AssertionError("Complete terminal output must bypass event recovery") + yield + + response = ResponsesToCompletionBridgeHandler._coerce_response_object(terminal, None, unavailable_events()) + + assert response is terminal + assert response.output == [_output_item_done_event()["item"]] + + +def test_recovery_keeps_empty_terminal_when_no_output_can_be_recovered(): + terminal = _empty_responses_response() + + response = ResponsesToCompletionBridgeHandler._coerce_response_object(terminal, None, (object(),)) + + assert response is terminal + assert response.output == [] + + @pytest.mark.asyncio async def test_acompletion_streams_completed_model_response(): """A streaming request whose bridge call comes back already completed must still be From 9124f0981db8ac5bebf16c0217fc7025bb4e59d2 Mon Sep 17 00:00:00 2001 From: Daniel Phang Date: Sat, 26 Sep 2026 22:13:18 -0700 Subject: [PATCH 5/7] fix(responses): drop citation offset recursion and test the real stream iterator The code-quality recursive_detector gate rejected _offset_annotation because it called itself to reach the nested url_citation form. The nesting is only one level deep, so a flat _offset_citation_indices helper replaces the recursion The bridge recovery tests used handwritten stand-ins for LiteLLM's own Responses streaming iterators, which tests/unit/AGENTS.md forbids. They also skipped the part that matters most: whether the typed events the real iterator yields pass _is_recoverable_stream_event and dump into payloads that recovery can read. The tests now run SyncResponsesAPIStreamingIterator and ResponsesAPIStreamingIterator over an httpx SSE body for both output_item.done and output_text.done events, and all four cases fail when recovery is disabled --- .../transformation.py | 15 ++- ...itellm_responses_transformation_handler.py | 94 ++++++++++++++----- 2 files changed, 78 insertions(+), 31 deletions(-) diff --git a/litellm/completion_extras/litellm_responses_transformation/transformation.py b/litellm/completion_extras/litellm_responses_transformation/transformation.py index ca4dbcfa163..9b2f6511b84 100644 --- a/litellm/completion_extras/litellm_responses_transformation/transformation.py +++ b/litellm/completion_extras/litellm_responses_transformation/transformation.py @@ -79,16 +79,21 @@ _RESPONSES_API_ONLY_FIELDS: Final = frozenset((*Response.model_fields, *Response ) -def _offset_annotation(annotation: Mapping[str, object], offset: int) -> dict[str, object]: - citation: Final = annotation.get("url_citation") - if isinstance(citation, dict): - return {**annotation, "url_citation": _offset_annotation(citation, offset)} +def _offset_citation_indices(fields: Mapping[str, object], offset: int) -> dict[str, object]: return { key: value + offset if key in ("start_index", "end_index") and isinstance(value, int) else value - for key, value in annotation.items() + for key, value in fields.items() } +def _offset_annotation(annotation: Mapping[str, object], offset: int) -> dict[str, object]: + """Shift citation indices in either the flat Responses form or the nested Chat Completions ``url_citation`` form.""" + citation: Final = annotation.get("url_citation") + if isinstance(citation, dict): + return {**annotation, "url_citation": _offset_citation_indices(citation, offset)} + return _offset_citation_indices(annotation, offset) + + def _provider_metadata(response_fields: Mapping[str, object] | None) -> Mapping[str, object]: return MappingProxyType( { diff --git a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py index 6f88a4b5bcf..8752cc7f969 100644 --- a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py +++ b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py @@ -1,7 +1,8 @@ +import json from datetime import datetime -from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch +import httpx import pytest from openai.types.responses import ResponseOutputItemDoneEvent @@ -11,6 +12,8 @@ from litellm.completion_extras.litellm_responses_transformation.handler import ( ) from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper +from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig +from litellm.responses.streaming_iterator import ResponsesAPIStreamingIterator, SyncResponsesAPIStreamingIterator from litellm.types.llms.openai import ResponsesAPIResponse from litellm.types.router import GenericLiteLLMParams from litellm.types.utils import ModelResponse @@ -244,40 +247,79 @@ def _output_item_done_event() -> dict: } -class _SyncResponsesStream: - def __init__(self, response: ResponsesAPIResponse): - self.completed_response = SimpleNamespace(response=response) - self._hidden_params = {} - - def __iter__(self): - return iter((_output_item_done_event(), self.completed_response)) +def _output_text_done_event() -> dict: + return { + "type": "response.output_text.done", + "sequence_number": 1, + "item_id": "msg_from_stream", + "output_index": 0, + "content_index": 0, + "text": "Recovered from stream", + } -class _AsyncResponsesStream: - def __init__(self, response: ResponsesAPIResponse): - self.completed_response = SimpleNamespace(response=response) - self._hidden_params = {} - - async def __aiter__(self): - for event in (_output_item_done_event(), self.completed_response): - yield event +def _empty_completed_event() -> dict: + return { + "type": "response.completed", + "sequence_number": 2, + "response": { + "id": "resp_test", + "object": "response", + "created_at": 0, + "status": "completed", + "model": "gpt-5.4", + "output": [], + "parallel_tool_calls": True, + "tool_choice": "auto", + "tools": [], + }, + } -def test_collect_response_from_stream_recovers_output_items(): - bridge = ResponsesToCompletionBridgeHandler() - - response = bridge._collect_response_from_stream(_SyncResponsesStream(_empty_responses_response())) - - assert response.output == [_output_item_done_event()["item"]] +def _sse_http_response(*events: dict) -> httpx.Response: + body = "".join(f"event: {event['type']}\ndata: {json.dumps(event)}\n\n" for event in events) + return httpx.Response(200, content=body.encode(), request=httpx.Request("POST", "https://example.com/responses")) +def _real_stream_iterator_kwargs(recoverable_event: dict) -> dict: + """Feed the real Responses streaming iterator an SSE body whose terminal event has empty output.""" + return { + "response": _sse_http_response(recoverable_event, _empty_completed_event()), + "model": "gpt-5.4", + "responses_api_provider_config": OpenAIResponsesAPIConfig(), + "logging_obj": _bridge_kwargs(stream=True)["logging_obj"], + "custom_llm_provider": "chatgpt", + } + + +def _recovered_texts(response: ResponsesAPIResponse) -> list[str]: + return [part["text"] for item in response.output for part in item["content"]] + + +_RECOVERABLE_STREAM_EVENTS = pytest.mark.parametrize( + "recoverable_event", + [_output_item_done_event(), _output_text_done_event()], + ids=["output_item_done", "output_text_done"], +) + + +@_RECOVERABLE_STREAM_EVENTS +def test_collect_response_from_stream_recovers_output_items(recoverable_event): + stream = SyncResponsesAPIStreamingIterator(**_real_stream_iterator_kwargs(recoverable_event)) + + response = ResponsesToCompletionBridgeHandler()._collect_response_from_stream(stream) + + assert _recovered_texts(response) == ["Recovered from stream"] + + +@_RECOVERABLE_STREAM_EVENTS @pytest.mark.asyncio -async def test_collect_response_from_async_stream_recovers_output_items(): - bridge = ResponsesToCompletionBridgeHandler() +async def test_collect_response_from_async_stream_recovers_output_items(recoverable_event): + stream = ResponsesAPIStreamingIterator(**_real_stream_iterator_kwargs(recoverable_event)) - response = await bridge._collect_response_from_stream_async(_AsyncResponsesStream(_empty_responses_response())) + response = await ResponsesToCompletionBridgeHandler()._collect_response_from_stream_async(stream) - assert response.output == [_output_item_done_event()["item"]] + assert _recovered_texts(response) == ["Recovered from stream"] @pytest.mark.parametrize( From be1fff7491e2622b817e2064d7a185aa9c7c1822 Mon Sep 17 00:00:00 2001 From: Daniel Phang Date: Sun, 27 Sep 2026 09:46:18 -0700 Subject: [PATCH 6/7] test(responses): type stream fixtures and remove redundant docstrings --- .../transformation.py | 1 - ...itellm_responses_transformation_handler.py | 52 +++++++++++++------ 2 files changed, 36 insertions(+), 17 deletions(-) diff --git a/litellm/completion_extras/litellm_responses_transformation/transformation.py b/litellm/completion_extras/litellm_responses_transformation/transformation.py index 9b2f6511b84..37913e112cf 100644 --- a/litellm/completion_extras/litellm_responses_transformation/transformation.py +++ b/litellm/completion_extras/litellm_responses_transformation/transformation.py @@ -87,7 +87,6 @@ def _offset_citation_indices(fields: Mapping[str, object], offset: int) -> dict[ def _offset_annotation(annotation: Mapping[str, object], offset: int) -> dict[str, object]: - """Shift citation indices in either the flat Responses form or the nested Chat Completions ``url_citation`` form.""" citation: Final = annotation.get("url_citation") if isinstance(citation, dict): return {**annotation, "url_citation": _offset_citation_indices(citation, offset)} diff --git a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py index 8752cc7f969..95bf6e5e2e5 100644 --- a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py +++ b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py @@ -1,10 +1,14 @@ import json from datetime import datetime +from itertools import chain +from typing import Final from unittest.mock import AsyncMock, MagicMock, patch import httpx import pytest from openai.types.responses import ResponseOutputItemDoneEvent +from pydantic import JsonValue +from typing_extensions import ReadOnly, TypedDict import litellm from litellm.completion_extras.litellm_responses_transformation.handler import ( @@ -227,7 +231,7 @@ def _empty_responses_response() -> ResponsesAPIResponse: return ResponsesAPIResponse.model_construct(output=[], error=None) -def _output_item_done_event() -> dict: +def _output_item_done_event() -> dict[str, JsonValue]: return { "type": "response.output_item.done", "output_index": 0, @@ -247,7 +251,7 @@ def _output_item_done_event() -> dict: } -def _output_text_done_event() -> dict: +def _output_text_done_event() -> dict[str, JsonValue]: return { "type": "response.output_text.done", "sequence_number": 1, @@ -258,7 +262,7 @@ def _output_text_done_event() -> dict: } -def _empty_completed_event() -> dict: +def _empty_completed_event() -> dict[str, JsonValue]: return { "type": "response.completed", "sequence_number": 2, @@ -276,27 +280,43 @@ def _empty_completed_event() -> dict: } -def _sse_http_response(*events: dict) -> httpx.Response: - body = "".join(f"event: {event['type']}\ndata: {json.dumps(event)}\n\n" for event in events) +def _sse_http_response(*events: dict[str, JsonValue]) -> httpx.Response: + body: Final = "".join(f"event: {event['type']}\ndata: {json.dumps(event)}\n\n" for event in events) return httpx.Response(200, content=body.encode(), request=httpx.Request("POST", "https://example.com/responses")) -def _real_stream_iterator_kwargs(recoverable_event: dict) -> dict: - """Feed the real Responses streaming iterator an SSE body whose terminal event has empty output.""" +class _StreamIteratorKwargs(TypedDict): + response: ReadOnly[httpx.Response] + model: ReadOnly[str] + responses_api_provider_config: ReadOnly[OpenAIResponsesAPIConfig] + logging_obj: ReadOnly[LiteLLMLogging] + custom_llm_provider: ReadOnly[str] + + +def _real_stream_iterator_kwargs(recoverable_event: dict[str, JsonValue]) -> _StreamIteratorKwargs: return { "response": _sse_http_response(recoverable_event, _empty_completed_event()), "model": "gpt-5.4", "responses_api_provider_config": OpenAIResponsesAPIConfig(), - "logging_obj": _bridge_kwargs(stream=True)["logging_obj"], + "logging_obj": LiteLLMLogging( + litellm_call_id="test-call", + call_type="completion", + model="gpt-5.4", + messages=[{"role": "user", "content": "hi"}], + function_id="fn-id", + stream=True, + start_time=datetime(2026, 1, 1), + ), "custom_llm_provider": "chatgpt", } def _recovered_texts(response: ResponsesAPIResponse) -> list[str]: - return [part["text"] for item in response.output for part in item["content"]] + content_parts: Final = chain.from_iterable(item["content"] for item in response.output) + return [part["text"] for part in content_parts] -_RECOVERABLE_STREAM_EVENTS = pytest.mark.parametrize( +_RECOVERABLE_STREAM_EVENTS: Final = pytest.mark.parametrize( "recoverable_event", [_output_item_done_event(), _output_text_done_event()], ids=["output_item_done", "output_text_done"], @@ -304,20 +324,20 @@ _RECOVERABLE_STREAM_EVENTS = pytest.mark.parametrize( @_RECOVERABLE_STREAM_EVENTS -def test_collect_response_from_stream_recovers_output_items(recoverable_event): - stream = SyncResponsesAPIStreamingIterator(**_real_stream_iterator_kwargs(recoverable_event)) +def test_collect_response_from_stream_recovers_output_items(recoverable_event: dict[str, JsonValue]) -> None: + stream: Final = SyncResponsesAPIStreamingIterator(**_real_stream_iterator_kwargs(recoverable_event)) - response = ResponsesToCompletionBridgeHandler()._collect_response_from_stream(stream) + response: Final = ResponsesToCompletionBridgeHandler()._collect_response_from_stream(stream) assert _recovered_texts(response) == ["Recovered from stream"] @_RECOVERABLE_STREAM_EVENTS @pytest.mark.asyncio -async def test_collect_response_from_async_stream_recovers_output_items(recoverable_event): - stream = ResponsesAPIStreamingIterator(**_real_stream_iterator_kwargs(recoverable_event)) +async def test_collect_response_from_async_stream_recovers_output_items(recoverable_event: dict[str, JsonValue]) -> None: + stream: Final = ResponsesAPIStreamingIterator(**_real_stream_iterator_kwargs(recoverable_event)) - response = await ResponsesToCompletionBridgeHandler()._collect_response_from_stream_async(stream) + response: Final = await ResponsesToCompletionBridgeHandler()._collect_response_from_stream_async(stream) assert _recovered_texts(response) == ["Recovered from stream"] From cfc7d488cdd6b69df77f0788e74b82fb0d1e8f36 Mon Sep 17 00:00:00 2001 From: Daniel Phang Date: Sun, 27 Sep 2026 09:53:14 -0700 Subject: [PATCH 7/7] test(responses): complete recovery regression annotations --- ...itellm_responses_transformation_handler.py | 21 +++++++------- ...responses_transformation_transformation.py | 28 ++++++++++--------- 2 files changed, 26 insertions(+), 23 deletions(-) diff --git a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py index 95bf6e5e2e5..871f2a3ff70 100644 --- a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py +++ b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py @@ -1,4 +1,5 @@ import json +from collections.abc import Iterator from datetime import datetime from itertools import chain from typing import Final @@ -350,10 +351,10 @@ async def test_collect_response_from_async_stream_recovers_output_items(recovera ], ids=["validated-terminal", "partial-terminal"], ) -def test_recovery_accepts_sdk_events_and_dictionary_terminals(terminal_payload): - event = ResponseOutputItemDoneEvent(**_output_item_done_event(), sequence_number=1) +def test_recovery_accepts_sdk_events_and_dictionary_terminals(terminal_payload: dict[str, JsonValue]) -> None: + event: Final = ResponseOutputItemDoneEvent(**_output_item_done_event(), sequence_number=1) - response = ResponsesToCompletionBridgeHandler._coerce_response_object( + response: Final = ResponsesToCompletionBridgeHandler._coerce_response_object( terminal_payload, {"headers": {"x-request-id": "req_test"}}, (object(), event), @@ -364,23 +365,23 @@ def test_recovery_accepts_sdk_events_and_dictionary_terminals(terminal_payload): assert terminal_payload["output"] == [] -def test_recovery_preserves_complete_response_without_consuming_events(): - terminal = ResponsesAPIResponse.model_construct(output=[_output_item_done_event()["item"]]) +def test_recovery_preserves_complete_response_without_consuming_events() -> None: + terminal: Final = ResponsesAPIResponse.model_construct(output=[_output_item_done_event()["item"]]) - def unavailable_events(): + def unavailable_events() -> Iterator[object]: raise AssertionError("Complete terminal output must bypass event recovery") yield - response = ResponsesToCompletionBridgeHandler._coerce_response_object(terminal, None, unavailable_events()) + response: Final = ResponsesToCompletionBridgeHandler._coerce_response_object(terminal, None, unavailable_events()) assert response is terminal assert response.output == [_output_item_done_event()["item"]] -def test_recovery_keeps_empty_terminal_when_no_output_can_be_recovered(): - terminal = _empty_responses_response() +def test_recovery_keeps_empty_terminal_when_no_output_can_be_recovered() -> None: + terminal: Final = _empty_responses_response() - response = ResponsesToCompletionBridgeHandler._coerce_response_object(terminal, None, (object(),)) + response: Final = ResponsesToCompletionBridgeHandler._coerce_response_object(terminal, None, (object(),)) assert response is terminal assert response.output == [] diff --git a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py index 85364d1fc69..6a87345e4a7 100644 --- a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py +++ b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py @@ -612,7 +612,9 @@ def test_transform_response_recovers_empty_output_from_raw_sse(): @pytest.mark.parametrize("nested_citation", [False, True]) @pytest.mark.parametrize("recovery_path", ["raw_sse", "bridge"]) -def test_transform_response_recovers_text_and_citation_offsets(nested_citation, recovery_path): +def test_transform_response_recovers_text_and_citation_offsets( + nested_citation: bool, recovery_path: Literal["raw_sse", "bridge"] +) -> None: from copy import deepcopy from litellm.completion_extras.litellm_responses_transformation.handler import ( @@ -622,12 +624,12 @@ def test_transform_response_recovers_text_and_citation_offsets(nested_citation, LiteLLMResponsesTransformationHandler, ) - handler = LiteLLMResponsesTransformationHandler() - citation = {"start_index": 0, "end_index": 5, "title": "Source", "url": "https://example.com"} - annotation = ( + handler: Final = LiteLLMResponsesTransformationHandler() + citation: Final = {"start_index": 0, "end_index": 5, "title": "Source", "url": "https://example.com"} + annotation: Final = ( {"type": "url_citation", "url_citation": citation} if nested_citation else {"type": "url_citation", **citation} ) - events = ( + events: Final = ( { "type": "response.output_text.done", "output_index": 0, @@ -645,18 +647,18 @@ def test_transform_response_recovers_text_and_citation_offsets(nested_citation, "annotations": [annotation], }, ) - original_events = deepcopy(events) - raw_sse = "\n".join(f"data: {json.dumps(event)}" for event in events) - raw_response = ( + original_events: Final = deepcopy(events) + raw_sse: Final = "\n".join(f"data: {json.dumps(event)}" for event in events) + raw_response: Final = ( ResponsesToCompletionBridgeHandler._coerce_response_object(_make_empty_responses_api_response(), None, events) if recovery_path == "bridge" else _make_empty_responses_api_response() ) - model_response = _make_empty_model_response() - logging_obj = Mock() + model_response: Final = _make_empty_model_response() + logging_obj: Final = Mock() logging_obj.model_call_details = {"original_response": raw_sse if recovery_path == "raw_sse" else None} - result = handler.transform_response( + result: Final = handler.transform_response( model="gpt-5.4", raw_response=raw_response, model_response=model_response, @@ -670,8 +672,8 @@ def test_transform_response_recovers_text_and_citation_offsets(nested_citation, assert len(result.choices) == 1 assert result.choices[0].message.content == "Hello, world!" - shifted_citation = {**citation, "start_index": 7, "end_index": 12} - shifted_annotation = ( + shifted_citation: Final = {**citation, "start_index": 7, "end_index": 12} + shifted_annotation: Final = ( {"type": "url_citation", "url_citation": shifted_citation} if nested_citation else {"type": "url_citation", **shifted_citation}