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_org TEXT,
-    provider_project TEXT,
+    provider_base_url TEXT NOT NULL DEFAULT '',
+    provider_org TEXT NOT NULL DEFAULT '',
+    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,
-                    provider_org,
-                    provider_project,
+                    base_url,
+                    p_org,
+                    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 IFNULL(provider_org, '') = IFNULL(?, '')
-                  AND IFNULL(provider_project, '') = IFNULL(?, '')
+                  AND provider_base_url = ?
+                  AND provider_org = ?
+                  AND provider_project = ?
                   AND provider_issue_id = ?
                 """,
-                (
-                    provider,
-                    provider_base_url,
-                    provider_org,
-                    provider_project,
-                    provider_issue_id,
-                ),
+                (provider, base_url, p_org, p_project, provider_issue_id),
             ).fetchone()
             return dict(row)