fix(proxy): match OpenAI's Responses error event and stream late passthrough blocks as block responses

litellm/proxy/common_request_processing.py

@@ -51,7 +51,7 @@ from litellm.constants import (
     STREAM_SSE_KEEPALIVE_PING_BYTES,
     UNSAFE_PROXY_RESPONSE_HEADERS,
 )
-from litellm.integrations.custom_guardrail import CustomGuardrail
+from litellm.integrations.custom_guardrail import CustomGuardrail, ModifyResponseException
 from litellm.integrations.otel.runtime import phase_event
 from litellm.litellm_core_utils.bug_report import (
     allowlisted,
@@ -89,6 +89,7 @@ from litellm.litellm_core_utils.streaming_handler import (
     backfill_missing_cache_usage_fields,
 )
 from litellm.litellm_core_utils.tool_fields import DropPolicy, MalformedField, drop_policy, resolve_tool_fields
+from litellm.llms.openai.responses.guardrail_translation.handler import OpenAIResponsesHandler
 from litellm.proxy._types import LiteLLMRoutes, ProxyErrorTypes, ProxyException, UserAPIKeyAuth
 from litellm.proxy.auth.auth_checks import (
     can_key_call_resolved_model,
@@ -1196,6 +1197,8 @@ 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),)
 

litellm/proxy/common_utils/responses_stream_errors.py

@@ -109,6 +109,12 @@ def _status_error_code(status_code: int | None) -> str:
     return _CLIENT_ERROR_CODES.get(status_code, "invalid_request_error")
 
 
+def _status_of(code: str | int | None) -> int | None:
+    if isinstance(code, int):
+        return code
+    return int(code) if code is not None and code.isdecimal() else None
+
+
 def _response_error_code(details: _FailureDetails) -> str:
     for value in (details.code, details.type):
         if value == "insufficient_quota":
@@ -151,9 +157,13 @@ 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,
+                "param": error.get("param"),
                 "sequence_number": self.sequence_number + 1,
                 "error": error,
             },

litellm/proxy/response_polling/background_streaming.py

