fix(cache): prompt cache served wrong responses — v2 key space + strict cacheability

app/config.py

@@ -52,9 +52,16 @@ class Settings(BaseSettings):
     port: int = 8000
     log_level: str = "info"
 
-    # ── Encryption (auto-generated on first run if empty) ──
+    # ── Encryption ──
+    # NOTE: credential_encryption_key is NOT auto-generated. If empty, the
+    # app seals provider keys with a publicly-known dev key (with a loud
+    # warning), and packages.db.guards refuses to boot once real
+    # credentials are at stake. Generate one: `openssl rand -hex 32`.
     credential_encryption_key: str = ""
     api_key_pepper: str = ""
+    # Explicit opt-out from the startup guard that refuses to run with the
+    # publicly-known dev encryption key when real credentials are at stake.
+    allow_insecure_dev_key: bool = False
 
     # ── Provider keys via env (alternative to UI-stored keys) ──
     # Keep in sync with `_PROVIDERS_FROM_ENV` above. Pydantic-settings reads

app/main.py

@@ -22,7 +22,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
 
     from app.config import get_settings
     from packages.db import session as session_mod
-    from packages.db.engine import dispose_engine, get_engine
+    from packages.db.engine import dispose_engine, get_engine, redacted_url
     from packages.db.models.base import Base
 
     settings = get_settings()
@@ -39,21 +39,37 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
         ),
     )
     log = structlog.get_logger()
-    log.info("lite_starting", database_url=settings.database_url)
+    # Redact: on the documented Postgres path the raw URL carries the DB
+    # password, and structured logs are retained by hosted aggregators.
+    log.info("lite_starting", database_url=redacted_url(settings.database_url))
 
     engine = get_engine(settings.database_url)
 
     async with engine.begin() as conn:
         await conn.run_sync(Base.metadata.create_all)
 
+    # Fail closed before any traffic can be served: refuse to boot when
+    # provider credentials are (or would be) sealed with the publicly-known
+    # dev encryption key. Runs after create_all so a fresh database's empty
+    # provider_keys table counts as "no credentials at risk".
+    from packages.db.guards import assert_credential_encryption_ready
+
+    await assert_credential_encryption_ready(
+        make_session=async_sessionmaker(engine, expire_on_commit=False),
+        database_url=settings.database_url,
+        allow_insecure_dev_key=settings.allow_insecure_dev_key,
+    )
+
     session_mod._session_factory = async_sessionmaker(engine, expire_on_commit=False)
 
     from app.seed import seed_initial_state
 
     async with session_mod._session_factory() as s:
         seed = await seed_initial_state(s)
         if seed.created and seed.api_key:
-            log.info("seed_complete", api_key=seed.api_key)
+            # No key material in the structured event: logs are retained by
+            # aggregators. The print() below is the one-time delivery channel.
+            log.info("seed_complete", workspace_id=seed.workspace_id)
             print(f"\n  ✓ orcarouter-lite ready. API key: {seed.api_key}\n")
 
     from app import cache_invalidation_bus
@@ -68,6 +84,29 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     await dispose_engine()
 
 
+def unhandled_exception_handler(request, exc: Exception):
+    """Last-resort handler: log everything, return an opaque envelope.
+
+    The traceback MUST be recorded here — this is the only place an
+    arbitrary exception surfaces, and without it every production 500 is
+    undebuggable.
+    """
+    structlog.get_logger().exception(
+        "unhandled_exception",
+        path=str(request.url.path),
+        error=str(exc),
+    )
+    return JSONResponse(
+        status_code=500,
+        content={
+            "error": {
+                "message": "Internal server error",
+                "type": "server_error",
+            }
+        },
+    )
+
+
 def create_app() -> FastAPI:
     app = FastAPI(
         title="OrcaRouter Lite",
@@ -106,17 +145,7 @@ async def val_exc_handler(_req, exc: RequestValidationError):
             content={"error": {"message": msg, "type": "validation_error"}},
         )
 
-    @app.exception_handler(Exception)
-    async def unhandled(_req, exc: Exception):
-        return JSONResponse(
-            status_code=500,
-            content={
-                "error": {
-                    "message": "Internal server error",
-                    "type": "server_error",
-                }
-            },
-        )
+    app.add_exception_handler(Exception, unhandled_exception_handler)
 
     from app.middleware.auth import AuthMiddleware
 

app/middleware/auth.py

@@ -10,9 +10,13 @@
 
 import json
 
+import structlog
+
 from packages.auth.key_validator import AuthError, validate_api_key
 from packages.db import session as session_mod
 
