fix: wire ADR-0012 per-user rate limit on /ask (#261)

services/retriever/.env.example

@@ -59,6 +59,10 @@ GCP_PROJECT_ID=
 # OpenTelemetry — leave empty for local dev (no collector running by default).
 OTEL_EXPORTER_OTLP_ENDPOINT=
 
+# Rate limiting (ADR-0012: slowapi, per-user limit on /api/v1/ask).
+# RATE_LIMIT_ENABLED=true
+# RATE_LIMIT_ASK=10/minute
+
 # App
 DEBUG=true
 ALLOWED_ORIGINS=["http://localhost:5173"]

services/retriever/pyproject.toml

@@ -33,6 +33,7 @@ dependencies = [
     "python-multipart>=0.0.18",
     "docling>=2.94.0",
     "docling-core[chunking-openai]>=2.0",
+    "slowapi>=0.1.9",
     # Direct floor to force the transitive transformers pin above the RCE in
     # CVE-2026-4372 (Dependabot alert #95). transformers arrives via
     # docling-core[chunking-openai] and docling-ibm-models. See issue #188.
@@ -114,6 +115,11 @@ module = ["docling.*", "docling_core.*"]
 ignore_missing_imports = true
 follow_imports = "skip"
 
+[[tool.mypy.overrides]]
+module = ["slowapi", "slowapi.*"]
+ignore_missing_imports = true
+follow_imports = "skip"
+
 # boto3/botocore are an optional, lazily-imported dependency for R2 storage and
 # are not installed; their imports live inside build_r2_storage.
 [[tool.mypy.overrides]]

services/retriever/src/retriever/config.py

@@ -134,6 +134,10 @@ class Settings(BaseSettings):
     #                   only for gateways that actually implement it.
     moderation_backend: Literal["guardrails", "openai_api"] = "guardrails"
 
+    # Rate limiting (ADR-0012)
+    rate_limit_enabled: bool = True
+    rate_limit_ask: str = "10/minute"
+
     # RAG
     rag_top_k: int = 5
 

services/retriever/src/retriever/infrastructure/rate_limit.py

@@ -0,0 +1,122 @@
+# Copyright (C) 2025 Backchain LLC
+# SPDX-License-Identifier: Apache-2.0
+
+"""Per-user request rate limiting for the RAG ask endpoint (ADR-0012).
+
+Guards ``POST /api/v1/ask`` against denial-of-wallet: a single authenticated,
+subscribed token could otherwise drive unbounded paid LLM inference. The
+limiter keys on the caller's Supabase user id (``AuthUser.sub``), stashed onto
+``request.state.rate_limit_key`` by a route dependency, so the limit tracks
+one bucket per user rather than one bucket per process or per remote address.
+
+Per-tenant limiting is a documented forward extension only, not implemented
+here: the JWT this service validates carries no tenant claim, so keying on
+the constant ``DEFAULT_TENANT_ID`` would collapse every user in the system
+into a single global bucket rather than isolating tenants from each other.
+
+Storage is in-memory (``memory://``, slowapi's default) -- per ADR-0012 this
+resets on restart and limits are per-instance, which is acceptable for a
+single-instance deployment.
+"""
+
+from __future__ import annotations
+
+import time
+from collections.abc import Callable
+from typing import cast
+
+from fastapi import FastAPI, Request
+from limits import parse_many
+from slowapi import Limiter
+from slowapi.errors import RateLimitExceeded
+from starlette.responses import JSONResponse
+
+from retriever.config import get_settings
+
+# Fallback Retry-After (seconds) when slowapi hasn't recorded window state on
+# the request (defensive; should not happen once configure_rate_limiting has
+# wired the limiter onto the app).
+_DEFAULT_RETRY_AFTER_SECONDS = 60
+
+RATE_LIMIT_MESSAGE = (
+    "Too many requests. Please wait a moment before asking another question."
+)
+
+
+def rate_limit_key(request: Request) -> str:
+    """Resolve the slowapi bucket key for the current request.
+
+    Prefers the per-user identity stashed by
+    :func:`retriever.modules.rag.routes._stash_rate_limit_identity` (the
+    authenticated user's ``sub``); falls back to the remote address for any
+    request that reaches the limiter without that identity attached.
+    """
+    stashed = getattr(request.state, "rate_limit_key", None)
+    if isinstance(stashed, str) and stashed:
+        return stashed
+    return request.client.host if request.client else "127.0.0.1"
+
+
+limiter = Limiter(key_func=rate_limit_key, headers_enabled=False)
+
+
+def _ask_limit_value() -> str:
+    """Return the current ``/ask`` limit string, re-read from settings."""
+    return get_settings().rate_limit_ask
+
+
+def _rate_limiting_disabled() -> bool:
+    """Return True when rate limiting is disabled via settings."""
+    return not get_settings().rate_limit_enabled
+
+
+# slowapi re-evaluates both callables on every request, so toggling
+# RATE_LIMIT_ENABLED or RATE_LIMIT_ASK at runtime (e.g. via env in tests)
+# takes effect without re-registering the decorator.
+_ask_limit_decorator = limiter.limit(
+    _ask_limit_value, exempt_when=_rate_limiting_disabled
+)
+
+
+def ask_rate_limit[F: Callable[..., object]](func: F) -> F:
+    """Typed shim around ``limiter.limit`` for the ``/ask`` endpoint.
+
+    slowapi's ``Limiter.limit`` is untyped, which trips mypy --strict's
+    ``disallow_untyped_decorators``. Casting the wrapped callable back to
+    ``F`` keeps the decorated route's signature visible to type checkers.
+    """
+    return cast("F", _ask_limit_decorator(func))
+
+
+def _retry_after_seconds(request: Request) -> int:
+    """Compute a Retry-After value (seconds) from slowapi's window state."""
+    current = getattr(request.state, "view_rate_limit", None)
+    if current is None:
+        return _DEFAULT_RETRY_AFTER_SECONDS
+    window_stats = limiter.limiter.get_window_stats(current[0], *current[1])
+    return max(1, int(1 + window_stats[0] - time.time()))
+
+
+async def rate_limit_exceeded_handler(request: Request, exc: Exception) -> JSONResponse:
+    """Render the ADR-0012 429 body for a :class:`RateLimitExceeded` error."""
+    retry_after = _retry_after_seconds(request)
+    return JSONResponse(
+        status_code=429,
+        content={
+            "error": "rate_limit_exceeded",
+            "message": RATE_LIMIT_MESSAGE,
+            "retry_after": retry_after,
+        },
+        headers={"Retry-After": str(retry_after)},
+    )
+
+
+def configure_rate_limiting(app: FastAPI) -> None:
+    """Wire the ADR-0012 limiter and its 429 handler onto ``app``.
+
+    Parses the configured ``/ask`` limit up front so an invalid limit string
+    fails fast at startup rather than on the first request.
+    """
+    parse_many(get_settings().rate_limit_ask)
+    app.state.limiter = limiter
+    app.add_exception_handler(RateLimitExceeded, rate_limit_exceeded_handler)

services/retriever/src/retriever/main.py

@@ -27,6 +27,7 @@
     RequestIdMiddleware,
 )
 from retriever.infrastructure.observability.tracing import configure_tracing
+from retriever.infrastructure.rate_limit import configure_rate_limiting
 from retriever.modules.auth import require_subscription
 from retriever.modules.documents.routes import router as documents_router
 from retriever.modules.messages.routes import router as messages_router
@@ -174,6 +175,11 @@ def create_app() -> FastAPI:
         allow_headers=["Authorization", "Content-Type", "X-Request-ID"],
     )
 
