feat: restrict model classes, quotas, and privacy per tenant

README.md

@@ -143,6 +143,40 @@ be served. A request can never introduce a model path, revision, or adapter.
 Send `routing.domain` to request a domain adapter; the router applies the promoted adapter
 with the largest measured quality gain for that base revision and task, or none at all.
 
+## Tenants
+
+What a caller may use is governance and lives in the catalog; the credential that proves which
+tenant is calling stays in the environment and is never committed.
+
+```bash
+ROUTER_TENANT_KEYS="support-tooling:<key>,clinical-research:<key>"
+```
+
+Keys listed bare in `ROUTER_API_KEYS` belong to the `default` tenant, so an existing deployment
+keeps working unchanged. Each tenant in [`config/registry.yaml`](config/registry.yaml) may set:
+
+| Field | Effect |
+|---|---|
+| `permitted_tiers` / `permitted_models` | Hard filter applied before scoring; it can never be outscored. Empty means unrestricted. |
+| `quota_requests_per_minute` | The tenant's own limit; absent defers to the platform default. |
+| `minimum_privacy` | A floor, never a ceiling. The request is raised to it before anything reads the class. |
+| `allow_external_fallback` | `false` denies external routing outright; absent defers to platform policy. |
+| `quality_floor` | Raises, and never lowers, the floor a request asked for. |
+
+An absent field never tightens an existing deployment: it defers to platform policy rather than
+implying a restriction.
+
+Quota and cache are scoped to the tenant rather than the credential, so rotating a key neither
+resets a quota nor orphans a cache, and two keys for one tenant draw on one quota.
+
+The privacy floor is applied before the cache lookup, not at routing. A tenant handling regulated
+data cannot declare its traffic `public` and so become eligible for external routing or semantic
+reuse, and it can never read an entry another tenant stored under `public`. When the class is
+raised the route reason says so — the change is attributed, not silent.
+
+A request that no entitled model can serve is refused with `422` rather than downgraded to a
+model never validated for the task.
+
 ## Observability
 
 `GET /metrics` returns Prometheus exposition text and is intentionally unauthenticated so

config/registry.yaml

@@ -179,3 +179,33 @@ deployments:
     model_checksums:
       small-specialist: sha256:mock-small-previous
     stage: deprecated
+
+# Per-tenant entitlement (sections 7.1 and 14). Entitlements are governance and
+# belong here; the credentials that bind a caller to a tenant are supplied by
+# ROUTER_TENANT_KEYS from the secret manager and are never committed.
+tenants:
+  - id: default
+    description: Platform default for callers with no dedicated entitlement.
+
+  - id: support-tooling
+    description: Ticket triage and field extraction only; no reasoning tier.
+    permitted_tiers: [small-specialist, general-local]
+    quota_requests_per_minute: 240
+    minimum_privacy: private
+
+  - id: clinical-research
+    description: Regulated data; never leaves the private inference environment.
+    permitted_tiers: [small-specialist, general-local, high-capability]
+    quota_requests_per_minute: 60
+    # Raised to restricted no matter what the caller declares, so this tenant
+    # can never become eligible for external routing or semantic reuse.
+    minimum_privacy: restricted
+    quality_floor: 0.85
+    # Stated explicitly rather than relying on the restricted floor alone.
+    allow_external_fallback: false
+
+  - id: public-demo
+    description: Public demo traffic; may use the approved external fallback.
+    permitted_models: [small-specialist]
+    quota_requests_per_minute: 30
+    allow_external_fallback: true

src/llm_router/admission.py

@@ -36,13 +36,18 @@ def __init__(self, requests_per_minute: int) -> None:
         self._events: dict[str, deque[float]] = defaultdict(deque)
         self._lock = asyncio.Lock()
 
-    async def consume(self, subject: str, *, now: float | None = None) -> None:
+    async def consume(
+        self, subject: str, *, now: float | None = None, limit: int | None = None
+    ) -> None:
+        """Consume one request, against the subject's own limit when it has one."""
+
         timestamp = time.monotonic() if now is None else now
         cutoff = timestamp - 60
+        effective_limit = self._limit if limit is None else limit
         async with self._lock:
             events = self._events[subject]
             while events and events[0] <= cutoff:
                 events.popleft()
-            if len(events) >= self._limit:
+            if len(events) >= effective_limit:
                 raise QuotaExceededError("request quota exceeded")
             events.append(timestamp)