+logger = structlog.get_logger()
+
 SKIP_AUTH_PATHS: set[str] = {
     "/health",
     "/health/ready",
@@ -95,6 +99,9 @@ async def __call__(self, scope, receive, send):
             await _send_error(send, e.status_code, e.message, "auth_error")
             return
         except Exception:
+            # A DB failure during key validation must be visible — without
+            # this log the operator only sees unexplained 503s.
+            logger.exception("auth_middleware_error", path=path)
             await _send_error(send, 503, "Service temporarily unavailable", "server_error")
             return
 

app/prompt_cache.py

@@ -33,15 +33,42 @@ def cache_key(
     tools: list[dict] | None,
     response_format: dict | None,
     seed: int | None,
+    max_tokens: int | None = None,
+    top_p: float | None = None,
+    stop=None,
+    n: int | None = None,
+    tool_choice=None,
+    presence_penalty: float | None = None,
+    frequency_penalty: float | None = None,
 ) -> str:
-    """Deterministic SHA-256 key for the cacheable inputs."""
+    """Deterministic SHA-256 key over EVERY output-affecting input.
+
+    v2: the original key covered only six fields, so e.g. a max_tokens=16
+    request collided with an identical max_tokens=4000 request and served
+    the truncated answer as a HIT. The version field keeps old entries
+    from ever matching the new key space.
+    """
     payload = {
+        "v": 2,
         "model": model,
         "messages": messages,
-        "temperature": temperature if temperature is not None else 0.0,
+        # Preserve the raw value, INCLUDING an omitted temperature (None).
+        # Coercing None to 0.0 here made `{seed, temp omitted}` (actually
+        # generated at the provider default of 1.0) and `{seed, temp: 0.0}`
+        # hash to the SAME key — serving a temperature-0 generation to a
+        # temperature-1 client. The seed pins determinism *per parameter
+        # combination*, not across different temperatures.
+        "temperature": temperature,
         "tools": tools or None,
         "response_format": response_format or None,
         "seed": seed,
+        "max_tokens": max_tokens,
+        "top_p": top_p,
+        "stop": stop,
+        "n": n,
+        "tool_choice": tool_choice,
+        "presence_penalty": presence_penalty,
+        "frequency_penalty": frequency_penalty,
     }
     blob = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()
     return hashlib.sha256(blob).hexdigest()
@@ -52,16 +79,24 @@ def is_cacheable(body: dict) -> bool:
 
     - Streaming responses are skipped (caching SSE chunks correctly is more
       trouble than it's worth for v1).
-    - Non-zero temperature without an explicit seed → non-deterministic, skip.
-    - With a seed, any temperature is fine — the seed pins the output.
+    - With an explicit seed, any sampling params are fine — the seed pins
+      the output.
+    - Otherwise the temperature must be EXPLICITLY zero AND top_p must not
+      narrow the distribution. An OMITTED temperature means the provider
+      default (1.0 — maximally non-deterministic), never cacheable.
     """
     if body.get("stream"):
         return False
-    temperature = body.get("temperature", 0.0) or 0.0
     seed = body.get("seed")
-    if temperature == 0.0:
+    if seed is not None:
         return True
-    return seed is not None
+    temperature = body.get("temperature")
+    if temperature is None or float(temperature) != 0.0:
+        return False
+    top_p = body.get("top_p")
+    if top_p is not None and float(top_p) != 1.0:
+        return False
+    return True
 
 
 # ── Backends ──────────────────────────────────────────────────────────

app/routes/chat.py

@@ -30,6 +30,7 @@
 from app.deps import get_db, get_key_context
 from app.quality_scores import resolve_model_metrics
 from app.schemas import ChatCompletionRequest
+from packages.auth.spend import budget_exceeded, get_lifetime_spend_microcents
 from packages.auth.types import KeyContext
 from packages.db.models.request_log import RequestLog
 from packages.litellm_adapter.catalog import CATALOG, CATALOG_BY_ID
@@ -153,7 +154,12 @@ async def _build_log_row(
             output_tokens=output_t,
             fallback_model=requested_model,
         ),
-        latency_ms=meta.get("latency_ms", latency_ms),
+        # `or latency_ms` (not `is None`): the streaming aggregator and the
+        # cache-hit path both emit a literal 0 when no real measurement
+        # exists, and 0 must fall back to the wall-clock measurement — a
+        # `default=` would silently accept the bogus zero and poison
+        # /v1/analytics/latency percentiles.
+        latency_ms=meta.get("latency_ms") or latency_ms,
         status_code=status_code,
         error_type=error_type,
         is_streaming=body.stream,
@@ -267,6 +273,21 @@ async def chat_completions(
             detail=f"Model '{body.model}' is not allowed for this API key",
         )
 
+    # Budget enforcement: `budget_limit_cents` is a lifetime cap on this
+    # key's billable (status < 400) spend. Checked before any routing,
+    # resolution, or cache work so an exhausted key costs the operator
+    # nothing — no upstream attempt, no cache fill.
+    if kc.budget_limit_cents is not None:
+        spend = await get_lifetime_spend_microcents(db, str(kc.key_id))
+        if budget_exceeded(spend, kc.budget_limit_cents):
+            raise HTTPException(
+                status_code=429,
+                detail=(
+                    "API key budget exhausted "
+                    f"({spend} of {kc.budget_limit_cents * 10_000} microcents spent)."
+                ),
+            )
+
     client = await router_cache.get_router(db)
     raw_strategy = getattr(client, "strategy", None)
     strategy = raw_strategy if isinstance(raw_strategy, str) and raw_strategy else "balanced"
@@ -401,6 +422,13 @@ async def chat_completions(
             tools=completion_kwargs.get("tools"),
             response_format=completion_kwargs.get("response_format"),
             seed=completion_kwargs.get("seed"),
+            max_tokens=completion_kwargs.get("max_tokens"),
+            top_p=completion_kwargs.get("top_p"),
+            stop=completion_kwargs.get("stop"),
+            n=completion_kwargs.get("n"),
+            tool_choice=completion_kwargs.get("tool_choice"),
+            presence_penalty=completion_kwargs.get("presence_penalty"),
+            frequency_penalty=completion_kwargs.get("frequency_penalty"),
         )
         cached = await prompt_cache.get_backend().get(cache_lookup_key)
         if cached is not None:
@@ -420,7 +448,10 @@ async def chat_completions(
             response={
                 "model": cached_model,
                 "usage": cache_hit_response.get("usage", {}),
-                "_orca_meta": {"provider": "cache", "latency_ms": 0},
+                # No latency_ms here: _build_log_row falls back to the
+                # measured serve time. Logging a hardcoded 0 dragged the
+                # analytics percentiles toward zero on every cache hit.
+                "_orca_meta": {"provider": "cache"},
             },
             status_code=200, error_type=None, started_perf=started_perf,
             strategy=strategy,

app/routes/keys.py

@@ -5,7 +5,7 @@
 from datetime import datetime, timezone
 
 from fastapi import APIRouter, Depends, HTTPException, Response
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -20,16 +20,46 @@
 
 class CreateKey(BaseModel):
     name: str
+    # Optional restrictions for child keys. Only reachable by unrestricted
+    # callers (require_unrestricted above), so a restricted key can never
+    # mint a sibling with looser limits than its own — it can't mint at all.
+    model_allowlist: list[str] | None = None
+    budget_limit_cents: int | None = Field(default=None, gt=0)
+
+
+def require_unrestricted(kc: KeyContext) -> None:
+    """Key management is reserved for unrestricted keys.
+
+    A key that carries any restriction (`model_allowlist` or
+    `budget_limit_cents`) must not be able to mint, list, or revoke other
+    keys — otherwise it could create a sibling with no restrictions and
+    trivially bypass its own allowlist/budget. Unrestricted keys already
+    hold the maximum privilege this single-workspace edition exposes
+    (same trust level as PUT /v1/providers/*), so denying restricted keys
+    here grants nothing to anyone; it only closes the escalation path.
+    """
+    if kc.model_allowlist is not None or kc.budget_limit_cents is not None:
+        raise HTTPException(
+            status_code=403,
+            detail=(
+                "Restricted API keys cannot manage keys. "
+                "Use an unrestricted key."
+            ),
+        )
 
 
 @router.get("")
 async def list_keys(
-    _kc: KeyContext = Depends(get_key_context),
+    kc: KeyContext = Depends(get_key_context),
     db: AsyncSession = Depends(get_db),
 ) -> dict:
+    require_unrestricted(kc)
     rows = (
         await db.execute(
-            select(ApiKey).where(ApiKey.is_deleted == 0).order_by(ApiKey.created_at)
+            select(ApiKey).where(
+                ApiKey.workspace_id == kc.workspace_id,
+                ApiKey.is_deleted == 0,
+            ).order_by(ApiKey.created_at)
         )
     ).scalars().all()
     return {
@@ -54,12 +84,15 @@ async def create_key(
     kc: KeyContext = Depends(get_key_context),
     db: AsyncSession = Depends(get_db),
 ) -> dict:
+    require_unrestricted(kc)
     full_key, key_hash, key_prefix = generate_api_key()
     row = ApiKey(
         workspace_id=kc.workspace_id,
         name=body.name,
         key_hash=key_hash,
         key_prefix=key_prefix,
+        model_allowlist=body.model_allowlist,
+        budget_limit_cents=body.budget_limit_cents,
     )
     db.add(row)
     await db.commit()
@@ -70,18 +103,28 @@ async def create_key(
         "name": row.name,
         "key_prefix": row.key_prefix,
         "api_key": full_key,  # plaintext shown ONCE
+        "model_allowlist": row.model_allowlist,
+        "budget_limit_cents": row.budget_limit_cents,
     }
 
 
 @router.delete("/{key_id}", status_code=204)
 async def revoke_key(
     key_id: str,
-    _kc: KeyContext = Depends(get_key_context),
+    kc: KeyContext = Depends(get_key_context),
     db: AsyncSession = Depends(get_db),
 ) -> Response:
+    require_unrestricted(kc)
     row = (
         await db.execute(
-            select(ApiKey).where(ApiKey.id == key_id, ApiKey.is_deleted == 0)
+            select(ApiKey).where(
+                ApiKey.id == key_id,
+                # Workspace scoping: without this, any key could revoke any
+                # other workspace's keys (the write path has always been
+                # scoped; the read/delete paths were not).
+                ApiKey.workspace_id == kc.workspace_id,
+                ApiKey.is_deleted == 0,
+            )
         )
     ).scalar_one_or_none()
     if row is None: