feat(providers): retry upstream HTTP 503 like 429

Extend execute_with_retry with retryable_upstream_status for 429/503
shared backoff, reactive block, and TRACE provider.retry.scheduled.
Map InternalServerError(503) to APIError with preserved status.
Native transport closes-and-raises on 503 like 429 for clean retries.
Add NIM/OpenAI-compat and rate limiter tests.

providers/anthropic_messages.py

@@ -360,9 +360,9 @@ class AnthropicMessagesTransport(BaseProvider):
             try:
 
                 async def _validated_stream_send() -> httpx.Response:
-                    """Send request; raise inside retry loop on 429 so rate limiter can backoff."""
+                    """Send request; raise inside retry loop on 429/503 so limiter can backoff."""
                     send_response = await self._send_stream_request(body)
-                    if send_response.status_code == 429:
+                    if send_response.status_code in (429, 503):
                         await send_response.aclose()
                         send_response.raise_for_status()
                     if send_response.status_code != 200:

providers/error_mapping.py

@@ -50,6 +50,14 @@ def map_error(
         return InvalidRequestError(message, raw_error=str(e))
     if isinstance(e, openai.InternalServerError):
         raw_message = str(e)
+        sdk_status = getattr(e, "status_code", None)
+        if sdk_status == 503:
+            stable = APIError("_", status_code=503)
+            return APIError(
+                get_user_facing_error_message(stable),
+                status_code=503,
+                raw_error=str(e),
+            )
         if "overloaded" in raw_message.lower() or "capacity" in raw_message.lower():
             return OverloadedError(message, raw_error=raw_message)
         return APIError(message, status_code=500, raw_error=str(e))

providers/rate_limit.py

@@ -12,10 +12,32 @@ import openai
 from loguru import logger
 
 from core.rate_limit import StrictSlidingWindowLimiter
+from core.trace import trace_event
 
 T = TypeVar("T")
 
 
+def retryable_upstream_status(exc: BaseException) -> int | None:
+    """Return HTTP-like status codes that qualify for reactive backoff retries.
+
+    ``429`` and ``503`` use the same exponential backoff plus scoped limiter
+    blocking semantics as today's rate-limit path.
+    """
+    if isinstance(exc, openai.RateLimitError):
+        return 429
+    if isinstance(exc, httpx.HTTPStatusError):
+        status = exc.response.status_code
+        if status in (429, 503):
+            return status
+        return None
+    if isinstance(exc, openai.APIError):
+        status = getattr(exc, "status_code", None)
+        if isinstance(status, int) and status == 503:
+            return 503
+        return None
+    return None
+
+
 class GlobalRateLimiter:
     """
     Global singleton rate limiter that blocks all requests
@@ -26,7 +48,7 @@ class GlobalRateLimiter:
     may be open simultaneously, independent of the sliding window.
 
     Proactive limits - throttles requests to stay within API limits.
-    Reactive limits - pauses all requests when a 429 is hit.
+    Reactive limits - pauses all requests when a 429 or 503 retry backoff is active.
     Concurrency limit - caps simultaneously open streams.
     """
 
@@ -203,10 +225,11 @@ class GlobalRateLimiter:
         jitter: float = 1.0,
         **kwargs: Any,
     ) -> Any:
-        """Execute an async callable with rate limiting and retry on 429.
+        """Execute an async callable with rate limiting and retry on transient limits.
 
-        Waits for the proactive limiter before each attempt. On 429, applies
-        exponential backoff with jitter before retrying.
+        Waits for the proactive limiter before each attempt. On ``429`` (rate limit)
+        or ``503`` (service unavailable), applies exponential backoff with jitter
+        and sets the reactive block before retrying.
 
         Args:
             fn: Async callable to execute.
