wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

fix(automation): reconcile new backup alert cards

Browse Source
This commit is contained in:
ogt
2026-07-15 04:59:24 +08:00
parent 1190c737c5
commit 098bdd81fe
2 changed files with 428 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

330
apps/api/src/jobs/backup_restore_legacy_backfill_job.py
View File

@@ -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)