+    # ADR-0012 rate limiting: registered after CORS so a 429 raised inside the
+    # router stack still exits back out through the CORS middleware above it
+    # and carries Access-Control-Allow-Origin.
+    configure_rate_limiting(app)
+
     # /api/v1 routes require an active "retriever" subscription (claim-based,
     # no DB read — see docs/subscriptions.md). Health has no /api/v1 prefix
     # and stays ungated so orchestrators can probe liveness without a token.

services/retriever/src/retriever/modules/rag/routes.py

@@ -9,10 +9,11 @@
 from typing import Annotated
 
 import structlog
-from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi import APIRouter, Depends, HTTPException, Request, status
 from pydantic import BaseModel, ConfigDict, Field
 
 from retriever.config import get_settings
+from retriever.infrastructure.rate_limit import ask_rate_limit
 from retriever.models.user import DEFAULT_TENANT_ID
 from retriever.modules.auth import AuthUser, require_auth
 from retriever.modules.messages.repos import MessageRepository
@@ -65,10 +66,28 @@ def _to_ask_response(rag_response: RAGResponse) -> AskResponse:
     )
 
 
+async def _stash_rate_limit_identity(
+    request: Request,
+    user: Annotated[AuthUser, Depends(require_auth)],
+) -> AuthUser:
+    """Expose the authenticated user's id to slowapi's rate-limit key_func.
+
+    Chains off ``require_auth`` (FastAPI caches the dependency per request,
+    so this costs no extra JWT decode) and stashes ``user.sub`` onto
+    ``request.state`` where :func:`retriever.infrastructure.rate_limit.rate_limit_key`
+    reads it. This keys the ADR-0012 limit per authenticated user rather than
+    per remote address.
+    """
+    request.state.rate_limit_key = user.sub
+    return user
+
+
 @router.post("/ask", response_model=AskResponse)
