From 9d4bab3b700c1170b7451bedfbda67cf830f58f3 Mon Sep 17 00:00:00 2001 From: zoroyihan7 Date: Tue, 8 Sep 2026 10:46:16 +0000 Subject: [PATCH 1/7] fix(responses): emit typed streaming failure events --- litellm/exceptions.py | 3 +- .../common_utils/responses_stream_errors.py | 119 ++++++++++++++++++ litellm/proxy/proxy_server.py | 21 +++- .../proxy/response_api_endpoints/endpoints.py | 6 +- litellm/responses/streaming_iterator.py | 5 + .../proxy_server/test_streaming_helpers.py | 85 ++++++++++++- .../response_api_endpoints/test_endpoints.py | 95 +++++++++++++- 7 files changed, 328 insertions(+), 6 deletions(-) create mode 100644 litellm/proxy/common_utils/responses_stream_errors.py diff --git a/litellm/exceptions.py b/litellm/exceptions.py index f9215267bf3..318b227ffa8 100644 --- a/litellm/exceptions.py +++ b/litellm/exceptions.py @@ -789,6 +789,7 @@ class APIError(openai.APIError): litellm_debug_info: str | None = None, max_retries: int | None = None, num_retries: int | None = None, + body: object | None = None, ): self.status_code = status_code self.message = f"litellm.APIError: {message}" @@ -799,7 +800,7 @@ class APIError(openai.APIError): self.num_retries = num_retries if request is None: request = httpx.Request(method="POST", url="https://api.openai.com/v1") - super().__init__(self.message, request=request, body=None) + super().__init__(self.message, request=request, body=body) def __str__(self): _message = self.message diff --git a/litellm/proxy/common_utils/responses_stream_errors.py b/litellm/proxy/common_utils/responses_stream_errors.py new file mode 100644 index 00000000000..a3bee06e912 --- /dev/null +++ b/litellm/proxy/common_utils/responses_stream_errors.py @@ -0,0 +1,119 @@ +import time +from collections.abc import Mapping +from types import MappingProxyType +from typing import Final + +from pydantic import BaseModel, ConfigDict + +from litellm._logging import redact_internal_details_from_client_message +from litellm._uuid import uuid +from litellm.exceptions import MidStreamFallbackError +from litellm.types.llms.openai import ResponseFailedEvent, ResponsesAPIResponse, ResponsesAPIStreamEvents + + +class _ResponseIdentity(BaseModel): + model_config = ConfigDict(frozen=True, from_attributes=True) + + id: str | None = None + model: str | None = None + created_at: int | None = None + + +class _StreamEvent(BaseModel): + model_config = ConfigDict(frozen=True, from_attributes=True) + + type: str | None = None + sequence_number: int | None = None + response: _ResponseIdentity | None = None + + +class _FailureDetails(BaseModel): + model_config = ConfigDict(frozen=True, from_attributes=True) + + message: str | None = None + code: str | int | None = None + type: str | None = None + status_code: int | None = None + + +def _original_failure(exception: Exception) -> Exception: + if isinstance(exception, MidStreamFallbackError) and exception.original_exception is not None: + return _original_failure(exception.original_exception) + return exception + + +def _response_error_code(details: _FailureDetails) -> str: + for value in (details.code, details.type): + if value == "insufficient_quota": + return "insufficient_quota" + if value in (429, "429") or isinstance(value, str) and value.startswith("rate_limit"): + return "rate_limit_exceeded" + if isinstance(details.code, str) and details.code and not details.code.isdecimal(): + return details.code + if details.status_code == 429: + return "rate_limit_exceeded" + return "server_error" + + +class ResponsesStreamErrorState: + def __init__(self) -> None: + self.response_id: str | None = None + self.model: str | None = None + self.created_at: int | None = None + self.sequence_number = -1 + self.terminal_emitted = False + + @staticmethod + def observe_chunk(chunk: object) -> _StreamEvent | None: + if not isinstance(chunk, (BaseModel, Mapping)): + return None + return _StreamEvent.model_validate(chunk) + + def mark_emitted(self, event: _StreamEvent | None) -> None: + if event is None: + return + if event.sequence_number is not None: + self.sequence_number = max(self.sequence_number, event.sequence_number) + if event.response is not None: + self.response_id = event.response.id or self.response_id + 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"): + self.terminal_emitted = True + + def format_failure(self, exception: Exception) -> str | None: + if self.terminal_emitted: + return None + original: Final = _original_failure(exception) + details: Final = _FailureDetails.model_validate(original) + response: Final = ResponsesAPIResponse.model_validate( + MappingProxyType( + { + "id": self.response_id or f"resp_{uuid.uuid4().hex}", + "object": "response", + "created_at": self.created_at if self.created_at is not None else int(time.time()), + "model": self.model, + "status": "failed", + "output": (), + "error": MappingProxyType( + { + "code": _response_error_code(details), + "message": redact_internal_details_from_client_message(details.message or str(original)), + } + ), + } + ) + ) + event: Final = ResponseFailedEvent.model_validate( + MappingProxyType( + { + "type": ResponsesAPIStreamEvents.RESPONSE_FAILED, + "response": response, + "sequence_number": self.sequence_number + 1, + } + ) + ) + payload: Final = event.model_dump_json(exclude_none=True) + self.terminal_emitted = True + return f"event: response.failed\ndata: {payload}\n\n" diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 32b6b841af7..9f1d69acefa 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -385,6 +385,7 @@ from litellm.proxy.common_utils.periodic_reload_schedule import ( ) from litellm.proxy.common_utils.proxy_state import ProxyState from litellm.proxy.common_utils.reset_budget_job import ResetBudgetJob +from litellm.proxy.common_utils.responses_stream_errors import ResponsesStreamErrorState from litellm.proxy.common_utils.scheduled_job_stagger import ( apply_scheduled_job_stagger, attach_job_timing_logger, @@ -8723,10 +8724,13 @@ async def async_data_generator( user_api_key_dict: UserAPIKeyAuth, request_data: dict, request: Request | None = None, + *, + responses_stream_errors: bool = False, ): verbose_proxy_logger.debug("inside generator") stream_completed = False client_disconnected = False + error_state: Final = ResponsesStreamErrorState() if responses_stream_errors else None try: error_message: str | None = None requested_model_from_client: Final = _get_client_requested_model_for_streaming(request_data=request_data) @@ -8837,6 +8841,7 @@ async def async_data_generator( fallback_metadata_event_sent = True continue + responses_event: Final = error_state.observe_chunk(chunk) if error_state is not None else None raw_passthrough = False if isinstance(chunk, BaseModel): chunk = _serialize_streaming_chunk(chunk) @@ -8871,8 +8876,13 @@ async def async_data_generator( if not raw_passthrough: try: - yield _format_streaming_sse_chunk(chunk=chunk) + formatted_chunk: Final = _format_streaming_sse_chunk(chunk=chunk) + if error_state is not None: + error_state.mark_emitted(responses_event) + yield formatted_chunk except Exception as e: + if error_state is not None: + raise yield f"data: {e}\n\n" if pending_fallback_event: @@ -8922,6 +8932,12 @@ async def async_data_generator( e, ) + if error_state is not None: + stream_completed = True + error_frame: Final = error_state.format_failure(e) + if error_frame is not None: + yield error_frame + return if isinstance(e, HTTPException): raise e elif isinstance(e, StreamingCallbackError): @@ -8958,12 +8974,15 @@ def select_data_generator( user_api_key_dict: UserAPIKeyAuth, request_data: dict, request: Request | None = None, + *, + responses_stream_errors: bool = False, ): return async_data_generator( response=response, user_api_key_dict=user_api_key_dict, request_data=request_data, request=request, + responses_stream_errors=responses_stream_errors, ) diff --git a/litellm/proxy/response_api_endpoints/endpoints.py b/litellm/proxy/response_api_endpoints/endpoints.py index 5907ffc64eb..3202fabe74e 100644 --- a/litellm/proxy/response_api_endpoints/endpoints.py +++ b/litellm/proxy/response_api_endpoints/endpoints.py @@ -3,6 +3,7 @@ import json import time from collections.abc import AsyncIterator, Awaitable, Mapping from enum import Enum +from functools import partial from types import MappingProxyType from typing import TYPE_CHECKING, Any, Final, NamedTuple, Protocol, cast, get_args from uuid import uuid4 @@ -243,6 +244,7 @@ async def responses_api( version, ) + native_data_generator: Final = partial(select_data_generator, responses_stream_errors=True) data = await _read_request_body(request=request) # Check if polling via cache should be used for this request @@ -329,7 +331,7 @@ async def responses_api( llm_router=llm_router, proxy_config=proxy_config, proxy_logging_obj=proxy_logging_obj, - select_data_generator=select_data_generator, + select_data_generator=native_data_generator, user_model=user_model, user_temperature=user_temperature, user_request_timeout=user_request_timeout, @@ -355,7 +357,7 @@ async def responses_api( llm_router=llm_router, general_settings=general_settings, proxy_config=proxy_config, - select_data_generator=select_data_generator, + select_data_generator=native_data_generator, model=None, user_model=user_model, user_temperature=user_temperature, diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index 9f9016c5a7f..1134f6b07e3 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -579,6 +579,11 @@ class BaseResponsesAPIStreamingIterator: message=error_message, llm_provider=self.custom_llm_provider or "", model=self.model or "", + body={ # mutable-ok: OpenAI APIError reads code/type only from a dict body + "code": error_code, + "type": error_type, + "message": error_message, + }, ) if 400 <= status_code < 500 and status_code != 429: raise mapped_exception diff --git a/tests/test_litellm/proxy/proxy_server/test_streaming_helpers.py b/tests/test_litellm/proxy/proxy_server/test_streaming_helpers.py index 87e10ce7e8d..6f3a47a4b79 100644 --- a/tests/test_litellm/proxy/proxy_server/test_streaming_helpers.py +++ b/tests/test_litellm/proxy/proxy_server/test_streaming_helpers.py @@ -17,15 +17,18 @@ from __future__ import annotations import asyncio import json +from collections.abc import AsyncIterator +from typing import Final, Literal from unittest.mock import AsyncMock, MagicMock import pytest from fastapi import Response from fastapi.responses import StreamingResponse +from pydantic import BaseModel import litellm -from litellm.constants import RETURN_RAW_MODEL_NAME_METADATA_KEY import litellm.proxy.proxy_server as ps +from litellm.constants import RETURN_RAW_MODEL_NAME_METADATA_KEY from litellm.proxy._types import UserAPIKeyAuth from litellm.proxy.proxy_server import ( _apply_streaming_chunk_hooks, @@ -42,6 +45,12 @@ from litellm.proxy.proxy_server import ( data_generator, select_data_generator, ) +from litellm.types.llms.openai import ( + ResponseCompletedEvent, + ResponseCreatedEvent, + ResponseFailedEvent, + ResponsesAPIResponse, +) from litellm.types.utils import Delta, ModelResponseStream, StreamingChoices, Usage from .conftest import normalize @@ -872,6 +881,80 @@ async def test_async_data_generator_mid_stream_exception_yields_error_payload( assert any(isinstance(item, str) and item.startswith('data: {"error":') for item in out) +@pytest.mark.asyncio +@pytest.mark.parametrize("terminal", ["completed", "serialization_failure", "failure_after_completed"]) +async def test_responses_stream_keeps_tool_deltas_and_only_emits_a_valid_terminal( + terminal: Literal["completed", "serialization_failure", "failure_after_completed"], +) -> None: + class ToolDelta(BaseModel): + type: Literal["response.function_call_arguments.delta"] + sequence_number: int + item_id: str + output_index: int + delta: str + + class UnserializableTerminal(BaseModel): + type: Literal["response.completed"] + sequence_number: int + response: ResponsesAPIResponse + invalid: object + + response: Final = ResponsesAPIResponse(id="resp_visible", created_at=1, model="gpt-6-astra", output=[]) + created: Final = ResponseCreatedEvent.model_validate( + {"type": "response.created", "sequence_number": 0, "response": response} + ) + completed: Final = ResponseCompletedEvent.model_validate( + {"type": "response.completed", "sequence_number": 2, "response": response} + ) + tool_delta: Final = ToolDelta( + type="response.function_call_arguments.delta", sequence_number=1, item_id="fc_stream_error", + output_index=0, delta='{"path":"partial', + ) + + async def upstream() -> AsyncIterator[BaseModel]: + yield created + yield tool_delta + yield ( + UnserializableTerminal(type="response.completed", sequence_number=2, response=response, invalid=object()) + if terminal == "serialization_failure" else completed + ) + if terminal == "failure_after_completed": + raise litellm.APIError(status_code=500, message="Stream close failed", llm_provider="openai", model="gpt-6-astra") + + frames: Final = [ + frame + async for frame in select_data_generator( + response=upstream(), + user_api_key_dict=_user_auth(), + request_data={}, + responses_stream_errors=True, + ) + ] + decoded: Final = tuple(frame.decode() if isinstance(frame, bytes) else frame for frame in frames) + event_frames: Final = tuple(frame for frame in decoded if frame != "data: [DONE]\n\n") + payloads: Final = tuple( + json.loads(next(line[6:] for line in frame.splitlines() if line.startswith("data: "))) + for frame in event_frames + ) + + assert payloads[0]["response"]["id"] == "resp_visible" + assert payloads[1] == tool_delta.model_dump() + assert len(payloads) == 3 + if terminal == "serialization_failure": + failure: Final = ResponseFailedEvent.model_validate(payloads[-1]) + assert event_frames[-1].startswith("event: response.failed\n") + 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 "serialize" in failure.response.error["message"].lower() + assert payloads[-1]["sequence_number"] > payloads[1]["sequence_number"] + else: + assert payloads[-1]["type"] == "response.completed" + assert payloads[-1]["sequence_number"] == 2 + assert "error" not in payloads[-1] + + # --------------------------------------------------------------------------- # select_data_generator # --------------------------------------------------------------------------- diff --git a/tests/test_litellm/proxy/response_api_endpoints/test_endpoints.py b/tests/test_litellm/proxy/response_api_endpoints/test_endpoints.py index d7010de6405..9e2f70a3d64 100644 --- a/tests/test_litellm/proxy/response_api_endpoints/test_endpoints.py +++ b/tests/test_litellm/proxy/response_api_endpoints/test_endpoints.py @@ -3,10 +3,12 @@ Test for response_api_endpoints/endpoints.py """ import unittest -from typing import Any +from typing import Any, Final, Literal from unittest.mock import AsyncMock, MagicMock, patch +import httpx import pytest +import respx from fastapi.testclient import TestClient from httpx import Response @@ -14,6 +16,97 @@ import litellm from litellm.proxy.proxy_server import app +@pytest.mark.asyncio +@pytest.mark.parametrize( + "path,error_kind", + [ + ("/v1/responses", "rate_limit"), + ("/v1/responses", "numeric_rate_limit"), + ("/v1/responses", "server_error"), + ("/v1/responses", "response_failed"), + ("/cursor/chat/completions", "server_error"), + ("/v1/chat/completions", "server_error"), + ], +) +async def test_streaming_upstream_errors_keep_the_client_protocol( + monkeypatch: pytest.MonkeyPatch, + path: str, + error_kind: Literal["rate_limit", "numeric_rate_limit", "server_error", "response_failed"], +) -> None: + import litellm.proxy.proxy_server as ps + from litellm.proxy.auth.user_api_key_auth import user_api_key_auth + + model: Final = "gpt-6-astra" + message: Final = "Upstream cannot complete this response" + code: Final = { + "rate_limit": "rate_limit_exceeded", "numeric_rate_limit": "429", + "server_error": "server_error", "response_failed": "server_error", + }[error_kind] + error: Final = {"message": message, "code": code, "type": None, "param": "input"} + response: Final = {"id": "resp_upstream", "object": "response", "created_at": 1, + "status": "in_progress", "model": model, "output": [], + "parallel_tool_calls": True, "tool_choice": "auto", "tools": []} + created: Final = {"type": "response.created", "sequence_number": 0, "response": response} + tool_added: Final = {"type": "response.output_item.added", "sequence_number": 1, "output_index": 0, + "item": {"type": "function_call", "id": "fc_partial", "call_id": "call_partial", + "name": "read_file", "arguments": "", "status": "in_progress"}} + tool_delta: Final = {"type": "response.function_call_arguments.delta", "sequence_number": 2, + "item_id": "fc_partial", "output_index": 0, "delta": '{"path":"partial'} + failed: Final = ( + {"type": "response.failed", "sequence_number": 9, + "response": {**response, "status": "failed", "error": error}} + if error_kind == "response_failed" else {"type": "error", "error": error} + ) + chat: Final = {"id": "chatcmpl_partial", "object": "chat.completion.chunk", "created": 1, + "model": model, "choices": [{"index": 0, "delta": {"content": "partial"}, + "finish_reason": None}]} + is_chat: Final = path == "/v1/chat/completions" + upstream_events: Final = (chat, {"error": error}) if is_chat else (created, tool_added, tool_delta, failed) + wire: Final = "".join("data: " + json.dumps(event) + "\n\n" for event in upstream_events) + upstream_url: Final = "https://streaming.example/v1" + router: Final = litellm.Router( + model_list=[{"model_name": model, "litellm_params": { + "model": "openai/" + model, "api_base": upstream_url, "api_key": "fixture-key"}}], + num_retries=0, + ) + monkeypatch.setattr(ps, "llm_router", router) + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + monkeypatch.setitem(app.dependency_overrides, user_api_key_auth, _auth_override) + with respx.mock as transport: + transport.post(upstream_url + ("/chat/completions" if is_chat else "/responses")).respond( + 200, content=wire, headers={"Content-Type": "text/event-stream"} + ) + async with httpx.AsyncClient(transport=httpx.ASGITransport(app), base_url="http://testserver") as client: + result: Final = await client.post( + path, json={"model": model, "stream": True, + **({"messages": [{"role": "user", "content": "hello"}]} if is_chat else {"input": "hello"})}, + ) + frames: Final = tuple(frame for frame in result.text.split("\n\n") if "data: " in frame) + events: Final = tuple( + json.loads(next(line[6:] for line in frame.splitlines() if line.startswith("data: "))) + for frame in frames if "data: [DONE]" not in frame + ) + + assert result.status_code == 200, result.text + assert message in result.text + if path == "/v1/responses": + assert frames[-1].startswith("event: response.failed\n"), result.text + assert [event["type"] for event in events] == [ + "response.created", "response.output_item.added", "response.function_call_arguments.delta", "response.failed" + ] + assert events[2]["delta"] == tool_delta["delta"] + assert events[-1]["sequence_number"] == events[-2]["sequence_number"] + 1 + assert events[-1]["response"]["id"] == events[0]["response"]["id"] + assert events[-1]["response"]["status"] == "failed" + assert events[-1]["response"]["error"]["code"] == ( + "rate_limit_exceeded" if error_kind in ("rate_limit", "numeric_rate_limit") else "server_error" + ) + else: + assert events[0]["object"] == "chat.completion.chunk", result.text + assert "response.failed" not in result.text + assert "error" in events[-1] + + class TestResponsesAPIEndpoints(unittest.TestCase): @pytest.mark.asyncio @patch("litellm.proxy.proxy_server.llm_router") From 089703d10b2eebbec1a5b780494686451ab72161 Mon Sep 17 00:00:00 2001 From: zoroyihan7 Date: Tue, 8 Sep 2026 11:02:38 +0000 Subject: [PATCH 2/7] fix(responses): preserve failure metadata at streaming boundaries --- .../common_utils/responses_stream_errors.py | 45 +++++++++---- litellm/proxy/proxy_server.py | 9 +-- .../proxy_server/test_streaming_helpers.py | 63 ++++++++++++++++--- .../response_api_endpoints/test_endpoints.py | 28 ++++++--- 4 files changed, 114 insertions(+), 31 deletions(-) diff --git a/litellm/proxy/common_utils/responses_stream_errors.py b/litellm/proxy/common_utils/responses_stream_errors.py index a3bee06e912..356e948a2df 100644 --- a/litellm/proxy/common_utils/responses_stream_errors.py +++ b/litellm/proxy/common_utils/responses_stream_errors.py @@ -1,9 +1,10 @@ import time from collections.abc import Mapping +from http import HTTPStatus from types import MappingProxyType from typing import Final -from pydantic import BaseModel, ConfigDict +from pydantic import BaseModel, ConfigDict, field_validator from litellm._logging import redact_internal_details_from_client_message from litellm._uuid import uuid @@ -35,6 +36,16 @@ class _FailureDetails(BaseModel): type: str | None = None status_code: int | None = None + @field_validator("code", mode="before") + @classmethod + def normalize_code(cls, value: object) -> str | int | None: + return value if isinstance(value, (str, int)) and not isinstance(value, bool) else None + + @field_validator("type", mode="before") + @classmethod + def normalize_type(cls, value: object) -> str | None: + return value if isinstance(value, str) else None + def _original_failure(exception: Exception) -> Exception: if isinstance(exception, MidStreamFallbackError) and exception.original_exception is not None: @@ -50,9 +61,21 @@ def _response_error_code(details: _FailureDetails) -> str: return "rate_limit_exceeded" if isinstance(details.code, str) and details.code and not details.code.isdecimal(): return details.code - if details.status_code == 429: - return "rate_limit_exceeded" - return "server_error" + match details.status_code: + case HTTPStatus.UNAUTHORIZED: + return "authentication_error" + case HTTPStatus.FORBIDDEN: + return "permission_error" + case HTTPStatus.NOT_FOUND: + return "not_found_error" + case HTTPStatus.REQUEST_TIMEOUT: + return "request_timeout" + case HTTPStatus.TOO_MANY_REQUESTS: + return "rate_limit_exceeded" + case int(status) if HTTPStatus.BAD_REQUEST <= status < HTTPStatus.INTERNAL_SERVER_ERROR: + return "invalid_request_error" + case _: + return "server_error" class ResponsesStreamErrorState: @@ -62,16 +85,15 @@ class ResponsesStreamErrorState: self.created_at: int | None = None self.sequence_number = -1 self.terminal_emitted = False + self._pending_event: _StreamEvent | None = None - @staticmethod - def observe_chunk(chunk: object) -> _StreamEvent | None: - if not isinstance(chunk, (BaseModel, Mapping)): - return None - return _StreamEvent.model_validate(chunk) + def observe_chunk(self, chunk: object) -> None: + self._pending_event = _StreamEvent.model_validate(chunk) if isinstance(chunk, (BaseModel, Mapping)) else None - def mark_emitted(self, event: _StreamEvent | None) -> None: + def mark_emitted(self, frame: str | bytes) -> str | bytes: + event: Final = self._pending_event if event is None: - return + return frame if event.sequence_number is not None: self.sequence_number = max(self.sequence_number, event.sequence_number) if event.response is not None: @@ -81,6 +103,7 @@ class ResponsesStreamErrorState: self.created_at = event.response.created_at if event.type in ("response.completed", "response.failed", "response.incomplete"): self.terminal_emitted = True + return frame def format_failure(self, exception: Exception) -> str | None: if self.terminal_emitted: diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 9f1d69acefa..870cd78aa85 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -8841,7 +8841,8 @@ async def async_data_generator( fallback_metadata_event_sent = True continue - responses_event: Final = error_state.observe_chunk(chunk) if error_state is not None else None + if error_state is not None: + error_state.observe_chunk(cast(object, chunk)) # cast-ok: the helper validates legacy untyped chunks raw_passthrough = False if isinstance(chunk, BaseModel): chunk = _serialize_streaming_chunk(chunk) @@ -8876,10 +8877,10 @@ async def async_data_generator( if not raw_passthrough: try: - formatted_chunk: Final = _format_streaming_sse_chunk(chunk=chunk) if error_state is not None: - error_state.mark_emitted(responses_event) - yield formatted_chunk + yield error_state.mark_emitted(_format_streaming_sse_chunk(chunk=chunk)) + else: + yield _format_streaming_sse_chunk(chunk=chunk) except Exception as e: if error_state is not None: raise diff --git a/tests/test_litellm/proxy/proxy_server/test_streaming_helpers.py b/tests/test_litellm/proxy/proxy_server/test_streaming_helpers.py index 6f3a47a4b79..53e055882e8 100644 --- a/tests/test_litellm/proxy/proxy_server/test_streaming_helpers.py +++ b/tests/test_litellm/proxy/proxy_server/test_streaming_helpers.py @@ -21,9 +21,11 @@ from collections.abc import AsyncIterator from typing import Final, Literal from unittest.mock import AsyncMock, MagicMock +import httpx import pytest -from fastapi import Response +from fastapi import HTTPException, Response from fastapi.responses import StreamingResponse +from openai import APIError as OpenAIAPIError from pydantic import BaseModel import litellm @@ -882,9 +884,44 @@ async def test_async_data_generator_mid_stream_exception_yields_error_payload( @pytest.mark.asyncio -@pytest.mark.parametrize("terminal", ["completed", "serialization_failure", "failure_after_completed"]) +@pytest.mark.parametrize( + "terminal,upstream_error,expected_code", + [ + ("completed", None, None), + ("serialization_failure", None, "server_error"), + ("failure_after_completed", None, None), + pytest.param( + "upstream_failure", + litellm.AuthenticationError( + message="Upstream rejected request", llm_provider="openai", model="gpt-6-astra" + ), + "authentication_error", id="authentication_error", + ), + pytest.param( + "upstream_failure", + OpenAIAPIError( + message="Upstream rejected request", + request=httpx.Request("POST", "https://streaming.example/v1/responses"), + body={"code": {"reason": "overloaded"}, "type": {"unexpected": "object"}}, + ), + "server_error", id="structured_provider_error_fields", + ), + *( + pytest.param( + "upstream_failure", HTTPException(status_code=status, detail="Upstream rejected request"), + code, id=f"http_{status}", + ) + for status, code in ( + (400, "invalid_request_error"), (403, "permission_error"), (404, "not_found_error"), + (408, "request_timeout"), (422, "invalid_request_error"), (500, "server_error"), (503, "server_error"), + ) + ), + ], +) async def test_responses_stream_keeps_tool_deltas_and_only_emits_a_valid_terminal( - terminal: Literal["completed", "serialization_failure", "failure_after_completed"], + terminal: Literal["completed", "serialization_failure", "failure_after_completed", "upstream_failure"], + upstream_error: HTTPException | OpenAIAPIError | None, + expected_code: str | None, ) -> None: class ToolDelta(BaseModel): type: Literal["response.function_call_arguments.delta"] @@ -910,16 +947,23 @@ async def test_responses_stream_keeps_tool_deltas_and_only_emits_a_valid_termina type="response.function_call_arguments.delta", sequence_number=1, item_id="fc_stream_error", output_index=0, delta='{"path":"partial', ) + original_status: Final = ( + upstream_error.status_code if isinstance(upstream_error, (HTTPException, litellm.AuthenticationError)) else None + ) async def upstream() -> AsyncIterator[BaseModel]: yield created yield tool_delta + if upstream_error is not None: + raise upstream_error yield ( UnserializableTerminal(type="response.completed", sequence_number=2, response=response, invalid=object()) if terminal == "serialization_failure" else completed ) if terminal == "failure_after_completed": - raise litellm.APIError(status_code=500, message="Stream close failed", llm_provider="openai", model="gpt-6-astra") + raise litellm.APIError( + status_code=500, message="Stream close failed", llm_provider="openai", model="gpt-6-astra" + ) frames: Final = [ frame @@ -940,14 +984,19 @@ async def test_responses_stream_keeps_tool_deltas_and_only_emits_a_valid_termina assert payloads[0]["response"]["id"] == "resp_visible" assert payloads[1] == tool_delta.model_dump() assert len(payloads) == 3 - if terminal == "serialization_failure": + if terminal in ("serialization_failure", "upstream_failure"): failure: Final = ResponseFailedEvent.model_validate(payloads[-1]) assert event_frames[-1].startswith("event: response.failed\n") 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 "serialize" in failure.response.error["message"].lower() + assert failure.response.error["code"] == expected_code + if upstream_error is None: + assert "serialize" in failure.response.error["message"].lower() + else: + assert "Upstream rejected request" in failure.response.error["message"] + if isinstance(upstream_error, (HTTPException, litellm.AuthenticationError)): + assert upstream_error.status_code == original_status assert payloads[-1]["sequence_number"] > payloads[1]["sequence_number"] else: assert payloads[-1]["type"] == "response.completed" diff --git a/tests/test_litellm/proxy/response_api_endpoints/test_endpoints.py b/tests/test_litellm/proxy/response_api_endpoints/test_endpoints.py index 9e2f70a3d64..7b79e1b8613 100644 --- a/tests/test_litellm/proxy/response_api_endpoints/test_endpoints.py +++ b/tests/test_litellm/proxy/response_api_endpoints/test_endpoints.py @@ -61,7 +61,9 @@ async def test_streaming_upstream_errors_keep_the_client_protocol( "model": model, "choices": [{"index": 0, "delta": {"content": "partial"}, "finish_reason": None}]} is_chat: Final = path == "/v1/chat/completions" - upstream_events: Final = (chat, {"error": error}) if is_chat else (created, tool_added, tool_delta, failed) + partial: Final = path != "/v1/responses" or error_kind in ("numeric_rate_limit", "response_failed") + response_events: Final = (created, tool_added, tool_delta, failed) if partial else (failed,) + upstream_events: Final = (chat, {"error": error}) if is_chat else response_events wire: Final = "".join("data: " + json.dumps(event) + "\n\n" for event in upstream_events) upstream_url: Final = "https://streaming.example/v1" router: Final = litellm.Router( @@ -78,8 +80,10 @@ async def test_streaming_upstream_errors_keep_the_client_protocol( ) async with httpx.AsyncClient(transport=httpx.ASGITransport(app), base_url="http://testserver") as client: result: Final = await client.post( - path, json={"model": model, "stream": True, - **({"messages": [{"role": "user", "content": "hello"}]} if is_chat else {"input": "hello"})}, + path, json={ + "model": model, "stream": True, + **({"messages": [{"role": "user", "content": "hello"}]} if is_chat else {"input": "hello"}), + }, ) frames: Final = tuple(frame for frame in result.text.split("\n\n") if "data: " in frame) events: Final = tuple( @@ -91,12 +95,18 @@ async def test_streaming_upstream_errors_keep_the_client_protocol( assert message in result.text if path == "/v1/responses": assert frames[-1].startswith("event: response.failed\n"), result.text - assert [event["type"] for event in events] == [ - "response.created", "response.output_item.added", "response.function_call_arguments.delta", "response.failed" - ] - assert events[2]["delta"] == tool_delta["delta"] - assert events[-1]["sequence_number"] == events[-2]["sequence_number"] + 1 - assert events[-1]["response"]["id"] == events[0]["response"]["id"] + if partial: + assert [event["type"] for event in events] == [ + "response.created", "response.output_item.added", + "response.function_call_arguments.delta", "response.failed", + ] + assert events[2]["delta"] == tool_delta["delta"] + assert events[-1]["sequence_number"] == events[-2]["sequence_number"] + 1 + assert events[-1]["response"]["id"] == events[0]["response"]["id"] + else: + assert [event["type"] for event in events] == ["response.failed"] + assert events[0]["sequence_number"] == 0 + assert events[0]["response"]["id"].startswith("resp_") assert events[-1]["response"]["status"] == "failed" assert events[-1]["response"]["error"]["code"] == ( "rate_limit_exceeded" if error_kind in ("rate_limit", "numeric_rate_limit") else "server_error" From dc7895c1eacf55733953c3f802f46ea9103a76b4 Mon Sep 17 00:00:00 2001 From: zoroyihan7 Date: Tue, 8 Sep 2026 11:44:41 +0000 Subject: [PATCH 3/7] fix(responses): satisfy streaming regression checks --- litellm/proxy/common_utils/responses_stream_errors.py | 7 ++++--- .../test_response_polling_pre_call_checks.py | 1 - 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/litellm/proxy/common_utils/responses_stream_errors.py b/litellm/proxy/common_utils/responses_stream_errors.py index 356e948a2df..706b4c298d7 100644 --- a/litellm/proxy/common_utils/responses_stream_errors.py +++ b/litellm/proxy/common_utils/responses_stream_errors.py @@ -48,9 +48,10 @@ class _FailureDetails(BaseModel): def _original_failure(exception: Exception) -> Exception: - if isinstance(exception, MidStreamFallbackError) and exception.original_exception is not None: - return _original_failure(exception.original_exception) - return exception + current = exception # rebind-ok: the recursion gate requires iterative wrapper traversal + while isinstance(current, MidStreamFallbackError) and current.original_exception is not None: + current = current.original_exception + return current def _response_error_code(details: _FailureDetails) -> str: diff --git a/tests/proxy_unit_tests/test_response_polling_pre_call_checks.py b/tests/proxy_unit_tests/test_response_polling_pre_call_checks.py index 459834d0fd2..3fabdcefe5a 100644 --- a/tests/proxy_unit_tests/test_response_polling_pre_call_checks.py +++ b/tests/proxy_unit_tests/test_response_polling_pre_call_checks.py @@ -130,7 +130,6 @@ class TestPollingEndpointPreCallGuard: "litellm.proxy.proxy_server.proxy_config": MagicMock(), "litellm.proxy.proxy_server.proxy_logging_obj": AsyncMock(), "litellm.proxy.proxy_server.redis_usage_cache": AsyncMock(), - "litellm.proxy.proxy_server.select_data_generator": None, "litellm.proxy.proxy_server.user_api_base": None, "litellm.proxy.proxy_server.user_max_tokens": None, "litellm.proxy.proxy_server.user_model": None, From df1b3c849b76b4dff04441a5838427b5b6be9827 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Fri, 18 Sep 2026 13:58:53 -0700 Subject: [PATCH 4/7] fix(responses): keep upstream error details in response.failed RateLimitError and InternalServerError now carry the provider body, so the OpenAI exception mapper keeps upstream codes like cyber_policy and the upstream message instead of a generic mapped one The proxy's response.failed event prefers the upstream body's code, message, and type over the mapped exception's, and numeric error codes in an error event map to their own HTTP status --- litellm/exceptions.py | 6 ++-- .../exception_mapping_utils.py | 5 +++ litellm/proxy/_lazy_openapi_snapshot.json | 2 +- .../common_utils/responses_stream_errors.py | 21 +++++++++++- litellm/responses/streaming_iterator.py | 13 +++++--- .../test_exception_mapping_utils.py | 23 +++++++++++++ .../proxy_server/test_streaming_helpers.py | 16 +++++++++ .../response_api_endpoints/test_endpoints.py | 17 ++++++---- .../test_streaming_iterator_error_events.py | 33 +++++++++++++++++++ 9 files changed, 120 insertions(+), 16 deletions(-) diff --git a/litellm/exceptions.py b/litellm/exceptions.py index c638e8a0f86..de9f5c692a1 100644 --- a/litellm/exceptions.py +++ b/litellm/exceptions.py @@ -464,6 +464,7 @@ class RateLimitError(openai.RateLimitError): rate_limit_type: str | RateLimitType | None = None, headers: dict[str, str] | None = None, detail: Any = None, + body: object | None = None, ): self.status_code = 429 self.message = f"litellm.RateLimitError: {message}" @@ -507,7 +508,7 @@ class RateLimitError(openai.RateLimitError): ), ) super().__init__( - self.message, response=self.response, body=None + self.message, response=self.response, body=body ) # Call the base class constructor with the parameters it needs self.code = "429" self.type = "throttling_error" @@ -765,6 +766,7 @@ class InternalServerError(openai.InternalServerError): litellm_debug_info: str | None = None, max_retries: int | None = None, num_retries: int | None = None, + body: object | None = None, ): self.status_code = 500 self.message = f"litellm.InternalServerError: {message}" @@ -783,7 +785,7 @@ class InternalServerError(openai.InternalServerError): ), ) super().__init__( - self.message, response=self.response, body=None + self.message, response=self.response, body=body ) # Call the base class constructor with the parameters it needs def __str__(self): diff --git a/litellm/litellm_core_utils/exception_mapping_utils.py b/litellm/litellm_core_utils/exception_mapping_utils.py index 70675966dfc..61e2698dd6f 100644 --- a/litellm/litellm_core_utils/exception_mapping_utils.py +++ b/litellm/litellm_core_utils/exception_mapping_utils.py @@ -307,6 +307,7 @@ def _map_openai_exception( model=model, llm_provider=custom_llm_provider, response=response, + body=getattr(original_exception, "body", None), ) elif ExceptionCheckers.is_error_str_context_window_exceeded(error_str): raise ContextWindowExceededError( @@ -381,6 +382,7 @@ def _map_openai_exception( message=f"{exception_provider} - {message}", model=model, llm_provider=custom_llm_provider, + body=getattr(original_exception, "body", None), ) elif "Request too large" in error_str: raise RateLimitError( @@ -389,6 +391,7 @@ def _map_openai_exception( llm_provider=custom_llm_provider, response=response, litellm_debug_info=extra_information, + body=getattr(original_exception, "body", None), ) elif ( "The api_key client option must be set either by passing api_key to the client or by setting the OPENAI_API_KEY environment variable" @@ -460,6 +463,7 @@ def _map_openai_exception( llm_provider=custom_llm_provider, response=response, litellm_debug_info=extra_information, + body=getattr(original_exception, "body", None), ) elif original_exception.status_code == 500: raise InternalServerError( @@ -468,6 +472,7 @@ def _map_openai_exception( llm_provider=custom_llm_provider, response=response, litellm_debug_info=extra_information, + body=getattr(original_exception, "body", None), ) elif original_exception.status_code == 502: raise BadGatewayError( diff --git a/litellm/proxy/_lazy_openapi_snapshot.json b/litellm/proxy/_lazy_openapi_snapshot.json index 8a8d08c6887..2aa2cf15ac1 100644 --- a/litellm/proxy/_lazy_openapi_snapshot.json +++ b/litellm/proxy/_lazy_openapi_snapshot.json @@ -19394,7 +19394,7 @@ } } }, - "description": "\n Unified rate-limit error.\n\n Every rate-limit condition surfaced by litellm \u2014 whether it originated from\n an upstream LLM provider, a vendor batch endpoint, or one of litellm's own\n proxy-side limiters (parallel-requests, dynamic-rate, batch-rate, budget,\n max-iterations, etc.) \u2014 is raised as an instance of this class.\n\n The :attr:`category` attribute lets callers distinguish the source. See\n :class:`RateLimitErrorCategory` for the available values.\n " + "description": "\nUnified rate-limit error.\n\nEvery rate-limit condition surfaced by litellm \u2014 whether it originated from\nan upstream LLM provider, a vendor batch endpoint, or one of litellm's own\nproxy-side limiters (parallel-requests, dynamic-rate, batch-rate, budget,\nmax-iterations, etc.) \u2014 is raised as an instance of this class.\n\nThe :attr:`category` attribute lets callers distinguish the source. See\n:class:`RateLimitErrorCategory` for the available values.\n" }, "500": { "content": { diff --git a/litellm/proxy/common_utils/responses_stream_errors.py b/litellm/proxy/common_utils/responses_stream_errors.py index 706b4c298d7..63f881c126e 100644 --- a/litellm/proxy/common_utils/responses_stream_errors.py +++ b/litellm/proxy/common_utils/responses_stream_errors.py @@ -36,6 +36,11 @@ class _FailureDetails(BaseModel): type: str | None = None status_code: int | None = None + @field_validator("message", mode="before") + @classmethod + def normalize_message(cls, value: object) -> str | None: + return value if isinstance(value, str) else None + @field_validator("code", mode="before") @classmethod def normalize_code(cls, value: object) -> str | int | None: @@ -54,6 +59,20 @@ def _original_failure(exception: Exception) -> Exception: return current +def _failure_details(original: Exception) -> _FailureDetails: + mapped: Final = _FailureDetails.model_validate(original) + body: Final = getattr(original, "body", None) + if not isinstance(body, Mapping): + return mapped + upstream: Final = _FailureDetails.model_validate(body) + return _FailureDetails( + message=upstream.message or mapped.message, + code=upstream.code if upstream.code is not None else mapped.code, + type=upstream.type or mapped.type, + status_code=mapped.status_code, + ) + + def _response_error_code(details: _FailureDetails) -> str: for value in (details.code, details.type): if value == "insufficient_quota": @@ -110,7 +129,7 @@ class ResponsesStreamErrorState: if self.terminal_emitted: return None original: Final = _original_failure(exception) - details: Final = _FailureDetails.model_validate(original) + details: Final = _failure_details(original) response: Final = ResponsesAPIResponse.model_validate( MappingProxyType( { diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index 8d766cf1cd0..9e7cfc0fbe7 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -211,18 +211,21 @@ def _error_event_fields(error_obj: object) -> tuple[str, str | None, str | None] raw_code = None message: Final = str(raw_message) if raw_message is not None else "Response API in-stream error" error_type: Final = raw_type if isinstance(raw_type, str) else None - code: Final = raw_code if isinstance(raw_code, str) else None + code: Final = str(raw_code) if isinstance(raw_code, (str, int)) and not isinstance(raw_code, bool) else None return message, error_type, code +def _status_code_for_error_field(field: str) -> int | None: + if field.isdecimal() and 400 <= int(field) <= 599: + return int(field) + return _ERROR_CODE_HTTP_STATUS.get(field) + + def _status_code_for_error_fields(error_type: str | None, error_code: str | None) -> int: fields: Final = tuple(field for field in (error_code, error_type) if field is not None) if any(field.startswith("rate_limit") or field == "insufficient_quota" for field in fields): return 429 - return next( - (_ERROR_CODE_HTTP_STATUS[field] for field in fields if field in _ERROR_CODE_HTTP_STATUS), - 500, - ) + return next((status for status in map(_status_code_for_error_field, fields) if status is not None), 500) def _mid_stream_fallback_eligible(mapped_exception: Exception) -> bool: diff --git a/tests/test_litellm/litellm_core_utils/test_exception_mapping_utils.py b/tests/test_litellm/litellm_core_utils/test_exception_mapping_utils.py index acc6248bf3e..0970526956e 100644 --- a/tests/test_litellm/litellm_core_utils/test_exception_mapping_utils.py +++ b/tests/test_litellm/litellm_core_utils/test_exception_mapping_utils.py @@ -1437,6 +1437,29 @@ def test_openai_compatible_vendor_400_keeps_body_but_not_headers(): assert not exc_info.value.response.headers +@pytest.mark.parametrize( + ("status_code", "mapped_class"), [(429, litellm.RateLimitError), (500, litellm.InternalServerError)] +) +def test_openai_429_and_500_keep_body(status_code: int, mapped_class: type[openai.APIError]): + with pytest.raises(mapped_class) as exc_info: + exception_type( + model="gpt-5.4-mini", + original_exception=_openai_handler_error( + "server_error", {}, status_code=status_code, message="upstream cannot complete this response" + ), + custom_llm_provider="openai", + completion_kwargs={}, + extra_kwargs={}, + ) + + assert exc_info.value.body == { + **_GUARDRAIL_BLOCK_ERROR, + "type": "server_error", + "code": str(status_code), + "message": "upstream cannot complete this response", + } + + def test_litellm_proxy_repeated_response_header_keeps_each_value(): repeated = [("x-litellm-call-id", "call-guardrail"), ("set-cookie", "a=1"), ("set-cookie", "b=2")] diff --git a/tests/test_litellm/proxy/proxy_server/test_streaming_helpers.py b/tests/test_litellm/proxy/proxy_server/test_streaming_helpers.py index 53e055882e8..9adc2b80741 100644 --- a/tests/test_litellm/proxy/proxy_server/test_streaming_helpers.py +++ b/tests/test_litellm/proxy/proxy_server/test_streaming_helpers.py @@ -883,6 +883,13 @@ async def test_async_data_generator_mid_stream_exception_yields_error_payload( assert any(isinstance(item, str) and item.startswith('data: {"error":') for item in out) +_UPSTREAM_BODY: Final = { + "code": "cyber_policy", + "message": "Upstream rejected request: flagged for possible cybersecurity risk", + "type": None, +} + + @pytest.mark.asyncio @pytest.mark.parametrize( "terminal,upstream_error,expected_code", @@ -906,6 +913,13 @@ async def test_async_data_generator_mid_stream_exception_yields_error_payload( ), "server_error", id="structured_provider_error_fields", ), + pytest.param( + "upstream_failure", + litellm.InternalServerError( + message="Upstream rejected request", llm_provider="openai", model="gpt-6-astra", body=_UPSTREAM_BODY + ), + "cyber_policy", id="upstream_body_code_and_message", + ), *( pytest.param( "upstream_failure", HTTPException(status_code=status, detail="Upstream rejected request"), @@ -995,6 +1009,8 @@ async def test_responses_stream_keeps_tool_deltas_and_only_emits_a_valid_termina assert "serialize" in failure.response.error["message"].lower() else: assert "Upstream rejected request" in failure.response.error["message"] + if isinstance(upstream_error, litellm.InternalServerError): + assert failure.response.error["message"] == _UPSTREAM_BODY["message"] if isinstance(upstream_error, (HTTPException, litellm.AuthenticationError)): assert upstream_error.status_code == original_status assert payloads[-1]["sequence_number"] > payloads[1]["sequence_number"] diff --git a/tests/test_litellm/proxy/response_api_endpoints/test_endpoints.py b/tests/test_litellm/proxy/response_api_endpoints/test_endpoints.py index 7b79e1b8613..1fb3fef8fb3 100644 --- a/tests/test_litellm/proxy/response_api_endpoints/test_endpoints.py +++ b/tests/test_litellm/proxy/response_api_endpoints/test_endpoints.py @@ -24,6 +24,7 @@ from litellm.proxy.proxy_server import app ("/v1/responses", "numeric_rate_limit"), ("/v1/responses", "server_error"), ("/v1/responses", "response_failed"), + ("/v1/responses", "cyber_policy"), ("/cursor/chat/completions", "server_error"), ("/v1/chat/completions", "server_error"), ], @@ -31,7 +32,7 @@ from litellm.proxy.proxy_server import app async def test_streaming_upstream_errors_keep_the_client_protocol( monkeypatch: pytest.MonkeyPatch, path: str, - error_kind: Literal["rate_limit", "numeric_rate_limit", "server_error", "response_failed"], + error_kind: Literal["rate_limit", "numeric_rate_limit", "server_error", "response_failed", "cyber_policy"], ) -> None: import litellm.proxy.proxy_server as ps from litellm.proxy.auth.user_api_key_auth import user_api_key_auth @@ -40,7 +41,7 @@ async def test_streaming_upstream_errors_keep_the_client_protocol( message: Final = "Upstream cannot complete this response" code: Final = { "rate_limit": "rate_limit_exceeded", "numeric_rate_limit": "429", - "server_error": "server_error", "response_failed": "server_error", + "server_error": "server_error", "response_failed": "server_error", "cyber_policy": "cyber_policy", }[error_kind] error: Final = {"message": message, "code": code, "type": None, "param": "input"} response: Final = {"id": "resp_upstream", "object": "response", "created_at": 1, @@ -55,13 +56,13 @@ async def test_streaming_upstream_errors_keep_the_client_protocol( failed: Final = ( {"type": "response.failed", "sequence_number": 9, "response": {**response, "status": "failed", "error": error}} - if error_kind == "response_failed" else {"type": "error", "error": error} + if error_kind in ("response_failed", "cyber_policy") else {"type": "error", "error": error} ) chat: Final = {"id": "chatcmpl_partial", "object": "chat.completion.chunk", "created": 1, "model": model, "choices": [{"index": 0, "delta": {"content": "partial"}, "finish_reason": None}]} is_chat: Final = path == "/v1/chat/completions" - partial: Final = path != "/v1/responses" or error_kind in ("numeric_rate_limit", "response_failed") + partial: Final = path != "/v1/responses" or error_kind in ("numeric_rate_limit", "response_failed", "cyber_policy") response_events: Final = (created, tool_added, tool_delta, failed) if partial else (failed,) upstream_events: Final = (chat, {"error": error}) if is_chat else response_events wire: Final = "".join("data: " + json.dumps(event) + "\n\n" for event in upstream_events) @@ -108,9 +109,11 @@ async def test_streaming_upstream_errors_keep_the_client_protocol( assert events[0]["sequence_number"] == 0 assert events[0]["response"]["id"].startswith("resp_") assert events[-1]["response"]["status"] == "failed" - assert events[-1]["response"]["error"]["code"] == ( - "rate_limit_exceeded" if error_kind in ("rate_limit", "numeric_rate_limit") else "server_error" - ) + assert events[-1]["response"]["error"]["code"] == { + "rate_limit": "rate_limit_exceeded", "numeric_rate_limit": "rate_limit_exceeded", + "server_error": "server_error", "response_failed": "server_error", "cyber_policy": "cyber_policy", + }[error_kind] + assert events[-1]["response"]["error"]["message"] == message else: assert events[0]["object"] == "chat.completion.chunk", result.text assert "response.failed" not in result.text diff --git a/tests/test_litellm/responses/test_streaming_iterator_error_events.py b/tests/test_litellm/responses/test_streaming_iterator_error_events.py index 3d7c220804a..e7cf09909fe 100644 --- a/tests/test_litellm/responses/test_streaming_iterator_error_events.py +++ b/tests/test_litellm/responses/test_streaming_iterator_error_events.py @@ -340,6 +340,36 @@ def test_maybe_raise_for_response_failed_event_with_dict_error(): assert exc_info.value.status_code == 429 +@pytest.mark.parametrize("code", [429, "429"]) +def test_response_failed_numeric_code_maps_to_its_http_status(code: int | str): + iterator = _make_iterator() + mock_response_obj = Mock() + mock_response_obj.error = {"code": code, "message": "throttled"} + chunk = Mock() + chunk.type = "response.failed" + chunk.response = mock_response_obj + with pytest.raises(MidStreamFallbackError) as exc_info: + iterator._maybe_raise_for_error_event(chunk) + assert exc_info.value.status_code == 429 + assert isinstance(exc_info.value.original_exception, litellm.RateLimitError) + + +def test_response_failed_unknown_code_keeps_upstream_code_and_message_on_mapped_exception(): + iterator = _make_iterator() + upstream_message = "This content was flagged for possible cybersecurity risk." + mock_response_obj = Mock() + mock_response_obj.error = {"code": "cyber_policy", "message": upstream_message} + chunk = Mock() + chunk.type = "response.failed" + chunk.response = mock_response_obj + with pytest.raises(MidStreamFallbackError) as exc_info: + iterator._maybe_raise_for_error_event(chunk) + mapped = exc_info.value.original_exception + assert isinstance(mapped, litellm.InternalServerError) + assert mapped.code == "cyber_policy" + assert mapped.body == {"message": upstream_message, "type": None, "code": "cyber_policy"} + + def test_maybe_raise_for_error_event_null_error_obj(): """error chunk with no error field: message and code default; wrapped as 500.""" iterator = _make_iterator() @@ -523,6 +553,9 @@ def test_every_openai_sdk_response_error_code_has_explicit_status_mapping(): ("failed_to_download_image", 400), ("image_file_not_found", 400), ("totally_unknown_future_code", 500), + ("429", 429), + ("503", 503), + ("200", 500), ], ) def test_status_code_for_documented_response_error_codes(code: str, expected_status: int): From eb96d885ceff1b8daa6c04330e819e39653ba33b Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Fri, 18 Sep 2026 14:16:47 -0700 Subject: [PATCH 5/7] fix(proxy): end failed responses streams with [DONE] Emit data: [DONE] after event: response.failed, and after a late failure when a terminal event already went out, so OpenAI SDK clients see the same stream end as a completed response. Restore the lazy OpenAPI snapshot to its Python 3.12 rendering, which is what CI regenerates. --- litellm/proxy/_lazy_openapi_snapshot.json | 2 +- litellm/proxy/proxy_server.py | 6 ++++-- .../proxy/proxy_server/test_streaming_helpers.py | 2 ++ .../proxy/response_api_endpoints/test_endpoints.py | 3 ++- 4 files changed, 9 insertions(+), 4 deletions(-) diff --git a/litellm/proxy/_lazy_openapi_snapshot.json b/litellm/proxy/_lazy_openapi_snapshot.json index 2aa2cf15ac1..8a8d08c6887 100644 --- a/litellm/proxy/_lazy_openapi_snapshot.json +++ b/litellm/proxy/_lazy_openapi_snapshot.json @@ -19394,7 +19394,7 @@ } } }, - "description": "\nUnified rate-limit error.\n\nEvery rate-limit condition surfaced by litellm \u2014 whether it originated from\nan upstream LLM provider, a vendor batch endpoint, or one of litellm's own\nproxy-side limiters (parallel-requests, dynamic-rate, batch-rate, budget,\nmax-iterations, etc.) \u2014 is raised as an instance of this class.\n\nThe :attr:`category` attribute lets callers distinguish the source. See\n:class:`RateLimitErrorCategory` for the available values.\n" + "description": "\n Unified rate-limit error.\n\n Every rate-limit condition surfaced by litellm \u2014 whether it originated from\n an upstream LLM provider, a vendor batch endpoint, or one of litellm's own\n proxy-side limiters (parallel-requests, dynamic-rate, batch-rate, budget,\n max-iterations, etc.) \u2014 is raised as an instance of this class.\n\n The :attr:`category` attribute lets callers distinguish the source. See\n :class:`RateLimitErrorCategory` for the available values.\n " }, "500": { "content": { diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 9cd233e5ad5..38a883a3b1b 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -8865,6 +8865,7 @@ def _format_streaming_sse_chunk(chunk: str | bytes) -> str | bytes: _SSE_FRAME_DELIMITERS: Final = ("\r\n\r\n", "\n\n", "\r\r") +_OPENAI_STREAM_DONE_FRAME: Final = "data: [DONE]\n\n" _MAX_RAW_SSE_BUFFER_CHARS: Final = 8 * 1024 * 1024 @@ -9272,8 +9273,7 @@ async def async_data_generator( yield error_message # OpenAI-compatible streams terminate with data: [DONE]; Google GenAI (?alt=sse) does not. if not request_data.get("_litellm_skip_openai_stream_done"): - done_message: Final = "[DONE]" - yield f"data: {done_message}\n\n" + yield _OPENAI_STREAM_DONE_FRAME except (asyncio.CancelledError, GeneratorExit): # Client disconnected mid-stream. CancelledError / GeneratorExit are # BaseException, so they bypass the success/failure logging callbacks @@ -9303,6 +9303,8 @@ async def async_data_generator( error_frame: Final = error_state.format_failure(e) if error_frame is not None: yield error_frame + if not request_data.get("_litellm_skip_openai_stream_done"): + yield _OPENAI_STREAM_DONE_FRAME return if isinstance(e, HTTPException): raise e diff --git a/tests/test_litellm/proxy/proxy_server/test_streaming_helpers.py b/tests/test_litellm/proxy/proxy_server/test_streaming_helpers.py index 9adc2b80741..86dd356e5f5 100644 --- a/tests/test_litellm/proxy/proxy_server/test_streaming_helpers.py +++ b/tests/test_litellm/proxy/proxy_server/test_streaming_helpers.py @@ -995,6 +995,8 @@ async def test_responses_stream_keeps_tool_deltas_and_only_emits_a_valid_termina for frame in event_frames ) + assert decoded[-1] == "data: [DONE]\n\n" + assert len(decoded) == len(event_frames) + 1 assert payloads[0]["response"]["id"] == "resp_visible" assert payloads[1] == tool_delta.model_dump() assert len(payloads) == 3 diff --git a/tests/test_litellm/proxy/response_api_endpoints/test_endpoints.py b/tests/test_litellm/proxy/response_api_endpoints/test_endpoints.py index 1fb3fef8fb3..63faef2366f 100644 --- a/tests/test_litellm/proxy/response_api_endpoints/test_endpoints.py +++ b/tests/test_litellm/proxy/response_api_endpoints/test_endpoints.py @@ -95,7 +95,8 @@ async def test_streaming_upstream_errors_keep_the_client_protocol( assert result.status_code == 200, result.text assert message in result.text if path == "/v1/responses": - assert frames[-1].startswith("event: response.failed\n"), result.text + assert frames[-1] == "data: [DONE]", result.text + assert frames[-2].startswith("event: response.failed\n"), result.text if partial: assert [event["type"] for event in events] == [ "response.created", "response.output_item.added", From 76d1abba723116da947f81368fe70939ad976b3c Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Fri, 18 Sep 2026 14:57:15 -0700 Subject: [PATCH 6/7] refactor(responses): map status codes to error codes with a lookup Ends _response_error_code in an unconditional return so CodeQL stops flagging mixed explicit and implicit returns. No behavior change: every status maps as before. --- .../common_utils/responses_stream_errors.py | 33 ++++++++++--------- 1 file changed, 18 insertions(+), 15 deletions(-) diff --git a/litellm/proxy/common_utils/responses_stream_errors.py b/litellm/proxy/common_utils/responses_stream_errors.py index 63f881c126e..25b83a5ba34 100644 --- a/litellm/proxy/common_utils/responses_stream_errors.py +++ b/litellm/proxy/common_utils/responses_stream_errors.py @@ -73,6 +73,23 @@ def _failure_details(original: Exception) -> _FailureDetails: ) +_CLIENT_ERROR_CODES: Final = MappingProxyType( + { + int(HTTPStatus.UNAUTHORIZED): "authentication_error", + int(HTTPStatus.FORBIDDEN): "permission_error", + int(HTTPStatus.NOT_FOUND): "not_found_error", + int(HTTPStatus.REQUEST_TIMEOUT): "request_timeout", + int(HTTPStatus.TOO_MANY_REQUESTS): "rate_limit_exceeded", + } +) + + +def _status_error_code(status_code: int | None) -> str: + if status_code is None or not HTTPStatus.BAD_REQUEST <= status_code < HTTPStatus.INTERNAL_SERVER_ERROR: + return "server_error" + return _CLIENT_ERROR_CODES.get(status_code, "invalid_request_error") + + def _response_error_code(details: _FailureDetails) -> str: for value in (details.code, details.type): if value == "insufficient_quota": @@ -81,21 +98,7 @@ def _response_error_code(details: _FailureDetails) -> str: return "rate_limit_exceeded" if isinstance(details.code, str) and details.code and not details.code.isdecimal(): return details.code - match details.status_code: - case HTTPStatus.UNAUTHORIZED: - return "authentication_error" - case HTTPStatus.FORBIDDEN: - return "permission_error" - case HTTPStatus.NOT_FOUND: - return "not_found_error" - case HTTPStatus.REQUEST_TIMEOUT: - return "request_timeout" - case HTTPStatus.TOO_MANY_REQUESTS: - return "rate_limit_exceeded" - case int(status) if HTTPStatus.BAD_REQUEST <= status < HTTPStatus.INTERNAL_SERVER_ERROR: - return "invalid_request_error" - case _: - return "server_error" + return _status_error_code(details.status_code) class ResponsesStreamErrorState: From c181c927d0b7a4ad214a4dd160c2f16f1373385e Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Fri, 18 Sep 2026 15:58:15 -0700 Subject: [PATCH 7/7] fix(proxy): record response.failed frames in background polling --- .../response_polling/background_streaming.py | 7 +++- .../test_response_polling_handler.py | 38 +++++++++++++++++++ 2 files changed, 43 insertions(+), 2 deletions(-) diff --git a/litellm/proxy/response_polling/background_streaming.py b/litellm/proxy/response_polling/background_streaming.py index fac45d4391c..b13042dfb6c 100644 --- a/litellm/proxy/response_polling/background_streaming.py +++ b/litellm/proxy/response_polling/background_streaming.py @@ -75,6 +75,10 @@ class _StreamEventParser: parse: Callable[[str], _StreamEvent] = staticmethod(json.loads) +def _sse_frame_data(frame: str) -> str | None: + return next((line[6:].strip() for line in frame.splitlines() if line.startswith("data: ")), None) + + async def _never_receive() -> Message: await asyncio.Event().wait() raise AssertionError("unreachable") @@ -224,8 +228,7 @@ async def background_streaming_task( if isinstance(chunk, bytes): chunk = chunk.decode("utf-8") - if isinstance(chunk, str) and chunk.startswith("data: "): - chunk_data = chunk[6:].strip() + if isinstance(chunk, str) and (chunk_data := _sse_frame_data(chunk)) is not None: if chunk_data == "[DONE]": break diff --git a/tests/proxy_unit_tests/test_response_polling_handler.py b/tests/proxy_unit_tests/test_response_polling_handler.py index 467c1332325..81ca0114a8d 100644 --- a/tests/proxy_unit_tests/test_response_polling_handler.py +++ b/tests/proxy_unit_tests/test_response_polling_handler.py @@ -1482,6 +1482,44 @@ class TestBackgroundStreamingTerminalEvents: assert final_call.kwargs["status"] == "failed" assert final_call.kwargs["error"] == error_payload + @pytest.mark.asyncio + async def test_named_event_failed_frame_sets_failed_status_and_error(self): + from litellm.proxy.response_polling.background_streaming import ( + background_streaming_task, + ) + + error_payload = { + "code": "cyber_policy", + "message": "Your request was flagged for possible cybersecurity risk and was not completed", + } + failed_event = { + "type": "response.failed", + "sequence_number": 5, + "response": {"id": "resp_123", "status": "failed", "error": error_payload, "output": []}, + } + + async def _body_iterator(): + yield b'data: {"type": "response.in_progress"}\n\n' + yield f"event: response.failed\ndata: {json.dumps(failed_event)}\n\n".encode() + yield b"data: [DONE]\n\n" + + mock_response = Mock() + mock_response.body_iterator = _body_iterator() + handler = AsyncMock(spec=ResponsePollingHandler) + kwargs = _make_background_streaming_kwargs("poll_named_event", 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 = handler.update_state.call_args_list[-1] + assert final_call.kwargs["status"] == "failed" + assert final_call.kwargs["error"] == error_payload + @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"""