Skip to content

Commit 4104382

Browse files
brownjuly2003-codeJuliaEdomclaude
authored
feat(serving): PostgresControlPlaneStore — the scale profile ships (ADR 0010 slice 5) (#135)
* feat(serving): PostgresControlPlaneStore — the scale profile ships (ADR 0010 slice 5) 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 <noreply@anthropic.com> * 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 <noreply@anthropic.com> * fix(tests): stub psycopg in the postgres resolution test (CI unit env has no optional extras) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: JuliaEdom <uedomskikh@gmail.com> Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
1 parent b901b5d commit 4104382

25 files changed

Lines changed: 2904 additions & 166 deletions

‎.github/workflows/ci.yml‎

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -177,14 +177,30 @@ jobs:
177177
CLICKHOUSE_USER: agentflow
178178
CLICKHOUSE_PASSWORD: agentflow
179179
CLICKHOUSE_DB: agentflow
180+
postgres:
181+
# Live coverage for PostgresControlPlaneStore (ADR 0010 slice 5);
182+
# test_control_plane_postgres_live.py skips itself when
183+
# AGENTFLOW_TEST_PG_DSN is absent.
184+
image: postgres:17
185+
ports:
186+
- 5432:5432
187+
env:
188+
POSTGRES_USER: agentflow
189+
POSTGRES_PASSWORD: agentflow
190+
POSTGRES_DB: agentflow
191+
options: >-
192+
--health-cmd "pg_isready -U agentflow"
193+
--health-interval 5s
194+
--health-timeout 5s
195+
--health-retries 10
180196
steps:
181197
- uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
182198
- uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0
183199
with:
184200
python-version: "3.11"
185201
- name: Install dependencies
186202
run: |
187-
pip install -e ".[dev,cloud]"
203+
pip install -e ".[dev,cloud,postgres]"
188204
pip install -e "./sdk"
189205
- name: Prepare pytest temp directory
190206
run: mkdir -p .tmp
@@ -201,6 +217,7 @@ jobs:
201217
CLICKHOUSE_LIVE_USER: agentflow
202218
CLICKHOUSE_LIVE_PASSWORD: agentflow
203219
CLICKHOUSE_LIVE_DATABASE: agentflow
220+
AGENTFLOW_TEST_PG_DSN: postgresql://agentflow:agentflow@localhost:5432/agentflow
204221
run: pytest tests/integration/ -v --tb=short
205222

206223
helm-schema-live:

‎CHANGELOG.md‎

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,63 @@ All notable changes to AgentFlow are documented in this file.
44

55
## [Unreleased]
66

7+
### Added — PostgresControlPlaneStore: the scale profile ships (ADR 0010 slice 5, 2026-07-03)
8+
9+
- **New `src/serving/control_plane/postgres.py`** — all six control-plane
10+
state classes as PostgreSQL tables behind the existing port, with the claim
11+
semantics the embedded adapter only satisfies degenerately made real:
12+
enqueue-win by `INSERT .. ON CONFLICT DO NOTHING` rowcount, queue/outbox
13+
claims by `FOR UPDATE SKIP LOCKED` + a self-expiring `lease_expires_at`
14+
(work-stealing across replicas, no leader election; a crashed owner's rows
15+
become due again on lease expiry), invariant 8 as an ordinary transaction
16+
(every store method is one transaction: commit on success, rollback on any
17+
exception). Payloads stay TEXT/JSON-string so callers see the embedded
18+
adapter's shapes. One connection per call — pooling stays out of ADR scope.
19+
Selection: `AGENTFLOW_CONTROLPLANE_STORE=postgres` +
20+
`AGENTFLOW_CONTROLPLANE_PG_DSN` (+ optional
21+
`AGENTFLOW_CONTROLPLANE_LEASE_SECONDS`); the slice-1 `NotImplementedError`
22+
ratchet is gone, and a missing DSN or missing `psycopg` fails the boot
23+
loudly — never a silent fallback to embedded. `psycopg` is a new optional
24+
extra (`pip install agentflow-runtime[postgres]`), the `redis` import
25+
pattern.
26+
- **Webhook registrations (state class 5) move behind the port** — the
27+
sharpest split-brain of the ADR's inventory was still a per-pod YAML read
28+
outside the port after slices 1–4. New port methods
29+
`load_webhook_registrations`/`save_webhook_registrations`;
30+
`load_webhooks`/`save_webhooks`/`create_webhook`/`list_webhooks`/
31+
`get_webhook`/`deactivate_webhook` now take `app` and resolve the store
32+
inside (the same move the alert-rule helpers made in slice 2). The embedded
33+
adapter keeps the byte-compatible `config/webhooks.yaml`.
34+
- **Alert-tick single-flight (ADR 0010 §2) wired into the dispatcher** — new
35+
port methods `claim_alert_tick`/`complete_alert_tick`;
36+
`AlertDispatcher.dispatch_alerts` claims each rule before evaluating (a
37+
lost claim = another replica owns that rule's tick) and persists advanced
38+
rule state **per rule** in the same transaction as the claim release — the
39+
old full-set save would let two replicas advancing different rules clobber
40+
each other's runtime state. Embedded grants every claim (one process), so
41+
the single-replica profile behaves as before; a CRUD full-set save on
42+
PostgreSQL upserts by id and does not release an in-flight claim.
43+
- **The postgres profile shares one store across every consumer**: `main.py`
44+
injects the app-wide store into `AuthManager` and `OutboxProcessor` when
45+
the profile is external (embedded keeps its historical private stores);
46+
the analytics entry points (`analytics.py`, `routers/admin.py`,
47+
`admin_ui.py`'s QPS tile) accept the store handle and route through
48+
`AuthManager.store`, so usage/sessions land in PostgreSQL instead of a
49+
per-pod DuckDB file.
50+
- **Verified live (standalone PostgreSQL 17.5, no Docker): 31/31 probes** —
51+
the ADR's named suite (parallel claim exclusivity, lease-expiry re-drive,
52+
restart re-drive, enqueue-win uniqueness, outbox↔dead-letter atomicity
53+
incl. rollback halves, alert-tick single-flight) plus a full contract
54+
parity sweep and an end-to-end app test (two boots on the postgres profile
55+
see each other's webhook registration; usage accounting lands in PG):
56+
`docs/perf/control-plane-pg-verify-2026-07-03.md`. The same suite runs in
57+
CI against a new `postgres:17` service in the integration job and
58+
self-skips where `AGENTFLOW_TEST_PG_DSN` is absent.
59+
- Helm is untouched by design: the values schema still pins
60+
`controlPlane.store=embedded` and the chart still refuses multi-replica
61+
renders — the enum extension and env/secret wiring are rollout slice 6,
62+
which this slice unblocks.
63+
764
### Added — API-usage accounting and session analytics behind the ControlPlaneStore port (ADR 0010 slice 4, 2026-07-02)
865

966
- **`api_usage`** (per-tenant/per-key request counters) and **`api_sessions`**

‎docs/decisions/0010-control-plane-externalization-postgres.md‎

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -186,7 +186,17 @@ ClickHouse is consumed), and `psycopg` joins the optional dependencies.
186186
5. `PostgresControlPlaneStore` + live verification (standalone-PG probe
187187
suite: parallel claim exclusivity, lease expiry re-drive, restart
188188
re-drive, enqueue-win uniqueness, outbox↔dead-letter atomicity) +
189-
CI integration coverage on the existing PG service.
189+
CI integration coverage (a `postgres:17` service added to the CI
190+
integration job). Executed 2026-07-03 with two scope additions the
191+
extraction slices had left open, both required for §1's "all six state
192+
classes" to hold: **webhook registrations** (class 5 — the sharpest
193+
split-brain — was still a per-pod YAML read outside the port; the
194+
registration CRUD now resolves the store from ``app``, embedded keeps the
195+
byte-compatible YAML) and **§2's ``claim_alert_tick`` /
196+
``complete_alert_tick``** wired into the dispatcher (per-rule state
197+
persistence — a full-set save would let two replicas clobber each other's
198+
rule runtime state). Verified live: 31/31 probes,
199+
`docs/perf/control-plane-pg-verify-2026-07-03.md`.
190200
6. Helm wiring (`controlPlane.store=postgres` profile: env + secret, schema
191201
enum extension — the render gate then admits multi-replica) and cutover
192202
plan Phase 3 execution: kind staging at `replicaCount=2`, verifying
@@ -213,8 +223,9 @@ ClickHouse is consumed), and `psycopg` joins the optional dependencies.
213223
- Per-pod event scanning is N× read amplification on the serving backend at
214224
scale (accepted; the scan is bounded and cheap, and consolidating scanners
215225
is a later topology refinement, cf. option 3).
216-
- Until slice 5 lands, multi-replica is simply impossible to render — a
217-
deliberate fail-closed period.
226+
- Until slice 6 extends the chart, multi-replica is simply impossible to
227+
render — a deliberate fail-closed period (the slice-5 adapter is app-side;
228+
the schema enum still pins `embedded` until the helm profile ships).
218229