@@ -329,7 +329,7 @@ async def background_streaming_task(
 
                         elif event_type == "error":
                             terminal_status = "failed"
-                            terminal_error = event.get("error")
+                            terminal_error = {"code": event.get("code"), "message": event.get("message")}
 
                         elif event_type in (
                             "response.completed",

tests/unit/proxy/proxy_server/test_streaming_helpers.py

@@ -172,6 +172,9 @@ async def test_a_guardrail_error_string_mid_responses_stream_ends_it_as_a_named_
     assert event_line == "event: error"
     assert json.loads(data_line.removeprefix("data: ")) == {
         "type": "error",
+        "code": "server_error",
+        "message": "response withheld",
+        "param": None,
         "sequence_number": 1,
         "error": guardrail_error,
     }

tests/unit/proxy/response_polling/test_response_polling_handler.py

@@ -17,7 +17,7 @@ following the OpenAI Response API format.
 import asyncio
 import json
 from datetime import datetime, timezone
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Final, Optional
 from unittest.mock import AsyncMock, Mock, patch
 
 import pytest
@@ -1529,17 +1529,17 @@ class TestBackgroundStreamingTerminalEvents:
             background_streaming_task,
         )
 
-        late_frames = RESPONSES_LATE_SSE_FRAMING.failure(HTTPException(status_code=429, detail="rate limited"))
+        late_frames: Final = RESPONSES_LATE_SSE_FRAMING.failure(HTTPException(status_code=429, detail="rate limited"))
 
         async def _body_iterator():
             yield b": ping\n\n"
             for frame in late_frames:
                 yield frame.encode()
 
-        mock_response = Mock()
+        mock_response: Final = Mock()
         mock_response.body_iterator = _body_iterator()
-        handler = AsyncMock(spec=ResponsePollingHandler)
-        kwargs = _make_background_streaming_kwargs("poll_late_failure", handler)
+        handler: Final = AsyncMock(spec=ResponsePollingHandler)
+        kwargs: Final = _make_background_streaming_kwargs("poll_late_failure", 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"
@@ -1549,10 +1549,9 @@ class TestBackgroundStreamingTerminalEvents:
             )
             await background_streaming_task(**kwargs)
 
-        final_call = handler.update_state.call_args_list[-1]
+        final_call: Final = handler.update_state.call_args_list[-1]
         assert final_call.kwargs["status"] == "failed"
-        assert final_call.kwargs["error"]["message"] == "rate limited"
-        assert final_call.kwargs["error"]["code"] == "429"
+        assert final_call.kwargs["error"] == {"code": "rate_limit_exceeded", "message": "rate limited"}
 
     @pytest.mark.asyncio
     async def test_response_incomplete_sets_incomplete_status_and_details(self):

tests/unit/proxy/test_common_request_processing.py

@@ -34,6 +34,7 @@ from litellm.constants import (
     STREAM_SSE_KEEPALIVE_PING_BYTES,
 )
 from litellm.integrations.custom_logger import CustomLogger
+from litellm.integrations.custom_guardrail import ModifyResponseException
 from litellm.integrations.opentelemetry import UserAPIKeyAuth
 from litellm.proxy.common_request_processing import (
     ProxyBaseLLMRequestProcessing,
@@ -10428,18 +10429,23 @@ async def _slow_dict_detail_failure() -> Response:
 
 
 @pytest.mark.parametrize(
-    "produce,expected_error",
+    "produce,expected_code,expected_error",
     [
-        (_slow_rate_limit_failure, {"message": "rate limited", "code": "429"}),
-        (_slow_guardrail_block_body, {"message": "blocked", "type": "guardrail_error", "code": "400"}),
+        (_slow_rate_limit_failure, "rate_limit_exceeded", {"message": "rate limited", "code": "429"}),
+        (
+            _slow_guardrail_block_body,
+            "invalid_request_error",
+            {"message": "blocked", "type": "guardrail_error", "code": "400"},
+        ),
         (
             _slow_dict_detail_failure,
+            "invalid_request_error",
             {"message": "blocked by policy", "code": "400", "provider_specific_fields": dict(_POLICY_BLOCK_DETAIL)},
         ),
     ],
 )
 def test_ttft_keepalive_ends_a_late_responses_failure_with_an_error_event_and_no_done(
-    produce: Callable[[], Awaitable[Response]], expected_error: Mapping[str, object]
+    produce: Callable[[], Awaitable[Response]], expected_code: str, expected_error: Mapping[str, object]
 ) -> None:
     async def collect_and_check() -> None:
         response: Final = await open_sse_before_first_byte(
@@ -10453,6 +10459,7 @@ def test_ttft_keepalive_ends_a_late_responses_failure_with_an_error_event_and_no
         event: Final = json.loads(data_line.removeprefix("data: "))
         assert event_line == "event: error"
         assert event["type"] == "error"
+        assert (event["code"], event["message"]) == (expected_code, expected_error["message"])
         assert {key: event["error"].get(key) for key in expected_error} == expected_error
 
     loop: Final = AutojumpClockLoop()
@@ -10463,6 +10470,34 @@ def test_ttft_keepalive_ends_a_late_responses_failure_with_an_error_event_and_no
         loop.close()
 
 
+async def _slow_passthrough_block() -> Response:
+    await asyncio.sleep(0.2)
+    raise ModifyResponseException(message="blocked by policy", model="gpt-4o", request_data={})
+
+
+def test_ttft_keepalive_ends_a_late_responses_passthrough_block_with_the_block_response() -> None:
+    async def collect_and_check() -> None:
+        response: Final = await open_sse_before_first_byte(
+            _slow_passthrough_block(), ping_interval_seconds=0.05, late_framing=RESPONSES_LATE_SSE_FRAMING
+        )
+        collected: Final = await _drain(response)
+        frames: Final = tuple(chunk.decode() for chunk in collected if chunk != TTFT_PING)
+        names: Final = tuple(frame.split("\n", 1)[0] for frame in frames)
+        completed: Final = json.loads(frames[-1].split("\n", 1)[1].removeprefix("data: "))
+
+        assert collected[0] == TTFT_PING
+        assert (names[0], names[-1]) == ("event: response.created", "event: response.completed")
+        assert "event: error" not in names
+        assert completed["response"]["output"][0]["content"][0]["text"] == "blocked by policy"
+
+    loop: Final = AutojumpClockLoop()
+    try:
+        loop.run_until_complete(collect_and_check())
+    finally:
+        loop.run_until_complete(loop.shutdown_asyncgens())
+        loop.close()
+
+
 @pytest.mark.asyncio
 async def test_ttft_keepalive_closes_the_upstream_stream_it_relayed():
     """Starlette never calls the produced response, so its own cleanup never runs