diff --git a/CHANGELOG.md b/CHANGELOG.md index 37f2ee3e..a297128a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,36 @@ All notable changes to AgentFlow are documented in this file. ## [Unreleased] +### Added — API-usage accounting and session analytics behind the ControlPlaneStore port (ADR 0010 slice 4, 2026-07-02) + +- **`api_usage`** (per-tenant/per-key request counters) and **`api_sessions`** + (per-request latency/entity/query telemetry) moved behind + `ControlPlaneStore`. Unlike every prior slice, this state was never on + `query_engine._conn` — `AuthManager.db_path` always resolves to its own + DuckDB file, independent of `DUCKDB_PATH` — so `EmbeddedControlPlaneStore` + gains a second, orthogonal `usage_db_path_provider` alongside its existing + `conn_provider`/`alert_rules_path_provider`, and `AuthManager` builds a + private embedded store bound to its own path (mirrors `OutboxProcessor`'s + additive-`store` pattern) rather than sharing the app-wide store. +- **Scope widened past `usage_table.py` + `analytics.py`**: `KeyRotator` + (`key_rotation.py`) queried `api_usage` directly for old-key-usage stats, + and the admin dashboard (`admin_ui.py`) queried `api_sessions` directly for + its QPS tile — both bypassed every prior slice's port and are covered here + too, so a PostgreSQL swap (slice 5) doesn't leave two call sites hard-wired + to a local DuckDB file. +- `record_usage`'s Windows file-lock fallback (temp-path rename on + `duckdb.IOException`) and `record_api_session`'s best-effort + log-and-return-on-exhaustion (vs. `record_api_usage`'s raise, which + `record_usage` depends on to skip its post-insert audit publish) both moved + into the store verbatim. `usage_table.py` / `analytics.py` / `key_rotation.py` + / `admin_ui.py` keep their pre-port public/test-facing signatures — only + the SQL and connection handling moved. +- New `EmbeddedControlPlaneStore` methods: `ensure_usage_schema`, + `record_api_usage`, `get_usage_by_tenant`, `get_usage_by_key`, + `get_old_key_usage_by_key_id`, `record_api_session`, `get_usage_analytics`, + `get_top_queries`, `get_top_entities`, `get_latency_analytics`, + `get_anomalies`, `get_queries_per_second_last_minute`. + ### Added — Replay outbox + dead-letter behind the ControlPlaneStore port (ADR 0010 slice 3, 2026-07-02) - **Invariant 8 preserved verbatim**: `mark_outbox_sent` flips an outbox row diff --git a/docs/decisions/0010-control-plane-externalization-postgres.md b/docs/decisions/0010-control-plane-externalization-postgres.md index 30e625ec..2f4a8398 100644 --- a/docs/decisions/0010-control-plane-externalization-postgres.md +++ b/docs/decisions/0010-control-plane-externalization-postgres.md @@ -178,8 +178,11 @@ ClickHouse is consumed), and `psycopg` joins the optional dependencies. 2. Alert history + alert rules/runtime-state repository behind the port (embedded impl keeps the YAML file format). 3. Outbox + dead-letter behind the port (invariant 8 preserved verbatim). -4. Usage/sessions behind the port (pooled writes replace the - per-request file connections in `usage_table.py`). +4. Usage/sessions behind the port — `usage_table.py`, `analytics.py`, + `key_rotation.py`'s old-key-usage queries and `admin_ui.py`'s QPS tile all + queried `api_usage`/`api_sessions` directly; all four move behind the port + (pure extraction, per-request file connections unchanged — pooling is not + part of this ADR). 5. `PostgresControlPlaneStore` + live verification (standalone-PG probe suite: parallel claim exclusivity, lease expiry re-drive, restart re-drive, enqueue-win uniqueness, outbox↔dead-letter atomicity) + diff --git a/src/serving/api/analytics.py b/src/serving/api/analytics.py index ac0cba57..a891ab31 100644 --- a/src/serving/api/analytics.py +++ b/src/serving/api/analytics.py @@ -1,6 +1,5 @@ import json import os -import re import threading import time from collections.abc import Awaitable, Callable @@ -14,10 +13,21 @@ from starlette.responses import Response from starlette.types import Message +from src.serving.control_plane import EmbeddedControlPlaneStore from src.serving.duckdb_connection import connect_duckdb logger = structlog.get_logger() + +def _usage_store(db_path: Path | str) -> EmbeddedControlPlaneStore: + # ADR 0010 slice 4: the SQL for the functions below lives behind the + # ControlPlaneStore port (control_plane/embedded.py) so a future + # PostgreSQL adapter can serve them too. A fresh, cheap wrapper per call — + # nothing is connected until a method on it runs — matches these + # functions' pre-port behavior of opening their own connection each call. + return EmbeddedControlPlaneStore(usage_db_path_provider=lambda: db_path) + + AnalyticsMiddleware = Callable[ [Request, Callable[[Request], Awaitable[Response]]], Awaitable[Response], @@ -166,72 +176,7 @@ def get_usage_analytics( window: str = "24h", tenant: str | None = None, ) -> dict: - interval = _window_to_interval(window) - ensure_analytics_table(db_path) - conn = connect_duckdb(db_path) - try: - if tenant: - rows = conn.execute( - """ - SELECT - tenant, - COUNT(*) AS total_requests, - ROUND(AVG(CASE WHEN status_code >= 400 THEN 1.0 ELSE 0.0 END), 4) AS error_rate, - ROUND(AVG(CASE WHEN cache_hit THEN 1.0 ELSE 0.0 END), 4) AS cache_hit_rate, - ROUND(AVG(duration_ms), 3) AS avg_duration_ms - FROM api_sessions - WHERE tenant IS NOT NULL - AND ts >= CURRENT_TIMESTAMP - CAST(? AS INTERVAL) - AND tenant = ? - GROUP BY tenant - ORDER BY tenant - """, - [interval, tenant], - ).fetchall() - else: - rows = conn.execute( - """ - SELECT - tenant, - COUNT(*) AS total_requests, - ROUND(AVG(CASE WHEN status_code >= 400 THEN 1.0 ELSE 0.0 END), 4) AS error_rate, - ROUND(AVG(CASE WHEN cache_hit THEN 1.0 ELSE 0.0 END), 4) AS cache_hit_rate, - ROUND(AVG(duration_ms), 3) AS avg_duration_ms - FROM api_sessions - WHERE tenant IS NOT NULL - AND ts >= CURRENT_TIMESTAMP - CAST(? AS INTERVAL) - GROUP BY tenant - ORDER BY tenant - """, - [interval], - ).fetchall() - tenants = [] - for tenant_name, total_requests, error_rate, cache_hit_rate, avg_duration_ms in rows: - top_endpoints = conn.execute( - """ - SELECT endpoint - FROM api_sessions - WHERE tenant = ? - AND ts >= CURRENT_TIMESTAMP - CAST(? AS INTERVAL) - GROUP BY endpoint - ORDER BY COUNT(*) DESC, endpoint - LIMIT 3 - """, - [tenant_name, interval], - ).fetchall() - tenants.append( - { - "tenant": tenant_name, - "total_requests": total_requests, - "error_rate": float(error_rate or 0.0), - "cache_hit_rate": float(cache_hit_rate or 0.0), - "top_endpoints": [item[0] for item in top_endpoints], - "avg_duration_ms": float(avg_duration_ms or 0.0), - } - ) - return {"window": window, "tenants": tenants} - finally: - conn.close() + return _usage_store(db_path).get_usage_analytics(window=window, tenant=tenant) def get_top_queries( @@ -240,30 +185,7 @@ def get_top_queries( limit: int = 10, window: str = "24h", ) -> dict: - interval = _window_to_interval(window) - ensure_analytics_table(db_path) - conn = connect_duckdb(db_path) - try: - rows = conn.execute( - """ - SELECT query_text, COUNT(*) AS frequency - FROM api_sessions - WHERE query_text IS NOT NULL - AND ts >= CURRENT_TIMESTAMP - CAST(? AS INTERVAL) - GROUP BY query_text - ORDER BY frequency DESC, query_text - LIMIT ? - """, - [interval, limit], - ).fetchall() - return { - "window": window, - "queries": [ - {"query": query_text, "count": frequency} for query_text, frequency in rows - ], - } - finally: - conn.close() + return _usage_store(db_path).get_top_queries(limit=limit, window=window) def get_top_entities( @@ -272,35 +194,7 @@ def get_top_entities( limit: int = 10, window: str = "24h", ) -> dict: - interval = _window_to_interval(window) - ensure_analytics_table(db_path) - conn = connect_duckdb(db_path) - try: - rows = conn.execute( - """ - SELECT entity_type, entity_id, COUNT(*) AS frequency - FROM api_sessions - WHERE entity_id IS NOT NULL - AND ts >= CURRENT_TIMESTAMP - CAST(? AS INTERVAL) - GROUP BY entity_type, entity_id - ORDER BY frequency DESC, entity_type, entity_id - LIMIT ? - """, - [interval, limit], - ).fetchall() - return { - "window": window, - "entities": [ - { - "entity_type": entity_type, - "entity_id": entity_id, - "count": frequency, - } - for entity_type, entity_id, frequency in rows - ], - } - finally: - conn.close() + return _usage_store(db_path).get_top_entities(limit=limit, window=window) def get_latency_analytics( @@ -308,119 +202,11 @@ def get_latency_analytics( *, window: str = "24h", ) -> dict: - interval = _window_to_interval(window) - ensure_analytics_table(db_path) - conn = connect_duckdb(db_path) - try: - rows = conn.execute( - """ - SELECT - endpoint, - COUNT(*) AS requests, - ROUND(quantile_cont(duration_ms, 0.50), 3) AS p50_ms, - ROUND(quantile_cont(duration_ms, 0.95), 3) AS p95_ms, - ROUND(quantile_cont(duration_ms, 0.99), 3) AS p99_ms - FROM api_sessions - WHERE ts >= CURRENT_TIMESTAMP - CAST(? AS INTERVAL) - GROUP BY endpoint - ORDER BY endpoint - """, - [interval], - ).fetchall() - return { - "window": window, - "endpoints": [ - { - "endpoint": endpoint, - "requests": requests, - "p50_ms": float(p50_ms or 0.0), - "p95_ms": float(p95_ms or 0.0), - "p99_ms": float(p99_ms or 0.0), - } - for endpoint, requests, p50_ms, p95_ms, p99_ms in rows - ], - } - finally: - conn.close() + return _usage_store(db_path).get_latency_analytics(window=window) def get_anomalies(db_path: Path | str, *, window: str = "24h") -> dict: - interval = _window_to_interval(window) - ensure_analytics_table(db_path) - conn = connect_duckdb(db_path) - try: - rows = conn.execute( - """ - WITH hourly AS ( - SELECT - tenant, - date_trunc('hour', ts) AS hour_bucket, - COUNT(*) AS requests - FROM api_sessions - WHERE tenant IS NOT NULL - AND ts >= CURRENT_TIMESTAMP - CAST(? AS INTERVAL) - GROUP BY tenant, hour_bucket - ), - latest AS ( - SELECT tenant, MAX(hour_bucket) AS current_hour - FROM hourly - GROUP BY tenant - ), - current_hour AS ( - SELECT - hourly.tenant, - hourly.hour_bucket, - hourly.requests AS current_hour_requests - FROM hourly - JOIN latest - ON latest.tenant = hourly.tenant - AND latest.current_hour = hourly.hour_bucket - ), - historical AS ( - SELECT - current_hour.tenant, - ROUND(AVG(hourly.requests), 1) AS hourly_average - FROM current_hour - JOIN hourly - ON hourly.tenant = current_hour.tenant - AND hourly.hour_bucket < current_hour.hour_bucket - GROUP BY current_hour.tenant - ), - scored AS ( - SELECT - current_hour.tenant, - current_hour.current_hour_requests, - historical.hourly_average, - ROUND( - current_hour.current_hour_requests - / NULLIF(historical.hourly_average, 0), - 2 - ) AS spike_ratio - FROM current_hour - JOIN historical - ON historical.tenant = current_hour.tenant - ) - SELECT tenant, current_hour_requests, hourly_average, spike_ratio - FROM scored - WHERE spike_ratio > 3 - ORDER BY spike_ratio DESC, tenant - """, - [interval], - ).fetchall() - return { - "window": window, - "anomalies": [ - { - "tenant": tenant, - "current_hour_requests": current_hour_requests, - "hourly_average": float(hourly_average or 0.0), - "spike_ratio": float(spike_ratio or 0.0), - } - for tenant, current_hour_requests, hourly_average, spike_ratio in rows - ], - } - finally: - conn.close() + return _usage_store(db_path).get_anomalies(window=window) def _schedule_session_write(db_path: Path | str, request_id: str, record: dict) -> None: @@ -432,80 +218,11 @@ def _schedule_session_write(db_path: Path | str, request_id: str, record: dict) def _insert_session(db_path: Path | str, request_id: str, record: dict) -> None: - for attempt in range(10): - try: - conn = connect_duckdb(db_path) - except duckdb.Error as exc: - if attempt == 9: - logger.warning( - "analytics_session_write_skipped", - stage="connect", - db_path=str(db_path), - request_id=request_id, - tenant=record.get("tenant"), - endpoint=record.get("endpoint"), - attempts=attempt + 1, - error=str(exc), - exc_info=True, - ) - return - time.sleep(0.01 * (attempt + 1)) - continue - - try: - conn.execute( - """ - INSERT OR REPLACE INTO api_sessions ( - request_id, - tenant, - key_name, - endpoint, - method, - status_code, - duration_ms, - cache_hit, - entity_type, - entity_id, - metric_name, - query_engine, - query_text - ) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - """, - [ - request_id, - record["tenant"], - record["key_name"], - record["endpoint"], - record["method"], - record["status_code"], - record["duration_ms"], - record["cache_hit"], - record["entity_type"], - record["entity_id"], - record["metric_name"], - record["query_engine"], - record["query_text"], - ], - ) - return - except duckdb.Error as exc: - if attempt == 9: - logger.warning( - "analytics_session_write_skipped", - stage="insert", - db_path=str(db_path), - request_id=request_id, - tenant=record.get("tenant"), - endpoint=record.get("endpoint"), - attempts=attempt + 1, - error=str(exc), - exc_info=True, - ) - return - time.sleep(0.01 * (attempt + 1)) - finally: - conn.close() + # Deliberately does NOT call ensure_analytics_table: the table is + # guaranteed to exist by main.py's boot-time call, and re-checking it on + # every background write would be wasted work on the hot path (see + # test_insert_session_uses_existing_schema_without_rechecking). + _usage_store(db_path).record_api_session(request_id, record) def _build_session_record( @@ -561,15 +278,3 @@ def _build_session_record( "query_engine": query_engine, "query_text": query_text, } - - -def _window_to_interval(window: str) -> str: - match = re.fullmatch(r"(\d+)([mhd])", window.strip()) - if match is None: - raise ValueError("Invalid window. Use formats like 15m, 1h, or 7d.") - value, unit = match.groups() - if unit == "m": - return f"{value} minutes" - if unit == "h": - return f"{value} hours" - return f"{value} days" diff --git a/src/serving/api/auth/key_rotation.py b/src/serving/api/auth/key_rotation.py index 46a31a43..4f6a8521 100644 --- a/src/serving/api/auth/key_rotation.py +++ b/src/serving/api/auth/key_rotation.py @@ -4,18 +4,14 @@ import re import secrets import threading -import time from datetime import UTC, datetime, timedelta -import duckdb - try: import yaml except ImportError: # pragma: no cover yaml = None # type: ignore[assignment] from src.serving.api.security import compute_key_lookup, hash_api_key -from src.serving.duckdb_connection import connect_duckdb from .manager import ApiKeysConfig, AuthManager, KeyCreateRequest, TenantKey @@ -198,68 +194,17 @@ def shutdown(self) -> None: self.cancel_rotation_cleanup_timers() def old_key_usage_by_key_id(self) -> dict[str, int]: - for attempt in range(10): - try: - conn = connect_duckdb(self._manager.db_path) - except duckdb.Error: - if attempt == 9: - raise - time.sleep(0.01 * (attempt + 1)) - continue - - try: - rows = conn.execute( - """ - SELECT key_id, COUNT(*) AS requests_last_hour - FROM api_usage - WHERE key_slot = 'previous' - AND ts >= CURRENT_TIMESTAMP - INTERVAL '1 hour' - AND key_id IS NOT NULL - GROUP BY key_id - """ - ).fetchall() - return dict(rows) - except duckdb.Error: - if attempt == 9: - raise - time.sleep(0.01 * (attempt + 1)) - finally: - conn.close() - return {} + # ADR 0010 slice 4: routed through the ControlPlaneStore port — + # was a direct connect_duckdb(self._manager.db_path) query. + return self._manager.store.get_old_key_usage_by_key_id() def old_key_usage_last_hour(self, key_id: str) -> int: return self.old_key_usage_by_key_id().get(key_id, 0) def _usage_by_key(self) -> dict[tuple[str, str], int]: - for attempt in range(10): - try: - conn = connect_duckdb(self._manager.db_path) - except duckdb.Error: - if attempt == 9: - raise - time.sleep(0.01 * (attempt + 1)) - continue - - try: - rows = conn.execute( - """ - SELECT tenant, key_name, COUNT(*) AS requests_last_24h - FROM api_usage - WHERE ts >= CURRENT_TIMESTAMP - INTERVAL '24 hours' - GROUP BY tenant, key_name - """ - ).fetchall() - return { - (tenant, key_name): requests_last_24h - for tenant, key_name, requests_last_24h in rows - } - except duckdb.Error: - if attempt == 9: - raise - time.sleep(0.01 * (attempt + 1)) - finally: - conn.close() - return {} + # ADR 0010 slice 4: routed through the ControlPlaneStore port — + # was a direct connect_duckdb(self._manager.db_path) query. + return self._manager.store.get_usage_by_key() def write_config(self, config: ApiKeysConfig) -> None: if self._manager.api_keys_path is None: diff --git a/src/serving/api/auth/manager.py b/src/serving/api/auth/manager.py index d8c94e8a..a1421432 100644 --- a/src/serving/api/auth/manager.py +++ b/src/serving/api/auth/manager.py @@ -36,6 +36,7 @@ verify_api_key, ) from src.serving.audit_publisher import AuditPublisher, build_audit_publisher_from_env +from src.serving.control_plane import ControlPlaneStore, EmbeddedControlPlaneStore if TYPE_CHECKING: from .key_rotation import KeyRotator @@ -117,6 +118,8 @@ def __init__( rate_limiter: RateLimiter | None = None, redis_url: str | None = None, audit_publisher: AuditPublisher | None = None, + *, + store: ControlPlaneStore | None = None, ) -> None: self.api_keys_path = Path(api_keys_path) if api_keys_path else None resolved_db_path = Path(db_path) @@ -130,6 +133,16 @@ def __init__( suffix = pipeline_path.suffix or ".duckdb" resolved_db_path = pipeline_path.with_name(f"{pipeline_path.stem}_api{suffix}") self.db_path = resolved_db_path + # ADR 0010 slice 4: usage/session accounting goes through the + # ControlPlaneStore port. When no store is injected (the common + # case), build a private embedded store bound to this manager's own + # usage db path — never the app's shared query_engine connection; + # usage/session state has always lived in its own file (see the + # port module's docstring on why this is not the outbox's + # :memory:-vs-file duality). + self.store: ControlPlaneStore = store or EmbeddedControlPlaneStore( + usage_db_path_provider=lambda: self.db_path + ) self.admin_key = admin_key if admin_key is not None else os.getenv("AGENTFLOW_ADMIN_KEY") self.security_config_path = Path(security_config_path) self.security_policy = load_security_policy(self.security_config_path) diff --git a/src/serving/api/auth/usage_table.py b/src/serving/api/auth/usage_table.py index df5c8892..026c654b 100644 --- a/src/serving/api/auth/usage_table.py +++ b/src/serving/api/auth/usage_table.py @@ -1,84 +1,26 @@ -"""Per-tenant API-usage table — DuckDB schema + read/write helpers. +"""Per-tenant API-usage accounting — thin wrappers over `AuthManager.store`. Extracted from `middleware.py` per audit L-C4 (2026-05-25): DB schema management and INSERT/SELECT helpers don't belong alongside the ASGI middleware. Imported lazily from `AuthManager` to avoid an import -cycle with `manager.py`. +cycle with `manager.py`. The actual DuckDB schema/SQL now lives behind the +`ControlPlaneStore` port (ADR 0010 slice 4, `control_plane/embedded.py`) — +these functions keep their pre-port signatures because `AuthManager` and +tests call them directly. """ from __future__ import annotations -import os -import time -from pathlib import Path from typing import TYPE_CHECKING -import duckdb import structlog -from src.serving.duckdb_connection import connect_duckdb - if TYPE_CHECKING: from .manager import AuthManager, TenantKey def ensure_usage_table(manager: AuthManager) -> None: - for attempt in range(10): - try: - conn = connect_duckdb(manager.db_path) - except duckdb.IOException as exc: - if ( - os.getenv("AGENTFLOW_USAGE_DB_PATH") is None - and manager.db_path.name == "agentflow_api.duckdb" - ): - from src.serving.api import auth as auth_package - - fallback_path = ( - Path(os.getenv("TEMP", ".")) - / f"agentflow_api_{os.getpid()}_{time.time_ns()}.duckdb" - ) - auth_package.logger.warning( - "usage_db_path_fallback", - original=str(manager.db_path), - fallback=str(fallback_path), - error=str(exc), - ) - manager.db_path = fallback_path - conn = connect_duckdb(manager.db_path) - else: - if attempt == 9: - raise - time.sleep(0.01 * (attempt + 1)) - continue - except duckdb.Error: - if attempt == 9: - raise - time.sleep(0.01 * (attempt + 1)) - continue - - try: - conn.execute( - """ - CREATE TABLE IF NOT EXISTS api_usage ( - tenant TEXT, - key_name TEXT, - endpoint TEXT, - ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP - ) - """ - ) - columns = {row[1] for row in conn.execute("PRAGMA table_info('api_usage')").fetchall()} - if "key_id" not in columns: - conn.execute("ALTER TABLE api_usage ADD COLUMN key_id TEXT") - if "key_slot" not in columns: - conn.execute("ALTER TABLE api_usage ADD COLUMN key_slot TEXT") - return - except duckdb.Error: - if attempt == 9: - raise - time.sleep(0.01 * (attempt + 1)) - finally: - conn.close() + manager.store.ensure_usage_schema() def record_usage(manager: AuthManager, tenant_key: TenantKey, endpoint: str) -> None: @@ -90,69 +32,28 @@ def record_usage(manager: AuthManager, tenant_key: TenantKey, endpoint: str) -> "key_id": tenant_key.key_id, "key_slot": tenant_key.matched_slot, } - inserted = False - for attempt in range(10): - try: - conn = connect_duckdb(manager.db_path) - except duckdb.Error: - if attempt == 9: - raise - time.sleep(0.01 * (attempt + 1)) - continue - - try: - conn.execute( - """ - INSERT INTO api_usage (tenant, key_name, endpoint, key_id, key_slot) - VALUES (?, ?, ?, ?, ?) - """, - [ - tenant_key.tenant, - tenant_key.name, - endpoint, - tenant_key.key_id, - tenant_key.matched_slot, - ], - ) - inserted = True - break - except duckdb.Error: - if attempt == 9: - raise - time.sleep(0.01 * (attempt + 1)) - finally: - conn.close() - - # Audit publish is intentionally outside the DB retry loop: a publish - # failure must not trigger another INSERT (H-C3 / audit-2026-05). - if inserted: - try: - manager.audit_publisher.publish(payload) - except Exception: - structlog.get_logger(__name__).warning( - "audit_publish_failed", - tenant=tenant_key.tenant, - endpoint=endpoint, - key_id=tenant_key.key_id, - exc_info=True, - ) + # Audit publish is intentionally outside the DB retry loop (delegated to + # the store below): a publish failure must not trigger another INSERT + # (H-C3 / audit-2026-05). If the insert never succeeds, record_api_usage + # raises and this function never reaches the publish call. + manager.store.record_api_usage( + tenant=tenant_key.tenant, + key_name=tenant_key.name, + endpoint=endpoint, + key_id=tenant_key.key_id, + key_slot=tenant_key.matched_slot, + ) + try: + manager.audit_publisher.publish(payload) + except Exception: + structlog.get_logger(__name__).warning( + "audit_publish_failed", + tenant=tenant_key.tenant, + endpoint=endpoint, + key_id=tenant_key.key_id, + exc_info=True, + ) def usage_by_tenant(manager: AuthManager) -> list[dict]: - conn = connect_duckdb(manager.db_path) - try: - rows = conn.execute( - """ - SELECT tenant, COUNT(*) AS requests_last_24h - FROM api_usage - WHERE ts >= CURRENT_TIMESTAMP - INTERVAL '24 hours' - GROUP BY tenant - ORDER BY tenant - """ - ).fetchall() - finally: - conn.close() - return [ - {"tenant": tenant, "requests_last_24h": requests_last_24h} - for tenant, requests_last_24h in rows - ] + return manager.store.get_usage_by_tenant() diff --git a/src/serving/api/routers/admin_ui.py b/src/serving/api/routers/admin_ui.py index d7966a58..c9438f95 100644 --- a/src/serving/api/routers/admin_ui.py +++ b/src/serving/api/routers/admin_ui.py @@ -2,7 +2,6 @@ from pathlib import Path -import duckdb from fastapi import APIRouter, Depends, Request from fastapi.responses import HTMLResponse from fastapi.templating import Jinja2Templates @@ -10,10 +9,9 @@ from starlette.datastructures import State from starlette.responses import Response -from src.serving.api.analytics import ensure_analytics_table from src.serving.api.auth import require_admin_key from src.serving.cache import ENTITY_TTL_SECONDS -from src.serving.duckdb_connection import connect_duckdb +from src.serving.control_plane import EmbeddedControlPlaneStore router = APIRouter( prefix="/admin", @@ -78,19 +76,7 @@ def _cache_stats(state: State) -> dict[str, object]: def _qps_last_minute(db_path: Path | str) -> float: - ensure_analytics_table(db_path) - conn = connect_duckdb(db_path) - try: - row = conn.execute( - """ - SELECT COUNT(*) - FROM api_sessions - WHERE ts >= CURRENT_TIMESTAMP - INTERVAL '1 minute' - """ - ).fetchone() - requests_last_minute = row[0] if row else 0 - except duckdb.Error: - return 0.0 - finally: - conn.close() - return round(float(requests_last_minute) / 60.0, 2) + # ADR 0010 slice 4: routed through the ControlPlaneStore port — was a + # direct connect_duckdb(db_path) query. + store = EmbeddedControlPlaneStore(usage_db_path_provider=lambda: db_path) + return store.get_queries_per_second_last_minute() diff --git a/src/serving/control_plane/__init__.py b/src/serving/control_plane/__init__.py index 9db0f985..adb91c53 100644 --- a/src/serving/control_plane/__init__.py +++ b/src/serving/control_plane/__init__.py @@ -9,6 +9,8 @@ from .embedded import ( EmbeddedControlPlaneStore, ensure_alert_history_table, + ensure_api_sessions_table, + ensure_api_usage_table, ensure_dead_letter_table, ensure_outbox_table, ensure_webhook_deliveries_table, @@ -29,6 +31,8 @@ "OutboxEntry", "WebhookQueueRow", "ensure_alert_history_table", + "ensure_api_sessions_table", + "ensure_api_usage_table", "ensure_dead_letter_table", "ensure_outbox_table", "ensure_webhook_deliveries_table", diff --git a/src/serving/control_plane/embedded.py b/src/serving/control_plane/embedded.py index 709144f8..8d3df50e 100644 --- a/src/serving/control_plane/embedded.py +++ b/src/serving/control_plane/embedded.py @@ -15,16 +15,23 @@ from __future__ import annotations import json +import os +import re +import time from collections.abc import Callable, Sequence from datetime import UTC, datetime, timedelta from pathlib import Path import duckdb +import structlog from src.db_concurrency import catalog_ddl_lock +from src.serving.duckdb_connection import connect_duckdb from .store import ControlPlaneStore, OutboxEntry, WebhookQueueRow +logger = structlog.get_logger() + try: import yaml except ImportError: # pragma: no cover @@ -173,6 +180,77 @@ def ensure_dead_letter_table(conn: duckdb.DuckDBPyConnection) -> None: ) +def ensure_api_usage_table(conn: duckdb.DuckDBPyConnection) -> None: + # Moved verbatim from auth/usage_table.py in ADR 0010 slice 4. Runs on a + # dedicated per-call connection (never the shared query_engine conn — see + # the store module docstring), so unlike its ensure_* siblings above it + # needs no catalog_ddl_lock. + conn.execute( + """ + CREATE TABLE IF NOT EXISTS api_usage ( + tenant TEXT, + key_name TEXT, + endpoint TEXT, + ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """ + ) + columns = {row[1] for row in conn.execute("PRAGMA table_info('api_usage')").fetchall()} + if "key_id" not in columns: + conn.execute("ALTER TABLE api_usage ADD COLUMN key_id TEXT") + if "key_slot" not in columns: + conn.execute("ALTER TABLE api_usage ADD COLUMN key_slot TEXT") + + +def ensure_api_sessions_table(conn: duckdb.DuckDBPyConnection) -> None: + # Moved verbatim from api/analytics.py in ADR 0010 slice 4 (that module's + # own path-based `ensure_analytics_table` stays put — it is independently + # pinned by main.py's boot call and a middleware test's monkeypatch — so + # this is a second, conn-based copy for the store's own methods). + conn.execute( + """ + CREATE TABLE IF NOT EXISTS api_sessions ( + request_id TEXT PRIMARY KEY, + tenant TEXT, + key_name TEXT, + endpoint TEXT, + method TEXT, + status_code INTEGER, + duration_ms FLOAT, + cache_hit BOOLEAN, + entity_type TEXT, + metric_name TEXT, + query_engine TEXT, + ts TIMESTAMP DEFAULT NOW() + ) + """ + ) + existing_columns = { + row[1] for row in conn.execute("PRAGMA table_info('api_sessions')").fetchall() + } + for column_name, column_type in ( + ("entity_id", "TEXT"), + ("query_text", "TEXT"), + ): + if column_name not in existing_columns: + conn.execute(f"ALTER TABLE api_sessions ADD COLUMN {column_name} {column_type}") + + +def _window_to_interval(window: str) -> str: + # Moved verbatim from api/analytics.py in ADR 0010 slice 4 — DuckDB + # interval-literal syntax is an adapter detail; a future PostgreSQL + # adapter parses `window` into its own interval syntax. + match = re.fullmatch(r"(\d+)([mhd])", window.strip()) + if match is None: + raise ValueError("Invalid window. Use formats like 15m, 1h, or 7d.") + value, unit = match.groups() + if unit == "m": + return f"{value} minutes" + if unit == "h": + return f"{value} hours" + return f"{value} days" + + class EmbeddedControlPlaneStore(ControlPlaneStore): """Control-plane state on the embedded serving DuckDB connection (queue, log and history tables) plus the YAML-backed alert-rule repository. @@ -187,12 +265,21 @@ class EmbeddedControlPlaneStore(ControlPlaneStore): def __init__( self, - conn_provider: Callable[[], duckdb.DuckDBPyConnection], + conn_provider: Callable[[], duckdb.DuckDBPyConnection] | None = None, *, alert_rules_path_provider: Callable[[], Path] | None = None, + usage_db_path_provider: Callable[[], Path | str] | None = None, ) -> None: self._conn_provider = conn_provider self._alert_rules_path_provider = alert_rules_path_provider + self._usage_db_path_provider = usage_db_path_provider + # Set once by _ensure_usage_db_connection's IOException fallback and + # then sticky for the rest of this store's lifetime — mirrors the + # pre-port code permanently reassigning `AuthManager.db_path` in + # place (module docstring: usage/session state is never on the + # shared conn_provider connection, so this override never touches + # the app's query engine). + self._usage_db_path_override: Path | None = None @property def _alert_rules_path(self) -> Path: @@ -206,8 +293,25 @@ def _alert_rules_path(self) -> Path: @property def _conn(self) -> duckdb.DuckDBPyConnection: + if self._conn_provider is None: + raise RuntimeError( + "EmbeddedControlPlaneStore was constructed without a " + "conn_provider; webhook/alert/outbox methods are unavailable." + ) return self._conn_provider() + @property + def _usage_db_path(self) -> Path: + if self._usage_db_path_override is not None: + return self._usage_db_path_override + if self._usage_db_path_provider is None: + raise RuntimeError( + "EmbeddedControlPlaneStore was constructed without a " + "usage_db_path_provider; usage/session methods are " + "unavailable." + ) + return Path(self._usage_db_path_provider()) + # --- webhook durable delivery queue -------------------------------------- def enqueue_webhook_delivery( @@ -874,3 +978,504 @@ def get_dead_letter_stats(self, tenant_id: str) -> dict: for hour, count in trend_rows ], } + + # --- API usage accounting ------------------------------------------------- + + def ensure_usage_schema(self) -> None: + # Moved verbatim from auth/usage_table.py's ensure_usage_table + # (ADR 0010 slice 4), including the Windows file-lock fallback: if + # the configured usage db path can't be opened, fall back to a + # per-process temp file and stick with it for this store's lifetime. + for attempt in range(10): + try: + conn = connect_duckdb(self._usage_db_path) + except duckdb.IOException as exc: + if ( + os.getenv("AGENTFLOW_USAGE_DB_PATH") is None + and self._usage_db_path.name == "agentflow_api.duckdb" + ): + fallback_path = ( + Path(os.getenv("TEMP", ".")) + / f"agentflow_api_{os.getpid()}_{time.time_ns()}.duckdb" + ) + logger.warning( + "usage_db_path_fallback", + original=str(self._usage_db_path), + fallback=str(fallback_path), + error=str(exc), + ) + self._usage_db_path_override = fallback_path + conn = connect_duckdb(self._usage_db_path) + else: + if attempt == 9: + raise + time.sleep(0.01 * (attempt + 1)) + continue + except duckdb.Error: + if attempt == 9: + raise + time.sleep(0.01 * (attempt + 1)) + continue + + try: + ensure_api_usage_table(conn) + return + except duckdb.Error: + if attempt == 9: + raise + time.sleep(0.01 * (attempt + 1)) + finally: + conn.close() + + def record_api_usage( + self, + *, + tenant: str, + key_name: str, + endpoint: str, + key_id: str | None, + key_slot: str, + ) -> None: + for attempt in range(10): + try: + conn = connect_duckdb(self._usage_db_path) + except duckdb.Error: + if attempt == 9: + raise + time.sleep(0.01 * (attempt + 1)) + continue + + try: + conn.execute( + """ + INSERT INTO api_usage (tenant, key_name, endpoint, key_id, key_slot) + VALUES (?, ?, ?, ?, ?) + """, + [tenant, key_name, endpoint, key_id, key_slot], + ) + return + except duckdb.Error: + if attempt == 9: + raise + time.sleep(0.01 * (attempt + 1)) + finally: + conn.close() + + def get_usage_by_tenant(self) -> list[dict]: + conn = connect_duckdb(self._usage_db_path) + try: + rows = conn.execute( + """ + SELECT tenant, COUNT(*) AS requests_last_24h + FROM api_usage + WHERE ts >= CURRENT_TIMESTAMP - INTERVAL '24 hours' + GROUP BY tenant + ORDER BY tenant + """ + ).fetchall() + finally: + conn.close() + return [ + {"tenant": tenant, "requests_last_24h": requests_last_24h} + for tenant, requests_last_24h in rows + ] + + def get_usage_by_key(self) -> dict[tuple[str, str], int]: + for attempt in range(10): + try: + conn = connect_duckdb(self._usage_db_path) + except duckdb.Error: + if attempt == 9: + raise + time.sleep(0.01 * (attempt + 1)) + continue + + try: + rows = conn.execute( + """ + SELECT tenant, key_name, COUNT(*) AS requests_last_24h + FROM api_usage + WHERE ts >= CURRENT_TIMESTAMP - INTERVAL '24 hours' + GROUP BY tenant, key_name + """ + ).fetchall() + return { + (tenant, key_name): requests_last_24h + for tenant, key_name, requests_last_24h in rows + } + except duckdb.Error: + if attempt == 9: + raise + time.sleep(0.01 * (attempt + 1)) + finally: + conn.close() + return {} + + def get_old_key_usage_by_key_id(self) -> dict[str, int]: + for attempt in range(10): + try: + conn = connect_duckdb(self._usage_db_path) + except duckdb.Error: + if attempt == 9: + raise + time.sleep(0.01 * (attempt + 1)) + continue + + try: + rows = conn.execute( + """ + SELECT key_id, COUNT(*) AS requests_last_hour + FROM api_usage + WHERE key_slot = 'previous' + AND ts >= CURRENT_TIMESTAMP - INTERVAL '1 hour' + AND key_id IS NOT NULL + GROUP BY key_id + """ + ).fetchall() + return dict(rows) + except duckdb.Error: + if attempt == 9: + raise + time.sleep(0.01 * (attempt + 1)) + finally: + conn.close() + return {} + + # --- API session analytics ------------------------------------------------ + + def record_api_session(self, request_id: str, record: dict) -> None: + for attempt in range(10): + try: + conn = connect_duckdb(self._usage_db_path) + except duckdb.Error as exc: + if attempt == 9: + logger.warning( + "analytics_session_write_skipped", + stage="connect", + db_path=str(self._usage_db_path), + request_id=request_id, + tenant=record.get("tenant"), + endpoint=record.get("endpoint"), + attempts=attempt + 1, + error=str(exc), + exc_info=True, + ) + return + time.sleep(0.01 * (attempt + 1)) + continue + + try: + conn.execute( + """ + INSERT OR REPLACE INTO api_sessions ( + request_id, + tenant, + key_name, + endpoint, + method, + status_code, + duration_ms, + cache_hit, + entity_type, + entity_id, + metric_name, + query_engine, + query_text + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + [ + request_id, + record["tenant"], + record["key_name"], + record["endpoint"], + record["method"], + record["status_code"], + record["duration_ms"], + record["cache_hit"], + record["entity_type"], + record["entity_id"], + record["metric_name"], + record["query_engine"], + record["query_text"], + ], + ) + return + except duckdb.Error as exc: + if attempt == 9: + logger.warning( + "analytics_session_write_skipped", + stage="insert", + db_path=str(self._usage_db_path), + request_id=request_id, + tenant=record.get("tenant"), + endpoint=record.get("endpoint"), + attempts=attempt + 1, + error=str(exc), + exc_info=True, + ) + return + time.sleep(0.01 * (attempt + 1)) + finally: + conn.close() + + def get_usage_analytics(self, *, window: str = "24h", tenant: str | None = None) -> dict: + interval = _window_to_interval(window) + conn = connect_duckdb(self._usage_db_path) + try: + ensure_api_sessions_table(conn) + if tenant: + rows = conn.execute( + """ + SELECT + tenant, + COUNT(*) AS total_requests, + ROUND(AVG(CASE WHEN status_code >= 400 THEN 1.0 ELSE 0.0 END), 4) + AS error_rate, + ROUND(AVG(CASE WHEN cache_hit THEN 1.0 ELSE 0.0 END), 4) + AS cache_hit_rate, + ROUND(AVG(duration_ms), 3) AS avg_duration_ms + FROM api_sessions + WHERE tenant IS NOT NULL + AND ts >= CURRENT_TIMESTAMP - CAST(? AS INTERVAL) + AND tenant = ? + GROUP BY tenant + ORDER BY tenant + """, + [interval, tenant], + ).fetchall() + else: + rows = conn.execute( + """ + SELECT + tenant, + COUNT(*) AS total_requests, + ROUND(AVG(CASE WHEN status_code >= 400 THEN 1.0 ELSE 0.0 END), 4) + AS error_rate, + ROUND(AVG(CASE WHEN cache_hit THEN 1.0 ELSE 0.0 END), 4) + AS cache_hit_rate, + ROUND(AVG(duration_ms), 3) AS avg_duration_ms + FROM api_sessions + WHERE tenant IS NOT NULL + AND ts >= CURRENT_TIMESTAMP - CAST(? AS INTERVAL) + GROUP BY tenant + ORDER BY tenant + """, + [interval], + ).fetchall() + tenants = [] + for tenant_name, total_requests, error_rate, cache_hit_rate, avg_duration_ms in rows: + top_endpoints = conn.execute( + """ + SELECT endpoint + FROM api_sessions + WHERE tenant = ? + AND ts >= CURRENT_TIMESTAMP - CAST(? AS INTERVAL) + GROUP BY endpoint + ORDER BY COUNT(*) DESC, endpoint + LIMIT 3 + """, + [tenant_name, interval], + ).fetchall() + tenants.append( + { + "tenant": tenant_name, + "total_requests": total_requests, + "error_rate": float(error_rate or 0.0), + "cache_hit_rate": float(cache_hit_rate or 0.0), + "top_endpoints": [item[0] for item in top_endpoints], + "avg_duration_ms": float(avg_duration_ms or 0.0), + } + ) + return {"window": window, "tenants": tenants} + finally: + conn.close() + + def get_top_queries(self, *, limit: int = 10, window: str = "24h") -> dict: + interval = _window_to_interval(window) + conn = connect_duckdb(self._usage_db_path) + try: + ensure_api_sessions_table(conn) + rows = conn.execute( + """ + SELECT query_text, COUNT(*) AS frequency + FROM api_sessions + WHERE query_text IS NOT NULL + AND ts >= CURRENT_TIMESTAMP - CAST(? AS INTERVAL) + GROUP BY query_text + ORDER BY frequency DESC, query_text + LIMIT ? + """, + [interval, limit], + ).fetchall() + return { + "window": window, + "queries": [ + {"query": query_text, "count": frequency} for query_text, frequency in rows + ], + } + finally: + conn.close() + + def get_top_entities(self, *, limit: int = 10, window: str = "24h") -> dict: + interval = _window_to_interval(window) + conn = connect_duckdb(self._usage_db_path) + try: + ensure_api_sessions_table(conn) + rows = conn.execute( + """ + SELECT entity_type, entity_id, COUNT(*) AS frequency + FROM api_sessions + WHERE entity_id IS NOT NULL + AND ts >= CURRENT_TIMESTAMP - CAST(? AS INTERVAL) + GROUP BY entity_type, entity_id + ORDER BY frequency DESC, entity_type, entity_id + LIMIT ? + """, + [interval, limit], + ).fetchall() + return { + "window": window, + "entities": [ + { + "entity_type": entity_type, + "entity_id": entity_id, + "count": frequency, + } + for entity_type, entity_id, frequency in rows + ], + } + finally: + conn.close() + + def get_latency_analytics(self, *, window: str = "24h") -> dict: + interval = _window_to_interval(window) + conn = connect_duckdb(self._usage_db_path) + try: + ensure_api_sessions_table(conn) + rows = conn.execute( + """ + SELECT + endpoint, + COUNT(*) AS requests, + ROUND(quantile_cont(duration_ms, 0.50), 3) AS p50_ms, + ROUND(quantile_cont(duration_ms, 0.95), 3) AS p95_ms, + ROUND(quantile_cont(duration_ms, 0.99), 3) AS p99_ms + FROM api_sessions + WHERE ts >= CURRENT_TIMESTAMP - CAST(? AS INTERVAL) + GROUP BY endpoint + ORDER BY endpoint + """, + [interval], + ).fetchall() + return { + "window": window, + "endpoints": [ + { + "endpoint": endpoint, + "requests": requests, + "p50_ms": float(p50_ms or 0.0), + "p95_ms": float(p95_ms or 0.0), + "p99_ms": float(p99_ms or 0.0), + } + for endpoint, requests, p50_ms, p95_ms, p99_ms in rows + ], + } + finally: + conn.close() + + def get_anomalies(self, *, window: str = "24h") -> dict: + interval = _window_to_interval(window) + conn = connect_duckdb(self._usage_db_path) + try: + ensure_api_sessions_table(conn) + rows = conn.execute( + """ + WITH hourly AS ( + SELECT + tenant, + date_trunc('hour', ts) AS hour_bucket, + COUNT(*) AS requests + FROM api_sessions + WHERE tenant IS NOT NULL + AND ts >= CURRENT_TIMESTAMP - CAST(? AS INTERVAL) + GROUP BY tenant, hour_bucket + ), + latest AS ( + SELECT tenant, MAX(hour_bucket) AS current_hour + FROM hourly + GROUP BY tenant + ), + current_hour AS ( + SELECT + hourly.tenant, + hourly.hour_bucket, + hourly.requests AS current_hour_requests + FROM hourly + JOIN latest + ON latest.tenant = hourly.tenant + AND latest.current_hour = hourly.hour_bucket + ), + historical AS ( + SELECT + current_hour.tenant, + ROUND(AVG(hourly.requests), 1) AS hourly_average + FROM current_hour + JOIN hourly + ON hourly.tenant = current_hour.tenant + AND hourly.hour_bucket < current_hour.hour_bucket + GROUP BY current_hour.tenant + ), + scored AS ( + SELECT + current_hour.tenant, + current_hour.current_hour_requests, + historical.hourly_average, + ROUND( + current_hour.current_hour_requests + / NULLIF(historical.hourly_average, 0), + 2 + ) AS spike_ratio + FROM current_hour + JOIN historical + ON historical.tenant = current_hour.tenant + ) + SELECT tenant, current_hour_requests, hourly_average, spike_ratio + FROM scored + WHERE spike_ratio > 3 + ORDER BY spike_ratio DESC, tenant + """, + [interval], + ).fetchall() + return { + "window": window, + "anomalies": [ + { + "tenant": tenant, + "current_hour_requests": current_hour_requests, + "hourly_average": float(hourly_average or 0.0), + "spike_ratio": float(spike_ratio or 0.0), + } + for tenant, current_hour_requests, hourly_average, spike_ratio in rows + ], + } + finally: + conn.close() + + def get_queries_per_second_last_minute(self) -> float: + conn = connect_duckdb(self._usage_db_path) + try: + ensure_api_sessions_table(conn) + row = conn.execute( + """ + SELECT COUNT(*) + FROM api_sessions + WHERE ts >= CURRENT_TIMESTAMP - INTERVAL '1 minute' + """ + ).fetchone() + requests_last_minute = row[0] if row else 0 + except duckdb.Error: + return 0.0 + finally: + conn.close() + return round(float(requests_last_minute) / 60.0, 2) diff --git a/src/serving/control_plane/store.py b/src/serving/control_plane/store.py index ad46cc86..9025e7e5 100644 --- a/src/serving/control_plane/store.py +++ b/src/serving/control_plane/store.py @@ -48,6 +48,24 @@ ``CREATE TABLE IF NOT EXISTS`` inside these methods would silently recreate a table a test dropped mid-scenario to simulate a transaction failure, defeating that fault injection. + +Slice 4 covers API-usage accounting (``api_usage``: per-tenant/per-key +request counters, including the rotation status endpoint's "requests on the +old key" query) and API-session analytics (``api_sessions``: per-request +latency/entity/query telemetry powering the admin usage dashboards). Unlike +every store above, this state was NEVER on ``query_engine._conn`` — not even +in the outbox's ``:memory:`` special case — because ``AuthManager.db_path`` +resolves to its own DuckDB file, independent of ``DUCKDB_PATH``, even when +the query engine runs on a file backend (see ``AuthManager.__init__``'s +sibling-path derivation). The embedded adapter preserves that separateness: +its usage/session methods open dedicated connections against +``usage_db_path_provider`` rather than the shared ``conn_provider``. The +scope turned out wider than "usage_table.py + analytics.py": ``KeyRotator`` +(``key_rotation.py``) queries ``api_usage`` directly for old-key-usage +stats, and the admin dashboard (``admin_ui.py``) queries ``api_sessions`` +directly for its QPS tile — both bypassed the port until this slice and are +covered here too, or a PostgreSQL swap (slice 5) would leave two call sites +still hard-wired to a local DuckDB file. """ from __future__ import annotations @@ -299,6 +317,81 @@ def get_dead_letter_stats(self, tenant_id: str) -> dict: """``{"counts": {reason: count}, "last_24h": int, "trend": [...]}`` for one tenant's active (``failed``) dead-letter events.""" + # --- API usage accounting (per-tenant/per-key request counters) ---------- + + @abstractmethod + def ensure_usage_schema(self) -> None: + """Idempotently create the ``api_usage`` table. Called once at + ``AuthManager`` construction — not lazily inside the methods below + (mirrors the pre-port ``ensure_usage_table`` call site, main.py's + lifespan, verbatim).""" + + @abstractmethod + def record_api_usage( + self, + *, + tenant: str, + key_name: str, + endpoint: str, + key_id: str | None, + key_slot: str, + ) -> None: + """Append one ``api_usage`` row for a completed authenticated + request. Raises on exhausted retries — a caller (``record_usage``) + depends on the exception to skip its post-insert audit publish.""" + + @abstractmethod + def get_usage_by_tenant(self) -> list[dict]: + """``{"tenant": ..., "requests_last_24h": ...}`` per tenant.""" + + @abstractmethod + def get_usage_by_key(self) -> dict[tuple[str, str], int]: + """``{(tenant, key_name): requests_last_24h}`` for every active key.""" + + @abstractmethod + def get_old_key_usage_by_key_id(self) -> dict[str, int]: + """``{key_id: requests_last_hour}`` for requests served on a + rotated-out (``previous``) key slot — powers the rotation-status + endpoint's "requests on the old key" figure.""" + + # --- API session analytics (per-request latency/entity/query log) ------- + + @abstractmethod + def record_api_session(self, request_id: str, record: dict) -> None: + """Insert-or-replace one ``api_sessions`` row, keyed on + ``request_id`` (idempotent — a retried background write must not + double-count). Best-effort: logs and returns on exhausted retries + rather than raising, unlike ``record_api_usage``.""" + + @abstractmethod + def get_usage_analytics(self, *, window: str = "24h", tenant: str | None = None) -> dict: + """Per-tenant request volume/error-rate/cache-hit-rate/latency and + top endpoints over ``window`` (e.g. ``"15m"``, ``"1h"``, ``"7d"``).""" + + @abstractmethod + def get_top_queries(self, *, limit: int = 10, window: str = "24h") -> dict: + """Most frequent ``/v1/query`` question texts over ``window``.""" + + @abstractmethod + def get_top_entities(self, *, limit: int = 10, window: str = "24h") -> dict: + """Most frequently fetched ``(entity_type, entity_id)`` pairs over + ``window``.""" + + @abstractmethod + def get_latency_analytics(self, *, window: str = "24h") -> dict: + """Per-endpoint request count and p50/p95/p99 latency over + ``window``.""" + + @abstractmethod + def get_anomalies(self, *, window: str = "24h") -> dict: + """Per-tenant hours whose request volume spikes >3x their own + historical hourly average within ``window``.""" + + @abstractmethod + def get_queries_per_second_last_minute(self) -> float: + """Average QPS over the trailing 60s of ``api_sessions`` rows — + powers the admin-UI live dashboard tile.""" + def get_control_plane_store(app: FastAPI) -> ControlPlaneStore: """Resolve the app's control-plane store, creating the configured one on diff --git a/tests/integration/test_rotation.py b/tests/integration/test_rotation.py index 6917d576..f5cbb3c3 100644 --- a/tests/integration/test_rotation.py +++ b/tests/integration/test_rotation.py @@ -1,256 +1,259 @@ -import time -from pathlib import Path - -import duckdb -import pytest -from fastapi.testclient import TestClient - -import src.serving.api.auth.key_rotation as key_rotation_module -from src.serving.api.auth import AuthManager -from src.serving.api.main import app - -pytestmark = pytest.mark.integration - - -@pytest.fixture -def client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): - monkeypatch.setenv("DUCKDB_PATH", str(tmp_path / "pipeline.duckdb")) - monkeypatch.setenv("AGENTFLOW_USAGE_DB_PATH", str(tmp_path / "usage.duckdb")) - monkeypatch.setenv("AGENTFLOW_ROTATION_GRACE_PERIOD_SECONDS", "30") - - api_keys_path = tmp_path / "config" / "api_keys.yaml" - api_keys_path.parent.mkdir(parents=True, exist_ok=True) - api_keys_path.write_text( - ( - "keys:\n" - ' - key: "rotation-acme-key"\n' - ' name: "Rotation Agent"\n' - ' tenant: "acme"\n' - " rate_limit_rpm: 100\n" - " allowed_entity_types: null\n" - ' created_at: "2026-04-10"\n' - ), - encoding="utf-8", - newline="\n", - ) - - with TestClient(app) as c: - manager = AuthManager( - api_keys_path=api_keys_path, - db_path=tmp_path / "usage.duckdb", - admin_key="admin-secret", - ) - manager.load() - manager.ensure_usage_table() - c.app.state.auth_manager = manager - yield c - manager.shutdown() - - -@pytest.fixture -def expiring_client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): - monkeypatch.setenv("DUCKDB_PATH", str(tmp_path / "pipeline.duckdb")) - monkeypatch.setenv("AGENTFLOW_USAGE_DB_PATH", str(tmp_path / "usage.duckdb")) - monkeypatch.setenv("AGENTFLOW_ROTATION_GRACE_PERIOD_SECONDS", "1") - - api_keys_path = tmp_path / "config" / "api_keys.yaml" - api_keys_path.parent.mkdir(parents=True, exist_ok=True) - api_keys_path.write_text( - ( - "keys:\n" - ' - key: "rotation-acme-key"\n' - ' name: "Rotation Agent"\n' - ' tenant: "acme"\n' - " rate_limit_rpm: 100\n" - " allowed_entity_types: null\n" - ' created_at: "2026-04-10"\n' - ), - encoding="utf-8", - newline="\n", - ) - - with TestClient(app) as c: - manager = AuthManager( - api_keys_path=api_keys_path, - db_path=tmp_path / "usage.duckdb", - admin_key="admin-secret", - ) - manager.load() - manager.ensure_usage_table() - c.app.state.auth_manager = manager - yield c - manager.shutdown() - - -def test_rotation_endpoints_require_admin_key(client: TestClient): - key_id = client.app.state.auth_manager.list_keys_with_usage()[0]["key_id"] - - rotate = client.post(f"/v1/admin/keys/{key_id}/rotate") - status_response = client.get(f"/v1/admin/keys/{key_id}/rotation-status") - revoke = client.post(f"/v1/admin/keys/{key_id}/revoke-old") - - assert rotate.status_code == 401 - assert status_response.status_code == 401 - assert revoke.status_code == 401 - - -def test_rotation_status_is_idle_when_no_rotation_in_progress(client: TestClient): - key_id = client.app.state.auth_manager.list_keys_with_usage()[0]["key_id"] - - response = client.get( - f"/v1/admin/keys/{key_id}/rotation-status", - headers={"X-Admin-Key": "admin-secret"}, - ) - - assert response.status_code == 200 - assert response.json() == { - "phase": "idle", - "old_key_active_until": None, - "requests_on_old_key_last_hour": 0, - } - - -def test_rotate_accepts_old_and_new_keys_during_grace_period(client: TestClient): - key_id = client.app.state.auth_manager.list_keys_with_usage()[0]["key_id"] - - rotate = client.post( - f"/v1/admin/keys/{key_id}/rotate", - headers={"X-Admin-Key": "admin-secret"}, - ) - - assert rotate.status_code == 200 - payload = rotate.json() - assert payload["new_key"] != "rotation-acme-key" - assert payload["expires_at"] - - old_response = client.get("/v1/metrics/revenue", headers={"X-API-Key": "rotation-acme-key"}) - new_response = client.get("/v1/metrics/revenue", headers={"X-API-Key": payload["new_key"]}) - - assert old_response.status_code == 200 - assert new_response.status_code == 200 - - -def test_rotation_status_retries_transient_usage_db_lock( - client: TestClient, - monkeypatch: pytest.MonkeyPatch, -): - manager = client.app.state.auth_manager - key_id = manager.list_keys_with_usage()[0]["key_id"] - real_connect = key_rotation_module.duckdb.connect - attempts = 0 - - def flaky_connect(path: str): - nonlocal attempts - attempts += 1 - if attempts == 1: - raise duckdb.IOException("usage db is locked") - return real_connect(path) - - monkeypatch.setattr(key_rotation_module.duckdb, "connect", flaky_connect) - - response = client.get( - f"/v1/admin/keys/{key_id}/rotation-status", - headers={"X-Admin-Key": "admin-secret"}, - ) - - assert response.status_code == 200 - assert response.json()["requests_on_old_key_last_hour"] == 0 - assert attempts >= 2 - - -def test_rotation_status_counts_old_key_requests(client: TestClient): - key_id = client.app.state.auth_manager.list_keys_with_usage()[0]["key_id"] - rotate = client.post( - f"/v1/admin/keys/{key_id}/rotate", - headers={"X-Admin-Key": "admin-secret"}, - ) - new_key = rotate.json()["new_key"] - - client.get("/v1/metrics/revenue", headers={"X-API-Key": "rotation-acme-key"}) - client.get("/v1/entity/order/ORD-20260412-0001", headers={"X-API-Key": "rotation-acme-key"}) - client.get("/v1/metrics/revenue", headers={"X-API-Key": new_key}) - - response = client.get( - f"/v1/admin/keys/{key_id}/rotation-status", - headers={"X-Admin-Key": "admin-secret"}, - ) - - assert response.status_code == 200 - assert response.json()["phase"] == "grace_period" - assert response.json()["old_key_active_until"] is not None - assert response.json()["requests_on_old_key_last_hour"] == 2 - - -def test_revoke_old_key_invalidates_only_previous_key(client: TestClient): - key_id = client.app.state.auth_manager.list_keys_with_usage()[0]["key_id"] - rotate = client.post( - f"/v1/admin/keys/{key_id}/rotate", - headers={"X-Admin-Key": "admin-secret"}, - ) - new_key = rotate.json()["new_key"] - - revoke = client.post( - f"/v1/admin/keys/{key_id}/revoke-old", - headers={"X-Admin-Key": "admin-secret"}, - ) - old_response = client.get("/v1/metrics/revenue", headers={"X-API-Key": "rotation-acme-key"}) - new_response = client.get("/v1/metrics/revenue", headers={"X-API-Key": new_key}) - - assert revoke.status_code == 200 - assert revoke.json() == {"revoked": True} - assert old_response.status_code == 401 - assert new_response.status_code == 200 - - -def test_rotate_rejects_second_rotation_during_grace_period(client: TestClient): - key_id = client.app.state.auth_manager.list_keys_with_usage()[0]["key_id"] - - first = client.post( - f"/v1/admin/keys/{key_id}/rotate", - headers={"X-Admin-Key": "admin-secret"}, - ) - second = client.post( - f"/v1/admin/keys/{key_id}/rotate", - headers={"X-Admin-Key": "admin-secret"}, - ) - - assert first.status_code == 200 - assert second.status_code == 409 - - -def test_expired_grace_period_revokes_old_key_automatically(expiring_client: TestClient): - key_id = expiring_client.app.state.auth_manager.list_keys_with_usage()[0]["key_id"] - rotate = expiring_client.post( - f"/v1/admin/keys/{key_id}/rotate", - headers={"X-Admin-Key": "admin-secret"}, - ) - new_key = rotate.json()["new_key"] - - deadline = time.monotonic() + 3 - status_response = expiring_client.get( - f"/v1/admin/keys/{key_id}/rotation-status", - headers={"X-Admin-Key": "admin-secret"}, - ) - while time.monotonic() < deadline: - status_response = expiring_client.get( - f"/v1/admin/keys/{key_id}/rotation-status", - headers={"X-Admin-Key": "admin-secret"}, - ) - if status_response.status_code == 200 and status_response.json()["phase"] == "idle": - break - time.sleep(0.1) - - old_response = expiring_client.get( - "/v1/metrics/revenue", - headers={"X-API-Key": "rotation-acme-key"}, - ) - new_response = expiring_client.get( - "/v1/metrics/revenue", - headers={"X-API-Key": new_key}, - ) - - assert rotate.status_code == 200 - assert old_response.status_code == 401 - assert status_response.status_code == 200 - assert status_response.json()["phase"] == "idle" - assert new_response.status_code == 200 +import time +from pathlib import Path + +import duckdb +import pytest +from fastapi.testclient import TestClient + +import src.serving.control_plane.embedded as embedded_module +from src.serving.api.auth import AuthManager +from src.serving.api.main import app + +pytestmark = pytest.mark.integration + + +@pytest.fixture +def client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + monkeypatch.setenv("DUCKDB_PATH", str(tmp_path / "pipeline.duckdb")) + monkeypatch.setenv("AGENTFLOW_USAGE_DB_PATH", str(tmp_path / "usage.duckdb")) + monkeypatch.setenv("AGENTFLOW_ROTATION_GRACE_PERIOD_SECONDS", "30") + + api_keys_path = tmp_path / "config" / "api_keys.yaml" + api_keys_path.parent.mkdir(parents=True, exist_ok=True) + api_keys_path.write_text( + ( + "keys:\n" + ' - key: "rotation-acme-key"\n' + ' name: "Rotation Agent"\n' + ' tenant: "acme"\n' + " rate_limit_rpm: 100\n" + " allowed_entity_types: null\n" + ' created_at: "2026-04-10"\n' + ), + encoding="utf-8", + newline="\n", + ) + + with TestClient(app) as c: + manager = AuthManager( + api_keys_path=api_keys_path, + db_path=tmp_path / "usage.duckdb", + admin_key="admin-secret", + ) + manager.load() + manager.ensure_usage_table() + c.app.state.auth_manager = manager + yield c + manager.shutdown() + + +@pytest.fixture +def expiring_client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + monkeypatch.setenv("DUCKDB_PATH", str(tmp_path / "pipeline.duckdb")) + monkeypatch.setenv("AGENTFLOW_USAGE_DB_PATH", str(tmp_path / "usage.duckdb")) + monkeypatch.setenv("AGENTFLOW_ROTATION_GRACE_PERIOD_SECONDS", "1") + + api_keys_path = tmp_path / "config" / "api_keys.yaml" + api_keys_path.parent.mkdir(parents=True, exist_ok=True) + api_keys_path.write_text( + ( + "keys:\n" + ' - key: "rotation-acme-key"\n' + ' name: "Rotation Agent"\n' + ' tenant: "acme"\n' + " rate_limit_rpm: 100\n" + " allowed_entity_types: null\n" + ' created_at: "2026-04-10"\n' + ), + encoding="utf-8", + newline="\n", + ) + + with TestClient(app) as c: + manager = AuthManager( + api_keys_path=api_keys_path, + db_path=tmp_path / "usage.duckdb", + admin_key="admin-secret", + ) + manager.load() + manager.ensure_usage_table() + c.app.state.auth_manager = manager + yield c + manager.shutdown() + + +def test_rotation_endpoints_require_admin_key(client: TestClient): + key_id = client.app.state.auth_manager.list_keys_with_usage()[0]["key_id"] + + rotate = client.post(f"/v1/admin/keys/{key_id}/rotate") + status_response = client.get(f"/v1/admin/keys/{key_id}/rotation-status") + revoke = client.post(f"/v1/admin/keys/{key_id}/revoke-old") + + assert rotate.status_code == 401 + assert status_response.status_code == 401 + assert revoke.status_code == 401 + + +def test_rotation_status_is_idle_when_no_rotation_in_progress(client: TestClient): + key_id = client.app.state.auth_manager.list_keys_with_usage()[0]["key_id"] + + response = client.get( + f"/v1/admin/keys/{key_id}/rotation-status", + headers={"X-Admin-Key": "admin-secret"}, + ) + + assert response.status_code == 200 + assert response.json() == { + "phase": "idle", + "old_key_active_until": None, + "requests_on_old_key_last_hour": 0, + } + + +def test_rotate_accepts_old_and_new_keys_during_grace_period(client: TestClient): + key_id = client.app.state.auth_manager.list_keys_with_usage()[0]["key_id"] + + rotate = client.post( + f"/v1/admin/keys/{key_id}/rotate", + headers={"X-Admin-Key": "admin-secret"}, + ) + + assert rotate.status_code == 200 + payload = rotate.json() + assert payload["new_key"] != "rotation-acme-key" + assert payload["expires_at"] + + old_response = client.get("/v1/metrics/revenue", headers={"X-API-Key": "rotation-acme-key"}) + new_response = client.get("/v1/metrics/revenue", headers={"X-API-Key": payload["new_key"]}) + + assert old_response.status_code == 200 + assert new_response.status_code == 200 + + +def test_rotation_status_retries_transient_usage_db_lock( + client: TestClient, + monkeypatch: pytest.MonkeyPatch, +): + manager = client.app.state.auth_manager + key_id = manager.list_keys_with_usage()[0]["key_id"] + # ADR 0010 slice 4: the usage-stat connect now happens in + # EmbeddedControlPlaneStore (control_plane/embedded.py), not + # key_rotation.py directly. + real_connect = embedded_module.duckdb.connect + attempts = 0 + + def flaky_connect(path: str): + nonlocal attempts + attempts += 1 + if attempts == 1: + raise duckdb.IOException("usage db is locked") + return real_connect(path) + + monkeypatch.setattr(embedded_module.duckdb, "connect", flaky_connect) + + response = client.get( + f"/v1/admin/keys/{key_id}/rotation-status", + headers={"X-Admin-Key": "admin-secret"}, + ) + + assert response.status_code == 200 + assert response.json()["requests_on_old_key_last_hour"] == 0 + assert attempts >= 2 + + +def test_rotation_status_counts_old_key_requests(client: TestClient): + key_id = client.app.state.auth_manager.list_keys_with_usage()[0]["key_id"] + rotate = client.post( + f"/v1/admin/keys/{key_id}/rotate", + headers={"X-Admin-Key": "admin-secret"}, + ) + new_key = rotate.json()["new_key"] + + client.get("/v1/metrics/revenue", headers={"X-API-Key": "rotation-acme-key"}) + client.get("/v1/entity/order/ORD-20260412-0001", headers={"X-API-Key": "rotation-acme-key"}) + client.get("/v1/metrics/revenue", headers={"X-API-Key": new_key}) + + response = client.get( + f"/v1/admin/keys/{key_id}/rotation-status", + headers={"X-Admin-Key": "admin-secret"}, + ) + + assert response.status_code == 200 + assert response.json()["phase"] == "grace_period" + assert response.json()["old_key_active_until"] is not None + assert response.json()["requests_on_old_key_last_hour"] == 2 + + +def test_revoke_old_key_invalidates_only_previous_key(client: TestClient): + key_id = client.app.state.auth_manager.list_keys_with_usage()[0]["key_id"] + rotate = client.post( + f"/v1/admin/keys/{key_id}/rotate", + headers={"X-Admin-Key": "admin-secret"}, + ) + new_key = rotate.json()["new_key"] + + revoke = client.post( + f"/v1/admin/keys/{key_id}/revoke-old", + headers={"X-Admin-Key": "admin-secret"}, + ) + old_response = client.get("/v1/metrics/revenue", headers={"X-API-Key": "rotation-acme-key"}) + new_response = client.get("/v1/metrics/revenue", headers={"X-API-Key": new_key}) + + assert revoke.status_code == 200 + assert revoke.json() == {"revoked": True} + assert old_response.status_code == 401 + assert new_response.status_code == 200 + + +def test_rotate_rejects_second_rotation_during_grace_period(client: TestClient): + key_id = client.app.state.auth_manager.list_keys_with_usage()[0]["key_id"] + + first = client.post( + f"/v1/admin/keys/{key_id}/rotate", + headers={"X-Admin-Key": "admin-secret"}, + ) + second = client.post( + f"/v1/admin/keys/{key_id}/rotate", + headers={"X-Admin-Key": "admin-secret"}, + ) + + assert first.status_code == 200 + assert second.status_code == 409 + + +def test_expired_grace_period_revokes_old_key_automatically(expiring_client: TestClient): + key_id = expiring_client.app.state.auth_manager.list_keys_with_usage()[0]["key_id"] + rotate = expiring_client.post( + f"/v1/admin/keys/{key_id}/rotate", + headers={"X-Admin-Key": "admin-secret"}, + ) + new_key = rotate.json()["new_key"] + + deadline = time.monotonic() + 3 + status_response = expiring_client.get( + f"/v1/admin/keys/{key_id}/rotation-status", + headers={"X-Admin-Key": "admin-secret"}, + ) + while time.monotonic() < deadline: + status_response = expiring_client.get( + f"/v1/admin/keys/{key_id}/rotation-status", + headers={"X-Admin-Key": "admin-secret"}, + ) + if status_response.status_code == 200 and status_response.json()["phase"] == "idle": + break + time.sleep(0.1) + + old_response = expiring_client.get( + "/v1/metrics/revenue", + headers={"X-API-Key": "rotation-acme-key"}, + ) + new_response = expiring_client.get( + "/v1/metrics/revenue", + headers={"X-API-Key": new_key}, + ) + + assert rotate.status_code == 200 + assert old_response.status_code == 401 + assert status_response.status_code == 200 + assert status_response.json()["phase"] == "idle" + assert new_response.status_code == 200 diff --git a/tests/unit/test_audit_publisher.py b/tests/unit/test_audit_publisher.py index 09a25d76..69628b91 100644 --- a/tests/unit/test_audit_publisher.py +++ b/tests/unit/test_audit_publisher.py @@ -112,9 +112,9 @@ def test_record_usage_skips_publish_when_all_inserts_fail( import duckdb - from src.serving.api.auth import usage_table as usage_table_module + from src.serving.control_plane import embedded as embedded_module - real_connect = usage_table_module.connect_duckdb + real_connect = embedded_module.connect_duckdb call_count = {"n": 0} class _InsertFailingConn: @@ -133,9 +133,11 @@ def _always_fail_insert(db_path): # type: ignore[no-untyped-def] call_count["n"] += 1 return _InsertFailingConn(real_connect(db_path)) - monkeypatch.setattr(usage_table_module, "connect_duckdb", _always_fail_insert) + # ADR 0010 slice 4: record_usage's retry-loop-with-connect now lives in + # EmbeddedControlPlaneStore.record_api_usage, not usage_table.py. + monkeypatch.setattr(embedded_module, "connect_duckdb", _always_fail_insert) # Also speed up the retry loop so the test does not spend ~0.55s sleeping. - monkeypatch.setattr(usage_table_module.time, "sleep", lambda _seconds: None) + monkeypatch.setattr(embedded_module.time, "sleep", lambda _seconds: None) with pytest.raises(duckdb.Error): record_usage(manager, tenant_key, "/v1/entity/order") diff --git a/tests/unit/test_key_rotation.py b/tests/unit/test_key_rotation.py index 30ec859e..01818d0f 100644 --- a/tests/unit/test_key_rotation.py +++ b/tests/unit/test_key_rotation.py @@ -14,6 +14,7 @@ import pytest import src.serving.api.auth.key_rotation as key_rotation_module +import src.serving.control_plane.embedded as embedded_module from src.serving.api.auth.key_rotation import rotate_all_keys from src.serving.api.auth.manager import AuthManager, KeyCreateRequest, TenantKey @@ -245,7 +246,9 @@ def test_schedule_cleanup_noops_when_already_expired(self, manager: AuthManager) def test_usage_query_retries_on_transient_duckdb_error( self, manager: AuthManager, monkeypatch: pytest.MonkeyPatch ) -> None: - real_connect = key_rotation_module.connect_duckdb + # ADR 0010 slice 4: the retry-loop-with-connect for these usage-stat + # queries now lives in EmbeddedControlPlaneStore, not key_rotation.py. + real_connect = embedded_module.connect_duckdb calls = {"n": 0} def flaky_connect(path: object) -> object: @@ -254,10 +257,10 @@ def flaky_connect(path: object) -> object: raise duckdb.Error("database is locked") return real_connect(path) - monkeypatch.setattr(key_rotation_module, "connect_duckdb", flaky_connect) + monkeypatch.setattr(embedded_module, "connect_duckdb", flaky_connect) # list_keys_with_usage runs the usage queries; the first connect raises - # a transient error and the rotator must retry rather than propagate. + # a transient error and the store must retry rather than propagate. manager.list_keys_with_usage() assert calls["n"] >= 2 @@ -265,7 +268,7 @@ def flaky_connect(path: object) -> object: def test_old_key_usage_retries_on_transient_duckdb_error( self, manager: AuthManager, monkeypatch: pytest.MonkeyPatch ) -> None: - real_connect = key_rotation_module.connect_duckdb + real_connect = embedded_module.connect_duckdb calls = {"n": 0} def flaky_connect(path: object) -> object: @@ -274,7 +277,7 @@ def flaky_connect(path: object) -> object: raise duckdb.Error("database is locked") return real_connect(path) - monkeypatch.setattr(key_rotation_module, "connect_duckdb", flaky_connect) + monkeypatch.setattr(embedded_module, "connect_duckdb", flaky_connect) # old_key_usage_last_hour runs its own connect+query with the same retry # guard; a transient error on the first attempt must be retried. diff --git a/tests/unit/test_key_rotation_mutation.py b/tests/unit/test_key_rotation_mutation.py index 889de610..ac4639d2 100644 --- a/tests/unit/test_key_rotation_mutation.py +++ b/tests/unit/test_key_rotation_mutation.py @@ -15,12 +15,14 @@ fable_handoff.md cont.16-23): 1. **duckdb-free.** Importing ``serving.api.auth.key_rotation`` runs the auth - package ``__init__`` (``import duckdb`` + the key_rotation / usage_table - import chain) and key_rotation itself does ``import duckdb`` + - ``from ...duckdb_connection import connect_duckdb``. Real duckdb's lazy - ``_duckdb._sqltypes`` import crashes mutmut's coverage-instrumented stats - pass (the same break ci.yml works around with ``coverage run``). A fake - top-level ``duckdb`` module satisfies the import chain; the three usage-stat + package ``__init__`` (``import duckdb``) and, since ADR 0010 slice 4, + ``key_rotation`` -> ``manager`` -> ``control_plane`` -> ``embedded`` pulls in + ``import duckdb`` there instead (key_rotation itself no longer imports + duckdb directly — its usage-stat methods delegate to + ``AuthManager.store``). Real duckdb's lazy ``_duckdb._sqltypes`` import + crashes mutmut's coverage-instrumented stats pass (the same break ci.yml + works around with ``coverage run``). A fake top-level ``duckdb`` module + satisfies the import chain either way; the three usage-stat methods that actually *call* duckdb (``old_key_usage_by_key_id`` / ``_usage_by_key`` / ``old_key_usage_last_hour``) are stubbed on the rotator in the tests that need them, so their bodies stay uncovered and are NOT mutated.