From df60e36d07bd39d765c1d737ba98313a3214b06b Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Wed, 15 Jul 2026 08:33:50 -0700 Subject: [PATCH] fix(responses): end stream cleanly on transport error after terminal event --- litellm/responses/streaming_iterator.py | 12 ++ .../responses/test_streaming_iterator.py | 143 ++++++++++++++++-- 2 files changed, 140 insertions(+), 15 deletions(-) diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index eb78e6f9c8d..357a7ecefe6 100644 --- a/litellm/responses/streaming_iterator.py +++ b/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 diff --git a/tests/test_litellm/responses/test_streaming_iterator.py b/tests/test_litellm/responses/test_streaming_iterator.py index 3e4b07fce18..5b0f40fdf27 100644 --- a/tests/test_litellm/responses/test_streaming_iterator.py +++ b/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