Add API key usage tracking (SEL-346)

.github/workflows/ci.yml

@@ -822,3 +822,4 @@ jobs:
 
       - name: Check doc references resolve
         run: python scripts/check_doc_references.py
+

app/core/auth.py

@@ -11,8 +11,9 @@
 from dataclasses import dataclass, replace
 from typing import Annotated
 
-from fastapi import Depends, Header, HTTPException, Request, Security, status
+from fastapi import Depends, Header, HTTPException, Security, status
 from fastapi.security import APIKeyHeader, HTTPAuthorizationCredentials, HTTPBearer
+from starlette.requests import Request
 from ..db.tenant_rls import bind_wallet_scope, rls_admin_scope
 from .config import get_settings
 from .oidc_iga import EnterprisePrincipal, IGADecision, IGAError, is_iga_issuer_token
@@ -134,6 +135,13 @@ class CredentialAcceptance:
     "credential_acceptance", default=None
 )
 
+# ContextVar to store the authenticated context for usage tracking middleware.
+# Set by get_auth_context when credentials are accepted, so middleware can
+# access key_id and tenant without re-parsing headers.
+CURRENT_AUTH_CONTEXT: ContextVar[AuthContext | None] = ContextVar(
+    "current_auth_context", default=None
+)
+
 
 async def get_auth_context(
     api_key: str | None = Security(api_key_header),
@@ -166,6 +174,10 @@ async def get_auth_context(
     acceptance = CREDENTIAL_ACCEPTANCE.get()
     if acceptance is not None:
         acceptance.accepted = True
+
+    # Store auth context for usage tracking middleware
+    CURRENT_AUTH_CONTEXT.set(context)
+
     return context
 
 

app/db/models.py

@@ -419,6 +419,32 @@ class APIKeyModel(SQLModel, table=True):
     model_config = {"arbitrary_types_allowed": True}
 
 
+class APIKeyUsageEventModel(SQLModel, table=True):
+    """
+    Append-only API key usage event log for per-call usage tracking.
+
+    Records method, route template, status code, and latency per authenticated
+    call. No request bodies, IPs, headers, or PII. Used for demo key
+    analytics and API usage monitoring.
+    """
+
+    __tablename__ = "api_key_usage_events"
+    __table_args__ = (
+        Index("ix_api_key_usage_events_key_id_created_at", "key_id", "created_at"),
+    )
+
+    id: str = Field(primary_key=True, max_length=64)
+    key_id: str = Field(max_length=50, foreign_key="api_keys.key_id", index=True)
+    tenant: Optional[str] = Field(default=None, max_length=32)
+    method: str = Field(max_length=10)
+    route_template: str = Field(max_length=255)
+    status_code: int
+    latency_ms: int
+    created_at: datetime = Field(sa_type=NaiveUTCDateTime, default_factory=utc_now)
+
+    model_config = {"arbitrary_types_allowed": True}
+
+
 class TelemetryEventModel(SQLModel, table=True):
     """
     Time-series telemetry events for the Autonomous PM.

app/main.py

@@ -661,6 +661,11 @@ async def periodic_cleanup():
 # Rate limiting (enforces documented 120 req/min per API key)
 app.add_middleware(RateLimitMiddleware)
 
+# API key usage tracking (records method, route, status, latency per authenticated call)
+from app.middleware.api_key_usage_tracking import APIKeyUsageTrackingMiddleware
+
+app.add_middleware(APIKeyUsageTrackingMiddleware)
+
 # Stash (method, path) for auth-time demo-tenant route confinement. No
 # authentication, no branching, negligible cost: only demo-tenant
 # credentials read this value. Added after (so it wraps) the rate limiter,

app/middleware/api_key_usage_tracking.py

@@ -0,0 +1,114 @@
+"""
+API Key Usage Tracking Middleware
+
+Records API key usage events after each authenticated request completes.
+Captures method, route template, status code, and latency without blocking
+the response path. A failed write never fails the request.
+"""
+
+import asyncio
+import logging
+import time
+from typing import Callable
+
+from fastapi import Request, Response
+from starlette.middleware.base import BaseHTTPMiddleware
+
+from ..core.auth import CREDENTIAL_ACCEPTANCE, CURRENT_AUTH_CONTEXT
+
+logger = logging.getLogger(__name__)
+
+
+class APIKeyUsageTrackingMiddleware(BaseHTTPMiddleware):
+    """
+    Records API key usage events after authenticated requests complete.
+
+    This middleware runs after authentication resolves an API key and captures:
+    - key_id: The authenticated API key
+    - tenant: The tenant label from the key
+    - method: HTTP method
+    - route_template: The matched route template (not the raw path)
+    - status_code: Response status code
+    - latency_ms: Request duration in milliseconds
+
+    A failed write is logged and never blocks the response.
+    """
+
+    async def dispatch(self, request: Request, call_next: Callable) -> Response:
+        start_time = time.time()
+
+        # Store start time in request state so we can track it after auth
+        request.state.usage_tracking_start = start_time
+
+        # Process the request (this will call auth and handlers)
+        response = await call_next(request)
+
+        # Only record for authenticated requests
+        acceptance = CREDENTIAL_ACCEPTANCE.get()
+        if acceptance is None or not acceptance.accepted:
+            return response
+
+        # Extract AuthContext from ContextVar (set by get_auth_context)
+        auth_context = CURRENT_AUTH_CONTEXT.get()
+        if auth_context is None:
+            # Not authenticated or auth context not available
+            return response
+
+        if auth_context.key_id is None:
+            # Bootstrap admin or static dev key - no usage tracking
+            return response
+
+        # Calculate latency
+        latency_ms = int((time.time() - start_time) * 1000)
+
+        # Determine route template
+        route = request.scope.get("route")
+        if route and hasattr(route, "path"):
+            route_template = route.path
+        else:
+            # Fallback to raw path if route template unavailable
+            route_template = request.url.path
+
+        # Record usage event asynchronously (fire and forget)
+        asyncio.create_task(
+            self._record_usage_event(
+                key_id=auth_context.key_id,
+                tenant=auth_context.tenant,
+                method=request.method,
+                route_template=route_template,
+                status_code=response.status_code,
+                latency_ms=latency_ms,
+            )
+        )
+
+        return response
+
+    async def _record_usage_event(
+        self,
+        key_id: str,
+        tenant: str | None,
+        method: str,
+        route_template: str,
+        status_code: int,
+        latency_ms: int,
+    ) -> None:
+        """
+        Record the usage event. This runs asynchronously and never blocks.
+        """
+        try:
+            from ..services.api_key_usage import record_api_key_usage_event
+
+            await record_api_key_usage_event(
+                key_id=key_id,
+                tenant=tenant,
+                method=method,
+                route_template=route_template,
+                status_code=status_code,
+                latency_ms=latency_ms,
+            )
+        except Exception as e:
+            # Catch-all: even if the import or call fails, don't block
+            logger.error(
+                f"Failed to record API key usage event for {key_id}: {e}",
+                exc_info=True,
+            )

app/routers/api_keys.py

@@ -5,6 +5,7 @@
 """
 
 from fastapi import APIRouter, Depends, Header, HTTPException, Request, status
+from sqlalchemy.ext.asyncio import AsyncSession  # noqa: F401
 
 from ..core.dependencies import get_agent_money
 
@@ -430,3 +431,77 @@ async def get_rotation_logs(
             status_code=404,
             detail=_not_found("wallet_not_found", str(e)),
         )
+
+
+@router.get(
+    "/usage/daily",
+    summary="Get daily API key usage statistics",
+    description=(
+        "Get aggregated daily usage statistics for API keys. "
+        "Admin-only endpoint; demo keys receive 403. "
+        "Returns call counts, error rates, latency percentiles, and first/last call timestamps."
+    ),
+)
+async def get_daily_usage_stats(
+    key_id: str | None = None,
+    limit: int = 100,
+    auth: AuthContext = Depends(get_auth_context),
+):
+    """
+    Get daily API key usage statistics.
+
+    Admin-only endpoint. Demo tenant keys are refused.
+    Exposes the api_key_usage_daily view aggregated by key_id and day.
+    """
+    auth.require_bootstrap_admin()
+
+    from sqlalchemy import text
+    from ..db.database import get_session_factory
+
+    session_factory = get_session_factory()
+    async with session_factory() as session:
+        # Query the view directly
+        query = text(
+            """
+            SELECT
+                key_id,
+                day,
+                calls,
+                count_4xx,
+                count_5xx,
+                p50_latency_ms,
+                p95_latency_ms,
+                first_call,
+                last_call
+            FROM api_key_usage_daily
+            WHERE (:key_id IS NULL OR key_id = :key_id)
+            ORDER BY day DESC, key_id
+            LIMIT :limit
+            """
+        )
+        result = await session.execute(
+            query,
+            {"key_id": key_id, "limit": limit},
+        )
+        rows = result.fetchall()
+
+        return [
+            {
+                "key_id": row[0],
+                "day": row[1]
+                if isinstance(row[1], str)
+                else (row[1].isoformat() if row[1] else None),
+                "calls": row[2],
+                "count_4xx": row[3],
+                "count_5xx": row[4],
+                "p50_latency_ms": row[5],
+                "p95_latency_ms": row[6],
+                "first_call": row[7]
+                if isinstance(row[7], str)
+                else (row[7].isoformat() if row[7] else None),
+                "last_call": row[8]
+                if isinstance(row[8], str)
+                else (row[8].isoformat() if row[8] else None),
+            }
+            for row in rows
+        ]

app/services/api_key_usage.py

@@ -0,0 +1,73 @@
+"""
+API Key Usage Event Tracking
+
+Records per-call usage events for API keys after successful authentication.
+Events are written asynchronously without blocking the request path.
+A failed write never fails the request.
+"""
+
+import logging
+from typing import Optional
+from uuid import uuid4
+
+from sqlalchemy.ext.asyncio import AsyncSession  # noqa: F401
+
+from ..core.time import to_naive_utc, utc_now
+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: Optional[str],
+    method: str,
+    route_template: str,
+    status_code: int,
+    latency_ms: int,
+) -> None:
+    """
+    Record one API key usage event asynchronously.
+
+    Args:
+        key_id: The API key ID that authenticated
+        tenant: The tenant label (None = normal, "demo" = demo tenant)
+        method: HTTP method (GET, POST, etc.)
+        route_template: The route template (not the raw path with IDs)
+        status_code: HTTP response status code
+        latency_ms: Request latency in milliseconds
+
+    This function never raises. A failed write is logged and the request
+    continues unchanged. No request bodies, IPs, headers, or PII are stored.
+    """
+    try:
+        event_id = f"ue_{uuid4().hex[:16]}"
+        now = to_naive_utc(utc_now())
+
+        event = APIKeyUsageEventModel(
+            id=event_id,
+            key_id=key_id,
+            tenant=tenant,
+            method=method[:10],  # Truncate to column size
+            route_template=route_template[:255],  # Truncate to column size
+            status_code=status_code,
+            latency_ms=latency_ms,
+            created_at=now,
+        )
+
+        session_factory = get_session_factory()
+        async with session_factory() as session:
+            session.add(event)
+            await session.commit()
+
+        logger.debug(
+            f"Recorded usage event {event_id} for key {key_id}: "
+            f"{method} {route_template} -> {status_code} ({latency_ms}ms)"
+        )
+    except Exception as e:
+        # A failed write never fails the request. Log and continue.
+        logger.error(
+            f"Failed to record API key usage event for key {key_id}: {e}",
+            exc_info=True,
+        )