219230
## Follow-up
220231

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,65 @@
1+
# Control plane on PostgreSQL — live verification (ADR 0010 rollout slice 5)
2+
3+
**Date:** 2026-07-03
4+
**Environment:** standalone PostgreSQL 17.5 (EDB windows-x64 binaries, no
5+
Docker, no service install: `initdb` + `pg_ctl`, port 55433, trust auth,
6+
user/db `agentflow`) — the same no-Docker standalone-PG recipe that verified
7+
the vault governance layer
8+
(`vault-pii-governance-pg-verify-2026-07-02.md`). Adapter under test:
9+
`src/serving/control_plane/postgres.py` (`PostgresControlPlaneStore`),
10+
driven by the probe suite the ADR names for this slice:
11+
`tests/integration/test_control_plane_postgres_live.py` with
12+
`AGENTFLOW_TEST_PG_DSN=postgresql://agentflow@127.0.0.1:55433/agentflow`.
13+
14+
**Result: 31/31 probes passed** (`pytest`, 19.45s; psycopg 3.3.4). The same
15+
suite runs in CI against the `postgres:17` service container added to the
16+
`test-integration` job (it self-skips when the DSN env var is absent, the
17+
`test_clickhouse_backend_live.py` pattern).
18+
19+
## The ADR's named probes
20+
21+
| Probe (ADR 0010 § Rollout 5) | Test | Result |
22+
| --- | --- | --- |
23+
| Enqueue-win uniqueness | 8 threads race `enqueue_webhook_delivery` on one (webhook, event) → exactly **1** `True`, 1 row | passed |
24+
| Parallel claim exclusivity | 4 threads claim 10 due rows concurrently → no row handed out twice, none lost (`FOR UPDATE SKIP LOCKED`) | passed |
25+
| 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 |
26+
| Restart re-drive | pending row enqueued by one store instance is claimed by a **fresh** instance, canonical body verbatim | passed |
27+
| 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 |
28+
| 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 |
29+
30+
## Contract parity sweep
31+
32+
Every port method exercised against live PostgreSQL with the same assertions
33+
the embedded adapter's unit pins make: webhook outcome state machine
34+
(backoff → `dead` at max, success → `delivered`, park), oldest-first claim
35+
ordering under `limit`, attempt-log and alert-history roundtrips (newest
36+
first, JSON payload decoded), webhook-registration and alert-rule
37+
repositories (order preserved via a `position` column, full-set save =
38+
YAML-replace semantics, ids required), dead-letter reads (tenant scoping,
39+
reason filter, pagination, stats + trend), usage accounting (per tenant /
40+
per key / old-key-slot hour window), session analytics (idempotent
41+
insert-or-replace on `request_id`, usage/top-queries/top-entities/latency
42+
percentiles/anomalies/QPS shapes, malformed window → `ValueError`, QPS
43+
degrades to 0.0 on an unreachable server).
44+
45+
## End to end: the app itself on the postgres profile
46+
47+
`test_app_on_postgres_profile_shares_state_across_boots` boots the real
48+
FastAPI app twice with `AGENTFLOW_CONTROLPLANE_STORE=postgres`:
49+
50+
- boot #1 resolves `PostgresControlPlaneStore` in the lifespan,
51+
`AuthManager.store` **is** the shared app-wide store (slice 5 injection in
52+
`main.py`), and `POST /v1/webhooks` registers a webhook;
53+
- boot #2 (a second pod, in production terms) sees that registration through
54+
`GET /v1/webhooks` — the class-5 per-pod YAML split-brain is gone;
55+
- `api_usage` rows for the authenticated tenant landed in PostgreSQL, not in
56+
a local DuckDB file.
57+
58+
## Notes
59+
60+
- Claim leases default to 300s (`AGENTFLOW_CONTROLPLANE_LEASE_SECONDS`
61+
overrides); alert-tick leases are 120s (`AlertDispatcher.tick_lease_seconds`).
62+
- Payload columns are TEXT holding JSON strings — callers see the same
63+
"string, you decode it" shape as on the embedded adapter.
64+
- One connection per method call, no pooling (out of ADR 0010 scope,
65+
recorded follow-up).

