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).

control_plane_db.py

@@ -317,13 +317,68 @@ class ControlPlaneDB:
             finally:
                 conn.close()
 
+    # Observation fields that must agree before collapsing legacy duplicates.
+    # Deleting a row would drop any of these; silent loss is forbidden (#619 RC3).
+    _INCIDENT_LINK_OBS_COMPARE_FIELDS: tuple[str, ...] = (
+        "provider_short_id",
+        "provider_permalink",
+        "fingerprint",
+        "gitea_org",
+        "gitea_repo",
+        "gitea_issue_number",
+        "linked_pr_numbers",
+        "first_seen",
+        "last_seen",
+        "event_count",
+        "status",
+        "release_resolved_at",
+        "last_sync_at",
+    )
+
+    @staticmethod
+    def _norm_incident_obs_value(field: str, value: Any) -> Any:
+        """Normalize optional observation values for equality during migration.
+
+        NULL and empty string are treated as equivalent for optional text
+        fields (legacy rows often omit them). Integer fields keep NULL
+        distinct from zero so event_count=0 vs NULL is not collapsed away.
+        """
+        if field in ("gitea_issue_number", "event_count"):
+            if value is None or value == "":
+                return None
+            return int(value)
+        if value is None:
+            return ""
+        return str(value)
+
+    def _incident_link_observation_identity(
+        self, row: sqlite3.Row, cols: set[str]
+    ) -> tuple[Any, ...]:
+        """Stable comparable identity of all meaningful observation metadata."""
+        parts: list[Any] = []
+        for field in self._INCIDENT_LINK_OBS_COMPARE_FIELDS:
+            if field not in cols:
+                continue
+            # Row keys match column names; missing keys treated as absent.
+            try:
+                raw = row[field]
+            except (IndexError, KeyError):
+                raw = None
+            parts.append((field, self._norm_incident_obs_value(field, raw)))
+        return tuple(parts)
+
     def _migrate_incident_links_null_scope(self, conn: sqlite3.Connection) -> None:
-        """Collapse legacy NULL-scope duplicates, then normalize to '' (#619 RC2).
+        """Collapse legacy NULL-scope duplicates, then normalize to '' (#619 RC).
 
         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 ''.
+        Gitea targets **or** any other meaningful observation metadata conflict
+        within a group (no silent data loss), then coerce NULLs to ''.
+
+        Policy: **fail closed** — never invent a merge of fingerprint/status/
+        event_count/etc. Only identical (after optional-text NULL≈'') rows may
+        collapse to the lowest ``link_id``.
         """
         cols = {
             row[1]
@@ -332,17 +387,41 @@ class ControlPlaneDB:
         if not cols:
             return
 
+        # Build SELECT from present columns so partial legacy schemas still migrate.
+        # Scope fields are normalized in the SELECT aliases used for grouping.
+        required = ("link_id", "provider", "provider_issue_id", "gitea_org", "gitea_repo", "gitea_issue_number")
+        if not all(c in cols for c in required):
+            return
+
+        select_parts = [
+            "link_id",
+            "provider",
+            "IFNULL(provider_base_url, '') AS base_url"
+            if "provider_base_url" in cols
+            else "'' AS base_url",
+            "IFNULL(provider_org, '') AS p_org"
+            if "provider_org" in cols
+            else "'' AS p_org",
+            "IFNULL(provider_project, '') AS p_project"
+            if "provider_project" in cols
+            else "'' AS p_project",
+            "provider_issue_id",
+            "gitea_org",
+            "gitea_repo",
+            "gitea_issue_number",
+        ]
+        for field in self._INCIDENT_LINK_OBS_COMPARE_FIELDS:
+            if field in (
+                "gitea_org",
+                "gitea_repo",
+                "gitea_issue_number",
+            ):
+                continue  # already selected
+            if field in cols:
+                select_parts.append(field)
+
         rows = conn.execute(
-            """
+            f"SELECT {', '.join(select_parts)} FROM incident_links ORDER BY link_id ASC"
-            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]] = {}
@@ -374,7 +453,20 @@ class ControlPlaneDB:
                     "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)
+            # Same Gitea target is not enough: fingerprint/status/event_count/
+            # permalinks/timestamps/etc. must also agree or we lose data.
+            obs_identities = {
+                self._incident_link_observation_identity(m, cols) for m in members
+            }
+            if len(obs_identities) > 1:
+                raise ControlPlaneError(
+                    "incident_links migration fail closed: conflicting "
+                    "observation metadata for provider key "
+                    f"{key!r} (same Gitea target but differing fingerprint/"
+                    "status/event_count/permalink/timestamps/or related fields); "
+                    "refusing silent discard of duplicate rows"
+                )
+            # lowest link_id kept (ORDER BY link_id ASC)
             for m in members[1:]:
                 to_delete.append(int(m["link_id"]))
 

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

@@ -29,6 +29,7 @@
 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
 

tests/test_control_plane_db.py

@@ -686,6 +686,93 @@ class ControlPlaneDBTest(unittest.TestCase):
             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(