Merge pull request 'feat(control-plane): durable MCP session checkpoint schema (Closes #660)' (#881) from feat/issue-660-session-checkpoint-schema into master

control_plane_db.py

@@ -30,6 +30,7 @@ from datetime import datetime, timedelta, timezone
 from typing import Any, Iterator, Sequence
 
 import dependency_graph
+import gitea_audit
 
 SCHEMA_VERSION = 5
 
@@ -177,6 +178,51 @@ CREATE TABLE IF NOT EXISTS dependency_edges (
     )
 );
 
+-- Durable MCP session checkpoints (#660, umbrella #655 child; #628 handoff
+-- goal). A restart otherwise loses session identity, stage, lease ownership,
+-- and next action, forcing human reconstruction. Each row is the *current*
+-- recoverable state of one session's work on one work unit; stage transitions
+-- upsert the row and audit the change to ``events``. Creating the table is the
+-- v4->v5 migration: additive, idempotent, and it never touches prior tables.
+--
+-- ``lease_id`` / ``assignment_id`` are recorded as soft references (plain TEXT,
+-- no enforced FK) exactly as dependency_edges references issues/PRs by number:
+-- a checkpoint is a recovery artifact that must survive the deletion of the
+-- lease it names, so a hard FK would fail the pre-drain write the issue
+-- requires to succeed. ``work_number`` uses 0 (never a real issue/PR number)
+-- as the "session-level, no specific work" sentinel so UNIQUE is NULL-safe.
+-- Every free-text / JSON field is redaction-filtered before storage (AC4).
+CREATE TABLE IF NOT EXISTS session_checkpoints (
+    checkpoint_id TEXT PRIMARY KEY,
+    remote TEXT NOT NULL,
+    org TEXT NOT NULL,
+    repo TEXT NOT NULL,
+    session_id TEXT NOT NULL,
+    provider_identity TEXT NOT NULL DEFAULT '',
+    role TEXT NOT NULL DEFAULT '',
+    work_kind TEXT NOT NULL DEFAULT '' CHECK (work_kind IN ('issue', 'pr', '')),
+    work_number INTEGER NOT NULL DEFAULT 0,
+    worktree_path TEXT NOT NULL DEFAULT '',
+    branch TEXT NOT NULL DEFAULT '',
+    head_sha TEXT NOT NULL DEFAULT '',
+    capabilities TEXT NOT NULL DEFAULT '[]',
+    lease_id TEXT NOT NULL DEFAULT '',
+    assignment_id TEXT NOT NULL DEFAULT '',
+    workflow_stage TEXT NOT NULL DEFAULT '',
+    last_completed_action TEXT NOT NULL DEFAULT '',
+    current_operation TEXT NOT NULL DEFAULT '',
+    pending_mutation TEXT NOT NULL DEFAULT '',
+    evidence TEXT NOT NULL DEFAULT '{}',
+    blocker TEXT NOT NULL DEFAULT '',
+    next_valid_action TEXT NOT NULL DEFAULT '',
+    recovery_instructions TEXT NOT NULL DEFAULT '',
+    checkpoint_schema_version INTEGER NOT NULL DEFAULT 5,
+    status TEXT NOT NULL DEFAULT 'active',
+    created_at TEXT NOT NULL,
+    updated_at TEXT NOT NULL,
+    UNIQUE (remote, org, repo, session_id, work_kind, work_number)
+);
+
 CREATE INDEX IF NOT EXISTS idx_leases_work_status ON leases(work_item_id, status);
 -- Reverse lookup ("what waits on this target") is the query automatic
 -- resumption needs, so it gets its own index alongside the forward one.
@@ -186,6 +232,12 @@ CREATE INDEX IF NOT EXISTS idx_dependency_edges_target
     ON dependency_edges(remote, org, repo, target_kind, target_number);
 CREATE INDEX IF NOT EXISTS idx_assignments_session ON assignments(session_id, status);
 CREATE INDEX IF NOT EXISTS idx_incident_gitea ON incident_links(gitea_org, gitea_repo, gitea_issue_number);
+-- Resume queries hit by session (what was this session doing) and by work unit
+-- (who was checkpointed on this issue/PR), so both get an index.
+CREATE INDEX IF NOT EXISTS idx_session_checkpoints_session
+    ON session_checkpoints(remote, org, repo, session_id);
+CREATE INDEX IF NOT EXISTS idx_session_checkpoints_work
+    ON session_checkpoints(remote, org, repo, work_kind, work_number);
 
 -- Model usage, token cost, latency, and performance events (#651)
 CREATE TABLE IF NOT EXISTS usage_events (
@@ -2598,3 +2650,378 @@ class ControlPlaneDB:
             edge["prior_state"] = prior_state
             edge["state_changed"] = prior_state != state_norm
             return edge
+
+    # ------------------------------------------------------------------
+    # Session checkpoints (#660) — durable, redacted, reconcile-on-boot.
+    # ------------------------------------------------------------------
+
+    # Free-text columns that carry human/agent-authored strings. Each is
+    # redaction-filtered before storage so an accidentally pasted token or
+    # credential URL can never land in a checkpoint (#660 AC4).
+    _CHECKPOINT_TEXT_FIELDS: tuple[str, ...] = (
+        "provider_identity",
+        "role",
+        "worktree_path",
+        "branch",
+        "head_sha",
+        "lease_id",
+        "assignment_id",
+        "workflow_stage",
+        "last_completed_action",
+        "current_operation",
+        "blocker",
+        "next_valid_action",
+        "recovery_instructions",
+        "status",
+    )
+
+    # JSON columns whose *decoded* structure is redacted recursively.
+    _CHECKPOINT_JSON_FIELDS: tuple[str, ...] = (
+        "capabilities",
+        "pending_mutation",
+        "evidence",
+    )
+
+    # Minimum a checkpoint must carry to be a usable recovery record. The drain
+    # gate writes with ``require_complete=True``; an incomplete write fails
+    # closed so drain cannot complete on a checkpoint that can't resume (#660
+    # "fail closed if checkpoint incomplete during drain").
+    REQUIRED_CHECKPOINT_FIELDS_FOR_DRAIN: tuple[str, ...] = (
+        "session_id",
+        "role",
+        "workflow_stage",
+        "next_valid_action",
+        "recovery_instructions",
+    )
+
+    @classmethod
+    def checkpoint_completeness(cls, record: dict[str, Any]) -> list[str]:
+        """Return the drain-required fields that are missing/blank in *record*.
+
+        Pure (no DB). An empty list means the record is drain-complete.
+        """
+        missing: list[str] = []
+        for field in cls.REQUIRED_CHECKPOINT_FIELDS_FOR_DRAIN:
+            if not str(record.get(field) or "").strip():
+                missing.append(field)
+        return missing
+
+    def write_session_checkpoint(
+        self,
+        *,
+        remote: str,
+        org: str,
+        repo: str,
+        session_id: str,
+        provider_identity: str = "",
+        role: str = "",
+        work_kind: str | None = None,
+        work_number: int | None = None,
+        worktree_path: str = "",
+        branch: str = "",
+        head_sha: str = "",
+        capabilities: Any = None,
+        lease_id: str = "",
+        assignment_id: str = "",
+        workflow_stage: str = "",
+        last_completed_action: str = "",
+        current_operation: str = "",
+        pending_mutation: Any = None,
+        evidence: Any = None,
+        blocker: str = "",
+        next_valid_action: str = "",
+        recovery_instructions: str = "",
+        status: str = "active",
+        require_complete: bool = False,
+    ) -> dict[str, Any]:
+        """Upsert the current checkpoint for one session's work on one unit.
+
+        Keyed by ``(remote, org, repo, session_id, work_kind, work_number)`` so
+        a stage transition refreshes the single current row (transitions are
+        audited to ``events``) rather than appending unbounded history — the
+        row is *resumable state*, matching the dependency-edge current-state
+        model.
+
+        Every string / JSON field is redacted via ``gitea_audit.redact`` before
+        it touches the DB. With ``require_complete=True`` (the drain gate path)
+        a record missing any drain-required field raises ``ControlPlaneError``
+        and writes nothing, so drain cannot proceed on an unrecoverable record.
+        """
+        if not str(session_id or "").strip():
+            raise ControlPlaneError("session_id is required for a checkpoint (fail closed)")
+
+        # Normalize the work-unit key. Absent/blank work collapses to the
+        # ('', 0) session-level sentinel so UNIQUE stays NULL-safe.
+        if work_kind and str(work_kind).strip():
+            kind_norm = dependency_graph.normalize_work_kind(work_kind)
+            number_norm = int(work_number) if work_number is not None else 0
+        else:
+            kind_norm = ""
+            number_norm = 0
+
+        # Assemble, then redact the whole record in one pass.
+        raw_record: dict[str, Any] = {
+            "provider_identity": provider_identity or "",
+            "role": role or "",
+            "worktree_path": worktree_path or "",
+            "branch": branch or "",
+            "head_sha": head_sha or "",
+            "lease_id": lease_id or "",
+            "assignment_id": assignment_id or "",
+            "workflow_stage": workflow_stage or "",
+            "last_completed_action": last_completed_action or "",
+            "current_operation": current_operation or "",
+            "blocker": blocker or "",
+            "next_valid_action": next_valid_action or "",
+            "recovery_instructions": recovery_instructions or "",
+            "status": (status or "active"),
+            "session_id": session_id,
+            "capabilities": capabilities if capabilities is not None else [],
+            "pending_mutation": pending_mutation if pending_mutation is not None else {},
+            "evidence": evidence if evidence is not None else {},
+        }
+        clean = gitea_audit.redact(raw_record)
+
+        if require_complete:
+            missing = self.checkpoint_completeness(clean)
+            if missing:
+                raise ControlPlaneError(
+                    "checkpoint incomplete for drain; missing "
+                    f"{', '.join(missing)} (fail closed)"
+                )
+
+        capabilities_json = json.dumps(clean.get("capabilities") or [])
+        pending_json = json.dumps(clean.get("pending_mutation") or {})
+        evidence_json = json.dumps(clean.get("evidence") or {})
+        now_s = _ts()
+
+        with self._tx() as conn:
+            existing = conn.execute(
+                """
+                SELECT * FROM session_checkpoints
+                WHERE remote = ? AND org = ? AND repo = ?
+                  AND session_id = ? AND work_kind = ? AND work_number = ?
+                """,
+                (remote, org, repo, session_id, kind_norm, number_norm),
+            ).fetchone()
+
+            if existing is None:
+                checkpoint_id = uuid.uuid4().hex
+                conn.execute(
+                    """
+                    INSERT INTO session_checkpoints(
+                        checkpoint_id, remote, org, repo, session_id,
+                        provider_identity, role, work_kind, work_number,
+                        worktree_path, branch, head_sha, capabilities,
+                        lease_id, assignment_id, workflow_stage,
+                        last_completed_action, current_operation,
+                        pending_mutation, evidence, blocker, next_valid_action,
+                        recovery_instructions, checkpoint_schema_version,
+                        status, created_at, updated_at
+                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
+                              ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                    """,
+                    (
+                        checkpoint_id, remote, org, repo, session_id,
+                        clean["provider_identity"], clean["role"], kind_norm,
+                        number_norm, clean["worktree_path"], clean["branch"],
+                        clean["head_sha"], capabilities_json, clean["lease_id"],
+                        clean["assignment_id"], clean["workflow_stage"],
+                        clean["last_completed_action"], clean["current_operation"],
+                        pending_json, evidence_json, clean["blocker"],
+                        clean["next_valid_action"], clean["recovery_instructions"],
+                        SCHEMA_VERSION, clean["status"], now_s, now_s,
+                    ),
+                )
+            else:
+                checkpoint_id = str(existing["checkpoint_id"])
+                conn.execute(
+                    """
+                    UPDATE session_checkpoints
+                       SET provider_identity = ?, role = ?, worktree_path = ?,
+                           branch = ?, head_sha = ?, capabilities = ?,
+                           lease_id = ?, assignment_id = ?, workflow_stage = ?,
+                           last_completed_action = ?, current_operation = ?,
+                           pending_mutation = ?, evidence = ?, blocker = ?,
+                           next_valid_action = ?, recovery_instructions = ?,
+                           checkpoint_schema_version = ?, status = ?,
+                           updated_at = ?
+                     WHERE checkpoint_id = ?
+                    """,
+                    (
+                        clean["provider_identity"], clean["role"],
+                        clean["worktree_path"], clean["branch"], clean["head_sha"],
+                        capabilities_json, clean["lease_id"], clean["assignment_id"],
+                        clean["workflow_stage"], clean["last_completed_action"],
+                        clean["current_operation"], pending_json, evidence_json,
+                        clean["blocker"], clean["next_valid_action"],
+                        clean["recovery_instructions"], SCHEMA_VERSION,
+                        clean["status"], now_s, checkpoint_id,
+                    ),
+                )
+                prior_stage = str(existing["workflow_stage"] or "")
+                new_stage = str(clean["workflow_stage"] or "")
+                if prior_stage != new_stage:
+                    conn.execute(
+                        """
+                        INSERT INTO events(work_item_id, event_type, message, created_at)
+                        VALUES (NULL, 'session_checkpoint_stage_change', ?, ?)
+                        """,
+                        (
+                            f"checkpoint {checkpoint_id} session {session_id} "
+                            f"stage {prior_stage or '(none)'} -> "
+                            f"{new_stage or '(none)'}",
+                            now_s,
+                        ),
+                    )
+
+            row = conn.execute(
+                "SELECT * FROM session_checkpoints WHERE checkpoint_id = ?",
+                (checkpoint_id,),
+            ).fetchone()
+            return self._session_checkpoint_row(row) or {}
+
+    @staticmethod
+    def _session_checkpoint_row(row: sqlite3.Row | None) -> dict[str, Any] | None:
+        """Return a stored checkpoint as a plain dict with JSON fields decoded."""
+        if row is None:
+            return None
+        record = dict(row)
+        for field, empty in (
+            ("capabilities", []),
+            ("pending_mutation", {}),
+            ("evidence", {}),
+        ):
+            raw = record.get(field)
+            try:
+                record[field] = json.loads(raw) if raw else empty
+            except (TypeError, ValueError):
+                # A row written by an older/foreign writer must not break reads.
+                record[field] = {"unparsed": str(raw)}
+        return record
+
+    def get_session_checkpoint(
+        self,
+        *,
+        remote: str,
+        org: str,
+        repo: str,
+        session_id: str,
+        work_kind: str | None = None,
+        work_number: int | None = None,
+    ) -> dict[str, Any] | None:
+        """Return the current checkpoint for one session's work unit, or None."""
+        if work_kind and str(work_kind).strip():
+            kind_norm = dependency_graph.normalize_work_kind(work_kind)
+            number_norm = int(work_number) if work_number is not None else 0
+        else:
+            kind_norm = ""
+            number_norm = 0
+        with self._tx(immediate=False) as conn:
+            row = conn.execute(
+                """
+                SELECT * FROM session_checkpoints
+                WHERE remote = ? AND org = ? AND repo = ?
+                  AND session_id = ? AND work_kind = ? AND work_number = ?
+                """,
+                (remote, org, repo, session_id, kind_norm, number_norm),
+            ).fetchone()
+        return self._session_checkpoint_row(row)
+
+    def list_session_checkpoints(
+        self,
+        *,
+        remote: str | None = None,
+        org: str | None = None,
+        repo: str | None = None,
+        session_id: str | None = None,
+        work_kind: str | None = None,
+        work_number: int | None = None,
+        status: str | None = None,
+        limit: int = 500,
+    ) -> list[dict[str, Any]]:
+        """Return stored checkpoints, filtered. Newest updated first."""
+        clauses: list[str] = []
+        params: list[Any] = []
+        if remote:
+            clauses.append("remote = ?")
+            params.append(remote)
+        if org:
+            clauses.append("org = ?")
+            params.append(org)
+        if repo:
+            clauses.append("repo = ?")
+            params.append(repo)
+        if session_id:
+            clauses.append("session_id = ?")
+            params.append(session_id)
+        if work_kind:
+            clauses.append("work_kind = ?")
+            params.append(dependency_graph.normalize_work_kind(work_kind))
+        if work_number is not None:
+            clauses.append("work_number = ?")
+            params.append(int(work_number))
+        if status:
+            clauses.append("status = ?")
+            params.append(status)
+
+        sql = "SELECT * FROM session_checkpoints"
+        if clauses:
+            sql += " WHERE " + " AND ".join(clauses)
+        sql += " ORDER BY updated_at DESC LIMIT ?"
+        params.append(int(limit))
+
+        with self._tx(immediate=False) as conn:
+            rows = conn.execute(sql, params).fetchall()
+        return [
+            record
+            for record in (self._session_checkpoint_row(r) for r in rows)
+            if record
+        ]
+
+    @staticmethod
+    def reconcile_session_checkpoint(
+        checkpoint: dict[str, Any],
+        *,
+        live_head_sha: str | None = None,
+        live_lease_active: bool | None = None,
+        live_lease_id: str | None = None,
+    ) -> dict[str, Any]:
+        """Diagnose a stored checkpoint against live Git/Gitea/lease state.
+
+        Pure (no DB) and **never restores** — it returns a diagnosis and the
+        caller decides. A head that no longer matches, or a lease that is gone
+        or reassigned, marks the checkpoint stale so resumption reconciles
+        instead of blindly restoring stale ownership (#660 AC3).
+
+        A ``None`` live input means "unknown, not checked" and never flags a
+        mismatch on its own.
+        """
+        stored_head = str(checkpoint.get("head_sha") or "")
+        stored_lease = str(checkpoint.get("lease_id") or "")
+
+        head_mismatch = bool(
+            live_head_sha is not None
+            and stored_head
+            and stored_head != str(live_head_sha)
+        )
+        lease_mismatch = False
+        if stored_lease:
+            if live_lease_active is False:
+                lease_mismatch = True
+            elif live_lease_id is not None and str(live_lease_id) != stored_lease:
+                lease_mismatch = True
+
+        stale = head_mismatch or lease_mismatch
+        return {
+            "checkpoint_id": checkpoint.get("checkpoint_id"),
+            "session_id": checkpoint.get("session_id"),
+            "stale": stale,
+            "head_mismatch": head_mismatch,
+            "lease_mismatch": lease_mismatch,
+            "stored_head_sha": stored_head,
+            "live_head_sha": None if live_head_sha is None else str(live_head_sha),
+            "stored_lease_id": stored_lease,
+            "live_lease_id": None if live_lease_id is None else str(live_lease_id),
+            "reconcile_action": "reconcile_required" if stale else "safe_to_resume",
+        }

docs/architecture/control-plane-db-substrate.md

@@ -14,12 +14,13 @@
 
 | Capability | Notes |
 |------------|--------|
-| Schema | `sessions`, `work_items`, `leases`, `assignments`, `terminal_locks`, `events`, `incident_links` |
+| Schema | `sessions`, `work_items`, `leases`, `assignments`, `terminal_locks`, `events`, `incident_links`, `session_checkpoints` |
 | Atomic assign+lease | `ControlPlaneDB.assign_and_lease` — one `BEGIN IMMEDIATE` transaction |
 | Mutation gate | `require_valid_assignment` — live lease + allowed action + non-terminal work + non-stale head |
 | Heartbeat / release / expire | Lease lifecycle helpers |
 | Terminal-lock index | Routing signal for #600 (terminal path first) |
 | `incident_links` | Provider-neutral link model for #612 — **not** assignable work; scope keys NULL-safe |
+| `session_checkpoints` | Durable session resume state for #660 — redacted at write, reconciled (never restored) on boot |
 
 ## Hard rules (enforced in code)
 
@@ -92,6 +93,44 @@ Module: `incident_bridge.py` · MCP tools: `gitea_observability_*`
 - Provider tokens never appear in issue bodies, links, or tool results.
 - Allocator sees bridge work only after a Gitea issue exists.
 
+## Session checkpoints (#660)
+
+Module: `control_plane_db.py` · Table: `session_checkpoints` · Parent **#655** · Soft-depends **#659** · Vision **#652** · Roadmap **#653**
+
+Workflow state that lived only in MCP process memory or chat did not survive a restart, so session identity, stage, lease ownership, and the next valid action had to be reconstructed by hand. The `session_checkpoints` table makes that state durable.
+
+### Schema version
+
+Rows carry `checkpoint_schema_version` (currently **5**, tracking the module-level `SCHEMA_VERSION`), so a reader can tell which field set a record was written under. The row identity is
+`UNIQUE (remote, org, repo, session_id, work_kind, work_number)` — one current-state row per session per work unit, upserted rather than appended. A session-level checkpoint that is not bound to an issue or PR uses the `work_number = 0` sentinel with an empty `work_kind`.
+
+| Group | Columns |
+|-------|---------|
+| Identity | `session_id`, `provider_identity`, `role`, `remote`/`org`/`repo` |
+| Work unit | `work_kind` (`issue`/`pr`/empty), `work_number`, `worktree_path`, `branch`, `head_sha` |
+| Ownership | `capabilities` (JSON), `lease_id`, `assignment_id` |
+| Progress | `workflow_stage`, `last_completed_action`, `current_operation`, `pending_mutation` (JSON) |
+| Recovery | `evidence` (JSON), `blocker`, `next_valid_action`, `recovery_instructions` |
+| Bookkeeping | `checkpoint_schema_version`, `status`, `created_at`, `updated_at` |
+
+### API
+
+| Method | Purpose |
+|--------|---------|
+| `write_session_checkpoint` | Upsert the current-state row; redacts, and optionally enforces drain completeness |
+| `get_session_checkpoint` | Read one checkpoint by session + work unit |
+| `list_session_checkpoints` | List checkpoints by session or by work unit |
+| `reconcile_session_checkpoint` | Pure diagnosis of a stored checkpoint against live head/lease state |
+| `checkpoint_completeness` | Pure — the drain-required fields missing from a record |
+
+### Hard rules
+
+1. **Redaction at write.** Free-text columns are redaction-filtered and JSON columns are redacted recursively before storage, so a pasted token or credential URL cannot land in a checkpoint. Secrets are never stored (#660 AC4).
+2. **Reconcile, never blind-restore.** `reconcile_session_checkpoint` returns a diagnosis (`stale`, `head_mismatch`, `lease_mismatch`, `reconcile_action`) and the caller decides. A stored head that no longer matches live Git, or a lease that is gone or reassigned, marks the checkpoint stale (#660 AC3).
+3. **Unknown live state is not a mismatch.** A `None` live input means "not checked" and never flags staleness on its own.
+4. **Drain fails closed.** A write with `require_complete=True` refuses when any of `session_id`, `role`, `workflow_stage`, `next_valid_action`, `recovery_instructions` is missing or blank, so drain cannot complete on a checkpoint that could not resume.
+5. **No transcript storage.** Checkpoints hold resume state, not conversation history.
+
 ## Non-goals (intentionally deferred)
 
 - Full unsupervised watchdog auto-filing (prefer explicit reconcile first)

tests/test_control_plane_db.py

@@ -11,6 +11,7 @@ from datetime import timedelta
 
 from control_plane_db import (
     ControlPlaneDB,
+    ControlPlaneError,
     InvalidWorkKindError,
     LeaseRequiredError,
     WORK_KINDS,
@@ -820,5 +821,229 @@ class ControlPlaneDBTest(unittest.TestCase):
         self.assertEqual(n, 1)
 
 
+class SessionCheckpointTest(unittest.TestCase):
+    """Durable MCP session checkpoint schema, redaction, and reconcile (#660)."""
+
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self._tmp.name, "cp.sqlite3")
+        self.db = ControlPlaneDB(self.db_path)
+
+    def tearDown(self) -> None:
+        self._tmp.cleanup()
+
+    def _write(self, **overrides):
+        base = dict(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            session_id="prgs-author-1-abc",
+            role="author",
+            work_kind="issue",
+            work_number=660,
+            branch="feat/issue-660-session-checkpoint-schema",
+            head_sha="deadbeef",
+            lease_id="lease-1",
+            workflow_stage="implementing",
+            last_completed_action="wrote schema",
+            next_valid_action="write tests",
+            recovery_instructions="re-lock #660 then continue tests",
+        )
+        base.update(overrides)
+        return self.db.write_session_checkpoint(**base)
+
+    # AC1 — schema documented and versioned.
+    def test_table_exists_and_row_carries_schema_version(self) -> None:
+        import sqlite3
+
+        conn = sqlite3.connect(self.db_path)
+        try:
+            names = {
+                r[0]
+                for r in conn.execute(
+                    "SELECT name FROM sqlite_master WHERE type='table'"
+                ).fetchall()
+            }
+        finally:
+            conn.close()
+        self.assertIn("session_checkpoints", names)
+        record = self._write()
+        self.assertEqual(record["checkpoint_schema_version"], 5)
+
+    # AC2 — checkpoints written for multi-role session fixtures.
+    def test_multi_role_fixtures_each_get_a_row(self) -> None:
+        roles = [
+            ("prgs-author-1", "author", "issue", 660),
+            ("prgs-reviewer-2", "reviewer", "pr", 795),
+            ("prgs-merger-3", "merger", "pr", 862),
+            ("prgs-controller-4", "controller", "issue", 653),
+        ]
+        for session_id, role, kind, number in roles:
+            self._write(
+                session_id=session_id,
+                role=role,
+                work_kind=kind,
+                work_number=number,
+                lease_id=f"lease-{session_id}",
+            )
+        rows = self.db.list_session_checkpoints(remote="prgs")
+        self.assertEqual(len(rows), 4)
+        self.assertEqual(
+            {r["role"] for r in rows},
+            {"author", "reviewer", "merger", "controller"},
+        )
+
+    def test_upsert_is_current_state_and_audits_stage_change(self) -> None:
+        first = self._write(workflow_stage="implementing")
+        second = self._write(workflow_stage="testing")
+        self.assertEqual(first["checkpoint_id"], second["checkpoint_id"])
+        rows = self.db.list_session_checkpoints(
+            remote="prgs", session_id="prgs-author-1-abc"
+        )
+        self.assertEqual(len(rows), 1)
+        self.assertEqual(rows[0]["workflow_stage"], "testing")
+
+        import sqlite3
+
+        conn = sqlite3.connect(self.db_path)
+        try:
+            n = conn.execute(
+                "SELECT COUNT(*) FROM events "
+                "WHERE event_type = 'session_checkpoint_stage_change'"
+            ).fetchone()[0]
+        finally:
+            conn.close()
+        self.assertEqual(n, 1)
+
+    def test_get_and_roundtrip_json_fields(self) -> None:
+        self._write(
+            capabilities=["gitea.repo.commit", "gitea.pr.create"],
+            evidence={"tests": "4 passing"},
+            pending_mutation={"op": "commit_files", "files": ["control_plane_db.py"]},
+        )
+        got = self.db.get_session_checkpoint(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            session_id="prgs-author-1-abc",
+            work_kind="issue",
+            work_number=660,
+        )
+        self.assertIsNotNone(got)
+        self.assertEqual(got["capabilities"], ["gitea.repo.commit", "gitea.pr.create"])
+        self.assertEqual(got["evidence"], {"tests": "4 passing"})
+        self.assertEqual(got["pending_mutation"]["op"], "commit_files")
+
+    # AC4 — no secrets in stored records.
+    def test_secrets_are_redacted_before_storage(self) -> None:
+        self._write(
+            recovery_instructions=(
+                "resume with Authorization: Bearer sk-supersecrettoken then retry"
+            ),
+            evidence={"authorization": "Bearer sk-anothersecret"},
+            pending_mutation={"url": "https://user:[email protected]/repo.git"},
+        )
+        import sqlite3
+
+        conn = sqlite3.connect(self.db_path)
+        try:
+            row = conn.execute(
+                "SELECT recovery_instructions, evidence, pending_mutation "
+                "FROM session_checkpoints"
+            ).fetchone()
+        finally:
+            conn.close()
+        blob = " ".join(str(v) for v in row)
+        self.assertNotIn("sk-supersecrettoken", blob)
+        self.assertNotIn("sk-anothersecret", blob)
+        self.assertNotIn("password", blob)
+        self.assertIn("REDACTED", blob)
+
+    # AC3 — reconcile detects stale head / lease mismatch.
+    def test_reconcile_flags_stale_head(self) -> None:
+        record = self._write(head_sha="aaaa1111")
+        result = self.db.reconcile_session_checkpoint(
+            record, live_head_sha="bbbb2222", live_lease_active=True,
+            live_lease_id="lease-1",
+        )
+        self.assertTrue(result["stale"])
+        self.assertTrue(result["head_mismatch"])
+        self.assertFalse(result["lease_mismatch"])
+        self.assertEqual(result["reconcile_action"], "reconcile_required")
+
+    def test_reconcile_flags_dead_lease(self) -> None:
+        record = self._write(lease_id="lease-1", head_sha="aaaa1111")
+        result = self.db.reconcile_session_checkpoint(
+            record, live_head_sha="aaaa1111", live_lease_active=False,
+        )
+        self.assertTrue(result["stale"])
+        self.assertFalse(result["head_mismatch"])
+        self.assertTrue(result["lease_mismatch"])
+
+    def test_reconcile_reassigned_lease_is_stale(self) -> None:
+        record = self._write(lease_id="lease-1")
+        result = self.db.reconcile_session_checkpoint(
+            record, live_lease_active=True, live_lease_id="lease-999",
+        )
+        self.assertTrue(result["lease_mismatch"])
+
+    def test_reconcile_clean_state_is_safe_to_resume(self) -> None:
+        record = self._write(head_sha="aaaa1111", lease_id="lease-1")
+        result = self.db.reconcile_session_checkpoint(
+            record, live_head_sha="aaaa1111", live_lease_active=True,
+            live_lease_id="lease-1",
+        )
+        self.assertFalse(result["stale"])
+        self.assertEqual(result["reconcile_action"], "safe_to_resume")
+
+    def test_unknown_live_state_never_flags_mismatch(self) -> None:
+        record = self._write(head_sha="aaaa1111", lease_id="lease-1")
+        result = self.db.reconcile_session_checkpoint(record)
+        self.assertFalse(result["stale"])
+
+    # Drain gate — fail closed when a checkpoint is incomplete.
+    def test_drain_requires_complete_checkpoint(self) -> None:
+        with self.assertRaises(ControlPlaneError):
+            self.db.write_session_checkpoint(
+                remote="prgs",
+                org="Scaled-Tech-Consulting",
+                repo="Gitea-Tools",
+                session_id="prgs-author-1-abc",
+                role="author",
+                workflow_stage="draining",
+                # next_valid_action + recovery_instructions intentionally absent
+                require_complete=True,
+            )
+        # Nothing was written.
+        rows = self.db.list_session_checkpoints(remote="prgs")
+        self.assertEqual(rows, [])
+
+    def test_drain_write_succeeds_when_complete(self) -> None:
+        record = self._write(require_complete=True)
+        self.assertEqual(record["status"], "active")
+        self.assertEqual(self.db.checkpoint_completeness(record), [])
+
+    def test_missing_session_id_fails_closed(self) -> None:
+        with self.assertRaises(ControlPlaneError):
+            self.db.write_session_checkpoint(
+                remote="prgs", org="o", repo="r", session_id="",
+            )
+
+    def test_session_level_checkpoint_uses_sentinel_key(self) -> None:
+        # No work unit -> ('', 0) sentinel; a second session-level write upserts.
+        self.db.write_session_checkpoint(
+            remote="prgs", org="o", repo="r", session_id="s-sess",
+            workflow_stage="idle",
+        )
+        self.db.write_session_checkpoint(
+            remote="prgs", org="o", repo="r", session_id="s-sess",
+            workflow_stage="booting",
+        )
+        rows = self.db.list_session_checkpoints(remote="prgs", session_id="s-sess")
+        self.assertEqual(len(rows), 1)
+        self.assertEqual(rows[0]["work_kind"], "")
+        self.assertEqual(rows[0]["work_number"], 0)
+
+
 if __name__ == "__main__":
     unittest.main()