SEL-346: AMW API: per-key usage events and daily per-key summary

gateway/app/core/api_key_usage.py

@@ -0,0 +1,89 @@
+"""Metadata-only HTTP usage events for database-backed API keys."""
+
+import logging
+import time
+from uuid import uuid4
+
+from starlette.types import ASGIApp, Message, Receive, Scope, Send
+
+from .auth import CREDENTIAL_ACCEPTANCE
+from ..db.database import get_session_factory
+from ..db.models import APIKeyUsageEventModel
+
+logger = logging.getLogger(__name__)
+
+
+async def record_api_key_usage_event(
+    *,
+    key_id: str,
+    tenant: str | None,
+    method: str,
+    endpoint: str,
+    status_code: int,
+    latency_ms: int,
+) -> None:
+    """Append one event using a separate session after the response."""
+    async with get_session_factory()() as session:
+        session.add(
+            APIKeyUsageEventModel(
+                id=str(uuid4()),
+                key_id=key_id,
+                tenant=tenant,
+                method=method,
+                endpoint=endpoint,
+                status_code=status_code,
+                latency_ms=latency_ms,
+            )
+        )
+        await session.commit()
+
+
+class APIKeyUsageMiddleware:
+    """Record the resolved key and final HTTP status once per API request."""
+
+    def __init__(self, app: ASGIApp) -> None:
+        self.app = app
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+
+        # RateLimitMiddleware sets this shared holder before entering this
+        # layer; auth fills its key identity in the child task after lookup.
+        acceptance = CREDENTIAL_ACCEPTANCE.get()
+        if acceptance is None:
+            await self.app(scope, receive, send)
+            return
+
+        started = time.perf_counter()
+        status_code = 500
+
+        async def send_with_status(message: Message) -> None:
+            nonlocal status_code
+            if message["type"] == "http.response.start":
+                status_code = message["status"]
+            await send(message)
+
+        try:
+            await self.app(scope, receive, send_with_status)
+        finally:
+            # route.path is the registered template. Never substitute scope's
+            # raw path, which could contain IDs, query values, or user text.
+            endpoint = getattr(scope.get("route"), "path", None)
+            if acceptance.key_id and endpoint:
+                latency_ms = max(0, int((time.perf_counter() - started) * 1000))
+                try:
+                    await record_api_key_usage_event(
+                        key_id=acceptance.key_id,
+                        tenant=acceptance.tenant,
+                        method=scope.get("amw_original_method", scope["method"]),
+                        endpoint=endpoint,
+                        status_code=status_code,
+                        latency_ms=latency_ms,
+                    )
+                except Exception as exc:  # noqa: BLE001 - usage cannot fail API calls
+                    logger.warning(
+                        "api_key_usage_event_write_failed error_type=%s",
+                        type(exc).__name__,
+                    )

gateway/app/core/auth.py

@@ -100,7 +100,9 @@ def require_bootstrap_admin(self) -> None:
         )
 
 
-async def _key_tenant_for_auth(key_tenant: str | None, wallet_id: str) -> str | None:
+async def _key_tenant_for_auth(
+    key_tenant: str | None, wallet_id: str, key_id: str | None = None
+) -> str | None:
     """Apply demo confinement to the key and its owning wallet."""
     from ..services.demo_tenant_boundary import (
         DEMO_KEY_ROUTE_FORBIDDEN,
@@ -112,6 +114,11 @@ async def _key_tenant_for_auth(key_tenant: str | None, wallet_id: str) -> str |
     )
 
     tenant = await resolve_effective_tenant(key_tenant, wallet_id)
+    if key_id is not None:
+        acceptance = CREDENTIAL_ACCEPTANCE.get()
+        if acceptance is not None:
+            acceptance.key_id = key_id
+            acceptance.tenant = tenant
     if tenant != DEMO_TENANT_LABEL:
         return tenant
     if not get_settings().ENABLE_DEMO_TENANT:
@@ -148,9 +155,11 @@ async def _key_tenant_for_auth(key_tenant: str | None, wallet_id: str) -> str |
 
 @dataclass
 class CredentialAcceptance:
-    """Whether ``get_auth_context`` accepted credentials during one request."""
+    """Accepted credentials and any database key resolved for the request."""
 
     accepted: bool = False
+    key_id: str | None = None
+    tenant: str | None = None
 
 
 # The converse signal, for RateLimitMiddleware's shared pre-auth bucket. A
@@ -190,6 +199,8 @@ async def get_auth_context(
     acceptance = CREDENTIAL_ACCEPTANCE.get()
     if acceptance is not None:
         acceptance.accepted = True
+        acceptance.key_id = context.key_id
+        acceptance.tenant = context.tenant
     return context
 
 
@@ -311,7 +322,9 @@ async def _resolve_auth_context(
 
     if candidate:
         if _key_is_live(candidate, utc_now()):
-            await _key_tenant_for_auth(candidate.tenant, candidate.wallet_id)
+            await _key_tenant_for_auth(
+                candidate.tenant, candidate.wallet_id, candidate.key_id
+            )
         try:
             db_key = await key_service.validate_key(stripped)
         except RuntimeError:
@@ -322,7 +335,9 @@ async def _resolve_auth_context(
     if db_key:
         from ..services.demo_tenant_boundary import parse_allowlist
 
-        tenant = await _key_tenant_for_auth(db_key.tenant, db_key.wallet_id)
+        tenant = await _key_tenant_for_auth(
+            db_key.tenant, db_key.wallet_id, db_key.key_id
+        )
         return AuthContext(
             source="db",
             raw_key=stripped,
@@ -464,7 +479,9 @@ async def _auth_from_jwt(token: str) -> AuthContext:
         and candidate.wallet_id == payload.sub
         and _key_is_live(candidate, utc_now())
     ):
-        await _key_tenant_for_auth(candidate.tenant, candidate.wallet_id)
+        await _key_tenant_for_auth(
+            candidate.tenant, candidate.wallet_id, candidate.key_id
+        )
 
     if not await key_service.consume_derived_key_use(payload.key_id, payload.sub):
         raise HTTPException(
@@ -507,7 +524,9 @@ async def _auth_from_jwt(token: str) -> AuthContext:
         )
     from ..services.demo_tenant_boundary import parse_allowlist
 
-    tenant = await _key_tenant_for_auth(key_row.tenant, key_row.wallet_id)
+    tenant = await _key_tenant_for_auth(
+        key_row.tenant, key_row.wallet_id, key_row.key_id
+    )
     return AuthContext(
         source="jwt",
         # Not a credential: a stable, non-secret handle for the API key that

gateway/app/db/api_key_usage_view.py

@@ -0,0 +1,58 @@
+"""Portable daily API-key usage view definition.
+
+Timestamps in this database are naive UTC. Percentiles use continuous linear
+interpolation, matching PostgreSQL's percentile_cont on both supported SQL
+dialects. The SQLite development schema and Alembic migration share this SQL.
+"""
+
+
+def _percentile_expression(percent: int) -> str:
+    return (
+        "SUM(CASE "
+        f"WHEN latency_rank = p{percent}_floor_rank "
+        f"THEN latency_ms * (100 - p{percent}_remainder) / 100.0 "
+        f"WHEN latency_rank = p{percent}_floor_rank + 1 "
+        f"THEN latency_ms * p{percent}_remainder / 100.0 "
+        f"ELSE 0 END) AS p{percent}_latency_ms"
+    )
+
+
+CREATE_API_KEY_USAGE_DAILY_VIEW_SQL = f"""
+CREATE VIEW api_key_usage_daily AS
+WITH ranked AS (
+    SELECT
+        key_id,
+        DATE(created_at) AS day,
+        status_code,
+        latency_ms,
+        created_at,
+        ROW_NUMBER() OVER (
+            PARTITION BY key_id, DATE(created_at)
+            ORDER BY latency_ms, id
+        ) AS latency_rank,
+        COUNT(*) OVER (
+            PARTITION BY key_id, DATE(created_at)
+        ) AS call_count
+    FROM api_key_usage_events
+), positioned AS (
+    SELECT
+        *,
+        ((call_count - 1) * 50) / 100 + 1 AS p50_floor_rank,
+        ((call_count - 1) * 50) % 100 AS p50_remainder,
+        ((call_count - 1) * 95) / 100 + 1 AS p95_floor_rank,
+        ((call_count - 1) * 95) % 100 AS p95_remainder
+    FROM ranked
+)
+SELECT
+    key_id,
+    day,
+    COUNT(*) AS calls,
+    SUM(CASE WHEN status_code BETWEEN 400 AND 499 THEN 1 ELSE 0 END) AS calls_4xx,
+    SUM(CASE WHEN status_code BETWEEN 500 AND 599 THEN 1 ELSE 0 END) AS calls_5xx,
+    {_percentile_expression(50)},
+    {_percentile_expression(95)},
+    MIN(created_at) AS first_call_at,
+    MAX(created_at) AS last_call_at
+FROM positioned
+GROUP BY key_id, day
+"""

gateway/app/db/database.py

@@ -35,6 +35,7 @@
 from ..core.db_urls import as_sqlalchemy_url
 from ..core.time import to_naive_utc
 from ..core.trust_mode import is_production_like_environment
+from .api_key_usage_view import CREATE_API_KEY_USAGE_DAILY_VIEW_SQL
 
 logger = logging.getLogger(__name__)
 
@@ -61,9 +62,22 @@
         "human_approvals",
         "receipt_resolutions",
         "resolution_chain_heads",
+        "api_key_usage_events",
     }
 )
 REQUIRED_TRUST_COLUMNS = {
+    "api_key_usage_events": frozenset(
+        {
+            "id",
+            "key_id",
+            "tenant",
+            "method",
+            "endpoint",
+            "status_code",
+            "latency_ms",
+            "created_at",
+        }
+    ),
     "mcp_dispatch_attempts": frozenset({"dispatch_claim_hash"}),
     "resolution_chain_heads": frozenset({"last_sequence", "last_hash"}),
     "receipt_resolutions": frozenset(
@@ -303,6 +317,10 @@ def _missing_required_tables(sync_conn) -> set[str]:  # noqa: ANN001
     return set(REQUIRED_TRUST_TABLES) - existing
 
 
+def _missing_required_views(sync_conn) -> set[str]:  # noqa: ANN001
+    return {"api_key_usage_daily"} - set(sa_inspect(sync_conn).get_view_names())
+
+
 def _missing_required_columns(sync_conn) -> set[str]:  # noqa: ANN001
     inspector = sa_inspect(sync_conn)
     existing_tables = set(inspector.get_table_names())
@@ -359,6 +377,7 @@ async def verify_required_schema(engine: AsyncEngine) -> None:
     async with engine.connect() as conn:
         missing = await conn.run_sync(_missing_required_tables)
         missing_columns = await conn.run_sync(_missing_required_columns)
+        missing_views = await conn.run_sync(_missing_required_views)
         stale = await conn.run_sync(_stale_alembic_message)
     # A stamped database can be both behind head and missing tables introduced
     # by later revisions. Report the revision mismatch first: it is the
@@ -381,6 +400,12 @@ async def verify_required_schema(engine: AsyncEngine) -> None:
             "the API (normally `alembic upgrade head`; review an unstamped "
             "legacy schema before stamping its exact existing revision)."
         )
+    if missing_views:
+        raise SchemaInitError(
+            "Required database views missing: "
+            + ", ".join(sorted(missing_views))
+            + ". Bring the schema to the packaged Alembic head before starting the API."
+        )
 
 
 async def init_db() -> None:
@@ -403,6 +428,8 @@ async def init_db() -> None:
     if allows_metadata_create_all(dialect_name=dialect, environment=environment):
         async with engine.begin() as conn:
             await conn.run_sync(SQLModel.metadata.create_all)
+            if "api_key_usage_daily" in await conn.run_sync(_missing_required_views):
+                await conn.exec_driver_sql(CREATE_API_KEY_USAGE_DAILY_VIEW_SQL)
             missing_columns = await conn.run_sync(_missing_required_columns)
         if missing_columns:
             raise SchemaInitError(

gateway/app/db/models.py

@@ -290,9 +290,9 @@ class APIKeyModel(SQLModel, table=True):
     )
     expires_at: Optional[datetime] = Field(sa_type=NaiveUTCDateTime, default=None)
 
-    # Bounded-use keys: None = unlimited. use_count only advances through the
-    # guarded UPDATE in APIKeyService.validate_key so the cap cannot be
-    # overshot by concurrent requests.
+    # Bounded-use keys: None = unlimited. use_count tracks capped-key budget
+    # consumption, not total calls. api_key_usage_events is the call source
+    # of truth; it also includes unbounded keys and response outcomes.
     max_uses: Optional[int] = Field(default=None)
     use_count: int = Field(default=0)
 
@@ -314,6 +314,25 @@ class APIKeyModel(SQLModel, table=True):
     model_config = {"arbitrary_types_allowed": True}
 
 
+class APIKeyUsageEventModel(SQLModel, table=True):
+    """One metadata-only HTTP outcome for a resolved database API key."""
+
+    __tablename__ = "api_key_usage_events"
+    __table_args__ = (
+        Index("ix_api_key_usage_events_key_created", "key_id", "created_at"),
+    )
+
+    id: str = Field(primary_key=True, max_length=36)
+    # Historical events survive key revocation or deletion.
+    key_id: str = Field(max_length=50)
+    tenant: Optional[str] = Field(default=None, max_length=32)
+    method: str = Field(max_length=16)
+    endpoint: str = Field(max_length=512)
+    status_code: int
+    latency_ms: int
+    created_at: datetime = Field(sa_type=NaiveUTCDateTime, default_factory=utc_now)
+
+
 class TelemetryEventModel(SQLModel, table=True):
     """
     Time-series telemetry events for the Autonomous PM.

gateway/app/main.py

@@ -26,6 +26,7 @@
 from fastapi.responses import HTMLResponse, JSONResponse
 
 from .core.auth import AuthContext, get_auth_context
+from .core.api_key_usage import APIKeyUsageMiddleware
 from .core.build_metadata import get_build_commit_sha, get_build_provenance
 from .core.config import get_settings
 from .core.durable_state import (
@@ -581,6 +582,10 @@ def build_openapi_servers(
 
 # --- Middleware Stack ---
 
+# Run inside the rate limiter so its accepted-credential holder is shared
+# with auth; this records at most one event after the routed call completes.
+app.add_middleware(APIKeyUsageMiddleware)
+
 # Rate limiting (enforces documented 120 req/min per API key)
 app.add_middleware(RateLimitMiddleware)
 
@@ -799,6 +804,7 @@ def mount_dormant_trust_surfaces(application: FastAPI) -> None:
             else True
         ),
     )
+app.include_router(api_keys.admin_router)
 
 # Self-serve dev keys: the handler is triple-gated at runtime (its own
 # ENABLE_DEV_KEY_SELF_PROVISION flag answers 404, production-like environments

gateway/app/middleware/head_method.py

@@ -42,6 +42,7 @@ async def __call__(
         # Copy before mutating: the original scope belongs to the server.
         get_scope = dict(scope)
         get_scope["method"] = "GET"
+        get_scope["amw_original_method"] = "HEAD"
 
         async def send_without_body(message: dict[str, Any]) -> None:
             if message["type"] == "http.response.body":