fix(automation): domain-separate backup fingerprints

apps/api/src/services/backup_restore_alertmanager_ingress.py

@@ -40,6 +40,9 @@ BACKUP_RESTORE_AGENT99_ROUTE_ID = "agent99:backup_health:BackupCheck"
 BACKUP_RESTORE_DISPATCH_OWNER = "alertmanager_raw_signal"
 BACKUP_RESTORE_OWNER_SCHEMA_VERSION = "backup_restore_ingress_owner_v1"
 APPROVAL_FINGERPRINT_MAX_LENGTH = 64
+APPROVAL_FINGERPRINT_HASH_DOMAIN = (
+    "awoooi.backup_restore.approval_fingerprint.v2"
+)
 BACKUP_RESTORE_PROHIBITED_ACTIONS = (
     "run_backup",
     "run_restore",
@@ -57,12 +60,76 @@ def _stable_digest(*parts: object) -> str:
 
 
 def _approval_storage_fingerprint(source_fingerprint: str) -> str:
-    """Fit a source identity into approval_records.fingerprint VARCHAR(64)."""
+    """Return the domain-separated durable identity for every source value.
+
+    Hashing only values longer than the PostgreSQL column made a long source
+    collide with a raw 64-character source equal to that source's SHA-256.
+    Hash every value in the lane with a domain tag so the two source domains
+    remain distinct while still fitting ``approval_records.fingerprint``.
+    """
 
     normalized = str(source_fingerprint or "").strip()
-    if len(normalized) <= APPROVAL_FINGERPRINT_MAX_LENGTH:
-        return normalized
-    return hashlib.sha256(normalized.encode("utf-8")).hexdigest()
+    payload = f"{APPROVAL_FINGERPRINT_HASH_DOMAIN}\x1f{normalized}"
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+
+def _approval_fingerprint_lookup_candidates(
+    source_fingerprint: str,
+) -> tuple[str, ...]:
+    """Return v2 first, followed by the one compatible legacy encoding.
+
+    The v1 implementation stored short values verbatim and unscoped-hashed
+    long values.  Both representations remain read-compatible during the
+    rollout, but new rows always use the domain-separated v2 fingerprint.
+    """
+
+    normalized = str(source_fingerprint or "").strip()
+    canonical = _approval_storage_fingerprint(normalized)
+    legacy = (
+        normalized
+        if len(normalized) <= APPROVAL_FINGERPRINT_MAX_LENGTH
+        else hashlib.sha256(normalized.encode("utf-8")).hexdigest()
+    )
+    return tuple(dict.fromkeys((canonical, legacy)))
+
+
+def _approval_matches_source_fingerprint(
+    approval: Any,
+    source_fingerprint: str,
+) -> bool:
+    """Verify a lookup row before reusing its canonical Incident binding."""
+
+    metadata = getattr(approval, "metadata", None)
+    if not isinstance(metadata, dict):
+        return False
+    persisted_source = str(metadata.get("source_fingerprint") or "").strip()
+    return persisted_source == str(source_fingerprint or "").strip()
+
+
+async def _find_compatible_source_approval(
+    approval_service: Any,
+    *,
+    source_fingerprint: str,
+) -> Any | None:
+    """Read v2/v1 identities without accepting a cross-domain collision."""
+
+    for storage_fingerprint in _approval_fingerprint_lookup_candidates(
+        source_fingerprint
+    ):
+        recent = await approval_service.find_by_fingerprint(
+            fingerprint=storage_fingerprint,
+            debounce_minutes=30,
+        )
+        if recent is None:
+            continue
+        if _approval_matches_source_fingerprint(recent, source_fingerprint):
+            return recent
+        logger.warning(
+            "backup_restore_approval_fingerprint_identity_mismatch",
+            storage_fingerprint=storage_fingerprint,
+            approval_id=_value(getattr(recent, "id", "")),
+        )
+    return None
 
 
 def build_backup_restore_ingress_owner(
@@ -252,9 +319,9 @@ async def process_backup_restore_alertmanager_signal(
     created_owner = approval is None
     legacy_incident_id = ""
     if approval is None:
-        recent = await approval_service.find_by_fingerprint(
-            fingerprint=approval_storage_fingerprint,
-            debounce_minutes=30,
+        recent = await _find_compatible_source_approval(
+            approval_service,
+            source_fingerprint=source_fingerprint,
         )
         legacy_incident_id = _value(getattr(recent, "incident_id", ""))
         request = _backup_restore_approval_request(