From 8bee637466072548bc92c266200747a3bb428f47 Mon Sep 17 00:00:00 2001 From: ogt Date: Tue, 14 Jul 2026 11:24:50 +0800 Subject: [PATCH] fix(automation): bind backup approvals to full owner --- .../backup_restore_alertmanager_ingress.py | 375 +++++++++++++--- ...test_backup_restore_prod_db_regressions.py | 231 ++++++++-- ...est_backup_restore_alertmanager_ingress.py | 421 +++++++++++++++--- 3 files changed, 875 insertions(+), 152 deletions(-) diff --git a/apps/api/src/services/backup_restore_alertmanager_ingress.py b/apps/api/src/services/backup_restore_alertmanager_ingress.py index c2880d17f..f0d6a7260 100644 --- a/apps/api/src/services/backup_restore_alertmanager_ingress.py +++ b/apps/api/src/services/backup_restore_alertmanager_ingress.py @@ -9,6 +9,7 @@ write escrow markers, or read secrets. from __future__ import annotations import hashlib +import json from typing import Any from uuid import NAMESPACE_URL, UUID, uuid5 @@ -38,10 +39,14 @@ logger = get_logger("awoooi.backup_restore_alertmanager_ingress") 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" +BACKUP_RESTORE_LEGACY_OWNER_SCHEMA_VERSION = "backup_restore_ingress_owner_v1" +BACKUP_RESTORE_OWNER_SCHEMA_VERSION = "backup_restore_ingress_owner_v2" +BACKUP_RESTORE_OWNER_KEY_ENCODING = "canonical_json_v1" +BACKUP_RESTORE_LEGACY_BACKFILL_PREFIX = "legacy-backup-occurrence:" APPROVAL_FINGERPRINT_MAX_LENGTH = 64 -APPROVAL_FINGERPRINT_HASH_DOMAIN = ( - "awoooi.backup_restore.approval_fingerprint.v2" +APPROVAL_FINGERPRINT_HASH_DOMAIN = "awoooi.backup_restore.approval_fingerprint.v2" +APPROVAL_OWNER_FINGERPRINT_HASH_DOMAIN = ( + "awoooi.backup_restore.approval_owner_fingerprint.v3" ) BACKUP_RESTORE_PROHIBITED_ACTIONS = ( "run_backup", @@ -59,13 +64,72 @@ def _stable_digest(*parts: object) -> str: return hashlib.sha256(value.encode("utf-8")).hexdigest() -def _approval_storage_fingerprint(source_fingerprint: str) -> str: - """Return the domain-separated durable identity for every source value. +def _canonical_json(value: Any) -> str: + return json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) - 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``. + +def _canonical_source_occurrence( + *, + source_fingerprint: str, + source_event_fingerprint: str, + source_started_at: str, +) -> dict[str, str]: + event_fingerprint = str(source_event_fingerprint or "").strip() + started_at = str(source_started_at or "").strip() + return { + "source_event_fingerprint": event_fingerprint, + "source_started_at": started_at, + "fallback_source_fingerprint": ( + "" + if event_fingerprint or started_at + else str(source_fingerprint or "").strip() + ), + } + + +def _legacy_backup_restore_ingress_owner( + *, + project_id: str, + source_fingerprint: str, + source_event_fingerprint: str, + source_started_at: str, +) -> dict[str, str]: + """Rebuild the exact v1 IDs solely for bounded compatibility lookup.""" + + source_occurrence = ( + "|".join( + value for value in (source_event_fingerprint, source_started_at) if value + ) + or source_fingerprint + ) + owner_key = ( + f"{BACKUP_RESTORE_LEGACY_OWNER_SCHEMA_VERSION}:{project_id}:" + f"{source_fingerprint}:{source_occurrence}" + ) + digest = _stable_digest(owner_key) + return { + "schema_version": BACKUP_RESTORE_LEGACY_OWNER_SCHEMA_VERSION, + "project_id": project_id, + "source_fingerprint": source_fingerprint, + "source_occurrence": source_occurrence, + "source_occurrence_hash": digest, + "approval_id": str(uuid5(NAMESPACE_URL, owner_key)), + "incident_id": f"INC-BRR-{digest[:16].upper()}", + "work_item_id": f"backupcheck:{project_id}:{digest[:32]}", + } + + +def _approval_storage_fingerprint(source_fingerprint: str) -> str: + """Return the transitional v2 source-only storage fingerprint. + + This representation remains a lookup alias for rows written by the first + bounded-fingerprint rollout. New rows use a v3 fingerprint over the full + canonical owner identity instead of only the source fingerprint. """ normalized = str(source_fingerprint or "").strip() @@ -73,59 +137,153 @@ def _approval_storage_fingerprint(source_fingerprint: str) -> str: 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. +def _approval_owner_storage_fingerprint(owner: dict[str, Any]) -> str: + identity = owner.get("owner_identity") + if not isinstance(identity, dict): + raise ValueError("backup_restore_canonical_owner_identity_required") + payload = f"{APPROVAL_OWNER_FINGERPRINT_HASH_DOMAIN}\x1f{_canonical_json(identity)}" + return hashlib.sha256(payload.encode("utf-8")).hexdigest() - 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. + +def _approval_fingerprint_lookup_candidates( + owner: dict[str, Any], +) -> tuple[str, ...]: + """Return full-owner v3, source-only v2, then the v1 source encoding. + + Every source-only candidate is only an index probe. A row is reusable + solely after its complete deterministic owner metadata also matches. """ - normalized = str(source_fingerprint or "").strip() - canonical = _approval_storage_fingerprint(normalized) + normalized = str(owner.get("source_fingerprint") or "").strip() + canonical = _approval_owner_storage_fingerprint(owner) + source_only_v2 = _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))) + return tuple(dict.fromkeys((canonical, source_only_v2, legacy))) -def _approval_matches_source_fingerprint( +def _legacy_owner_reuse_is_unambiguous(owner: dict[str, Any]) -> bool: + """Reject v1 occurrence shapes that cannot be losslessly decoded.""" + + event_fingerprint = str(owner.get("source_event_fingerprint") or "") + started_at = str(owner.get("source_started_at") or "") + if "|" in event_fingerprint or "|" in started_at: + return False + # With exactly one component, v1 cannot tell whether the value was the + # event fingerprint or startsAt because it discarded empty fields. + return bool(event_fingerprint) == bool(started_at) + + +def _approval_matches_owner_identity( approval: Any, - source_fingerprint: str, + *, + owner: dict[str, Any], + legacy_owner: dict[str, str], ) -> bool: - """Verify a lookup row before reusing its canonical Incident binding.""" + """Require the complete deterministic owner before approval reuse.""" 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() + approval_id = _value(getattr(approval, "id", "")) + allowed_ids = { + _value(owner.get("canonical_approval_id")), + _value(owner.get("legacy_approval_id")), + } + if approval_id not in allowed_ids: + return False + + persisted_schema = _value(metadata.get("schema_version")) + if persisted_schema == BACKUP_RESTORE_OWNER_SCHEMA_VERSION: + persisted_identity = metadata.get("owner_identity") + return bool( + isinstance(persisted_identity, dict) + and persisted_identity == owner.get("owner_identity") + and _value(metadata.get("owner_key_encoding")) + == BACKUP_RESTORE_OWNER_KEY_ENCODING + and _value(metadata.get("owner_key_sha256")) + == _value(owner.get("owner_key_sha256")) + and _value(metadata.get("approval_id")) == approval_id + ) + + if persisted_schema != BACKUP_RESTORE_LEGACY_OWNER_SCHEMA_VERSION: + return False + if not _legacy_owner_reuse_is_unambiguous(owner): + return False + return bool( + approval_id == legacy_owner["approval_id"] + and _value(metadata.get("approval_id")) == legacy_owner["approval_id"] + and _value(metadata.get("project_id")) == legacy_owner["project_id"] + and _value(metadata.get("source_fingerprint")) + == legacy_owner["source_fingerprint"] + and _value(metadata.get("source_occurrence")) + == legacy_owner["source_occurrence"] + and _value(metadata.get("source_occurrence_hash")) + == legacy_owner["source_occurrence_hash"] + and _value(metadata.get("incident_id")) == legacy_owner["incident_id"] + and _value(metadata.get("work_item_id")) == legacy_owner["work_item_id"] + ) -async def _find_compatible_source_approval( +def _owner_with_id_scheme( + owner: dict[str, Any], + *, + scheme: str, +) -> dict[str, Any]: + if scheme not in {"canonical_json_v2", "legacy_colon_v1"}: + raise ValueError("backup_restore_owner_id_scheme_invalid") + prefix = "canonical" if scheme == "canonical_json_v2" else "legacy" + selected = dict(owner) + selected.update( + { + "approval_id": selected[f"{prefix}_approval_id"], + "incident_id": selected[f"{prefix}_incident_id"], + "work_item_id": selected[f"{prefix}_work_item_id"], + "identity_id_scheme": scheme, + "legacy_identity_compatibility": scheme == "legacy_colon_v1", + } + ) + return selected + + +def _owner_for_approval( + owner: dict[str, Any], + approval: Any, +) -> dict[str, Any]: + approval_id = _value(getattr(approval, "id", "")) + if approval_id == _value(owner.get("canonical_approval_id")): + return _owner_with_id_scheme(owner, scheme="canonical_json_v2") + if approval_id == _value(owner.get("legacy_approval_id")): + return _owner_with_id_scheme(owner, scheme="legacy_colon_v1") + raise ValueError("backup_restore_approval_owner_id_not_deterministic") + + +async def _find_compatible_owner_approval( approval_service: Any, *, - source_fingerprint: str, + owner: dict[str, Any], + legacy_owner: dict[str, str], ) -> Any | None: - """Read v2/v1 identities without accepting a cross-domain collision.""" + """Probe all storage generations without accepting an owner collision.""" - for storage_fingerprint in _approval_fingerprint_lookup_candidates( - source_fingerprint - ): + for storage_fingerprint in _approval_fingerprint_lookup_candidates(owner): 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): + if _approval_matches_owner_identity( + recent, + owner=owner, + legacy_owner=legacy_owner, + ): return recent logger.warning( - "backup_restore_approval_fingerprint_identity_mismatch", + "backup_restore_approval_owner_identity_mismatch", storage_fingerprint=storage_fingerprint, approval_id=_value(getattr(recent, "id", "")), ) @@ -145,29 +303,41 @@ def build_backup_restore_ingress_owner( fingerprint = str(source_fingerprint or "").strip() if not fingerprint: raise ValueError("backup_restore_source_fingerprint_required") - source_occurrence = "|".join( - value - for value in ( - str(source_event_fingerprint or "").strip(), - str(source_started_at or "").strip(), - ) - if value - ) or fingerprint - owner_key = ( - f"{BACKUP_RESTORE_OWNER_SCHEMA_VERSION}:{project}:" - f"{fingerprint}:{source_occurrence}" + event_fingerprint = str(source_event_fingerprint or "").strip() + started_at = str(source_started_at or "").strip() + source_occurrence = _canonical_source_occurrence( + source_fingerprint=fingerprint, + source_event_fingerprint=event_fingerprint, + source_started_at=started_at, ) - digest = _stable_digest(owner_key) - approval_id = uuid5(NAMESPACE_URL, owner_key) - return { + owner_identity = { "schema_version": BACKUP_RESTORE_OWNER_SCHEMA_VERSION, "project_id": project, "source_fingerprint": fingerprint, "source_occurrence": source_occurrence, + } + owner_key = _canonical_json(owner_identity) + digest = hashlib.sha256(owner_key.encode("utf-8")).hexdigest() + legacy_owner = _legacy_backup_restore_ingress_owner( + project_id=project, + source_fingerprint=fingerprint, + source_event_fingerprint=event_fingerprint, + source_started_at=started_at, + ) + owner = { + **owner_identity, + "owner_identity": owner_identity, + "owner_key_encoding": BACKUP_RESTORE_OWNER_KEY_ENCODING, + "owner_key_sha256": digest, + "source_event_fingerprint": event_fingerprint, + "source_started_at": started_at, "source_occurrence_hash": digest, - "approval_id": str(approval_id), - "incident_id": f"INC-BRR-{digest[:16].upper()}", - "work_item_id": f"backupcheck:{project}:{digest[:32]}", + "canonical_approval_id": str(uuid5(NAMESPACE_URL, owner_key)), + "canonical_incident_id": f"INC-BRR-{digest[:16].upper()}", + "canonical_work_item_id": f"backupcheck:{project}:{digest[:32]}", + "legacy_approval_id": legacy_owner["approval_id"], + "legacy_incident_id": legacy_owner["incident_id"], + "legacy_work_item_id": legacy_owner["work_item_id"], "route_id": BACKUP_RESTORE_AGENT99_ROUTE_ID, "kind": "backup_health", "mode": "BackupCheck", @@ -177,6 +347,15 @@ def build_backup_restore_ingress_owner( "read_only": True, "prohibited_actions": list(BACKUP_RESTORE_PROHIBITED_ACTIONS), } + # Production legacy-backfill claims already persist v1 IDs in their item + # payload. Preserve those IDs while upgrading their metadata and storage + # fingerprint; all other new episodes use collision-safe canonical IDs. + scheme = ( + "legacy_colon_v1" + if fingerprint.startswith(BACKUP_RESTORE_LEGACY_BACKFILL_PREFIX) + else "canonical_json_v2" + ) + return _owner_with_id_scheme(owner, scheme=scheme) def _backup_restore_approval_request( @@ -310,31 +489,83 @@ async def process_backup_restore_alertmanager_signal( source_event_fingerprint=source_event_fingerprint, source_started_at=source_started_at, ) - approval_storage_fingerprint = _approval_storage_fingerprint( - source_fingerprint + legacy_owner = _legacy_backup_restore_ingress_owner( + project_id=str(owner["project_id"]), + source_fingerprint=str(owner["source_fingerprint"]), + source_event_fingerprint=str(owner["source_event_fingerprint"]), + source_started_at=str(owner["source_started_at"]), ) approval_service = get_approval_service() - owner_approval_id = UUID(str(owner["approval_id"])) - approval = await approval_service.get_approval(owner_approval_id) - created_owner = approval is None - legacy_incident_id = "" - if approval is None: - recent = await _find_compatible_source_approval( - approval_service, - source_fingerprint=source_fingerprint, + approval = None + mismatched_owner_ids: set[str] = set() + owner_candidate_ids = tuple( + dict.fromkeys( + ( + _value(owner.get("approval_id")), + _value(owner.get("canonical_approval_id")), + _value(owner.get("legacy_approval_id")), + ) ) - legacy_incident_id = _value(getattr(recent, "incident_id", "")) + ) + for candidate_id in owner_candidate_ids: + candidate = await approval_service.get_approval(UUID(candidate_id)) + if candidate is None: + continue + if _approval_matches_owner_identity( + candidate, + owner=owner, + legacy_owner=legacy_owner, + ): + approval = candidate + owner = _owner_for_approval(owner, candidate) + break + mismatched_owner_ids.add(candidate_id) + logger.warning( + "backup_restore_deterministic_owner_identity_mismatch", + approval_id=candidate_id, + ) + + if approval is None: + recent = await _find_compatible_owner_approval( + approval_service, + owner=owner, + legacy_owner=legacy_owner, + ) + if recent is not None: + approval = recent + owner = _owner_for_approval(owner, recent) + + created_owner = False + if approval is None: + # A v1 delimiter collision may occupy the legacy UUID. New work uses + # the canonical JSON UUID; a mismatch on that UUID itself fails closed. + canonical_id = _value(owner.get("canonical_approval_id")) + if canonical_id in mismatched_owner_ids: + return { + "schema_version": "backup_restore_ingress_result_v1", + "status": "durable_owner_identity_mismatch_no_dispatch", + "owner": owner, + "dispatch_performed": False, + "runtime_execution_authorized": False, + "runtime_closure_verified": False, + } + if _value(owner.get("approval_id")) in mismatched_owner_ids: + owner = _owner_with_id_scheme( + owner, + scheme="canonical_json_v2", + ) + owner_approval_id = UUID(str(owner["approval_id"])) request = _backup_restore_approval_request( owner=owner, target_resource=target_resource, message=message, - incident_id=legacy_incident_id, ) try: approval = await approval_service.create_approval_with_fingerprint( request=request, - fingerprint=approval_storage_fingerprint, + fingerprint=_approval_owner_storage_fingerprint(owner), ) + created_owner = True except Exception: # A concurrent producer may have inserted the deterministic owner. # Read it back; never create another random owner or dispatch twice. @@ -342,11 +573,18 @@ async def process_backup_restore_alertmanager_signal( if approval is None: raise else: + owner_approval_id = UUID(str(owner["approval_id"])) incremented = await approval_service.increment_hit_count(owner_approval_id) if incremented is not None: approval = incremented - if _value(getattr(approval, "id", "")) != str(owner_approval_id): + if _value(getattr(approval, "id", "")) != str( + owner_approval_id + ) or not _approval_matches_owner_identity( + approval, + owner=owner, + legacy_owner=legacy_owner, + ): return { "schema_version": "backup_restore_ingress_result_v1", "status": "durable_owner_identity_mismatch_no_dispatch", @@ -384,6 +622,11 @@ async def process_backup_restore_alertmanager_signal( if ( bound_approval is None or _value(getattr(bound_approval, "incident_id", "")) != incident_id + or not _approval_matches_owner_identity( + bound_approval, + owner=owner, + legacy_owner=legacy_owner, + ) ): return { "schema_version": "backup_restore_ingress_result_v1", @@ -479,8 +722,7 @@ async def process_backup_restore_alertmanager_signal( dispatch_identity and readback_identity and all( - _value(dispatch_identity.get(field)) - == _value(readback_identity.get(field)) + _value(dispatch_identity.get(field)) == _value(readback_identity.get(field)) for field in ("incident_id", "run_id", "trace_id", "work_item_id") ) ) @@ -491,8 +733,7 @@ async def process_backup_restore_alertmanager_signal( receipt_persisted = bool(readback and identity_matches) verifier = ( dict(readback.get("verifier")) - if isinstance(readback, dict) - and isinstance(readback.get("verifier"), dict) + if isinstance(readback, dict) and isinstance(readback.get("verifier"), dict) else {} ) verifier_status = _value(verifier.get("status")) or "not_started" diff --git a/apps/api/tests/integration/test_backup_restore_prod_db_regressions.py b/apps/api/tests/integration/test_backup_restore_prod_db_regressions.py index 72d83838d..715b72a54 100644 --- a/apps/api/tests/integration/test_backup_restore_prod_db_regressions.py +++ b/apps/api/tests/integration/test_backup_restore_prod_db_regressions.py @@ -23,9 +23,7 @@ def _explicit_local_test_database_url() -> str: if not database_url: pytest.skip("BACKUP_RESTORE_TEST_DATABASE_URL is required for real PG tests") - parsed = urlsplit( - database_url.replace("postgresql+asyncpg://", "postgresql://", 1) - ) + parsed = urlsplit(database_url.replace("postgresql+asyncpg://", "postgresql://", 1)) if parsed.hostname not in {None, "", "localhost", "127.0.0.1", "::1"}: pytest.fail("backup/restore PG regression must target a local database") database_name = parsed.path.rsplit("/", 1)[-1].strip().lower() @@ -45,7 +43,8 @@ async def test_backfill_run_inserts_satisfy_no_default_cost_column() -> None: async with engine.connect() as connection: transaction = await connection.begin() try: - await connection.execute(text(""" + await connection.execute( + text(""" CREATE TEMP TABLE awooop_run_state ( run_id UUID PRIMARY KEY, project_id VARCHAR(64) NOT NULL, @@ -69,7 +68,8 @@ async def test_backfill_run_inserts_satisfy_no_default_cost_column() -> None: timeout_at TIMESTAMPTZ, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ) ON COMMIT DROP - """)) + """) + ) await connection.execute( backfill._CURSOR_INSERT_SQL, { @@ -77,9 +77,7 @@ async def test_backfill_run_inserts_satisfy_no_default_cost_column() -> None: "project_id": "awoooi", "agent_id": backfill.BACKFILL_CURSOR_AGENT_ID, "input_sha256": "a" * 64, - "error_detail": json.dumps( - backfill._initial_cursor("awoooi") - ), + "error_detail": json.dumps(backfill._initial_cursor("awoooi")), }, ) await connection.execute( @@ -127,19 +125,23 @@ async def test_backfill_run_inserts_satisfy_no_default_cost_column() -> None: assert finished.scalar_one() == item_run_id rows = ( - await connection.execute(text(""" + ( + await connection.execute( + text(""" SELECT run_id, cost_usd FROM awooop_run_state ORDER BY run_id - """)) - ).mappings().all() + """) + ) + ) + .mappings() + .all() + ) assert {row["run_id"] for row in rows} == { cursor_run_id, item_run_id, } - assert {row["cost_usd"] for row in rows} == { - Decimal("0.0000") - } + assert {row["cost_usd"] for row in rows} == {Decimal("0.0000")} finally: await transaction.rollback() finally: @@ -166,12 +168,14 @@ async def test_domain_separated_fingerprints_avoid_cross_domain_pg_collision() - async with engine.connect() as connection: transaction = await connection.begin() try: - await connection.execute(text(""" + await connection.execute( + text(""" CREATE TEMP TABLE approval_records ( source_identity TEXT NOT NULL, fingerprint VARCHAR(64) NOT NULL UNIQUE ) ON COMMIT DROP - """)) + """) + ) await connection.execute( text(""" INSERT INTO approval_records ( @@ -186,12 +190,18 @@ async def test_domain_separated_fingerprints_avoid_cross_domain_pg_collision() - }, ) persisted = ( - await connection.execute(text(""" + ( + await connection.execute( + text(""" SELECT source_identity, fingerprint FROM approval_records ORDER BY source_identity - """)) - ).mappings().all() + """) + ) + ) + .mappings() + .all() + ) assert len(persisted) == 2 assert {row["fingerprint"] for row in persisted} == { stored_long_fingerprint, @@ -204,6 +214,159 @@ async def test_domain_separated_fingerprints_avoid_cross_domain_pg_collision() - await engine.dispose() +@pytest.mark.asyncio +async def test_full_owner_pg_replay_does_not_reuse_resolved_prior_occurrence() -> None: + """A source-only v2 row cannot bind a later startsAt to its Incident.""" + + engine = create_async_engine(_explicit_local_test_database_url(), echo=False) + source_fingerprint = "legacy-backup-occurrence:" + ("a" * 64) + common = { + "project_id": "awoooi", + "source_fingerprint": source_fingerprint, + "source_event_fingerprint": "legacy-backup:message:hash", + } + previous_owner = ( + backup_restore_alertmanager_ingress.build_backup_restore_ingress_owner( + **common, + source_started_at="2026-07-11T10:00:00Z", + ) + ) + previous_legacy = ( + backup_restore_alertmanager_ingress._legacy_backup_restore_ingress_owner( + **common, + source_started_at="2026-07-11T10:00:00Z", + ) + ) + current_owner = ( + backup_restore_alertmanager_ingress.build_backup_restore_ingress_owner( + **common, + source_started_at="2026-07-12T10:00:00Z", + ) + ) + current_legacy = ( + backup_restore_alertmanager_ingress._legacy_backup_restore_ingress_owner( + **common, + source_started_at="2026-07-12T10:00:00Z", + ) + ) + source_only_v2 = backup_restore_alertmanager_ingress._approval_storage_fingerprint( + source_fingerprint + ) + current_v3 = ( + backup_restore_alertmanager_ingress._approval_owner_storage_fingerprint( + current_owner + ) + ) + assert previous_owner["approval_id"] == previous_legacy["approval_id"] + + try: + async with engine.connect() as connection: + transaction = await connection.begin() + try: + await connection.execute( + text(""" + CREATE TEMP TABLE approval_owner_replay ( + id UUID PRIMARY KEY, + fingerprint VARCHAR(64) NOT NULL, + incident_id TEXT NOT NULL, + status TEXT NOT NULL, + extra_metadata JSONB NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ) ON COMMIT DROP + """) + ) + await connection.execute( + text(""" + INSERT INTO approval_owner_replay ( + id, fingerprint, incident_id, status, extra_metadata + ) VALUES ( + :id, :fingerprint, :incident_id, 'resolved', + CAST(:extra_metadata AS JSONB) + ) + """), + { + "id": UUID(previous_owner["approval_id"]), + "fingerprint": source_only_v2, + "incident_id": previous_owner["incident_id"], + "extra_metadata": json.dumps(previous_legacy), + }, + ) + + async def find_by_fingerprint( + *, + fingerprint: str, + debounce_minutes: int, + ): + assert debounce_minutes == 30 + row = ( + ( + await connection.execute( + text(""" + SELECT id, incident_id, status, extra_metadata + FROM approval_owner_replay + WHERE fingerprint = :fingerprint + ORDER BY created_at DESC + LIMIT 1 + """), + {"fingerprint": fingerprint}, + ) + ) + .mappings() + .one_or_none() + ) + if row is None: + return None + return SimpleNamespace( + id=row["id"], + incident_id=row["incident_id"], + status=row["status"], + metadata=dict(row["extra_metadata"]), + ) + + service = SimpleNamespace(find_by_fingerprint=find_by_fingerprint) + incompatible = await ( + backup_restore_alertmanager_ingress._find_compatible_owner_approval( + service, + owner=current_owner, + legacy_owner=current_legacy, + ) + ) + assert incompatible is None + + await connection.execute( + text(""" + INSERT INTO approval_owner_replay ( + id, fingerprint, incident_id, status, extra_metadata + ) VALUES ( + :id, :fingerprint, :incident_id, 'approved', + CAST(:extra_metadata AS JSONB) + ) + """), + { + "id": UUID(current_owner["approval_id"]), + "fingerprint": current_v3, + "incident_id": current_owner["incident_id"], + "extra_metadata": json.dumps(current_owner), + }, + ) + replay = await ( + backup_restore_alertmanager_ingress._find_compatible_owner_approval( + service, + owner=current_owner, + legacy_owner=current_legacy, + ) + ) + assert replay is not None + assert str(replay.id) == current_owner["approval_id"] + assert replay.incident_id == current_owner["incident_id"] + assert replay.incident_id != previous_owner["incident_id"] + assert len(current_v3) == 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, @@ -217,9 +380,7 @@ async def test_direct_readback_rewrites_ssl_and_keeps_rls_context_in_transaction ): separator = "&" if "?" in sqlalchemy_url else "?" sqlalchemy_url = f"{sqlalchemy_url}{separator}ssl=disable" - direct_url = platform_operator_service._asyncpg_direct_database_url( - sqlalchemy_url - ) + direct_url = platform_operator_service._asyncpg_direct_database_url(sqlalchemy_url) direct_query = dict(parse_qsl(urlsplit(direct_url).query)) assert "ssl" not in direct_query assert direct_query["sslmode"] == "disable" @@ -271,16 +432,18 @@ async def test_direct_readback_rewrites_ssl_and_keeps_rls_context_in_transaction ) """, UUID("66cdb6ad-46a4-48f5-9d3b-b1ac9c0b2e92"), - json.dumps({ - "ai_automation_alert_card": { - "event_type": "backup_restore_escrow_signal", - "lane": "backup_restore_escrow_triage", - "target": "backup_restore", - "delivery_receipt_readback_required": True, - "runtime_write_gate_count": 0, - }, - "source_refs": {"fingerprints": ["backup-fixture"]}, - }), + json.dumps( + { + "ai_automation_alert_card": { + "event_type": "backup_restore_escrow_signal", + "lane": "backup_restore_escrow_triage", + "target": "backup_restore", + "delivery_receipt_readback_required": True, + "runtime_write_gate_count": 0, + }, + "source_refs": {"fingerprints": ["backup-fixture"]}, + } + ), ) original_connect = asyncpg.connect @@ -316,6 +479,4 @@ async def test_direct_readback_rewrites_ssl_and_keeps_rls_context_in_transaction assert summary["backup_restore_total"] == 1 assert len(rows) == 1 assert rows[0]["project_id"] == "awoooi" - assert rows[0]["alert_card"]["event_type"] == ( - "backup_restore_escrow_signal" - ) + assert rows[0]["alert_card"]["event_type"] == ("backup_restore_escrow_signal") diff --git a/apps/api/tests/test_backup_restore_alertmanager_ingress.py b/apps/api/tests/test_backup_restore_alertmanager_ingress.py index 8ea3f1f74..9b12ecc0b 100644 --- a/apps/api/tests/test_backup_restore_alertmanager_ingress.py +++ b/apps/api/tests/test_backup_restore_alertmanager_ingress.py @@ -78,6 +78,14 @@ def test_backup_ingress_owner_is_stable_per_source_occurrence() -> None: assert first == replay assert new_firing["approval_id"] != first["approval_id"] + assert first["schema_version"] == "backup_restore_ingress_owner_v2" + assert first["owner_key_encoding"] == "canonical_json_v1" + assert first["identity_id_scheme"] == "canonical_json_v2" + assert first["source_occurrence"] == { + "source_event_fingerprint": "native-fingerprint", + "source_started_at": "2026-07-11T10:00:00Z", + "fallback_source_fingerprint": "", + } assert first["route_id"] == "agent99:backup_health:BackupCheck" assert first["single_dispatch_owner"] == "alertmanager_raw_signal" assert first["telegram_role"] == "receipt_projection_only" @@ -85,6 +93,87 @@ def test_backup_ingress_owner_is_stable_per_source_occurrence() -> None: assert first["read_only"] is True +def test_backup_owner_canonical_framing_breaks_colon_and_pipe_collision() -> None: + left = ingress.build_backup_restore_ingress_owner( + project_id="alpha:beta", + source_fingerprint="gamma", + source_event_fingerprint="delta|epsilon", + source_started_at="zeta", + ) + right = ingress.build_backup_restore_ingress_owner( + project_id="alpha", + source_fingerprint="beta:gamma", + source_event_fingerprint="delta", + source_started_at="epsilon|zeta", + ) + left_legacy = ingress._legacy_backup_restore_ingress_owner( + project_id="alpha:beta", + source_fingerprint="gamma", + source_event_fingerprint="delta|epsilon", + source_started_at="zeta", + ) + right_legacy = ingress._legacy_backup_restore_ingress_owner( + project_id="alpha", + source_fingerprint="beta:gamma", + source_event_fingerprint="delta", + source_started_at="epsilon|zeta", + ) + + assert left_legacy["approval_id"] == right_legacy["approval_id"] + assert left["approval_id"] != right["approval_id"] + assert left["owner_key_sha256"] != right["owner_key_sha256"] + assert ingress._legacy_owner_reuse_is_unambiguous(left) is False + assert ingress._legacy_owner_reuse_is_unambiguous(right) is False + + +def test_backup_owner_canonical_framing_breaks_empty_component_collision() -> None: + event_only = ingress.build_backup_restore_ingress_owner( + project_id="awoooi", + source_fingerprint="source-fingerprint", + source_event_fingerprint="same-value", + source_started_at="", + ) + started_only = ingress.build_backup_restore_ingress_owner( + project_id="awoooi", + source_fingerprint="source-fingerprint", + source_event_fingerprint="", + source_started_at="same-value", + ) + + assert event_only["legacy_approval_id"] == started_only["legacy_approval_id"] + assert event_only["approval_id"] != started_only["approval_id"] + assert ingress._legacy_owner_reuse_is_unambiguous(event_only) is False + assert ingress._legacy_owner_reuse_is_unambiguous(started_only) is False + + +def test_backup_legacy_89_char_claim_keeps_existing_deterministic_ids() -> None: + source = "legacy-backup-occurrence:" + ("a" * 64) + kwargs = { + "project_id": "awoooi", + "source_fingerprint": source, + "source_event_fingerprint": "legacy-backup:message:hash", + "source_started_at": "2026-07-11T10:00:00Z", + } + owner = ingress.build_backup_restore_ingress_owner(**kwargs) + legacy = ingress._legacy_backup_restore_ingress_owner(**kwargs) + + assert len(source) == 89 + assert owner["schema_version"] == "backup_restore_ingress_owner_v2" + assert owner["identity_id_scheme"] == "legacy_colon_v1" + assert owner["legacy_identity_compatibility"] is True + assert owner["approval_id"] == legacy["approval_id"] + assert owner["incident_id"] == legacy["incident_id"] + assert owner["work_item_id"] == legacy["work_item_id"] + assert owner["approval_id"] == "0896722a-a4ea-574b-9c1a-c389b30c14ae" + assert owner["incident_id"] == "INC-BRR-544862858131AD18" + assert owner["work_item_id"] == ( + "backupcheck:awoooi:544862858131ad1803baa9fea9f27903" + ) + assert ingress._approval_owner_storage_fingerprint(owner) != ( + ingress._approval_storage_fingerprint(source) + ) + + def test_backup_approval_fingerprint_is_domain_separated_and_bounded() -> None: short = "source-fingerprint" long = "legacy-backup-occurrence:" + ("a" * 64) @@ -92,9 +181,7 @@ def test_backup_approval_fingerprint_is_domain_separated_and_bounded() -> None: stored_short = ingress._approval_storage_fingerprint(short) stored_long = ingress._approval_storage_fingerprint(long) - stored_deliberate_raw = ingress._approval_storage_fingerprint( - legacy_long_digest - ) + stored_deliberate_raw = ingress._approval_storage_fingerprint(legacy_long_digest) assert stored_short != short assert stored_long != legacy_long_digest @@ -102,21 +189,51 @@ def test_backup_approval_fingerprint_is_domain_separated_and_bounded() -> None: assert len(stored_short) == ingress.APPROVAL_FINGERPRINT_MAX_LENGTH assert len(stored_long) == ingress.APPROVAL_FINGERPRINT_MAX_LENGTH assert ingress._approval_storage_fingerprint(long) == stored_long - assert ingress._approval_fingerprint_lookup_candidates(short) == ( + short_owner = ingress.build_backup_restore_ingress_owner( + project_id="awoooi", + source_fingerprint=short, + source_event_fingerprint="short-occurrence", + ) + long_owner = ingress.build_backup_restore_ingress_owner( + project_id="awoooi", + source_fingerprint=long, + source_event_fingerprint="long-occurrence", + ) + short_candidates = ingress._approval_fingerprint_lookup_candidates(short_owner) + long_candidates = ingress._approval_fingerprint_lookup_candidates(long_owner) + assert short_candidates == ( + ingress._approval_owner_storage_fingerprint(short_owner), stored_short, short, ) - assert ingress._approval_fingerprint_lookup_candidates(long) == ( + assert long_candidates == ( + ingress._approval_owner_storage_fingerprint(long_owner), stored_long, legacy_long_digest, ) + assert short_candidates[0] != long_candidates[0] + assert len(short_candidates[0]) == ingress.APPROVAL_FINGERPRINT_MAX_LENGTH @pytest.mark.asyncio async def test_backup_approval_legacy_lookup_rejects_deliberate_collision() -> None: long = "legacy-backup-occurrence:" + ("a" * 64) deliberate_raw = hashlib.sha256(long.encode("utf-8")).hexdigest() - canonical, legacy = ingress._approval_fingerprint_lookup_candidates(long) + owner = ingress.build_backup_restore_ingress_owner( + project_id="awoooi", + source_fingerprint=long, + source_event_fingerprint="legacy-backup:message:hash", + source_started_at="2026-07-11T10:00:00Z", + ) + legacy_owner = ingress._legacy_backup_restore_ingress_owner( + project_id="awoooi", + source_fingerprint=long, + source_event_fingerprint="legacy-backup:message:hash", + source_started_at="2026-07-11T10:00:00Z", + ) + canonical, source_only_v2, legacy = ingress._approval_fingerprint_lookup_candidates( + owner + ) assert legacy == deliberate_raw wrong_source = SimpleNamespace( @@ -131,23 +248,38 @@ async def test_backup_approval_legacy_lookup_rejects_deliberate_collision() -> N assert debounce_minutes == 30 return wrong_source if fingerprint == legacy else None - result = await ingress._find_compatible_source_approval( + result = await ingress._find_compatible_owner_approval( SimpleNamespace(find_by_fingerprint=find_by_fingerprint), - source_fingerprint=long, + owner=owner, + legacy_owner=legacy_owner, ) assert result is None - assert calls == [canonical, legacy] + assert calls == [canonical, source_only_v2, legacy] @pytest.mark.asyncio async def test_backup_approval_legacy_89_char_lookup_remains_compatible() -> None: source = "legacy-backup-occurrence:" + ("a" * 64) - canonical, legacy = ingress._approval_fingerprint_lookup_candidates(source) + owner = ingress.build_backup_restore_ingress_owner( + project_id="awoooi", + source_fingerprint=source, + source_event_fingerprint="legacy-backup:message:hash", + source_started_at="2026-07-11T10:00:00Z", + ) + legacy_owner = ingress._legacy_backup_restore_ingress_owner( + project_id="awoooi", + source_fingerprint=source, + source_event_fingerprint="legacy-backup:message:hash", + source_started_at="2026-07-11T10:00:00Z", + ) + canonical, source_only_v2, legacy = ingress._approval_fingerprint_lookup_candidates( + owner + ) existing = SimpleNamespace( - id=UUID("22222222-2222-4222-8222-222222222222"), - incident_id="INC-LEGACY-89-CHAR", - metadata={"source_fingerprint": source}, + id=UUID(legacy_owner["approval_id"]), + incident_id=legacy_owner["incident_id"], + metadata=dict(legacy_owner), ) calls: list[str] = [] @@ -156,14 +288,51 @@ async def test_backup_approval_legacy_89_char_lookup_remains_compatible() -> Non assert debounce_minutes == 30 return existing if fingerprint == legacy else None - result = await ingress._find_compatible_source_approval( + result = await ingress._find_compatible_owner_approval( SimpleNamespace(find_by_fingerprint=find_by_fingerprint), - source_fingerprint=source, + owner=owner, + legacy_owner=legacy_owner, ) assert result is existing - assert result.incident_id == "INC-LEGACY-89-CHAR" - assert calls == [canonical, legacy] + assert result.incident_id == legacy_owner["incident_id"] + assert calls == [canonical, source_only_v2, legacy] + + +@pytest.mark.asyncio +async def test_backup_source_only_v2_lookup_reuses_only_full_legacy_owner() -> None: + source = "legacy-backup-occurrence:" + ("b" * 64) + kwargs = { + "project_id": "awoooi", + "source_fingerprint": source, + "source_event_fingerprint": "legacy-backup:message-two:hash", + "source_started_at": "2026-07-11T11:00:00Z", + } + owner = ingress.build_backup_restore_ingress_owner(**kwargs) + legacy_owner = ingress._legacy_backup_restore_ingress_owner(**kwargs) + canonical, source_only_v2, _legacy = ( + ingress._approval_fingerprint_lookup_candidates(owner) + ) + existing = SimpleNamespace( + id=UUID(legacy_owner["approval_id"]), + incident_id=legacy_owner["incident_id"], + metadata=dict(legacy_owner), + ) + calls: list[str] = [] + + async def find_by_fingerprint(*, fingerprint: str, debounce_minutes: int): + calls.append(fingerprint) + assert debounce_minutes == 30 + return existing if fingerprint == source_only_v2 else None + + result = await ingress._find_compatible_owner_approval( + SimpleNamespace(find_by_fingerprint=find_by_fingerprint), + owner=owner, + legacy_owner=legacy_owner, + ) + + assert result is existing + assert calls == [canonical, source_only_v2] @pytest.mark.asyncio @@ -207,9 +376,7 @@ async def test_alertmanager_type1_backup_queues_durable_owner_before_info_shortc "component": "backup_restore", "namespace": "awoooi-prod", }, - annotations={ - "summary": "backup freshness missing; readback required" - }, + annotations={"summary": "backup freshness missing; readback required"}, startsAt=datetime.now(UTC).isoformat(), fingerprint="native-backup-fingerprint", generatorURL="http://alertmanager/graph", @@ -249,23 +416,34 @@ class _FakeApprovalService: async def find_by_fingerprint(self, **_kwargs): self.calls.append("legacy_owner_read") - expected = ingress._approval_fingerprint_lookup_candidates( - str(_signal_kwargs()["source_fingerprint"]) + expected_owner = ingress.build_backup_restore_ingress_owner( + project_id=str(_signal_kwargs()["project_id"]), + source_fingerprint=str(_signal_kwargs()["source_fingerprint"]), + source_event_fingerprint=str(_signal_kwargs()["source_event_fingerprint"]), + source_started_at=str(_signal_kwargs()["source_started_at"]), ) + expected = ingress._approval_fingerprint_lookup_candidates(expected_owner) assert _kwargs["fingerprint"] in expected return None async def create_approval_with_fingerprint(self, *, request, fingerprint): self.calls.append("owner_create") self.created_request = request - assert fingerprint == ingress._approval_storage_fingerprint( - str(_signal_kwargs()["source_fingerprint"]) + expected_owner = ingress.build_backup_restore_ingress_owner( + project_id=str(_signal_kwargs()["project_id"]), + source_fingerprint=str(_signal_kwargs()["source_fingerprint"]), + source_event_fingerprint=str(_signal_kwargs()["source_event_fingerprint"]), + source_started_at=str(_signal_kwargs()["source_started_at"]), + ) + assert fingerprint == ingress._approval_owner_storage_fingerprint( + expected_owner ) self.approval = SimpleNamespace( id=UUID(request.metadata["preallocated_approval_id"]), incident_id=request.incident_id, hit_count=1, risk_level=request.risk_level, + metadata=dict(request.metadata), ) return self.approval @@ -307,12 +485,14 @@ async def test_backup_ingress_binds_canonical_incident_then_dispatches_and_reads 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", - "trace_id": "00-11111111111111111111111111111111-2222222222222222-01", - "work_item_id": kwargs["work_item_id"], - }) + identity.update( + { + "incident_id": incident_id, + "run_id": "12345678-1234-5678-9234-567812345678", + "trace_id": "00-11111111111111111111111111111111-2222222222222222-01", + "work_item_id": kwargs["work_item_id"], + } + ) return { "status": "dispatched", "dispatchPerformed": True, @@ -361,12 +541,8 @@ async def test_backup_ingress_binds_canonical_incident_then_dispatches_and_reads assert calls.index("owner_create") < calls.index("incident_create") assert calls.index("incident_create") < calls.index("incident_bind") assert calls.index("incident_bind") < calls.index("incident_canonical_readback") - assert calls.index("incident_canonical_readback") < calls.index( - "agent99_dispatch" - ) - assert calls.index("agent99_dispatch") < calls.index( - "dispatch_receipt_readback" - ) + assert calls.index("incident_canonical_readback") < calls.index("agent99_dispatch") + assert calls.index("agent99_dispatch") < calls.index("dispatch_receipt_readback") assert result["status"] == "backupcheck_dispatched_verifier_pending" assert result["receipt_persisted"] is True assert result["accepted"] is True @@ -379,9 +555,12 @@ async def test_backup_ingress_binds_canonical_incident_then_dispatches_and_reads assert calls.count("owner_create") == 1 assert calls.count("incident_create") == 1 assert calls.count("owner_recurrence") == 1 - assert approval_service.approval.fingerprint == hashlib.sha256( - str(_signal_kwargs()["source_fingerprint"]).encode("utf-8") - ).hexdigest() + assert ( + approval_service.approval.fingerprint + == hashlib.sha256( + str(_signal_kwargs()["source_fingerprint"]).encode("utf-8") + ).hexdigest() + ) assert approval_service.created_request.risk_level.value == "low" assert approval_service.created_request.metadata["telegram_role"] == ( "receipt_projection_only" @@ -389,6 +568,146 @@ async def test_backup_ingress_binds_canonical_incident_then_dispatches_and_reads assert "run_restore" in result["prohibited_actions"] +@pytest.mark.asyncio +async def test_same_source_new_occurrence_does_not_inherit_resolved_incident( + monkeypatch: pytest.MonkeyPatch, +) -> None: + previous_kwargs = _signal_kwargs() + current_kwargs = { + **previous_kwargs, + "alert_id": "backup-alert-new-occurrence", + "source_started_at": "2026-07-12T10:00:00Z", + } + previous_owner = ingress.build_backup_restore_ingress_owner( + project_id=str(previous_kwargs["project_id"]), + source_fingerprint=str(previous_kwargs["source_fingerprint"]), + source_event_fingerprint=str(previous_kwargs["source_event_fingerprint"]), + source_started_at=str(previous_kwargs["source_started_at"]), + ) + current_owner = ingress.build_backup_restore_ingress_owner( + project_id=str(current_kwargs["project_id"]), + source_fingerprint=str(current_kwargs["source_fingerprint"]), + source_event_fingerprint=str(current_kwargs["source_event_fingerprint"]), + source_started_at=str(current_kwargs["source_started_at"]), + ) + previous = SimpleNamespace( + id=UUID(previous_owner["approval_id"]), + incident_id=previous_owner["incident_id"], + hit_count=1, + metadata=dict(previous_owner), + status="approved", + ) + + class OccurrenceApprovalService: + def __init__(self) -> None: + self.created = None + self.created_request = None + + async def get_approval(self, approval_id: UUID): + if self.created is not None and self.created.id == approval_id: + return self.created + if previous.id == approval_id: + return previous + return None + + async def find_by_fingerprint( + self, + *, + fingerprint: str, + debounce_minutes: int, + ): + assert debounce_minutes == 30 + if fingerprint in { + ingress._approval_storage_fingerprint( + str(current_kwargs["source_fingerprint"]) + ), + hashlib.sha256( + str(current_kwargs["source_fingerprint"]).encode("utf-8") + ).hexdigest(), + }: + return previous + return None + + async def create_approval_with_fingerprint(self, *, request, fingerprint): + assert request.incident_id is None + assert fingerprint == ingress._approval_owner_storage_fingerprint( + current_owner + ) + self.created_request = request + self.created = SimpleNamespace( + id=UUID(request.metadata["preallocated_approval_id"]), + incident_id=None, + hit_count=1, + metadata=dict(request.metadata), + status="approved", + ) + return self.created + + async def increment_hit_count(self, _approval_id: UUID): + raise AssertionError("new occurrence must not increment prior owner") + + async def update_incident_id(self, approval_id: UUID, incident_id: str): + assert self.created is not None + assert approval_id == self.created.id + self.created.incident_id = incident_id + + approval_service = OccurrenceApprovalService() + + async def create_incident(**kwargs): + assert kwargs["approval_id"] == current_owner["approval_id"] + assert kwargs["canonical_incident_id"] == current_owner["incident_id"] + return str(kwargs["canonical_incident_id"]) + + async def canonical_readback(incident_id: str, *, project_id: str): + assert incident_id == current_owner["incident_id"] + assert incident_id != previous_owner["incident_id"] + assert project_id == "awoooi" + return SimpleNamespace(status="investigating", persisted_to_pg=True) + + identity = { + "incident_id": current_owner["incident_id"], + "run_id": "42345678-1234-5678-9234-567812345678", + "trace_id": "00-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-bbbbbbbbbbbbbbbb-01", + "work_item_id": current_owner["work_item_id"], + } + + async def dispatch(**kwargs): + assert kwargs["incident_id"] == current_owner["incident_id"] + assert kwargs["approval_id"] == current_owner["approval_id"] + return { + "status": "dispatched", + "dispatchPerformed": True, + "identity": identity, + } + + async def readback(**kwargs): + assert kwargs["incident_id"] == current_owner["incident_id"] + return { + "identity": identity, + "dispatch_receipt": {"accepted": True, "inbox_triggered": True}, + "verifier": {"status": "pending"}, + } + + monkeypatch.setattr(ingress, "get_approval_service", lambda: approval_service) + monkeypatch.setattr(ingress, "create_incident_for_approval", create_incident) + monkeypatch.setattr( + ingress, + "get_incident_service", + lambda: SimpleNamespace(get_for_readback=canonical_readback), + ) + monkeypatch.setattr(ingress, "bridge_alertmanager_to_agent99", dispatch) + monkeypatch.setattr(ingress, "read_agent99_dispatch_receipt", readback) + monkeypatch.setattr(ingress, "record_alertmanager_event", AsyncMock()) + + result = await ingress.process_backup_restore_alertmanager_signal(**current_kwargs) + + assert result["status"] == "backupcheck_dispatched_verifier_pending" + assert result["owner_created"] is True + assert result["incident_id"] == current_owner["incident_id"] + assert result["incident_id"] != previous_owner["incident_id"] + assert approval_service.created_request.incident_id is None + + class _FakeDispatchLedger: def __init__(self) -> None: self.receipt = None @@ -452,17 +771,19 @@ async def test_backup_bridge_postgres_identity_deduplicates_transport( ) monkeypatch.setattr( "src.services.agent99_sre_bridge.dispatch_agent99_sre_alert_with_receipt", - lambda payload: transported.append(payload) - or { - "schema_version": "agent99_sre_dispatch_receipt_v1", - "status": "accepted_inbox_triggered", - "transport": "relay", - "alert_id": str(payload["id"]), - "kind": str(payload["kind"]), - "accepted": True, - "inbox_triggered": True, - "delivery_certainty": "delivered", - }, + lambda payload: ( + transported.append(payload) + or { + "schema_version": "agent99_sre_dispatch_receipt_v1", + "status": "accepted_inbox_triggered", + "transport": "relay", + "alert_id": str(payload["id"]), + "kind": str(payload["kind"]), + "accepted": True, + "inbox_triggered": True, + "delivery_certainty": "delivered", + } + ), ) common = { "alertname": "BackupCredentialEscrowEvidenceMissing",