fix(storage): a re-keyed raw replaces its stale verdict-cache row

polylogue/storage/raw_authority_verdict_cache.py

@@ -183,6 +183,16 @@ def write_raw_authority_verdict_cache(
             "DELETE FROM raw_authority_verdicts WHERE logical_source_key = ?",
             (logical_source_key,),
         )
+        # A raw re-keyed since it was cached (a pending key resolved after
+        # parse) still holds its row under the old key; raw_id is unique, so
+        # that row goes too or the insert fails and the converger stalls.
+        raw_ids = list(verdicts)
+        for offset in range(0, len(raw_ids), 500):
+            batch = raw_ids[offset : offset + 500]
+            conn.execute(
+                f"DELETE FROM raw_authority_verdicts WHERE raw_id IN ({','.join('?' for _ in batch)})",
+                batch,
+            )
         conn.executemany(
             """
             INSERT INTO raw_authority_verdicts

tests/unit/storage/test_raw_authority_verdict_cache.py

@@ -304,3 +304,22 @@ def _fail(*args: object, **kwargs: object) -> dict[str, RawAuthorityVerdict]:
 
     assert outcome.warmed_cohorts == 0
     assert outcome.pending_cohorts is False
+
+
+def test_rekeyed_raw_replaces_its_stale_cache_row(tmp_path: Path) -> None:
+    """A raw cached under one logical key is written again under another.
+
+    Anti-vacuity: dropping the raw_id delete in the writer makes this raise
+    ``IntegrityError: UNIQUE constraint failed: raw_authority_verdicts.raw_id``,
+    which stalled the converger 84 times in rehearsal-10 (2026-09-05).
+    """
+    initialize_active_archive_root(tmp_path)
+    with ArchiveStore.open_existing(tmp_path, read_only=False) as archive:
+        _bind_full(archive, raw_id="moved", payload=b"one\n", logical_source_key="codex:pending")
+        write_raw_authority_verdict_cache(
+            archive, "codex:pending", {"moved": RawAuthorityVerdict.VERIFIED}, now_ms=1000
+        )
+        write_raw_authority_verdict_cache(archive, "codex:s1", {"moved": RawAuthorityVerdict.VERIFIED}, now_ms=2000)
+        conn = archive._ensure_source_conn()
+        rows = conn.execute("SELECT logical_source_key FROM raw_authority_verdicts WHERE raw_id = 'moved'").fetchall()
+    assert [row[0] for row in rows] == ["codex:s1"]