feat(serving): alert history + alert-rule repository behind ControlPlaneStore (ADR 0010 slice 2)

CHANGELOG.md

@@ -4,6 +4,30 @@ All notable changes to AgentFlow are documented in this file.
 
 ## [Unreleased]
 
+### Added — Alert history + alert-rule repository behind the ControlPlaneStore port (ADR 0010 slice 2, 2026-07-02)
+
+- **Alert delivery history** (`alert_history`) moved behind `ControlPlaneStore`
+  (`log_alert_delivery` / `get_alert_delivery_history`) — `alerts/escalation.py`
+  and `routers/alerts.py` no longer reach `query_engine._conn`; the DDL
+  (`ensure_alert_history_table`) moved to `control_plane/embedded.py` next to
+  its webhook sibling, same catalog-DDL-lock discipline.
+- **Alert-rule repository** (`config/alerts.yaml`, including mutable runtime
+  state — `state`, `fired_at`, `last_escalation_level`, flap window, cooldown)
+  moved behind the port (`load_alert_rules` / `save_alert_rules`); the
+  embedded adapter keeps the exact YAML file format, so `config/alerts.yaml`
+  is unchanged on disk. `create_alert` / `list_alerts` / `get_alert` /
+  `update_alert` / `deactivate_alert` now take the FastAPI `app` (resolving
+  the store internally) instead of a bare config `Path`.
+- `src/serving/api/alerts/history.py` removed — its logic split between the
+  store port (history log) and `alerts/dispatcher.py` (rule repository
+  callers); the `get_alert_history` / `ensure_alert_history_table` backwards-
+  compatible re-exports on `alert_dispatcher.py` are retired with it (both
+  were internal DB-plumbing, not public SDK surface).
+- Structural ratchet test extended to `alerts/dispatcher.py`,
+  `alerts/escalation.py` and `routers/alerts.py` — none may reach
+  `query_engine._conn` directly. New `tests/unit/test_control_plane_store.py`
+  coverage for the alert history log and the YAML rule-repository round-trip.
+
 ### 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

src/serving/api/alert_dispatcher.py

@@ -2,8 +2,11 @@
 
 Historical import path ``src.serving.api.alert_dispatcher`` kept for
 ``main.py``/routers. The clock, logger, and HTTP client live in their
