fix(agent): replay failed checks after catalog drift

apps/api/src/services/awooop_ansible_check_mode_service.py

@@ -2174,6 +2174,138 @@ async def claim_stale_pending_check_modes(
     return claims
 
 
+async def claim_catalog_drift_failed_check_modes(
+    *,
+    project_id: str = "awoooi",
+    limit: int = 1,
+    candidate_max_age_hours: int | None = None,
+) -> list[AnsibleCheckModeClaim]:
+    """Replay a failed check once when its canonical check playbook changed."""
+
+    claims: list[AnsibleCheckModeClaim] = []
+    max_age_hours = (
+        candidate_max_age_hours
+        or settings.AWOOOP_ANSIBLE_CHECK_MODE_CANDIDATE_MAX_AGE_HOURS
+    )
+    async with get_db_context(project_id) as db:
+        result = await db.execute(
+            text("""
+                SELECT
+                    check_mode.op_id,
+                    check_mode.parent_op_id,
+                    coalesce(
+                        check_mode.input ->> 'incident_id',
+                        check_mode.incident_id::text
+                    ) AS incident_id,
+                    check_mode.input
+                FROM automation_operation_log check_mode
+                WHERE check_mode.operation_type = 'ansible_check_mode_executed'
+                  AND check_mode.status = 'failed'
+                  AND check_mode.created_at >= NOW() - (
+                      :candidate_max_age_hours * INTERVAL '1 hour'
+                  )
+                  AND NOT EXISTS (
+                      SELECT 1
+                      FROM automation_operation_log replay
+                      WHERE replay.input ->> 'replay_of_check_mode_op_id'
+                            = check_mode.op_id::text
+                  )
+                ORDER BY check_mode.created_at DESC
+                LIMIT :scan_limit
+                FOR UPDATE SKIP LOCKED
+            """),
+            {
+                "candidate_max_age_hours": max(1, max_age_hours),
+                "scan_limit": min(100, max(10, limit * 10)),
+            },
+        )
+        for row_value in result.mappings().all():
+            row = dict(row_value)
+            previous_input = _json_loads(row.get("input"))
+            previous_check_path = str(
+                previous_input.get("check_mode_playbook_path")
+                or previous_input.get("playbook_path")
+                or ""
+            )
+            canonical_claim = _claim_from_stale_check_mode_row(row)
+            if (
+                canonical_claim is None
+                or not previous_check_path
+                or canonical_claim.playbook_path == previous_check_path
+            ):
+                continue
+            replay_input = {
+                **canonical_claim.input_payload,
+                "catalog_drift_replay": True,
+                "replay_of_check_mode_op_id": str(row.get("op_id") or ""),
+                "previous_check_mode_playbook_path": previous_check_path,
+            }
+            inserted = await db.execute(
+                text("""
+                    INSERT INTO automation_operation_log (
+                        operation_type, actor, status, incident_id,
+                        input, output, dry_run_result,
+                        parent_op_id, tags
+                    ) VALUES (
+                        'ansible_check_mode_executed',
+                        'ansible_check_mode_worker',
+                        'pending',
+                        :incident_db_id,
+                        CAST(:input AS jsonb),
+                        '{}'::jsonb,
+                        CAST(:dry_run_result AS jsonb),
+                        CAST(:parent_op_id AS uuid),
+                        :tags
+                    )
+                    RETURNING op_id
+                """),
+                {
+                    "incident_db_id": _automation_operation_log_incident_id(
+                        canonical_claim.incident_id
+                    ),
+                    "input": json.dumps(replay_input, ensure_ascii=False),
+                    "dry_run_result": json.dumps(
+                        {
+                            "check_mode_executed": False,
+                            "apply_executed": False,
+                            "claim_state": "catalog_drift_replay",
+                            "controlled_apply_allowed": bool(
+                                replay_input.get("controlled_apply_allowed")
+                            ),
+                            "controlled_apply_blocker": replay_input.get(
+                                "controlled_apply_blocker"
+                            ),
+                        },
+                        ensure_ascii=False,
+                    ),
+                    "parent_op_id": canonical_claim.source_candidate_op_id,
+                    "tags": [
+                        "ansible",
+                        "check_mode",
+                        "pending",
+                        "catalog_drift_replay",
+                    ],
+                },
+            )
+            op_id = str(inserted.scalar_one())
+            claims.append(
+                AnsibleCheckModeClaim(
+                    op_id=op_id,
+                    source_candidate_op_id=canonical_claim.source_candidate_op_id,
+                    incident_id=canonical_claim.incident_id,
+                    catalog_id=canonical_claim.catalog_id,
+                    playbook_path=canonical_claim.playbook_path,
+                    apply_playbook_path=canonical_claim.apply_playbook_path,
+                    inventory_hosts=canonical_claim.inventory_hosts,
+                    risk_level=canonical_claim.risk_level,
+                    input_payload=replay_input,
+                )
+            )
+            if len(claims) >= max(1, limit):
+                break
+    return claims
+
+
 async def recent_ansible_transport_blockers(
     *,
     project_id: str = "awoooi",
@@ -2565,6 +2697,18 @@ async def run_pending_check_modes_once(
         stale_after_seconds=max(300, effective_timeout_seconds + 120),
     )
     remaining_limit = max(0, limit - len(reclaimed_claims))
+    catalog_replay_claims = (
+        await claim_catalog_drift_failed_check_modes(
+            project_id=project_id,
+            limit=remaining_limit,
+            candidate_max_age_hours=(
+                settings.AWOOOP_ANSIBLE_CHECK_MODE_CANDIDATE_MAX_AGE_HOURS
+            ),
+        )
+        if remaining_limit
+        else []
+    )
+    remaining_limit = max(0, remaining_limit - len(catalog_replay_claims))
     fresh_claims = (
         await claim_pending_check_modes(
             project_id=project_id,
@@ -2576,7 +2720,7 @@ async def run_pending_check_modes_once(
         if remaining_limit
         else []
     )
-    claims = [*reclaimed_claims, *fresh_claims]
+    claims = [*reclaimed_claims, *catalog_replay_claims, *fresh_claims]
     completed = 0
     failed = 0
     apply_completed = 0
@@ -2606,6 +2750,7 @@ async def run_pending_check_modes_once(
     return {
         "claimed": len(claims),
         "reclaimed": len(reclaimed_claims),
+        "catalog_replayed": len(catalog_replay_claims),
         "completed": completed,
         "failed": failed,
         "apply_completed": apply_completed,

apps/api/tests/test_awooop_truth_chain_service.py

@@ -39,6 +39,7 @@ from src.services.awooop_ansible_check_mode_service import (
     build_ansible_post_apply_runtime_stage_receipts,
     build_ansible_pre_apply_runtime_stage_receipts,
     build_ansible_timeline_runtime_stage_receipt,
+    claim_catalog_drift_failed_check_modes,
     claim_pending_check_modes,
     claim_stale_pending_check_modes,
     detect_ansible_transport_blockers,
@@ -1692,6 +1693,28 @@ def test_stale_check_mode_row_is_revalidated_with_canonical_run_id() -> None:
     assert claim.playbook_path == "infra/ansible/playbooks/188-momo-backup-user.yml"
 
 
+def test_failed_nginx_check_mode_revalidates_to_readonly_catalog_path() -> None:
+    source_candidate_op_id = "00000000-0000-0000-0000-000000000093"
+    claim = _claim_from_stale_check_mode_row({
+        "op_id": "00000000-0000-0000-0000-000000000094",
+        "parent_op_id": source_candidate_op_id,
+        "incident_id": "INC-20260710-F87642",
+        "input": {
+            "incident_id": "INC-20260710-F87642",
+            "source_candidate_op_id": source_candidate_op_id,
+            "catalog_id": "ansible:nginx-sync",
+            "catalog_playbook_path": "infra/ansible/playbooks/nginx-sync.yml",
+            "check_mode_playbook_path": "infra/ansible/playbooks/nginx-sync.yml",
+            "inventory_hosts": ["host_110", "host_188"],
+            "risk_level": "high",
+        },
+    })
+
+    assert claim is not None
+    assert claim.playbook_path == "infra/ansible/playbooks/nginx-sync-readonly.yml"
+    assert claim.apply_playbook_path == "infra/ansible/playbooks/nginx-sync.yml"
+
+
 def test_stale_check_mode_reclaim_uses_lease_lock_and_current_policy() -> None:
     source = inspect.getsource(claim_stale_pending_check_modes)
     run_source = inspect.getsource(run_pending_check_modes_once)
@@ -1705,6 +1728,16 @@ def test_stale_check_mode_reclaim_uses_lease_lock_and_current_policy() -> None:
     assert "effective_timeout_seconds + 120" in run_source
 
 
+def test_failed_check_mode_catalog_drift_replays_once() -> None:
+    source = inspect.getsource(claim_catalog_drift_failed_check_modes)
+    run_source = inspect.getsource(run_pending_check_modes_once)
+
+    assert "replay_of_check_mode_op_id" in source
+    assert "catalog_drift_replay" in source
+    assert "FOR UPDATE SKIP LOCKED" in source
+    assert "claim_catalog_drift_failed_check_modes" in run_source
+
+
 def test_ansible_apply_receipt_backfill_queries_existing_apply_rows() -> None:
     source = inspect.getsource(backfill_missing_auto_repair_execution_receipts_once)