fix(automation): recover verified backup backfill failures

apps/api/src/jobs/backup_restore_legacy_backfill_job.py

@@ -43,6 +43,7 @@ BACKFILL_RESULT_SCHEMA_VERSION = "backup_restore_legacy_backfill_result_v1"
 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"
 BACKUP_RESTORE_EVENT_TYPE = "backup_restore_escrow_signal"
 BACKUP_RESTORE_LANE = "backup_restore_escrow_triage"
 PROJECTION_SCHEMA_VERSION = "telegram_agent99_dispatch_receipt_projection_v1"
@@ -582,6 +583,58 @@ _ITEM_STATE_COUNTS_SQL = text(
 """
 )
 
+_REQUEUE_REPAIRED_DBAPI_FAILURES_SQL = text(
+    """
+    WITH repairable AS (
+        SELECT item.run_id
+        FROM awooop_run_state item
+        WHERE item.project_id = :project_id
+          AND item.agent_id = :agent_id
+          AND item.state = 'failed'
+          AND item.error_code = 'E-BACKFILL-EXHAUSTED'
+          AND COALESCE(NULLIF(item.error_detail, ''), '{}')::jsonb #>>
+                '{last_result,reason}' = 'processor_error:DBAPIError'
+          AND COALESCE(NULLIF(item.error_detail, ''), '{}')::jsonb #>>
+                '{item,runtime_execution_authorized}' = 'false'
+          AND COALESCE(
+                COALESCE(NULLIF(item.error_detail, ''), '{}')::jsonb #>>
+                    '{item,recovery_generation}',
+                ''
+              ) <> :repair_generation
+          AND item.worker_id IS NULL
+          AND item.lease_until IS NULL
+        ORDER BY item.created_at ASC, item.run_id ASC
+        FOR UPDATE SKIP LOCKED
+        LIMIT :limit
+    )
+    UPDATE awooop_run_state item
+    SET state = 'waiting_tool',
+        attempt_count = 0,
+        max_attempts = GREATEST(item.max_attempts, 5),
+        worker_id = NULL,
+        lease_until = NULL,
+        heartbeat_at = NOW(),
+        completed_at = NULL,
+        error_code = NULL,
+        error_detail = JSONB_SET(
+            JSONB_SET(
+                COALESCE(NULLIF(item.error_detail, ''), '{}')::jsonb,
+                '{item,recovery_generation}',
+                TO_JSONB(CAST(:repair_generation AS TEXT)),
+                TRUE
+            ),
+            '{item,recovery_reason}',
+            TO_JSONB(CAST('verified_database_contract_repair' AS TEXT)),
+            TRUE
+        )
+    FROM repairable
+    WHERE item.project_id = :project_id
+      AND item.run_id = repairable.run_id
+      AND item.agent_id = :agent_id
+    RETURNING item.run_id
+"""
+)
+
 
 @dataclass(frozen=True)
 class LegacyBackupClaim:
@@ -1187,6 +1240,44 @@ async def _claim_legacy_backup_items(
     return claims
 
 
+async def _requeue_repaired_dbapi_failures(
+    *,
+    project_id: str,
+    limit: int,
+) -> int:
+    """Requeue one bounded historical DB-contract failure generation.
+
+    The original attempts predate the explicit ``cost_usd`` insert and typed
+    state binds.  Only those exact, no-runtime-write failures are eligible,
+    and the recovery generation is persisted inside the stable item envelope
+    so the same source row cannot be reset repeatedly.
+    """
+
+    bounded_limit = max(1, min(int(limit), MAX_SCOPE_ROWS))
+    async with get_db_context(project_id) as db:
+        result = await db.execute(
+            _REQUEUE_REPAIRED_DBAPI_FAILURES_SQL,
+            {
+                "project_id": project_id,
+                "agent_id": BACKFILL_ITEM_AGENT_ID,
+                "repair_generation": BACKFILL_DB_CONTRACT_REPAIR_GENERATION,
+                "limit": bounded_limit,
+            },
+        )
+        rows = result.mappings().all()
+    requeued = len(rows)
+    if requeued:
+        logger.info(
+            "backup_restore_legacy_backfill_db_contract_failures_requeued",
+            project_id=project_id,
+            repair_generation=BACKFILL_DB_CONTRACT_REPAIR_GENERATION,
+            requeued=requeued,
+            runtime_execution_authorized=False,
+            telegram_dispatch_performed=False,
+        )
+    return requeued
+
+
 async def _load_legacy_backup_card(
     *,
     project_id: str,
@@ -1876,6 +1967,8 @@ async def run_backup_restore_legacy_backfill_once(
         "failed": 0,
         "retried": 0,
         "lease_lost": 0,
+        "requeued_repaired_dbapi": 0,
+        "backfill_state_write_performed": False,
         "dispatch_total": 0,
         "source_total_at_start": None,
         "source_exhausted": False,
@@ -1892,6 +1985,13 @@ async def run_backup_restore_legacy_backfill_once(
         "error": None,
     }
     try:
+        if processor is process_backup_restore_alertmanager_signal:
+            requeued = await _requeue_repaired_dbapi_failures(
+                project_id=normalized_project,
+                limit=max_rows,
+            )
+            stats["requeued_repaired_dbapi"] = requeued
+            stats["backfill_state_write_performed"] = requeued > 0
         reserved = await _reserve_legacy_backup_page(
             project_id=normalized_project,
             scan_limit=scan_limit,