fix(alerting): cluster legacy backup incident replay

apps/api/src/core/config.py

@@ -1125,10 +1125,11 @@ class Settings(BaseSettings):
         description="Timeout for fail-open Agent99 SRE alert relay POSTs.",
     )
     ENABLE_BACKUP_RESTORE_LEGACY_BACKFILL: bool = Field(
-        default=True,
+        default=False,
         description=(
             "Enable bounded durable BackupCheck replay for legacy backup/restore "
-            "Telegram receipt cards when a trusted Agent99 relay is configured."
+            "Telegram receipt cards when a trusted Agent99 relay is configured. "
+            "Keep disabled until the recurrence-cluster canary is verified."
         ),
     )
     BACKUP_RESTORE_LEGACY_BACKFILL_STARTUP_SLEEP_SECONDS: float = Field(

apps/api/src/jobs/backup_restore_legacy_backfill_job.py

@@ -37,7 +37,7 @@ from src.services.backup_restore_alertmanager_ingress import (
 
 logger = structlog.get_logger(__name__)
 
-BACKFILL_SCHEMA_VERSION = "backup_restore_legacy_backfill_v1"
+BACKFILL_SCHEMA_VERSION = "backup_restore_legacy_backfill_v2"
 BACKFILL_CURSOR_SCHEMA_VERSION = "backup_restore_legacy_backfill_cursor_v1"
 BACKFILL_RESULT_SCHEMA_VERSION = "backup_restore_legacy_backfill_result_v1"
 BACKFILL_CURSOR_AGENT_ID = "backup_restore_legacy_backfill_cursor"
@@ -59,6 +59,9 @@ MAX_SCOPE_ROWS = 100
 REPO_KNOWN_LAN_RELAY_HOST = "192.168.0.99"
 REPO_KNOWN_LAN_RELAY_PORT = 8787
 REPO_KNOWN_LAN_RELAY_PATH = "/agent99/sre-alert"
+LEGACY_OCCURRENCE_SCHEMA_VERSION = "backup_restore_legacy_occurrence_v1"
+RECURRENCE_OCCURRENCE_SCHEMA_VERSION = "backup_restore_recurrence_occurrence_v2"
+RECURRENCE_FINGERPRINT_PREFIX = "legacy-backup-recurrence:"
 
 Processor = Callable[..., Awaitable[dict[str, Any]]]
 
@@ -878,21 +881,22 @@ def build_legacy_backup_occurrence(
         raise ValueError("backup_backfill_source_fingerprint_required")
     fingerprint_hash = _sha256(fingerprint)[:16]
     occurrence_id = f"legacy-backup:{source_message}:{fingerprint_hash}"
-    dispatch_source_fingerprint = "legacy-backup-occurrence:" + _sha256(
-        f"{project}\x1f{source_message}\x1f{fingerprint}"
+    recurrence_id = f"{RECURRENCE_FINGERPRINT_PREFIX}{fingerprint_hash}"
+    dispatch_source_fingerprint = RECURRENCE_FINGERPRINT_PREFIX + _sha256(
+        f"{project}\x1f{fingerprint}"
     )
     owner = build_backup_restore_ingress_owner(
         project_id=project,
         source_fingerprint=dispatch_source_fingerprint,
-        source_event_fingerprint=occurrence_id,
-        source_started_at=queued,
+        source_event_fingerprint=recurrence_id,
+        source_started_at="",
     )
     item_run_id = uuid5(
         NAMESPACE_URL,
         f"{BACKFILL_SCHEMA_VERSION}:{project}:{occurrence_id}",
     )
     return {
-        "schema_version": "backup_restore_legacy_occurrence_v1",
+        "schema_version": RECURRENCE_OCCURRENCE_SCHEMA_VERSION,
         "project_id": project,
         "source_message_id": source_message,
         "source_fingerprint": fingerprint,
@@ -900,6 +904,7 @@ def build_legacy_backup_occurrence(
         "dispatch_source_fingerprint": dispatch_source_fingerprint,
         "source_queued_at": queued,
         "occurrence_id": occurrence_id,
+        "recurrence_id": recurrence_id,
         "backfill_run_id": str(item_run_id),
         "incident_id": owner["incident_id"],
         "approval_id": owner["approval_id"],
@@ -992,7 +997,7 @@ def _item_payload_from_row(
     else:
         missing_hash = _sha256(f"missing:{project_id}:{message_id}")[:16]
         occurrence = {
-            "schema_version": "backup_restore_legacy_occurrence_v1",
+            "schema_version": RECURRENCE_OCCURRENCE_SCHEMA_VERSION,
             "project_id": project_id,
             "source_message_id": message_id,
             "source_fingerprint": "",
@@ -1000,6 +1005,7 @@ def _item_payload_from_row(
             "dispatch_source_fingerprint": "",
             "source_queued_at": queued_at,
             "occurrence_id": f"legacy-backup-blocked:{message_id}:{missing_hash}",
+            "recurrence_id": None,
             "backfill_run_id": str(
                 uuid5(
                     NAMESPACE_URL,
@@ -1760,6 +1766,22 @@ async def _replay_backfill_claim(
     processor: Processor,
     relay_auth_mode: str = "injected_processor",
 ) -> str:
+    if claim.item.get("schema_version") == LEGACY_OCCURRENCE_SCHEMA_VERSION:
+        result = _public_result(
+            status="superseded_unique_incident_replay_no_dispatch",
+            occurrence=claim.item,
+            reason="superseded_by_recurrence_cluster_v2",
+            auth_mode=relay_auth_mode,
+        )
+        finished = await _finish_backfill_claim(
+            project_id=project_id,
+            claim=claim,
+            state="completed",
+            result=result,
+            error_code=None,
+        )
+        return "deduplicated" if finished else "lease_lost"
+
     card = await _load_legacy_backup_card(project_id=project_id, claim=claim)
     if card is None:
         result = _public_result(
@@ -1887,8 +1909,8 @@ async def _replay_backfill_claim(
             },
             annotations={"summary": "Read-only BackupCheck historical evidence replay"},
             source_fingerprint=occurrence["dispatch_source_fingerprint"],
-            source_event_fingerprint=occurrence["occurrence_id"],
-            source_started_at=card.queued_at,
+            source_event_fingerprint=occurrence["recurrence_id"],
+            source_started_at="",
             alert_category="backup",
             notification_type="TYPE-1",
             source_url=None,

apps/api/tests/test_backup_restore_legacy_backfill_job.py

@@ -10,6 +10,7 @@ from uuid import NAMESPACE_URL, UUID, uuid5
 
 import pytest
 
+from src.core.config import Settings
 from src.core.context import (
     clear_project_context,
     get_current_project_id,
@@ -159,17 +160,57 @@ def test_legacy_occurrence_identity_is_stable_and_source_specific() -> None:
     )
 
     assert first == replay
-    for other in (next_message, next_fingerprint):
-        assert other["occurrence_id"] != first["occurrence_id"]
-        assert other["incident_id"] != first["incident_id"]
-        assert other["work_item_id"] != first["work_item_id"]
-        assert other["backfill_run_id"] != first["backfill_run_id"]
+    assert next_message["occurrence_id"] != first["occurrence_id"]
+    assert next_message["backfill_run_id"] != first["backfill_run_id"]
+    assert next_message["recurrence_id"] == first["recurrence_id"]
+    assert next_message["incident_id"] == first["incident_id"]
+    assert next_message["work_item_id"] == first["work_item_id"]
+    for key in ("recurrence_id", "incident_id", "work_item_id"):
+        assert next_fingerprint[key] != first[key]
+    assert next_fingerprint["backfill_run_id"] != first["backfill_run_id"]
     assert first["source_message_id"] in first["occurrence_id"]
-    assert first["dispatch_source_fingerprint"].startswith("legacy-backup-occurrence:")
+    assert first["dispatch_source_fingerprint"].startswith(
+        job.RECURRENCE_FINGERPRINT_PREFIX
+    )
     assert first["controlled_apply"] is False
     assert first["telegram_dispatch_performed"] is False
 
 
+def test_legacy_replay_is_disabled_until_recurrence_canary_is_verified() -> None:
+    assert (
+        Settings.model_fields["ENABLE_BACKUP_RESTORE_LEGACY_BACKFILL"].default
+        is False
+    )
+
+
+@pytest.mark.asyncio
+async def test_v1_unique_incident_claim_is_superseded_without_dispatch(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    claim = _claim_from_row(1)
+    claim.item["schema_version"] = job.LEGACY_OCCURRENCE_SCHEMA_VERSION
+    processor = AsyncMock()
+    finish = AsyncMock(return_value=True)
+    monkeypatch.setattr(job, "_finish_backfill_claim", finish)
+
+    result = await job._replay_backfill_claim(
+        project_id="awoooi",
+        claim=claim,
+        processor=processor,
+    )
+
+    assert result == "deduplicated"
+    processor.assert_not_awaited()
+    finish.assert_awaited_once()
+    call = finish.await_args.kwargs
+    assert call["state"] == "completed"
+    assert call["error_code"] is None
+    assert call["result"]["status"] == (
+        "superseded_unique_incident_replay_no_dispatch"
+    )
+    assert call["result"]["reason"] == "superseded_by_recurrence_cluster_v2"
+
+
 def test_cursor_never_reports_completed_when_any_item_failed() -> None:
     detail = {
         "source_exhausted": True,
@@ -776,7 +817,9 @@ async def test_exact_98_legacy_cards_replay_in_bounded_restart_safe_batches(
         assert kwargs["project_id"] == "awoooi"
         assert kwargs["labels"]["event_type"] == job.BACKUP_RESTORE_EVENT_TYPE
         assert kwargs["labels"]["lane"] == job.BACKUP_RESTORE_LANE
-        assert kwargs["source_event_fingerprint"].startswith("legacy-backup:")
+        assert kwargs["source_event_fingerprint"].startswith(
+            job.RECURRENCE_FINGERPRINT_PREFIX
+        )
         processor_calls.append(kwargs)
         return _durable_dispatch_result(kwargs)
 
@@ -871,10 +914,10 @@ async def test_exact_98_legacy_cards_replay_in_bounded_restart_safe_batches(
         len({call["labels"]["legacy_source_message_id"] for call in processor_calls})
         == 98
     )
-    assert len({call["source_fingerprint"] for call in processor_calls}) == 98
-    assert len({value["incident_id"] for value in projected.values()}) == 98
-    assert len({value["work_item_id"] for value in projected.values()}) == 98
-    assert len({value["run_id"] for value in projected.values()}) == 98
+    assert len({call["source_fingerprint"] for call in processor_calls}) == 1
+    assert len({value["incident_id"] for value in projected.values()}) == 1
+    assert len({value["work_item_id"] for value in projected.values()}) == 1
+    assert len({value["run_id"] for value in projected.values()}) == 1
     assert {value["auth_mode"] for value in projected.values()} == {
         "injected_processor"
     }