Merge pull request #34539 from BerriAI/litellm_fix_responses_bridge_streaming_contract

fix(responses_bridge): keep one chat completion id per stream and always stream completed responses

litellm/completion_extras/litellm_responses_transformation/handler.py

@@ -209,7 +209,15 @@ class ResponsesToCompletionBridgeHandler:
                 json_mode=kwargs.get("json_mode"),
             )
         elif isinstance(result, ModelResponse):
-            return result
+            if not stream:
+                return result
+            return self._completed_response_as_stream(
+                response=result,
+                model=model,
+                custom_llm_provider=custom_llm_provider,
+                logging_obj=logging_obj,
+                json_mode=kwargs.get("json_mode"),
+            )
         elif not stream:
             responses_api_response = self._collect_response_from_stream(result)
             return self.transformation_handler.transform_response(
@@ -299,7 +307,15 @@ class ResponsesToCompletionBridgeHandler:
                 json_mode=kwargs.get("json_mode"),
             )
         elif isinstance(result, ModelResponse):
-            return result
+            if not stream:
+                return result
+            return self._completed_response_as_stream(
+                response=result,
+                model=model,
+                custom_llm_provider=custom_llm_provider,
+                logging_obj=logging_obj,
+                json_mode=kwargs.get("json_mode"),
+            )
         elif not stream:
             responses_api_response = await self._collect_response_from_stream_async(result)
             return self.transformation_handler.transform_response(
@@ -331,6 +347,25 @@ class ResponsesToCompletionBridgeHandler:
             )
             return self._apply_post_stream_processing(streamwrapper, model, custom_llm_provider)
 
