ControlPlaneStore port + embedded adapter — webhook queue/log (ADR 0010 slice 1)

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)