From 5ed9814fab2c243be825fd01964d73f8522630e5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=A3=8E=E6=9C=88?= <1759519137@qq.com> Date: Sat, 10 Oct 2026 15:03:31 +0800 Subject: [PATCH] fix(proxy): emit Responses-shaped error frame on /v1/responses stream failures (#45406) --- litellm/proxy/proxy_server.py | 26 +++++- .../proxy_server/test_streaming_helpers.py | 84 ++++++++++++++++++- 2 files changed, 106 insertions(+), 4 deletions(-) diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index dac46855fd6..ff5cbdb46f4 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -10061,18 +10061,37 @@ def _make_keepalive_resolver(request_data: Mapping[str, object]) -> Callable[[ob return _resolve +def _is_responses_stream_route(request: Request | None) -> bool: + """True when `request` targets a Responses API route (`/v1/responses`, ...). + + Only the native `POST /v1/responses` route binds + `responses_stream_errors=True`; callers that serve the same routes through + the shared generator (other `/responses*` endpoints, wrappers around + `select_data_generator`) don't. The request itself carries the route, so + the frame shape is derived from it rather than leaking the generic + chat-completions `{"error": ...}` frame to Responses clients. + """ + path: Final = getattr(getattr(request, "url", None), "path", None) + if not isinstance(path, str): + return False + return any(segment == "responses" for segment in path.split("/")) + + async def async_data_generator( response, user_api_key_dict: UserAPIKeyAuth, request_data: dict, request: Request | None = None, *, - responses_stream_errors: bool = False, + responses_stream_errors: bool | None = None, ): verbose_proxy_logger.debug("inside generator") stream_completed = False client_disconnected = False - error_state: Final = ResponsesStreamErrorState() if responses_stream_errors else None + serve_responses_stream_errors: Final = ( + _is_responses_stream_route(request) if responses_stream_errors is None else responses_stream_errors + ) + error_state: Final = ResponsesStreamErrorState() if serve_responses_stream_errors else None needs_iterator_wrap: Final = proxy_logging_obj.needs_iterator_wrap() stream_iterator: Final[AsyncIterator[object]] = ( proxy_logging_obj.async_post_call_streaming_iterator_hook( @@ -10323,8 +10342,9 @@ def select_data_generator( request_data: dict, request: Request | None = None, *, - responses_stream_errors: bool = False, + responses_stream_errors: bool | None = None, ): + """`responses_stream_errors=None` auto-detects Responses routes from `request`.""" return async_data_generator( response=response, user_api_key_dict=user_api_key_dict, diff --git a/tests/unit/proxy/proxy_server/test_streaming_helpers.py b/tests/unit/proxy/proxy_server/test_streaming_helpers.py index 09d71cb4ead..b161f387c32 100644 --- a/tests/unit/proxy/proxy_server/test_streaming_helpers.py +++ b/tests/unit/proxy/proxy_server/test_streaming_helpers.py @@ -23,7 +23,7 @@ from unittest.mock import AsyncMock, MagicMock import httpx import pytest -from fastapi import HTTPException, Response +from fastapi import HTTPException, Request, Response from fastapi.responses import StreamingResponse from openai import APIError as OpenAIAPIError from pydantic import BaseModel @@ -1044,11 +1044,93 @@ async def test_responses_stream_keeps_tool_deltas_and_only_emits_a_valid_termina assert "error" not in payloads[-1] +def _http_request(path: str) -> Request: + return Request( + { + "type": "http", + "method": "POST", + "path": path, + "headers": [], + "query_string": b"", + "scheme": "http", + "server": ("testserver", 80), + } + ) + + +async def _responses_upstream_failing_mid_stream() -> AsyncIterator[BaseModel]: + response: Final = ResponsesAPIResponse(id="resp_visible", created_at=1, model="gpt-6-astra", output=[]) + yield ResponseCreatedEvent.model_validate( + {"type": "response.created", "sequence_number": 0, "response": response} + ) + raise litellm.ServiceUnavailableError( + message="Our servers are currently overloaded. Please try again later.", + llm_provider="openai", + model="gpt-6-astra", + ) + + +async def _frames_for_route(request: Request, **kwargs: object) -> tuple[str, ...]: + frames: Final = [ + frame + async for frame in select_data_generator( + response=_responses_upstream_failing_mid_stream(), + user_api_key_dict=_user_auth(), + request_data={}, + request=request, + **kwargs, + ) + ] + return tuple(frame.decode() if isinstance(frame, bytes) else frame for frame in frames) + + # --------------------------------------------------------------------------- # select_data_generator # --------------------------------------------------------------------------- +@pytest.mark.asyncio +@pytest.mark.parametrize("path", ["/v1/responses", "/responses", "/openai/v1/responses"]) +async def test_responses_route_without_explicit_flag_emits_responses_error_frame(path: str) -> None: + """issue #45406: a flag-less caller streaming `/` must not end + the stream with the chat-completions `{"error": ...}` frame. The route in the + request selects the Responses terminal event, whose payload carries the typed + error (and, unlike the generic frame, a `type` field).""" + decoded: Final = await _frames_for_route(_http_request(path)) + + assert decoded[-1] == "data: [DONE]\n\n" + failure_frame: Final = decoded[-2] + assert failure_frame.startswith("event: response.failed\n") + failure: Final = ResponseFailedEvent.model_validate( + json.loads(next(line[6:] for line in failure_frame.splitlines() if line.startswith("data: "))) + ) + assert failure.response.id == "resp_visible" + assert failure.response.status == "failed" + assert failure.response.error is not None + assert failure.response.error["code"] == "server_error" + assert "Our servers are currently overloaded" in failure.response.error["message"] + assert not any(frame.startswith('data: {"error"') for frame in decoded) + + +@pytest.mark.asyncio +async def test_chat_route_without_explicit_flag_keeps_generic_error_frame() -> None: + """Only Responses routes change shape: `/v1/chat/completions` streams keep the + historical `data: {"error": ...}` frame.""" + decoded: Final = await _frames_for_route(_http_request("/v1/chat/completions")) + + assert any(frame.startswith('data: {"error"') for frame in decoded) + assert not any(frame.startswith("event: response.failed") for frame in decoded) + + +@pytest.mark.asyncio +async def test_responses_route_explicit_opt_out_keeps_generic_error_frame() -> None: + """An explicit `responses_stream_errors=False` wins over route detection.""" + decoded: Final = await _frames_for_route(_http_request("/v1/responses"), responses_stream_errors=False) + + assert any(frame.startswith('data: {"error"') for frame in decoded) + assert not any(frame.startswith("event: response.failed") for frame in decoded) + + @pytest.mark.asyncio async def test_select_data_generator_returns_async_generator(monkeypatch): _patch_logging_flags(monkeypatch)