fix: fail closed on conflicting incident_links observation metadata

Legacy NULL-scope dedupe only compared Gitea targets, so duplicate rows
with the same provider key and issue but different fingerprint/status/
event_count/etc. collapsed to the lowest link_id and silently dropped
observation data. Migration now compares all meaningful observation
fields and refuses to discard conflicts (#619 RC3 / #613).

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

@@ -0,0 +1,53 @@
+# Control-plane DB substrate (#613)
+
+**Status:** Implemented (SQLite single-writer MVP)
+
+**ADR:** [`mcp-allocator-control-plane-observability-adr.md`](mcp-allocator-control-plane-observability-adr.md)
+
+**Module:** `control_plane_db.py`
+
+## Architecture statement
+
+> **DB coordinates, Gitea records, Sentry/GlitchTip observe; the bridge is the only path that turns observations into Gitea work.**
+
+## What this ships
+
+| Capability | Notes |
+|------------|--------|
+| Schema | `sessions`, `work_items`, `leases`, `assignments`, `terminal_locks`, `events`, `incident_links` |
+| 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 |
+
+## Hard rules (enforced in code)
+
+1. Assignable `work_items.kind` ∈ {`issue`, `pr`} only — **never** raw Sentry/GlitchTip incidents.
+2. Two concurrent sessions cannot both receive an active assignment on the same open work item (second gets `wait`).
+3. Merged/closed work items return `no_safe_work` at assign time, and `require_valid_assignment` fails closed if the work item becomes terminal later.
+4. Assignments pin `expected_head_sha`; mutations fail closed if the work item head drifts.
+5. SQLite path is the **single-writer MVP** (`GITEA_CONTROL_PLANE_DB`, default under `~/.cache/gitea-tools/control-plane/`). Multi-session multi-host production requires **Postgres** or a **single allocator daemon** (ADR §6).
+6. `incident_links` optional scope fields are stored as empty strings (never NULL) so UNIQUE is canonical across minimal upserts.
+7. Legacy `incident_links` migration collapses NULL-scope duplicates **only** when Gitea targets **and** all meaningful observation metadata agree (fingerprint, status, event_count, permalink, timestamps, linked PRs, etc.). Conflicting metadata fails closed — no silent discard.
+
+## Dependency chain
+
+```text
+#613 control-plane DB (this)  →  #600 allocator API  →  #612 incident bridge
+```
+
+- **#600** must call this substrate (not file locks / comment-only leases alone) for completion.
+- **#612** must write `incident_links` here and create **Gitea issues**; the allocator assigns those issues, not raw incidents.
+
+## Non-goals (intentionally deferred)
+
+- Full `gitea_allocate_next_work` routing policy (REQUEST_CHANGES → author, etc.) — **#600**
+- Sentry/GlitchTip provider adapters and auto-watchdog — **#612**
+- Gitea comment/label mirror writers (may be added by allocator tooling later)
+
+## Tests
+
+```bash
+python3 -m pytest tests/test_control_plane_db.py -q
+```

gitea_mcp_server.py

@@ -3095,26 +3095,22 @@ def check_review_decision_gate(
         )
 
     prior = list(lock.get("live_mutations") or [])
-    ready_head = lock.get("ready_expected_head_sha")
+    if prior and not lock.get("correction_authorized"):
-    if stale_review_decision_lock.prior_live_mutations_block_boundary(
+        reasons.append(
-        lock, pr_number=pr_number, expected_head_sha=ready_head
+            "live review mutation already recorded in this run; only one live "
+            "review mutation is allowed unless "
+            "gitea_authorize_review_correction was invoked (fail closed)"
+        )
+    elif (
+        action in _TERMINAL_REVIEW_ACTIONS
+        and any(m.get("action") in _TERMINAL_REVIEW_ACTIONS for m in prior)
+        and not lock.get("correction_authorized")
     ):
-        if prior and not lock.get("correction_authorized"):
+        reasons.append(
-            reasons.append(
+            "terminal review decision already submitted on this PR in this "
-                "live review mutation already recorded for this PR head in this "
+            "run; blocked unless an operator-approved correction was "
-                "run; only one live review mutation is allowed per head unless "
+            "authorized (fail closed)"
-                "gitea_authorize_review_correction was invoked (fail closed, #620)"
+        )
-            )
-        elif (
-            action in _TERMINAL_REVIEW_ACTIONS
-            and any(m.get("action") in _TERMINAL_REVIEW_ACTIONS for m in prior)
-            and not lock.get("correction_authorized")
-        ):
-            reasons.append(
-                "terminal review decision already submitted for this PR head in "
-                "this run; blocked unless an operator-approved correction was "
-                "authorized (fail closed, #620)"
-            )
 
     return reasons
 
@@ -3122,18 +3118,12 @@ def check_review_decision_gate(
 def record_live_review_mutation(pr_number: int, action: str, review_id: int | None = None):
     lock = _load_review_decision_lock() or {}
     mutations = list(lock.get("live_mutations") or [])
-    head_sha = stale_review_decision_lock.normalize_head_sha(
+    mutations.append({
-        lock.get("ready_expected_head_sha")
-    )
-    entry = {
         "pr_number": pr_number,
         "action": action,
         "review_id": review_id,
         "review_state": action,
-    }
+    })
-    if head_sha:
-        entry["head_sha"] = head_sha
-    mutations.append(entry)
     lock["live_mutations"] = mutations
     if lock.get("correction_authorized"):
         lock["correction_authorized"] = False
@@ -3141,28 +3131,17 @@ def record_live_review_mutation(pr_number: int, action: str, review_id: int | No
     _save_review_decision_lock(lock)
 
 
-def terminal_review_hard_stop_reasons(
+def terminal_review_hard_stop_reasons(pr_number: int, operation: str) -> list[str]:
-    pr_number: int,
+    """Session hard-stop after a terminal live review mutation (#332).
-    operation: str,
-    expected_head_sha: str | None = None,
-) -> list[str]:
-    """Session hard-stop after a terminal live review mutation (#332 / #620).
 
-    After a terminal verdict is consumed for a given PR **head**, the only
+    After a terminal verdict is consumed in this run, the only permitted
-    permitted continuation is the merge sequence for that same approved PR
+    continuation is the merge sequence for the same PR that was approved.
-    (merge does not require a new head). A REQUEST_CHANGES (or an approval of
+    A REQUEST_CHANGES (or an approval of a different PR) blocks every
-    a different PR, or the same head) blocks further review/mark-ready/merge
+    further review/mark-ready/merge mutation. An operator-approved
-    mutations for that boundary.
+    correction (#211) re-opens the review path only — never a cross-PR
+    merge.
 
-    #620: when *expected_head_sha* differs from the last terminal's reviewed
+    *operation* is one of 'merge', 'mark_ready', or 'review'.
-    head on the **same open PR**, mark_ready / review / resume may proceed so
-    a fresh formal decision can be recorded for the new head. Historical
-    mutations are preserved.
-
-    An operator-approved correction (#211) re-opens the review path only —
-    never a cross-PR merge.
-
-    *operation* is one of 'merge', 'mark_ready', 'review', or 'resume'.
     Returns [] when the operation may proceed.
     """
     lock = _load_review_decision_lock()
@@ -3181,19 +3160,8 @@ def terminal_review_hard_stop_reasons(
         and last.get("pr_number") == pr_number
     ):
         return []
-    if operation in ("mark_ready", "review", "resume") and lock.get(
+    if operation in ("mark_ready", "review") and lock.get("correction_authorized"):
-        "correction_authorized"
-    ):
         return []
-    # #620: same PR, different reviewed head → fresh decision boundary.
-    if stale_review_decision_lock.terminal_boundary_allows_fresh_decision(
-        lock,
-        pr_number=pr_number,
-        expected_head_sha=expected_head_sha,
-        operation=operation,
-    ):
-        return []
-    locked_head = stale_review_decision_lock.mutation_head_sha(last, lock)
     if last.get("action") == "approve":
         guidance = (
             f"only the merge sequence for approved PR "
@@ -3201,22 +3169,15 @@ def terminal_review_hard_stop_reasons(
         )
     else:
         guidance = "the session must stop and produce a final report"
-    head_note = (
-        f" at head {locked_head[:12]}…"
-        if locked_head
-        else ""
-    )
     recovery = (
         "if that PR is already merged/closed, use "
         "gitea_cleanup_stale_review_decision_lock (apply=true) after live-state "
-        "proof (#594); if the PR is still open but the head moved, mark/submit "
+        "proof (#594); never delete session-state files by hand"
-        "with the new expected_head_sha (#620); never delete session-state "
-        "files by hand"
     )
     return [
         "terminal review mutation already consumed in this run "
-        f"({last.get('action')} on PR #{last.get('pr_number')}{head_note}); "
+        f"({last.get('action')} on PR #{last.get('pr_number')}); {guidance} "
-        f"{guidance} (fail closed, #332); {recovery}"
+        f"(fail closed, #332); {recovery}"
     ]
 
 
@@ -3839,11 +3800,18 @@ def gitea_mark_final_review_decision(
             "reasons": workflow_blockers + (
                 review_workflow_load.recovery_handoff_without_replay()),
         }
-    hard_stop = terminal_review_hard_stop_reasons(
+    hard_stop = terminal_review_hard_stop_reasons(pr_number, "mark_ready")
-        pr_number, "mark_ready", expected_head_sha=expected_head_sha
-    )
     if hard_stop:
         return {"marked_ready": False, "reasons": hard_stop}
+    if lock.get("live_mutations") and not lock.get("correction_authorized"):
+        return {
+            "marked_ready": False,
+            "reasons": [
+                "cannot mark final decision after a live review mutation was "
+                "already recorded in this run unless an operator-approved "
+                "correction was authorized"
+            ],
+        }
     if action not in _REVIEW_ACTIONS:
         return {
             "marked_ready": False,
@@ -3860,17 +3828,6 @@ def gitea_mark_final_review_decision(
                 "decision (fail closed, #399)"
             ],
         }
-    if stale_review_decision_lock.prior_live_mutations_block_boundary(
-        lock, pr_number=pr_number, expected_head_sha=expected_head_sha
-    ):
-        return {
-            "marked_ready": False,
-            "reasons": [
-                "cannot mark final decision after a live review mutation was "
-                "already recorded for this PR head unless an operator-approved "
-                "correction was authorized (fail closed, #620)"
-            ],
-        }
     elig = gitea_check_pr_eligibility(
         pr_number=pr_number,
         action="review",
@@ -3923,8 +3880,6 @@ def gitea_mark_final_review_decision(
                     "#332)"
                 ],
             }
-    # Preserve historical terminal head boundaries before advancing ready_* (#620).
-    stale_review_decision_lock.backfill_terminal_heads_from_ready(lock)
     lock["final_review_decision_ready"] = True
     lock["ready_pr_number"] = pr_number
     lock["ready_action"] = action
@@ -3942,7 +3897,6 @@ def gitea_mark_final_review_decision(
         "org": org,
         "repo": repo,
         "final_review_decision_ready": True,
-        "head_scoped": True,
         "reasons": [],
     }
 
@@ -4109,12 +4063,6 @@ def gitea_cleanup_stale_review_decision_lock(
         "cleanup_allowed": assessment.get("cleanup_allowed"),
         "last_terminal_pr": assessment.get("last_terminal_pr"),
         "last_terminal_action": assessment.get("last_terminal_action"),
-        "locked_head_sha": assessment.get("locked_head_sha"),
-        "current_pr_head_sha": assessment.get("current_pr_head_sha"),
-        "stale_by_head": assessment.get("stale_by_head"),
-        "fresh_review_on_current_head_allowed": assessment.get(
-            "fresh_review_on_current_head_allowed"
-        ),
         "pr_state": assessment.get("pr_state"),
         "pr_merged": assessment.get("pr_merged"),
         "pr_merged_or_closed": assessment.get("pr_merged_or_closed"),

stale_review_decision_lock.py

@@ -1,4 +1,4 @@
-"""Stale / moot #332 review-decision lock detection and cleanup policy (#594/#620).
+"""Stale / moot #332 review-decision lock detection and cleanup policy (#594).
 
 #332 correctly hard-stops a reviewer session after a terminal live review
 mutation. After #559 those locks are durable on disk, so they can outlive the
@@ -6,12 +6,6 @@ work they protect: when the referenced PR is already merged or closed, no
 same-PR merge sequence remains, yet the durable ledger still blocks unrelated
 new terminal reviews.
 
-#620 scopes the terminal boundary by **reviewed head SHA**: a prior
-REQUEST_CHANGES (or APPROVE) on head A must not block a fresh formal decision
-on head B of the **same open PR**. Same-head duplicates remain fail-closed.
-Open-PR lock *cleanup* remains forbidden unless the PR is truly merged/closed
-(#594); new-head re-review is allowed without deleting the durable lock.
-
 This module is pure policy (no Gitea I/O). The MCP tool
 ``gitea_cleanup_stale_review_decision_lock`` fetches live PR state, calls these
 helpers, and only then clears durable state when cleanup is allowed.
@@ -29,45 +23,6 @@ from typing import Any
 TERMINAL_REVIEW_ACTIONS = frozenset({"approve", "request_changes"})
 
 
-def normalize_head_sha(value: Any) -> str | None:
-    """Normalize a git SHA for equality comparison; None if empty/invalid."""
-    if value is None:
-        return None
-    text = str(value).strip().lower()
-    if not text:
-        return None
-    return text
-
-
-def heads_equal(a: Any, b: Any) -> bool:
-    na = normalize_head_sha(a)
-    nb = normalize_head_sha(b)
-    if not na or not nb:
-        return False
-    return na == nb
-
-
-def mutation_head_sha(mutation: dict | None, lock: dict | None = None) -> str | None:
-    """Head SHA that bounds a live mutation (#620).
-
-    Prefers fields recorded on the mutation itself. For *legacy* mutations
-    that predate head-scoping, falls back to the lock's
-    ``ready_expected_head_sha`` only when that ready binding is for the same
-    PR number (the frozen durable PR #619-style ledger).
-    """
-    if not isinstance(mutation, dict):
-        return None
-    for key in ("head_sha", "expected_head_sha", "reviewed_head_sha"):
-        head = normalize_head_sha(mutation.get(key))
-        if head:
-            return head
-    if not isinstance(lock, dict):
-        return None
-    if lock.get("ready_pr_number") != mutation.get("pr_number"):
-        return None
-    return normalize_head_sha(lock.get("ready_expected_head_sha"))
-
-
 def last_terminal_mutation(lock: dict | None) -> dict | None:
     """Return the last terminal live mutation on *lock*, or None."""
     if not lock:
@@ -80,125 +35,18 @@ def last_terminal_mutation(lock: dict | None) -> dict | None:
     return terminals[-1] if terminals else None
 
 
-def prior_live_mutations_block_boundary(
-    lock: dict | None,
-    *,
-    pr_number: int,
-    expected_head_sha: str | None,
-) -> bool:
-    """True when prior live mutations block a new decision for PR+head (#620).
-
-    Historical terminals on a **different head of the same PR** do not block.
-    Any prior mutation on another PR, the same head, or without a comparable
-    head SHA fails closed (blocks).
-
-    Head resolution uses ``mutation_head_sha(m, lock)`` so pre-#620 ledgers
-    that only stored ``ready_expected_head_sha`` still compare correctly
-    (PR #619-style).
-    """
-    if not lock:
-        return False
-    if lock.get("correction_authorized"):
-        return False
-    prior = list(lock.get("live_mutations") or [])
-    if not prior:
-        return False
-    target = normalize_head_sha(expected_head_sha)
-    for m in prior:
-        if not isinstance(m, dict):
-            return True
-        m_pr = m.get("pr_number")
-        m_head = mutation_head_sha(m, lock)
-        if (
-            m_pr == pr_number
-            and target
-            and m_head
-            and not heads_equal(m_head, target)
-        ):
-            # Same open PR, different reviewed head — historical boundary only.
-            continue
-        return True
-    return False
-
-
-def terminal_boundary_allows_fresh_decision(
-    lock: dict | None,
-    *,
-    pr_number: int,
-    expected_head_sha: str | None,
-    operation: str,
-) -> bool:
-    """Whether #332 hard-stop should yield for a new PR+head boundary (#620).
-
-    Only for ``mark_ready`` / ``review`` / ``resume`` on the **same PR** when
-    the requested head differs from the last terminal's head. Merge is never
-    reopened by head change (approved head merge path is separate).
-    """
-    if operation not in ("mark_ready", "review", "resume"):
-        return False
-    if not lock:
-        return False
-    if lock.get("correction_authorized"):
-        return True
-    last = last_terminal_mutation(lock)
-    if last is None:
-        return True
-    if last.get("pr_number") != pr_number:
-        return False
-    locked_head = mutation_head_sha(last, lock)
-    target = normalize_head_sha(expected_head_sha)
-    if not locked_head or not target:
-        return False
-    return not heads_equal(locked_head, target)
-
-
-def backfill_terminal_heads_from_ready(lock: dict) -> dict:
-    """Stamp head_sha onto legacy terminal mutations before overwriting ready_*.
-
-    Called when marking a fresh decision on a new head so historical rows keep
-    their original boundary after ``ready_expected_head_sha`` advances (#620).
-    """
-    if not isinstance(lock, dict):
-        return lock
-    ready_pr = lock.get("ready_pr_number")
-    ready_head = normalize_head_sha(lock.get("ready_expected_head_sha"))
-    if ready_pr is None or not ready_head:
-        return lock
-    mutations = list(lock.get("live_mutations") or [])
-    changed = False
-    for m in mutations:
-        if not isinstance(m, dict):
-            continue
-        if m.get("action") not in TERMINAL_REVIEW_ACTIONS:
-            continue
-        if m.get("pr_number") != ready_pr:
-            continue
-        if mutation_head_sha(m):
-            continue
-        m["head_sha"] = ready_head
-        changed = True
-    if changed:
-        lock["live_mutations"] = mutations
-    return lock
-
-
 def classify_pr_live_state(pr_live: dict | None) -> dict[str, Any]:
     """Normalize a Gitea PR payload into merge/closed flags."""
     pr = pr_live or {}
     merged = bool(pr.get("merged") or pr.get("merged_at"))
     state = (pr.get("state") or "").strip().lower() or None
     closed = state == "closed"
-    head = pr.get("head") if isinstance(pr.get("head"), dict) else {}
-    head_sha = normalize_head_sha(
-        pr.get("head_sha") or pr.get("head_commit_sha") or head.get("sha")
-    )
     return {
         "pr_state": state,
         "pr_merged": merged,
         "pr_merged_or_closed": merged or closed,
         "merge_commit_sha": pr.get("merge_commit_sha")
         or pr.get("merged_commit_sha"),
-        "current_pr_head_sha": head_sha,
     }
 
 
@@ -208,7 +56,6 @@ def lock_summary(lock: dict | None) -> dict[str, Any] | None:
         return None
     last = last_terminal_mutation(lock)
     mutations = list(lock.get("live_mutations") or [])
-    last_head = mutation_head_sha(last, lock) if last else None
     return {
         "task": lock.get("task"),
         "remote": lock.get("remote"),
@@ -220,21 +67,16 @@ def lock_summary(lock: dict | None) -> dict[str, Any] | None:
         "session_profile": lock.get("session_profile"),
         "ready_pr_number": lock.get("ready_pr_number"),
         "ready_action": lock.get("ready_action"),
-        "ready_expected_head_sha": normalize_head_sha(
-            lock.get("ready_expected_head_sha")
-        ),
         "live_mutations_count": len(mutations),
         "last_terminal": (
             {
                 "pr_number": last.get("pr_number"),
                 "action": last.get("action"),
                 "review_id": last.get("review_id"),
-                "head_sha": last_head,
             }
             if last
             else None
         ),
-        "locked_head_sha": last_head,
         "correction_authorized": bool(lock.get("correction_authorized")),
         "updated_at": lock.get("updated_at") or lock.get("recorded_at"),
     }
@@ -246,16 +88,13 @@ def assess_stale_review_decision_lock(
     pr_live: dict | None = None,
     pr_lookup_error: str | None = None,
     active_profile_identity: str | None = None,
-    current_pr_head_sha: str | None = None,
 ) -> dict[str, Any]:
-    """Assess whether a durable review-decision lock is stale/moot (#594/#620).
+    """Assess whether a durable review-decision lock is stale/moot (#594).
 
     *pr_live* must be the live Gitea payload for the **last terminal
     mutation's PR**. When no lock / no terminal mutation exists, cleanup is
     not applicable. When the PR is still open (or lookup fails), cleanup is
-    forbidden (#594). Open PR + different live head reports
+    forbidden and #332 hard-stop remains in force.
-    ``stale_by_head`` / ``fresh_review_on_current_head_allowed`` (#620)
-    without enabling cleanup.
     """
     summary = lock_summary(lock)
     result: dict[str, Any] = {
@@ -263,10 +102,6 @@ def assess_stale_review_decision_lock(
         "lock_summary": summary,
         "last_terminal_pr": None,
         "last_terminal_action": None,
-        "locked_head_sha": None,
-        "current_pr_head_sha": normalize_head_sha(current_pr_head_sha),
-        "stale_by_head": False,
-        "fresh_review_on_current_head_allowed": False,
         "is_moot": False,
         "cleanup_allowed": False,
         "profile_match": True,
@@ -306,10 +141,8 @@ def assess_stale_review_decision_lock(
 
     pr_number = last.get("pr_number")
     action = last.get("action")
-    locked_head = mutation_head_sha(last, lock)
     result["last_terminal_pr"] = pr_number
     result["last_terminal_action"] = action
-    result["locked_head_sha"] = locked_head
 
     if pr_number is None:
         result["reasons"].append(
@@ -332,46 +165,25 @@ def assess_stale_review_decision_lock(
         return result
 
     live = classify_pr_live_state(pr_live)
-    current_head = normalize_head_sha(
-        current_pr_head_sha or live.get("current_pr_head_sha")
-    )
     result.update(
         {
             "pr_state": live["pr_state"],
             "pr_merged": live["pr_merged"],
             "pr_merged_or_closed": live["pr_merged_or_closed"],
             "merge_commit_sha": live["merge_commit_sha"],
-            "current_pr_head_sha": current_head,
         }
     )
 
     if not live["pr_merged_or_closed"]:
-        stale_by_head = bool(
+        result["reasons"].append(
-            locked_head and current_head and not heads_equal(locked_head, current_head)
+            f"terminal review mutation on open PR #{pr_number} is still active "
+            f"({action}); #332 hard-stop remains — cleanup forbidden (#594)"
         )
-        result["stale_by_head"] = stale_by_head
-        # Fresh formal decision is allowed on the new head without cleanup (#620).
-        result["fresh_review_on_current_head_allowed"] = stale_by_head
-        if stale_by_head:
-            result["reasons"].append(
-                f"terminal {action} on PR #{pr_number} is bound to head "
-                f"{locked_head[:12]}…; live head is {current_head[:12]}… — "
-                "fresh formal review on current head is allowed without "
-                "cleanup (fail closed for cleanup, #620); #332 same-head "
-                "duplicate protection remains"
-            )
-        else:
-            result["reasons"].append(
-                f"terminal review mutation on open PR #{pr_number} is still active "
-                f"({action}); #332 hard-stop remains — cleanup forbidden (#594)"
-            )
         return result
 
     # Merged or closed: lock is moot. Same-PR merge continuation is impossible.
     result["is_moot"] = True
     result["cleanup_allowed"] = True
-    result["stale_by_head"] = False
-    result["fresh_review_on_current_head_allowed"] = False
     result["reasons"].append(
         f"terminal mutation ({action} on PR #{pr_number}) is moot: PR is "
         f"{'merged' if live['pr_merged'] else 'closed'}; canonical cleanup allowed (#594)"

tests/test_control_plane_db.py

@@ -0,0 +1,824 @@
+"""Tests for control-plane DB substrate (#613)."""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import threading
+import unittest
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from datetime import timedelta
+
+from control_plane_db import (
+    ControlPlaneDB,
+    InvalidWorkKindError,
+    LeaseRequiredError,
+    WORK_KINDS,
+    _ts,
+    _utc_now,
+)
+
+
+class ControlPlaneDBTest(unittest.TestCase):
+    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 test_schema_and_architecture_meta(self) -> None:
+        import sqlite3
+
+        conn = sqlite3.connect(self.db_path)
+        try:
+            rows = dict(conn.execute("SELECT key, value FROM schema_meta").fetchall())
+        finally:
+            conn.close()
+        self.assertEqual(rows["schema_version"], "2")
+        self.assertIn("DB coordinates", rows["architecture"])
+        self.assertIn("bridge", rows["architecture"].lower())
+
+    def test_rejects_raw_incident_as_work_kind(self) -> None:
+        with self.assertRaises(InvalidWorkKindError):
+            self.db.upsert_work_item(
+                remote="prgs",
+                org="org",
+                repo="repo",
+                kind="sentry_incident",
+                number=1,
+            )
+        with self.assertRaises(InvalidWorkKindError):
+            self.db.assign_and_lease(
+                session_id="s1",
+                role="author",
+                remote="prgs",
+                org="org",
+                repo="repo",
+                kind="glitchtip_incident",
+                number=9,
+            )
+        self.assertEqual(WORK_KINDS, frozenset({"issue", "pr"}))
+
+    def test_atomic_assign_and_lease_fields(self) -> None:
+        self.db.upsert_session(session_id="s-a", role="author", profile="prgs-author")
+        result = self.db.assign_and_lease(
+            session_id="s-a",
+            role="author",
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            kind="issue",
+            number=613,
+            expected_head_sha="abc123",
+            allowed_actions=("implement", "comment"),
+            forbidden_actions=("approve", "merge"),
+        )
+        self.assertEqual(result.outcome, "assigned")
+        self.assertIsNotNone(result.assignment_id)
+        self.assertIsNotNone(result.lease_id)
+        self.assertEqual(result.role, "author")
+        self.assertEqual(result.work_kind, "issue")
+        self.assertEqual(result.work_number, 613)
+        self.assertEqual(result.expected_head_sha, "abc123")
+        self.assertIn("implement", result.allowed_actions)
+        self.assertIn("merge", result.forbidden_actions)
+        self.assertIsNotNone(result.expires_at)
+
+    def test_second_session_waits_on_foreign_lease(self) -> None:
+        self.db.upsert_session(session_id="s1", role="author")
+        self.db.upsert_session(session_id="s2", role="author")
+        first = self.db.assign_and_lease(
+            session_id="s1",
+            role="author",
+            remote="prgs",
+            org="o",
+            repo="r",
+            kind="pr",
+            number=100,
+            expected_head_sha="deadbeef",
+        )
+        self.assertEqual(first.outcome, "assigned")
+        second = self.db.assign_and_lease(
+            session_id="s2",
+            role="author",
+            remote="prgs",
+            org="o",
+            repo="r",
+            kind="pr",
+            number=100,
+            expected_head_sha="deadbeef",
+        )
+        self.assertEqual(second.outcome, "wait")
+        self.assertEqual(second.owner_session_id, "s1")
+
+    def test_owner_resume_refreshes_lease(self) -> None:
+        self.db.upsert_session(session_id="s1", role="reviewer")
+        a = self.db.assign_and_lease(
+            session_id="s1",
+            role="reviewer",
+            remote="prgs",
+            org="o",
+            repo="r",
+            kind="pr",
+            number=50,
+            expected_head_sha="head-50",
+        )
+        b = self.db.assign_and_lease(
+            session_id="s1",
+            role="reviewer",
+            remote="prgs",
+            org="o",
+            repo="r",
+            kind="pr",
+            number=50,
+            expected_head_sha="head-50",
+        )
+        self.assertEqual(b.outcome, "assigned")
+        self.assertEqual(b.lease_id, a.lease_id)
+        self.assertIn("owner-resume", b.reason)
+
+    def test_require_valid_assignment_gates_mutations(self) -> None:
+        self.db.upsert_session(session_id="s1", role="author")
+        self.db.assign_and_lease(
+            session_id="s1",
+            role="author",
+            remote="prgs",
+            org="o",
+            repo="r",
+            kind="issue",
+            number=7,
+            allowed_actions=("implement",),
+            forbidden_actions=("merge",),
+        )
+        proof = self.db.require_valid_assignment(
+            session_id="s1",
+            remote="prgs",
+            org="o",
+            repo="r",
+            kind="issue",
+            number=7,
+            action="implement",
+        )
+        self.assertEqual(proof["session_id"], "s1")
+        with self.assertRaises(LeaseRequiredError):
+            self.db.require_valid_assignment(
+                session_id="s1",
+                remote="prgs",
+                org="o",
+                repo="r",
+                kind="issue",
+                number=7,
+                action="merge",
+            )
+        with self.assertRaises(LeaseRequiredError):
+            self.db.require_valid_assignment(
+                session_id="s-other",
+                remote="prgs",
+                org="o",
+                repo="r",
+                kind="issue",
+                number=7,
+                action="implement",
+            )
+
+    def test_expired_lease_allows_reassign(self) -> None:
+        self.db.upsert_session(session_id="s1", role="author")
+        self.db.upsert_session(session_id="s2", role="author")
+        past = _utc_now() - timedelta(hours=1)
+        # Create lease already expired by using negative TTL edge via direct assign then expire
+        assigned = self.db.assign_and_lease(
+            session_id="s1",
+            role="author",
+            remote="prgs",
+            org="o",
+            repo="r",
+            kind="issue",
+            number=3,
+            lease_ttl_seconds=1,
+        )
+        self.assertEqual(assigned.outcome, "assigned")
+        # Force expiry in DB
+        import sqlite3
+
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "UPDATE leases SET expires_at = ? WHERE lease_id = ?",
+                (_ts(past), assigned.lease_id),
+            )
+            conn.commit()
+        finally:
+            conn.close()
+        n = self.db.expire_stale_leases()
+        self.assertGreaterEqual(n, 1)
+        second = self.db.assign_and_lease(
+            session_id="s2",
+            role="author",
+            remote="prgs",
+            org="o",
+            repo="r",
+            kind="issue",
+            number=3,
+        )
+        self.assertEqual(second.outcome, "assigned")
+        self.assertEqual(second.session_id, "s2")
+
+    def test_merged_work_never_assigned(self) -> None:
+        self.db.upsert_session(session_id="s1", role="merger")
+        self.db.upsert_work_item(
+            remote="prgs",
+            org="o",
+            repo="r",
+            kind="pr",
+            number=99,
+            state="merged",
+        )
+        result = self.db.assign_and_lease(
+            session_id="s1",
+            role="merger",
+            remote="prgs",
+            org="o",
+            repo="r",
+            kind="pr",
+            number=99,
+            expected_head_sha="merged-head",
+        )
+        self.assertEqual(result.outcome, "no_safe_work")
+
+    def test_four_concurrent_sessions_unique_assignments(self) -> None:
+        """Four concurrent assigners on four different issues — all succeed uniquely.
+
+        Also two concurrent assigners on the *same* issue: at most one assigned.
+        """
+        for i in range(4):
+            self.db.upsert_session(session_id=f"sess-{i}", role="author")
+
+        def claim_unique(i: int):
+            return self.db.assign_and_lease(
+                session_id=f"sess-{i}",
+                role="author",
+                remote="prgs",
+                org="o",
+                repo="r",
+                kind="issue",
+                number=1000 + i,
+            )
+
+        with ThreadPoolExecutor(max_workers=4) as pool:
+            results = [f.result() for f in as_completed([pool.submit(claim_unique, i) for i in range(4)])]
+        self.assertEqual({r.outcome for r in results}, {"assigned"})
+        numbers = sorted(r.work_number for r in results)
+        self.assertEqual(numbers, [1000, 1001, 1002, 1003])
+
+        # Contention on one item
+        self.db.upsert_session(session_id="c1", role="author")
+        self.db.upsert_session(session_id="c2", role="author")
+        self.db.upsert_session(session_id="c3", role="author")
+        self.db.upsert_session(session_id="c4", role="author")
+        barrier = threading.Barrier(4)
+        outcomes: list[str] = []
+        lock = threading.Lock()
+
+        def contend(sid: str) -> None:
+            barrier.wait()
+            res = self.db.assign_and_lease(
+                session_id=sid,
+                role="author",
+                remote="prgs",
+                org="o",
+                repo="r",
+                kind="issue",
+                number=7777,
+            )
+            with lock:
+                outcomes.append(res.outcome)
+
+        threads = [threading.Thread(target=contend, args=(f"c{i}",)) for i in range(1, 5)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+        self.assertEqual(outcomes.count("assigned"), 1)
+        self.assertEqual(outcomes.count("wait"), 3)
+
+    def test_terminal_lock_index(self) -> None:
+        self.db.set_terminal_lock(
+            remote="prgs",
+            org="o",
+            repo="r",
+            terminal_pr=332,
+            review_id="rev-1",
+            decision="approve",
+        )
+        row = self.db.get_active_terminal_lock(remote="prgs", org="o", repo="r")
+        self.assertIsNotNone(row)
+        assert row is not None
+        self.assertEqual(row["terminal_pr"], 332)
+        self.assertEqual(row["status"], "active")
+
+    def test_incident_links_not_work_items(self) -> None:
+        link = self.db.upsert_incident_link(
+            provider="sentry",
+            provider_base_url="https://sentry.prgs.cc",
+            provider_org="prgs",
+            provider_project="gitea-tools-mcp",
+            provider_issue_id="12345",
+            gitea_org="Scaled-Tech-Consulting",
+            gitea_repo="Gitea-Tools",
+            gitea_issue_number=9001,
+            fingerprint="fp-1",
+        )
+        self.assertEqual(link["gitea_issue_number"], 9001)
+        found = self.db.get_incident_link_for_gitea_issue(
+            gitea_org="Scaled-Tech-Consulting",
+            gitea_repo="Gitea-Tools",
+            gitea_issue_number=9001,
+        )
+        self.assertIsNotNone(found)
+        # Linking does not create a work_item of incident kind
+        with self.assertRaises(InvalidWorkKindError):
+            self.db.upsert_work_item(
+                remote="prgs",
+                org="Scaled-Tech-Consulting",
+                repo="Gitea-Tools",
+                kind="sentry",
+                number=12345,
+            )
+
+    def test_heartbeat_and_release(self) -> None:
+        self.db.upsert_session(session_id="s1", role="author")
+        a = self.db.assign_and_lease(
+            session_id="s1",
+            role="author",
+            remote="prgs",
+            org="o",
+            repo="r",
+            kind="issue",
+            number=1,
+        )
+        hb = self.db.heartbeat_lease(a.lease_id, session_id="s1")
+        self.assertEqual(hb["lease_id"], a.lease_id)
+        self.db.release_lease(a.lease_id, session_id="s1")
+        # After release another session can claim
+        self.db.upsert_session(session_id="s2", role="author")
+        b = self.db.assign_and_lease(
+            session_id="s2",
+            role="author",
+            remote="prgs",
+            org="o",
+            repo="r",
+            kind="issue",
+            number=1,
+        )
+        self.assertEqual(b.outcome, "assigned")
+        self.assertEqual(b.session_id, "s2")
+
+    def test_require_valid_assignment_rejects_stale_head(self) -> None:
+        """Assignment must not authorize mutations after work-item head drifts."""
+        self.db.upsert_session(session_id="s1", role="author")
+        self.db.assign_and_lease(
+            session_id="s1",
+            role="author",
+            remote="prgs",
+            org="o",
+            repo="r",
+            kind="pr",
+            number=42,
+            expected_head_sha="head-v1",
+            allowed_actions=("implement",),
+        )
+        # Head drifts after assignment
+        self.db.upsert_work_item(
+            remote="prgs",
+            org="o",
+            repo="r",
+            kind="pr",
+            number=42,
+            current_head_sha="head-v2",
+        )
+        with self.assertRaises(LeaseRequiredError) as ctx:
+            self.db.require_valid_assignment(
+                session_id="s1",
+                remote="prgs",
+                org="o",
+                repo="r",
+                kind="pr",
+                number=42,
+                action="implement",
+            )
+        self.assertIn("stale head", str(ctx.exception).lower())
+
+    def test_require_valid_assignment_rejects_terminal_state(self) -> None:
+        """Assignment must not authorize mutations after work item is merged/closed."""
+        self.db.upsert_session(session_id="s1", role="author")
+        self.db.assign_and_lease(
+            session_id="s1",
+            role="author",
+            remote="prgs",
+            org="o",
+            repo="r",
+            kind="pr",
+            number=55,
+            expected_head_sha="abc",
+            allowed_actions=("implement",),
+        )
+        self.db.upsert_work_item(
+            remote="prgs",
+            org="o",
+            repo="r",
+            kind="pr",
+            number=55,
+            state="merged",
+            current_head_sha="abc",
+        )
+        with self.assertRaises(LeaseRequiredError) as ctx:
+            self.db.require_valid_assignment(
+                session_id="s1",
+                remote="prgs",
+                org="o",
+                repo="r",
+                kind="pr",
+                number=55,
+                action="implement",
+            )
+        self.assertIn("terminal", str(ctx.exception).lower())
+
+        self.db.upsert_work_item(
+            remote="prgs",
+            org="o",
+            repo="r",
+            kind="issue",
+            number=56,
+            state="open",
+        )
+        self.db.assign_and_lease(
+            session_id="s1",
+            role="author",
+            remote="prgs",
+            org="o",
+            repo="r",
+            kind="issue",
+            number=56,
+            allowed_actions=("implement",),
+        )
+        self.db.upsert_work_item(
+            remote="prgs",
+            org="o",
+            repo="r",
+            kind="issue",
+            number=56,
+            state="closed",
+        )
+        with self.assertRaises(LeaseRequiredError):
+            self.db.require_valid_assignment(
+                session_id="s1",
+                remote="prgs",
+                org="o",
+                repo="r",
+                kind="issue",
+                number=56,
+                action="implement",
+            )
+
+    def test_pr_assignment_requires_expected_head_sha(self) -> None:
+        """PR assign and mutation must fail closed without a head pin."""
+        self.db.upsert_session(session_id="s1", role="author")
+        with self.assertRaises(LeaseRequiredError) as ctx:
+            self.db.assign_and_lease(
+                session_id="s1",
+                role="author",
+                remote="prgs",
+                org="o",
+                repo="r",
+                kind="pr",
+                number=88,
+                expected_head_sha=None,
+                allowed_actions=("implement",),
+            )
+        self.assertIn("expected_head_sha", str(ctx.exception))
+
+        # Legacy path: force an unpinned PR assignment into the DB, then
+        # populate head and prove mutation is still rejected.
+        import sqlite3
+
+        self.db.upsert_work_item(
+            remote="prgs",
+            org="o",
+            repo="r",
+            kind="pr",
+            number=89,
+            current_head_sha=None,
+        )
+        conn = sqlite3.connect(self.db_path)
+        try:
+            wid = conn.execute(
+                "SELECT work_item_id FROM work_items WHERE kind='pr' AND number=89"
+            ).fetchone()[0]
+            conn.execute(
+                """
+                INSERT INTO sessions(session_id, role, started_at, last_heartbeat_at, status)
+                VALUES ('legacy', 'author', '2020-01-01T00:00:00Z', '2020-01-01T00:00:00Z', 'active')
+                """
+            )
+            conn.execute(
+                """
+                INSERT INTO leases(
+                    lease_id, work_item_id, session_id, role, phase,
+                    expires_at, heartbeat_at, status
+                ) VALUES (
+                    'lease-legacy', ?, 'legacy', 'author', 'claimed',
+                    '2099-01-01T00:00:00Z', '2020-01-01T00:00:00Z', 'active'
+                )
+                """,
+                (wid,),
+            )
+            conn.execute(
+                """
+                INSERT INTO assignments(
+                    assignment_id, work_item_id, session_id, lease_id,
+                    allowed_actions, forbidden_actions, expected_head_sha,
+                    role, status, created_at
+                ) VALUES (
+                    'asn-legacy', ?, 'legacy', 'lease-legacy',
+                    '["implement"]', '["merge"]', NULL,
+                    'author', 'active', '2020-01-01T00:00:00Z'
+                )
+                """,
+                (wid,),
+            )
+            conn.commit()
+        finally:
+            conn.close()
+        self.db.upsert_work_item(
+            remote="prgs",
+            org="o",
+            repo="r",
+            kind="pr",
+            number=89,
+            current_head_sha="populated-head",
+        )
+        with self.assertRaises(LeaseRequiredError) as ctx2:
+            self.db.require_valid_assignment(
+                session_id="legacy",
+                remote="prgs",
+                org="o",
+                repo="r",
+                kind="pr",
+                number=89,
+                action="implement",
+            )
+        self.assertIn("expected_head_sha pin", str(ctx2.exception))
+
+    def test_migrate_duplicate_null_scope_incident_links(self) -> None:
+        """Legacy NULL-scope duplicates must migrate without UNIQUE crash."""
+        import sqlite3
+
+        from control_plane_db import ControlPlaneDB, ControlPlaneError
+
+        # Build a v1-like table with nullable scope columns and insert dups
+        # that collapse under normalization, then open ControlPlaneDB on it.
+        path = os.path.join(self._tmp.name, "legacy_dups.sqlite3")
+        conn = sqlite3.connect(path)
+        try:
+            conn.executescript(
+                """
+                CREATE TABLE schema_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
+                CREATE TABLE incident_links (
+                    link_id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    provider TEXT NOT NULL,
+                    provider_base_url TEXT,
+                    provider_org TEXT,
+                    provider_project TEXT,
+                    provider_issue_id TEXT NOT NULL,
+                    provider_short_id TEXT,
+                    provider_permalink TEXT,
+                    fingerprint TEXT,
+                    gitea_org TEXT NOT NULL,
+                    gitea_repo TEXT NOT NULL,
+                    gitea_issue_number INTEGER NOT NULL,
+                    linked_pr_numbers TEXT,
+                    first_seen TEXT,
+                    last_seen TEXT,
+                    event_count INTEGER,
+                    status TEXT NOT NULL DEFAULT 'open',
+                    release_resolved_at TEXT,
+                    last_sync_at TEXT,
+                    UNIQUE (
+                        provider, provider_base_url, provider_org,
+                        provider_project, provider_issue_id
+                    )
+                );
+                INSERT INTO incident_links(
+                    provider, provider_base_url, provider_org, provider_project,
+                    provider_issue_id, gitea_org, gitea_repo, gitea_issue_number, status
+                ) VALUES
+                    ('sentry', NULL, NULL, NULL, 'dup-1', 'org', 'repo', 10, 'open'),
+                    ('sentry', NULL, NULL, NULL, 'dup-1', 'org', 'repo', 10, 'open');
+                """
+            )
+            # SQLite allows two NULL-scope rows with same provider/issue under UNIQUE.
+            n = conn.execute("SELECT COUNT(*) FROM incident_links").fetchone()[0]
+            self.assertEqual(n, 2)
+            conn.commit()
+        finally:
+            conn.close()
+
+        db = ControlPlaneDB(path)
+        conn2 = sqlite3.connect(path)
+        try:
+            n2 = conn2.execute("SELECT COUNT(*) FROM incident_links").fetchone()[0]
+            rows = conn2.execute(
+                "SELECT provider_base_url, provider_org, provider_project, gitea_issue_number "
+                "FROM incident_links"
+            ).fetchall()
+        finally:
+            conn2.close()
+        self.assertEqual(n2, 1)
+        self.assertEqual(rows[0][0], "")
+        self.assertEqual(rows[0][1], "")
+        self.assertEqual(rows[0][2], "")
+        self.assertEqual(rows[0][3], 10)
+        # Touch to silence unused import in type checkers if needed
+        self.assertTrue(issubclass(ControlPlaneError, Exception))
+        del db
+
+    def test_migrate_conflicting_duplicate_incident_links_fails_closed(self) -> None:
+        """Conflicting Gitea targets for the same provider key must fail closed."""
+        import sqlite3
+        from control_plane_db import ControlPlaneDB, ControlPlaneError
+
+        path = os.path.join(self._tmp.name, "legacy_conflict.sqlite3")
+        conn = sqlite3.connect(path)
+        try:
+            conn.executescript(
+                """
+                CREATE TABLE incident_links (
+                    link_id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    provider TEXT NOT NULL,
+                    provider_base_url TEXT,
+                    provider_org TEXT,
+                    provider_project TEXT,
+                    provider_issue_id TEXT NOT NULL,
+                    gitea_org TEXT NOT NULL,
+                    gitea_repo TEXT NOT NULL,
+                    gitea_issue_number INTEGER NOT NULL,
+                    status TEXT NOT NULL DEFAULT 'open',
+                    UNIQUE (
+                        provider, provider_base_url, provider_org,
+                        provider_project, provider_issue_id
+                    )
+                );
+                INSERT INTO incident_links(
+                    provider, provider_base_url, provider_org, provider_project,
+                    provider_issue_id, gitea_org, gitea_repo, gitea_issue_number
+                ) VALUES
+                    ('sentry', NULL, NULL, NULL, 'dup-c', 'org', 'repo', 1),
+                    ('sentry', NULL, NULL, NULL, 'dup-c', 'org', 'repo', 2);
+                """
+            )
+            conn.commit()
+        finally:
+            conn.close()
+
+        with self.assertRaises(ControlPlaneError) as ctx:
+            ControlPlaneDB(path)
+        self.assertIn("conflicting", str(ctx.exception).lower())
+
+    def test_migrate_conflicting_observation_metadata_fails_closed(self) -> None:
+        """Same provider key + same Gitea target but differing obs metadata must fail closed.
+
+        Regression for silent data loss: migration used to keep lowest link_id and
+        delete peers after comparing only Gitea targets (#619 RC3).
+        """
+        import sqlite3
+        from control_plane_db import ControlPlaneDB, ControlPlaneError
+
+        path = os.path.join(self._tmp.name, "legacy_meta_conflict.sqlite3")
+        conn = sqlite3.connect(path)
+        try:
+            conn.executescript(
+                """
+                CREATE TABLE schema_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
+                CREATE TABLE incident_links (
+                    link_id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    provider TEXT NOT NULL,
+                    provider_base_url TEXT,
+                    provider_org TEXT,
+                    provider_project TEXT,
+                    provider_issue_id TEXT NOT NULL,
+                    provider_short_id TEXT,
+                    provider_permalink TEXT,
+                    fingerprint TEXT,
+                    gitea_org TEXT NOT NULL,
+                    gitea_repo TEXT NOT NULL,
+                    gitea_issue_number INTEGER NOT NULL,
+                    linked_pr_numbers TEXT,
+                    first_seen TEXT,
+                    last_seen TEXT,
+                    event_count INTEGER,
+                    status TEXT NOT NULL DEFAULT 'open',
+                    release_resolved_at TEXT,
+                    last_sync_at TEXT,
+                    UNIQUE (
+                        provider, provider_base_url, provider_org,
+                        provider_project, provider_issue_id
+                    )
+                );
+                INSERT INTO incident_links(
+                    provider, provider_base_url, provider_org, provider_project,
+                    provider_issue_id, provider_permalink, fingerprint,
+                    gitea_org, gitea_repo, gitea_issue_number,
+                    event_count, status, first_seen, last_seen
+                ) VALUES
+                    (
+                        'sentry', NULL, NULL, NULL, 'dup-meta',
+                        'https://sentry.example/issues/1', 'fingerprint-A',
+                        'org', 'repo', 42,
+                        1, 'open', '2026-01-01T00:00:00Z', '2026-01-01T01:00:00Z'
+                    ),
+                    (
+                        'sentry', NULL, NULL, NULL, 'dup-meta',
+                        'https://sentry.example/issues/1', 'fingerprint-B',
+                        'org', 'repo', 42,
+                        99, 'resolved', '2026-01-01T00:00:00Z', '2026-01-02T00:00:00Z'
+                    );
+                """
+            )
+            n = conn.execute("SELECT COUNT(*) FROM incident_links").fetchone()[0]
+            self.assertEqual(n, 2)
+            conn.commit()
+        finally:
+            conn.close()
+
+        with self.assertRaises(ControlPlaneError) as ctx:
+            ControlPlaneDB(path)
+        msg = str(ctx.exception).lower()
+        self.assertIn("conflicting", msg)
+        self.assertIn("observation metadata", msg)
+
+        # Rows must still be present — migration must not delete before failing.
+        conn2 = sqlite3.connect(path)
+        try:
+            remaining = conn2.execute("SELECT COUNT(*) FROM incident_links").fetchone()[0]
+            fps = {
+                r[0]
+                for r in conn2.execute(
+                    "SELECT fingerprint FROM incident_links ORDER BY link_id"
+                ).fetchall()
+            }
+        finally:
+            conn2.close()
+        self.assertEqual(remaining, 2)
+        self.assertEqual(fps, {"fingerprint-A", "fingerprint-B"})
+
+    def test_incident_links_minimal_upsert_is_canonical(self) -> None:
+        """Repeated minimal upserts must update one row (NULL-safe uniqueness)."""
+        a = self.db.upsert_incident_link(
+            provider="sentry",
+            provider_issue_id="inc-1",
+            gitea_org="org",
+            gitea_repo="repo",
+            gitea_issue_number=1,
+        )
+        b = self.db.upsert_incident_link(
+            provider="sentry",
+            provider_issue_id="inc-1",
+            gitea_org="org",
+            gitea_repo="repo",
+            gitea_issue_number=2,
+            # omit optional scope fields again
+        )
+        c = self.db.upsert_incident_link(
+            provider="sentry",
+            provider_issue_id="inc-1",
+            gitea_org="org",
+            gitea_repo="repo",
+            gitea_issue_number=3,
+            provider_base_url=None,
+            provider_org="",
+            provider_project="  ",
+        )
+        self.assertEqual(a["link_id"], b["link_id"])
+        self.assertEqual(b["link_id"], c["link_id"])
+        self.assertEqual(c["gitea_issue_number"], 3)
+        self.assertEqual(c["provider_base_url"], "")
+        self.assertEqual(c["provider_org"], "")
+        self.assertEqual(c["provider_project"], "")
+
+        import sqlite3
+
+        conn = sqlite3.connect(self.db_path)
+        try:
+            n = conn.execute(
+                "SELECT COUNT(*) FROM incident_links WHERE provider_issue_id = ?",
+                ("inc-1",),
+            ).fetchone()[0]
+        finally:
+            conn.close()
+        self.assertEqual(n, 1)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_head_scoped_review_decision_lock.py

@@ -1,397 +0,0 @@
-"""Head-scoped #332 review-decision locks (#620).
-
-Proves:
-  * REQUEST_CHANGES on head A does not block mark_ready/submit on head B
-  * APPROVE on head B is allowed after RC on head A (same open PR)
-  * same-head duplicate terminal remains blocked
-  * open-PR cleanup remains forbidden; assessment reports stale_by_head
-  * historical mutations keep their head_sha (preserved ledger)
-  * PR #619-style: old RC at 036b78e…, current head 2429d9d…, approve allowed
-"""
-
-from __future__ import annotations
-
-import os
-import unittest
-from unittest.mock import patch
-
-import mcp_server
-import stale_review_decision_lock as srdl
-
-HEAD_A = "036b78e31ea5d036452b3a52e0e086b01e0c763f"
-HEAD_B = "2429d9d2e826e519eb8eb4ade45987c00838aefb"
-HEAD_C = "c" * 40
-
-
-def _lock(mutations=None, correction=False, ready_pr=None, ready_head=None, ready_action=None):
-    return {
-        "task": "review_pr",
-        "remote": "prgs",
-        "org": "Scaled-Tech-Consulting",
-        "repo": "Gitea-Tools",
-        "session_pid": os.getpid(),
-        "session_profile": "prgs-reviewer",
-        "session_profile_lock": "prgs-reviewer",
-        "profile_identity": "prgs-reviewer",
-        "final_review_decision_ready": False,
-        "ready_pr_number": ready_pr,
-        "ready_action": ready_action,
-        "ready_expected_head_sha": ready_head,
-        "ready_remote": "prgs" if ready_pr else None,
-        "ready_org": "Scaled-Tech-Consulting" if ready_pr else None,
-        "ready_repo": "Gitea-Tools" if ready_pr else None,
-        "live_mutations": list(mutations or []),
-        "correction_authorized": correction,
-        "correction_reason": None,
-    }
-
-
-RC_619_LEGACY = {
-    "pr_number": 619,
-    "action": "request_changes",
-    "review_id": 410,
-    "review_state": "request_changes",
-    # no head_sha — pre-#620 durable ledger shape
-}
-RC_619_HEAD_A = {
-    "pr_number": 619,
-    "action": "request_changes",
-    "review_id": 410,
-    "review_state": "request_changes",
-    "head_sha": HEAD_A,
-}
-APPROVE_619_HEAD_B = {
-    "pr_number": 619,
-    "action": "approve",
-    "review_id": 411,
-    "review_state": "approve",
-    "head_sha": HEAD_B,
-}
-
-
-def _seed(mutations=None, **kwargs):
-    import review_workflow_load
-
-    review_workflow_load.record_review_workflow_load(mcp_server.PROJECT_ROOT)
-    mcp_server._save_review_decision_lock(_lock(mutations, **kwargs))
-    mcp_server.gitea_load_review_workflow()
-
-
-def _open_pr(pr_number=619, head=HEAD_B):
-    return {
-        "number": pr_number,
-        "state": "open",
-        "merged": False,
-        "merged_at": None,
-        "head": {"sha": head},
-    }
-
-
-def _no_lease():
-    return {"block": False, "reasons": [], "mutation_allowed": True}
-
-
-def _feedback(blocking=False, stale=False, success=True):
-    return {
-        "success": success,
-        "has_blocking_change_requests": blocking,
-        "review_feedback_stale": stale,
-        "current_head_sha": HEAD_B,
-    }
-
-
-class TestPureHeadScopeHelpers(unittest.TestCase):
-    def test_mutation_head_prefers_recorded_sha(self):
-        m = {"pr_number": 1, "head_sha": HEAD_A}
-        self.assertEqual(srdl.mutation_head_sha(m), HEAD_A)
-
-    def test_legacy_mutation_uses_ready_expected_for_same_pr(self):
-        lock = _lock(
-            [RC_619_LEGACY],
-            ready_pr=619,
-            ready_head=HEAD_A,
-            ready_action="request_changes",
-        )
-        self.assertEqual(srdl.mutation_head_sha(RC_619_LEGACY, lock), HEAD_A)
-
-    def test_legacy_mutation_ignores_ready_for_other_pr(self):
-        lock = _lock([RC_619_LEGACY], ready_pr=1, ready_head=HEAD_A)
-        self.assertIsNone(srdl.mutation_head_sha(RC_619_LEGACY, lock))
-
-    def test_prior_blocks_same_head_not_other_head(self):
-        lock = _lock([RC_619_HEAD_A])
-        self.assertTrue(
-            srdl.prior_live_mutations_block_boundary(
-                lock, pr_number=619, expected_head_sha=HEAD_A
-            )
-        )
-        self.assertFalse(
-            srdl.prior_live_mutations_block_boundary(
-                lock, pr_number=619, expected_head_sha=HEAD_B
-            )
-        )
-
-    def test_backfill_stamps_legacy_terminals(self):
-        lock = _lock(
-            [dict(RC_619_LEGACY)],
-            ready_pr=619,
-            ready_head=HEAD_A,
-            ready_action="request_changes",
-        )
-        srdl.backfill_terminal_heads_from_ready(lock)
-        self.assertEqual(lock["live_mutations"][0]["head_sha"], HEAD_A)
-
-
-class TestHardStopHeadScope(unittest.TestCase):
-    def tearDown(self):
-        mcp_server._save_review_decision_lock(None)
-        mcp_server.review_workflow_load.clear_review_workflow_load()
-
-    def test_rc_head_a_allows_mark_ready_head_b(self):
-        _seed(
-            [RC_619_HEAD_A],
-            ready_pr=619,
-            ready_head=HEAD_A,
-            ready_action="request_changes",
-        )
-        self.assertEqual(
-            mcp_server.terminal_review_hard_stop_reasons(
-                619, "mark_ready", expected_head_sha=HEAD_B
-            ),
-            [],
-        )
-
-    def test_rc_head_a_blocks_mark_ready_same_head(self):
-        _seed(
-            [RC_619_HEAD_A],
-            ready_pr=619,
-            ready_head=HEAD_A,
-            ready_action="request_changes",
-        )
-        reasons = mcp_server.terminal_review_hard_stop_reasons(
-            619, "mark_ready", expected_head_sha=HEAD_A
-        )
-        self.assertTrue(reasons)
-        self.assertIn("#332", reasons[0])
-
-    def test_rc_head_a_blocks_other_pr(self):
-        _seed([RC_619_HEAD_A], ready_pr=619, ready_head=HEAD_A)
-        reasons = mcp_server.terminal_review_hard_stop_reasons(
-            700, "mark_ready", expected_head_sha=HEAD_B
-        )
-        self.assertTrue(reasons)
-
-    def test_legacy_619_ledger_allows_new_head(self):
-        """PR #619-style durable lock without mutation head_sha."""
-        _seed(
-            [RC_619_LEGACY],
-            ready_pr=619,
-            ready_head=HEAD_A,
-            ready_action="request_changes",
-        )
-        self.assertEqual(
-            mcp_server.terminal_review_hard_stop_reasons(
-                619, "mark_ready", expected_head_sha=HEAD_B
-            ),
-            [],
-        )
-
-
-class TestMarkFinalAndRecord(unittest.TestCase):
-    def tearDown(self):
-        mcp_server._save_review_decision_lock(None)
-        mcp_server.review_workflow_load.clear_review_workflow_load()
-
-    def _mark(self, pr, action, head, feedback=None):
-        with patch("mcp_server._list_pr_lease_comments", return_value=[]), patch(
-            "mcp_server._pr_work_lease_reviewer_block", return_value=_no_lease()
-        ), patch.object(
-            mcp_server,
-            "gitea_get_pr_review_feedback",
-            return_value=feedback or _feedback(blocking=False, stale=True),
-        ), patch.object(
-            mcp_server,
-            "gitea_check_pr_eligibility",
-            return_value={"eligible": True, "head_sha": head},
-        ):
-            return mcp_server.gitea_mark_final_review_decision(
-                pr_number=pr,
-                action=action,
-                expected_head_sha=head,
-                remote="prgs",
-                org="Scaled-Tech-Consulting",
-                repo="Gitea-Tools",
-            )
-
-    def test_rc_then_rc_on_new_head(self):
-        _seed(
-            [RC_619_HEAD_A],
-            ready_pr=619,
-            ready_head=HEAD_A,
-            ready_action="request_changes",
-        )
-        # Prior RC at head A is unresolved but head moved → feedback stale.
-        res = self._mark(
-            619,
-            "request_changes",
-            HEAD_B,
-            feedback=_feedback(blocking=True, stale=True),
-        )
-        self.assertTrue(res.get("marked_ready"), res)
-        lock = mcp_server._load_review_decision_lock()
-        # Historical head A preserved on mutation after backfill.
-        heads = {
-            m.get("head_sha")
-            for m in lock["live_mutations"]
-            if m.get("action") == "request_changes"
-        }
-        self.assertIn(HEAD_A, heads)
-        self.assertEqual(lock["ready_expected_head_sha"], HEAD_B)
-
-    def test_rc_then_approve_on_new_head(self):
-        _seed(
-            [RC_619_HEAD_A],
-            ready_pr=619,
-            ready_head=HEAD_A,
-            ready_action="request_changes",
-        )
-        res = self._mark(619, "approve", HEAD_B)
-        self.assertTrue(res.get("marked_ready"), res)
-        self.assertTrue(res.get("head_scoped"))
-
-    def test_same_head_rc_still_blocked_by_hard_stop(self):
-        _seed(
-            [RC_619_HEAD_A],
-            ready_pr=619,
-            ready_head=HEAD_A,
-            ready_action="request_changes",
-        )
-        res = self._mark(619, "approve", HEAD_A)
-        self.assertFalse(res.get("marked_ready"))
-        self.assertTrue(any("#332" in r for r in res.get("reasons") or []))
-
-    def test_pr619_style_legacy_lock_approve_on_current_head(self):
-        _seed(
-            [RC_619_LEGACY],
-            ready_pr=619,
-            ready_head=HEAD_A,
-            ready_action="request_changes",
-        )
-        res = self._mark(619, "approve", HEAD_B)
-        self.assertTrue(res.get("marked_ready"), res)
-        lock = mcp_server._load_review_decision_lock()
-        # Backfill should have stamped HEAD_A onto the legacy mutation.
-        self.assertEqual(lock["live_mutations"][0].get("head_sha"), HEAD_A)
-        self.assertEqual(lock["ready_expected_head_sha"], HEAD_B)
-
-    def test_record_live_mutation_stores_head(self):
-        _seed(ready_pr=619, ready_head=HEAD_B, ready_action="approve")
-        lock = mcp_server._load_review_decision_lock()
-        lock["final_review_decision_ready"] = True
-        lock["ready_pr_number"] = 619
-        lock["ready_expected_head_sha"] = HEAD_B
-        mcp_server._save_review_decision_lock(lock)
-        mcp_server.record_live_review_mutation(619, "approve", review_id=99)
-        stored = mcp_server._load_review_decision_lock()["live_mutations"][-1]
-        self.assertEqual(stored["head_sha"], HEAD_B)
-        self.assertEqual(stored["pr_number"], 619)
-
-    def test_submit_gate_allows_new_head_after_prior_terminal(self):
-        """check_review_decision_gate must not block different-head submit."""
-        mutations = [RC_619_HEAD_A]
-        _seed(
-            mutations,
-            ready_pr=619,
-            ready_head=HEAD_B,
-            ready_action="approve",
-        )
-        lock = mcp_server._load_review_decision_lock()
-        lock["final_review_decision_ready"] = True
-        lock["ready_pr_number"] = 619
-        lock["ready_action"] = "approve"
-        lock["ready_expected_head_sha"] = HEAD_B
-        lock["ready_remote"] = "prgs"
-        lock["ready_org"] = "Scaled-Tech-Consulting"
-        lock["ready_repo"] = "Gitea-Tools"
-        mcp_server._save_review_decision_lock(lock)
-        reasons = mcp_server.check_review_decision_gate(
-            619,
-            "approve",
-            final_review_decision_ready=True,
-            remote="prgs",
-            org="Scaled-Tech-Consulting",
-            repo="Gitea-Tools",
-        )
-        self.assertEqual(reasons, [], reasons)
-
-    def test_submit_gate_blocks_same_head_duplicate(self):
-        mutations = [RC_619_HEAD_A]
-        _seed(
-            mutations,
-            ready_pr=619,
-            ready_head=HEAD_A,
-            ready_action="request_changes",
-        )
-        lock = mcp_server._load_review_decision_lock()
-        lock["final_review_decision_ready"] = True
-        lock["ready_pr_number"] = 619
-        lock["ready_action"] = "request_changes"
-        lock["ready_expected_head_sha"] = HEAD_A
-        lock["ready_remote"] = "prgs"
-        lock["ready_org"] = "Scaled-Tech-Consulting"
-        lock["ready_repo"] = "Gitea-Tools"
-        mcp_server._save_review_decision_lock(lock)
-        reasons = mcp_server.check_review_decision_gate(
-            619,
-            "request_changes",
-            final_review_decision_ready=True,
-            remote="prgs",
-            org="Scaled-Tech-Consulting",
-            repo="Gitea-Tools",
-        )
-        self.assertTrue(reasons)
-
-
-class TestAssessmentOpenPrHead(unittest.TestCase):
-    def test_open_pr_stale_by_head_no_cleanup(self):
-        lock = _lock(
-            [RC_619_HEAD_A],
-            ready_pr=619,
-            ready_head=HEAD_A,
-            ready_action="request_changes",
-        )
-        a = srdl.assess_stale_review_decision_lock(
-            lock, pr_live=_open_pr(619, HEAD_B)
-        )
-        self.assertTrue(a["has_lock"])
-        self.assertFalse(a["is_moot"])
-        self.assertFalse(a["cleanup_allowed"])
-        self.assertTrue(a["stale_by_head"])
-        self.assertTrue(a["fresh_review_on_current_head_allowed"])
-        self.assertEqual(a["locked_head_sha"], HEAD_A)
-        self.assertEqual(a["current_pr_head_sha"], HEAD_B)
-
-    def test_open_pr_same_head_no_fresh(self):
-        lock = _lock([RC_619_HEAD_A], ready_pr=619, ready_head=HEAD_A)
-        a = srdl.assess_stale_review_decision_lock(
-            lock, pr_live=_open_pr(619, HEAD_A)
-        )
-        self.assertFalse(a["stale_by_head"])
-        self.assertFalse(a["fresh_review_on_current_head_allowed"])
-        self.assertFalse(a["cleanup_allowed"])
-
-    def test_historical_mutation_preserved_in_summary(self):
-        lock = _lock(
-            [RC_619_HEAD_A, APPROVE_619_HEAD_B],
-            ready_pr=619,
-            ready_head=HEAD_B,
-            ready_action="approve",
-        )
-        summary = srdl.lock_summary(lock)
-        self.assertEqual(summary["live_mutations_count"], 2)
-        self.assertEqual(summary["last_terminal"]["head_sha"], HEAD_B)
-        self.assertEqual(summary["locked_head_sha"], HEAD_B)
-
-
-if __name__ == "__main__":
-    unittest.main()