fix: fail closed on stale/terminal assignments; canonicalize incident_links

Address REQUEST_CHANGES on PR #619:

- require_valid_assignment rejects terminal work states and head drift
- incident_links scope keys normalize NULL/blank to '' for UNIQUE
- remove trailing whitespace in control-plane-db-substrate.md

control_plane_db.py

@@ -29,11 +29,14 @@ from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 from typing import Any, Iterator, Sequence
 
-SCHEMA_VERSION = 1
+SCHEMA_VERSION = 2
 
 # Assignable work kinds only — raw monitoring incidents are never work items.
 WORK_KINDS = frozenset({"issue", "pr"})
 
+# Work item states that permanently revoke assignment/lease authority.
+TERMINAL_WORK_STATES = frozenset({"merged", "closed"})
+
 DEFAULT_LEASE_TTL_SECONDS = 4 * 3600
 
 # Environment: path for SQLite MVP (single-writer).
@@ -120,12 +123,13 @@ CREATE TABLE IF NOT EXISTS events (
 );
 
 -- Provider-neutral incident ↔ Gitea link model (ADR §8–9). Not assignable work.
+-- Optional scope fields are stored as '' (never NULL) so UNIQUE is NULL-safe.
 CREATE TABLE IF NOT EXISTS incident_links (
     link_id INTEGER PRIMARY KEY AUTOINCREMENT,
     provider TEXT NOT NULL,
-    provider_base_url TEXT,
+    provider_base_url TEXT NOT NULL DEFAULT '',
-    provider_org TEXT,
+    provider_org TEXT NOT NULL DEFAULT '',
-    provider_project TEXT,
+    provider_project TEXT NOT NULL DEFAULT '',
     provider_issue_id TEXT NOT NULL,
     provider_short_id TEXT,
     provider_permalink TEXT,
@@ -172,6 +176,17 @@ def _parse_ts(value: str | None) -> datetime | None:
         return None
 
 
+def _norm_scope(value: str | None) -> str:
+    """Normalize optional incident-link scope keys for NULL-safe uniqueness.
+
+    Empty / whitespace / None all become '' so SQLite UNIQUE treats them as one
+    canonical key component (SQLite treats multiple NULLs as distinct).
+    """
+    if value is None:
+        return ""
+    return str(value).strip()
+
+
 def default_db_path() -> str:
     raw = (os.environ.get(DB_PATH_ENV) or DEFAULT_DB_PATH).strip()
     return raw or DEFAULT_DB_PATH
@@ -285,6 +300,7 @@ class ControlPlaneDB:
             conn = self._connect()
             try:
                 conn.executescript(_SCHEMA_SQL)
+                self._migrate_incident_links_null_scope(conn)
                 conn.execute(
                     "INSERT OR REPLACE INTO schema_meta(key, value) VALUES (?, ?)",
                     ("schema_version", str(SCHEMA_VERSION)),
@@ -301,6 +317,35 @@ class ControlPlaneDB:
             finally:
                 conn.close()
 
+    def _migrate_incident_links_null_scope(self, conn: sqlite3.Connection) -> None:
+        """Coerce NULL scope columns to '' and collapse any duplicate keys (#619 RC)."""
+        cols = {
+            row[1]
+            for row in conn.execute("PRAGMA table_info(incident_links)").fetchall()
+        }
+        if not cols:
+            return
+        for col in ("provider_base_url", "provider_org", "provider_project"):
+            if col in cols:
+                conn.execute(
+                    f"UPDATE incident_links SET {col} = '' WHERE {col} IS NULL"
+                )
+        # Drop duplicates keeping lowest link_id (canonical row).
+        conn.execute(
+            """
+            DELETE FROM incident_links
+            WHERE link_id NOT IN (
+                SELECT MIN(link_id)
+                FROM incident_links
+                GROUP BY provider,
+                         IFNULL(provider_base_url, ''),
+                         IFNULL(provider_org, ''),
+                         IFNULL(provider_project, ''),
+                         provider_issue_id
+            )
+            """
+        )
+
     # ── sessions ──────────────────────────────────────────────────────────
 
     def upsert_session(
@@ -730,7 +775,12 @@ class ControlPlaneDB:
         number: int,
         action: str,
     ) -> dict[str, Any]:
-        """Gate a mutation: require active assignment+lease for this session/work."""
+        """Gate a mutation: require active assignment+lease for this session/work.
+
+        Fail-closed when the work item is terminal (merged/closed) or when the
+        assignment's expected_head_sha no longer matches the work item head
+        (stale-head after assignment time).
+        """
         kind_norm = (kind or "").strip().lower()
         if kind_norm not in WORK_KINDS:
             raise InvalidWorkKindError(f"invalid kind '{kind}'")
@@ -746,6 +796,12 @@ class ControlPlaneDB:
                 raise LeaseRequiredError(
                     f"no work_item for {kind_norm}#{number}; assign first"
                 )
+            work_state = (work["state"] or "").strip().lower()
+            if work_state in TERMINAL_WORK_STATES:
+                raise LeaseRequiredError(
+                    f"work item {kind_norm}#{number} is terminal state "
+                    f"'{work_state}'; assignment no longer authorizes mutations"
+                )
             self._expire_stale_leases_conn(conn, work_item_id=int(work["work_item_id"]))
             asn = conn.execute(
                 """
@@ -769,6 +825,14 @@ class ControlPlaneDB:
             exp = _parse_ts(asn["expires_at"])
             if exp and exp <= _utc_now():
                 raise LeaseRequiredError(f"lease {asn['lease_id']} expired")
+            # Stale-head: pinned expected_head_sha must still match live work item.
+            assigned_head = (asn["expected_head_sha"] or "").strip()
+            current_head = (work["current_head_sha"] or "").strip()
+            if assigned_head and assigned_head != current_head:
+                raise LeaseRequiredError(
+                    f"assignment {asn['assignment_id']} stale head: expected "
+                    f"{assigned_head!r} but work item head is {current_head!r}"
+                )
             allowed = json.loads(asn["allowed_actions"])
             forbidden = json.loads(asn["forbidden_actions"])
             if action in forbidden:
@@ -867,6 +931,10 @@ class ControlPlaneDB:
         status: str = "open",
     ) -> dict[str, Any]:
         now_s = _ts()
+        # Canonical key: never store NULL in UNIQUE scope columns.
+        base_url = _norm_scope(provider_base_url)
+        p_org = _norm_scope(provider_org)
+        p_project = _norm_scope(provider_project)
         with self._tx() as conn:
             conn.execute(
                 """
@@ -892,9 +960,9 @@ class ControlPlaneDB:
                 """,
                 (
                     provider,
-                    provider_base_url,
+                    base_url,
-                    provider_org,
+                    p_org,
-                    provider_project,
+                    p_project,
                     provider_issue_id,
                     provider_short_id,
                     provider_permalink,
@@ -914,18 +982,12 @@ class ControlPlaneDB:
                 """
                 SELECT * FROM incident_links
                 WHERE provider = ?
-                  AND IFNULL(provider_base_url, '') = IFNULL(?, '')
+                  AND provider_base_url = ?
-                  AND IFNULL(provider_org, '') = IFNULL(?, '')
+                  AND provider_org = ?
-                  AND IFNULL(provider_project, '') = IFNULL(?, '')
+                  AND provider_project = ?
                   AND provider_issue_id = ?
                 """,
-                (
+                (provider, base_url, p_org, p_project, provider_issue_id),
-                    provider,
-                    provider_base_url,
-                    provider_org,
-                    provider_project,
-                    provider_issue_id,
-                ),
             ).fetchone()
             return dict(row)
 

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

@@ -1,7 +1,9 @@
 # Control-plane DB substrate (#613)
 
-**Status:** Implemented (SQLite single-writer MVP)  
+**Status:** Implemented (SQLite single-writer MVP)
-**ADR:** [`mcp-allocator-control-plane-observability-adr.md`](mcp-allocator-control-plane-observability-adr.md)  
+
+**ADR:** [`mcp-allocator-control-plane-observability-adr.md`](mcp-allocator-control-plane-observability-adr.md)
+
 **Module:** `control_plane_db.py`
 
 ## Architecture statement
@@ -14,17 +16,19 @@
 |------------|--------|
 | 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` — workers need a live assignment/lease |
+| 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 |
+| `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`.
+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. 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).
+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.
 
 ## Dependency chain
 

tests/test_control_plane_db.py

@@ -36,7 +36,7 @@ class ControlPlaneDBTest(unittest.TestCase):
             rows = dict(conn.execute("SELECT key, value FROM schema_meta").fetchall())
         finally:
             conn.close()
-        self.assertEqual(rows["schema_version"], "1")
+        self.assertEqual(rows["schema_version"], "2")
         self.assertIn("DB coordinates", rows["architecture"])
         self.assertIn("bridge", rows["architecture"].lower())
 
@@ -372,6 +372,159 @@ class ControlPlaneDBTest(unittest.TestCase):
         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_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()