diff --git a/litellm/proxy/common_request_processing.py b/litellm/proxy/common_request_processing.py index 6cdab2daa82..10a5c6d61ee 100644 --- a/litellm/proxy/common_request_processing.py +++ b/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),) diff --git a/litellm/proxy/common_utils/responses_stream_errors.py b/litellm/proxy/common_utils/responses_stream_errors.py index 132155d9492..05ec7603a3d 100644 --- a/litellm/proxy/common_utils/responses_stream_errors.py +++ b/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, }, diff --git a/litellm/proxy/response_polling/background_streaming.py b/litellm/proxy/response_polling/background_streaming.py index 975d700688e..7411e204d90 100644 --- a/litellm/proxy/response_polling/background_streaming.py +++ b/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", diff --git a/tests/unit/proxy/proxy_server/test_streaming_helpers.py b/tests/unit/proxy/proxy_server/test_streaming_helpers.py index b01487af5d1..fb8ff519fbc 100644 --- a/tests/unit/proxy/proxy_server/test_streaming_helpers.py +++ b/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, } 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 8affc8a6e19..f398d5acb1c 100644 --- a/tests/unit/proxy/response_polling/test_response_polling_handler.py +++ b/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): diff --git a/tests/unit/proxy/test_common_request_processing.py b/tests/unit/proxy/test_common_request_processing.py index 944432f91b1..63324c11930 100644 --- a/tests/unit/proxy/test_common_request_processing.py +++ b/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