From 098bdd81fe5f1d2c4d649f9c92dfef35fcc57bd8 Mon Sep 17 00:00:00 2001 From: ogt Date: Wed, 15 Jul 2026 04:59:24 +0800 Subject: [PATCH] fix(automation): reconcile new backup alert cards --- .../backup_restore_legacy_backfill_job.py | 330 +++++++++++++++++- ...test_backup_restore_legacy_backfill_job.py | 103 ++++++ 2 files changed, 428 insertions(+), 5 deletions(-) diff --git a/apps/api/src/jobs/backup_restore_legacy_backfill_job.py b/apps/api/src/jobs/backup_restore_legacy_backfill_job.py index 453b4cf0d..04060260c 100644 --- a/apps/api/src/jobs/backup_restore_legacy_backfill_job.py +++ b/apps/api/src/jobs/backup_restore_legacy_backfill_job.py @@ -44,6 +44,9 @@ BACKFILL_CURSOR_AGENT_ID = "backup_restore_legacy_backfill_cursor" BACKFILL_ITEM_AGENT_ID = "backup_restore_legacy_backfill" BACKFILL_IDEMPOTENCY_CHANNEL = "backup_restore_backfill" BACKFILL_DB_CONTRACT_REPAIR_GENERATION = "cost_usd_state_bind_v1" +LIVE_RECONCILE_SCHEMA_VERSION = "backup_restore_outbound_reconciler_v1" +LIVE_RECONCILE_AGENT_ID = "backup_restore_outbound_reconciler" +LIVE_RECONCILE_IDEMPOTENCY_CHANNEL = "backup_restore_outbound_reconcile" BACKUP_RESTORE_EVENT_TYPE = "backup_restore_escrow_signal" BACKUP_RESTORE_LANE = "backup_restore_escrow_triage" PROJECTION_SCHEMA_VERSION = "telegram_agent99_dispatch_receipt_projection_v1" @@ -252,6 +255,98 @@ _SOURCE_SNAPSHOT_SQL = text( """ ) +_LIVE_SOURCE_SCAN_SQL = text( + """ + SELECT + m.message_id, + m.queued_at, + COALESCE(m.source_envelope, '{}'::jsonb) #>> + '{source_refs,fingerprints,0}' AS source_fingerprint, + COALESCE(m.source_envelope, '{}'::jsonb) #>> + '{ai_automation_alert_card,event_type}' AS event_type, + COALESCE(m.source_envelope, '{}'::jsonb) #>> + '{ai_automation_alert_card,lane}' AS lane, + COALESCE( + NULLIF( + COALESCE(m.source_envelope, '{}'::jsonb) #>> + '{ai_automation_alert_card,target}', + '' + ), + 'backup_restore' + ) AS target + FROM awooop_outbound_message m + WHERE m.project_id = :project_id + AND m.channel_type = 'telegram' + AND COALESCE(m.source_envelope, '{}'::jsonb) #>> + '{ai_automation_alert_card,event_type}' = :event_type + AND COALESCE( + NULLIF( + COALESCE(m.source_envelope, '{}'::jsonb) #>> + '{ai_automation_alert_card,target}', + '' + ), + 'backup_restore' + ) = 'backup_restore' + AND ( + COALESCE(m.source_envelope, '{}'::jsonb) #>> + '{agent99_dispatch_receipt,schema_version}' + <> :projection_schema + OR COALESCE( + COALESCE(m.source_envelope, '{}'::jsonb) #>> + '{agent99_dispatch_receipt,receipt_persisted}', + '' + ) <> 'true' + OR COALESCE( + COALESCE(m.source_envelope, '{}'::jsonb) #>> + '{agent99_dispatch_receipt,accepted}', + '' + ) <> 'true' + OR COALESCE( + COALESCE(m.source_envelope, '{}'::jsonb) #>> + '{agent99_dispatch_receipt,inbox_triggered}', + '' + ) <> 'true' + OR NULLIF( + COALESCE(m.source_envelope, '{}'::jsonb) #>> + '{agent99_dispatch_receipt,run_id}', + '' + ) IS NULL + OR NULLIF( + COALESCE(m.source_envelope, '{}'::jsonb) #>> + '{agent99_dispatch_receipt,trace_id}', + '' + ) IS NULL + OR NULLIF( + COALESCE(m.source_envelope, '{}'::jsonb) #>> + '{agent99_dispatch_receipt,work_item_id}', + '' + ) IS NULL + OR COALESCE( + COALESCE(m.source_envelope, '{}'::jsonb) #>> + '{agent99_dispatch_receipt,kind}', + '' + ) <> 'backup_health' + OR COALESCE( + COALESCE(m.source_envelope, '{}'::jsonb) #>> + '{agent99_dispatch_receipt,suggested_mode}', + '' + ) <> 'BackupCheck' + ) + AND NOT EXISTS ( + SELECT 1 + FROM awooop_run_state item + WHERE item.project_id = m.project_id + AND item.agent_id IN (:backfill_agent_id, :live_agent_id) + AND item.trigger_ref IN ( + ('legacy-backup:' || CAST(m.message_id AS TEXT)), + ('backup-signal:' || CAST(m.message_id AS TEXT)) + ) + ) + ORDER BY m.queued_at ASC, m.message_id ASC + LIMIT :limit +""" +) + _ITEM_RUN_INSERT_SQL = text( """ INSERT INTO awooop_run_state ( @@ -260,7 +355,7 @@ _ITEM_RUN_INSERT_SQL = text( is_shadow, input_sha256, cost_usd, step_count, error_detail, timeout_at ) VALUES ( :run_id, :project_id, :agent_id, 'waiting_tool', - 0, 5, 'backfill_replay', :trigger_ref, + 0, 5, :trigger_type, :trigger_ref, FALSE, :input_sha256, 0.0000, 1, :error_detail, NOW() + INTERVAL '7 days' ) @@ -643,6 +738,7 @@ class LegacyBackupClaim: attempt_count: int max_attempts: int item: dict[str, Any] + agent_id: str = BACKFILL_ITEM_AGENT_ID @dataclass(frozen=True) @@ -1122,6 +1218,7 @@ async def _reserve_legacy_backup_page( "run_id": item_run_id, "project_id": project_id, "agent_id": BACKFILL_ITEM_AGENT_ID, + "trigger_type": "backfill_replay", "trigger_ref": (f"legacy-backup:{item['source_message_id']}")[:256], "input_sha256": _sha256(_stable_json(item)), "error_detail": _stable_json({"item": item}), @@ -1187,10 +1284,72 @@ async def _reserve_legacy_backup_page( } +async def _reserve_live_backup_cards( + *, + project_id: str, + limit: int, +) -> dict[str, int]: + """Reserve newly emitted incomplete backup cards without reopening history.""" + + bounded_limit = max(1, min(int(limit), MAX_BATCH_LIMIT)) + async with get_db_context(project_id) as db: + result = await db.execute( + _LIVE_SOURCE_SCAN_SQL, + { + "project_id": project_id, + "event_type": BACKUP_RESTORE_EVENT_TYPE, + "projection_schema": PROJECTION_SCHEMA_VERSION, + "backfill_agent_id": BACKFILL_ITEM_AGENT_ID, + "live_agent_id": LIVE_RECONCILE_AGENT_ID, + "limit": bounded_limit, + }, + ) + rows = [_row_mapping(row) for row in result.mappings().all()] + reserved = 0 + deduplicated = 0 + for row in rows: + item = _item_payload_from_row(project_id=project_id, row=row) + item["reconciliation_mode"] = "continuous_outbound" + item_run_id = UUID(str(item["backfill_run_id"])) + await db.execute( + _ITEM_RUN_INSERT_SQL, + { + "run_id": item_run_id, + "project_id": project_id, + "agent_id": LIVE_RECONCILE_AGENT_ID, + "trigger_type": "outbound_reconcile", + "trigger_ref": ( + f"backup-signal:{item['source_message_id']}" + )[:256], + "input_sha256": _sha256(_stable_json(item)), + "error_detail": _stable_json({"item": item}), + }, + ) + idempotency = await db.execute( + _ITEM_IDEMPOTENCY_INSERT_SQL, + { + "project_id": project_id, + "channel_type": LIVE_RECONCILE_IDEMPOTENCY_CHANNEL, + "provider_event_id": item["occurrence_id"], + "run_id": item_run_id, + }, + ) + if idempotency.scalar_one_or_none() is not None: + reserved += 1 + else: + deduplicated += 1 + return { + "scanned": len(rows), + "reserved": reserved, + "deduplicated": deduplicated, + } + + async def _claim_legacy_backup_items( *, project_id: str, limit: int, + agent_id: str = BACKFILL_ITEM_AGENT_ID, ) -> list[LegacyBackupClaim]: bounded_limit = max(1, min(int(limit), MAX_BATCH_LIMIT)) claim_token = str(uuid4()) @@ -1199,7 +1358,7 @@ async def _claim_legacy_backup_items( _EXHAUST_EXPIRED_CLAIMS_SQL, { "project_id": project_id, - "agent_id": BACKFILL_ITEM_AGENT_ID, + "agent_id": agent_id, }, ) exhausted_run_ids = exhausted_result.mappings().all() @@ -1214,7 +1373,7 @@ async def _claim_legacy_backup_items( _CLAIM_SQL, { "project_id": project_id, - "agent_id": BACKFILL_ITEM_AGENT_ID, + "agent_id": agent_id, "limit": bounded_limit, "claim_token": claim_token, "lease_seconds": CLAIM_LEASE_SECONDS, @@ -1235,6 +1394,7 @@ async def _claim_legacy_backup_items( attempt_count=int(mapping.get("attempt_count") or 0), max_attempts=int(mapping.get("max_attempts") or 0), item=dict(item), + agent_id=agent_id, ) ) return claims @@ -1463,6 +1623,14 @@ async def _persist_legacy_dispatch_projection( ) -> str: backfill_receipt = { "schema_version": BACKFILL_SCHEMA_VERSION, + "reconciliation_schema_version": ( + LIVE_RECONCILE_SCHEMA_VERSION + if claim.agent_id == LIVE_RECONCILE_AGENT_ID + else BACKFILL_SCHEMA_VERSION + ), + "reconciliation_mode": str( + claim.item.get("reconciliation_mode") or "legacy_snapshot" + ), "status": "dispatch_receipt_projected", "source_message_id": card.message_id, "source_fingerprint_hash": occurrence["source_fingerprint_hash"], @@ -1490,7 +1658,7 @@ async def _persist_legacy_dispatch_projection( "source_lane": str(claim.item.get("lane") or BACKUP_RESTORE_LANE), "source_target": str(claim.item.get("target") or "backup_restore"), "claim_run_id": claim.run_id, - "agent_id": BACKFILL_ITEM_AGENT_ID, + "agent_id": claim.agent_id, "claim_token": claim.claim_token, } async with get_db_context(project_id) as db: @@ -1575,7 +1743,7 @@ async def _finish_backfill_claim( "error_detail": _stable_json(envelope), "project_id": project_id, "run_id": claim.run_id, - "agent_id": BACKFILL_ITEM_AGENT_ID, + "agent_id": claim.agent_id, "claim_token": claim.claim_token, }, ) @@ -1737,6 +1905,11 @@ async def _replay_backfill_claim( occurrence=occurrence, relay_auth_mode=relay_auth_mode, ) + projection["source"] = ( + LIVE_RECONCILE_SCHEMA_VERSION + if claim.agent_id == LIVE_RECONCILE_AGENT_ID + else BACKFILL_SCHEMA_VERSION + ) projection_status = await _persist_legacy_dispatch_projection( project_id=project_id, claim=claim, @@ -2064,6 +2237,151 @@ async def run_backup_restore_legacy_backfill_once( return stats +async def _live_reconcile_state_counts(*, project_id: str) -> dict[str, int]: + async with get_db_context(project_id) as db: + result = await db.execute( + _ITEM_STATE_COUNTS_SQL, + { + "project_id": project_id, + "agent_id": LIVE_RECONCILE_AGENT_ID, + }, + ) + row = result.mappings().one_or_none() + mapping = _row_mapping(row) if row is not None else {} + return { + "item_total": int(mapping.get("item_total") or 0), + "remaining_total": int(mapping.get("remaining_total") or 0), + "completed_total": int(mapping.get("completed_total") or 0), + "failed_total": int(mapping.get("failed_total") or 0), + } + + +async def run_backup_restore_outbound_reconciler_once( + *, + project_id: str = "awoooi", + limit: int = MAX_BATCH_LIMIT, + processor: Processor = process_backup_restore_alertmanager_signal, +) -> dict[str, Any]: + """Reconcile newly emitted backup cards in bounded continuous batches.""" + + normalized_project = str(project_id or "").strip() + if not normalized_project: + return { + "schema_version": LIVE_RECONCILE_SCHEMA_VERSION, + "status": "project_context_missing_fail_closed", + "runtime_execution_authorized": False, + "telegram_dispatch_performed": False, + } + relay_auth_mode = "injected_processor" + relay_preflight: dict[str, Any] | None = None + if processor is process_backup_restore_alertmanager_signal: + relay_preflight = build_backup_restore_backfill_relay_preflight() + if relay_preflight["ready"] is not True: + return { + "schema_version": LIVE_RECONCILE_SCHEMA_VERSION, + "status": "relay_preflight_failed_fail_closed", + "relay_preflight": relay_preflight, + "runtime_execution_authorized": False, + "telegram_dispatch_performed": False, + } + relay_auth_mode = str(relay_preflight["auth_mode"]) + stats: dict[str, Any] = { + "schema_version": LIVE_RECONCILE_SCHEMA_VERSION, + "project_id": normalized_project, + "status": "in_progress", + "auth_mode": relay_auth_mode, + "scanned": 0, + "reserved": 0, + "claimed": 0, + "completed": 0, + "deduplicated": 0, + "failed": 0, + "retried": 0, + "lease_lost": 0, + "dispatch_total": 0, + "runtime_execution_authorized": False, + "telegram_dispatch_performed": False, + "production_write_performed": False, + "prohibited_actions": list(BACKUP_RESTORE_PROHIBITED_ACTIONS), + "error": None, + } + try: + reserved = await _reserve_live_backup_cards( + project_id=normalized_project, + limit=limit, + ) + stats.update(reserved) + claims = await _claim_legacy_backup_items( + project_id=normalized_project, + limit=limit, + agent_id=LIVE_RECONCILE_AGENT_ID, + ) + stats["claimed"] = len(claims) + for claim in claims: + outcome = await _replay_backfill_claim( + project_id=normalized_project, + claim=claim, + processor=processor, + relay_auth_mode=relay_auth_mode, + ) + if outcome == "completed": + stats["completed"] += 1 + stats["dispatch_total"] += 1 + elif outcome == "deduplicated": + stats["deduplicated"] += 1 + elif outcome == "retried": + stats["retried"] += 1 + elif outcome == "lease_lost": + stats["lease_lost"] += 1 + else: + stats["failed"] += 1 + counts = await _live_reconcile_state_counts(project_id=normalized_project) + stats.update(counts) + if counts["failed_total"] > 0: + stats["status"] = "blocked_failed_items_fail_closed" + elif counts["remaining_total"] > 0: + stats["status"] = "in_progress" + elif stats["scanned"] or stats["claimed"]: + stats["status"] = "reconciled" + else: + stats["status"] = "idle" + except Exception as exc: + stats["status"] = "database_or_worker_failure_fail_closed" + stats["error"] = type(exc).__name__ + stats["error_sqlstate"] = _database_sqlstate(exc) + logger.warning( + "backup_restore_outbound_reconciler_tick_failed", + project_id=normalized_project, + error_type=type(exc).__name__, + error_sqlstate=stats["error_sqlstate"], + runtime_execution_authorized=False, + telegram_dispatch_performed=False, + ) + logger.info("backup_restore_outbound_reconciler_tick", **stats) + return stats + + +async def run_backup_restore_outbound_reconciler_loop() -> None: + """Continuously reconcile new backup cards after the legacy cohort.""" + + while True: + sleep_seconds = float(settings.BACKUP_RESTORE_LEGACY_BACKFILL_INTERVAL_SECONDS) + try: + await run_backup_restore_outbound_reconciler_once( + project_id="awoooi", + limit=settings.BACKUP_RESTORE_LEGACY_BACKFILL_BATCH_LIMIT, + ) + except asyncio.CancelledError: + raise + except Exception as exc: + logger.warning( + "backup_restore_outbound_reconciler_loop_failed", + error_type=type(exc).__name__, + ) + sleep_seconds = max(sleep_seconds, 60.0) + await asyncio.sleep(sleep_seconds) + + async def run_backup_restore_legacy_backfill_loop() -> None: """Drain the historical backlog in bounded, restart-safe batches.""" @@ -2103,6 +2421,7 @@ async def run_backup_restore_legacy_backfill_loop() -> None: source_total_at_start=result.get("source_total_at_start"), telegram_dispatch_performed=False, ) + await run_backup_restore_outbound_reconciler_loop() return if result.get("status") in { "blocked_scope_cap", @@ -2115,6 +2434,7 @@ async def run_backup_restore_legacy_backfill_loop() -> None: status=result.get("status"), telegram_dispatch_performed=False, ) + await run_backup_restore_outbound_reconciler_loop() return if result.get("error"): sleep_seconds = max(sleep_seconds, 60.0) diff --git a/apps/api/tests/test_backup_restore_legacy_backfill_job.py b/apps/api/tests/test_backup_restore_legacy_backfill_job.py index 178c1e321..88aee8c88 100644 --- a/apps/api/tests/test_backup_restore_legacy_backfill_job.py +++ b/apps/api/tests/test_backup_restore_legacy_backfill_job.py @@ -527,6 +527,102 @@ async def test_db_contract_repair_requeues_only_one_bounded_generation( assert "attempt_count = 0" in sql +@pytest.mark.asyncio +async def test_live_reconciler_reserves_only_unowned_incomplete_cards( + monkeypatch: pytest.MonkeyPatch, +) -> None: + rows = [_source_row(201), _source_row(202)] + inserted_agents: list[str] = [] + trigger_refs: list[str] = [] + + class FakeDb: + async def execute( + self, + statement: Any, + params: dict[str, Any], + ) -> _Result: + if statement is job._LIVE_SOURCE_SCAN_SQL: + assert params["project_id"] == "awoooi" + assert params["backfill_agent_id"] == job.BACKFILL_ITEM_AGENT_ID + assert params["live_agent_id"] == job.LIVE_RECONCILE_AGENT_ID + assert params["limit"] == job.MAX_BATCH_LIMIT + return _Result(rows=rows) + if statement is job._ITEM_RUN_INSERT_SQL: + inserted_agents.append(str(params["agent_id"])) + trigger_refs.append(str(params["trigger_ref"])) + return _Result() + if statement is job._ITEM_IDEMPOTENCY_INSERT_SQL: + assert params["channel_type"] == job.LIVE_RECONCILE_IDEMPOTENCY_CHANNEL + return _Result(value=params["run_id"]) + raise AssertionError(f"unexpected statement: {statement}") + + monkeypatch.setattr( + job, + "get_db_context", + lambda _project_id: _DbContext(FakeDb()), + ) + + result = await job._reserve_live_backup_cards( + project_id="awoooi", + limit=job.MAX_BATCH_LIMIT + 10, + ) + + assert result == {"scanned": 2, "reserved": 2, "deduplicated": 0} + assert inserted_agents == [job.LIVE_RECONCILE_AGENT_ID] * 2 + assert all(value.startswith("backup-signal:") for value in trigger_refs) + sql = str(job._LIVE_SOURCE_SCAN_SQL) + assert "NOT EXISTS" in sql + assert "legacy-backup:" in sql + assert "backup-signal:" in sql + + +@pytest.mark.asyncio +async def test_live_reconciler_projects_bounded_agent99_receipt( + monkeypatch: pytest.MonkeyPatch, +) -> None: + claim = job.LegacyBackupClaim( + **{ + **_claim_from_row(203).__dict__, + "agent_id": job.LIVE_RECONCILE_AGENT_ID, + } + ) + reserve = AsyncMock( + return_value={"scanned": 1, "reserved": 1, "deduplicated": 0} + ) + claims = AsyncMock(return_value=[claim]) + replay = AsyncMock(return_value="completed") + counts = AsyncMock( + return_value={ + "item_total": 1, + "remaining_total": 0, + "completed_total": 1, + "failed_total": 0, + } + ) + processor = AsyncMock() + monkeypatch.setattr(job, "_reserve_live_backup_cards", reserve) + monkeypatch.setattr(job, "_claim_legacy_backup_items", claims) + monkeypatch.setattr(job, "_replay_backfill_claim", replay) + monkeypatch.setattr(job, "_live_reconcile_state_counts", counts) + + result = await job.run_backup_restore_outbound_reconciler_once( + project_id="awoooi", + limit=7, + processor=processor, + ) + + assert result["status"] == "reconciled" + assert result["dispatch_total"] == 1 + assert result["runtime_execution_authorized"] is False + assert result["telegram_dispatch_performed"] is False + claims.assert_awaited_once_with( + project_id="awoooi", + limit=7, + agent_id=job.LIVE_RECONCILE_AGENT_ID, + ) + replay.assert_awaited_once() + + @pytest.mark.asyncio @pytest.mark.parametrize( ("item_total", "completed_total", "expected_status", "expected_cursor_state"), @@ -1437,6 +1533,7 @@ async def test_one_time_loop_exits_after_durable_snapshot_completes( } ) sleep = AsyncMock() + live_loop = AsyncMock() monkeypatch.setattr(job.settings, "ENABLE_BACKUP_RESTORE_LEGACY_BACKFILL", True) monkeypatch.setattr( job.settings, @@ -1450,11 +1547,17 @@ async def test_one_time_loop_exits_after_durable_snapshot_completes( 0.0, ) monkeypatch.setattr(job, "run_backup_restore_legacy_backfill_once", tick) + monkeypatch.setattr( + job, + "run_backup_restore_outbound_reconciler_loop", + live_loop, + ) monkeypatch.setattr(job.asyncio, "sleep", sleep) await job.run_backup_restore_legacy_backfill_loop() tick.assert_awaited_once() + live_loop.assert_awaited_once() sleep.assert_awaited_once_with(0.0)