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 <bliu1@berkeley.edu>

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,

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 = (

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(