diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2d50bf70..5526e1b2 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -177,6 +177,22 @@ jobs: CLICKHOUSE_USER: agentflow CLICKHOUSE_PASSWORD: agentflow CLICKHOUSE_DB: agentflow + postgres: + # Live coverage for PostgresControlPlaneStore (ADR 0010 slice 5); + # test_control_plane_postgres_live.py skips itself when + # AGENTFLOW_TEST_PG_DSN is absent. + image: postgres:17 + ports: + - 5432:5432 + env: + POSTGRES_USER: agentflow + POSTGRES_PASSWORD: agentflow + POSTGRES_DB: agentflow + options: >- + --health-cmd "pg_isready -U agentflow" + --health-interval 5s + --health-timeout 5s + --health-retries 10 steps: - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0 @@ -184,7 +200,7 @@ jobs: python-version: "3.11" - name: Install dependencies run: | - pip install -e ".[dev,cloud]" + pip install -e ".[dev,cloud,postgres]" pip install -e "./sdk" - name: Prepare pytest temp directory run: mkdir -p .tmp @@ -201,6 +217,7 @@ jobs: CLICKHOUSE_LIVE_USER: agentflow CLICKHOUSE_LIVE_PASSWORD: agentflow CLICKHOUSE_LIVE_DATABASE: agentflow + AGENTFLOW_TEST_PG_DSN: postgresql://agentflow:agentflow@localhost:5432/agentflow run: pytest tests/integration/ -v --tb=short helm-schema-live: diff --git a/CHANGELOG.md b/CHANGELOG.md index a297128a..97803efc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,63 @@ All notable changes to AgentFlow are documented in this file. ## [Unreleased] +### Added — PostgresControlPlaneStore: the scale profile ships (ADR 0010 slice 5, 2026-07-03) + +- **New `src/serving/control_plane/postgres.py`** — all six control-plane + state classes as PostgreSQL tables behind the existing port, with the claim + semantics the embedded adapter only satisfies degenerately made real: + enqueue-win by `INSERT .. ON CONFLICT DO NOTHING` rowcount, queue/outbox + claims by `FOR UPDATE SKIP LOCKED` + a self-expiring `lease_expires_at` + (work-stealing across replicas, no leader election; a crashed owner's rows + become due again on lease expiry), invariant 8 as an ordinary transaction + (every store method is one transaction: commit on success, rollback on any + exception). Payloads stay TEXT/JSON-string so callers see the embedded + adapter's shapes. One connection per call — pooling stays out of ADR scope. + Selection: `AGENTFLOW_CONTROLPLANE_STORE=postgres` + + `AGENTFLOW_CONTROLPLANE_PG_DSN` (+ optional + `AGENTFLOW_CONTROLPLANE_LEASE_SECONDS`); the slice-1 `NotImplementedError` + ratchet is gone, and a missing DSN or missing `psycopg` fails the boot + loudly — never a silent fallback to embedded. `psycopg` is a new optional + extra (`pip install agentflow-runtime[postgres]`), the `redis` import + pattern. +- **Webhook registrations (state class 5) move behind the port** — the + sharpest split-brain of the ADR's inventory was still a per-pod YAML read + outside the port after slices 1–4. New port methods + `load_webhook_registrations`/`save_webhook_registrations`; + `load_webhooks`/`save_webhooks`/`create_webhook`/`list_webhooks`/ + `get_webhook`/`deactivate_webhook` now take `app` and resolve the store + inside (the same move the alert-rule helpers made in slice 2). The embedded + adapter keeps the byte-compatible `config/webhooks.yaml`. +- **Alert-tick single-flight (ADR 0010 §2) wired into the dispatcher** — new + port methods `claim_alert_tick`/`complete_alert_tick`; + `AlertDispatcher.dispatch_alerts` claims each rule before evaluating (a + lost claim = another replica owns that rule's tick) and persists advanced + rule state **per rule** in the same transaction as the claim release — the + old full-set save would let two replicas advancing different rules clobber + each other's runtime state. Embedded grants every claim (one process), so + the single-replica profile behaves as before; a CRUD full-set save on + PostgreSQL upserts by id and does not release an in-flight claim. +- **The postgres profile shares one store across every consumer**: `main.py` + injects the app-wide store into `AuthManager` and `OutboxProcessor` when + the profile is external (embedded keeps its historical private stores); + the analytics entry points (`analytics.py`, `routers/admin.py`, + `admin_ui.py`'s QPS tile) accept the store handle and route through + `AuthManager.store`, so usage/sessions land in PostgreSQL instead of a + per-pod DuckDB file. +- **Verified live (standalone PostgreSQL 17.5, no Docker): 31/31 probes** — + the ADR's named suite (parallel claim exclusivity, lease-expiry re-drive, + restart re-drive, enqueue-win uniqueness, outbox↔dead-letter atomicity + incl. rollback halves, alert-tick single-flight) plus a full contract + parity sweep and an end-to-end app test (two boots on the postgres profile + see each other's webhook registration; usage accounting lands in PG): + `docs/perf/control-plane-pg-verify-2026-07-03.md`. The same suite runs in + CI against a new `postgres:17` service in the integration job and + self-skips where `AGENTFLOW_TEST_PG_DSN` is absent. +- Helm is untouched by design: the values schema still pins + `controlPlane.store=embedded` and the chart still refuses multi-replica + renders — the enum extension and env/secret wiring are rollout slice 6, + which this slice unblocks. + ### 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`** diff --git a/docs/decisions/0010-control-plane-externalization-postgres.md b/docs/decisions/0010-control-plane-externalization-postgres.md index 2f4a8398..9a57eb5f 100644 --- a/docs/decisions/0010-control-plane-externalization-postgres.md +++ b/docs/decisions/0010-control-plane-externalization-postgres.md @@ -186,7 +186,17 @@ ClickHouse is consumed), and `psycopg` joins the optional dependencies. 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) + - CI integration coverage on the existing PG service. + CI integration coverage (a `postgres:17` service added to the CI + integration job). Executed 2026-07-03 with two scope additions the + extraction slices had left open, both required for §1's "all six state + classes" to hold: **webhook registrations** (class 5 — the sharpest + split-brain — was still a per-pod YAML read outside the port; the + registration CRUD now resolves the store from ``app``, embedded keeps the + byte-compatible YAML) and **§2's ``claim_alert_tick`` / + ``complete_alert_tick``** wired into the dispatcher (per-rule state + persistence — a full-set save would let two replicas clobber each other's + rule runtime state). Verified live: 31/31 probes, + `docs/perf/control-plane-pg-verify-2026-07-03.md`. 6. Helm wiring (`controlPlane.store=postgres` profile: env + secret, schema enum extension — the render gate then admits multi-replica) and cutover plan Phase 3 execution: kind staging at `replicaCount=2`, verifying @@ -213,8 +223,9 @@ ClickHouse is consumed), and `psycopg` joins the optional dependencies. - Per-pod event scanning is N× read amplification on the serving backend at scale (accepted; the scan is bounded and cheap, and consolidating scanners is a later topology refinement, cf. option 3). -- Until slice 5 lands, multi-replica is simply impossible to render — a - deliberate fail-closed period. +- Until slice 6 extends the chart, multi-replica is simply impossible to + render — a deliberate fail-closed period (the slice-5 adapter is app-side; + the schema enum still pins `embedded` until the helm profile ships). ## Follow-up diff --git a/docs/perf/control-plane-pg-verify-2026-07-03.md b/docs/perf/control-plane-pg-verify-2026-07-03.md new file mode 100644 index 00000000..deb9d402 --- /dev/null +++ b/docs/perf/control-plane-pg-verify-2026-07-03.md @@ -0,0 +1,65 @@ +# Control plane on PostgreSQL — live verification (ADR 0010 rollout slice 5) + +**Date:** 2026-07-03 +**Environment:** standalone PostgreSQL 17.5 (EDB windows-x64 binaries, no +Docker, no service install: `initdb` + `pg_ctl`, port 55433, trust auth, +user/db `agentflow`) — the same no-Docker standalone-PG recipe that verified +the vault governance layer +(`vault-pii-governance-pg-verify-2026-07-02.md`). Adapter under test: +`src/serving/control_plane/postgres.py` (`PostgresControlPlaneStore`), +driven by the probe suite the ADR names for this slice: +`tests/integration/test_control_plane_postgres_live.py` with +`AGENTFLOW_TEST_PG_DSN=postgresql://agentflow@127.0.0.1:55433/agentflow`. + +**Result: 31/31 probes passed** (`pytest`, 19.45s; psycopg 3.3.4). The same +suite runs in CI against the `postgres:17` service container added to the +`test-integration` job (it self-skips when the DSN env var is absent, the +`test_clickhouse_backend_live.py` pattern). + +## The ADR's named probes + +| Probe (ADR 0010 § Rollout 5) | Test | Result | +| --- | --- | --- | +| Enqueue-win uniqueness | 8 threads race `enqueue_webhook_delivery` on one (webhook, event) → exactly **1** `True`, 1 row | passed | +| Parallel claim exclusivity | 4 threads claim 10 due rows concurrently → no row handed out twice, none lost (`FOR UPDATE SKIP LOCKED`) | passed | +| Lease-expiry re-drive | row claimed with a 0.4s lease, owner "crashes" → invisible while leased, claimable again after expiry; outcome writes clear the lease so backoff alone governs | passed | +| Restart re-drive | pending row enqueued by one store instance is claimed by a **fresh** instance, canonical body verbatim | passed | +| Outbox↔dead-letter atomicity (invariant 8) | `mark_outbox_sent` flips both rows in one transaction; with `dead_letter_events` dropped mid-scenario the outbox flip **rolls back** (row stays `pending`); `enqueue_outbox_replay` rolls back symmetrically | passed | +| Alert-tick single-flight (§2) | second claimant loses; `complete_alert_tick` persists the advanced record and releases in one transaction; a stale claim self-expires; a concurrent CRUD full-set save does **not** release an in-flight claim | passed | + +## Contract parity sweep + +Every port method exercised against live PostgreSQL with the same assertions +the embedded adapter's unit pins make: webhook outcome state machine +(backoff → `dead` at max, success → `delivered`, park), oldest-first claim +ordering under `limit`, attempt-log and alert-history roundtrips (newest +first, JSON payload decoded), webhook-registration and alert-rule +repositories (order preserved via a `position` column, full-set save = +YAML-replace semantics, ids required), dead-letter reads (tenant scoping, +reason filter, pagination, stats + trend), usage accounting (per tenant / +per key / old-key-slot hour window), session analytics (idempotent +insert-or-replace on `request_id`, usage/top-queries/top-entities/latency +percentiles/anomalies/QPS shapes, malformed window → `ValueError`, QPS +degrades to 0.0 on an unreachable server). + +## End to end: the app itself on the postgres profile + +`test_app_on_postgres_profile_shares_state_across_boots` boots the real +FastAPI app twice with `AGENTFLOW_CONTROLPLANE_STORE=postgres`: + +- boot #1 resolves `PostgresControlPlaneStore` in the lifespan, + `AuthManager.store` **is** the shared app-wide store (slice 5 injection in + `main.py`), and `POST /v1/webhooks` registers a webhook; +- boot #2 (a second pod, in production terms) sees that registration through + `GET /v1/webhooks` — the class-5 per-pod YAML split-brain is gone; +- `api_usage` rows for the authenticated tenant landed in PostgreSQL, not in + a local DuckDB file. + +## Notes + +- Claim leases default to 300s (`AGENTFLOW_CONTROLPLANE_LEASE_SECONDS` + overrides); alert-tick leases are 120s (`AlertDispatcher.tick_lease_seconds`). +- Payload columns are TEXT holding JSON strings — callers see the same + "string, you decode it" shape as on the embedded adapter. +- One connection per method call, no pooling (out of ADR 0010 scope, + recorded follow-up). diff --git a/docs/security-audit.md b/docs/security-audit.md index 1d2d7344..e6b87488 100644 --- a/docs/security-audit.md +++ b/docs/security-audit.md @@ -80,7 +80,7 @@ Audit finding A-4 flagged the dynamic-SQL surface as "one careless edit away fro - Interpolated **identifiers** come from a fixed in-code allowlist, the semantic catalog, trusted backend config, live schema introspection (`PRAGMA table_info`), `_IDENTIFIER_RE`, or `sqlglot` validation. - Interpolated **values** are either bound as `?` parameters, fixed literals, integers, or regex-extracted tokens that exclude SQL metacharacters and are additionally quoted via `_quote_literal` / `_sql_str_literal`. -No site interpolates unbound, unquoted request data. There are **no Class-A (migratable value-interpolation) sites remaining**: the hot entity/metric paths already bind values (`use_query_params` on DuckDB, `_quote_literal` elsewhere) and the operational routers already pass values through `?`. The remaining suppressions are Class-B (identifiers / structural fragments that cannot be parameterized). +No site interpolates unbound, unquoted request data. There are **no Class-A (migratable value-interpolation) sites remaining**: the hot entity/metric paths already bind values (`use_query_params` on DuckDB, `_quote_literal` elsewhere) and the operational routers already pass values through `?`. The remaining suppressions are Class-B (identifiers / structural fragments that cannot be parameterized). The `PostgresControlPlaneStore` sites added by ADR 0010 slice 5 (`control_plane/postgres.py`, reviewed 2026-07-03) interpolate only a table name that is a module literal at exactly two call sites; every value binds via `%s`. The number of suppressions per file is pinned by `test_interpolated_sql_nosec_surface_is_pinned` — a new site (even inside an already-listed file) or a new file fails CI and forces a review. Each suppression's per-line rationale comment is enforced by `test_nosec_comments_carry_reason`. diff --git a/helm/agentflow/templates/deployment.yaml b/helm/agentflow/templates/deployment.yaml index e002b9d3..39c863a1 100644 --- a/helm/agentflow/templates/deployment.yaml +++ b/helm/agentflow/templates/deployment.yaml @@ -2,7 +2,7 @@ {{- fail "DuckDB persistence requires a single writer replica: set replicaCount=1 and autoscaling.maxReplicas=1, or disable persistence and use ephemeral storage." }} {{- end }} {{- if and (or (gt (int .Values.replicaCount) 1) (and .Values.autoscaling.enabled (gt (int .Values.autoscaling.maxReplicas) 1))) (or (ne .Values.controlPlane.store "postgres") (ne .Values.serving.backend "clickhouse")) }} -{{- fail "Multi-replica requires BOTH an external serving engine (serving.backend=clickhouse, ADR 0006/0007) AND an external control-plane store (controlPlane.store=postgres, ADR 0009/0010): the embedded per-pod store forks webhook/alert/outbox/usage state across replicas (duplicate deliveries, split alert history). Until the ADR 0010 rollout ships the postgres store, keep replicaCount=1 and autoscaling.maxReplicas=1." }} +{{- fail "Multi-replica requires BOTH an external serving engine (serving.backend=clickhouse, ADR 0006/0007) AND an external control-plane store (controlPlane.store=postgres, ADR 0009/0010): the embedded per-pod store forks webhook/alert/outbox/usage state across replicas (duplicate deliveries, split alert history). The app-side postgres adapter shipped with ADR 0010 slice 5; until the chart profile for it lands (rollout slice 6), keep replicaCount=1 and autoscaling.maxReplicas=1." }} {{- end }} {{- $secretName := .Values.secrets.existingSecret | default (include "agentflow.fullname" .) }} apiVersion: apps/v1 diff --git a/helm/agentflow/values.yaml b/helm/agentflow/values.yaml index 2269ff57..b254e47c 100644 --- a/helm/agentflow/values.yaml +++ b/helm/agentflow/values.yaml @@ -134,13 +134,16 @@ serving: existingSecret: "" passwordKey: clickhouse-password -# Control-plane state (ADR 0009 / ADR 0010): webhook queue+log, alert rules and -# history, outbox, dead-letter and usage accounting. 'embedded' = per-pod -# DuckDB + config files — correct for the single-replica profile, a split-brain -# at replicaCount > 1 (duplicate deliveries, forked alert state). ADR 0010 -# externalizes this state to PostgreSQL behind a ControlPlaneStore port; until -# that adapter ships, the values schema pins store=embedded (fail-closed -# ratchet) and the chart refuses any multi-replica render. +# Control-plane state (ADR 0009 / ADR 0010): webhook queue+log, registrations, +# alert rules and history, outbox, dead-letter and usage accounting. +# 'embedded' = per-pod DuckDB + config files — correct for the single-replica +# profile, a split-brain at replicaCount > 1 (duplicate deliveries, forked +# alert state). The PostgresControlPlaneStore adapter shipped with ADR 0010 +# rollout slice 5 (app-side: AGENTFLOW_CONTROLPLANE_STORE=postgres + +# AGENTFLOW_CONTROLPLANE_PG_DSN); the chart profile for it (env + secret +# wiring, schema enum extension) lands with rollout slice 6 — until then the +# values schema pins store=embedded (fail-closed ratchet) and the chart +# refuses any multi-replica render. controlPlane: store: embedded diff --git a/pyproject.toml b/pyproject.toml index 4a03450b..34468a4b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -37,6 +37,12 @@ cloud = [ "boto3>=1.35,<2", "pyiceberg[pyiceberg-core]>=0.7,<1", ] +postgres = [ + # ADR 0010 slice 5: PostgresControlPlaneStore (scale profile). Optional + # exactly like redis — the embedded default profile needs none of it, + # and control_plane/postgres.py degrades to a clear boot error without it. + "psycopg[binary]>=3.2,<4", +] load = [ "locust>=2.29,<3", # benchmark_freshness.py: production QueryCache semantics (TTL + key @@ -158,6 +164,11 @@ ignore = ["S101", "S311"] "src/serving/api/routers/lineage.py" = [ "S608", ] +# Interpolations are internal literals only (a lease-expiry SQL fragment, +# optional filter clauses, table-name constants); all values bind via %s. +"src/serving/control_plane/postgres.py" = [ + "S608", +] "src/serving/api/routers/slo.py" = [ "S608", ] @@ -284,7 +295,10 @@ editable-installs = [".[dev]"] editable-installs = [".[dev,cloud]"] [tool.agentflow.dependency-profiles.profiles.test-sdk] -editable-installs = [".[dev,cloud]", "./sdk"] +# postgres: the CI integration job drives the PostgresControlPlaneStore live +# suite against its postgres:17 service (ADR 0010 slice 5), so it needs the +# optional psycopg extra. +editable-installs = [".[dev,cloud,postgres]", "./sdk"] [tool.agentflow.dependency-profiles.profiles.test-integrations] editable-installs = [".[dev,cloud]", "./sdk", "./integrations[mcp]"] diff --git a/src/serving/api/alerts/dispatcher.py b/src/serving/api/alerts/dispatcher.py index 49a02689..144e2966 100644 --- a/src/serving/api/alerts/dispatcher.py +++ b/src/serving/api/alerts/dispatcher.py @@ -223,6 +223,13 @@ def __init__(self, app: FastAPI, poll_interval_seconds: float = 60.0) -> None: self.app = app self.poll_interval_seconds = poll_interval_seconds self.backoff_seconds = [1.0, 5.0, 25.0] + # How long one rule's tick claim is held before it self-expires + # (ADR 0010 §2). Longer than the poll interval so a healthy evaluation + # (deliveries retry with backoff) finishes inside its lease; short + # enough that a crashed claim owner only silences a rule for two + # ticks. The embedded store grants claims unconditionally, so this + # only matters on the PostgreSQL profile. + self.tick_lease_seconds = 120.0 self._task: asyncio.Task | None = None def start(self) -> None: @@ -250,19 +257,36 @@ async def run(self) -> None: async def dispatch_alerts(self) -> int: from .escalation import dispatch_alert + store = get_control_plane_store(self.app) alerts = load_alerts(self.app) now = datetime.now(UTC) triggered = 0 - changed = False - for index, alert in enumerate(alerts): + for alert in alerts: if not alert.active: continue - updated_alert, alert_changed, alert_triggered = await dispatch_alert(self, alert, now) - alerts[index] = updated_alert + # Single-flight each rule across replicas (ADR 0010 §2): only the + # claim winner evaluates and pages; a lost claim means another pod + # owns this rule's tick. The embedded store grants every claim + # (one process), so the single-replica profile behaves as before. + if not store.claim_alert_tick(alert.id, lease_seconds=self.tick_lease_seconds): + continue + try: + updated_alert, alert_changed, alert_triggered = await dispatch_alert( + self, alert, now + ) + except BaseException: + # Release the claim so the next tick retries this rule + # immediately instead of waiting out the lease. + store.complete_alert_tick(alert.id, record=None) + raise triggered += alert_triggered - changed = changed or alert_changed - if changed: - save_alerts(self.app, alerts) + # Rule state advances per rule, in the same transaction as the + # claim release — a full-set save here would let two replicas + # advancing different rules clobber each other's runtime state. + store.complete_alert_tick( + alert.id, + record=updated_alert.model_dump(mode="json") if alert_changed else None, + ) return triggered async def send_test_alert(self, alert: AlertRule) -> dict: diff --git a/src/serving/api/analytics.py b/src/serving/api/analytics.py index a891ab31..0726bff9 100644 --- a/src/serving/api/analytics.py +++ b/src/serving/api/analytics.py @@ -13,19 +13,22 @@ from starlette.responses import Response from starlette.types import Message -from src.serving.control_plane import EmbeddedControlPlaneStore +from src.serving.control_plane import ControlPlaneStore, EmbeddedControlPlaneStore from src.serving.duckdb_connection import connect_duckdb logger = structlog.get_logger() -def _usage_store(db_path: Path | str) -> EmbeddedControlPlaneStore: +def _usage_store(source: ControlPlaneStore | Path | str) -> ControlPlaneStore: # 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) + # ControlPlaneStore port (control_plane/embedded.py). Slice 5 makes the + # entry points polymorphic: callers on the scale profile hand in the + # shared store (AuthManager.store, a PostgresControlPlaneStore there); + # a path keeps the pre-port behavior — a fresh, cheap embedded wrapper + # per call, nothing connected until a method on it runs. + if isinstance(source, ControlPlaneStore): + return source + return EmbeddedControlPlaneStore(usage_db_path_provider=lambda: source) AnalyticsMiddleware = Callable[ @@ -122,7 +125,7 @@ async def receive() -> Message: # un-throttled DB write/thread spawn. (audit_30_06_26.md S1) if getattr(request.state, "tenant_key", None) is not None: _schedule_session_write( - request.app.state.auth_manager.db_path, + request.app.state.auth_manager.store, request_id, _build_session_record( request=request, @@ -153,7 +156,7 @@ async def receive() -> Message: background = BackgroundTasks([background]) background.add_task( _schedule_session_write, - request.app.state.auth_manager.db_path, + request.app.state.auth_manager.store, request_id, _build_session_record( request=request, @@ -171,58 +174,61 @@ async def receive() -> Message: def get_usage_analytics( - db_path: Path | str, + source: ControlPlaneStore | Path | str, *, window: str = "24h", tenant: str | None = None, ) -> dict: - return _usage_store(db_path).get_usage_analytics(window=window, tenant=tenant) + return _usage_store(source).get_usage_analytics(window=window, tenant=tenant) def get_top_queries( - db_path: Path | str, + source: ControlPlaneStore | Path | str, *, limit: int = 10, window: str = "24h", ) -> dict: - return _usage_store(db_path).get_top_queries(limit=limit, window=window) + return _usage_store(source).get_top_queries(limit=limit, window=window) def get_top_entities( - db_path: Path | str, + source: ControlPlaneStore | Path | str, *, limit: int = 10, window: str = "24h", ) -> dict: - return _usage_store(db_path).get_top_entities(limit=limit, window=window) + return _usage_store(source).get_top_entities(limit=limit, window=window) def get_latency_analytics( - db_path: Path | str, + source: ControlPlaneStore | Path | str, *, window: str = "24h", ) -> dict: - return _usage_store(db_path).get_latency_analytics(window=window) + return _usage_store(source).get_latency_analytics(window=window) -def get_anomalies(db_path: Path | str, *, window: str = "24h") -> dict: - return _usage_store(db_path).get_anomalies(window=window) +def get_anomalies(source: ControlPlaneStore | Path | str, *, window: str = "24h") -> dict: + return _usage_store(source).get_anomalies(window=window) -def _schedule_session_write(db_path: Path | str, request_id: str, record: dict) -> None: +def _schedule_session_write( + source: ControlPlaneStore | Path | str, request_id: str, record: dict +) -> None: threading.Thread( target=_insert_session, - args=(db_path, request_id, record), + args=(source, request_id, record), daemon=True, ).start() -def _insert_session(db_path: Path | str, request_id: str, record: dict) -> None: +def _insert_session(source: ControlPlaneStore | Path | str, request_id: str, record: dict) -> None: # 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 + # guaranteed to exist by main.py's boot-time call (embedded profile; the + # postgres adapter creates its schema once per process), 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) + _usage_store(source).record_api_session(request_id, record) def _build_session_record( diff --git a/src/serving/api/main.py b/src/serving/api/main.py index 18480b53..cbdff2cd 100644 --- a/src/serving/api/main.py +++ b/src/serving/api/main.py @@ -53,7 +53,7 @@ ) from src.serving.api.webhook_dispatcher import WebhookDispatcher from src.serving.cache import QueryCache -from src.serving.control_plane import get_control_plane_store +from src.serving.control_plane import control_plane_store_kind, get_control_plane_store from src.serving.db_pool import DuckDBPool from src.serving.semantic_layer.catalog import DataCatalog from src.serving.semantic_layer.query_engine import QueryEngine @@ -162,10 +162,24 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: value=os.getenv("CACHE_TTL_SECONDS"), fallback=30, ) + # Control-plane store (ADR 0010): resolve eagerly so a misconfigured + # AGENTFLOW_CONTROLPLANE_STORE fails the boot, not the first delivery. + # Reset any instance cached by a previous lifespan of this process-wide + # app — the query engine above is fresh, the store must bind to it. + app.state.control_plane_store = None + control_plane_store = get_control_plane_store(app) + # On the external (postgres) profile every control-plane consumer shares + # the one store (slice 5); the embedded profile injects nothing, so the + # consumers keep building their historical private stores (usage on its + # own file, outbox on the engine's conn/path) exactly as before. + shared_control_plane_store = ( + control_plane_store if control_plane_store_kind() != "embedded" else None + ) app.state.auth_manager = AuthManager( api_keys_path=os.getenv("AGENTFLOW_API_KEYS_FILE"), db_path=os.getenv("AGENTFLOW_USAGE_DB_PATH", "agentflow_api.duckdb"), admin_key=os.getenv("AGENTFLOW_ADMIN_KEY"), + store=shared_control_plane_store, ) app.state.auth_manager.load() if app.state.demo_mode: @@ -195,13 +209,11 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: ) app.state.auth_manager.register_signal_handlers() app.state.auth_manager.ensure_usage_table() - ensure_analytics_table(app.state.auth_manager.db_path) - # Control-plane store (ADR 0010): resolve eagerly so a misconfigured - # AGENTFLOW_CONTROLPLANE_STORE fails the boot, not the first delivery. - # Reset any instance cached by a previous lifespan of this process-wide - # app — the query engine above is fresh, the store must bind to it. - app.state.control_plane_store = None - get_control_plane_store(app) + if shared_control_plane_store is None: + # Embedded-profile bootstrap of the local api_sessions file table; the + # postgres adapter creates its whole schema (sessions included) once + # per process, so a stray local DuckDB file would be dead weight. + ensure_analytics_table(app.state.auth_manager.db_path) app.state.webhook_dispatcher = WebhookDispatcher(app) original_dispatch_new_events = app.state.webhook_dispatcher.dispatch_new_events @@ -217,7 +229,9 @@ async def dispatch_new_events_with_cache_invalidation() -> None: app.state.alert_dispatcher = AlertDispatcher(app) if getattr(app.state, "alert_dispatcher_autostart", True): app.state.alert_dispatcher.start() - if app.state.query_engine._db_path == ":memory:": + if shared_control_plane_store is not None: + app.state.outbox_processor = OutboxProcessor(store=shared_control_plane_store) + elif app.state.query_engine._db_path == ":memory:": app.state.outbox_processor = OutboxProcessor(conn=app.state.query_engine._conn) else: app.state.outbox_processor = OutboxProcessor( diff --git a/src/serving/api/routers/admin.py b/src/serving/api/routers/admin.py index 6605e4b8..d83830e9 100644 --- a/src/serving/api/routers/admin.py +++ b/src/serving/api/routers/admin.py @@ -99,7 +99,7 @@ async def get_analytics_usage( ) -> dict[str, object]: manager = get_auth_manager(request) try: - return get_usage_analytics(manager.db_path, window=window, tenant=tenant) + return get_usage_analytics(manager.store, window=window, tenant=tenant) except ValueError as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc @@ -112,7 +112,7 @@ async def get_analytics_top_queries( ) -> dict[str, object]: manager = get_auth_manager(request) try: - return get_top_queries(manager.db_path, limit=limit, window=window) + return get_top_queries(manager.store, limit=limit, window=window) except ValueError as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc @@ -125,7 +125,7 @@ async def get_analytics_top_entities( ) -> dict[str, object]: manager = get_auth_manager(request) try: - return get_top_entities(manager.db_path, limit=limit, window=window) + return get_top_entities(manager.store, limit=limit, window=window) except ValueError as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc @@ -137,7 +137,7 @@ async def get_analytics_latency( ) -> dict[str, object]: manager = get_auth_manager(request) try: - return get_latency_analytics(manager.db_path, window=window) + return get_latency_analytics(manager.store, window=window) except ValueError as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc @@ -149,6 +149,6 @@ async def get_analytics_anomalies( ) -> dict[str, object]: manager = get_auth_manager(request) try: - return get_anomalies(manager.db_path, window=window) + return get_anomalies(manager.store, window=window) except ValueError as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc diff --git a/src/serving/api/routers/admin_ui.py b/src/serving/api/routers/admin_ui.py index c9438f95..044b0d3e 100644 --- a/src/serving/api/routers/admin_ui.py +++ b/src/serving/api/routers/admin_ui.py @@ -11,7 +11,7 @@ from src.serving.api.auth import require_admin_key from src.serving.cache import ENTITY_TTL_SECONDS -from src.serving.control_plane import EmbeddedControlPlaneStore +from src.serving.control_plane import ControlPlaneStore, EmbeddedControlPlaneStore router = APIRouter( prefix="/admin", @@ -53,7 +53,7 @@ async def _build_context(request: Request, *, partial: bool) -> dict[str, object "key_usage": manager.list_keys_with_usage(), "db_pool": state.db_pool.stats(), "cache_stats": _cache_stats(state), - "qps_1m": await run_in_threadpool(_qps_last_minute, manager.db_path), + "qps_1m": await run_in_threadpool(_qps_last_minute, manager.store), } @@ -75,8 +75,12 @@ def _cache_stats(state: State) -> dict[str, object]: } -def _qps_last_minute(db_path: Path | str) -> float: +def _qps_last_minute(source: ControlPlaneStore | Path | str) -> float: # 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) + # direct connect_duckdb(db_path) query. Slice 5: the admin dashboard + # hands in the manager's store (shared PostgreSQL store on the scale + # profile); a bare path still builds the embedded per-call wrapper. + if isinstance(source, ControlPlaneStore): + return source.get_queries_per_second_last_minute() + store = EmbeddedControlPlaneStore(usage_db_path_provider=lambda: source) return store.get_queries_per_second_last_minute() diff --git a/src/serving/api/routers/webhooks.py b/src/serving/api/routers/webhooks.py index f066434a..4e9ee4c3 100644 --- a/src/serving/api/routers/webhooks.py +++ b/src/serving/api/routers/webhooks.py @@ -13,7 +13,6 @@ create_webhook, deactivate_webhook, get_webhook, - get_webhook_config_path, list_webhooks, ) from src.serving.control_plane import get_control_plane_store @@ -40,7 +39,7 @@ async def register_webhook(payload: WebhookCreateRequest, request: Request) -> d except UnsafeEgressURLError as exc: raise HTTPException(status_code=400, detail=f"Unsafe webhook URL: {exc}") from exc registration = create_webhook( - get_webhook_config_path(request.app), + request.app, url=str(payload.url), tenant=_tenant(request), filters=payload.filters, @@ -50,7 +49,7 @@ async def register_webhook(payload: WebhookCreateRequest, request: Request) -> d @router.get("") async def list_my_webhooks(request: Request) -> dict[str, object]: - webhooks = list_webhooks(get_webhook_config_path(request.app), _tenant(request)) + webhooks = list_webhooks(request.app, _tenant(request)) # Exclude `secret` from list/read responses. Plaintext signing material # is returned only once on POST. Listing it again would let any tenant # API key recover signing secrets after creation (audit p2_2 #7). @@ -62,7 +61,7 @@ async def list_my_webhooks(request: Request) -> dict[str, object]: @router.delete("/{webhook_id}", status_code=status.HTTP_204_NO_CONTENT) async def unregister_webhook(webhook_id: str, request: Request) -> Response: removed = deactivate_webhook( - get_webhook_config_path(request.app), + request.app, webhook_id, _tenant(request), ) @@ -74,7 +73,7 @@ async def unregister_webhook(webhook_id: str, request: Request) -> Response: @router.post("/{webhook_id}/test") async def test_webhook(webhook_id: str, request: Request) -> dict[str, object]: registration = get_webhook( - get_webhook_config_path(request.app), + request.app, webhook_id, _tenant(request), ) @@ -95,7 +94,7 @@ async def test_webhook(webhook_id: str, request: Request) -> dict[str, object]: @router.get("/{webhook_id}/logs") async def webhook_logs(webhook_id: str, request: Request) -> dict[str, object]: registration = get_webhook( - get_webhook_config_path(request.app), + request.app, webhook_id, _tenant(request), ) diff --git a/src/serving/api/webhook_dispatcher.py b/src/serving/api/webhook_dispatcher.py index 37327e35..3adc4f1c 100644 --- a/src/serving/api/webhook_dispatcher.py +++ b/src/serving/api/webhook_dispatcher.py @@ -56,36 +56,33 @@ def get_webhook_config_path(app: FastAPI) -> Path: return Path(configured) if configured else DEFAULT_WEBHOOKS_CONFIG_PATH -def load_webhooks(path: Path) -> list[WebhookRegistration]: - if not path.exists(): - return [] - raw = path.read_text(encoding="utf-8") - if not raw.strip(): - return [] - data = yaml.safe_load(raw) if yaml is not None else json.loads(raw) - config = WebhookConfig.model_validate(data or {}) - return config.webhooks - - -def save_webhooks(path: Path, webhooks: list[WebhookRegistration]) -> None: - path.parent.mkdir(parents=True, exist_ok=True) +# The registration CRUD helpers below take ``app`` and resolve the +# control-plane store inside (ADR 0010 slice 5) — the same move the alert-rule +# helpers made in slice 2: registrations were the last control-plane state +# read from a per-pod file (config/webhooks.yaml) instead of the store, the +# exact split-brain the ADR's inventory calls the sharpest. The embedded +# adapter keeps the YAML file (via ``get_webhook_config_path``), so the +# single-replica profile and its on-disk format do not change. + + +def load_webhooks(app: FastAPI) -> list[WebhookRegistration]: + records = get_control_plane_store(app).load_webhook_registrations() + return WebhookConfig.model_validate({"webhooks": records}).webhooks + + +def save_webhooks(app: FastAPI, webhooks: list[WebhookRegistration]) -> None: payload = WebhookConfig(webhooks=webhooks).model_dump(mode="json") - content = ( - yaml.safe_dump(payload, sort_keys=False) - if yaml is not None - else json.dumps(payload, indent=2) - ) - path.write_text(content, encoding="utf-8") + get_control_plane_store(app).save_webhook_registrations(payload["webhooks"]) def create_webhook( - path: Path, + app: FastAPI, *, url: str, tenant: str, filters: WebhookFilters, ) -> WebhookRegistration: - webhooks = load_webhooks(path) + webhooks = load_webhooks(app) registration = WebhookRegistration( id=str(uuid.uuid4()), url=url, @@ -95,25 +92,25 @@ def create_webhook( created_at=datetime.now(UTC), ) webhooks.append(registration) - save_webhooks(path, webhooks) + save_webhooks(app, webhooks) return registration -def list_webhooks(path: Path, tenant: str) -> list[WebhookRegistration]: +def list_webhooks(app: FastAPI, tenant: str) -> list[WebhookRegistration]: return [ - webhook for webhook in load_webhooks(path) if webhook.tenant == tenant and webhook.active + webhook for webhook in load_webhooks(app) if webhook.tenant == tenant and webhook.active ] -def get_webhook(path: Path, webhook_id: str, tenant: str) -> WebhookRegistration | None: - for webhook in load_webhooks(path): +def get_webhook(app: FastAPI, webhook_id: str, tenant: str) -> WebhookRegistration | None: + for webhook in load_webhooks(app): if webhook.id == webhook_id and webhook.tenant == tenant and webhook.active: return webhook return None -def deactivate_webhook(path: Path, webhook_id: str, tenant: str) -> bool: - webhooks = load_webhooks(path) +def deactivate_webhook(app: FastAPI, webhook_id: str, tenant: str) -> bool: + webhooks = load_webhooks(app) changed = False for webhook in webhooks: if webhook.id == webhook_id and webhook.tenant == tenant and webhook.active: @@ -121,7 +118,7 @@ def deactivate_webhook(path: Path, webhook_id: str, tenant: str) -> bool: changed = True break if changed: - save_webhooks(path, webhooks) + save_webhooks(app, webhooks) return changed @@ -179,8 +176,7 @@ def mark_existing_events_seen(self) -> None: logger.warning("webhook_seen_init_failed", error=str(exc)) async def dispatch_new_events(self) -> None: - path = get_webhook_config_path(self.app) - webhooks = [webhook for webhook in load_webhooks(path) if webhook.active] + webhooks = [webhook for webhook in load_webhooks(self.app) if webhook.active] webhooks_by_tenant: dict[str, list[WebhookRegistration]] = {} for webhook in webhooks: webhooks_by_tenant.setdefault(webhook.tenant, []).append(webhook) @@ -417,9 +413,8 @@ async def process_delivery_queue(self) -> None: forever. Bounded by ``redrive_batch_size`` so one pass can't stall the loop on a large backlog.""" store = get_control_plane_store(self.app) - path = get_webhook_config_path(self.app) for row in store.claim_due_webhook_deliveries(limit=self.redrive_batch_size): - webhook = get_webhook(path, row.webhook_id, str(row.tenant or "default")) + webhook = get_webhook(self.app, row.webhook_id, str(row.tenant or "default")) if webhook is None: store.park_webhook_delivery( webhook_id=row.webhook_id, diff --git a/src/serving/control_plane/__init__.py b/src/serving/control_plane/__init__.py index adb91c53..35102678 100644 --- a/src/serving/control_plane/__init__.py +++ b/src/serving/control_plane/__init__.py @@ -1,9 +1,12 @@ """Control-plane state store (ADR 0009 / ADR 0010). ``ControlPlaneStore`` is the port; ``EmbeddedControlPlaneStore`` (DuckDB, -single-replica default) is the shipped adapter; ``PostgresControlPlaneStore`` -(scale profile) arrives with ADR 0010 rollout slice 5. Resolve the app's -store via ``get_control_plane_store`` — never through ``query_engine._conn``. +single-replica default) and ``PostgresControlPlaneStore`` (scale profile, +ADR 0010 rollout slice 5) are the adapters. Resolve the app's store via +``get_control_plane_store`` — never through ``query_engine._conn``. +``PostgresControlPlaneStore`` is intentionally NOT re-exported here: it +imports lazily (psycopg is an optional dependency, the ``redis`` pattern), +so reach it via ``src.serving.control_plane.postgres`` only when configured. """ from .embedded import ( @@ -17,19 +20,23 @@ ensure_webhook_delivery_queue_table, ) from .store import ( + CONTROL_PLANE_PG_DSN_ENV, CONTROL_PLANE_STORE_ENV, ControlPlaneStore, OutboxEntry, WebhookQueueRow, + control_plane_store_kind, get_control_plane_store, ) __all__ = [ + "CONTROL_PLANE_PG_DSN_ENV", "CONTROL_PLANE_STORE_ENV", "ControlPlaneStore", "EmbeddedControlPlaneStore", "OutboxEntry", "WebhookQueueRow", + "control_plane_store_kind", "ensure_alert_history_table", "ensure_api_sessions_table", "ensure_api_usage_table", diff --git a/src/serving/control_plane/embedded.py b/src/serving/control_plane/embedded.py index 8d3df50e..0c1ecbc8 100644 --- a/src/serving/control_plane/embedded.py +++ b/src/serving/control_plane/embedded.py @@ -269,10 +269,12 @@ def __init__( *, alert_rules_path_provider: Callable[[], Path] | None = None, usage_db_path_provider: Callable[[], Path | str] | None = None, + webhook_registrations_path_provider: Callable[[], Path] | 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 + self._webhook_registrations_path_provider = webhook_registrations_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 @@ -291,6 +293,16 @@ def _alert_rules_path(self) -> Path: ) return self._alert_rules_path_provider() + @property + def _webhook_registrations_path(self) -> Path: + if self._webhook_registrations_path_provider is None: + raise RuntimeError( + "EmbeddedControlPlaneStore was constructed without a " + "webhook_registrations_path_provider; webhook-registration " + "repository methods are unavailable." + ) + return self._webhook_registrations_path_provider() + @property def _conn(self) -> duckdb.DuckDBPyConnection: if self._conn_provider is None: @@ -570,6 +582,32 @@ def get_alert_delivery_history(self, alert_id: str, *, limit: int = 20) -> list[ pass return records + # --- webhook registration repository --------------------------------------- + + def load_webhook_registrations(self) -> list[dict]: + # Byte-compatible with the pre-port webhook_dispatcher.load_webhooks + # YAML round-trip (ADR 0010 slice 5) — existing config/webhooks.yaml + # files keep working unchanged. + path = self._webhook_registrations_path + if not path.exists(): + return [] + raw = path.read_text(encoding="utf-8") + if not raw.strip(): + return [] + data = yaml.safe_load(raw) if yaml is not None else json.loads(raw) + return list((data or {}).get("webhooks", [])) + + def save_webhook_registrations(self, registrations: list[dict]) -> None: + path = self._webhook_registrations_path + path.parent.mkdir(parents=True, exist_ok=True) + payload = {"webhooks": registrations} + content = ( + yaml.safe_dump(payload, sort_keys=False) + if yaml is not None + else json.dumps(payload, indent=2) + ) + path.write_text(content, encoding="utf-8") + # --- alert rule repository (mutable runtime state) ------------------------ def load_alert_rules(self) -> list[dict]: @@ -593,6 +631,25 @@ def save_alert_rules(self, rules: list[dict]) -> None: ) path.write_text(content, encoding="utf-8", newline="\n") + def claim_alert_tick(self, rule_id: str, *, lease_seconds: float) -> bool: + # One process, one dispatcher loop: every claim is granted — the same + # degenerate exclusivity as claim_due_webhook_deliveries above. The + # PostgreSQL adapter takes a real lease here (ADR 0010 §2). + return True + + def complete_alert_tick(self, rule_id: str, *, record: dict | None) -> None: + if record is None: + # Nothing advanced and embedded claims hold no lease to release. + return + rules = self.load_alert_rules() + for index, existing in enumerate(rules): + if existing.get("id") == rule_id: + rules[index] = record + break + else: + rules.append(record) + self.save_alert_rules(rules) + # --- replay outbox + dead-letter (invariant 8: one transaction) ----------- def ensure_outbox_schema(self) -> None: diff --git a/src/serving/control_plane/postgres.py b/src/serving/control_plane/postgres.py new file mode 100644 index 00000000..b3f63266 --- /dev/null +++ b/src/serving/control_plane/postgres.py @@ -0,0 +1,1353 @@ +"""PostgreSQL control-plane store — the scale profile (ADR 0010 slice 5). + +All six state classes from the ADR's inventory live in ordinary PostgreSQL +tables, and the claim semantics the port only satisfies degenerately on the +embedded adapter become real here: + +- ``enqueue_webhook_delivery`` wins by ``INSERT .. ON CONFLICT DO NOTHING`` + rowcount — exactly one replica inline-delivers a fresh enqueue. +- ``claim_due_webhook_deliveries`` / ``claim_due_outbox_entries`` take rows + with ``FOR UPDATE SKIP LOCKED`` and stamp a lease + (``lease_expires_at``): N replicas work-steal without leader election, and + a crashed owner's rows become due again when the lease runs out. +- ``claim_alert_tick`` single-flights each alert rule's evaluation via a + lease column on the rule row; ``complete_alert_tick`` releases the claim + and persists that rule's advanced runtime state in the same transaction. +- ``mark_outbox_sent`` / ``schedule_outbox_retry`` / ``enqueue_outbox_replay`` + keep the outbox↔dead-letter flip in one transaction (invariant 8) — here it + is simply *a* transaction, no manual BEGIN/ROLLBACK choreography. + +Design constraints inherited from the embedded adapter, kept deliberately: + +- **One connection per method call, no pool** — mirrors the pre-port + usage/session code opening a fresh file connection per request; pooling is + explicitly out of ADR 0010's scope and noted as a follow-up. +- **Every method is one transaction** — the ``_connect`` context manager + commits on success and rolls back on any exception, which is what makes + the invariant-8 methods atomic without adapter-specific ceremony. +- **JSON payloads are stored as TEXT** holding the caller's JSON string, + not ``jsonb`` — the port contract says payloads come back "as stored + (string or dict), the caller decodes", and the embedded adapter returns + strings; keeping strings here means callers see one shape on both + profiles. +- **Schema DDL runs once per store instance** (first use), never lazily + inside the write methods — the same fault-injection rule the port + docstring pins for the outbox tables: a test that drops a table + mid-scenario to simulate a failed transaction must see the failure, not a + silently recreated table. + +``psycopg`` (v3) is an optional dependency imported at module load with a +``None`` fallback, exactly like ``redis`` in the rate limiter: importing this +module is safe without it, constructing the store is not. +""" + +from __future__ import annotations + +import json +import os +import re +import threading +import time +from collections.abc import Sequence +from datetime import UTC, datetime, timedelta +from typing import TYPE_CHECKING, Any + +import structlog + +from .store import ( + CONTROL_PLANE_PG_DSN_ENV, + ControlPlaneStore, + OutboxEntry, + WebhookQueueRow, +) + +if TYPE_CHECKING: + from contextlib import AbstractContextManager + +try: + import psycopg +except ImportError: # pragma: no cover + psycopg = None # type: ignore[assignment] + +logger = structlog.get_logger() + +# How long a claimed webhook-queue / outbox row stays invisible to other +# claimants before it self-expires back to due. Long enough for a full +# delivery burst (3 HTTP attempts x timeout + backoff) per row across a +# claimed batch; short enough that a crashed pod's backlog resumes within +# minutes. Overridable per store via the constructor. +DEFAULT_CLAIM_LEASE_SECONDS = 300.0 + +_SCHEMA_STATEMENTS = ( + """ + CREATE TABLE IF NOT EXISTS webhook_delivery_queue ( + webhook_id TEXT NOT NULL, + event_id TEXT NOT NULL, + tenant TEXT, + event_type TEXT, + body TEXT, + status TEXT NOT NULL DEFAULT 'pending', + attempts INTEGER NOT NULL DEFAULT 0, + next_attempt_at TIMESTAMPTZ, + last_status_code INTEGER, + last_error TEXT, + lease_expires_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (webhook_id, event_id) + ) + """, + """ + CREATE INDEX IF NOT EXISTS webhook_delivery_queue_due_idx + ON webhook_delivery_queue (created_at) WHERE status = 'pending' + """, + """ + CREATE TABLE IF NOT EXISTS webhook_deliveries ( + delivery_id TEXT, + webhook_id TEXT, + event_id TEXT, + event_type TEXT, + attempt INTEGER, + status_code INTEGER, + success BOOLEAN, + error TEXT, + delivered_at TIMESTAMPTZ NOT NULL DEFAULT now() + ) + """, + """ + CREATE INDEX IF NOT EXISTS webhook_deliveries_webhook_idx + ON webhook_deliveries (webhook_id, delivered_at DESC) + """, + """ + CREATE TABLE IF NOT EXISTS alert_history ( + delivery_id TEXT, + alert_id TEXT, + alert_name TEXT, + metric TEXT, + current_value DOUBLE PRECISION, + previous_value DOUBLE PRECISION, + change_pct DOUBLE PRECISION, + threshold DOUBLE PRECISION, + condition TEXT, + metric_window TEXT, + tenant TEXT, + event_type TEXT, + status_code INTEGER, + success BOOLEAN, + error TEXT, + payload TEXT, + triggered_at TIMESTAMPTZ NOT NULL DEFAULT now() + ) + """, + """ + CREATE INDEX IF NOT EXISTS alert_history_alert_idx + ON alert_history (alert_id, triggered_at DESC) + """, + """ + CREATE TABLE IF NOT EXISTS webhook_registrations ( + id TEXT PRIMARY KEY, + position INTEGER NOT NULL, + record TEXT NOT NULL + ) + """, + """ + CREATE TABLE IF NOT EXISTS alert_rules ( + id TEXT PRIMARY KEY, + position INTEGER NOT NULL, + record TEXT NOT NULL, + tick_lease_expires_at TIMESTAMPTZ + ) + """, + """ + CREATE TABLE IF NOT EXISTS outbox ( + id TEXT PRIMARY KEY, + event_id TEXT NOT NULL, + payload TEXT NOT NULL, + topic TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + sent_at TIMESTAMPTZ, + status TEXT DEFAULT 'pending', + retry_count INTEGER DEFAULT 0, + next_attempt_at TIMESTAMPTZ DEFAULT now(), + last_error TEXT, + lease_expires_at TIMESTAMPTZ + ) + """, + """ + CREATE INDEX IF NOT EXISTS outbox_due_idx + ON outbox (created_at) WHERE status = 'pending' + """, + """ + CREATE TABLE IF NOT EXISTS dead_letter_events ( + event_id TEXT PRIMARY KEY, + tenant_id TEXT DEFAULT 'default', + event_type TEXT, + payload TEXT, + failure_reason TEXT, + failure_detail TEXT, + received_at TIMESTAMPTZ, + retry_count INTEGER DEFAULT 0, + last_retried_at TIMESTAMPTZ, + status TEXT DEFAULT 'failed' + ) + """, + """ + CREATE TABLE IF NOT EXISTS api_usage ( + tenant TEXT, + key_name TEXT, + endpoint TEXT, + ts TIMESTAMPTZ NOT NULL DEFAULT now(), + key_id TEXT, + key_slot TEXT + ) + """, + """ + CREATE INDEX IF NOT EXISTS api_usage_ts_idx ON api_usage (ts) + """, + """ + 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 DOUBLE PRECISION, + cache_hit BOOLEAN, + entity_type TEXT, + metric_name TEXT, + query_engine TEXT, + ts TIMESTAMPTZ NOT NULL DEFAULT now(), + entity_id TEXT, + query_text TEXT + ) + """, + """ + CREATE INDEX IF NOT EXISTS api_sessions_ts_idx ON api_sessions (ts) + """, +) + + +def _window_to_interval(window: str) -> str: + # Same grammar as the embedded adapter's parser; the ' minutes/hours/ + # days' strings it produces are valid PostgreSQL interval literals too, + # but parsing here (rather than passing user input through) keeps the + # ValueError contract for malformed windows. + 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 PostgresControlPlaneStore(ControlPlaneStore): + """Control-plane state in PostgreSQL behind the ``ControlPlaneStore`` + port. See the module docstring for the concurrency and storage-shape + contract.""" + + def __init__( + self, + dsn: str, + *, + claim_lease_seconds: float = DEFAULT_CLAIM_LEASE_SECONDS, + ) -> None: + if psycopg is None: # pragma: no cover - exercised via monkeypatch + raise RuntimeError( + "AGENTFLOW_CONTROLPLANE_STORE=postgres requires the optional " + "'psycopg' dependency (pip install psycopg[binary])." + ) + if not dsn: + raise ValueError("PostgresControlPlaneStore requires a non-empty DSN.") + self._dsn = dsn + self._claim_lease_seconds = float(claim_lease_seconds) + self._schema_ready = False + self._schema_lock = threading.Lock() + + # --- connection / schema plumbing ---------------------------------------- + + def _connect(self) -> AbstractContextManager[Any]: + # One connection = one transaction: psycopg's connection context + # manager commits on clean exit and rolls back on exception, which is + # exactly the invariant-8 semantics the port requires. + self._ensure_schema() + # Annotated hop: with psycopg absent (optional dependency), mypy sees + # the module as Any and warn_return_any would flag a bare return. + connection: AbstractContextManager[Any] = psycopg.connect(self._dsn) + return connection + + def _ensure_schema(self) -> None: + if self._schema_ready: + return + with self._schema_lock: + if self._schema_ready: + return + with psycopg.connect(self._dsn) as conn: + for statement in _SCHEMA_STATEMENTS: + conn.execute(statement) + # Once per store lifetime: the write methods below must never + # recreate a table mid-scenario (see the module docstring). + self._schema_ready = True + + # --- webhook durable delivery queue -------------------------------------- + + def enqueue_webhook_delivery( + self, + *, + webhook_id: str, + event_id: str, + tenant: str, + event_type: str, + body: str, + ) -> bool: + with self._connect() as conn: + cursor = conn.execute( + """ + INSERT INTO webhook_delivery_queue + (webhook_id, event_id, tenant, event_type, body, status, attempts, + next_attempt_at, created_at, updated_at) + VALUES (%s, %s, %s, %s, %s, 'pending', 0, now(), now(), now()) + ON CONFLICT (webhook_id, event_id) DO NOTHING + """, + (webhook_id, event_id, tenant, event_type, body), + ) + # Insert-win detection (ADR 0010 §2): rowcount is 1 only for the + # caller whose INSERT actually landed — the enqueue winner, who + # alone inline-delivers. + return bool(cursor.rowcount == 1) + + def claim_due_webhook_deliveries(self, *, limit: int) -> list[WebhookQueueRow]: + with self._connect() as conn: + rows = conn.execute( + """ + WITH due AS ( + SELECT webhook_id, event_id, created_at + FROM webhook_delivery_queue + WHERE status = 'pending' + AND (next_attempt_at IS NULL OR next_attempt_at <= now()) + AND (lease_expires_at IS NULL OR lease_expires_at <= now()) + ORDER BY created_at ASC + LIMIT %s + FOR UPDATE SKIP LOCKED + ) + UPDATE webhook_delivery_queue queue + SET lease_expires_at = now() + make_interval(secs => %s), + updated_at = now() + FROM due + WHERE queue.webhook_id = due.webhook_id + AND queue.event_id = due.event_id + RETURNING queue.webhook_id, queue.event_id, queue.tenant, + queue.event_type, queue.body, due.created_at + """, + (limit, self._claim_lease_seconds), + ).fetchall() + # UPDATE .. RETURNING does not guarantee row order; re-establish the + # oldest-first contract the dispatcher relies on. + rows.sort(key=lambda row: row[5]) + return [ + WebhookQueueRow( + webhook_id=webhook_id, + event_id=event_id, + tenant=tenant, + event_type=event_type, + body=body, + ) + for webhook_id, event_id, tenant, event_type, body, _created_at in rows + ] + + def record_webhook_delivery_outcome( + self, + *, + webhook_id: str, + event_id: str, + success: bool, + status_code: int | None, + error: str | None, + max_attempts: int, + backoff_seconds: Sequence[float], + ) -> None: + with self._connect() as conn: + if success: + conn.execute( + """ + UPDATE webhook_delivery_queue + SET status = 'delivered', last_status_code = %s, last_error = NULL, + lease_expires_at = NULL, updated_at = now() + WHERE webhook_id = %s AND event_id = %s + """, + (status_code, webhook_id, event_id), + ) + return + row = conn.execute( + "SELECT attempts FROM webhook_delivery_queue " + "WHERE webhook_id = %s AND event_id = %s FOR UPDATE", + (webhook_id, event_id), + ).fetchone() + attempts = (row[0] if row else 0) + 1 + if attempts >= max_attempts: + conn.execute( + """ + UPDATE webhook_delivery_queue + SET status = 'dead', attempts = %s, last_status_code = %s, + last_error = %s, next_attempt_at = NULL, + lease_expires_at = NULL, updated_at = now() + WHERE webhook_id = %s AND event_id = %s + """, + (attempts, status_code, error, webhook_id, event_id), + ) + return + delay = backoff_seconds[min(attempts - 1, len(backoff_seconds) - 1)] + conn.execute( + """ + UPDATE webhook_delivery_queue + SET status = 'pending', attempts = %s, last_status_code = %s, + last_error = %s, next_attempt_at = %s, + lease_expires_at = NULL, updated_at = now() + WHERE webhook_id = %s AND event_id = %s + """, + ( + attempts, + status_code, + error, + datetime.now(UTC) + timedelta(seconds=delay), + webhook_id, + event_id, + ), + ) + + def park_webhook_delivery(self, *, webhook_id: str, event_id: str, error: str) -> None: + with self._connect() as conn: + conn.execute( + """ + UPDATE webhook_delivery_queue + SET status = 'dead', last_error = %s, next_attempt_at = NULL, + lease_expires_at = NULL, updated_at = now() + WHERE webhook_id = %s AND event_id = %s + """, + (error, webhook_id, event_id), + ) + + # --- webhook delivery attempt log ---------------------------------------- + + def log_webhook_delivery( + self, + *, + delivery_id: str, + webhook_id: str, + event_id: str, + event_type: str, + attempt: int, + status_code: int | None, + success: bool, + error: str | None, + ) -> None: + with self._connect() as conn: + conn.execute( + """ + INSERT INTO webhook_deliveries ( + delivery_id, webhook_id, event_id, event_type, attempt, + status_code, success, error, delivered_at + ) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, now()) + """, + ( + delivery_id, + webhook_id, + event_id, + event_type, + attempt, + status_code, + success, + error, + ), + ) + + def get_webhook_delivery_logs(self, webhook_id: str, *, limit: int = 20) -> list[dict]: + with self._connect() as conn: + result = conn.execute( + """ + SELECT delivery_id, webhook_id, event_id, event_type, attempt, + status_code, success, error, delivered_at + FROM webhook_deliveries + WHERE webhook_id = %s + ORDER BY delivered_at DESC + LIMIT %s + """, + (webhook_id, limit), + ) + columns = [description.name for description in result.description] + return [dict(zip(columns, row, strict=False)) for row in result.fetchall()] + + # --- alert delivery history ----------------------------------------------- + + def log_alert_delivery( + self, + *, + delivery_id: str, + alert_id: str, + alert_name: str, + tenant: str, + metric: str, + current_value: float | None, + previous_value: float | None, + change_pct: float | None, + threshold: float, + condition: str, + window: str, + event_type: str, + status_code: int | None, + success: bool, + error: str | None, + payload: dict, + ) -> None: + with self._connect() as conn: + conn.execute( + """ + INSERT INTO alert_history ( + delivery_id, alert_id, alert_name, metric, current_value, + previous_value, change_pct, threshold, condition, metric_window, + tenant, event_type, status_code, success, error, payload, + triggered_at + ) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, + %s, now()) + """, + ( + delivery_id, + alert_id, + alert_name, + metric, + current_value, + previous_value, + change_pct, + threshold, + condition, + window, + tenant, + event_type, + status_code, + success, + error, + json.dumps(payload, sort_keys=True), + ), + ) + + def get_alert_delivery_history(self, alert_id: str, *, limit: int = 20) -> list[dict]: + with self._connect() as conn: + result = conn.execute( + """ + SELECT delivery_id, alert_id, alert_name, metric, current_value, + previous_value, change_pct, threshold, condition, + metric_window AS window, + tenant, event_type, status_code, success, error, payload, + triggered_at + FROM alert_history + WHERE alert_id = %s + ORDER BY triggered_at DESC + LIMIT %s + """, + (alert_id, limit), + ) + columns = [description.name for description in result.description] + records = [dict(zip(columns, row, strict=False)) for row in result.fetchall()] + for record in records: + payload = record.get("payload") + if isinstance(payload, str): + try: + record["payload"] = json.loads(payload) + except json.JSONDecodeError: + pass + return records + + # --- webhook registration repository --------------------------------------- + + def load_webhook_registrations(self) -> list[dict]: + with self._connect() as conn: + rows = conn.execute( + "SELECT record FROM webhook_registrations ORDER BY position ASC" + ).fetchall() + return [json.loads(record) for (record,) in rows] + + def save_webhook_registrations(self, registrations: list[dict]) -> None: + self._replace_record_set("webhook_registrations", registrations) + + # --- alert rule repository (mutable runtime state) ------------------------ + + def load_alert_rules(self) -> list[dict]: + with self._connect() as conn: + rows = conn.execute("SELECT record FROM alert_rules ORDER BY position ASC").fetchall() + return [json.loads(record) for (record,) in rows] + + def save_alert_rules(self, rules: list[dict]) -> None: + self._replace_record_set("alert_rules", rules) + + def _replace_record_set(self, table: str, records: list[dict]) -> None: + # Full-set save with the YAML file's replace semantics: rows missing + # from the incoming set disappear, existing rows are updated in place + # (alert_rules keeps its tick_lease_expires_at — a CRUD save must not + # release another replica's in-flight evaluation claim), new rows + # append. One transaction, so a concurrent reader never sees a + # half-written set. + ids: list[str] = [] + for record in records: + record_id = record.get("id") + if not record_id: + raise ValueError(f"{table} records require a non-empty 'id'.") + ids.append(str(record_id)) + # ``table`` is one of two module literals (see the call sites above); + # every value binds via %s. + delete_missing_sql = f"DELETE FROM {table} WHERE id != ALL(%s)" # nosec B608 + # table is a module literal (same rationale as above) + delete_all_sql = f"DELETE FROM {table}" # nosec B608 + upsert_sql = ( + # table is a module literal (same rationale as above) + f"INSERT INTO {table} (id, position, record) VALUES (%s, %s, %s) " # nosec B608 + "ON CONFLICT (id) DO UPDATE " + "SET position = EXCLUDED.position, record = EXCLUDED.record" + ) + with self._connect() as conn: + if ids: + conn.execute(delete_missing_sql, (ids,)) + else: + conn.execute(delete_all_sql) + for position, (record_id, record) in enumerate(zip(ids, records, strict=True)): + conn.execute( + upsert_sql, + (record_id, position, json.dumps(record, sort_keys=True)), + ) + + def claim_alert_tick(self, rule_id: str, *, lease_seconds: float) -> bool: + with self._connect() as conn: + cursor = conn.execute( + """ + UPDATE alert_rules + SET tick_lease_expires_at = now() + make_interval(secs => %s) + WHERE id = %s + AND (tick_lease_expires_at IS NULL OR tick_lease_expires_at <= now()) + """, + (lease_seconds, rule_id), + ) + # rowcount 0 = another replica holds this rule's tick (or the rule + # row is gone — either way, nothing to evaluate here). + return bool(cursor.rowcount == 1) + + def complete_alert_tick(self, rule_id: str, *, record: dict | None) -> None: + with self._connect() as conn: + if record is None: + conn.execute( + "UPDATE alert_rules SET tick_lease_expires_at = NULL WHERE id = %s", + (rule_id,), + ) + return + # State advance and claim release in the same transaction + # (ADR 0010 §2). + conn.execute( + """ + UPDATE alert_rules + SET record = %s, tick_lease_expires_at = NULL + WHERE id = %s + """, + (json.dumps(record, sort_keys=True), rule_id), + ) + + # --- replay outbox + dead-letter (invariant 8: one transaction) ----------- + + def ensure_outbox_schema(self) -> None: + self._ensure_schema() + + def claim_due_outbox_entries(self, *, limit: int = 100) -> list[OutboxEntry]: + with self._connect() as conn: + rows = conn.execute( + """ + WITH due AS ( + SELECT id, created_at + FROM outbox + WHERE status = 'pending' + AND (next_attempt_at IS NULL OR next_attempt_at <= now()) + AND (lease_expires_at IS NULL OR lease_expires_at <= now()) + ORDER BY created_at ASC + LIMIT %s + FOR UPDATE SKIP LOCKED + ) + UPDATE outbox + SET lease_expires_at = now() + make_interval(secs => %s) + FROM due + WHERE outbox.id = due.id + RETURNING outbox.id, outbox.event_id, outbox.payload, outbox.topic, + outbox.retry_count, due.created_at + """, + (limit, self._claim_lease_seconds), + ).fetchall() + rows.sort(key=lambda row: row[5]) + return [ + OutboxEntry( + id=row_id, event_id=event_id, payload=payload, topic=topic, retry_count=retry_count + ) + for row_id, event_id, payload, topic, retry_count, _created_at in rows + ] + + def get_pending_outbox_entry(self, outbox_id: str) -> OutboxEntry | None: + # Claim-by-id: the replay path inline-delivers the row it just + # inserted, so it must own it — if a background claimant on another + # replica got there first (rowcount 0), the replay stays pending and + # that claimant delivers it. At-least-once end to end, never twice + # from this seam. + with self._connect() as conn: + row = conn.execute( + """ + UPDATE outbox + SET lease_expires_at = now() + make_interval(secs => %s) + WHERE id = %s + AND status = 'pending' + AND (lease_expires_at IS NULL OR lease_expires_at <= now()) + RETURNING id, event_id, payload, topic, retry_count + """, + (self._claim_lease_seconds, outbox_id), + ).fetchone() + if row is None: + return None + row_id, event_id, payload, topic, retry_count = row + return OutboxEntry( + id=row_id, event_id=event_id, payload=payload, topic=topic, retry_count=retry_count + ) + + def mark_outbox_sent(self, *, outbox_id: str, event_id: str) -> None: + with self._connect() as conn: + conn.execute( + """ + UPDATE outbox + SET status = 'sent', sent_at = now(), last_error = NULL, + lease_expires_at = NULL + WHERE id = %s + """, + (outbox_id,), + ) + conn.execute( + "UPDATE dead_letter_events SET status = 'replayed' WHERE event_id = %s", + (event_id,), + ) + # Both updates share the method's transaction (invariant 8): the + # context manager commits them together or rolls both back. + + def schedule_outbox_retry( + self, + *, + outbox_id: str, + event_id: str, + retry_count: int, + error_message: str, + max_retries: int, + ) -> None: + status = "pending" + retry_delay_seconds = 2**retry_count + is_kafka_error = ( + error_message.startswith("KafkaError{") + or "Kafka message(s) were not delivered" in error_message + ) + if is_kafka_error: + retry_delay_seconds = max(retry_delay_seconds, 30) + next_attempt_at: datetime | None = datetime.now(UTC) + timedelta( + seconds=retry_delay_seconds + ) + if retry_count >= max_retries: + status = "failed" + next_attempt_at = None + with self._connect() as conn: + conn.execute( + """ + UPDATE outbox + SET status = %s, retry_count = %s, next_attempt_at = %s, + last_error = %s, lease_expires_at = NULL + WHERE id = %s + """, + (status, retry_count, next_attempt_at, error_message, outbox_id), + ) + if status == "failed": + conn.execute( + "UPDATE dead_letter_events SET status = 'failed' WHERE event_id = %s", + (event_id,), + ) + + def enqueue_outbox_replay( + self, + *, + outbox_id: str, + event_id: str, + payload: dict, + topic: str, + retry_count: int, + replayed_at: datetime, + ) -> None: + encoded_payload = json.dumps(payload) + with self._connect() as conn: + conn.execute( + """ + UPDATE dead_letter_events + SET payload = %s, status = 'replay_pending', retry_count = %s, + last_retried_at = %s + WHERE event_id = %s + """, + (encoded_payload, retry_count, replayed_at, event_id), + ) + conn.execute( + """ + INSERT INTO outbox ( + id, event_id, payload, topic, created_at, sent_at, status, + retry_count, next_attempt_at, last_error + ) + VALUES (%s, %s, %s, %s, %s, NULL, 'pending', 0, %s, NULL) + """, + (outbox_id, event_id, encoded_payload, topic, replayed_at, replayed_at), + ) + + def get_dead_letter_event_for_replay(self, event_id: str) -> dict | None: + with self._connect() as conn: + row = conn.execute( + "SELECT event_id, payload, retry_count FROM dead_letter_events WHERE event_id = %s", + (event_id,), + ).fetchone() + if row is None: + return None + return {"event_id": row[0], "payload": row[1], "retry_count": row[2]} + + def dismiss_dead_letter_event(self, event_id: str) -> None: + with self._connect() as conn: + conn.execute( + "UPDATE dead_letter_events SET status = 'dismissed' WHERE event_id = %s", + (event_id,), + ) + + def dead_letter_event_exists(self, event_id: str, tenant_id: str) -> bool: + with self._connect() as conn: + row = conn.execute( + """ + SELECT event_id + FROM dead_letter_events + WHERE event_id = %s AND COALESCE(tenant_id, 'default') = %s + """, + (event_id, tenant_id), + ).fetchone() + return row is not None + + def get_dead_letter_event(self, event_id: str, tenant_id: str) -> dict | None: + with self._connect() as conn: + row = conn.execute( + """ + SELECT event_id, event_type, payload, failure_reason, failure_detail, + received_at, retry_count, last_retried_at, status + FROM dead_letter_events + WHERE event_id = %s AND COALESCE(tenant_id, 'default') = %s + """, + (event_id, tenant_id), + ).fetchone() + if row is None: + return None + return { + "event_id": row[0], + "event_type": row[1], + "payload": row[2], + "failure_reason": row[3], + "failure_detail": row[4], + "received_at": row[5], + "retry_count": int(row[6] or 0), + "last_retried_at": row[7], + "status": row[8], + } + + def list_dead_letter_events( + self, + *, + tenant_id: str, + reason: str | None, + page: int, + page_size: int, + ) -> tuple[list[dict], int]: + # Two literal SQL branches instead of an interpolated filter clause — + # the same shape as the embedded adapter (and nothing for a SQL + # linter to squint at). + if reason is not None: + count_sql = ( + "SELECT COUNT(*) FROM dead_letter_events " + "WHERE status = 'failed' AND COALESCE(tenant_id, 'default') = %s " + "AND failure_reason = %s" + ) + page_sql = ( + "SELECT event_id, event_type, failure_reason, failure_detail, " + "received_at, retry_count, last_retried_at, status " + "FROM dead_letter_events " + "WHERE status = 'failed' AND COALESCE(tenant_id, 'default') = %s " + "AND failure_reason = %s " + "ORDER BY received_at DESC, event_id ASC LIMIT %s OFFSET %s" + ) + count_params: tuple = (tenant_id, reason) + else: + count_sql = ( + "SELECT COUNT(*) FROM dead_letter_events " + "WHERE status = 'failed' AND COALESCE(tenant_id, 'default') = %s" + ) + page_sql = ( + "SELECT event_id, event_type, failure_reason, failure_detail, " + "received_at, retry_count, last_retried_at, status " + "FROM dead_letter_events " + "WHERE status = 'failed' AND COALESCE(tenant_id, 'default') = %s " + "ORDER BY received_at DESC, event_id ASC LIMIT %s OFFSET %s" + ) + count_params = (tenant_id,) + offset = (page - 1) * page_size + with self._connect() as conn: + total_row = conn.execute(count_sql, count_params).fetchone() + total = int(total_row[0]) if total_row and total_row[0] is not None else 0 + rows = conn.execute(page_sql, (*count_params, page_size, offset)).fetchall() + items = [ + { + "event_id": row[0], + "event_type": row[1], + "failure_reason": row[2], + "failure_detail": row[3], + "received_at": row[4], + "retry_count": int(row[5] or 0), + "last_retried_at": row[6], + "status": row[7], + } + for row in rows + ] + return items, total + + def get_dead_letter_stats(self, tenant_id: str) -> dict: + with self._connect() as conn: + rows = conn.execute( + """ + SELECT failure_reason, COUNT(*) + FROM dead_letter_events + WHERE status = 'failed' + AND COALESCE(tenant_id, 'default') = %s + GROUP BY failure_reason + ORDER BY failure_reason + """, + (tenant_id,), + ).fetchall() + last_24h_row = conn.execute( + """ + SELECT COUNT(*) + FROM dead_letter_events + WHERE status = 'failed' + AND COALESCE(tenant_id, 'default') = %s + AND received_at >= now() - INTERVAL '24 hours' + """, + (tenant_id,), + ).fetchone() + trend_rows = conn.execute( + """ + SELECT DATE_TRUNC('hour', received_at) AS hour_bucket, COUNT(*) + FROM dead_letter_events + WHERE status = 'failed' + AND COALESCE(tenant_id, 'default') = %s + AND received_at >= now() - INTERVAL '24 hours' + GROUP BY hour_bucket + ORDER BY hour_bucket + """, + (tenant_id,), + ).fetchall() + return { + "counts": {str(reason): int(count) for reason, count in rows if reason is not None}, + "last_24h": int(last_24h_row[0]) if last_24h_row and last_24h_row[0] is not None else 0, + "trend": [ + { + "hour": hour.isoformat() if hasattr(hour, "isoformat") else str(hour), + "count": int(count), + } + for hour, count in trend_rows + ], + } + + # --- API usage accounting ------------------------------------------------- + + def ensure_usage_schema(self) -> None: + self._ensure_schema() + + def record_api_usage( + self, + *, + tenant: str, + key_name: str, + endpoint: str, + key_id: str | None, + key_slot: str, + ) -> None: + # Bounded retry on transient connection errors, then raise — the + # caller (record_usage) skips its audit publish on failure, exactly + # like the embedded adapter's file-lock retry loop. + last_error: Exception | None = None + for attempt in range(3): + try: + with self._connect() as conn: + conn.execute( + """ + INSERT INTO api_usage (tenant, key_name, endpoint, key_id, key_slot) + VALUES (%s, %s, %s, %s, %s) + """, + (tenant, key_name, endpoint, key_id, key_slot), + ) + return + except psycopg.OperationalError as exc: + last_error = exc + time.sleep(0.01 * (attempt + 1)) + assert last_error is not None + raise last_error + + def get_usage_by_tenant(self) -> list[dict]: + with self._connect() as conn: + rows = conn.execute( + """ + SELECT tenant, COUNT(*) AS requests_last_24h + FROM api_usage + WHERE ts >= now() - INTERVAL '24 hours' + GROUP BY tenant + ORDER BY tenant + """ + ).fetchall() + return [ + {"tenant": tenant, "requests_last_24h": int(requests_last_24h)} + for tenant, requests_last_24h in rows + ] + + def get_usage_by_key(self) -> dict[tuple[str, str], int]: + with self._connect() as conn: + rows = conn.execute( + """ + SELECT tenant, key_name, COUNT(*) AS requests_last_24h + FROM api_usage + WHERE ts >= now() - INTERVAL '24 hours' + GROUP BY tenant, key_name + """ + ).fetchall() + return { + (tenant, key_name): int(requests_last_24h) + for tenant, key_name, requests_last_24h in rows + } + + def get_old_key_usage_by_key_id(self) -> dict[str, int]: + with self._connect() as conn: + rows = conn.execute( + """ + SELECT key_id, COUNT(*) AS requests_last_hour + FROM api_usage + WHERE key_slot = 'previous' + AND ts >= now() - INTERVAL '1 hour' + AND key_id IS NOT NULL + GROUP BY key_id + """ + ).fetchall() + return {key_id: int(count) for key_id, count in rows} + + # --- API session analytics ------------------------------------------------ + + def record_api_session(self, request_id: str, record: dict) -> None: + try: + with self._connect() as conn: + conn.execute( + """ + INSERT 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 (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) + ON CONFLICT (request_id) DO UPDATE SET + tenant = EXCLUDED.tenant, + key_name = EXCLUDED.key_name, + endpoint = EXCLUDED.endpoint, + method = EXCLUDED.method, + status_code = EXCLUDED.status_code, + duration_ms = EXCLUDED.duration_ms, + cache_hit = EXCLUDED.cache_hit, + entity_type = EXCLUDED.entity_type, + entity_id = EXCLUDED.entity_id, + metric_name = EXCLUDED.metric_name, + query_engine = EXCLUDED.query_engine, + query_text = EXCLUDED.query_text + """, + ( + 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"], + ), + ) + except psycopg.Error as exc: + # Best-effort telemetry, same contract as the embedded adapter: + # log and return rather than failing the request path. + logger.warning( + "analytics_session_write_skipped", + stage="insert", + dsn=_masked_dsn(self._dsn), + request_id=request_id, + tenant=record.get("tenant"), + endpoint=record.get("endpoint"), + error=str(exc), + exc_info=True, + ) + + def get_usage_analytics(self, *, window: str = "24h", tenant: str | None = None) -> dict: + interval = _window_to_interval(window) + # Two literal SQL branches instead of an interpolated tenant clause — + # the same shape as the embedded adapter. + select_head = ( + "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)::numeric, 3) AS avg_duration_ms " + "FROM api_sessions " + "WHERE tenant IS NOT NULL AND ts >= now() - CAST(%s AS INTERVAL) " + ) + if tenant: + tenants_sql = select_head + "AND tenant = %s GROUP BY tenant ORDER BY tenant" + params: tuple = (interval, tenant) + else: + tenants_sql = select_head + "GROUP BY tenant ORDER BY tenant" + params = (interval,) + with self._connect() as conn: + rows = conn.execute(tenants_sql, params).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 = %s + AND ts >= now() - CAST(%s AS INTERVAL) + GROUP BY endpoint + ORDER BY COUNT(*) DESC, endpoint + LIMIT 3 + """, + (tenant_name, interval), + ).fetchall() + tenants.append( + { + "tenant": tenant_name, + "total_requests": int(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} + + def get_top_queries(self, *, limit: int = 10, window: str = "24h") -> dict: + interval = _window_to_interval(window) + with self._connect() as conn: + rows = conn.execute( + """ + SELECT query_text, COUNT(*) AS frequency + FROM api_sessions + WHERE query_text IS NOT NULL + AND ts >= now() - CAST(%s AS INTERVAL) + GROUP BY query_text + ORDER BY frequency DESC, query_text + LIMIT %s + """, + (interval, limit), + ).fetchall() + return { + "window": window, + "queries": [ + {"query": query_text, "count": int(frequency)} for query_text, frequency in rows + ], + } + + def get_top_entities(self, *, limit: int = 10, window: str = "24h") -> dict: + interval = _window_to_interval(window) + with self._connect() as conn: + rows = conn.execute( + """ + SELECT entity_type, entity_id, COUNT(*) AS frequency + FROM api_sessions + WHERE entity_id IS NOT NULL + AND ts >= now() - CAST(%s AS INTERVAL) + GROUP BY entity_type, entity_id + ORDER BY frequency DESC, entity_type, entity_id + LIMIT %s + """, + (interval, limit), + ).fetchall() + return { + "window": window, + "entities": [ + { + "entity_type": entity_type, + "entity_id": entity_id, + "count": int(frequency), + } + for entity_type, entity_id, frequency in rows + ], + } + + def get_latency_analytics(self, *, window: str = "24h") -> dict: + interval = _window_to_interval(window) + with self._connect() as conn: + rows = conn.execute( + """ + SELECT + endpoint, + COUNT(*) AS requests, + ROUND((percentile_cont(0.50) WITHIN GROUP (ORDER BY duration_ms))::numeric, + 3) AS p50_ms, + ROUND((percentile_cont(0.95) WITHIN GROUP (ORDER BY duration_ms))::numeric, + 3) AS p95_ms, + ROUND((percentile_cont(0.99) WITHIN GROUP (ORDER BY duration_ms))::numeric, + 3) AS p99_ms + FROM api_sessions + WHERE ts >= now() - CAST(%s AS INTERVAL) + GROUP BY endpoint + ORDER BY endpoint + """, + (interval,), + ).fetchall() + return { + "window": window, + "endpoints": [ + { + "endpoint": endpoint, + "requests": int(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 + ], + } + + def get_anomalies(self, *, window: str = "24h") -> dict: + interval = _window_to_interval(window) + with self._connect() as 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 >= now() - CAST(%s 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": int(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 + ], + } + + def get_queries_per_second_last_minute(self) -> float: + try: + with self._connect() as conn: + row = conn.execute( + """ + SELECT COUNT(*) + FROM api_sessions + WHERE ts >= now() - INTERVAL '1 minute' + """ + ).fetchone() + except psycopg.Error: + # Same degrade-to-zero contract as the embedded adapter's + # duckdb.Error guard: the admin tile shows 0.0 over failing. + return 0.0 + requests_last_minute = row[0] if row else 0 + return round(float(requests_last_minute) / 60.0, 2) + + +def _masked_dsn(dsn: str) -> str: + """DSN with any password masked, for log lines.""" + masked = re.sub(r"(password=)[^ ]+", r"\1***", dsn) + return re.sub(r"(://[^:/@]+:)[^@]+(@)", r"\1***\2", masked) + + +def resolve_postgres_store_from_env() -> PostgresControlPlaneStore: + """Build the scale-profile store from the environment (the selection + seam ``get_control_plane_store`` calls for ``postgres``). Fails loudly on + a missing DSN — silently falling back to embedded would re-open the + split-brain the render gate exists to prevent.""" + dsn = (os.getenv(CONTROL_PLANE_PG_DSN_ENV) or "").strip() + if not dsn: + raise ValueError( + "AGENTFLOW_CONTROLPLANE_STORE=postgres requires " + f"{CONTROL_PLANE_PG_DSN_ENV} to hold a PostgreSQL DSN." + ) + lease_env = (os.getenv("AGENTFLOW_CONTROLPLANE_LEASE_SECONDS") or "").strip() + if lease_env: + try: + lease_seconds = float(lease_env) + except ValueError: + raise ValueError( + "AGENTFLOW_CONTROLPLANE_LEASE_SECONDS must be a number of seconds, " + f"got {lease_env!r}." + ) from None + else: + lease_seconds = DEFAULT_CLAIM_LEASE_SECONDS + return PostgresControlPlaneStore(dsn, claim_lease_seconds=lease_seconds) diff --git a/src/serving/control_plane/store.py b/src/serving/control_plane/store.py index 9025e7e5..f4f45662 100644 --- a/src/serving/control_plane/store.py +++ b/src/serving/control_plane/store.py @@ -37,7 +37,24 @@ module imports this one to resolve the store — see ``get_control_plane_store`` below), so callers validate/serialize at the boundary, exactly like the embedded adapter's YAML round-trip already did before the port existed. The -outbox/dead-letter methods follow the same rule for the same reason. +outbox/dead-letter methods follow the same rule for the same reason, and so +does the webhook-registration repository (slice 5, below). + +Slice 5 ships the PostgreSQL adapter and closes the two per-pod gaps the +extraction slices left open: + +- **Webhook registrations** (state class 5 of the ADR's inventory — the + sharpest split-brain, a webhook registered on pod A that pod B has never + heard of) move behind ``load_webhook_registrations`` / + ``save_webhook_registrations``, mirroring the alert-rule repository: the + embedded adapter keeps the byte-compatible per-app YAML file, the + PostgreSQL adapter stores rows. +- **Alert tick single-flight** (``claim_alert_tick`` / ``complete_alert_tick``, + ADR 0010 §2): the dispatcher claims each rule before evaluating it and + completes the claim with that rule's advanced state. Rule state is + persisted *per rule*, not as a full-set save — with per-rule claims, two + replicas advancing different rules in overlapping ticks would clobber each + other's runtime state through a full-set write. The outbox/dead-letter write methods (``mark_outbox_sent``, ``schedule_outbox_retry``, ``enqueue_outbox_replay``, @@ -82,6 +99,15 @@ from fastapi import FastAPI CONTROL_PLANE_STORE_ENV = "AGENTFLOW_CONTROLPLANE_STORE" +CONTROL_PLANE_PG_DSN_ENV = "AGENTFLOW_CONTROLPLANE_PG_DSN" + + +def control_plane_store_kind() -> str: + """The configured adapter kind (``'embedded'`` when unset). Composition + seams that must branch per profile (main.py deciding whether the + outbox/auth consumers share the app-wide store) use this instead of + re-parsing the env var.""" + return (os.getenv(CONTROL_PLANE_STORE_ENV) or "embedded").strip().lower() @dataclass(frozen=True) @@ -210,6 +236,22 @@ def get_alert_delivery_history(self, alert_id: str, *, limit: int = 20) -> list[ Safe to call from a worker thread (adapters isolate the read — the embedded store opens a dedicated cursor per call, audit_30 A2).""" + # --- webhook registration repository -------------------------------------- + + @abstractmethod + def load_webhook_registrations(self) -> list[dict]: + """Return every webhook registration as a JSON-shaped record (the + caller validates each into ``WebhookRegistration``) — state class 5 of + the ADR 0010 inventory, the per-pod YAML whose split-brain motivated + the ADR. Same no-model rule as the alert-rule repository below: this + module must not import ``webhook_dispatcher`` (it imports this one).""" + + @abstractmethod + def save_webhook_registrations(self, registrations: list[dict]) -> None: + """Persist the full registration set verbatim (JSON-shaped records, + the caller's serialized ``WebhookRegistration.model_dump(mode="json")`` + output).""" + # --- alert rule repository (mutable runtime state) ------------------------ @abstractmethod @@ -223,6 +265,25 @@ def save_alert_rules(self, rules: list[dict]) -> None: """Persist the full alert-rule set verbatim (JSON-shaped records, the caller's serialized ``AlertRule.model_dump(mode="json")`` output).""" + @abstractmethod + def claim_alert_tick(self, rule_id: str, *, lease_seconds: float) -> bool: + """Claim one alert rule's evaluation tick for this worker (ADR 0010 + §2): only the claim winner evaluates and pages; a lost claim means + another pod owns this rule's tick, so N replicas never run N parallel + state machines for the same rule. The embedded adapter grants every + claim (one process); the PostgreSQL adapter takes a lease that expires + on its own — crash recovery without coordination.""" + + @abstractmethod + def complete_alert_tick(self, rule_id: str, *, record: dict | None) -> None: + """Release the rule's tick claim; when ``record`` is not ``None``, + persist that rule's advanced runtime state in the same transaction as + the release (ADR 0010 §2). ``record`` is the caller's serialized + ``AlertRule.model_dump(mode="json")``, exactly like + ``save_alert_rules`` — but scoped to one rule, so two replicas + advancing different rules in overlapping ticks cannot clobber each + other's state the way a full-set save would.""" + # --- replay outbox + dead-letter (invariant 8: one transaction) ----------- @abstractmethod @@ -398,21 +459,23 @@ def get_control_plane_store(app: FastAPI) -> ControlPlaneStore: first use and caching it on ``app.state`` (the lazy pattern mirrors ``ensure_alert_dispatcher`` so lightweight test stubs keep working). - ``AGENTFLOW_CONTROLPLANE_STORE`` selects the adapter. Fail-closed ratchet: - ``postgres`` is the ADR 0010 target profile and raises until the - ``PostgresControlPlaneStore`` adapter ships (rollout slice 5); anything - else but ``embedded`` is a configuration error. + ``AGENTFLOW_CONTROLPLANE_STORE`` selects the adapter: ``embedded`` + (default) or ``postgres`` (scale profile, slice 5 — requires + ``AGENTFLOW_CONTROLPLANE_PG_DSN`` and the optional ``psycopg`` + dependency; both fail the boot loudly when missing, never a silent + fallback to embedded). Anything else is a configuration error. """ store: ControlPlaneStore | None = getattr(app.state, "control_plane_store", None) if store is not None: return store - kind = (os.getenv(CONTROL_PLANE_STORE_ENV) or "embedded").strip().lower() + kind = control_plane_store_kind() if kind == "embedded": - # Deferred: src.serving.api.alerts.dispatcher imports this module at - # top level to resolve the store, so a module-level import here would - # cycle. Resolved lazily, exactly like the EmbeddedControlPlaneStore - # import above. + # Deferred: src.serving.api.alerts.dispatcher and webhook_dispatcher + # import this module at top level to resolve the store, so module-level + # imports here would cycle. Resolved lazily, exactly like the + # EmbeddedControlPlaneStore import above. from src.serving.api.alerts.dispatcher import get_alert_config_path + from src.serving.api.webhook_dispatcher import get_webhook_config_path from .embedded import EmbeddedControlPlaneStore @@ -422,13 +485,15 @@ def get_control_plane_store(app: FastAPI) -> ControlPlaneStore: store = EmbeddedControlPlaneStore( conn_provider=lambda: app.state.query_engine._conn, alert_rules_path_provider=lambda: get_alert_config_path(app), + webhook_registrations_path_provider=lambda: get_webhook_config_path(app), ) elif kind == "postgres": - raise NotImplementedError( - "AGENTFLOW_CONTROLPLANE_STORE=postgres is the ADR 0010 scale profile; " - "the PostgresControlPlaneStore adapter ships in rollout slice 5 — " - "until then only 'embedded' runs." - ) + # Deferred for a different reason than the embedded branch: psycopg is + # an optional dependency (the redis pattern), so the adapter module + # must not load unless this profile is actually configured. + from .postgres import resolve_postgres_store_from_env + + store = resolve_postgres_store_from_env() else: raise ValueError( f"Unknown control-plane store {kind!r} " diff --git a/tests/integration/test_control_plane_postgres_live.py b/tests/integration/test_control_plane_postgres_live.py new file mode 100644 index 00000000..bcfc9b07 --- /dev/null +++ b/tests/integration/test_control_plane_postgres_live.py @@ -0,0 +1,839 @@ +"""Live probes for ``PostgresControlPlaneStore`` (ADR 0010 rollout slice 5). + +This is the probe suite the ADR names for the slice: enqueue-win uniqueness +under parallel writers, parallel claim exclusivity, lease-expiry re-drive, +restart re-drive, outbox↔dead-letter transactional atomicity (invariant 8, +including the rollback half), alert-tick single-flight — plus a contract +parity sweep that exercises every port method against a real PostgreSQL so +the two adapters cannot drift. + +Needs a live server: set ``AGENTFLOW_TEST_PG_DSN`` (CI provides a +``postgres:17`` service; locally the standalone-PG recipe from +``docs/perf/vault-pii-governance-pg-verify-2026-07-02.md`` works). The whole +module skips when the env var is absent — the same self-skip pattern as +``test_clickhouse_backend_live.py``. +""" + +from __future__ import annotations + +import json +import os +import threading +import time +from concurrent.futures import ThreadPoolExecutor +from datetime import UTC, datetime, timedelta + +import pytest + +psycopg = pytest.importorskip("psycopg") + +from src.serving.control_plane.postgres import PostgresControlPlaneStore # noqa: E402 + +PG_DSN = os.getenv("AGENTFLOW_TEST_PG_DSN", "") + +pytestmark = pytest.mark.skipif( + not PG_DSN, + reason="AGENTFLOW_TEST_PG_DSN is not set; PostgresControlPlaneStore live probes need a server", +) + +_TABLES = ( + "webhook_delivery_queue", + "webhook_deliveries", + "alert_history", + "webhook_registrations", + "alert_rules", + "outbox", + "dead_letter_events", + "api_usage", + "api_sessions", +) + + +@pytest.fixture +def store() -> PostgresControlPlaneStore: + instance = PostgresControlPlaneStore(PG_DSN) + instance.ensure_outbox_schema() # creates the full schema + with psycopg.connect(PG_DSN) as conn: + for table in _TABLES: + conn.execute(f"TRUNCATE {table}") # noqa: S608 - table names are literals above + return instance + + +def _enqueue(store: PostgresControlPlaneStore, webhook_id: str, event_id: str) -> bool: + return store.enqueue_webhook_delivery( + webhook_id=webhook_id, + event_id=event_id, + tenant="acme", + event_type="order.created", + body=json.dumps({"event_id": event_id}), + ) + + +def _queue_row(webhook_id: str, event_id: str) -> tuple | None: + with psycopg.connect(PG_DSN) as conn: + return conn.execute( + "SELECT status, attempts, next_attempt_at, last_error, lease_expires_at " + "FROM webhook_delivery_queue WHERE webhook_id = %s AND event_id = %s", + (webhook_id, event_id), + ).fetchone() + + +def _stagger_created_at(table: str, key_column: str, key: str, offset_seconds: int) -> None: + with psycopg.connect(PG_DSN) as conn: + conn.execute( + f"UPDATE {table} SET created_at = now() + make_interval(secs => %s) " # noqa: S608 + f"WHERE {key_column} = %s", + (offset_seconds, key), + ) + + +def _seed_dead_letter(event_id: str, *, tenant_id: str = "acme", status: str = "failed") -> None: + with psycopg.connect(PG_DSN) as conn: + conn.execute( + """ + INSERT INTO dead_letter_events ( + event_id, tenant_id, event_type, payload, failure_reason, + failure_detail, received_at, retry_count, last_retried_at, status + ) VALUES (%s, %s, 'order.created', '{"a": 1}', 'semantic', 'x', now(), 0, NULL, %s) + """, + (event_id, tenant_id, status), + ) + + +def _seed_outbox(outbox_id: str, *, event_id: str = "evt-1", status: str = "pending") -> None: + with psycopg.connect(PG_DSN) as conn: + conn.execute( + """ + INSERT INTO outbox (id, event_id, payload, topic, status, retry_count, + next_attempt_at) + VALUES (%s, %s, '{"a": 1}', 'agentflow.orders', %s, 0, now()) + """, + (outbox_id, event_id, status), + ) + + +# --- ADR probe 1: enqueue-win uniqueness under parallel writers ------------------- + + +def test_parallel_enqueues_produce_exactly_one_winner(store: PostgresControlPlaneStore) -> None: + workers = 8 + barrier = threading.Barrier(workers) + + def race() -> bool: + barrier.wait() + return _enqueue(store, "wh-1", "e-contested") + + with ThreadPoolExecutor(max_workers=workers) as pool: + wins = list(pool.map(lambda _: race(), range(workers))) + + assert sum(wins) == 1 # ON CONFLICT DO NOTHING + rowcount: one inline delivery + with psycopg.connect(PG_DSN) as conn: + count = conn.execute("SELECT COUNT(*) FROM webhook_delivery_queue").fetchone()[0] + assert count == 1 + + +# --- ADR probe 2: parallel claim exclusivity --------------------------------------- + + +def test_parallel_claims_never_hand_the_same_row_to_two_workers( + store: PostgresControlPlaneStore, +) -> None: + for index in range(10): + _enqueue(store, "wh-1", f"e{index}") + workers = 4 + barrier = threading.Barrier(workers) + + def claim() -> list[str]: + barrier.wait() + return [row.event_id for row in store.claim_due_webhook_deliveries(limit=10)] + + with ThreadPoolExecutor(max_workers=workers) as pool: + batches = list(pool.map(lambda _: claim(), range(workers))) + + claimed = [event_id for batch in batches for event_id in batch] + assert len(claimed) == len(set(claimed)) # FOR UPDATE SKIP LOCKED: no double claim + assert set(claimed) == {f"e{index}" for index in range(10)} # nothing lost either + + +def test_claimed_rows_are_invisible_until_their_lease_expires( + store: PostgresControlPlaneStore, +) -> None: + _enqueue(store, "wh-1", "e1") + assert [row.event_id for row in store.claim_due_webhook_deliveries(limit=10)] == ["e1"] + # Still pending (the claim is a lease, not a state flip), but leased — + # a second worker sees nothing. + assert store.claim_due_webhook_deliveries(limit=10) == [] + status, _, _, _, lease = _queue_row("wh-1", "e1") + assert status == "pending" + assert lease is not None + + +# --- ADR probe 3: lease-expiry re-drive -------------------------------------------- + + +def test_expired_lease_makes_the_row_due_again(store: PostgresControlPlaneStore) -> None: + short_lease = PostgresControlPlaneStore(PG_DSN, claim_lease_seconds=0.4) + _enqueue(short_lease, "wh-1", "e1") + assert [row.event_id for row in short_lease.claim_due_webhook_deliveries(limit=10)] == ["e1"] + assert short_lease.claim_due_webhook_deliveries(limit=10) == [] + + time.sleep(0.6) + + # Crash recovery without coordination: the owner never reported an + # outcome, the lease ran out, any worker may claim the row again. + assert [row.event_id for row in store.claim_due_webhook_deliveries(limit=10)] == ["e1"] + + +def test_outcome_clears_the_lease_so_backoff_alone_governs_redrive( + store: PostgresControlPlaneStore, +) -> None: + _enqueue(store, "wh-1", "e1") + store.claim_due_webhook_deliveries(limit=10) + store.record_webhook_delivery_outcome( + webhook_id="wh-1", + event_id="e1", + success=False, + status_code=500, + error="boom", + max_attempts=5, + backoff_seconds=[0.1], + ) + status, attempts, next_at, _, lease = _queue_row("wh-1", "e1") + assert (status, attempts) == ("pending", 1) + assert next_at is not None + assert lease is None # the outcome released the claim + time.sleep(0.2) + assert [row.event_id for row in store.claim_due_webhook_deliveries(limit=10)] == ["e1"] + + +# --- ADR probe 4: restart re-drive -------------------------------------------------- + + +def test_pending_delivery_survives_a_new_store_instance( + store: PostgresControlPlaneStore, +) -> None: + _enqueue(store, "wh-1", "e1") + del store # simulate process exit; only the PostgreSQL rows remain + + reborn = PostgresControlPlaneStore(PG_DSN) + claimed = reborn.claim_due_webhook_deliveries(limit=10) + assert [row.event_id for row in claimed] == ["e1"] + assert claimed[0].body == json.dumps({"event_id": "e1"}) # canonical body verbatim + + +# --- webhook outcome state machine (parity with the embedded pins) ----------------- + + +def test_outcome_state_machine_backs_off_then_parks_dead( + store: PostgresControlPlaneStore, +) -> None: + _enqueue(store, "wh-1", "e1") + store.record_webhook_delivery_outcome( + webhook_id="wh-1", + event_id="e1", + success=False, + status_code=500, + error="boom", + max_attempts=2, + backoff_seconds=[10.0], + ) + status, attempts, next_at, _, _ = _queue_row("wh-1", "e1") + assert (status, attempts) == ("pending", 1) + assert next_at is not None + + store.record_webhook_delivery_outcome( + webhook_id="wh-1", + event_id="e1", + success=False, + status_code=500, + error="boom", + max_attempts=2, + backoff_seconds=[10.0], + ) + status, attempts, next_at, _, _ = _queue_row("wh-1", "e1") + assert (status, attempts) == ("dead", 2) + assert next_at is None + + +def test_outcome_success_marks_delivered_and_park_marks_dead( + store: PostgresControlPlaneStore, +) -> None: + _enqueue(store, "wh-1", "e1") + store.record_webhook_delivery_outcome( + webhook_id="wh-1", + event_id="e1", + success=True, + status_code=200, + error=None, + max_attempts=5, + backoff_seconds=[1.0], + ) + status, _, _, last_error, _ = _queue_row("wh-1", "e1") + assert (status, last_error) == ("delivered", None) + + _enqueue(store, "ghost", "e2") + store.park_webhook_delivery(webhook_id="ghost", event_id="e2", error="webhook removed") + status, _, next_at, last_error, _ = _queue_row("ghost", "e2") + assert (status, next_at, last_error) == ("dead", None, "webhook removed") + + +def test_claims_come_back_oldest_first_within_the_limit( + store: PostgresControlPlaneStore, +) -> None: + for index in range(3): + _enqueue(store, "wh-1", f"e{index}") + _stagger_created_at("webhook_delivery_queue", "event_id", f"e{index}", index) + + claimed = store.claim_due_webhook_deliveries(limit=2) + + assert [row.event_id for row in claimed] == ["e0", "e1"] + + +# --- attempt log + alert history (parity) ------------------------------------------ + + +def test_webhook_delivery_log_roundtrip_newest_first(store: PostgresControlPlaneStore) -> None: + for attempt in (1, 2): + store.log_webhook_delivery( + delivery_id=f"d{attempt}", + webhook_id="wh-1", + event_id="e1", + event_type="order.created", + attempt=attempt, + status_code=500 if attempt == 1 else 200, + success=attempt == 2, + error="boom" if attempt == 1 else None, + ) + with psycopg.connect(PG_DSN) as conn: + conn.execute( + "UPDATE webhook_deliveries SET delivered_at = now() + make_interval(secs => %s) " + "WHERE delivery_id = %s", + (attempt, f"d{attempt}"), + ) + + logs = store.get_webhook_delivery_logs("wh-1") + + assert [entry["delivery_id"] for entry in logs] == ["d2", "d1"] + assert logs[0]["success"] is True + assert store.get_webhook_delivery_logs("wh-other") == [] + assert len(store.get_webhook_delivery_logs("wh-1", limit=1)) == 1 + + +def test_alert_history_roundtrip_decodes_payload(store: PostgresControlPlaneStore) -> None: + store.log_alert_delivery( + delivery_id="d1", + alert_id="a1", + alert_name="High error rate", + tenant="acme", + metric="error_rate", + current_value=0.5, + previous_value=0.1, + change_pct=400.0, + threshold=0.1, + condition="above", + window="1h", + event_type="alert.triggered", + status_code=200, + success=True, + error=None, + payload={"alert_id": "a1", "status": "firing"}, + ) + + history = store.get_alert_delivery_history("a1") + + assert len(history) == 1 + assert history[0]["payload"] == {"alert_id": "a1", "status": "firing"} + assert history[0]["window"] == "1h" + assert store.get_alert_delivery_history("a-ghost") == [] + + +# --- webhook registration + alert rule repositories (parity) ----------------------- + + +def test_webhook_registrations_round_trip_and_full_replace( + store: PostgresControlPlaneStore, +) -> None: + first = [ + {"id": "wh-1", "url": "https://a.test/h", "tenant": "acme", "active": True}, + {"id": "wh-2", "url": "https://b.test/h", "tenant": "beta", "active": True}, + ] + store.save_webhook_registrations(first) + assert store.load_webhook_registrations() == first + + # Full-set save has the YAML file's replace semantics: wh-2 disappears. + second = [{"id": "wh-1", "url": "https://a.test/h", "tenant": "acme", "active": False}] + store.save_webhook_registrations(second) + assert store.load_webhook_registrations() == second + + store.save_webhook_registrations([]) + assert store.load_webhook_registrations() == [] + + +def test_alert_rules_round_trip_preserves_order(store: PostgresControlPlaneStore) -> None: + rules = [ + {"id": "a2", "name": "second-created-first", "state": "firing"}, + {"id": "a1", "name": "listed-after", "state": "ok"}, + ] + + store.save_alert_rules(rules) + + assert store.load_alert_rules() == rules # position column, not id order + + +def test_record_sets_require_ids(store: PostgresControlPlaneStore) -> None: + with pytest.raises(ValueError, match="'id'"): + store.save_alert_rules([{"name": "no id"}]) + + +# --- ADR probe 6: alert tick single-flight ------------------------------------------ + + +def test_alert_tick_claim_single_flights_and_releases( + store: PostgresControlPlaneStore, +) -> None: + store.save_alert_rules([{"id": "a1", "state": "ok"}]) + + assert store.claim_alert_tick("a1", lease_seconds=60.0) is True + # Second claimant (another replica's dispatcher tick) loses. + assert store.claim_alert_tick("a1", lease_seconds=60.0) is False + + # Completion releases the claim and persists the advanced state in the + # same transaction. + store.complete_alert_tick("a1", record={"id": "a1", "state": "firing"}) + assert store.load_alert_rules() == [{"id": "a1", "state": "firing"}] + assert store.claim_alert_tick("a1", lease_seconds=60.0) is True + + +def test_alert_tick_claim_expires_on_its_own(store: PostgresControlPlaneStore) -> None: + store.save_alert_rules([{"id": "a1", "state": "ok"}]) + assert store.claim_alert_tick("a1", lease_seconds=0.4) is True + assert store.claim_alert_tick("a1", lease_seconds=0.4) is False + time.sleep(0.6) + # A crashed claim owner silences a rule only until the lease runs out. + assert store.claim_alert_tick("a1", lease_seconds=0.4) is True + + +def test_crud_save_does_not_release_an_in_flight_tick_claim( + store: PostgresControlPlaneStore, +) -> None: + store.save_alert_rules([{"id": "a1", "state": "ok"}]) + assert store.claim_alert_tick("a1", lease_seconds=60.0) is True + + # A concurrent CRUD full-set save (update_alert / deactivate_alert on + # another pod) must not hand this rule's tick to a second evaluator. + store.save_alert_rules([{"id": "a1", "state": "ok", "name": "renamed"}]) + + assert store.claim_alert_tick("a1", lease_seconds=60.0) is False + + +def test_complete_alert_tick_without_record_only_releases( + store: PostgresControlPlaneStore, +) -> None: + store.save_alert_rules([{"id": "a1", "state": "ok"}]) + assert store.claim_alert_tick("a1", lease_seconds=60.0) is True + + store.complete_alert_tick("a1", record=None) + + assert store.load_alert_rules() == [{"id": "a1", "state": "ok"}] # untouched + assert store.claim_alert_tick("a1", lease_seconds=60.0) is True # released + + +# --- ADR probe 5: outbox↔dead-letter atomicity (invariant 8) ------------------------ + + +def test_mark_outbox_sent_flips_both_rows_in_one_transaction( + store: PostgresControlPlaneStore, +) -> None: + _seed_outbox("o1", event_id="e1") + _seed_dead_letter("e1") + + store.mark_outbox_sent(outbox_id="o1", event_id="e1") + + with psycopg.connect(PG_DSN) as conn: + outbox_status = conn.execute("SELECT status FROM outbox WHERE id = 'o1'").fetchone()[0] + dl_status = conn.execute( + "SELECT status FROM dead_letter_events WHERE event_id = 'e1'" + ).fetchone()[0] + assert (outbox_status, dl_status) == ("sent", "replayed") + + +def test_mark_outbox_sent_rolls_back_when_dead_letter_update_fails( + store: PostgresControlPlaneStore, +) -> None: + _seed_outbox("o1", event_id="e1") + with psycopg.connect(PG_DSN) as conn: + conn.execute("DROP TABLE dead_letter_events") + + try: + with pytest.raises(psycopg.Error): + store.mark_outbox_sent(outbox_id="o1", event_id="e1") + + # The transaction rolled back: the outbox flip did not survive alone. + with psycopg.connect(PG_DSN) as conn: + status = conn.execute("SELECT status FROM outbox WHERE id = 'o1'").fetchone()[0] + assert status == "pending" + finally: + PostgresControlPlaneStore(PG_DSN).ensure_outbox_schema() # restore the table + + +def test_enqueue_outbox_replay_rolls_back_when_outbox_insert_fails( + store: PostgresControlPlaneStore, +) -> None: + _seed_dead_letter("e1", status="failed") + with psycopg.connect(PG_DSN) as conn: + conn.execute("DROP TABLE outbox") + + try: + with pytest.raises(psycopg.Error): + store.enqueue_outbox_replay( + outbox_id="o1", + event_id="e1", + payload={"event_id": "e1"}, + topic="events.raw", + retry_count=1, + replayed_at=datetime.now(UTC), + ) + + with psycopg.connect(PG_DSN) as conn: + status = conn.execute( + "SELECT status FROM dead_letter_events WHERE event_id = 'e1'" + ).fetchone()[0] + assert status == "failed" # the dead-letter flip rolled back too + finally: + PostgresControlPlaneStore(PG_DSN).ensure_outbox_schema() + + +def test_enqueue_outbox_replay_marks_pending_and_inserts_in_one_transaction( + store: PostgresControlPlaneStore, +) -> None: + _seed_dead_letter("e1", status="failed") + replayed_at = datetime.now(UTC) + + store.enqueue_outbox_replay( + outbox_id="o1", + event_id="e1", + payload={"event_id": "e1", "total_amount": "9.99"}, + topic="events.raw", + retry_count=1, + replayed_at=replayed_at, + ) + + with psycopg.connect(PG_DSN) as conn: + dl_status, dl_retry = conn.execute( + "SELECT status, retry_count FROM dead_letter_events WHERE event_id = 'e1'" + ).fetchone() + outbox_row = conn.execute( + "SELECT event_id, topic, status FROM outbox WHERE id = 'o1'" + ).fetchone() + assert (dl_status, dl_retry) == ("replay_pending", 1) + assert outbox_row == ("e1", "events.raw", "pending") + + +def test_schedule_outbox_retry_backs_off_then_fails_and_dead_letters( + store: PostgresControlPlaneStore, +) -> None: + _seed_outbox("o1", event_id="e1") + _seed_dead_letter("e1") + + store.schedule_outbox_retry( + outbox_id="o1", event_id="e1", retry_count=1, error_message="boom", max_retries=2 + ) + with psycopg.connect(PG_DSN) as conn: + status, next_at = conn.execute( + "SELECT status, next_attempt_at FROM outbox WHERE id = 'o1'" + ).fetchone() + assert status == "pending" + assert next_at is not None + + store.schedule_outbox_retry( + outbox_id="o1", event_id="e1", retry_count=2, error_message="boom", max_retries=2 + ) + with psycopg.connect(PG_DSN) as conn: + status, next_at = conn.execute( + "SELECT status, next_attempt_at FROM outbox WHERE id = 'o1'" + ).fetchone() + dl_status = conn.execute( + "SELECT status FROM dead_letter_events WHERE event_id = 'e1'" + ).fetchone()[0] + assert (status, next_at, dl_status) == ("failed", None, "failed") + + +def test_schedule_outbox_retry_floors_kafka_shaped_errors_at_30s( + store: PostgresControlPlaneStore, +) -> None: + _seed_outbox("o1", event_id="e1") + + store.schedule_outbox_retry( + outbox_id="o1", + event_id="e1", + retry_count=1, + error_message="KafkaError{code=_MSG_TIMED_OUT}", + max_retries=5, + ) + + with psycopg.connect(PG_DSN) as conn: + next_at = conn.execute("SELECT next_attempt_at FROM outbox WHERE id = 'o1'").fetchone()[0] + assert next_at >= datetime.now(UTC) + timedelta(seconds=20) + + +def test_outbox_claims_are_leased_and_claim_by_id_is_exclusive( + store: PostgresControlPlaneStore, +) -> None: + for index in range(3): + _seed_outbox(f"o{index}", event_id=f"e{index}") + _stagger_created_at("outbox", "id", f"o{index}", index) + + claimed = store.claim_due_outbox_entries(limit=2) + assert [entry.id for entry in claimed] == ["o0", "o1"] + assert claimed[0].topic == "agentflow.orders" + + # o0/o1 are leased; only o2 is left for a second claimant. + assert [entry.id for entry in store.claim_due_outbox_entries(limit=10)] == ["o2"] + + # Claim-by-id (the replay inline path): everything is leased now. + assert store.get_pending_outbox_entry("o0") is None + store.mark_outbox_sent(outbox_id="o0", event_id="e0") + assert store.get_pending_outbox_entry("o0") is None # sent, not pending + + # A freshly inserted replay row is claimable by id exactly once. + _seed_outbox("o-replay", event_id="e-replay") + entry = store.get_pending_outbox_entry("o-replay") + assert entry is not None + assert entry.event_id == "e-replay" + assert store.get_pending_outbox_entry("o-replay") is None + + +# --- dead-letter reads (parity) ------------------------------------------------------ + + +def test_dead_letter_reads_are_tenant_scoped_and_paginate( + store: PostgresControlPlaneStore, +) -> None: + _seed_dead_letter("e1", tenant_id="acme") + _seed_dead_letter("e2", tenant_id="acme") + _seed_dead_letter("e-beta", tenant_id="beta") + with psycopg.connect(PG_DSN) as conn: + conn.execute( + "UPDATE dead_letter_events SET failure_reason = 'schema' WHERE event_id = 'e2'" + ) + + assert store.dead_letter_event_exists("e1", "acme") is True + assert store.dead_letter_event_exists("e1", "beta") is False + + record = store.get_dead_letter_event("e1", "acme") + assert record is not None + assert record["failure_reason"] == "semantic" + assert store.get_dead_letter_event("e1", "beta") is None + + items, total = store.list_dead_letter_events(tenant_id="acme", reason=None, page=1, page_size=1) + assert total == 2 + assert len(items) == 1 + + items, total = store.list_dead_letter_events( + tenant_id="acme", reason="schema", page=1, page_size=10 + ) + assert total == 1 + assert items[0]["event_id"] == "e2" + + stats = store.get_dead_letter_stats("acme") + assert stats["counts"] == {"semantic": 1, "schema": 1} + assert stats["last_24h"] == 2 + assert len(stats["trend"]) >= 1 + + replay_row = store.get_dead_letter_event_for_replay("e1") + assert replay_row is not None + assert json.loads(replay_row["payload"]) == {"a": 1} + assert store.get_dead_letter_event_for_replay("ghost") is None + + store.dismiss_dead_letter_event("e1") + with psycopg.connect(PG_DSN) as conn: + status = conn.execute( + "SELECT status FROM dead_letter_events WHERE event_id = 'e1'" + ).fetchone()[0] + assert status == "dismissed" + + +# --- usage accounting + session analytics (parity) ---------------------------------- + + +def test_usage_roundtrip_by_tenant_key_and_old_key_slot( + store: PostgresControlPlaneStore, +) -> None: + store.ensure_usage_schema() + for endpoint in ("/v1/query", "/v1/entity"): + store.record_api_usage( + tenant="acme", + key_name="agent", + endpoint=endpoint, + key_id="k1", + key_slot="current", + ) + store.record_api_usage( + tenant="beta", key_name="etl", endpoint="/v1/query", key_id="k-old", key_slot="previous" + ) + + assert store.get_usage_by_tenant() == [ + {"tenant": "acme", "requests_last_24h": 2}, + {"tenant": "beta", "requests_last_24h": 1}, + ] + assert store.get_usage_by_key() == {("acme", "agent"): 2, ("beta", "etl"): 1} + assert store.get_old_key_usage_by_key_id() == {"k-old": 1} + + +def _session_record(**overrides: object) -> dict: + record = { + "tenant": "acme", + "key_name": "agent", + "endpoint": "/v1/query", + "method": "POST", + "status_code": 200, + "duration_ms": 12.5, + "cache_hit": False, + "entity_type": None, + "entity_id": None, + "metric_name": None, + "query_engine": "duckdb", + "query_text": "revenue today", + } + record.update(overrides) + return record + + +def test_session_writes_are_idempotent_per_request_id( + store: PostgresControlPlaneStore, +) -> None: + store.record_api_session("r1", _session_record()) + # A retried background write must not double-count (insert-or-replace). + store.record_api_session("r1", _session_record(status_code=500)) + + with psycopg.connect(PG_DSN) as conn: + rows = conn.execute("SELECT status_code FROM api_sessions").fetchall() + assert rows == [(500,)] + + +def test_session_analytics_windows_and_shapes(store: PostgresControlPlaneStore) -> None: + store.record_api_session("r1", _session_record()) + store.record_api_session("r2", _session_record(status_code=500, cache_hit=True)) + store.record_api_session( + "r3", + _session_record( + tenant="beta", + endpoint="/v1/entity/order/ORD-1", + entity_type="order", + entity_id="ORD-1", + query_text=None, + ), + ) + + usage = store.get_usage_analytics(window="1h") + assert usage["window"] == "1h" + acme = next(item for item in usage["tenants"] if item["tenant"] == "acme") + assert acme["total_requests"] == 2 + assert acme["error_rate"] == pytest.approx(0.5) + assert acme["cache_hit_rate"] == pytest.approx(0.5) + assert acme["top_endpoints"] == ["/v1/query"] + + scoped = store.get_usage_analytics(window="1h", tenant="beta") + assert [item["tenant"] for item in scoped["tenants"]] == ["beta"] + + top_queries = store.get_top_queries(window="1h") + assert top_queries["queries"][0] == {"query": "revenue today", "count": 2} + + top_entities = store.get_top_entities(window="1h") + assert top_entities["entities"][0] == { + "entity_type": "order", + "entity_id": "ORD-1", + "count": 1, + } + + latency = store.get_latency_analytics(window="1h") + endpoints = {item["endpoint"]: item for item in latency["endpoints"]} + assert endpoints["/v1/query"]["requests"] == 2 + assert endpoints["/v1/query"]["p50_ms"] == pytest.approx(12.5) + + anomalies = store.get_anomalies(window="24h") + assert anomalies["anomalies"] == [] # no history to spike against + + assert store.get_queries_per_second_last_minute() == pytest.approx(3 / 60.0, abs=0.01) + + with pytest.raises(ValueError, match="Invalid window"): + store.get_usage_analytics(window="fortnight") + + +def test_qps_degrades_to_zero_when_the_server_is_unreachable() -> None: + unreachable = PostgresControlPlaneStore( + "postgresql://nobody@127.0.0.1:1/agentflow?connect_timeout=1" + ) + assert unreachable.get_queries_per_second_last_minute() == 0.0 + + +# --- end to end: the app on the postgres profile ------------------------------------ + + +def test_app_on_postgres_profile_shares_state_across_boots( + store: PostgresControlPlaneStore, monkeypatch: pytest.MonkeyPatch +) -> None: + """The split-brain the ADR set out to kill, demonstrated dead: a webhook + registered through one app boot is visible to a *fresh* boot (a second + pod, in production terms), because registrations, usage and sessions all + live in PostgreSQL — no per-pod YAML, no per-pod DuckDB file.""" + from datetime import datetime as _datetime + + from fastapi.testclient import TestClient + + from src.serving.api.auth import TenantKey + from src.serving.api.main import app + + monkeypatch.setenv("AGENTFLOW_CONTROLPLANE_STORE", "postgres") + monkeypatch.setenv("AGENTFLOW_CONTROLPLANE_PG_DSN", PG_DSN) + + previous_webhook_autostart = getattr(app.state, "webhook_dispatcher_autostart", True) + previous_alert_autostart = getattr(app.state, "alert_dispatcher_autostart", True) + app.state.webhook_dispatcher_autostart = False + app.state.alert_dispatcher_autostart = False + + def _authenticate(client: TestClient) -> None: + manager = client.app.state.auth_manager + manager.keys_by_value = { + "acme-key": TenantKey( + key="acme-key", + name="acme-agent", + tenant="acme", + rate_limit_rpm=100, + allowed_entity_types=None, + created_at=_datetime.now(UTC).date(), + ) + } + manager._rate_windows.clear() + + try: + with TestClient(app) as first_boot: + assert isinstance(first_boot.app.state.control_plane_store, PostgresControlPlaneStore) + # AuthManager shares the app-wide store on this profile (slice 5 + # injection in main.py) — usage/sessions land in PostgreSQL. + assert first_boot.app.state.auth_manager.store is ( + first_boot.app.state.control_plane_store + ) + _authenticate(first_boot) + response = first_boot.post( + "/v1/webhooks", + headers={"X-API-Key": "acme-key"}, + json={"url": "http://agent.test/webhook", "filters": {}}, + ) + assert response.status_code == 201 + webhook_id = response.json()["id"] + + with TestClient(app) as second_boot: + _authenticate(second_boot) + response = second_boot.get("/v1/webhooks", headers={"X-API-Key": "acme-key"}) + assert response.status_code == 200 + assert [item["id"] for item in response.json()["webhooks"]] == [webhook_id] + finally: + app.state.webhook_dispatcher_autostart = previous_webhook_autostart + app.state.alert_dispatcher_autostart = previous_alert_autostart + + with psycopg.connect(PG_DSN) as conn: + registrations = conn.execute("SELECT COUNT(*) FROM webhook_registrations").fetchone()[0] + usage_tenants = conn.execute("SELECT DISTINCT tenant FROM api_usage").fetchall() + assert registrations == 1 + assert usage_tenants == [("acme",)] # request accounting went to PostgreSQL too diff --git a/tests/unit/test_analytics_middleware.py b/tests/unit/test_analytics_middleware.py index 71c3e061..0827c52a 100644 --- a/tests/unit/test_analytics_middleware.py +++ b/tests/unit/test_analytics_middleware.py @@ -17,6 +17,7 @@ async def test_analytics_middleware_does_not_read_body_for_get_requests(tmp_path app = FastAPI() app.state.auth_manager = SimpleNamespace( db_path=tmp_path / "usage.duckdb", + store=analytics_module._usage_store(tmp_path / "usage.duckdb"), has_configured_keys=lambda: True, ) middleware = build_analytics_middleware() @@ -54,6 +55,7 @@ async def test_analytics_middleware_skips_logging_when_auth_is_open(tmp_path: Pa app = FastAPI() app.state.auth_manager = SimpleNamespace( db_path=tmp_path / "usage.duckdb", + store=analytics_module._usage_store(tmp_path / "usage.duckdb"), has_configured_keys=lambda: False, ) middleware = build_analytics_middleware() @@ -147,6 +149,7 @@ async def test_analytics_middleware_skips_unauthenticated_request(tmp_path: Path app = FastAPI() app.state.auth_manager = SimpleNamespace( db_path=tmp_path / "usage.duckdb", + store=analytics_module._usage_store(tmp_path / "usage.duckdb"), has_configured_keys=lambda: True, ) middleware = build_analytics_middleware() @@ -168,6 +171,7 @@ async def test_analytics_middleware_records_authenticated_request(tmp_path: Path app = FastAPI() app.state.auth_manager = SimpleNamespace( db_path=tmp_path / "usage.duckdb", + store=analytics_module._usage_store(tmp_path / "usage.duckdb"), has_configured_keys=lambda: True, ) middleware = build_analytics_middleware() diff --git a/tests/unit/test_blocking_routes_offloaded.py b/tests/unit/test_blocking_routes_offloaded.py index e5383ccb..69ef415f 100644 --- a/tests/unit/test_blocking_routes_offloaded.py +++ b/tests/unit/test_blocking_routes_offloaded.py @@ -15,7 +15,6 @@ import asyncio import time from datetime import UTC, datetime -from pathlib import Path from types import SimpleNamespace import httpx @@ -174,8 +173,7 @@ async def test_alert_history_does_not_block_event_loop(monkeypatch: pytest.Monke @pytest.mark.asyncio async def test_webhook_logs_does_not_block_event_loop(monkeypatch: pytest.MonkeyPatch) -> None: - # The registration lookup is a YAML read, not the DuckDB scan under test. - monkeypatch.setattr(webhooks_module, "get_webhook_config_path", lambda app: Path("unused")) + # The registration lookup is a store read, not the DuckDB scan under test. monkeypatch.setattr(webhooks_module, "get_webhook", lambda *a, **k: SimpleNamespace(id="W1")) app = FastAPI() app.state.query_engine = SimpleNamespace(_conn=_SleepyHistoryConn(delay_seconds=0.3)) diff --git a/tests/unit/test_control_plane_store.py b/tests/unit/test_control_plane_store.py index 3eb8733e..38f871d8 100644 --- a/tests/unit/test_control_plane_store.py +++ b/tests/unit/test_control_plane_store.py @@ -21,6 +21,7 @@ import pytest from src.serving.control_plane import ( + CONTROL_PLANE_PG_DSN_ENV, CONTROL_PLANE_STORE_ENV, EmbeddedControlPlaneStore, get_control_plane_store, @@ -317,6 +318,157 @@ def test_alert_rules_methods_require_a_path_provider(store: EmbeddedControlPlane store.load_alert_rules() +# --- webhook registration repository (YAML round-trip, slice 5) ------------------- + + +@pytest.fixture +def registration_store(tmp_path: Path) -> EmbeddedControlPlaneStore: + path = tmp_path / "webhooks.yaml" + return EmbeddedControlPlaneStore(webhook_registrations_path_provider=lambda: path) + + +def test_load_webhook_registrations_empty_when_file_is_missing( + registration_store: EmbeddedControlPlaneStore, +) -> None: + assert registration_store.load_webhook_registrations() == [] + + +def test_save_then_load_webhook_registrations_round_trips_verbatim( + registration_store: EmbeddedControlPlaneStore, +) -> None: + registrations = [ + {"id": "wh-1", "url": "https://a.test/h", "tenant": "acme", "active": True}, + {"id": "wh-2", "url": "https://b.test/h", "tenant": "beta", "active": False}, + ] + + registration_store.save_webhook_registrations(registrations) + + assert registration_store.load_webhook_registrations() == registrations + + +def test_load_webhook_registrations_reads_the_pre_port_yaml_shape(tmp_path: Path) -> None: + # Byte-compatibility pin: a config/webhooks.yaml written by the pre-port + # save_webhooks (a top-level ``webhooks:`` list) loads unchanged. + path = tmp_path / "webhooks.yaml" + path.write_text( + "webhooks:\n- id: wh-1\n url: https://a.test/h\n tenant: acme\n active: true\n", + encoding="utf-8", + ) + store = EmbeddedControlPlaneStore(webhook_registrations_path_provider=lambda: path) + + assert store.load_webhook_registrations() == [ + {"id": "wh-1", "url": "https://a.test/h", "tenant": "acme", "active": True} + ] + + +def test_webhook_registration_methods_require_a_path_provider( + store: EmbeddedControlPlaneStore, +) -> None: + with pytest.raises(RuntimeError, match="webhook_registrations_path_provider"): + store.load_webhook_registrations() + + +# --- alert tick claims (slice 5) --------------------------------------------------- + + +def test_embedded_claim_alert_tick_always_grants( + alert_store: EmbeddedControlPlaneStore, +) -> None: + # One process, one dispatcher loop: the embedded adapter satisfies the + # single-flight contract degenerately, like its claim_due siblings. + assert alert_store.claim_alert_tick("a1", lease_seconds=60.0) is True + assert alert_store.claim_alert_tick("a1", lease_seconds=60.0) is True + + +def test_embedded_complete_alert_tick_persists_only_that_rule( + alert_store: EmbeddedControlPlaneStore, +) -> None: + alert_store.save_alert_rules([{"id": "a1", "state": "ok"}, {"id": "a2", "state": "ok"}]) + + alert_store.complete_alert_tick("a1", record={"id": "a1", "state": "firing"}) + + assert alert_store.load_alert_rules() == [ + {"id": "a1", "state": "firing"}, + {"id": "a2", "state": "ok"}, + ] + + +def test_embedded_complete_alert_tick_without_record_is_a_no_op( + alert_store: EmbeddedControlPlaneStore, +) -> None: + alert_store.save_alert_rules([{"id": "a1", "state": "ok"}]) + + alert_store.complete_alert_tick("a1", record=None) + + assert alert_store.load_alert_rules() == [{"id": "a1", "state": "ok"}] + + +@pytest.mark.asyncio +async def test_dispatch_alerts_single_flights_rules_through_the_claim() -> None: + """ADR 0010 §2 wiring pin: the dispatcher evaluates only the rules whose + tick it claimed, and completes every claim it took — with the advanced + record when the rule changed, with ``None`` when it did not.""" + from src.serving.api.alerts import dispatcher as dispatcher_module + from src.serving.api.alerts import escalation as escalation_module + from src.serving.api.alerts.dispatcher import AlertDispatcher, AlertRule + + now = datetime.now(UTC) + rules = [ + AlertRule( + id=f"a{index}", + name=f"rule {index}", + tenant="acme", + metric="error_rate", + window="1h", + condition="above", + threshold=0.1, + webhook_url="https://example.test/hook", + secret="s", + created_at=now, + updated_at=now, + ) + for index in range(3) + ] + + class _ClaimingStubStore: + def __init__(self) -> None: + self.claims: list[str] = [] + self.completions: list[tuple[str, bool]] = [] + + def load_alert_rules(self) -> list[dict]: + return [rule.model_dump(mode="json") for rule in rules] + + def claim_alert_tick(self, rule_id: str, *, lease_seconds: float) -> bool: + self.claims.append(rule_id) + return rule_id != "a1" # a1's tick belongs to "another replica" + + def complete_alert_tick(self, rule_id: str, *, record: dict | None) -> None: + self.completions.append((rule_id, record is not None)) + + stub_store = _ClaimingStubStore() + app = SimpleNamespace(state=SimpleNamespace(control_plane_store=stub_store)) + dispatcher = AlertDispatcher(app) # type: ignore[arg-type] + + evaluated: list[str] = [] + + async def _fake_dispatch_alert(dispatcher_arg, alert, now_arg): + evaluated.append(alert.id) + # a2 advances state; a0 does not. + return alert, alert.id == "a2", 0 + + original = escalation_module.dispatch_alert + escalation_module.dispatch_alert = _fake_dispatch_alert # type: ignore[assignment] + try: + await dispatcher.dispatch_alerts() + finally: + escalation_module.dispatch_alert = original # type: ignore[assignment] + del dispatcher_module # imported for parity with the dispatcher under test + + assert stub_store.claims == ["a0", "a1", "a2"] + assert evaluated == ["a0", "a2"] # the lost claim was never evaluated + assert stub_store.completions == [("a0", False), ("a2", True)] + + # --- replay outbox + dead-letter (invariant 8) ------------------------------------ @@ -595,14 +747,59 @@ def test_get_store_defaults_to_embedded_and_caches_on_app_state( assert get_control_plane_store(app) is store # type: ignore[arg-type] -def test_get_store_postgres_is_a_fail_closed_ratchet_until_slice_5( +def test_get_store_postgres_requires_a_dsn( conn: duckdb.DuckDBPyConnection, monkeypatch: pytest.MonkeyPatch ) -> None: # ADR 0010: the scale profile must not silently fall back to the embedded - # (split-brain at replicaCount>1) store — it fails the boot instead, and - # this test is deleted only when PostgresControlPlaneStore ships. + # (split-brain at replicaCount>1) store — a missing DSN fails the boot. + monkeypatch.setenv(CONTROL_PLANE_STORE_ENV, "postgres") + monkeypatch.delenv(CONTROL_PLANE_PG_DSN_ENV, raising=False) + with pytest.raises(ValueError, match=CONTROL_PLANE_PG_DSN_ENV): + get_control_plane_store(_stub_app(conn)) # type: ignore[arg-type] + + +def test_get_store_postgres_resolves_the_adapter_and_caches_it( + conn: duckdb.DuckDBPyConnection, monkeypatch: pytest.MonkeyPatch +) -> None: + from src.serving.control_plane import postgres as postgres_module + from src.serving.control_plane.postgres import PostgresControlPlaneStore + + # The selection seam is under test, not psycopg: stub the module object so + # this resolves identically whether or not the optional dependency is + # installed (the CI unit job installs no optional extras). + monkeypatch.setattr(postgres_module, "psycopg", SimpleNamespace()) + monkeypatch.setenv(CONTROL_PLANE_STORE_ENV, "postgres") + monkeypatch.setenv(CONTROL_PLANE_PG_DSN_ENV, "postgresql://cp@localhost:5432/agentflow") + app = _stub_app(conn) + + store = get_control_plane_store(app) # type: ignore[arg-type] + + # Construction is connection-free (schema DDL runs on first method use), + # so resolution succeeds without a live server. + assert isinstance(store, PostgresControlPlaneStore) + assert app.state.control_plane_store is store + assert get_control_plane_store(app) is store # type: ignore[arg-type] + + +def test_get_store_postgres_fails_loudly_without_psycopg( + conn: duckdb.DuckDBPyConnection, monkeypatch: pytest.MonkeyPatch +) -> None: + from src.serving.control_plane import postgres as postgres_module + + monkeypatch.setenv(CONTROL_PLANE_STORE_ENV, "postgres") + monkeypatch.setenv(CONTROL_PLANE_PG_DSN_ENV, "postgresql://cp@localhost:5432/agentflow") + monkeypatch.setattr(postgres_module, "psycopg", None) + with pytest.raises(RuntimeError, match="psycopg"): + get_control_plane_store(_stub_app(conn)) # type: ignore[arg-type] + + +def test_get_store_postgres_rejects_a_malformed_lease_override( + conn: duckdb.DuckDBPyConnection, monkeypatch: pytest.MonkeyPatch +) -> None: monkeypatch.setenv(CONTROL_PLANE_STORE_ENV, "postgres") - with pytest.raises(NotImplementedError, match="slice 5"): + monkeypatch.setenv(CONTROL_PLANE_PG_DSN_ENV, "postgresql://cp@localhost:5432/agentflow") + monkeypatch.setenv("AGENTFLOW_CONTROLPLANE_LEASE_SECONDS", "soon") + with pytest.raises(ValueError, match="LEASE_SECONDS"): get_control_plane_store(_stub_app(conn)) # type: ignore[arg-type] diff --git a/tests/unit/test_security_tooling_policy.py b/tests/unit/test_security_tooling_policy.py index 4a30728c..a3e248c2 100644 --- a/tests/unit/test_security_tooling_policy.py +++ b/tests/unit/test_security_tooling_policy.py @@ -59,6 +59,12 @@ def test_sql_injection_checks_are_not_globally_suppressed() -> None: "src/processing/clickhouse_sink.py": 2, "src/serving/api/routers/lineage.py": 1, "src/serving/api/routers/slo.py": 4, + # ADR 0010 slice 5 (reviewed 2026-07-03): _replace_record_set interpolates + # only its `table` argument, a module literal at exactly two call sites + # (save_webhook_registrations / save_alert_rules); every value binds via + # %s. All other adapter SQL is literal (the lease fragment is inlined and + # the tenant/reason filters branch into full literal statements). + "src/serving/control_plane/postgres.py": 3, "src/serving/backends/clickhouse_backend.py": 7, "src/serving/backends/duckdb_backend.py": 2, "src/serving/semantic_layer/nl_engine.py": 6, diff --git a/tests/unit/test_webhook_dispatcher_unit.py b/tests/unit/test_webhook_dispatcher_unit.py index f884622e..365a6599 100644 --- a/tests/unit/test_webhook_dispatcher_unit.py +++ b/tests/unit/test_webhook_dispatcher_unit.py @@ -93,54 +93,66 @@ def config_path(tmp_path: Path) -> Path: return tmp_path / "webhooks.yaml" +def _registry_app(config_path: Path) -> SimpleNamespace: + # Registration CRUD resolves the control-plane store from the app + # (ADR 0010 slice 5); the embedded store persists registrations to the + # app's webhook_config_path YAML, exactly like the pre-port path-based + # helpers did. + return SimpleNamespace(state=SimpleNamespace(webhook_config_path=config_path)) + + def test_create_then_load_and_list_roundtrip(config_path: Path) -> None: + app = _registry_app(config_path) created = create_webhook( - config_path, + app, url="https://example.test/hook", tenant="acme", filters=WebhookFilters(event_types=["order"]), ) assert created.secret # a secret is generated - assert load_webhooks(config_path) # persisted - listed = list_webhooks(config_path, "acme") + assert config_path.exists() # persisted to the embedded profile's YAML + assert load_webhooks(app) + listed = list_webhooks(app, "acme") assert [w.id for w in listed] == [created.id] # tenant isolation: another tenant sees nothing - assert list_webhooks(config_path, "other") == [] + assert list_webhooks(app, "other") == [] def test_get_webhook_respects_tenant_and_activity(config_path: Path) -> None: + app = _registry_app(config_path) created = create_webhook( - config_path, + app, url="https://example.test/hook", tenant="acme", filters=WebhookFilters(), ) - assert get_webhook(config_path, created.id, "acme") is not None - assert get_webhook(config_path, created.id, "other") is None - assert get_webhook(config_path, "missing-id", "acme") is None + assert get_webhook(app, created.id, "acme") is not None + assert get_webhook(app, created.id, "other") is None + assert get_webhook(app, "missing-id", "acme") is None def test_deactivate_hides_webhook(config_path: Path) -> None: + app = _registry_app(config_path) created = create_webhook( - config_path, + app, url="https://example.test/hook", tenant="acme", filters=WebhookFilters(), ) - assert deactivate_webhook(config_path, created.id, "acme") is True - assert list_webhooks(config_path, "acme") == [] + assert deactivate_webhook(app, created.id, "acme") is True + assert list_webhooks(app, "acme") == [] # second deactivation is a no-op - assert deactivate_webhook(config_path, created.id, "acme") is False + assert deactivate_webhook(app, created.id, "acme") is False def test_load_webhooks_missing_or_empty_returns_empty(tmp_path: Path) -> None: - assert load_webhooks(tmp_path / "absent.yaml") == [] + assert load_webhooks(_registry_app(tmp_path / "absent.yaml")) == [] empty = tmp_path / "empty.yaml" empty.write_text(" \n", encoding="utf-8") - assert load_webhooks(empty) == [] + assert load_webhooks(_registry_app(empty)) == [] def test_matches_filters_event_type_prefix_and_exact() -> None: @@ -327,17 +339,11 @@ async def test_dispatch_isolates_webhook_failure_and_enqueues_all( "INSERT INTO pipeline_events VALUES " "('e1', 'orders.raw', 'acme', 'order.created', NOW())" ) - wh1 = create_webhook( - config_path, url="https://a.test/h1", tenant="acme", filters=WebhookFilters() - ) - wh2 = create_webhook( - config_path, url="https://b.test/h2", tenant="acme", filters=WebhookFilters() - ) - monkeypatch.setattr( - "src.serving.api.webhook_dispatcher.get_webhook_config_path", - lambda app: config_path, - ) - dispatcher = WebhookDispatcher(_stub_app(conn)) + app = _stub_app(conn) + app.state.webhook_config_path = config_path + wh1 = create_webhook(app, url="https://a.test/h1", tenant="acme", filters=WebhookFilters()) + wh2 = create_webhook(app, url="https://b.test/h2", tenant="acme", filters=WebhookFilters()) + dispatcher = WebhookDispatcher(app) async def _deliver(webhook: object, event: dict) -> dict: if getattr(webhook, "id", None) == wh1.id: @@ -395,12 +401,11 @@ def test_record_outcome_failure_reschedules_then_dies_at_max() -> None: def test_process_delivery_queue_redrives_due_pending(tmp_path: Path) -> None: conn = duckdb.connect(":memory:") try: - config_path = tmp_path / "webhooks.yaml" + app = _stub_app(conn) + app.state.webhook_config_path = tmp_path / "webhooks.yaml" created = create_webhook( - config_path, url="https://example.test/hook", tenant="acme", filters=WebhookFilters() + app, url="https://example.test/hook", tenant="acme", filters=WebhookFilters() ) - app = _stub_app(conn) - app.state.webhook_config_path = config_path dispatcher = WebhookDispatcher(app) dispatcher._enqueue_delivery(SimpleNamespace(id=created.id), _event("e1", tenant="acme")) dispatcher._record_delivery_outcome( @@ -450,12 +455,11 @@ async def _fake_deliver_body(*args, **kwargs): def test_process_delivery_queue_skips_not_due(tmp_path: Path) -> None: conn = duckdb.connect(":memory:") try: - config_path = tmp_path / "webhooks.yaml" + app = _stub_app(conn) + app.state.webhook_config_path = tmp_path / "webhooks.yaml" created = create_webhook( - config_path, url="https://example.test/hook", tenant="acme", filters=WebhookFilters() + app, url="https://example.test/hook", tenant="acme", filters=WebhookFilters() ) - app = _stub_app(conn) - app.state.webhook_config_path = config_path dispatcher = WebhookDispatcher(app) dispatcher._enqueue_delivery(SimpleNamespace(id=created.id), _event("e1", tenant="acme")) conn.execute( @@ -484,12 +488,11 @@ def test_pending_delivery_survives_a_new_dispatcher_instance(tmp_path: Path) -> in-memory seen-set cannot do this.""" conn = duckdb.connect(":memory:") try: - config_path = tmp_path / "webhooks.yaml" + app = _stub_app(conn) + app.state.webhook_config_path = tmp_path / "webhooks.yaml" created = create_webhook( - config_path, url="https://example.test/hook", tenant="acme", filters=WebhookFilters() + app, url="https://example.test/hook", tenant="acme", filters=WebhookFilters() ) - app = _stub_app(conn) - app.state.webhook_config_path = config_path first = WebhookDispatcher(app) first._enqueue_delivery(SimpleNamespace(id=created.id), _event("e1", tenant="acme"))