src/llm_router/app.py

@@ -5,7 +5,9 @@
 import uuid
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
+from dataclasses import dataclass
 from pathlib import Path
+from typing import Annotated
 
 import httpx
 from fastapi import Depends, FastAPI, Header, HTTPException, Request, Response, status
@@ -49,10 +51,29 @@
 )
 from llm_router.observability import Metrics
 from llm_router.redis_state import RedisCacheStore, RedisFixedWindowQuota, RedisLike
-from llm_router.registry import Registry, RegistryError, catalog_revisions, load_registry
+from llm_router.registry import (
+    Registry,
+    RegistryError,
+    TenantRecord,
+    catalog_revisions,
+    load_registry,
+    strictest_privacy,
+)
 from llm_router.routing import NoEligibleModelError, Router, default_model_profiles
 
 
+@dataclass(frozen=True)
+class Principal:
+    """The authenticated caller: which tenant, and which credential proved it.
+
+    Quota and cache scope use the tenant rather than the credential, so
+    rotating a key neither resets a tenant's quota nor orphans its cache.
+    """
+
+    tenant_id: str
+    credential_fingerprint: str
+
+
 def _redis_client(settings: Settings) -> RedisLike | None:
     """Build a shared-state client when a Redis URL is configured."""
 
@@ -162,7 +183,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         lifespan=lifespan,
     )
 
-    async def authenticate(authorization: str | None = Header(default=None)) -> str:
+    async def authenticate(authorization: str | None = Header(default=None)) -> Principal:
         prefix = "Bearer "
         if authorization is None or not authorization.startswith(prefix):
             raise HTTPException(
@@ -171,16 +192,25 @@ async def authenticate(authorization: str | None = Header(default=None)) -> str:
                 headers={"WWW-Authenticate": "Bearer"},
             )
         token = authorization.removeprefix(prefix)
-        if not any(
-            secrets.compare_digest(token, candidate)
-            for candidate in runtime_settings.accepted_api_keys
-        ):
+        # Every candidate is compared so the work does not depend on which
+        # credential matched, and the match itself stays constant time.
+        matched: str | None = None
+        for candidate, tenant_id in runtime_settings.tenant_by_key.items():
+            if secrets.compare_digest(token, candidate):
+                matched = tenant_id
+        if matched is None:
             raise HTTPException(
                 status_code=status.HTTP_401_UNAUTHORIZED,
                 detail="invalid bearer token",
                 headers={"WWW-Authenticate": "Bearer"},
             )
-        return hashlib.sha256(token.encode()).hexdigest()
+        return Principal(
+            tenant_id=matched,
+            credential_fingerprint=hashlib.sha256(token.encode()).hexdigest(),
+        )
+
+    def _entitlement(principal: Principal) -> TenantRecord | None:
+        return catalog.tenant(principal.tenant_id) if catalog is not None else None
 
     @app.exception_handler(NoEligibleModelError)
     async def no_model_handler(_: Request, error: NoEligibleModelError) -> JSONResponse:
