fix(responses): end stream cleanly on transport error after terminal event

litellm/responses/streaming_iterator.py

@@ -718,6 +718,12 @@ class ResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
         except StopAsyncIteration:
             # Normal end of stream - don't log as failure
             raise
+        except (httpx.ReadError, httpx.RemoteProtocolError) as e:
+            self.finished = True
+            if self.completed_response is None:
+                self._handle_failure(e)
+                raise
+            raise StopAsyncIteration from e
         except httpx.HTTPError as e:
             # Handle HTTP errors
             self.finished = True
@@ -794,6 +800,12 @@ class SyncResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
         except StopIteration:
             # Normal end of stream - don't log as failure
             raise
+        except (httpx.ReadError, httpx.RemoteProtocolError) as e:
+            self.finished = True
+            if self.completed_response is None:
+                self._handle_failure(e)
+                raise
+            raise StopIteration from e
         except httpx.HTTPError as e:
             # Handle HTTP errors
             self.finished = True

tests/test_litellm/responses/test_streaming_iterator.py

@@ -5,13 +5,18 @@ completion_start_time = end_time."""
 
 import json
 from datetime import datetime
+from typing import Optional
 from unittest.mock import Mock
 
+import httpx
 import pytest
 
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
 from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig
-from litellm.responses.streaming_iterator import ResponsesAPIStreamingIterator
+from litellm.responses.streaming_iterator import (
+    ResponsesAPIStreamingIterator,
+    SyncResponsesAPIStreamingIterator,
+)
 from litellm.types.llms.openai import (
     ResponseCompletedEvent,
     ResponsesAPIResponse,
@@ -23,19 +28,7 @@ def _sse_event(payload: dict) -> bytes:
     return f"data: {json.dumps(payload)}\n\n".encode("utf-8")
 
 
-def _make_iterator(
-    *,
-    sse_events: list[bytes],
-    logging_obj: LiteLLMLoggingObj,
-) -> ResponsesAPIStreamingIterator:
-    async def aiter_bytes():
-        for evt in sse_events:
-            yield evt
-
-    mock_response = Mock()
-    mock_response.headers = {}
-    mock_response.aiter_bytes = aiter_bytes
-
+def _mock_config() -> Mock:
     mock_config = Mock(spec=BaseResponsesAPIConfig)
     mock_responses_api_response = Mock(spec=ResponsesAPIResponse)
     mock_responses_api_response.id = "resp_ttft"
@@ -52,17 +45,68 @@ def _make_iterator(
         return stub
 
     mock_config.transform_streaming_response.side_effect = _transform
+    return mock_config
+
+
+def _make_iterator(
+    *,
+    sse_events: list[bytes],
+    logging_obj: LiteLLMLoggingObj,
+    trailing_error: Optional[Exception] = None,
+) -> ResponsesAPIStreamingIterator:
+    async def aiter_bytes():
+        for evt in sse_events:
+            yield evt
+        if trailing_error is not None:
+            raise trailing_error
+
+    mock_response = Mock()
+    mock_response.headers = {}
+    mock_response.aiter_bytes = aiter_bytes
 
     return ResponsesAPIStreamingIterator(
         response=mock_response,
         model="gpt-4o-mini",
-        responses_api_provider_config=mock_config,
+        responses_api_provider_config=_mock_config(),
         logging_obj=logging_obj,
         litellm_metadata={},
         custom_llm_provider="openai",
     )
 
 
+def _make_sync_iterator(
+    *,
+    sse_events: list[bytes],
+    logging_obj: LiteLLMLoggingObj,
+    trailing_error: Optional[Exception] = None,
+) -> SyncResponsesAPIStreamingIterator:
+    def iter_bytes():
+        for evt in sse_events:
+            yield evt
+        if trailing_error is not None:
+            raise trailing_error
+
+    mock_response = Mock()
+    mock_response.headers = {}
+    mock_response.iter_bytes = iter_bytes
+
+    return SyncResponsesAPIStreamingIterator(
+        response=mock_response,
+        model="gpt-4o-mini",
+        responses_api_provider_config=_mock_config(),
+        logging_obj=logging_obj,
+        litellm_metadata={},
+        custom_llm_provider="openai",
+    )
+
+
+def _logging_obj_stub() -> Mock:
+    logging_obj = Mock(spec=LiteLLMLoggingObj)
+    logging_obj.completion_start_time = None
+    logging_obj.model_call_details = {"litellm_params": {}}
+    return logging_obj
+
+
 @pytest.mark.asyncio
 async def test_responses_streaming_stamps_completion_start_time_on_first_chunk():
     """Without the fix, `logging_obj.completion_start_time` stays None across the
@@ -122,3 +166,72 @@ async def test_responses_streaming_does_not_reset_prior_completion_start_time():
 
     logging_obj._update_completion_start_time.assert_not_called()
     assert logging_obj.completion_start_time == prior
+
+
+_COMPLETE_STREAM_EVENTS = [
+    _sse_event({"type": "response.created"}),
+    _sse_event({"type": "response.output_text.delta", "delta": "hi"}),
+    _sse_event({"type": "response.completed"}),
+]
+
+_TRAILING_ERRORS = [
+    httpx.ReadError("Response payload is not completed"),
+    httpx.RemoteProtocolError("peer closed connection without sending complete message body"),
+]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("trailing_error", _TRAILING_ERRORS, ids=type)
+async def test_transport_error_after_completed_event_ends_stream_cleanly(trailing_error):
+    """A sloppy connection close after `response.completed` must not turn a
+    complete stream into an error (regression guard for the transport no longer
+    swallowing ClientPayloadError/TransferEncodingError)."""
+    iterator = _make_iterator(
+        sse_events=_COMPLETE_STREAM_EVENTS,
+        logging_obj=_logging_obj_stub(),
+        trailing_error=trailing_error,
+    )
+
+    seen = [event.type async for event in iterator]
+
+    assert ResponsesAPIStreamEvents.RESPONSE_COMPLETED in seen
+
+
+@pytest.mark.asyncio
+async def test_transport_error_before_completed_event_raises():
+    """A connection lost before any terminal event is a real failure and must
+    surface, not end the stream as if it completed."""
+    iterator = _make_iterator(
+        sse_events=_COMPLETE_STREAM_EVENTS[:-1],
+        logging_obj=_logging_obj_stub(),
+        trailing_error=httpx.ReadError("Response payload is not completed"),
+    )
+
+    with pytest.raises(httpx.ReadError):
+        async for _ in iterator:
+            pass
+
+
+@pytest.mark.parametrize("trailing_error", _TRAILING_ERRORS, ids=type)
+def test_sync_transport_error_after_completed_event_ends_stream_cleanly(trailing_error):
+    iterator = _make_sync_iterator(
+        sse_events=_COMPLETE_STREAM_EVENTS,
+        logging_obj=_logging_obj_stub(),
+        trailing_error=trailing_error,
+    )
+
+    seen = [event.type for event in iterator]
+
+    assert ResponsesAPIStreamEvents.RESPONSE_COMPLETED in seen
+
+
+def test_sync_transport_error_before_completed_event_raises():
+    iterator = _make_sync_iterator(
+        sse_events=_COMPLETE_STREAM_EVENTS[:-1],
+        logging_obj=_logging_obj_stub(),
+        trailing_error=httpx.ReadError("Response payload is not completed"),
+    )
+
+    with pytest.raises(httpx.ReadError):
+        for _ in iterator:
+            pass