‎docs/security-audit.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -80,7 +80,7 @@ Audit finding A-4 flagged the dynamic-SQL surface as "one careless edit away fro
8080
- 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.
8181
- 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`.
8282

83-
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).
83+
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`.
8484

8585
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`.
8686

‎helm/agentflow/templates/deployment.yaml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22
{{- fail "DuckDB persistence requires a single writer replica: set replicaCount=1 and autoscaling.maxReplicas=1, or disable persistence and use ephemeral storage." }}
33
{{- end }}
44
{{- 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")) }}
5-
{{- 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." }}
5+
{{- 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." }}
66
{{- end }}
77
{{- $secretName := .Values.secrets.existingSecret | default (include "agentflow.fullname" .) }}
88
apiVersion: apps/v1

‎helm/agentflow/values.yaml‎

Lines changed: 10 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -134,13 +134,16 @@ serving:
134134
existingSecret: ""
135135
passwordKey: clickhouse-password
136136

137-
# Control-plane state (ADR 0009 / ADR 0010): webhook queue+log, alert rules and
138-
# history, outbox, dead-letter and usage accounting. 'embedded' = per-pod
139-
# DuckDB + config files — correct for the single-replica profile, a split-brain
140-
# at replicaCount > 1 (duplicate deliveries, forked alert state). ADR 0010
141-
# externalizes this state to PostgreSQL behind a ControlPlaneStore port; until
142-
# that adapter ships, the values schema pins store=embedded (fail-closed
143-
# ratchet) and the chart refuses any multi-replica render.
137+
# Control-plane state (ADR 0009 / ADR 0010): webhook queue+log, registrations,
138+
# alert rules and history, outbox, dead-letter and usage accounting.
139+
# 'embedded' = per-pod DuckDB + config files — correct for the single-replica
140+
# profile, a split-brain at replicaCount > 1 (duplicate deliveries, forked
141+
# alert state). The PostgresControlPlaneStore adapter shipped with ADR 0010
142+
# rollout slice 5 (app-side: AGENTFLOW_CONTROLPLANE_STORE=postgres +
143+
# AGENTFLOW_CONTROLPLANE_PG_DSN); the chart profile for it (env + secret
144+
# wiring, schema enum extension) lands with rollout slice 6 — until then the
145+
# values schema pins store=embedded (fail-closed ratchet) and the chart
146+
# refuses any multi-replica render.
144147
controlPlane:
145148
store: embedded
146149

‎pyproject.toml‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,12 @@ cloud = [
3737
"boto3>=1.35,<2",
3838
"pyiceberg[pyiceberg-core]>=0.7,<1",
3939
]
40+
postgres = [
41+
# ADR 0010 slice 5: PostgresControlPlaneStore (scale profile). Optional
42+
# exactly like redis — the embedded default profile needs none of it,
43+
# and control_plane/postgres.py degrades to a clear boot error without it.
44+
"psycopg[binary]>=3.2,<4",
45+
]
4046
load = [
4147
"locust>=2.29,<3",
4248
# benchmark_freshness.py: production QueryCache semantics (TTL + key
@@ -158,6 +164,11 @@ ignore = ["S101", "S311"]
158164
"src/serving/api/routers/lineage.py" = [
159165
"S608",
160166
]
167+
# Interpolations are internal literals only (a lease-expiry SQL fragment,
168+
# optional filter clauses, table-name constants); all values bind via %s.
169+
"src/serving/control_plane/postgres.py" = [
170+
"S608",
171+
]
161172
"src/serving/api/routers/slo.py" = [
162173
"S608",
163174
]
@@ -284,7 +295,10 @@ editable-installs = [".[dev]"]
284295
editable-installs = [".[dev,cloud]"]
285296

286297
[tool.agentflow.dependency-profiles.profiles.test-sdk]
287-
editable-installs = [".[dev,cloud]", "./sdk"]
298+
# postgres: the CI integration job drives the PostgresControlPlaneStore live
299+
# suite against its postgres:17 service (ADR 0010 slice 5), so it needs the
300+
# optional psycopg extra.
301+
editable-installs = [".[dev,cloud,postgres]", "./sdk"]
288302

289303
[tool.agentflow.dependency-profiles.profiles.test-integrations]
290304
editable-installs = [".[dev,cloud]", "./sdk", "./integrations[mcp]"]

0 commit comments

Comments
 (0)