+@ask_rate_limit
 async def ask(
+    request: Request,
     body: AskRequest,
-    user: Annotated[AuthUser, Depends(require_auth)],
+    user: Annotated[AuthUser, Depends(_stash_rate_limit_identity)],
     rag_service: Annotated[RAGService, Depends(get_rag_service)],
     message_repo: Annotated[MessageRepository, Depends(get_message_repository)],
 ) -> AskResponse:

services/retriever/tests/conftest.py

@@ -20,6 +20,7 @@
     create_async_engine,
 )
 
+from retriever.infrastructure.rate_limit import limiter
 from retriever.models.base import Base
 
 TEST_DATABASE_URL = os.getenv(
@@ -34,6 +35,18 @@ def pytest_configure(config: pytest.Config) -> None:
     )
 
 
+@pytest.fixture(autouse=True)
+def _reset_rate_limiter() -> None:
+    """Reset slowapi's in-memory rate-limit windows between tests.
+
+    The limiter is a module-level singleton (see
+    retriever.infrastructure.rate_limit), so without this reset, request
+    counts from one test's /ask calls would bleed into the next test's
+    window and produce flaky 429s (e.g. in test_rag_routes.py).
+    """
+    limiter.reset()
+
+
 @pytest_asyncio.fixture
 async def db_engine() -> AsyncEngine:  # type: ignore[return]
     """Create all tables for one test; drop them after the test completes."""

services/retriever/tests/integration/test_rate_limit_and_concurrency.py

@@ -1,11 +1,11 @@
 """Concurrent-request handling and burst behavior for the retriever API.
 
-There is no app-level rate limiter in this service today (no slowapi
-middleware, no 429 responses anywhere in the routes). The burst test below
-therefore asserts service stability under load (no 5xx) rather than a
-specific 429 response, and is forward-compatible: the allowed-status set
-already includes 429 so this test keeps passing unmodified if a limiter is
-added later.
+``POST /api/v1/ask`` now carries the ADR-0012 per-user rate limit (slowapi,
+in-memory, keyed on the authenticated user's Supabase id). The burst test
+below targets ``GET /api/v1/documents``, which is unaffected by that limit,
+so it asserts service stability under load (no 5xx) rather than a specific
+429 response; its allowed-status set already includes 429, so it keeps
+passing unmodified if a limiter is ever extended to this endpoint too.
 """
 
 from __future__ import annotations

services/retriever/tests/test_config.py

@@ -164,3 +164,10 @@ def test_gateway_token_for_moderation_scope() -> None:
         settings.gateway_token_for("moderation").get_secret_value()
         == "moderation-token"
     )
+
+
+def test_rate_limit_defaults() -> None:
+    """Rate limiting is enabled by default with a 10/minute limit on /ask."""
+    settings = Settings(_env_file=None)  # type: ignore[call-arg]
+    assert settings.rate_limit_enabled is True
+    assert settings.rate_limit_ask == "10/minute"