+    def _completed_response_as_stream(
+        self,
+        response: "ModelResponse",
+        model: str,
+        custom_llm_provider: str,
+        logging_obj: "LiteLLMLoggingObj",
+        json_mode: bool | None,
+    ) -> "CustomStreamWrapper":
+        from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
+        from litellm.llms.base_llm.base_model_iterator import MockResponseIterator
+
+        streamwrapper = CustomStreamWrapper(
+            completion_stream=MockResponseIterator(model_response=response, json_mode=json_mode),
+            model=model,
+            custom_llm_provider=custom_llm_provider,
+            logging_obj=logging_obj,
+        )
+        return self._apply_post_stream_processing(streamwrapper, model, custom_llm_provider)
+
     @staticmethod
     def _apply_post_stream_processing(
         stream: "CustomStreamWrapper",

litellm/completion_extras/litellm_responses_transformation/transformation.py

@@ -1077,6 +1077,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
 class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
     def __init__(self, streaming_response, sync_stream: bool, json_mode: Optional[bool] = False):
         super().__init__(streaming_response, sync_stream, json_mode)
+        self._chat_completion_id: str | None = None
 
     def _handle_string_chunk(
         self, str_line: Union[str, "BaseModel"]
@@ -1384,4 +1385,13 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
             ModelResponseStream: OpenAI-formatted streaming chunk
         """
         verbose_logger.debug(f"Chat provider: transform_streaming_response called with chunk: {chunk}")
-        return OpenAiResponsesToChatCompletionStreamIterator.translate_responses_chunk_to_openai_stream(chunk)
+        return self._with_stream_scoped_id(
+            OpenAiResponsesToChatCompletionStreamIterator.translate_responses_chunk_to_openai_stream(chunk)
+        )
+
+    def _with_stream_scoped_id(self, chunk: "ModelResponseStream") -> "ModelResponseStream":
+        if self._chat_completion_id is None:
+            self._chat_completion_id = chunk.id
+        else:
+            chunk.id = self._chat_completion_id
+        return chunk

tests/e2e/coverage_registry/llm_conversational.yaml

@@ -5,6 +5,9 @@
 - {id: llm.chat_completions.openai.passthrough.nonstream.cost_logged, module: llm, tier: P1, subject_endpoint: chat_completions, route: openai, capability: basic, streaming: nonstream, assertions: [works, cost_logged], source: "test_passthrough_e2e.py", rationale: "OpenAI-format chat via the raw /openai/{endpoint} passthrough (/openai/v1/chat/completions); proxy swaps in OPENAI_API_KEY and still logs a costed pass_through_endpoint row (LIT-4752)"}
 - {id: llm.chat_completions.openai.tool_use.nonstream.works, module: llm, tier: P0, subject_endpoint: chat_completions, route: openai, capability: tool_use, streaming: nonstream, assertions: [works], source: "model_prices json", rationale: "OpenAI function_calling; high usage"}
 - {id: llm.chat_completions.openai.tool_use.stream.works, module: llm, tier: P0, subject_endpoint: chat_completions, route: openai, capability: tool_use, streaming: stream, assertions: [works], source: "model_prices json", rationale: "Tool calls over streaming"}
+- {id: llm.chat_completions.openai.basic.stream.bridge_shares_chunk_id, module: llm, tier: P0, subject_endpoint: chat_completions, route: openai, capability: basic, streaming: stream, assertions: [stable_chunk_id], source: "completion_extras/litellm_responses_transformation/transformation.py", rationale: "A responses-only model served over /chat/completions must stream every chunk under one chat completion id; per-chunk ids make id-accumulating SDKs drop the response", fail_before_fix: proven}
+- {id: llm.chat_completions.openai.basic.stream.bridge_streams_sse, module: llm, tier: P0, subject_endpoint: chat_completions, route: openai, capability: basic, streaming: stream, assertions: [works], source: "completion_extras/litellm_responses_transformation/handler.py", rationale: "The Responses bridge must answer a streaming chat request with real SSE (content deltas, finish_reason, [DONE]), never a completed response the SSE generator cannot iterate"}
+- {id: llm.chat_completions.openai.tool_use.stream.bridge_streams_tool_call, module: llm, tier: P0, subject_endpoint: chat_completions, route: openai, capability: tool_use, streaming: stream, assertions: [works], source: "completion_extras/litellm_responses_transformation/transformation.py", rationale: "Tool calls translated from Responses events must reassemble into one named call with parseable argument JSON over the bridged stream"}
 - {id: llm.chat_completions.openai.vision.nonstream.works, module: llm, tier: P0, subject_endpoint: chat_completions, route: openai, capability: vision, streaming: nonstream, assertions: [works], source: "model_prices json", rationale: "gpt-4o vision; high usage"}
 - {id: llm.chat_completions.openai.prompt_cache_5m.nonstream.works, module: llm, tier: P0, subject_endpoint: chat_completions, route: openai, capability: prompt_cache_5m, streaming: nonstream, assertions: [works], source: "model_prices json", rationale: "Prompt caching cost optimization"}
 - {id: llm.chat_completions.openai.service_tier.nonstream.works, module: llm, tier: P1, subject_endpoint: chat_completions, route: openai, capability: service_tier, streaming: nonstream, assertions: [works], source: "OpenAI service_tier param", rationale: "OpenAI scale-tier request option is forwarded and echoed"}

tests/e2e/e2e_http.py

@@ -137,6 +137,7 @@ class StreamingResponse(BaseModel):
     # quota) arrive as SSE error events inside an otherwise-successful response;
     # the consumed body is elided, so this is the only place they surface.
     stream_error: str | None = None
+    stream_done: bool = False
 
     @property
     def ok(self) -> bool:
@@ -408,6 +409,7 @@ def _streaming_outcome(resp: requests.Response, stream: bool) -> StreamingRespon
     chunks = 0
     stream_error: str | None = None
     stream_events: list[str] = []
+    stream_done = False
     for line in lines:
         if not line:
             continue
@@ -415,7 +417,9 @@ def _streaming_outcome(resp: requests.Response, stream: bool) -> StreamingRespon
         decoded_line = line.decode(errors="replace")
         if decoded_line.startswith("data: "):
             payload = decoded_line.removeprefix("data: ")
-            if payload != "[DONE]":
+            if payload == "[DONE]":
+                stream_done = True
+            else:
                 stream_events.append(payload)
         if stream_error is None and (
             line.startswith(b"event: error")
@@ -433,6 +437,7 @@ def _streaming_outcome(resp: requests.Response, stream: bool) -> StreamingRespon
         body="<streamed>",
         chunks=chunks,
         stream_events=stream_events,
+        stream_done=stream_done,
         stream_error=stream_error,
     )
 

tests/e2e/llm_translation/test_responses_bridge_streaming_e2e.py

@@ -0,0 +1,173 @@
+"""Live /chat/completions streaming through the Responses API bridge.
+
+Responses-only models (gpt-5.3-codex here, the same shape as the GPT-5.6 models
+customers reach over bedrock_mantle) cannot serve /chat/completions natively, so the
+proxy translates the request to /v1/responses and translates each Responses event back
+into a chat completion chunk. Two customer-visible contracts only hold on that path:
+
+- every chunk of one stream carries the same ``id`` (#32854). The bridge builds a chunk
+  per Responses event, so a regression there hands each chunk a fresh ``chatcmpl-<uuid>``
+  and SDKs that accumulate by id (openai-go's ChatCompletionAccumulator) silently drop
+  everything after the first chunk while the HTTP response still looks healthy
+- the bridge always answers a streaming request with a real SSE stream (#33154). When it
+  hands back an already-completed response instead, the proxy's SSE generator dies with
+  "'async for' requires an object with __aiter__ method" mid-stream
+"""
+
+from __future__ import annotations
+
+import pytest
+from pydantic import BaseModel
+
+from e2e_config import unique_marker
+from e2e_http import StreamingResponse
+from lifecycle import ResourceManager
+from models import ChatBody, ChatMessage, ChatTool, ChatToolFunction, LiteLLMParamsBody
+from passthrough_client import PassthroughClient
+
+pytestmark = pytest.mark.e2e
+
+RESPONSES_ONLY_BACKEND = "openai/gpt-5.3-codex"
+
+
+class _BridgeToolCallFunction(BaseModel):
+    name: str | None = None
+    arguments: str | None = None
+
+
+class _BridgeToolCall(BaseModel):
+    function: _BridgeToolCallFunction = _BridgeToolCallFunction()
+
+
+class _BridgeDelta(BaseModel):
+    content: str | None = None
+    tool_calls: list[_BridgeToolCall] | None = None
+
+
+class _BridgeChoice(BaseModel):
+    delta: _BridgeDelta = _BridgeDelta()
+    finish_reason: str | None = None
+
+
+class _BridgeChunk(BaseModel):
+    id: str
+    choices: list[_BridgeChoice] = []
+
+
+class _WeatherArgs(BaseModel):
+    location: str
+
+
+_WEATHER_TOOL = ChatTool(
+    function=ChatToolFunction(
+        name="get_weather",
+        description="Get the current weather for a location",
+        parameters={
+            "type": "object",
+            "properties": {"location": {"type": "string"}},
+            "required": ["location"],
+        },
+    )
+)
+
+
+def _bridge_chunks(result: StreamingResponse) -> list[_BridgeChunk]:
+    """Parse the SSE events of a bridged stream, failing loudly on a stream that never
+    established, carried an error event, or delivered no chunks."""
+    assert result.ok and result.is_streaming, f"bridged stream was not established: {result}"
+    assert result.stream_error is None, f"bridged stream carried an error event: {result.stream_error}"
+    chunks = [_BridgeChunk.model_validate_json(event) for event in result.stream_events]
+    assert chunks, f"bridged stream delivered no chunks: {result.body[:500]}"
+    return chunks
+
+
+class TestResponsesBridgeChatCompletionsStreaming:
+    @pytest.fixture
+    def bridged_model(self, client: PassthroughClient, resources: ResourceManager) -> str:
+        model = f"e2e-bridge-stream-{unique_marker()}"
+        model_id = client.proxy.create_model(
+            model,
+            LiteLLMParamsBody(model=RESPONSES_ONLY_BACKEND, api_key="os.environ/OPENAI_API_KEY"),
+        )
+        resources.defer(lambda: client.proxy.delete_model(model_id))
+        return model
+
+    @pytest.mark.covers(
+        "llm.chat_completions.openai.basic.stream.bridge_shares_chunk_id",
+        exercised_on=["chat_completions"],
+    )
+    def test_bridged_stream_shares_one_chunk_id(
+        self, client: PassthroughClient, resources: ResourceManager, bridged_model: str
+    ) -> None:
+        result = client.proxy.chat_stream(
+            resources.key(),
+            ChatBody(
+                model=bridged_model,
+                messages=[ChatMessage(role="user", content=f"Count from 1 to 5, one number per line. {unique_marker()}")],
+                max_tokens=64,
+                stream=True,
+            ),
+        )
+
+        chunks = _bridge_chunks(result)
+        ids = {chunk.id for chunk in chunks}
+        assert len(ids) == 1, f"bridged stream used {len(ids)} different chunk ids: {sorted(ids)[:5]}"
+        assert ids.pop().startswith("chatcmpl-"), f"bridged chunk id is not chat-completion shaped: {chunks[0].id}"
+
+    @pytest.mark.covers(
+        "llm.chat_completions.openai.basic.stream.bridge_streams_sse",
+        exercised_on=["chat_completions"],
+    )
+    def test_bridged_stream_delivers_content_finish_reason_and_done(
+        self, client: PassthroughClient, resources: ResourceManager, bridged_model: str
+    ) -> None:
+        result = client.proxy.chat_stream(
+            resources.key(),
+            ChatBody(
+                model=bridged_model,
+                messages=[ChatMessage(role="user", content=f"Reply with the single word pong. {unique_marker()}")],
+                max_tokens=32,
+                stream=True,
+            ),
+        )
+
+        chunks = _bridge_chunks(result)
+        content = "".join(choice.delta.content or "" for chunk in chunks for choice in chunk.choices)
+        assert content.strip(), f"bridged stream completed with no content deltas: {result.stream_events[:3]}"
+        assert any(
+            choice.finish_reason for chunk in chunks for choice in chunk.choices
+        ), f"bridged stream never emitted a finish_reason: {result.stream_events[-3:]}"
+        assert result.stream_done, f"bridged stream did not terminate with [DONE]: {result.stream_events[-2:]}"
+
+    @pytest.mark.covers(
+        "llm.chat_completions.openai.tool_use.stream.bridge_streams_tool_call",
+        exercised_on=["chat_completions"],
+    )
+    def test_bridged_stream_reassembles_tool_call(
+        self, client: PassthroughClient, resources: ResourceManager, bridged_model: str
+    ) -> None:
+        result = client.proxy.chat_stream(
+            resources.key(),
+            ChatBody(
+                model=bridged_model,
+                messages=[
+                    ChatMessage(
+                        role="user",
+                        content="What is the weather in San Francisco? Use the get_weather tool.",
+                    )
+                ],
+                tools=[_WEATHER_TOOL],
+                tool_choice="required",
+                max_tokens=256,
+                stream=True,
+            ),
+        )
+
+        chunks = _bridge_chunks(result)
+        calls = [call for chunk in chunks for choice in chunk.choices for call in (choice.delta.tool_calls or [])]
+        assert calls, f"bridged stream returned no tool call for a tool-forced prompt: {result.stream_events[:5]}"
+        name = "".join(call.function.name or "" for call in calls)
+        arguments = "".join(call.function.arguments or "" for call in calls)
+        assert name == "get_weather", f"bridged stream streamed the wrong tool name: {name!r}"
+        args = _WeatherArgs.model_validate_json(arguments)
+        assert args.location.strip(), f"bridged tool call arguments missing location: {arguments!r}"

tests/test_litellm/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py

@@ -203,3 +203,65 @@ async def test_acompletion_preserves_top_level_stream_flag_in_responses_request(
 
     assert result is stream
     assert transform_request.call_args.kwargs["optional_params"]["stream"] is True
+
+
+def _completed_chat_response() -> ModelResponse:
+    return ModelResponse(
+        id="chatcmpl-completed",
+        model="gpt-5.4",
+        choices=[
+            {
+                "index": 0,
+                "message": {"role": "assistant", "content": "pong"},
+                "finish_reason": "stop",
+            }
+        ],
+    )
+
+
+@pytest.mark.asyncio
+async def test_acompletion_streams_completed_model_response():
+    """A streaming request whose bridge call comes back already completed must still be
+    handed back as an async-iterable stream. Returning the bare ModelResponse crashed the
+    proxy's SSE generator with "'async for' requires an object with __aiter__ method".
+    Regression for #33154."""
+    completed = _completed_chat_response()
+    bridge = ResponsesToCompletionBridgeHandler()
+
+    with (
+        patch.object(
+            bridge.transformation_handler,
+            "transform_request",
+            return_value={"model": "gpt-5.4", "input": "hi"},
+        ),
+        patch("litellm.aresponses", new=AsyncMock(return_value=completed)),
+    ):
+        result = await bridge.acompletion(**_bridge_kwargs(stream=True))
+
+    assert isinstance(result, CustomStreamWrapper), f"streaming request got {type(result)}"
+    chunks = [chunk async for chunk in result]
+    assert "".join(
+        chunk.choices[0].delta.content or "" for chunk in chunks
+    ) == "pong", f"completed response did not stream its content: {chunks}"
+    assert [c for c in chunks if c.choices[0].finish_reason], "stream never emitted a finish_reason"
+
+
+def test_completion_streams_completed_model_response():
+    completed = _completed_chat_response()
+    bridge = ResponsesToCompletionBridgeHandler()
+
+    with (
+        patch.object(
+            bridge.transformation_handler,
+            "transform_request",
+            return_value={"model": "gpt-5.4", "input": "hi"},
+        ),
+        patch("litellm.responses", return_value=completed),
+    ):
+        result = bridge.completion(**_bridge_kwargs(stream=True))
+
+    assert isinstance(result, CustomStreamWrapper), f"streaming request got {type(result)}"
+    chunks = list(result)
+    assert "".join(chunk.choices[0].delta.content or "" for chunk in chunks) == "pong", (
+        f"completed response did not stream its content: {chunks}"
+    )

tests/test_litellm/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py

@@ -2855,6 +2855,41 @@ def test_streaming_function_call_tool_id_for_degenerate_call_id():
     assert stream_tool_id("fc_2", "call_tokyo") == "call_tokyo"
 
 
+def test_streaming_chunks_share_one_chat_completion_id():
+    """Every chunk of one streamed chat completion must carry the same ``id``, per the
+    OpenAI spec. The bridge builds a fresh ``ModelResponseStream`` per Responses event,
+    so without a stream-scoped id each chunk got a new ``chatcmpl-<uuid>`` and clients
+    that validate id consistency (openai-go's ChatCompletionAccumulator) silently
+    dropped every chunk after the first. Regression for #32854."""
+    from litellm.completion_extras.litellm_responses_transformation.transformation import (
+        OpenAiResponsesToChatCompletionStreamIterator,
+    )
+
+    iterator = OpenAiResponsesToChatCompletionStreamIterator(
+        streaming_response=None, sync_stream=True
+    )
+    events = [
+        {"type": "response.created", "response": {"id": "resp_abc", "output": []}},
+        {"type": "response.output_text.delta", "delta": "Hel"},
+        {"type": "response.output_text.delta", "delta": "lo"},
+        {
+            "type": "response.completed",
+            "response": {"id": "resp_abc", "output": [{"type": "message"}]},
+        },
+    ]
+
+    ids = [iterator.chunk_parser(event).id for event in events]
+
+    assert len(set(ids)) == 1, f"streamed chunks carried different ids: {ids}"
+    assert ids[0], "streamed chunks carried an empty id"
+
+    other_stream = OpenAiResponsesToChatCompletionStreamIterator(
+        streaming_response=None, sync_stream=True
+    )
+    assert (
+        other_stream.chunk_parser(events[1]).id != ids[0]
+    ), "a separate stream must get its own id, not a process-wide one"
+
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize(