diff --git a/CHANGELOG.md b/CHANGELOG.md
index a344d8b9..042aa586 100644
--- a/CHANGELOG.md
+++ b/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
diff --git a/src/serving/api/alert_dispatcher.py b/src/serving/api/alert_dispatcher.py
index b3103f95..9a1b3f24 100644
--- a/src/serving/api/alert_dispatcher.py
+++ b/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",
diff --git a/src/serving/api/alerts/__init__.py b/src/serving/api/alerts/__init__.py
index ff78dbec..16b780ea 100644
--- a/src/serving/api/alerts/__init__.py
+++ b/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",
diff --git a/src/serving/api/alerts/dispatcher.py b/src/serving/api/alerts/dispatcher.py
index 14930ad3..49a02689 100644
--- a/src/serving/api/alerts/dispatcher.py
+++ b/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:
diff --git a/src/serving/api/alerts/escalation.py b/src/serving/api/alerts/escalation.py
index 23464647..a499b9bf 100644
--- a/src/serving/api/alerts/escalation.py
+++ b/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,
diff --git a/src/serving/api/alerts/history.py b/src/serving/api/alerts/history.py
deleted file mode 100644
index 6c34eb94..00000000
--- a/src/serving/api/alerts/history.py
+++ /dev/null
@@ -1,119 +0,0 @@
-from __future__ import annotations
-
-import json
-from datetime import UTC, datetime
-from typing import TYPE_CHECKING
-
-import duckdb
-
-from src.db_concurrency import catalog_ddl_lock
-
-if TYPE_CHECKING:
-    from .dispatcher import AlertRule
-
-
-def ensure_alert_history_table(conn: duckdb.DuckDBPyConnection) -> None:
-    # Serialize the lazy DDL: the offloaded read handler calls this on a worker
-    # thread, and concurrent CREATE on a cold DuckDB catalog conflicts (across
-    # tables too). (audit_30 A2 follow-up: #120 offload race)
-    with catalog_ddl_lock:
-        conn.execute(
-            """
-            CREATE TABLE IF NOT EXISTS alert_history (
-                delivery_id VARCHAR,
-                alert_id VARCHAR,
-                alert_name VARCHAR,
-                metric VARCHAR,
-                current_value DOUBLE,
-                previous_value DOUBLE,
-                change_pct DOUBLE,
-                threshold DOUBLE,
-                condition VARCHAR,
-                metric_window VARCHAR,
-                tenant VARCHAR,
-                event_type VARCHAR,
-                status_code INTEGER,
-                success BOOLEAN,
-                error TEXT,
-                payload JSON,
-                triggered_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
-            )
-            """
-        )
-
-
-def get_alert_history(conn: duckdb.DuckDBPyConnection, alert_id: str) -> list[dict]:
-    ensure_alert_history_table(conn)
-    cursor = 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 = ?
-        ORDER BY triggered_at DESC
-        LIMIT 20
-        """,
-        [alert_id],
-    )
-    columns = [description[0] for description in cursor.description]
-    records = [dict(zip(columns, row, strict=False)) for row in cursor.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
-
-
-def log_alert_history(
-    conn: duckdb.DuckDBPyConnection,
-    *,
-    delivery_id: str,
-    alert: AlertRule,
-    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:
-    ensure_alert_history_table(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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
-        """,
-        [
-            delivery_id,
-            alert.id,
-            alert.name,
-            metric,
-            current_value,
-            previous_value,
-            change_pct,
-            threshold,
-            condition,
-            window,
-            alert.tenant,
-            event_type,
-            status_code,
-            success,
-            error,
-            json.dumps(payload, sort_keys=True),
-            datetime.now(UTC),
-        ],
-    )
diff --git a/src/serving/api/routers/alerts.py b/src/serving/api/routers/alerts.py
index c4b7f001..f608d381 100644
--- a/src/serving/api/routers/alerts.py
+++ b/src/serving/api/routers/alerts.py
@@ -10,12 +10,11 @@
     deactivate_alert,
     ensure_alert_dispatcher,
     get_alert,
-    get_alert_config_path,
-    get_alert_history,
     list_alerts,
     update_alert,
 )
 from src.serving.api.egress_guard import UnsafeEgressURLError, validate_public_url
+from src.serving.control_plane import get_control_plane_store
 
 router = APIRouter(prefix="/v1/alerts", tags=["alerts"])
 