-defining modules (``alerts.dispatcher``/``alerts.escalation``/``alerts.history``)
-and are patched there in tests — this module no longer tunnels them.
+defining modules (``alerts.dispatcher``/``alerts.escalation``) and are
+patched there in tests — this module no longer tunnels them. Alert-history
+persistence and the alert-rule repository moved behind ``ControlPlaneStore``
+in ADR 0010 slice 2 — resolve them via
+``src.serving.control_plane.get_control_plane_store``, not through this shim.
 """
 
 from __future__ import annotations
@@ -18,10 +21,8 @@
     create_alert,
     deactivate_alert,
     ensure_alert_dispatcher,
-    ensure_alert_history_table,
     get_alert,
     get_alert_config_path,
-    get_alert_history,
     list_alerts,
     load_alerts,
     save_alerts,
@@ -38,10 +39,8 @@
     "create_alert",
     "deactivate_alert",
     "ensure_alert_dispatcher",
-    "ensure_alert_history_table",
     "get_alert",
     "get_alert_config_path",
-    "get_alert_history",
     "list_alerts",
     "load_alerts",
     "save_alerts",

src/serving/api/alerts/__init__.py

@@ -15,7 +15,6 @@
     save_alerts,
     update_alert,
 )
-from .history import ensure_alert_history_table, get_alert_history
 
 __all__ = [
     "DEFAULT_ALERTS_CONFIG_PATH",
@@ -27,10 +26,8 @@
     "create_alert",
     "deactivate_alert",
     "ensure_alert_dispatcher",
-    "ensure_alert_history_table",
     "get_alert",
     "get_alert_config_path",
-    "get_alert_history",
     "list_alerts",
     "load_alerts",
     "save_alerts",

src/serving/api/alerts/dispatcher.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import asyncio
-import json
 import os
 import secrets
 import uuid
@@ -12,10 +11,7 @@
 import structlog
 from pydantic import BaseModel, Field, model_validator
 
-try:
-    import yaml
-except ImportError:  # pragma: no cover
-    yaml = None  # type: ignore[assignment]
+from src.serving.control_plane import get_control_plane_store
 
 if TYPE_CHECKING:
     from fastapi import FastAPI
@@ -92,30 +88,18 @@ def get_alert_config_path(app: FastAPI) -> Path:
     return Path(configured) if configured else DEFAULT_ALERTS_CONFIG_PATH
 
 
-def load_alerts(path: Path) -> list[AlertRule]:
-    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 = AlertConfig.model_validate(data or {})
-    return config.alerts
-
-
-def save_alerts(path: Path, alerts: list[AlertRule]) -> None:
-    path.parent.mkdir(parents=True, exist_ok=True)
-    payload = AlertConfig(alerts=alerts).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", newline="\n")
+def load_alerts(app: FastAPI) -> list[AlertRule]:
+    store = get_control_plane_store(app)
+    return [AlertRule.model_validate(record) for record in store.load_alert_rules()]
+
+
+def save_alerts(app: FastAPI, alerts: list[AlertRule]) -> None:
+    store = get_control_plane_store(app)
+    store.save_alert_rules([alert.model_dump(mode="json") for alert in alerts])
 
 
 def create_alert(
-    path: Path,
+    app: FastAPI,
     *,
     name: str,
     tenant: str,
@@ -126,7 +110,7 @@ def create_alert(
     webhook_url: str,
     cooldown_minutes: int,
 ) -> AlertRule:
-    alerts = load_alerts(path)
+    alerts = load_alerts(app)
     now = datetime.now(UTC)
     rule = AlertRule(
         id=str(uuid.uuid4()),
@@ -143,23 +127,23 @@ def create_alert(
         updated_at=now,
     )
     alerts.append(rule)
-    save_alerts(path, alerts)
+    save_alerts(app, alerts)
     return rule
 
 
-def list_alerts(path: Path, tenant: str) -> list[AlertRule]:
-    return [alert for alert in load_alerts(path) if alert.tenant == tenant and alert.active]
+def list_alerts(app: FastAPI, tenant: str) -> list[AlertRule]:
+    return [alert for alert in load_alerts(app) if alert.tenant == tenant and alert.active]
 
 
-def get_alert(path: Path, alert_id: str, tenant: str) -> AlertRule | None:
-    for alert in load_alerts(path):
+def get_alert(app: FastAPI, alert_id: str, tenant: str) -> AlertRule | None:
+    for alert in load_alerts(app):
         if alert.id == alert_id and alert.tenant == tenant and alert.active:
             return alert
     return None
 
 
-def update_alert(path: Path, alert_id: str, tenant: str, updates: dict) -> AlertRule | None:
-    alerts = load_alerts(path)
+def update_alert(app: FastAPI, alert_id: str, tenant: str, updates: dict) -> AlertRule | None:
+    alerts = load_alerts(app)
     for index, alert in enumerate(alerts):
         if alert.id != alert_id or alert.tenant != tenant or not alert.active:
             continue
@@ -168,13 +152,13 @@ def update_alert(path: Path, alert_id: str, tenant: str, updates: dict) -> Alert
         payload["updated_at"] = datetime.now(UTC)
         updated = AlertRule.model_validate(payload)
         alerts[index] = updated
-        save_alerts(path, alerts)
+        save_alerts(app, alerts)
         return updated
     return None
 
 
-def deactivate_alert(path: Path, alert_id: str, tenant: str) -> bool:
-    alerts = load_alerts(path)
+def deactivate_alert(app: FastAPI, alert_id: str, tenant: str) -> bool:
+    alerts = load_alerts(app)
     changed = False
     for index, alert in enumerate(alerts):
         if alert.id != alert_id or alert.tenant != tenant or not alert.active:
@@ -186,7 +170,7 @@ def deactivate_alert(path: Path, alert_id: str, tenant: str) -> bool:
         changed = True
         break
     if changed:
-        save_alerts(path, alerts)
+        save_alerts(app, alerts)
     return changed
 
 
@@ -266,8 +250,7 @@ async def run(self) -> None:
     async def dispatch_alerts(self) -> int:
         from .escalation import dispatch_alert
 
-        path = get_alert_config_path(self.app)
-        alerts = load_alerts(path)
+        alerts = load_alerts(self.app)
         now = datetime.now(UTC)
         triggered = 0
         changed = False
@@ -279,7 +262,7 @@ async def dispatch_alerts(self) -> int:
             triggered += alert_triggered
             changed = changed or alert_changed
         if changed:
-            save_alerts(path, alerts)
+            save_alerts(self.app, alerts)
         return triggered
 
     async def send_test_alert(self, alert: AlertRule) -> dict:

src/serving/api/alerts/escalation.py

@@ -10,9 +10,9 @@
 
 from src.serving.api.egress_guard import UnsafeEgressURLError, validate_public_url
 from src.serving.api.webhook_dispatcher import _event_body, _signature
+from src.serving.control_plane import get_control_plane_store
 
 from .evaluator import evaluate_rule
-from .history import ensure_alert_history_table, log_alert_history
 
 if TYPE_CHECKING:
     from .dispatcher import AlertDispatcher, AlertRule
@@ -248,8 +248,7 @@ async def deliver(
     change_pct: float | None = None,
     webhook_url: str | None = None,
 ) -> dict:
-    conn = dispatcher.app.state.query_engine._conn
-    ensure_alert_history_table(conn)
+    store = get_control_plane_store(dispatcher.app)
     delivery_id = str(uuid.uuid4())
     body = _event_body(payload)
     headers = {
@@ -270,10 +269,11 @@ async def deliver(
         await asyncio.to_thread(validate_public_url, target_url)
     except UnsafeEgressURLError as exc:
         error = f"unsafe egress URL: {exc}"
-        log_alert_history(
-            conn,
+        store.log_alert_delivery(
             delivery_id=delivery_id,
-            alert=alert,
+            alert_id=alert.id,
+            alert_name=alert.name,
+            tenant=alert.tenant,
             metric=alert.metric,
             current_value=current_value,
             previous_value=previous_value,
@@ -322,10 +322,11 @@ async def deliver(
                 ]
                 await asyncio.sleep(delay)
 
-    log_alert_history(
-        conn,
+    store.log_alert_delivery(
         delivery_id=delivery_id,
-        alert=alert,
+        alert_id=alert.id,
+        alert_name=alert.name,
+        tenant=alert.tenant,
         metric=alert.metric,
         current_value=current_value,
         previous_value=previous_value,