From 86360077c637cd915f4bee9e375ddc7f233218c8 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Wed, 28 Oct 2026 21:57:45 -0700 Subject: [PATCH] fix(proxy): keep the HTTP status on late Responses errors and end Responses streams on error events A late error body without a code now takes the response's status, an upstream error event counts as the stream's terminal event, and a Responses stream that fails before its first chunk ends with an error event instead of data: [DONE]. --- litellm/proxy/common_request_processing.py | 40 ++++++++++++++----- .../common_utils/responses_stream_errors.py | 2 +- .../proxy_server/test_streaming_helpers.py | 27 +++++++++++++ .../proxy/test_common_request_processing.py | 38 ++++++++++++++++-- 4 files changed, 93 insertions(+), 14 deletions(-) diff --git a/litellm/proxy/common_request_processing.py b/litellm/proxy/common_request_processing.py index 350f4d6cb9c..27164ba0c83 100644 --- a/litellm/proxy/common_request_processing.py +++ b/litellm/proxy/common_request_processing.py @@ -1196,6 +1196,10 @@ def _sse_error_frames(error_obj: Mapping[str, object]) -> tuple[str, str]: return f"data: {json.dumps({'error': error_obj}, default=str)}\n\n", "data: [DONE]\n\n" +def _responses_error_frames(error_obj: Mapping[str, object]) -> tuple[str, ...]: + return (ResponsesStreamErrorState().format_error(error_obj),) + + @dataclass(frozen=True, slots=True) class LateSseFraming: """How a response the keepalive already opened ends once its upstream fails late.""" @@ -1216,14 +1220,14 @@ def _openai_late_body_frames(produced: Response) -> tuple[bytes, ...]: def _responses_late_failure_frames(exc: Exception) -> tuple[str, ...]: if isinstance(exc, ModifyResponseException): return tuple(chunk.decode() for chunk in OpenAIResponsesHandler().build_block_sse_chunks(exc)) - _, error_obj = sse_error_payload(exc) - return (ResponsesStreamErrorState().format_error(error_obj),) + return _responses_error_frames(sse_error_payload(exc)[1]) def _responses_late_body_frames(produced: Response) -> tuple[bytes, ...]: body: Final = bytes(produced.body) - error_obj: Final = sse_error_object(body or b"{}") or {"message": body.decode(), "code": str(produced.status_code)} - return (ResponsesStreamErrorState().format_error(error_obj).encode(),) + error_obj: Final = sse_error_object(body or b"{}") or {"message": body.decode()} + with_status: Final = {**error_obj, "code": error_obj.get("code") or str(produced.status_code)} + return tuple(frame.encode() for frame in _responses_error_frames(with_status)) OPENAI_LATE_SSE_FRAMING: Final = LateSseFraming(failure=_openai_late_failure_frames, body=_openai_late_body_frames) @@ -1232,6 +1236,20 @@ RESPONSES_LATE_SSE_FRAMING: Final = LateSseFraming( ) +@dataclass(frozen=True, slots=True) +class FirstChunkSseFraming: + """How a stream's first chunk is read for an error, and how the stream ends when reading it fails.""" + + data_payload: SseDataPayload + error_frames: Callable[[Mapping[str, object]], tuple[str, ...]] + + +OPENAI_FIRST_CHUNK_FRAMING: Final = FirstChunkSseFraming(data_payload=_sse_data_payload, error_frames=_sse_error_frames) +RESPONSES_FIRST_CHUNK_FRAMING: Final = FirstChunkSseFraming( + data_payload=_responses_event_data_payload, error_frames=_responses_error_frames +) + + def _sse_stream_headers(headers: Mapping[str, str]) -> Mapping[str, str]: """`headers` plus the two that stop reverse proxies from buffering SSE (issue #28384).""" return MappingProxyType({**headers, **_TTFT_KEEPALIVE_HEADERS}) @@ -1260,7 +1278,7 @@ async def create_response( request: Request | None = None, refresh_headers: Callable[[], Awaitable[Mapping[str, str]]] | None = None, general_settings: Mapping[str, object] = _NO_GENERAL_SETTINGS, - first_chunk_payload: SseDataPayload = _sse_data_payload, + framing: FirstChunkSseFraming = OPENAI_FIRST_CHUNK_FRAMING, ) -> StreamingResponse | JSONResponse: """ Create streaming response, checking if the first chunk is an error. @@ -1296,7 +1314,9 @@ async def create_response( if first_chunk_value is not None: try: - error_code_from_chunk: Final = await _parse_event_data_for_error(first_chunk_value, first_chunk_payload) + error_code_from_chunk: Final = await _parse_event_data_for_error( + first_chunk_value, framing.data_payload + ) if error_code_from_chunk is not None: # First chunk is an error, stream hasn't really started yet # Should return standard JSON error response instead of SSE format @@ -1309,7 +1329,7 @@ async def create_response( # Parse error content error_dict: Final = with_call_id( JSON_OBJECT.validate_python( - _extract_error_from_sse_chunk(first_chunk_value, first_chunk_payload) + _extract_error_from_sse_chunk(first_chunk_value, framing.data_payload) ), error_body_call_id(general_settings, resolved_headers.get(LITELLM_CALL_ID_HEADER)), ) @@ -1352,7 +1372,7 @@ async def create_response( error_status, error_obj = sse_error_payload(e) async def error_gen_message() -> AsyncGenerator[str, None]: - for frame in _sse_error_frames(error_obj): + for frame in framing.error_frames(error_obj): yield frame return StreamingResponse( @@ -3286,8 +3306,8 @@ class ProxyBaseLLMRequestProcessing: request=request, refresh_headers=refresh_stream_headers, general_settings=general_settings, - first_chunk_payload=( - _responses_event_data_payload if route_type == "aresponses" else _sse_data_payload + framing=( + RESPONSES_FIRST_CHUNK_FRAMING if route_type == "aresponses" else OPENAI_FIRST_CHUNK_FRAMING ), ) diff --git a/litellm/proxy/common_utils/responses_stream_errors.py b/litellm/proxy/common_utils/responses_stream_errors.py index 560365a4cdb..4a806c32284 100644 --- a/litellm/proxy/common_utils/responses_stream_errors.py +++ b/litellm/proxy/common_utils/responses_stream_errors.py @@ -172,7 +172,7 @@ class ResponsesStreamErrorState: self.model = event.response.model or self.model if event.response.created_at is not None: self.created_at = event.response.created_at - if event.type in ("response.completed", "response.failed", "response.incomplete"): + if event.type in ("response.completed", "response.failed", "response.incomplete", "error"): self.terminal_emitted = True if not event.type or "\n" in event.type or "\r" in event.type: return frame diff --git a/tests/unit/proxy/proxy_server/test_streaming_helpers.py b/tests/unit/proxy/proxy_server/test_streaming_helpers.py index fb8ff519fbc..93eb8fa226b 100644 --- a/tests/unit/proxy/proxy_server/test_streaming_helpers.py +++ b/tests/unit/proxy/proxy_server/test_streaming_helpers.py @@ -146,6 +146,33 @@ async def test_responses_stream_dispatches_named_events_without_a_chat_done_sent assert payload["type"] == event_type assert json.loads(frames[1].split("data: ", 1)[1])["delta"] == "hello" + +@pytest.mark.asyncio +async def test_responses_stream_sends_no_failure_event_after_an_upstream_error_event() -> None: + response: Final = ResponsesAPIResponse(id="resp_error_then_raise", created_at=1, model="local-model", output=[]) + created: Final = ResponseCreatedEvent.model_validate( + {"type": "response.created", "sequence_number": 0, "response": response} + ) + error_event: Final = ErrorEvent.model_validate( + {"type": "error", "sequence_number": 1, "code": "server_error", "message": "upstream failed", "param": None} + ) + + async def upstream() -> AsyncIterator[BaseModel]: + yield created + yield error_event + raise litellm.APIError(status_code=500, message="Stream close failed", llm_provider="openai", model="local-model") + + frames: Final = tuple( + [ + frame.decode() if isinstance(frame, bytes) else frame + async for frame in select_data_generator( + response=upstream(), user_api_key_dict=_user_auth(), request_data={}, responses_stream_errors=True + ) + ] + ) + + assert tuple(frame.split("\n", 1)[0] for frame in frames) == ("event: response.created", "event: error") + @pytest.mark.asyncio async def test_a_guardrail_error_string_mid_responses_stream_ends_it_as_a_named_error_event() -> None: response: Final = ResponsesAPIResponse(id="resp_hook_error", created_at=1, model="local-model", output=[]) diff --git a/tests/unit/proxy/test_common_request_processing.py b/tests/unit/proxy/test_common_request_processing.py index 1b1c7ce1ccb..3f8ff1c9b14 100644 --- a/tests/unit/proxy/test_common_request_processing.py +++ b/tests/unit/proxy/test_common_request_processing.py @@ -50,7 +50,8 @@ from litellm.proxy.common_request_processing import ( _get_cost_breakdown_from_logging_obj, CostBreakdownHeaderValues, _has_attribute_error_in_chain, - _responses_event_data_payload, + OPENAI_FIRST_CHUNK_FRAMING, + RESPONSES_FIRST_CHUNK_FRAMING, include_guardrail_response_requested, is_azure_model_router_request, RESPONSES_LATE_SSE_FRAMING, @@ -2250,6 +2251,14 @@ class TestProxyBaseLLMRequestProcessing: ) +def _fail_stream_setup() -> str: + raise RuntimeError("stream setup failed") + + +async def _stream_failing_before_its_first_chunk() -> AsyncGenerator[str, None]: + yield _fail_stream_setup() + + @pytest.mark.asyncio class TestCommonRequestProcessingHelpers: async def consume_stream(self, streaming_response: StreamingResponse) -> list: @@ -2335,7 +2344,7 @@ class TestCommonRequestProcessingHelpers: yield f"event: error\ndata: {json.dumps({'type': 'error', 'sequence_number': 0, 'error': error})}\n\n" response: Final = await create_response( - responses_stream(), "text/event-stream", {}, first_chunk_payload=_responses_event_data_payload + responses_stream(), "text/event-stream", {}, framing=RESPONSES_FIRST_CHUNK_FRAMING ) assert isinstance(response, JSONResponse) @@ -2350,13 +2359,30 @@ class TestCommonRequestProcessingHelpers: yield f"event: error\ndata: {event}\n\n" response: Final = await create_response( - responses_stream(), "text/event-stream", {}, first_chunk_payload=_responses_event_data_payload + responses_stream(), "text/event-stream", {}, framing=RESPONSES_FIRST_CHUNK_FRAMING ) assert isinstance(response, JSONResponse) assert response.status_code == 400 assert json.loads(response.body) == {"error": error} + @pytest.mark.parametrize( + "framing,frame_starts", + [ + (OPENAI_FIRST_CHUNK_FRAMING, ('data: {"error": ', "data: [DONE]")), + (RESPONSES_FIRST_CHUNK_FRAMING, ('event: error\ndata: {"type": "error", "code": "server_error"',)), + ], + ) + async def test_a_stream_that_fails_before_its_first_chunk_ends_in_its_own_api_shape(self, framing, frame_starts): + response: Final = await create_response( + _stream_failing_before_its_first_chunk(), "text/event-stream", {}, framing=framing + ) + frames: Final = await self.consume_stream(response) + + assert response.status_code == 500 + assert len(frames) == len(frame_starts) + assert all(frame.startswith(start) for frame, start in zip(frames, frame_starts, strict=True)) + async def test_create_streaming_response_first_chunk_not_error(self): async def mock_generator(): yield 'data: {"content": "first part"}\n\n' @@ -10591,6 +10617,11 @@ _POLICY_BLOCK_DETAIL: Final = MappingProxyType( ) +async def _slow_codeless_rate_limit_body() -> Response: + await asyncio.sleep(0.2) + return JSONResponse(status_code=429, content={"error": {"message": "slow down", "code": None}}) + + async def _slow_dict_detail_failure() -> Response: await asyncio.sleep(0.2) raise HTTPException(status_code=400, detail=dict(_POLICY_BLOCK_DETAIL)) @@ -10600,6 +10631,7 @@ async def _slow_dict_detail_failure() -> Response: "produce,expected_code,expected_error", [ (_slow_rate_limit_failure, "rate_limit_exceeded", {"message": "rate limited", "code": "429"}), + (_slow_codeless_rate_limit_body, "rate_limit_exceeded", {"message": "slow down", "code": "429"}), ( _slow_guardrail_block_body, "invalid_request_error",