@@ -74,7 +73,7 @@ async def register_alert(payload: AlertCreateRequest, request: Request) -> dict[
     except UnsafeEgressURLError as exc:
         raise HTTPException(status_code=400, detail=f"Unsafe webhook URL: {exc}") from exc
     rule = create_alert(
-        get_alert_config_path(request.app),
+        request.app,
         name=payload.name,
         tenant=_tenant(request),
         metric=payload.metric,
@@ -90,7 +89,7 @@ async def register_alert(payload: AlertCreateRequest, request: Request) -> dict[
 
 @router.get("")
 async def list_my_alerts(request: Request) -> dict[str, object]:
-    alerts = list_alerts(get_alert_config_path(request.app), _tenant(request))
+    alerts = list_alerts(request.app, _tenant(request))
     # Exclude signing `secret` from list responses (audit p2_2 #7).
     return {"alerts": [alert.model_dump(mode="json", exclude={"secret"}) for alert in alerts]}
 
@@ -99,8 +98,7 @@ async def list_my_alerts(request: Request) -> dict[str, object]:
 async def modify_alert(
     alert_id: str, payload: AlertUpdateRequest, request: Request
 ) -> dict[str, object]:
-    path = get_alert_config_path(request.app)
-    existing = get_alert(path, alert_id, _tenant(request))
+    existing = get_alert(request.app, alert_id, _tenant(request))
     if existing is None:
         raise HTTPException(status_code=404, detail=f"Alert '{alert_id}' not found.")
 
@@ -114,7 +112,7 @@ async def modify_alert(
         except UnsafeEgressURLError as exc:
             raise HTTPException(status_code=400, detail=f"Unsafe webhook URL: {exc}") from exc
 
-    updated = update_alert(path, alert_id, _tenant(request), updates)
+    updated = update_alert(request.app, alert_id, _tenant(request), updates)
     if updated is None:
         raise HTTPException(status_code=404, detail=f"Alert '{alert_id}' not found.")
     # Exclude signing `secret` from update responses (audit p2_2 #7).
@@ -123,11 +121,7 @@ async def modify_alert(
 
 @router.delete("/{alert_id}", status_code=status.HTTP_204_NO_CONTENT)
 async def remove_alert(alert_id: str, request: Request) -> Response:
-    removed = deactivate_alert(
-        get_alert_config_path(request.app),
-        alert_id,
-        _tenant(request),
-    )
+    removed = deactivate_alert(request.app, alert_id, _tenant(request))
     if not removed:
         raise HTTPException(status_code=404, detail=f"Alert '{alert_id}' not found.")
     return Response(status_code=status.HTTP_204_NO_CONTENT)
@@ -135,7 +129,7 @@ async def remove_alert(alert_id: str, request: Request) -> Response:
 
 @router.post("/{alert_id}/test")
 async def test_alert(alert_id: str, request: Request) -> dict[str, object]:
-    rule = get_alert(get_alert_config_path(request.app), alert_id, _tenant(request))
+    rule = get_alert(request.app, alert_id, _tenant(request))
     if rule is None:
         raise HTTPException(status_code=404, detail=f"Alert '{alert_id}' not found.")
     return cast(dict[str, object], await ensure_alert_dispatcher(request.app).send_test_alert(rule))
@@ -143,7 +137,7 @@ async def test_alert(alert_id: str, request: Request) -> dict[str, object]:
 
 @router.get("/{alert_id}/history")
 async def alert_history(alert_id: str, request: Request) -> dict[str, object]:
-    rule = get_alert(get_alert_config_path(request.app), alert_id, _tenant(request))
+    rule = get_alert(request.app, alert_id, _tenant(request))
     if rule is None:
         raise HTTPException(status_code=404, detail=f"Alert '{alert_id}' not found.")
     history = await run_in_threadpool(_read_alert_history, request, alert_id)
@@ -151,11 +145,8 @@ async def alert_history(alert_id: str, request: Request) -> dict[str, object]:
 
 
 def _read_alert_history(request: Request, alert_id: str) -> list[dict]:
-    # The history 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_alert_history(cursor, alert_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_alert_delivery_history(alert_id)
diff --git a/src/serving/control_plane/__init__.py b/src/serving/control_plane/__init__.py
index 9f1b1691..faacd33b 100644
--- a/src/serving/control_plane/__init__.py
+++ b/src/serving/control_plane/__init__.py
@@ -8,6 +8,7 @@
 
 from .embedded import (
     EmbeddedControlPlaneStore,
+    ensure_alert_history_table,
     ensure_webhook_deliveries_table,
     ensure_webhook_delivery_queue_table,
 )
@@ -23,6 +24,7 @@
     "ControlPlaneStore",
     "EmbeddedControlPlaneStore",
     "WebhookQueueRow",
+    "ensure_alert_history_table",
     "ensure_webhook_deliveries_table",
     "ensure_webhook_delivery_queue_table",
     "get_control_plane_store",
diff --git a/src/serving/control_plane/embedded.py b/src/serving/control_plane/embedded.py
index 2bdc29be..907db0d5 100644
--- a/src/serving/control_plane/embedded.py
+++ b/src/serving/control_plane/embedded.py
@@ -14,8 +14,10 @@
 
 from __future__ import annotations
 
+import json
 from collections.abc import Callable, Sequence
 from datetime import UTC, datetime, timedelta
+from pathlib import Path
 
 import duckdb
 
@@ -23,6 +25,11 @@
 
 from .store import ControlPlaneStore, WebhookQueueRow
 
+try:
+    import yaml
+except ImportError:  # pragma: no cover
+    yaml = None  # type: ignore[assignment]
+
 
 def ensure_webhook_deliveries_table(conn: duckdb.DuckDBPyConnection) -> None:
     # Serialize the lazy DDL: the offloaded read handler calls this on a worker
@@ -85,16 +92,66 @@ def ensure_webhook_delivery_queue_table(conn: duckdb.DuckDBPyConnection) -> None
         )
 
 
+def ensure_alert_history_table(conn: duckdb.DuckDBPyConnection) -> None:
+    # Moved verbatim from alerts/history.py in ADR 0010 slice 2; same
+    # catalog-DDL-lock discipline as its ensure_webhook_* siblings above
+    # (audit_30 A2 follow-up: #120 offload race).
+    with catalog_ddl_lock:
+        conn.execute(
+            """
+            CREATE TABLE IF NOT EXISTS alert_history (
+                delivery_id VARCHAR,
+                alert_id VARCHAR,
+                alert_name VARCHAR,
+                metric VARCHAR,
+                current_value DOUBLE,
+                previous_value DOUBLE,
+                change_pct DOUBLE,
+                threshold DOUBLE,
+                condition VARCHAR,
+                metric_window VARCHAR,
+                tenant VARCHAR,
+                event_type VARCHAR,
+                status_code INTEGER,
+                success BOOLEAN,
+                error TEXT,
+                payload JSON,
+                triggered_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+            )
+            """
+        )
+
+
 class EmbeddedControlPlaneStore(ControlPlaneStore):
-    """Control-plane state on the embedded serving DuckDB connection.
+    """Control-plane state on the embedded serving DuckDB connection (queue,
+    log and history tables) plus the YAML-backed alert-rule repository.
 
     ``conn_provider`` is resolved per call (not captured once): tests and the
     lifespan may swap ``app.state.query_engine``, and the store must follow
     the live connection exactly like the pre-port ``_conn`` lookups did.
+    ``alert_rules_path_provider`` is resolved the same way — the alert config
+    path is per-app configurable (``app.state.alert_config_path``) and tests
+    swap it per case.
     """
 
-    def __init__(self, conn_provider: Callable[[], duckdb.DuckDBPyConnection]) -> None:
+    def __init__(
+        self,
+        conn_provider: Callable[[], duckdb.DuckDBPyConnection],
+        *,
+        alert_rules_path_provider: Callable[[], Path] | None = None,
+    ) -> None:
         self._conn_provider = conn_provider
+        self._alert_rules_path_provider = alert_rules_path_provider
+
+    @property
+    def _alert_rules_path(self) -> Path:
+        if self._alert_rules_path_provider is None:
+            raise RuntimeError(
+                "EmbeddedControlPlaneStore was constructed without an "
+                "alert_rules_path_provider; alert-rule repository methods "
+                "are unavailable."
+            )
+        return self._alert_rules_path_provider()
 
     @property
     def _conn(self) -> duckdb.DuckDBPyConnection:
@@ -270,3 +327,113 @@ def get_webhook_delivery_logs(self, webhook_id: str, *, limit: int = 20) -> list
             return [dict(zip(columns, row, strict=False)) for row in result.fetchall()]
         finally:
             cursor.close()
+
+    # --- 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:
+        conn = self._conn
+        ensure_alert_history_table(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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+            [
+                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),
+                datetime.now(UTC),
+            ],
+        )
+
+    def get_alert_delivery_history(self, alert_id: str, *, limit: int = 20) -> list[dict]:
+        # A dedicated cursor per read — not the shared connection — keeps
+        # concurrent reads on worker threads (run_in_threadpool) from colliding
+        # on the connection. (audit_30_06_26.md A2)
+        cursor = self._conn.cursor()
+        try:
+            ensure_alert_history_table(cursor)
+            result = cursor.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 = ?
+                ORDER BY triggered_at DESC
+                LIMIT ?
+                """,
+                [alert_id, limit],
+            )
+            columns = [description[0] for description in result.description]
+            records = [dict(zip(columns, row, strict=False)) for row in result.fetchall()]
+        finally:
+            cursor.close()
+        for record in records:
+            payload = record.get("payload")
+            if isinstance(payload, str):
+                try:
+                    record["payload"] = json.loads(payload)
+                except json.JSONDecodeError:
+                    pass
+        return records
+
+    # --- alert rule repository (mutable runtime state) ------------------------
+
+    def load_alert_rules(self) -> list[dict]:
+        path = self._alert_rules_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("alerts", []))
+
+    def save_alert_rules(self, rules: list[dict]) -> None:
+        path = self._alert_rules_path
+        path.parent.mkdir(parents=True, exist_ok=True)
+        payload = {"alerts": rules}
+        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")
diff --git a/src/serving/control_plane/store.py b/src/serving/control_plane/store.py
index 7b5f0953..2786f63e 100644
--- a/src/serving/control_plane/store.py
+++ b/src/serving/control_plane/store.py
@@ -19,9 +19,19 @@
   PostgreSQL adapter uses ``FOR UPDATE SKIP LOCKED`` plus a lease column so
   N replicas work-steal without leader election.
 
-Rollout slice 1 covers the webhook delivery queue + attempt log; alert
-rules/history, outbox/dead-letter and usage migrate behind this port in
-follow-up slices (see the ADR's rollout section).
+Rollout slice 1 covers the webhook delivery queue + attempt log. Slice 2
+(this module's alert methods) covers the alert delivery/history log and the
+alert-rule repository — including the rules' mutable runtime state (``state``,
+``fired_at``, ``last_escalation_level``, flap window, cooldown), the sharpest
+per-pod split-brain in the ADR's inventory. Outbox/dead-letter and usage
+migrate behind this port in follow-up slices (see the ADR's rollout section).
+
+The alert-rule repository methods (``load_alert_rules`` / ``save_alert_rules``)
+operate on plain JSON-shaped ``dict`` records rather than the ``AlertRule``
+pydantic model: the port must not import ``src.serving.api.alerts`` (that
+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.
 """
 
 from __future__ import annotations
@@ -121,6 +131,51 @@ def get_webhook_delivery_logs(self, webhook_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)."""
 
+    # --- alert delivery history (append-only) --------------------------------
+
+    @abstractmethod
+    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:
+        """Append one alert-delivery attempt to the ``alert_history`` log."""
+
+    @abstractmethod
+    def get_alert_delivery_history(self, alert_id: str, *, limit: int = 20) -> list[dict]:
+        """Most recent alert-history entries for one alert, newest first.
+
+        Safe to call from a worker thread (adapters isolate the read — the
+        embedded store opens a dedicated cursor per call, audit_30 A2)."""
+
+    # --- alert rule repository (mutable runtime state) ------------------------
+
+    @abstractmethod
+    def load_alert_rules(self) -> list[dict]:
+        """Return every alert rule as a JSON-shaped record (the caller
+        validates each into ``AlertRule``), including mutable runtime state
+        (``state``, ``fired_at``, ``last_escalation_level``, flap window)."""
+
+    @abstractmethod
+    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)."""
+
 
 def get_control_plane_store(app: FastAPI) -> ControlPlaneStore:
     """Resolve the app's control-plane store, creating the configured one on
@@ -137,12 +192,21 @@ def get_control_plane_store(app: FastAPI) -> ControlPlaneStore:
         return store
     kind = (os.getenv(CONTROL_PLANE_STORE_ENV) or "embedded").strip().lower()
     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.
+        from src.serving.api.alerts.dispatcher import get_alert_config_path
+
         from .embedded import EmbeddedControlPlaneStore
 
         # The one sanctioned reach into the engine's embedded connection: the
         # composition seam that binds the default profile's store to the
         # serving DuckDB. Everything above the port goes through the store.
-        store = EmbeddedControlPlaneStore(conn_provider=lambda: app.state.query_engine._conn)
+        store = EmbeddedControlPlaneStore(
+            conn_provider=lambda: app.state.query_engine._conn,
+            alert_rules_path_provider=lambda: get_alert_config_path(app),
+        )
     elif kind == "postgres":
         raise NotImplementedError(
             "AGENTFLOW_CONTROLPLANE_STORE=postgres is the ADR 0010 scale profile; "
diff --git a/tests/integration/test_alert_dedup.py b/tests/integration/test_alert_dedup.py
index 3d8a52e1..3108fd0e 100644
--- a/tests/integration/test_alert_dedup.py
+++ b/tests/integration/test_alert_dedup.py
@@ -11,8 +11,8 @@
 
 from src.serving.api.alerts import dispatcher as alerts_dispatcher
 from src.serving.api.alerts import escalation as alerts_escalation
-from src.serving.api.alerts import history as alerts_history
 from src.serving.api.main import app
+from src.serving.control_plane import embedded as control_plane_embedded
 
 
 class _FrozenDateTime:
@@ -67,7 +67,9 @@ def warning(self, event: str, **kwargs) -> None:
 def freeze_time(monkeypatch) -> _Clock:
     clock = _Clock()
     monkeypatch.setattr(alerts_dispatcher, "datetime", _FrozenDateTime)
-    monkeypatch.setattr(alerts_history, "datetime", _FrozenDateTime)
+    # log_alert_delivery's triggered_at default lives in the embedded store
+    # since ADR 0010 slice 2 (moved from alerts/history.py).
+    monkeypatch.setattr(control_plane_embedded, "datetime", _FrozenDateTime)
     return clock
 
 
diff --git a/tests/unit/test_alert_dispatcher_backcompat.py b/tests/unit/test_alert_dispatcher_backcompat.py
index ac35421d..850e2c6f 100644
--- a/tests/unit/test_alert_dispatcher_backcompat.py
+++ b/tests/unit/test_alert_dispatcher_backcompat.py
@@ -3,7 +3,6 @@
     AlertDispatcher,
     create_alert,
     ensure_alert_dispatcher,
-    get_alert_history,
 )
 
 
@@ -11,4 +10,3 @@ def test_alert_dispatcher_keeps_backwards_compatible_exports():
     assert alert_dispatcher.AlertDispatcher is AlertDispatcher
     assert alert_dispatcher.create_alert is create_alert
     assert alert_dispatcher.ensure_alert_dispatcher is ensure_alert_dispatcher
-    assert alert_dispatcher.get_alert_history is get_alert_history
diff --git a/tests/unit/test_alerts_router_unit.py b/tests/unit/test_alerts_router_unit.py
index caa19e03..edc109a1 100644
--- a/tests/unit/test_alerts_router_unit.py
+++ b/tests/unit/test_alerts_router_unit.py
@@ -12,7 +12,6 @@
 from __future__ import annotations
 
 from datetime import UTC, datetime
-from pathlib import Path
 from types import SimpleNamespace
 from typing import Any
 
@@ -84,13 +83,6 @@ async def send_test_alert(self, alert: AlertRule) -> dict[str, object]:
         return {"status": "sent", "alert_id": alert.id}
 
 
-@pytest.fixture(autouse=True)
-def _stub_config_path(monkeypatch: pytest.MonkeyPatch) -> None:
-    # Every endpoint resolves the config path first; pin it so no test touches
-    # the real config/alerts.yaml.
-    monkeypatch.setattr(alerts_module, "get_alert_config_path", lambda app: Path("alerts.yaml"))
-
-
 # ── _tenant ──────────────────────────────────────────────────────
 
 
@@ -131,7 +123,7 @@ async def test_register_alert_validates_creates_and_starts_dispatcher(
 ) -> None:
     created: dict[str, Any] = {}
 
-    def fake_create(path: Path, **kwargs: Any) -> AlertRule:
+    def fake_create(app: Any, **kwargs: Any) -> AlertRule:
         created.update(kwargs)
         return _rule(name=kwargs["name"], tenant=kwargs["tenant"])
 
@@ -179,7 +171,7 @@ async def test_register_alert_rejects_unknown_metric(monkeypatch: pytest.MonkeyP
 
 async def test_list_alerts_excludes_signing_secret(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr(
-        alerts_module, "list_alerts", lambda path, tenant: [_rule(), _rule(id="al-2")]
+        alerts_module, "list_alerts", lambda app, tenant: [_rule(), _rule(id="al-2")]
     )
     result = await list_my_alerts(_req())
 
@@ -191,16 +183,16 @@ async def test_list_alerts_excludes_signing_secret(monkeypatch: pytest.MonkeyPat
 
 
 async def test_modify_alert_not_found_raises_404(monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setattr(alerts_module, "get_alert", lambda path, alert_id, tenant: None)
+    monkeypatch.setattr(alerts_module, "get_alert", lambda app, alert_id, tenant: None)
     with pytest.raises(HTTPException) as exc:
         await modify_alert("al-x", AlertUpdateRequest(threshold=5.0), _req())
     assert exc.value.status_code == 404
 
 
 async def test_modify_alert_updates_and_excludes_secret(monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setattr(alerts_module, "get_alert", lambda path, alert_id, tenant: _rule())
+    monkeypatch.setattr(alerts_module, "get_alert", lambda app, alert_id, tenant: _rule())
     monkeypatch.setattr(
-        alerts_module, "update_alert", lambda path, alert_id, tenant, updates: _rule(threshold=5.0)
+        alerts_module, "update_alert", lambda app, alert_id, tenant, updates: _rule(threshold=5.0)
     )
     result = await modify_alert("al-1", AlertUpdateRequest(threshold=5.0), _req())
 
@@ -210,15 +202,15 @@ async def test_modify_alert_updates_and_excludes_secret(monkeypatch: pytest.Monk
 
 async def test_modify_alert_lost_race_returns_404(monkeypatch: pytest.MonkeyPatch) -> None:
     # get_alert sees the rule but update_alert finds it already gone.
-    monkeypatch.setattr(alerts_module, "get_alert", lambda path, alert_id, tenant: _rule())
-    monkeypatch.setattr(alerts_module, "update_alert", lambda path, alert_id, tenant, updates: None)
+    monkeypatch.setattr(alerts_module, "get_alert", lambda app, alert_id, tenant: _rule())
+    monkeypatch.setattr(alerts_module, "update_alert", lambda app, alert_id, tenant, updates: None)
     with pytest.raises(HTTPException) as exc:
         await modify_alert("al-1", AlertUpdateRequest(threshold=5.0), _req())
     assert exc.value.status_code == 404
 
 
 async def test_modify_alert_revalidates_new_metric(monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setattr(alerts_module, "get_alert", lambda path, alert_id, tenant: _rule())
+    monkeypatch.setattr(alerts_module, "get_alert", lambda app, alert_id, tenant: _rule())
     monkeypatch.setattr(
         alerts_module,
         "update_alert",
@@ -233,14 +225,14 @@ async def test_modify_alert_revalidates_new_metric(monkeypatch: pytest.MonkeyPat
 
 
 async def test_remove_alert_returns_204(monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setattr(alerts_module, "deactivate_alert", lambda path, alert_id, tenant: True)
+    monkeypatch.setattr(alerts_module, "deactivate_alert", lambda app, alert_id, tenant: True)
     result = await remove_alert("al-1", _req())
     assert isinstance(result, Response)
     assert result.status_code == 204
 
 
 async def test_remove_alert_not_found_raises_404(monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setattr(alerts_module, "deactivate_alert", lambda path, alert_id, tenant: False)
+    monkeypatch.setattr(alerts_module, "deactivate_alert", lambda app, alert_id, tenant: False)
     with pytest.raises(HTTPException) as exc:
         await remove_alert("al-x", _req())
     assert exc.value.status_code == 404
@@ -251,7 +243,7 @@ async def test_remove_alert_not_found_raises_404(monkeypatch: pytest.MonkeyPatch
 
 async def test_test_alert_sends_through_dispatcher(monkeypatch: pytest.MonkeyPatch) -> None:
     dispatcher = _FakeDispatcher()
-    monkeypatch.setattr(alerts_module, "get_alert", lambda path, alert_id, tenant: _rule())
+    monkeypatch.setattr(alerts_module, "get_alert", lambda app, alert_id, tenant: _rule())
     monkeypatch.setattr(alerts_module, "ensure_alert_dispatcher", lambda app: dispatcher)
 
     result = await _test_alert_endpoint("al-1", _req())
@@ -261,7 +253,7 @@ async def test_test_alert_sends_through_dispatcher(monkeypatch: pytest.MonkeyPat
 
 
 async def test_test_alert_not_found_raises_404(monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setattr(alerts_module, "get_alert", lambda path, alert_id, tenant: None)
+    monkeypatch.setattr(alerts_module, "get_alert", lambda app, alert_id, tenant: None)
     with pytest.raises(HTTPException) as exc:
         await _test_alert_endpoint("al-x", _req())
     assert exc.value.status_code == 404
@@ -270,29 +262,30 @@ async def test_test_alert_not_found_raises_404(monkeypatch: pytest.MonkeyPatch)
 # ── alert_history ────────────────────────────────────────────────
 
 
-class _CursorConn:
-    """A connection stub that yields a closeable cursor — the history read
-    offloads onto a dedicated cursor (audit_30 A2). get_alert_history is stubbed,
-    so the cursor itself is never queried."""
+class _FakeStore:
+    """Stands in for ``get_control_plane_store(app)`` — the history read goes
+    through the store port, not a raw ``query_engine._conn`` cursor."""
 
-    def cursor(self) -> _CursorConn:
-        return self
+    def __init__(self, history: list[dict]) -> None:
+        self._history = history
 
-    def close(self) -> None:
-        pass
+    def get_alert_delivery_history(self, alert_id: str) -> list[dict]:
+        return self._history
 
 
 async def test_alert_history_returns_records(monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setattr(alerts_module, "get_alert", lambda path, alert_id, tenant: _rule())
+    monkeypatch.setattr(alerts_module, "get_alert", lambda app, alert_id, tenant: _rule())
     monkeypatch.setattr(
-        alerts_module, "get_alert_history", lambda conn, alert_id: [{"fired_at": "2026-06-13"}]
+        alerts_module,
+        "get_control_plane_store",
+        lambda app: _FakeStore([{"fired_at": "2026-06-13"}]),
     )
-    result = await alert_history("al-1", _req(query_conn=_CursorConn()))
+    result = await alert_history("al-1", _req())
     assert result == {"history": [{"fired_at": "2026-06-13"}]}
 
 
 async def test_alert_history_not_found_raises_404(monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setattr(alerts_module, "get_alert", lambda path, alert_id, tenant: None)
+    monkeypatch.setattr(alerts_module, "get_alert", lambda app, alert_id, tenant: None)
     with pytest.raises(HTTPException) as exc:
         await alert_history("al-x", _req())
     assert exc.value.status_code == 404
diff --git a/tests/unit/test_blocking_routes_offloaded.py b/tests/unit/test_blocking_routes_offloaded.py
index 6af458b7..e5383ccb 100644
--- a/tests/unit/test_blocking_routes_offloaded.py
+++ b/tests/unit/test_blocking_routes_offloaded.py
@@ -164,7 +164,6 @@ async def test_deadletter_stats_does_not_block_event_loop() -> None:
 async def test_alert_history_does_not_block_event_loop(monkeypatch: pytest.MonkeyPatch) -> None:
     # The rule lookup is a YAML read, not the DuckDB scan under test — stub it so
     # the handler reaches the history scan with a non-None rule.
-    monkeypatch.setattr(alerts_module, "get_alert_config_path", lambda app: Path("unused"))
     monkeypatch.setattr(alerts_module, "get_alert", lambda *a, **k: SimpleNamespace(id="A1"))
     app = FastAPI()
     app.state.query_engine = SimpleNamespace(_conn=_SleepyHistoryConn(delay_seconds=0.3))
diff --git a/tests/unit/test_catalog_ddl_concurrency.py b/tests/unit/test_catalog_ddl_concurrency.py
index 02d201e8..08e35df2 100644
--- a/tests/unit/test_catalog_ddl_concurrency.py
+++ b/tests/unit/test_catalog_ddl_concurrency.py
@@ -7,12 +7,11 @@
 import pytest
 
 import src.processing.event_replayer as event_replayer_module
-import src.serving.api.alerts.history as history_module
 import src.serving.control_plane.embedded as control_plane_embedded_module
 from src.db_concurrency import catalog_ddl_lock
 from src.processing.event_replayer import ensure_dead_letter_table
-from src.serving.api.alerts.history import ensure_alert_history_table
 from src.serving.control_plane import (
+    ensure_alert_history_table,
     ensure_webhook_deliveries_table,
     ensure_webhook_delivery_queue_table,
 )
@@ -83,12 +82,11 @@ def test_ensure_tables_concurrency_safe_across_tables_cold_db() -> None:
 
 
 def test_catalog_ddl_lock_is_a_single_shared_instance() -> None:
-    # Every ensure_* helper (across all three modules) must guard DDL with the
+    # Every ensure_* helper (across both modules) must guard DDL with the
     # *same* lock instance, or the cross-table conflict resurfaces. Pins against a
     # future per-module lock.
     assert (
         event_replayer_module.catalog_ddl_lock
-        is history_module.catalog_ddl_lock
         is control_plane_embedded_module.catalog_ddl_lock
         is catalog_ddl_lock
     )
diff --git a/tests/unit/test_control_plane_store.py b/tests/unit/test_control_plane_store.py
index 98a87f66..e762ca38 100644
--- a/tests/unit/test_control_plane_store.py
+++ b/tests/unit/test_control_plane_store.py
@@ -1,10 +1,11 @@
-"""ADR 0010 slice 1: the ControlPlaneStore port and its embedded adapter.
+"""ADR 0010 slices 1-2: the ControlPlaneStore port and its embedded adapter.
 
 Covers the store-level contract the dispatcher relies on (enqueue-win
 semantics, due-claim ordering and bounds, the outcome state machine, parking,
-attempt-log roundtrip), the ``get_control_plane_store`` resolution/ratchet,
-and the structural pin that keeps the webhook path from re-growing direct
-``query_engine._conn`` reaches.
+attempt-log roundtrip), the alert-delivery history log and the YAML-backed
+alert-rule repository (slice 2), the ``get_control_plane_store``
+resolution/ratchet, and the structural pin that keeps the webhook and alert
+paths from re-growing direct ``query_engine._conn`` reaches.
 """
 
 from __future__ import annotations
@@ -40,6 +41,14 @@ def store(conn: duckdb.DuckDBPyConnection) -> EmbeddedControlPlaneStore:
     return EmbeddedControlPlaneStore(conn_provider=lambda: conn)
 
 
+@pytest.fixture
+def alert_store(conn: duckdb.DuckDBPyConnection, tmp_path: Path) -> EmbeddedControlPlaneStore:
+    path = tmp_path / "alerts.yaml"
+    return EmbeddedControlPlaneStore(
+        conn_provider=lambda: conn, alert_rules_path_provider=lambda: path
+    )
+
+
 def _enqueue(store: EmbeddedControlPlaneStore, webhook_id: str, event_id: str) -> bool:
     return store.enqueue_webhook_delivery(
         webhook_id=webhook_id,
@@ -205,6 +214,100 @@ def test_delivery_log_roundtrip_is_isolated_per_webhook_and_newest_first(
     assert store.get_webhook_delivery_logs("wh-1", limit=1)[0]["delivery_id"] in {"d1", "d2"}
 
 
+# --- alert delivery history -----------------------------------------------------
+
+
+def _log_alert(
+    store: EmbeddedControlPlaneStore, *, delivery_id: str, alert_id: str, success: bool
+) -> None:
+    store.log_alert_delivery(
+        delivery_id=delivery_id,
+        alert_id=alert_id,
+        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 if success else 500,
+        success=success,
+        error=None if success else "boom",
+        payload={"alert_id": alert_id, "status": "firing"},
+    )
+
+
+def test_alert_delivery_log_roundtrip_is_isolated_per_alert_and_newest_first(
+    alert_store: EmbeddedControlPlaneStore,
+) -> None:
+    _log_alert(alert_store, delivery_id="d1", alert_id="a1", success=False)
+    _log_alert(alert_store, delivery_id="d2", alert_id="a1", success=True)
+    _log_alert(alert_store, delivery_id="d-other", alert_id="a-other", success=True)
+
+    history = alert_store.get_alert_delivery_history("a1")
+
+    assert [entry["delivery_id"] for entry in history] == ["d2", "d1"]
+    assert history[0]["success"] is True
+    assert history[1]["error"] == "boom"
+    assert alert_store.get_alert_delivery_history("a-ghost") == []
+    assert alert_store.get_alert_delivery_history("a1", limit=1) == history[:1]
+
+
+def test_alert_delivery_history_decodes_json_payload(
+    alert_store: EmbeddedControlPlaneStore,
+) -> None:
+    _log_alert(alert_store, delivery_id="d1", alert_id="a1", success=True)
+
+    history = alert_store.get_alert_delivery_history("a1")
+
+    assert history[0]["payload"] == {"alert_id": "a1", "status": "firing"}
+
+
+# --- alert rule repository (YAML round-trip) --------------------------------------
+
+
+def test_load_alert_rules_returns_empty_list_when_file_is_missing(
+    alert_store: EmbeddedControlPlaneStore,
+) -> None:
+    assert alert_store.load_alert_rules() == []
+
+
+def test_save_then_load_alert_rules_round_trips_verbatim(
+    alert_store: EmbeddedControlPlaneStore,
+) -> None:
+    rules = [
+        {"id": "a1", "name": "High error rate", "state": "firing", "last_escalation_level": 2},
+        {"id": "a2", "name": "Low revenue", "state": "ok", "last_escalation_level": 0},
+    ]
+
+    alert_store.save_alert_rules(rules)
+
+    assert alert_store.load_alert_rules() == rules
+
+
+def test_save_alert_rules_creates_parent_directories(tmp_path: Path, conn) -> None:
+    nested_path = tmp_path / "nested" / "alerts.yaml"
+    store = EmbeddedControlPlaneStore(
+        conn_provider=lambda: conn, alert_rules_path_provider=lambda: nested_path
+    )
+
+    store.save_alert_rules([{"id": "a1"}])
+
+    assert nested_path.exists()
+    assert store.load_alert_rules() == [{"id": "a1"}]
+
+
+def test_alert_rules_methods_require_a_path_provider(store: EmbeddedControlPlaneStore) -> None:
+    # ``store`` (unlike ``alert_store``) was constructed without
+    # alert_rules_path_provider — mirrors a caller that only wired the webhook
+    # queue's conn_provider and forgot the alert-rule repository.
+    with pytest.raises(RuntimeError, match="alert_rules_path_provider"):
+        store.load_alert_rules()
+
+
 # --- get_control_plane_store resolution + ratchet -------------------------------
 
 
@@ -269,13 +372,16 @@ def test_get_store_follows_a_swapped_query_engine(
 
 
 def test_webhook_path_does_not_reach_into_the_engine_connection() -> None:
-    """ADR 0010 slice 1 ratchet: the dispatcher and the webhooks router go
-    through the ControlPlaneStore port; direct ``query_engine._conn`` reaches
-    must not re-grow there. The single sanctioned reach is the composition
-    seam in ``control_plane/store.py``."""
+    """ADR 0010 slices 1-2 ratchet: the webhook/alert dispatchers and their
+    routers go through the ControlPlaneStore port; direct ``query_engine._conn``
+    reaches must not re-grow there. The single sanctioned reach is the
+    composition seam in ``control_plane/store.py``."""
     for relative in (
         "src/serving/api/webhook_dispatcher.py",
         "src/serving/api/routers/webhooks.py",
+        "src/serving/api/alerts/dispatcher.py",
+        "src/serving/api/alerts/escalation.py",
+        "src/serving/api/routers/alerts.py",
     ):
         source = (PROJECT_ROOT / relative).read_text(encoding="utf-8")
         assert "query_engine._conn" not in source, relative