From 0d67be84860e21d9b0415dd6d5ee1c8b0e7ef514 Mon Sep 17 00:00:00 2001 From: JuliaEdom Date: Fri, 3 Jul 2026 01:06:05 +0300 Subject: [PATCH 1/3] =?UTF-8?q?feat(serving):=20PostgresControlPlaneStore?= =?UTF-8?q?=20=E2=80=94=20the=20scale=20profile=20ships=20(ADR=200010=20sl?= =?UTF-8?q?ice=205)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit All six control-plane state classes as PostgreSQL tables behind the ControlPlaneStore port, with real claim semantics: enqueue-win by ON CONFLICT rowcount, FOR UPDATE SKIP LOCKED + self-expiring leases for queue/outbox claims (work-stealing, no leader election), invariant 8 as an ordinary transaction. Selection via AGENTFLOW_CONTROLPLANE_STORE=postgres + AGENTFLOW_CONTROLPLANE_PG_DSN; psycopg is a new optional extra (the redis pattern); the slice-1 NotImplementedError ratchet is gone. Two scope additions the extraction slices left open, both needed for the ADR's "all six classes" to hold: - webhook registrations (class 5, the sharpest split-brain) move behind the port; registration CRUD takes `app` and resolves the store inside (the slice-2 alerts move); embedded keeps byte-compatible config/webhooks.yaml. - alert-tick single-flight (ADR §2): claim_alert_tick/complete_alert_tick wired into AlertDispatcher — per-rule state persistence in the same transaction as the claim release, replacing the full-set save that would let two replicas clobber each other's rule runtime state. On the postgres profile main.py injects the shared store into AuthManager and OutboxProcessor, and the analytics entry points route through AuthManager.store — usage/sessions land in PostgreSQL, not a per-pod file. Verified live (standalone PostgreSQL 17.5, no Docker): 31/31 probes — parallel claim exclusivity, lease-expiry re-drive, restart re-drive, enqueue-win uniqueness, outbox<->dead-letter atomicity incl. rollback halves, alert-tick single-flight, full contract parity sweep, and an end-to-end two-boot app test (docs/perf/control-plane-pg-verify-2026-07-03.md). The same suite runs in CI against a new postgres:17 service and self-skips without AGENTFLOW_TEST_PG_DSN. Helm untouched by design: the schema enum still pins embedded until slice 6 ships the chart profile. Full unit+property 1586, integration 237/19skip, mypy strict, ruff check+format, bandit surface pinned (B608 sites documented, A-4 registry). Co-Authored-By: Claude Fable 5 --- .github/workflows/ci.yml | 19 +- CHANGELOG.md | 57 + ...-control-plane-externalization-postgres.md | 17 +- .../control-plane-pg-verify-2026-07-03.md | 65 + docs/security-audit.md | 2 +- helm/agentflow/templates/deployment.yaml | 2 +- helm/agentflow/values.yaml | 17 +- pyproject.toml | 16 +- src/serving/api/alerts/dispatcher.py | 38 +- src/serving/api/analytics.py | 56 +- src/serving/api/main.py | 32 +- src/serving/api/routers/admin.py | 10 +- src/serving/api/routers/admin_ui.py | 14 +- src/serving/api/routers/webhooks.py | 11 +- src/serving/api/webhook_dispatcher.py | 61 +- src/serving/control_plane/__init__.py | 13 +- src/serving/control_plane/embedded.py | 57 + src/serving/control_plane/postgres.py | 1350 +++++++++++++++++ src/serving/control_plane/store.py | 95 +- .../test_control_plane_postgres_live.py | 839 ++++++++++ tests/unit/test_analytics_middleware.py | 4 + tests/unit/test_blocking_routes_offloaded.py | 4 +- tests/unit/test_control_plane_store.py | 200 ++- tests/unit/test_security_tooling_policy.py | 6 + tests/unit/test_webhook_dispatcher_unit.py | 77 +- 25 files changed, 2896 insertions(+), 166 deletions(-) create mode 100644 docs/perf/control-plane-pg-verify-2026-07-03.md create mode 100644 src/serving/control_plane/postgres.py create mode 100644 tests/integration/test_control_plane_postgres_live.py 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..0a06d7ba --- /dev/null +++ b/src/serving/control_plane/postgres.py @@ -0,0 +1,1350 @@ +"""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() + return psycopg.connect(self._dsn) + + 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..c26e5e21 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,54 @@ 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.postgres import PostgresControlPlaneStore + + 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")) From 06444a161a50e550dedd18cff6fde75affc2eefd Mon Sep 17 00:00:00 2001 From: JuliaEdom Date: Fri, 3 Jul 2026 01:11:38 +0300 Subject: [PATCH 2/3] fix(serving): keep mypy green when psycopg is absent (CI lint env) The lint job installs no optional extras, so psycopg resolves as Any and warn_return_any flags _connect's bare return; an annotated local pins the declared context-manager type in both environments. Co-Authored-By: Claude Fable 5 --- src/serving/control_plane/postgres.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/serving/control_plane/postgres.py b/src/serving/control_plane/postgres.py index 0a06d7ba..b3f63266 100644 --- a/src/serving/control_plane/postgres.py +++ b/src/serving/control_plane/postgres.py @@ -274,7 +274,10 @@ def _connect(self) -> AbstractContextManager[Any]: # manager commits on clean exit and rolls back on exception, which is # exactly the invariant-8 semantics the port requires. self._ensure_schema() - return psycopg.connect(self._dsn) + # 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: From a3d5f4563a9513d352eef0f0d6978d1077b11106 Mon Sep 17 00:00:00 2001 From: JuliaEdom Date: Fri, 3 Jul 2026 01:18:36 +0300 Subject: [PATCH 3/3] fix(tests): stub psycopg in the postgres resolution test (CI unit env has no optional extras) Co-Authored-By: Claude Fable 5 --- tests/unit/test_control_plane_store.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/tests/unit/test_control_plane_store.py b/tests/unit/test_control_plane_store.py index c26e5e21..38f871d8 100644 --- a/tests/unit/test_control_plane_store.py +++ b/tests/unit/test_control_plane_store.py @@ -761,8 +761,13 @@ def test_get_store_postgres_requires_a_dsn( 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)