feat: add provider retry with backoff and API rate limiting

Part A: OpenAI 429 errors trigger exponential backoff (1s, 2s, 4s)
before raising ProviderRateLimitError. Retry wraps the raw openai call
inside the existing error translation — not outside it.

Part B: In-memory sliding window rate limiter (10 RPM per IP default).
/health and /metrics exempt. 429 response with Retry-After header.

- RetryConfig added to AppConfig (max_retries, base_delay, max_delay)
- rate_limit_rpm added to ServingConfig
- RateLimitMiddleware registered in app.py
- 8 new tests (117 total), lint + types clean
- DECISIONS.md entries for both

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

DECISIONS.md

@@ -168,6 +168,32 @@ than a computed subset. The reranker's `top_k` handles truncation.
 This is simpler and more robust than computing an input size from
 per-system candidate counts.
 
+## Why provider retry with exponential backoff
+
+OpenAI returns 429 (rate limit) errors under load. Without retry logic,
+a single 429 causes a user-visible failure. We add exponential backoff:
+attempt after 1s, 2s, 4s. After 3 retries, raise `ProviderRateLimitError`
+so the middleware returns a clear 503.
+
+The retry wraps the raw `openai.RateLimitError` — it must fire BEFORE
+the error gets translated to `ProviderRateLimitError`, otherwise retry
+logic is dead code. Other errors (400, 401, timeout) fail immediately.
+
+## Why in-memory API rate limiting
+
+A public-facing API needs abuse protection. We use a simple in-memory
+sliding window limiter: 10 requests/minute per IP. Sufficient for a
+demo deployment; a production system would use Redis.
+
+Known limitation: the per-IP dict grows without bound across distinct
+IPs. Acceptable for Fly.io with auto-stop (memory resets). If running
+continuously under bot traffic, add a periodic sweep or switch to a
+TTL-based structure.
+
+Design choices:
+- `/health` and `/metrics` exempt: monitoring should never be rate-limited.
+- `Retry-After` header: follows HTTP 429 spec, lets clients back off.
+
 ## Why ranked_sources separate from deduplicated sources?
 
 The deduplicated `sources` list in `AgentResponse` is for the API

agent_bench/core/config.py

@@ -53,6 +53,12 @@ class RAGConfig(BaseModel):
     refusal_threshold: float = 0.0  # 0.0 = disabled (V1 behavior)
 
 
+class RetryConfig(BaseModel):
+    max_retries: int = 3
+    base_delay: float = 1.0  # seconds
+    max_delay: float = 8.0  # cap for exponential backoff
+
+
 class EmbeddingConfig(BaseModel):
     model: str = "all-MiniLM-L6-v2"
     cache_dir: str = ".cache/embeddings"
@@ -62,6 +68,7 @@ class ServingConfig(BaseModel):
     host: str = "0.0.0.0"
     port: int = 8000
     request_timeout_seconds: int = 30
+    rate_limit_rpm: int = 10  # requests per minute per IP
 
 
 class EvaluationConfig(BaseModel):
@@ -73,6 +80,7 @@ class AppConfig(BaseModel):
     agent: AgentConfig = AgentConfig()
     provider: ProviderConfig = ProviderConfig()
     rag: RAGConfig = RAGConfig()
+    retry: RetryConfig = RetryConfig()
     embedding: EmbeddingConfig = EmbeddingConfig()
     serving: ServingConfig = ServingConfig()
     evaluation: EvaluationConfig = EvaluationConfig()

agent_bench/core/provider.py

@@ -2,10 +2,13 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
 import time
 from abc import ABC, abstractmethod
 
+import structlog
+
 from agent_bench.core.config import AppConfig, load_config
 from agent_bench.core.types import (
     CompletionResponse,
@@ -16,6 +19,8 @@ from agent_bench.core.types import (
     ToolDefinition,
 )
 
+log = structlog.get_logger()
+
 
 class ProviderTimeoutError(Exception):
     """Raised when the LLM provider times out."""
@@ -173,7 +178,7 @@ class OpenAIProvider(LLMProvider):
         temperature: float = 0.0,
         max_tokens: int = 1024,
     ) -> CompletionResponse:
-        from openai import APITimeoutError
+        from openai import APITimeoutError, RateLimitError
 
         formatted_messages = format_messages_openai(messages)
         kwargs: dict = {
@@ -186,15 +191,30 @@ class OpenAIProvider(LLMProvider):
             kwargs["tools"] = self.format_tools(tools)
             kwargs["tool_choice"] = "auto"
 
+        retry_cfg = self.config.retry
         start = time.perf_counter()
-        try:
-            response = await self.client.chat.completions.create(**kwargs)
-        except APITimeoutError as e:
-            raise ProviderTimeoutError(f"OpenAI timed out: {e}") from e
-        except Exception as e:
-            if "insufficient_quota" in str(e) or "rate_limit" in str(e).lower():
-                raise ProviderRateLimitError(f"OpenAI rate limit / quota: {e}") from e
-            raise
+
+        for attempt in range(retry_cfg.max_retries + 1):
+            try:
+                response = await self.client.chat.completions.create(**kwargs)
+                break  # success
+            except RateLimitError as e:
+                if attempt == retry_cfg.max_retries:
+                    log.error("provider_rate_limited",
+                              attempts=attempt + 1, error=str(e))
+                    raise ProviderRateLimitError(
+                        f"Rate limited after {retry_cfg.max_retries} retries: {e}"
+                    ) from e
+                wait = min(
+                    retry_cfg.base_delay * (2 ** attempt),
+                    retry_cfg.max_delay,
+                )
+                log.warning("provider_retry",
+                            attempt=attempt + 1, wait_seconds=wait, error=str(e))
+                await asyncio.sleep(wait)
+            except APITimeoutError as e:
+                raise ProviderTimeoutError(f"OpenAI timed out: {e}") from e
+
         latency_ms = (time.perf_counter() - start) * 1000
 
         choice = response.choices[0]

agent_bench/serving/app.py

@@ -13,7 +13,7 @@ from agent_bench.core.provider import create_provider
 from agent_bench.rag.embedder import Embedder
 from agent_bench.rag.retriever import Retriever
 from agent_bench.rag.store import HybridStore
-from agent_bench.serving.middleware import MetricsCollector, RequestMiddleware
+from agent_bench.serving.middleware import MetricsCollector, RateLimitMiddleware, RequestMiddleware
 from agent_bench.serving.routes import router
 from agent_bench.tools.calculator import CalculatorTool
 from agent_bench.tools.registry import ToolRegistry
@@ -99,8 +99,9 @@ def create_app(config: AppConfig | None = None) -> FastAPI:
     app.state.start_time = time.time()
     app.state.metrics = metrics
 
-    # Middleware and routes
+    # Middleware and routes (order matters: rate limit checked first)
     app.add_middleware(RequestMiddleware)
+    app.add_middleware(RateLimitMiddleware, requests_per_minute=config.serving.rate_limit_rpm)
     app.include_router(router)
 
     return app

agent_bench/serving/middleware.py

@@ -1,10 +1,10 @@
-"""Request middleware: ID generation, logging, error handling, metrics."""
+"""Request middleware: ID generation, logging, error handling, metrics, rate limiting."""
 
 from __future__ import annotations
 
 import time
 import uuid
-from collections import deque
+from collections import defaultdict, deque
 
 import structlog
 from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
@@ -47,6 +47,44 @@ class MetricsCollector:
         return self.total_cost_usd / self.requests_total
 
 
+class RateLimitMiddleware(BaseHTTPMiddleware):
+    """In-memory sliding window rate limiter, per client IP."""
+
+    EXEMPT_PATHS = {"/health", "/metrics"}
+
+    def __init__(self, app: object, requests_per_minute: int = 10) -> None:
+        super().__init__(app)  # type: ignore[arg-type]
+        self.rpm = requests_per_minute
+        self.windows: dict[str, list[float]] = defaultdict(list)
+
+    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
+        if request.url.path in self.EXEMPT_PATHS:
+            return await call_next(request)
+
+        client_ip = request.client.host if request.client else "unknown"
+        now = time.time()
+        window_start = now - 60
+
+        # Prune timestamps outside the window
+        self.windows[client_ip] = [
+            t for t in self.windows[client_ip] if t > window_start
+        ]
+
+        if len(self.windows[client_ip]) >= self.rpm:
+            retry_after = max(1, int(60 - (now - self.windows[client_ip][0])))
+            logger.warning("rate_limited",
+                           client_ip=client_ip,
+                           requests_in_window=len(self.windows[client_ip]))
+            return JSONResponse(
+                status_code=429,
+                content={"error": "Rate limit exceeded", "retry_after": retry_after},
+                headers={"Retry-After": str(retry_after)},
+            )
+
+        self.windows[client_ip].append(now)
+        return await call_next(request)
+
+
 class RequestMiddleware(BaseHTTPMiddleware):
     """Adds request ID, timing, structured logging, and error handling."""
 

tests/test_provider.py

@@ -1,11 +1,20 @@
 """Tests for core types, config, and provider abstraction."""
 
+from unittest.mock import patch
+
 import pytest
 
-from agent_bench.core.config import AppConfig, ProviderConfig, load_config, load_task_config
+from agent_bench.core.config import (
+    AppConfig,
+    ProviderConfig,
+    RetryConfig,
+    load_config,
+    load_task_config,
+)
 from agent_bench.core.provider import (
     AnthropicProvider,
     MockProvider,
+    ProviderRateLimitError,
     create_provider,
     format_messages_openai,
     format_tools_openai,
@@ -395,3 +404,168 @@ class TestProviderFactory:
         config = AppConfig(provider=ProviderConfig(default="unknown"))
         with pytest.raises(ValueError, match="Unknown provider"):
             create_provider(config)
+
+
+# --- Retry logic ---
+
+
+class TestProviderRetry:
+    """Tests for OpenAI provider retry with exponential backoff."""
+
+    MOCK_SUCCESS_RESPONSE = {
+        "id": "chatcmpl-retry",
+        "object": "chat.completion",
+        "created": 1234567890,
+        "model": "gpt-4o-mini",
+        "choices": [
+            {
+                "index": 0,
+                "message": {
+                    "role": "assistant",
+                    "content": "Success after retry.",
+                    "tool_calls": None,
+                },
+                "finish_reason": "stop",
+            }
+        ],
+        "usage": {"prompt_tokens": 50, "completion_tokens": 10, "total_tokens": 60},
+    }
+
+    @pytest.mark.asyncio
+    async def test_retry_on_rate_limit(self, monkeypatch):
+        """Two failures then success — returns answer."""
+        monkeypatch.setenv("OPENAI_API_KEY", "test-key-fake")
+
+        import httpx
+        import respx
+
+        from agent_bench.core.provider import OpenAIProvider
+
+        config = AppConfig(
+            provider=ProviderConfig(default="openai"),
+            retry=RetryConfig(max_retries=3, base_delay=0.01, max_delay=0.1),
+        )
+        provider = OpenAIProvider(config)
+
+        call_count = 0
+
+        def side_effect(request):
+            nonlocal call_count
+            call_count += 1
+            if call_count <= 2:
+                return httpx.Response(429, json={"error": {"message": "Rate limit exceeded"}})
+            return httpx.Response(200, json=self.MOCK_SUCCESS_RESPONSE)
+
+        with respx.mock:
+            respx.post("https://api.openai.com/v1/chat/completions").mock(
+                side_effect=side_effect
+            )
+            from agent_bench.core.types import Message, Role
+
+            response = await provider.complete(
+                [Message(role=Role.USER, content="test")]
+            )
+
+        assert response.content == "Success after retry."
+        assert call_count == 3
+
+    @pytest.mark.asyncio
+    async def test_retry_exhausted(self, monkeypatch):
+        """All retries fail — raises ProviderRateLimitError."""
+        monkeypatch.setenv("OPENAI_API_KEY", "test-key-fake")
+
+        import httpx
+        import respx
+
+        from agent_bench.core.provider import OpenAIProvider
+
+        config = AppConfig(
+            provider=ProviderConfig(default="openai"),
+            retry=RetryConfig(max_retries=2, base_delay=0.01, max_delay=0.1),
+        )
+        provider = OpenAIProvider(config)
+
+        with respx.mock:
+            respx.post("https://api.openai.com/v1/chat/completions").mock(
+                return_value=httpx.Response(429, json={"error": {"message": "Rate limit"}})
+            )
+            from agent_bench.core.types import Message, Role
+
+            with pytest.raises(ProviderRateLimitError, match="Rate limited after"):
+                await provider.complete(
+                    [Message(role=Role.USER, content="test")]
+                )
+
+    @pytest.mark.asyncio
+    async def test_no_retry_on_other_errors(self, monkeypatch):
+        """Non-rate-limit errors fail immediately without retry."""
+        monkeypatch.setenv("OPENAI_API_KEY", "test-key-fake")
+
+        import httpx
+        import respx
+
+        from agent_bench.core.provider import OpenAIProvider
+
+        config = AppConfig(
+            provider=ProviderConfig(default="openai"),
+            retry=RetryConfig(max_retries=3, base_delay=0.01, max_delay=0.1),
+        )
+        provider = OpenAIProvider(config)
+
+        call_count = 0
+
+        def side_effect(request):
+            nonlocal call_count
+            call_count += 1
+            return httpx.Response(400, json={"error": {"message": "Bad request"}})
+
+        with respx.mock:
+            respx.post("https://api.openai.com/v1/chat/completions").mock(
+                side_effect=side_effect
+            )
+            from agent_bench.core.types import Message, Role
+
+            with pytest.raises(Exception):
+                await provider.complete(
+                    [Message(role=Role.USER, content="test")]
+                )
+
+        assert call_count == 1  # no retry
+
+    @pytest.mark.asyncio
+    async def test_retry_backoff_timing(self, monkeypatch):
+        """Verify exponential backoff delays between retries."""
+        monkeypatch.setenv("OPENAI_API_KEY", "test-key-fake")
+
+        import httpx
+        import respx
+
+        from agent_bench.core.provider import OpenAIProvider
+
+        config = AppConfig(
+            provider=ProviderConfig(default="openai"),
+            retry=RetryConfig(max_retries=3, base_delay=1.0, max_delay=8.0),
+        )
+        provider = OpenAIProvider(config)
+
+        sleep_calls: list[float] = []
+
+        async def mock_sleep(seconds):
+            sleep_calls.append(seconds)
+
+        with respx.mock, patch("asyncio.sleep", side_effect=mock_sleep):
+            respx.post("https://api.openai.com/v1/chat/completions").mock(
+                return_value=httpx.Response(429, json={"error": {"message": "Rate limit"}})
+            )
+            from agent_bench.core.types import Message, Role
+
+            with pytest.raises(ProviderRateLimitError):
+                await provider.complete(
+                    [Message(role=Role.USER, content="test")]
+                )
+
+        # 3 retries: delays should be 1.0, 2.0, 4.0
+        assert len(sleep_calls) == 3
+        assert sleep_calls[0] == pytest.approx(1.0)
+        assert sleep_calls[1] == pytest.approx(2.0)
+        assert sleep_calls[2] == pytest.approx(4.0)

tests/test_serving.py

@@ -11,7 +11,7 @@ from agent_bench.agents.orchestrator import Orchestrator
 from agent_bench.core.config import AppConfig, ProviderConfig
 from agent_bench.core.provider import MockProvider, ProviderTimeoutError
 from agent_bench.rag.store import HybridStore
-from agent_bench.serving.middleware import MetricsCollector, RequestMiddleware
+from agent_bench.serving.middleware import MetricsCollector, RateLimitMiddleware, RequestMiddleware
 from agent_bench.tools.calculator import CalculatorTool
 from agent_bench.tools.registry import ToolRegistry
 
@@ -174,3 +174,96 @@ class TestMiddleware:
         data = response.json()
         assert "request_id" in data
         assert "x-request-id" in response.headers
+
+
+# --- Rate limiting tests ---
+
+
+def _make_rate_limited_app(rpm: int = 3):
+    """Create a test app with rate limiting enabled."""
+    from fastapi import FastAPI
+
+    app = FastAPI(title="agent-bench-ratelimit")
+
+    registry = ToolRegistry()
+    registry.register(FakeSearchTool())
+    registry.register(CalculatorTool())
+
+    provider = MockProvider()
+    orchestrator = Orchestrator(provider=provider, registry=registry, max_iterations=3)
+
+    app.state.orchestrator = orchestrator
+    app.state.store = HybridStore(dimension=384)
+    app.state.config = AppConfig(provider=ProviderConfig(default="mock"))
+    app.state.system_prompt = "You are a test assistant."
+    app.state.start_time = time.time()
+    app.state.metrics = MetricsCollector()
+
+    app.add_middleware(RequestMiddleware)
+    app.add_middleware(RateLimitMiddleware, requests_per_minute=rpm)
+
+    from agent_bench.serving.routes import router
+
+    app.include_router(router)
+    return app
+
+
+@pytest.fixture
+def rate_limited_app():
+    return _make_rate_limited_app(rpm=3)
+
+
+class TestRateLimiting:
+    @pytest.mark.asyncio
+    async def test_allows_normal_traffic(self, rate_limited_app):
+        """Requests within the limit all succeed."""
+        async with AsyncClient(
+            transport=ASGITransport(app=rate_limited_app), base_url="http://test"
+        ) as client:
+            for _ in range(3):
+                response = await client.get("/health")
+                assert response.status_code == 200
+
+    @pytest.mark.asyncio
+    async def test_blocks_excess(self, rate_limited_app):
+        """Request beyond the limit gets 429."""
+        async with AsyncClient(
+            transport=ASGITransport(app=rate_limited_app), base_url="http://test"
+        ) as client:
+            # Use up the quota
+            for _ in range(3):
+                await client.post("/ask", json={"question": "test"})
+            # Next request should be blocked
+            response = await client.post("/ask", json={"question": "test"})
+            assert response.status_code == 429
+
+    @pytest.mark.asyncio
+    async def test_retry_after_header(self, rate_limited_app):
+        """429 response includes Retry-After header."""
+        async with AsyncClient(
+            transport=ASGITransport(app=rate_limited_app), base_url="http://test"
+        ) as client:
+            # Exhaust quota on non-exempt path
+            for _ in range(3):
+                await client.post("/ask", json={"question": "test"})
+            response = await client.post("/ask", json={"question": "test"})
+            assert response.status_code == 429
+            assert "retry-after" in response.headers
+            assert int(response.headers["retry-after"]) > 0
+
+    @pytest.mark.asyncio
+    async def test_health_exempt(self):
+        """Health endpoint is never rate limited."""
+        app = _make_rate_limited_app(rpm=2)
+        async with AsyncClient(
+            transport=ASGITransport(app=app), base_url="http://test"
+        ) as client:
+            # Exhaust quota on non-exempt path
+            for _ in range(2):
+                await client.post("/ask", json={"question": "test"})
+            # Health should still work
+            response = await client.get("/health")
+            assert response.status_code == 200
+            # But another ask should be blocked
+            response = await client.post("/ask", json={"question": "test"})
+            assert response.status_code == 429