@@ -222,43 +245,51 @@ class GlobalRateLimiter:
             The last exception if all retries are exhausted.
         """
         last_exc: Exception | None = None
+        total_attempts = 1 + max_retries
 
-        for attempt in range(1 + max_retries):
+        for attempt in range(total_attempts):
             await self.wait_if_blocked()
 
             try:
                 return await fn(*args, **kwargs)
-            except openai.RateLimitError as e:
-                last_exc = e
-                if attempt >= max_retries:
-                    logger.warning(
-                        f"Rate limit retry exhausted after {max_retries} retries"
-                    )
-                    break
-
-                delay = min(base_delay * (2**attempt), max_delay)
-                delay += random.uniform(0, jitter)
-                logger.warning(
-                    f"Rate limited (429), attempt {attempt + 1}/{max_retries + 1}. "
-                    f"Retrying in {delay:.1f}s..."
-                )
-                self.set_blocked(delay)
-                await asyncio.sleep(delay)
-            except httpx.HTTPStatusError as e:
-                if e.response.status_code != 429:
+            except Exception as e:
+                status = retryable_upstream_status(e)
+                if status is None:
                     raise
+
+                label = (
+                    "Rate limited (429)"
+                    if status == 429
+                    else "Upstream unavailable (503)"
+                )
                 last_exc = e
                 if attempt >= max_retries:
                     logger.warning(
-                        f"HTTP 429 retry exhausted after {max_retries} retries"
+                        "{} retry exhausted after {} retries (attempts={})",
+                        label,
+                        max_retries,
+                        total_attempts,
                     )
                     break
 
                 delay = min(base_delay * (2**attempt), max_delay)
                 delay += random.uniform(0, jitter)
+                attempt_no = attempt + 1
                 logger.warning(
-                    f"HTTP 429 from upstream, attempt {attempt + 1}/{max_retries + 1}. "
-                    f"Retrying in {delay:.1f}s..."
+                    "{}, attempt {}/{}. Retrying in {:.1f}s...",
+                    label,
+                    attempt_no,
+                    total_attempts,
+                    delay,
+                )
+                trace_event(
+                    stage="provider",
+                    event="provider.retry.scheduled",
+                    source="provider",
+                    status_code=status,
+                    attempt=attempt_no,
+                    max_attempts=total_attempts,
+                    delay_s=round(delay, 3),
                 )
                 self.set_blocked(delay)
                 await asyncio.sleep(delay)

tests/providers/test_anthropic_messages_429_retry.py

@@ -1,4 +1,4 @@
-"""Native Anthropic transport: HTTP 429 is retried inside execute_with_retry."""
+"""Native Anthropic transport: HTTP 429 and 503 are retried inside execute_with_retry."""
 
 from contextlib import asynccontextmanager
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -80,6 +80,108 @@ async def test_native_stream_retries_on_http_429_then_streams(provider_config):
         GlobalRateLimiter.reset_instance()
 
 
+@pytest.mark.asyncio
+async def test_native_stream_retries_on_http_503_then_streams(provider_config):
+    """First response 503 (closed), second 200 streams; send is called twice."""
+    GlobalRateLimiter.reset_instance()
+    try:
+        provider = NativeProvider(provider_config)
+        req = MockRequest()
+        request_obj = httpx.Request("POST", "https://custom.test/v1/messages")
+        ok_lines = [
+            "event: message_start",
+            'data: {"type":"message_start"}',
+            "",
+        ]
+        ok_response = FakeResponse(lines=ok_lines)
+        unavailable = FakeResponse(status_code=503, text="Service Unavailable")
+
+        send_calls = {"n": 0}
+
+        async def send_side_effect(*_a, **_kw):
+            send_calls["n"] += 1
+            if send_calls["n"] == 1:
+                return unavailable
+            return ok_response
+
+        with (
+            patch.object(provider._client, "build_request", return_value=request_obj),
+            patch.object(
+                provider._client,
+                "send",
+                new_callable=AsyncMock,
+                side_effect=send_side_effect,
+            ),
+            patch(
+                "asyncio.sleep",
+                new_callable=AsyncMock,
+            ),
+        ):
+            events = [e async for e in provider.stream_response(req)]
+
+        assert send_calls["n"] == 2
+        assert unavailable.is_closed
+        assert ok_response.is_closed
+        assert events == [
+            "event: message_start\n",
+            'data: {"type":"message_start"}\n',
+            "\n",
+        ]
+    finally:
+        GlobalRateLimiter.reset_instance()
+
+
+@pytest.mark.asyncio
+async def test_native_stream_503_retry_exhausted(provider_config):
+    """Repeated HTTP 503 exhausts execute_with_retry; emits overloaded-style message."""
+    GlobalRateLimiter.reset_instance()
+    try:
+
+        @asynccontextmanager
+        async def _slot():
+            yield
+
+        with patch("providers.anthropic_messages.GlobalRateLimiter") as mock_gl:
+            instance = mock_gl.get_scoped_instance.return_value
+            real = GlobalRateLimiter(
+                rate_limit=100,
+                rate_window=60,
+                max_concurrency=5,
+            )
+            instance.wait_if_blocked = real.wait_if_blocked
+            instance.execute_with_retry = real.execute_with_retry
+            instance.set_blocked = real.set_blocked
+            instance.concurrency_slot.side_effect = _slot
+
+            provider = NativeProvider(provider_config)
+            req = MockRequest()
+
+            unavailable = FakeResponse(status_code=503, text="Service Unavailable")
+
+            with (
+                patch.object(
+                    provider._client, "build_request", return_value=MagicMock()
+                ),
+                patch.object(
+                    provider._client,
+                    "send",
+                    new_callable=AsyncMock,
+                    return_value=unavailable,
+                ) as mock_send,
+                patch("asyncio.sleep", new_callable=AsyncMock),
+            ):
+                events = [e async for e in provider.stream_response(req)]
+
+            assert mock_send.await_count == 4
+            assert unavailable.is_closed
+            assert_canonical_stream_error_envelope(
+                events,
+                user_message_substr="Provider is currently overloaded",
+            )
+    finally:
+        GlobalRateLimiter.reset_instance()
+
+
 @pytest.mark.asyncio
 async def test_non_429_http_error_not_retried(provider_config):
     """HTTP 500 from upstream is not retried; single send."""

tests/providers/test_error_mapping.py

@@ -88,6 +88,18 @@ class TestMapError:
         assert isinstance(result, APIError)
         assert result.status_code == 500
 
+    def test_internal_server_error_503_maps_to_api_error_with_status(self):
+        """InternalServerError carrying HTTP 503 retains 503 for stable user messaging."""
+        exc = _make_openai_error(
+            openai.InternalServerError,
+            message="<html>503</html>",
+            status_code=503,
+        )
+        result = map_error(exc)
+        assert isinstance(result, APIError)
+        assert result.status_code == 503
+        assert "temporarily unavailable" in result.message.lower()
+
     def test_generic_api_error(self):
         """openai.APIError -> APIError with original status_code."""
         exc = _make_openai_error(

tests/providers/test_openai_compat_503_retry.py

@@ -0,0 +1,100 @@
+"""OpenAI-compat transports: HTTP 503 uses the same execute_with_retry path as 429."""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import openai
+import pytest
+from httpx import Request, Response
+
+from config.nim import NimSettings
+from providers.base import ProviderConfig
+from providers.nvidia_nim import NvidiaNimProvider
+from providers.rate_limit import GlobalRateLimiter
+from tests.providers.test_nvidia_nim import MockRequest
+
+
+def _internal_503() -> openai.InternalServerError:
+    return openai.InternalServerError(
+        "unavailable",
+        response=Response(503, request=Request("POST", "http://x")),
+        body={},
+    )
+
+
+@pytest.mark.asyncio
+async def test_nim_stream_retries_on_openai_503_then_streams():
+    GlobalRateLimiter.reset_instance()
+    try:
+        config = ProviderConfig(
+            api_key="test_key",
+            base_url="https://test.api.nvidia.com/v1",
+            rate_limit=100,
+            rate_window=60,
+            http_read_timeout=600.0,
+            http_write_timeout=15.0,
+            http_connect_timeout=5.0,
+        )
+        provider = NvidiaNimProvider(config, nim_settings=NimSettings())
+        req = MockRequest()
+
+        mock_chunk = MagicMock()
+        mock_chunk.choices = [
+            MagicMock(
+                delta=MagicMock(content="Hi", reasoning_content=""),
+                finish_reason=None,
+            )
+        ]
+        mock_chunk.usage = None
+
+        async def mock_stream():
+            yield mock_chunk
+
+        with (
+            patch.object(
+                provider._client.chat.completions,
+                "create",
+                new_callable=AsyncMock,
+            ) as mock_create,
+            patch("asyncio.sleep", new_callable=AsyncMock),
+        ):
+            mock_create.side_effect = [_internal_503(), mock_stream()]
+            events = [e async for e in provider.stream_response(req)]
+
+        assert mock_create.await_count == 2
+        assert any("Hi" in e for e in events)
+    finally:
+        GlobalRateLimiter.reset_instance()
+
+
+@pytest.mark.asyncio
+async def test_nim_stream_openai_503_exhausted_emits_user_message():
+    GlobalRateLimiter.reset_instance()
+    try:
+        config = ProviderConfig(
+            api_key="test_key",
+            base_url="https://test.api.nvidia.com/v1",
+            rate_limit=100,
+            rate_window=60,
+            http_read_timeout=600.0,
+            http_write_timeout=15.0,
+            http_connect_timeout=5.0,
+        )
+        provider = NvidiaNimProvider(config, nim_settings=NimSettings())
+        req = MockRequest()
+
+        with (
+            patch.object(
+                provider._client.chat.completions,
+                "create",
+                new_callable=AsyncMock,
+            ) as mock_create,
+            patch("asyncio.sleep", new_callable=AsyncMock),
+        ):
+            mock_create.side_effect = _internal_503()
+            events = [e async for e in provider.stream_response(req)]
+
+        assert mock_create.await_count == 4
+        blob = "".join(events)
+        assert "temporarily unavailable" in blob.lower()
+    finally:
+        GlobalRateLimiter.reset_instance()

tests/providers/test_provider_rate_limit.py

@@ -278,6 +278,88 @@ class TestProviderRateLimiter:
         assert result == "ok"
         assert call_count == 2
 
+    @pytest.mark.asyncio
+    async def test_execute_with_retry_succeeds_on_openai_internal_server_error_503(
+        self,
+    ):
+        """503 as openai.InternalServerError then success."""
+        import openai
+        from httpx import Request, Response
+
+        GlobalRateLimiter.reset_instance()
+        limiter = GlobalRateLimiter.get_instance(rate_limit=100, rate_window=60)
+
+        def make_503():
+            return openai.InternalServerError(
+                "unavailable",
+                response=Response(503, request=Request("POST", "http://x")),
+                body={},
+            )
+
+        call_count = 0
+
+        async def fail_then_ok():
+            nonlocal call_count
+            call_count += 1
+            if call_count == 1:
+                raise make_503()
+            return "ok"
+
+        result = await limiter.execute_with_retry(
+            fail_then_ok, max_retries=2, base_delay=0.01, max_delay=0.1, jitter=0
+        )
+        assert result == "ok"
+        assert call_count == 2
+
+    @pytest.mark.asyncio
+    async def test_execute_with_retry_succeeds_on_httpx_503(self):
+        """HTTP 503 as httpx.HTTPStatusError then success."""
+        import httpx
+        from httpx import Request, Response
+
+        limiter = GlobalRateLimiter.get_instance(rate_limit=100, rate_window=60)
+
+        call_count = 0
+
+        async def fail_then_ok():
+            nonlocal call_count
+            call_count += 1
+            if call_count == 1:
+                r = Response(503, request=Request("POST", "http://x"), text="busy")
+                raise httpx.HTTPStatusError(
+                    "Service Unavailable", request=r.request, response=r
+                )
+            return "ok"
+
+        result = await limiter.execute_with_retry(
+            fail_then_ok, max_retries=2, base_delay=0.01, max_delay=0.1, jitter=0
+        )
+        assert result == "ok"
+        assert call_count == 2
+
+    @pytest.mark.asyncio
+    async def test_execute_with_retry_exhaust_openai_503_raises(self):
+        """When all 503 retries exhausted (openai), last InternalServerError is raised."""
+        import openai
+        from httpx import Request, Response
+
+        GlobalRateLimiter.reset_instance()
+        limiter = GlobalRateLimiter.get_instance(rate_limit=100, rate_window=60)
+
+        exc = openai.InternalServerError(
+            "unavailable",
+            response=Response(503, request=Request("POST", "http://x")),
+            body={},
+        )
+
+        async def always_503():
+            raise exc
+
+        with pytest.raises(openai.InternalServerError):
+            await limiter.execute_with_retry(
+                always_503, max_retries=2, base_delay=0.01, max_delay=0.1, jitter=0
+            )
+
     @pytest.mark.asyncio
     async def test_max_concurrency_zero_raises(self):
         """max_concurrency <= 0 raises ValueError."""