@@ -409,12 +439,15 @@ async def _store_cache(
             scope = semantic_cache.scope(payload, tenant=tenant, model_revision=catalog_version)
             semantic_cache.store(scope, prompt, entry)
 
-    async def _consume_quota(subject: str) -> None:
+    async def _consume_quota(subject: str, tenant: TenantRecord | None) -> None:
+        # A tenant may carry its own limit; absent one the platform default
+        # applies, which is why None means "defer" rather than "unlimited".
+        limit = tenant.quota_requests_per_minute if tenant is not None else None
         if shared_quota is None:
-            await quota.consume(subject)
+            await quota.consume(subject, limit=limit)
             return
         window = int(time.time() // 60)
-        if not await shared_quota.consume(subject, window=window):
+        if not await shared_quota.consume(subject, window=window, limit=limit):
             raise QuotaExceededError("request quota exceeded")
 
     def _route_headers(decision: RouteDecision, cache_state: str) -> dict[str, str]:
@@ -486,10 +519,31 @@ async def _stream_completion(
     async def chat_completions(
         payload: ChatCompletionRequest,
         response: Response,
-        subject: str = Depends(authenticate),
+        principal: Annotated[Principal, Depends(authenticate)],
     ) -> ChatCompletionResponse | StreamingResponse:
         started = time.perf_counter()
-        await _consume_quota(subject)
+        tenant = _entitlement(principal)
+        # Quota and cache are scoped to the tenant, not the credential, so
+        # rotating a key neither resets a quota nor orphans a cache.
+        subject = principal.tenant_id
+        await _consume_quota(subject, tenant)
+
+        declared_privacy = payload.routing.privacy
+        if tenant is not None:
+            effective_privacy = strictest_privacy(declared_privacy, tenant.minimum_privacy)
+            if effective_privacy is not declared_privacy:
+                # Raised before anything reads the class, so cache eligibility,
+                # the cache key, and routing all agree on one value and a
+                # public-declared request can never reach restricted entries.
+                payload = payload.model_copy(
+                    update={
+                        "routing": payload.routing.model_copy(update={"privacy": effective_privacy})
+                    }
+                )
+        privacy_raised_from = (
+            declared_privacy if payload.routing.privacy is not declared_privacy else None
+        )
+
         prompt = payload.prompt
         cache_key = build_cache_key(
             payload, tenant=subject, model_revision=catalog_version, prompt=prompt
@@ -517,7 +571,18 @@ async def chat_completions(
 
         cached_task = decision_cache.get(prompt) if runtime_settings.cache_enabled else None
         telemetry.record_cache_event("router", "hit" if cached_task is not None else "miss")
-        decision = router.select(payload, task=cached_task)
+        decision = router.select(
+            payload,
+            task=cached_task,
+            permitted_models=(
+                catalog.permitted_models_for(tenant) if catalog is not None else None
+            ),
+            quality_floor=tenant.quality_floor if tenant is not None else 0.0,
+            tenant_allows_external=(
+                tenant.allow_external_fallback is not False if tenant is not None else True
+            ),
+            privacy_raised_from=privacy_raised_from,
+        )
         if runtime_settings.cache_enabled:
             decision_cache.set(prompt, decision.task)
         telemetry.record_route(decision, privacy=payload.routing.privacy.value)

src/llm_router/config.py

@@ -12,6 +12,11 @@ class Settings(BaseSettings):
 
     environment: str = "development"
     api_keys: str = "dev-key"
+    # Binds credentials to tenant ids as "tenant:key,tenant:key". Entitlements
+    # are governance and live in the catalog; the secret that proves which
+    # tenant is calling stays in the environment and is never committed.
+    tenant_keys: str = ""
+    default_tenant: str = "default"
     max_concurrency: int = Field(default=32, ge=1)
     admission_timeout_seconds: float = Field(default=0.25, gt=0)
     quota_requests_per_minute: int = Field(default=120, ge=1)
@@ -36,7 +41,24 @@ def reject_development_key_in_shared_environments(self) -> "Settings":
 
     @property
     def accepted_api_keys(self) -> frozenset[str]:
-        return frozenset(key.strip() for key in self.api_keys.split(",") if key.strip())
+        return frozenset(self.tenant_by_key)
+
+    @property
+    def tenant_by_key(self) -> dict[str, str]:
+        """Map every accepted credential to the tenant it authenticates.
+
+        Keys listed without a tenant belong to the default tenant, so an
+        existing ROUTER_API_KEYS deployment keeps working unchanged.
+        """
+
+        bindings = {
+            key.strip(): self.default_tenant for key in self.api_keys.split(",") if key.strip()
+        }
+        for entry in self.tenant_keys.split(","):
+            tenant, separator, key = entry.partition(":")
+            if separator and tenant.strip() and key.strip():
+                bindings[key.strip()] = tenant.strip()
+        return bindings
 
 
 @lru_cache

src/llm_router/redis_state.py

@@ -64,11 +64,11 @@ def __init__(self, client: RedisLike, requests_per_minute: int, *, prefix: str =
         self._limit = requests_per_minute
         self._prefix = prefix
 
-    async def consume(self, subject: str, *, window: int) -> bool:
+    async def consume(self, subject: str, *, window: int, limit: int | None = None) -> bool:
         """Return whether the request fits inside the caller's quota."""
 
         key = f"{self._prefix}{subject}:{window}"
         count = await self._client.incr(key)
         if count == 1:
             await self._client.expire(key, QUOTA_WINDOW_SECONDS * 2)
-        return count <= self._limit
+        return count <= (self._limit if limit is None else limit)