fix(automation): bound backup approval fingerprints

apps/api/tests/integration/test_backup_restore_prod_db_regressions.py

@@ -15,7 +15,7 @@ from sqlalchemy import text
 from sqlalchemy.ext.asyncio import create_async_engine
 
 from src.jobs import backup_restore_legacy_backfill_job as backfill
-from src.services import platform_operator_service
+from src.services import backup_restore_alertmanager_ingress, platform_operator_service
 
 
 def _explicit_local_test_database_url() -> str:
@@ -146,6 +146,42 @@ async def test_backfill_run_inserts_satisfy_no_default_cost_column() -> None:
         await engine.dispose()
 
 
+@pytest.mark.asyncio
+async def test_long_occurrence_fingerprint_fits_real_pg_varchar() -> None:
+    engine = create_async_engine(_explicit_local_test_database_url(), echo=False)
+    source_fingerprint = "legacy-backup-occurrence:" + ("a" * 64)
+    stored_fingerprint = (
+        backup_restore_alertmanager_ingress._approval_storage_fingerprint(
+            source_fingerprint
+        )
+    )
+    try:
+        async with engine.connect() as connection:
+            transaction = await connection.begin()
+            try:
+                await connection.execute(text("""
+                    CREATE TEMP TABLE approval_records (
+                        fingerprint VARCHAR(64) NOT NULL
+                    ) ON COMMIT DROP
+                """))
+                await connection.execute(
+                    text("""
+                        INSERT INTO approval_records (fingerprint)
+                        VALUES (:fingerprint)
+                    """),
+                    {"fingerprint": stored_fingerprint},
+                )
+                persisted = await connection.scalar(
+                    text("SELECT fingerprint FROM approval_records")
+                )
+                assert persisted == stored_fingerprint
+                assert len(persisted) == 64
+            finally:
+                await transaction.rollback()
+    finally:
+        await engine.dispose()
+
+
 @pytest.mark.asyncio
 async def test_direct_readback_rewrites_ssl_and_keeps_rls_context_in_transaction(
     monkeypatch: pytest.MonkeyPatch,

apps/api/tests/test_backup_restore_alertmanager_ingress.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import hashlib
 from datetime import UTC, datetime
 from types import SimpleNamespace
 from unittest.mock import AsyncMock
@@ -46,7 +47,7 @@ def _signal_kwargs() -> dict[str, object]:
             "lane": "backup_restore_escrow_triage",
         },
         "annotations": {"summary": "readback verifier required"},
-        "source_fingerprint": "source-fingerprint",
+        "source_fingerprint": "legacy-backup-occurrence:" + ("a" * 64),
         "source_event_fingerprint": "alertmanager-native-fingerprint",
         "source_started_at": "2026-07-11T10:00:00Z",
         "alert_category": "backup",
@@ -84,6 +85,17 @@ def test_backup_ingress_owner_is_stable_per_source_occurrence() -> None:
     assert first["read_only"] is True
 
 
+def test_backup_approval_fingerprint_fits_durable_varchar_boundary() -> None:
+    short = "source-fingerprint"
+    long = "legacy-backup-occurrence:" + ("a" * 64)
+
+    assert ingress._approval_storage_fingerprint(short) == short
+    stored = ingress._approval_storage_fingerprint(long)
+    assert stored == hashlib.sha256(long.encode("utf-8")).hexdigest()
+    assert len(stored) == ingress.APPROVAL_FINGERPRINT_MAX_LENGTH
+    assert ingress._approval_storage_fingerprint(long) == stored
+
+
 @pytest.mark.asyncio
 async def test_alertmanager_type1_backup_queues_durable_owner_before_info_shortcut(
     monkeypatch: pytest.MonkeyPatch,
@@ -167,12 +179,18 @@ class _FakeApprovalService:
 
     async def find_by_fingerprint(self, **_kwargs):
         self.calls.append("legacy_owner_read")
+        expected = ingress._approval_storage_fingerprint(
+            str(_signal_kwargs()["source_fingerprint"])
+        )
+        assert _kwargs["fingerprint"] == expected
         return None
 
     async def create_approval_with_fingerprint(self, *, request, fingerprint):
         self.calls.append("owner_create")
         self.created_request = request
-        assert fingerprint == "source-fingerprint"
+        assert fingerprint == ingress._approval_storage_fingerprint(
+            str(_signal_kwargs()["source_fingerprint"])
+        )
         self.approval = SimpleNamespace(
             id=UUID(request.metadata["preallocated_approval_id"]),
             incident_id=request.incident_id,
@@ -218,6 +236,7 @@ async def test_backup_ingress_binds_canonical_incident_then_dispatches_and_reads
         calls.append("agent99_dispatch")
         assert kwargs["incident_id"] == incident_id
         assert kwargs["route_id"] == "agent99:backup_health:BackupCheck"
+        assert kwargs["fingerprint"] == _signal_kwargs()["source_fingerprint"]
         identity.update({
             "incident_id": incident_id,
             "run_id": "12345678-1234-5678-9234-567812345678",