feat(serving): ControlPlaneStore port + embedded adapter — webhook queue/log behind it (ADR 0010 slice 1) (#131)

First rollout slice of ADR 0010: the webhook durable delivery queue and the
delivery attempt log move behind a ControlPlaneStore port. Pure extraction —
table DDL, SQL shapes, catalog-DDL-lock behavior and the dispatcher's public
method signatures are byte-compatible with the pre-port code; no behavior
change on the embedded (default) profile.

- NEW src/serving/control_plane/: store.py (port ABC + WebhookQueueRow +
  get_control_plane_store resolver), embedded.py (EmbeddedControlPlaneStore
  over the serving DuckDB; ensure_* DDL helpers moved here verbatim,
  including the shared catalog_ddl_lock discipline).
- Claim semantics are part of the port contract, not adapter detail:
  enqueue returns True only for the inserting caller (inline delivery
  belongs to the enqueue winner), claim_due returns rows this worker owns
  (embedded: degenerate single-process exclusivity; the PG adapter in
  slice 5 implements the same contract with FOR UPDATE SKIP LOCKED + lease).
- webhook_dispatcher and routers/webhooks.py no longer reach into
  query_engine._conn; the single sanctioned reach is the composition seam
  in control_plane/store.py (documented), pinned by a structural ratchet
  test. The outcome state machine (delivered / backoff-pending / dead at
  max attempts) moved into the store; retry policy stays dispatcher config.
- get_control_plane_store: lazy per-app resolution (mirrors
  ensure_alert_dispatcher so unit stubs keep working), eager boot-time
  resolution in the lifespan. AGENTFLOW_CONTROLPLANE_STORE env knob:
  'embedded' (default) runs; 'postgres' raises NotImplementedError until
  slice 5 ships the adapter (fail-closed ratchet, pinned by test); unknown
  values are a boot error.
- Tests: NEW tests/unit/test_control_plane_store.py (13 tests: enqueue-win,
  claim ordering/bounds/due-ness, outcome transitions, park, log roundtrip,
  resolver caching + ratchets + live-engine rebinding, structural _conn
  ratchet); test_webhook_dispatcher_unit / test_catalog_ddl_concurrency /
  integration test_webhooks rewired to the new module; the DDL-lock
  identity pin now covers control_plane.embedded.

Verified: unit 42 (store+dispatcher+ddl) + integration webhooks 12 + full
unit suite green locally; ruff check+format clean; mypy strict clean on all
touched src files.

Co-authored-by: JuliaEdom <uedomskikh@gmail.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: 3 people

CHANGELOG.md

@@ -4,6 +4,25 @@ All notable changes to AgentFlow are documented in this file.
 
 ## [Unreleased]
 
+### Added — ControlPlaneStore port + embedded adapter: webhook queue/log behind it (ADR 0010 slice 1, 2026-07-02)
+
+- **New `src/serving/control_plane/`** — the `ControlPlaneStore` port and its
+  `EmbeddedControlPlaneStore` (DuckDB) adapter; the webhook durable delivery
+  queue and the delivery attempt log are the first subsystem behind it. Pure
+  extraction: DDL, SQL shapes, the catalog-DDL-lock discipline and the
+  dispatcher's pinned method signatures are byte-compatible — no behavior
+  change on the embedded (default) profile.
+- Claim semantics are part of the port contract (enqueue-winner-only inline
+  delivery; `claim_due` ownership — degenerate in one process, `FOR UPDATE
+  SKIP LOCKED` + lease in the slice-5 PostgreSQL adapter); the outcome state
+  machine moved into the store so it can be one transaction on PostgreSQL,
+  while retry policy stays dispatcher configuration.
+- `webhook_dispatcher` and `routers/webhooks.py` no longer reach into
+  `query_engine._conn` — pinned by a structural ratchet test; the
+  `AGENTFLOW_CONTROLPLANE_STORE` knob resolves the adapter (`postgres` raises
+  until slice 5 ships — fail-closed, pinned by test; unknown values fail the
+  boot). New `tests/unit/test_control_plane_store.py` (12 tests).
+
 ### Added — Control-plane externalization decided; scaling gate enforced at render time (ADR 0010, 2026-07-02)
 
 - **New [ADR 0010](docs/decisions/0010-control-plane-externalization-postgres.md)**

src/serving/api/main.py

@@ -53,6 +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.db_pool import DuckDBPool
 from src.serving.semantic_layer.catalog import DataCatalog
 from src.serving.semantic_layer.query_engine import QueryEngine
@@ -195,6 +196,12 @@ 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)
     app.state.webhook_dispatcher = WebhookDispatcher(app)
     original_dispatch_new_events = app.state.webhook_dispatcher.dispatch_new_events
 

src/serving/api/routers/webhooks.py

@@ -12,11 +12,11 @@
     WebhookFilters,
     create_webhook,
     deactivate_webhook,
-    get_delivery_logs,
     get_webhook,
     get_webhook_config_path,
     list_webhooks,
 )
+from src.serving.control_plane import get_control_plane_store
 
 router = APIRouter(prefix="/v1/webhooks", tags=["webhooks"])
 
@@ -106,11 +106,8 @@ async def webhook_logs(webhook_id: str, request: Request) -> dict[str, object]:
 
 
 def _read_delivery_logs(request: Request, webhook_id: str) -> list[dict]:
-    # The log scan runs on a worker thread (run_in_threadpool); a dedicated
-    # cursor — not the shared connection — keeps concurrent reads on different
-    # threads from colliding on the connection. (audit_30_06_26.md A2)
-    cursor = request.app.state.query_engine._conn.cursor()
-    try:
-        return get_delivery_logs(cursor, webhook_id)
-    finally:
-        cursor.close()
+    # Runs on a worker thread (run_in_threadpool); the control-plane store
+    # isolates the read (a dedicated cursor per call in the embedded adapter)
+    # so concurrent reads on different threads don't collide on the shared
+    # connection. (audit_30_06_26.md A2)
+    return get_control_plane_store(request.app).get_webhook_delivery_logs(webhook_id)