fix: safe incident_links migration order; require PR head pin

Address remaining PR #619 blockers on head 783ea88:

- Deduplicate legacy NULL-scope incident_links before normalizing to ''
- Fail closed when duplicate rows disagree on Gitea target
- Require non-empty expected_head_sha for PR assign and mutation

control_plane_db.py

@@ -318,33 +318,75 @@ class ControlPlaneDB:
                 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)."""
+        """Collapse legacy NULL-scope duplicates, then normalize to '' (#619 RC2).
+
+        Order is mandatory: SQLite UNIQUE treats multiple NULLs as distinct, so
+        normalizing NULL→'' first can hit the UNIQUE constraint and abort
+        migration. Deduplicate under the *normalized* key first, fail closed if
+        Gitea targets conflict within a group, then coerce NULLs to ''.
+        """
         cols = {
             row[1]
             for row in conn.execute("PRAGMA table_info(incident_links)").fetchall()
         }
         if not cols:
             return
+
+        rows = conn.execute(
+            """
+            SELECT link_id, provider,
+                   IFNULL(provider_base_url, '') AS base_url,
+                   IFNULL(provider_org, '') AS p_org,
+                   IFNULL(provider_project, '') AS p_project,
+                   provider_issue_id,
+                   gitea_org, gitea_repo, gitea_issue_number
+            FROM incident_links
+            ORDER BY link_id ASC
+            """
+        ).fetchall()
+
+        groups: dict[tuple[str, str, str, str, str], list[sqlite3.Row]] = {}
+        for row in rows:
+            key = (
+                str(row["provider"]),
+                str(row["base_url"]),
+                str(row["p_org"]),
+                str(row["p_project"]),
+                str(row["provider_issue_id"]),
+            )
+            groups.setdefault(key, []).append(row)
+
+        to_delete: list[int] = []
+        for key, members in groups.items():
+            if len(members) < 2:
+                continue
+            # Canonical identity for observation links: Gitea issue target.
+            targets = {
+                (
+                    str(m["gitea_org"] or ""),
+                    str(m["gitea_repo"] or ""),
+                    int(m["gitea_issue_number"]),
+                )
+                for m in members
+            }
+            if len(targets) > 1:
+                raise ControlPlaneError(
+                    "incident_links migration fail closed: conflicting Gitea "
+                    f"targets for provider key {key!r}: {sorted(targets)}"
+                )
+            keep_id = int(members[0]["link_id"])  # lowest link_id (ORDER BY)
+            for m in members[1:]:
+                to_delete.append(int(m["link_id"]))
+
+        for link_id in to_delete:
+            conn.execute("DELETE FROM incident_links WHERE link_id = ?", (link_id,))
+
+        # Safe only after dedupe: normalize NULL scope fields to empty string.
         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 ──────────────────────────────────────────────────────────
 
@@ -481,6 +523,11 @@ class ControlPlaneDB:
             raise InvalidWorkKindError(
                 f"cannot assign kind '{kind}'; only {sorted(WORK_KINDS)}"
             )
+        head_pin = (expected_head_sha or "").strip()
+        if kind_norm == "pr" and not head_pin:
+            raise LeaseRequiredError(
+                f"pr#{int(number)} assignment requires non-empty expected_head_sha pin"
+            )
 
         allowed = tuple(allowed_actions or ("implement", "comment"))
         forbidden = tuple(
@@ -825,9 +872,15 @@ 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.
+            # Head pin: PRs require a non-empty expected_head_sha (fail closed).
             assigned_head = (asn["expected_head_sha"] or "").strip()
             current_head = (work["current_head_sha"] or "").strip()
+            if kind_norm == "pr" and not assigned_head:
+                raise LeaseRequiredError(
+                    f"assignment {asn['assignment_id']} for pr#{number} has no "
+                    f"expected_head_sha pin; PR mutations require a head pin"
+                )
+            # Stale-head: pinned expected_head_sha must still match live work item.
             if assigned_head and assigned_head != current_head:
                 raise LeaseRequiredError(
                     f"assignment {asn['assignment_id']} stale head: expected "

tests/test_control_plane_db.py

@@ -123,6 +123,7 @@ class ControlPlaneDBTest(unittest.TestCase):
             repo="r",
             kind="pr",
             number=50,
+            expected_head_sha="head-50",
         )
         b = self.db.assign_and_lease(
             session_id="s1",
@@ -132,6 +133,7 @@ class ControlPlaneDBTest(unittest.TestCase):
             repo="r",
             kind="pr",
             number=50,
+            expected_head_sha="head-50",
         )
         self.assertEqual(b.outcome, "assigned")
         self.assertEqual(b.lease_id, a.lease_id)
@@ -241,6 +243,7 @@ class ControlPlaneDBTest(unittest.TestCase):
             repo="r",
             kind="pr",
             number=99,
+            expected_head_sha="merged-head",
         )
         self.assertEqual(result.outcome, "no_safe_work")
 
@@ -479,6 +482,210 @@ class ControlPlaneDBTest(unittest.TestCase):
                 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_incident_links_minimal_upsert_is_canonical(self) -> None:
         """Repeated minimal upserts must update one row (NULL-safe uniqueness)."""
         a = self.db.upsert_incident_link(