From f343c2421a3554a43720420b3b7d104a7103986d Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Wed, 28 Oct 2026 12:01:53 -0700 Subject: [PATCH] fix(proxy): keep the reason when a background Responses request fails A background poll of a request a post-call guardrail blocked saved a null code and message, because the guardrail's error event nests them. A request whose first chunk was an error polled as completed with no error. Both now poll failed with the same code and message a late failure gets. --- .../common_utils/responses_stream_errors.py | 12 ++- .../response_polling/background_streaming.py | 21 +++-- .../proxy/response_polling/polling_handler.py | 2 +- .../test_response_polling_handler.py | 90 +++++++++++++++++++ 4 files changed, 116 insertions(+), 9 deletions(-) diff --git a/litellm/proxy/common_utils/responses_stream_errors.py b/litellm/proxy/common_utils/responses_stream_errors.py index 05ec7603a3d..c730a94dc62 100644 --- a/litellm/proxy/common_utils/responses_stream_errors.py +++ b/litellm/proxy/common_utils/responses_stream_errors.py @@ -126,6 +126,14 @@ def _response_error_code(details: _FailureDetails) -> str: return _status_error_code(details.status_code) +def responses_error_fields(error: Mapping[str, object]) -> Mapping[str, object]: + details: Final = _FailureDetails.model_validate(error) + return { + "code": _response_error_code(details.model_copy(update={"status_code": _status_of(details.code)})), + "message": details.message, + } + + class ResponsesStreamErrorState: def __init__(self) -> None: self.response_id: str | None = None @@ -157,12 +165,10 @@ class ResponsesStreamErrorState: return prefix.encode("utf-8") + frame if isinstance(frame, bytes) else prefix + frame def format_error(self, error: Mapping[str, object]) -> str: - details: Final = _FailureDetails.model_validate(error) payload: Final = json.dumps( { "type": ResponsesAPIStreamEvents.ERROR.value, - "code": _response_error_code(details.model_copy(update={"status_code": _status_of(details.code)})), - "message": details.message, + **responses_error_fields(error), "param": error.get("param"), "sequence_number": self.sequence_number + 1, "error": error, diff --git a/litellm/proxy/response_polling/background_streaming.py b/litellm/proxy/response_polling/background_streaming.py index 7411e204d90..ecf9bea4ab7 100644 --- a/litellm/proxy/response_polling/background_streaming.py +++ b/litellm/proxy/response_polling/background_streaming.py @@ -11,6 +11,7 @@ https://platform.openai.com/docs/api-reference/responses-streaming import asyncio import json from collections.abc import AsyncIterable, Callable, Mapping, Sequence +from http import HTTPStatus from typing import TYPE_CHECKING, Final, Protocol, TypeAlias, runtime_checkable from fastapi import Request, Response @@ -20,6 +21,7 @@ from typing_extensions import ReadOnly, TypedDict from litellm._logging import verbose_proxy_logger from litellm.proxy.auth.user_api_key_auth import UserAPIKeyAuth from litellm.proxy.common_request_processing import RESPONSES_LATE_SSE_FRAMING, ProxyBaseLLMRequestProcessing +from litellm.proxy.common_utils.responses_stream_errors import responses_error_fields, sse_error_object from litellm.proxy.response_polling.polling_handler import ResponsePollingHandler from litellm.types.llms.openai import ResponsesAPIStatus @@ -85,6 +87,14 @@ def _sse_frame_data(frame: str) -> str | None: return next((line[6:].strip() for line in frame.splitlines() if line.startswith("data: ")), None) +def _first_chunk_error(response: object) -> Mapping[str, object] | None: + if isinstance(response, _StreamsBody) or not isinstance(response, Response): + return None + if response.status_code < HTTPStatus.BAD_REQUEST: + return None + return responses_error_fields(sse_error_object(bytes(response.body)) or {"code": response.status_code}) + + async def _never_receive() -> Message: await asyncio.Event().wait() raise AssertionError("unreachable") @@ -329,7 +339,7 @@ async def background_streaming_task( elif event_type == "error": terminal_status = "failed" - terminal_error = {"code": event.get("code"), "message": event.get("message")} + terminal_error = responses_error_fields(event.get("error") or event) elif event_type in ( "response.completed", @@ -387,14 +397,15 @@ async def background_streaming_task( # Final flush to ensure all accumulated state is saved await flush_state_if_needed(force=True) - # Use the terminal status from the stream, default to "completed" - final_status: Final = terminal_status or "completed" + early_error: Final = _first_chunk_error(response) + final_error: Final = terminal_error if early_error is None else early_error + final_status: Final = "failed" if early_error is not None else terminal_status or "completed" await polling_handler.update_state( polling_id=polling_id, status=final_status, usage=usage_data, - error=terminal_error, + error=final_error, reasoning=reasoning_data, tool_choice=tool_choice_data, tools=tools_data, @@ -416,7 +427,7 @@ async def background_streaming_task( "Finished background streaming for %s, status=%s, error=%s, incomplete_details=%s, output_items=%s", polling_id, final_status, - terminal_error, + final_error, incomplete_details_data, len(output_items), ) diff --git a/litellm/proxy/response_polling/polling_handler.py b/litellm/proxy/response_polling/polling_handler.py index e50587b1a0e..158b1f0f63b 100644 --- a/litellm/proxy/response_polling/polling_handler.py +++ b/litellm/proxy/response_polling/polling_handler.py @@ -98,7 +98,7 @@ class ResponsePollingHandler: polling_id: str, status: ResponsesAPIStatus | None = None, usage: dict | None = None, - error: dict | None = None, + error: Mapping[str, object] | None = None, incomplete_details: dict | None = None, reasoning: dict | None = None, tool_choice: object | None = None, diff --git a/tests/unit/proxy/response_polling/test_response_polling_handler.py b/tests/unit/proxy/response_polling/test_response_polling_handler.py index f398d5acb1c..6cb9376c0e5 100644 --- a/tests/unit/proxy/response_polling/test_response_polling_handler.py +++ b/tests/unit/proxy/response_polling/test_response_polling_handler.py @@ -1553,6 +1553,96 @@ class TestBackgroundStreamingTerminalEvents: assert final_call.kwargs["status"] == "failed" assert final_call.kwargs["error"] == {"code": "rate_limit_exceeded", "message": "rate limited"} + @pytest.mark.asyncio + async def test_a_nested_only_error_event_keeps_its_reason(self): + from litellm.proxy.response_polling.background_streaming import ( + background_streaming_task, + ) + + nested_error: Final = {"type": "guardrail_error", "code": "400", "message": "output blocked by policy"} + error_event: Final = {"type": "error", "sequence_number": 9, "error": nested_error} + + async def _body_iterator(): + yield b'data: {"type": "response.in_progress"}\n\n' + yield f"event: error\ndata: {json.dumps(error_event)}\n\n".encode() + + mock_response: Final = Mock() + mock_response.body_iterator = _body_iterator() + handler: Final = AsyncMock(spec=ResponsePollingHandler) + kwargs: Final = _make_background_streaming_kwargs("poll_nested_error", handler) + + with patch( # test-quality-ok: the processor is built inside the task, same idiom as the sibling tests + "litellm.proxy.response_polling.background_streaming.ProxyBaseLLMRequestProcessing" + ) as MockProcessor: + MockProcessor.return_value.base_process_llm_request = AsyncMock( + return_value=mock_response + ) + await background_streaming_task(**kwargs) + + final_call: Final = handler.update_state.call_args_list[-1] + assert final_call.kwargs["status"] == "failed" + assert final_call.kwargs["error"] == {"code": "invalid_request_error", "message": "output blocked by policy"} + + @pytest.mark.asyncio + @pytest.mark.parametrize( + ("status_code", "body", "expected_error"), + [ + ( + 404, + json.dumps( + {"error": {"message": "model not found", "type": "invalid_request_error", "param": None, "code": "404"}} + ), + {"code": "not_found_error", "message": "model not found"}, + ), + (429, "Too Many Requests", {"code": "rate_limit_exceeded", "message": None}), + ], + ) + async def test_a_first_chunk_error_polls_failed_like_a_late_one(self, status_code, body, expected_error): + from fastapi import Response + + from litellm.proxy.response_polling.background_streaming import ( + background_streaming_task, + ) + + first_chunk_error: Final = Response(status_code=status_code, content=body) + handler: Final = AsyncMock(spec=ResponsePollingHandler) + kwargs: Final = _make_background_streaming_kwargs("poll_first_chunk_error", handler) + + with patch( # test-quality-ok: the processor is built inside the task, same idiom as the sibling tests + "litellm.proxy.response_polling.background_streaming.ProxyBaseLLMRequestProcessing" + ) as MockProcessor: + MockProcessor.return_value.base_process_llm_request = AsyncMock( + return_value=first_chunk_error + ) + await background_streaming_task(**kwargs) + + final_call: Final = handler.update_state.call_args_list[-1] + assert final_call.kwargs["status"] == "failed" + assert final_call.kwargs["error"] == expected_error + + @pytest.mark.asyncio + async def test_a_non_streamed_success_response_still_polls_completed(self): + from fastapi.responses import JSONResponse + + from litellm.proxy.response_polling.background_streaming import ( + background_streaming_task, + ) + + handler: Final = AsyncMock(spec=ResponsePollingHandler) + kwargs: Final = _make_background_streaming_kwargs("poll_non_streamed_success", handler) + + with patch( # test-quality-ok: the processor is built inside the task, same idiom as the sibling tests + "litellm.proxy.response_polling.background_streaming.ProxyBaseLLMRequestProcessing" + ) as MockProcessor: + MockProcessor.return_value.base_process_llm_request = AsyncMock( + return_value=JSONResponse(status_code=200, content={"id": "resp_123", "status": "completed"}) + ) + await background_streaming_task(**kwargs) + + final_call: Final = handler.update_state.call_args_list[-1] + assert final_call.kwargs["status"] == "completed" + assert final_call.kwargs["error"] is None + @pytest.mark.asyncio async def test_response_incomplete_sets_incomplete_status_and_details(self): """Test that a response.incomplete stream event results in incomplete status"""