From bc1309f18eb2c440782500e4c5c902f5e4e0e1de Mon Sep 17 00:00:00 2001 From: mrinal-berri Date: Fri, 6 Nov 2026 09:55:30 -0800 Subject: [PATCH] fix(responses): map streamed flex_unavailable errors to 429 not 500 (#50336) * fix(responses): map streamed flex_unavailable errors to 429 not 500 When OpenAI has no flex capacity, a service_tier flex request fails with type resource_unavailable / code flex_unavailable. Non-streamed, this arrives as HTTP 429 and litellm raises RateLimitError. Streamed /v1/responses delivers the same error as an in-stream event, and neither field was in _ERROR_CODE_HTTP_STATUS, so it defaulted to 500 InternalServerError. The HTTP status then depended only on stream: one provider outage looked like a 429 or a 500. The 500 reads as a server error in metrics and alerting, and callers that retry on 429 do not retry it. Map both resource_unavailable and flex_unavailable to 429 so the streamed path matches the non-streamed path. Fixes #49736 * test(responses): cover streamed flex outage end to end for error and response.failed events Streams the flex_unavailable payload from #49736 through litellm.aresponses with an injected client and asserts a 429 RateLimitError for both in-stream event kinds, matching the non-streamed path * test(responses): drop explanatory comment from flex status mapping test * test(integration): cover streamed flex outage status on responses and chat bridge Drives the flex_unavailable payload from #49736 through a real proxy as an in-stream error event and as response.failed. The native /v1/responses stream fails with rate_limit_exceeded and logs 429, and the chat bridge answers a 429 throttling error, where both reported a 500 before the mapping fix --------- Co-authored-by: BinYi LIU --- litellm/responses/streaming_iterator.py | 2 + .../test_responses_top_level_error_events.py | 34 ++++++++ .../test_streaming_iterator_error_events.py | 78 +++++++++++++++++++ 3 files changed, 114 insertions(+) diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index 288e4a7119d..0c002a770e8 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -250,6 +250,8 @@ _ERROR_CODE_HTTP_STATUS: Final[Mapping[str, int]] = MappingProxyType( "server_error": 500, "rate_limit_exceeded": 429, "insufficient_quota": 429, + "resource_unavailable": 429, + "flex_unavailable": 429, "vector_store_timeout": 504, "invalid_prompt": 400, "invalid_image": 400, diff --git a/tests/integration/streaming/test_responses_top_level_error_events.py b/tests/integration/streaming/test_responses_top_level_error_events.py index 63bdadceebc..2cbc4570dec 100644 --- a/tests/integration/streaming/test_responses_top_level_error_events.py +++ b/tests/integration/streaming/test_responses_top_level_error_events.py @@ -22,12 +22,14 @@ from integration._support.responses_stream import ( data_frames, delta, error_body, + error_event, frame, function_tools, healthy_stream, messages_body, openai_error_event, openai_error_stream, + response_object, serve, spend_row, sse_events, @@ -277,6 +279,38 @@ def test_native_responses_stream_fails_with_the_provider_code_and_message(gatewa _assert_failure_logged(response.headers["x-litellm-call-id"], "429", RATE_LIMIT_MESSAGE) +_FLEX_MESSAGE: Final = "Flex processing is temporarily unavailable. Please try again later or use standard processing." +_FLEX_ERROR: Final = {"type": "resource_unavailable", "code": "flex_unavailable", "message": _FLEX_MESSAGE} + + +def _flex_outage(identity: str, kind: str) -> tuple[bytes, ...]: + failure: Final = ( + error_event(_FLEX_ERROR) + if kind == "error" + else { + "type": "response.failed", + "sequence_number": 1, + "response": response_object(identity, "failed", error=_FLEX_ERROR), + } + ) + return (frame(created(identity)), frame(failure)) + + +@pytest.mark.parametrize("kind", ["error", "response.failed"]) +def test_native_responses_stream_flex_outage_fails_as_a_rate_limit(gateway: Gateway, kind: str) -> None: + response, error = _responses_frames(gateway, _flex_outage(_identity(), kind)) + assert error["code"] == "rate_limit_exceeded" and _FLEX_MESSAGE in string_value(error["message"]), error + _assert_failure_logged(response.headers["x-litellm-call-id"], "429", _FLEX_MESSAGE) + + +@pytest.mark.parametrize("kind", ["error", "response.failed"]) +def test_chat_stream_flex_outage_is_a_throttling_error_like_the_non_streamed_429(gateway: Gateway, kind: str) -> None: + response, _ = _chat(gateway, _flex_outage(_identity(), kind)) + assert response.status_code == 429, response.text + assert_provider_typed_rate_limit(error_body(response), _FLEX_MESSAGE) + _assert_failure_logged(response.headers["x-litellm-call-id"], "429", _FLEX_MESSAGE) + + def test_native_responses_stream_error_after_text_fails_with_the_provider_code(gateway: Gateway) -> None: identity: Final = _identity() stream: Final = ( diff --git a/tests/unit/responses/test_streaming_iterator_error_events.py b/tests/unit/responses/test_streaming_iterator_error_events.py index 14ecb185f06..0acf5d9ce62 100644 --- a/tests/unit/responses/test_streaming_iterator_error_events.py +++ b/tests/unit/responses/test_streaming_iterator_error_events.py @@ -37,6 +37,7 @@ from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging from litellm.litellm_core_utils.get_litellm_params import get_litellm_params from litellm.litellm_core_utils.litellm_logging import _get_provider_request_id from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig +from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler from litellm.responses.streaming_iterator import ( _ERROR_CODE_HTTP_STATUS, BaseResponsesAPIStreamingIterator, @@ -75,6 +76,83 @@ def _make_error_chunk(error_type: str, code: str, message: str = "err") -> Error return ErrorEvent(type=ResponsesAPIStreamEvents.ERROR, sequence_number=0, error=error_obj) +@pytest.mark.parametrize( + "error_type,error_code", + ( + ("resource_unavailable", "flex_unavailable"), + ("resource_unavailable", None), + (None, "flex_unavailable"), + ), +) +def test_streamed_flex_unavailable_maps_to_429(error_type: str | None, error_code: str | None) -> None: + assert _status_code_for_error_fields(error_type, error_code) == 429 + + +def test_streamed_unrelated_server_error_still_maps_to_500() -> None: + assert _status_code_for_error_fields("server_error", "server_error") == 500 + + +FLEX_UNAVAILABLE_ERROR: Final = { + "type": "resource_unavailable", + "code": "flex_unavailable", + "message": "Flex processing is temporarily unavailable. Please try again later or use standard processing.", +} + + +def _flex_outage_sse(kind: str) -> str: + pending: Final = { + "id": "resp_flex", + "object": "response", + "created_at": 1700000000, + "model": "gpt-5", + "status": "in_progress", + "output": [], + } + failure: Final = ( + {"type": "error", "sequence_number": 1, "error": FLEX_UNAVAILABLE_ERROR} + if kind == "error" + else { + "type": "response.failed", + "sequence_number": 1, + "response": {**pending, "status": "failed", "error": FLEX_UNAVAILABLE_ERROR}, + } + ) + events: Final = ({"type": "response.created", "sequence_number": 0, "response": pending}, failure) + return "".join(f"event: {event['type']}\ndata: {json.dumps(event)}\n\n" for event in events) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("kind", ("error", "response.failed")) +async def test_streamed_flex_outage_raises_rate_limit_like_the_non_streamed_path( + kind: str, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + litellm.in_memory_llm_clients_cache.flush_cache() + async_client: Final = AsyncHTTPHandler( + transport=httpx.MockTransport( + lambda request: httpx.Response( + 200, headers={"content-type": "text/event-stream"}, text=_flex_outage_sse(kind) + ) + ) + ) + try: + stream: Final = await litellm.aresponses( + model="openai/gpt-5", + input="hi", + service_tier="flex", + stream=True, + api_key="sk-test", + client=async_client, + ) + with pytest.raises(MidStreamFallbackError) as raised: + _ = [event async for event in stream] + finally: + await async_client.close() + assert raised.value.status_code == 429, raised.value + assert isinstance(raised.value.original_exception, litellm.RateLimitError), raised.value.original_exception + assert "Flex processing is temporarily unavailable" in str(raised.value.original_exception) + + @pytest.mark.parametrize("provider", ("openai", "azure", "bedrock")) @pytest.mark.parametrize("header", ("x-amzn-RequestId", "x-request-id", None)) @pytest.mark.parametrize(