fix(agent): replay repaired failed applies

apps/api/src/services/awooop_ansible_audit_service.py

@@ -445,7 +445,7 @@ _CATALOG: tuple[dict[str, Any], ...] = (
         "auto_apply_enabled": True,
         "approval_required": False,
         "risk_level": "low",
-        "catalog_revision": "2026-07-14-node-exporter-bounded-v4",
+        "catalog_revision": "2026-07-14-node-exporter-bounded-v5",
     },
     {
         "catalog_id": "ansible:110-devops-full-convergence",

apps/api/src/services/awooop_ansible_check_mode_service.py

@@ -1119,6 +1119,107 @@ def _claim_from_stale_check_mode_row(
     )
 
 
+def _claim_from_failed_apply_catalog_repair_row(
+    row: dict[str, Any],
+) -> AnsibleCheckModeClaim | None:
+    """Build a new bounded claim when a failed apply's catalog was repaired."""
+
+    input_payload = _json_loads(row.get("input"))
+    apply_op_id = str(row.get("op_id") or "")
+    source_candidate_op_id = str(
+        input_payload.get("source_candidate_op_id") or ""
+    )
+    incident_id = str(
+        row.get("incident_id")
+        or input_payload.get("incident_id")
+        or ""
+    )
+    catalog_id = str(input_payload.get("catalog_id") or "")
+    catalog_item = get_ansible_catalog_item(catalog_id) or {}
+    current_revision = str(catalog_item.get("catalog_revision") or "")
+    previous_revision = str(input_payload.get("catalog_revision") or "")
+    current_playbook_path = str(catalog_item.get("playbook_path") or "")
+    inventory_hosts = catalog_item.get("inventory_hosts") or []
+    if not (
+        apply_op_id
+        and source_candidate_op_id
+        and incident_id
+        and catalog_id
+        and current_revision
+        and current_revision != previous_revision
+        and current_playbook_path
+        and isinstance(inventory_hosts, list)
+        and inventory_hosts
+    ):
+        return None
+
+    try:
+        claim_input = build_ansible_check_mode_claim_input(
+            source_candidate_op_id=source_candidate_op_id,
+            candidate_input={
+                "incident_id": incident_id,
+                "approval_id": input_payload.get("approval_id"),
+                "proposal_risk_level": input_payload.get(
+                    "proposal_risk_level"
+                ),
+                "decision_path": "failed_apply_catalog_repair_replay",
+                "router_source_sha": input_payload.get("router_source_sha"),
+                "target_selector": input_payload.get("target_selector") or {},
+                "source_truth_diff": input_payload.get("source_truth_diff")
+                or {},
+                "risk_policy_decision": input_payload.get(
+                    "risk_policy_decision"
+                )
+                or {},
+                "executor_candidates": [
+                    {
+                        "catalog_id": catalog_id,
+                        "catalog_revision": current_revision,
+                        "playbook_path": current_playbook_path,
+                        "check_mode_playbook_path": str(
+                            catalog_item.get("check_mode_playbook_path")
+                            or current_playbook_path
+                        ),
+                        "inventory_hosts": inventory_hosts,
+                        "risk_level": catalog_item.get("risk_level"),
+                    }
+                ],
+            },
+        )
+    except ValueError:
+        return None
+
+    automation_run_id = str(
+        input_payload.get("automation_run_id")
+        or source_candidate_op_id
+    )
+    claim_input.update(
+        {
+            "automation_run_id": automation_run_id,
+            "catalog_repair_of_apply_op_id": apply_op_id,
+            "previous_catalog_revision": previous_revision or None,
+            "catalog_replay_revision": current_revision,
+            "apply_idempotency_key": (
+                f"ansible-repair-apply:{apply_op_id}:{current_revision}"
+            ),
+            "bounded_execution_scope": "catalog_repair_single_apply",
+        }
+    )
+    return AnsibleCheckModeClaim(
+        op_id="",
+        source_candidate_op_id=source_candidate_op_id,
+        incident_id=incident_id,
+        catalog_id=str(claim_input["catalog_id"]),
+        playbook_path=str(claim_input["playbook_path"]),
+        apply_playbook_path=str(claim_input["apply_playbook_path"]),
+        inventory_hosts=tuple(
+            str(host) for host in claim_input["inventory_hosts"]
+        ),
+        risk_level=str(claim_input.get("risk_level") or ""),
+        input_payload=claim_input,
+    )
+
+
 async def _record_auto_repair_execution_receipt(
     claim: AnsibleCheckModeClaim,
     result: AnsibleRunResult,
@@ -7715,6 +7816,129 @@ async def claim_catalog_drift_failed_check_modes(
     return claims
 
 
+async def claim_catalog_drift_failed_applies(
+    *,
+    project_id: str = "awoooi",
+    limit: int = 1,
+    candidate_max_age_hours: int | None = None,
+) -> list[AnsibleCheckModeClaim]:
+    """Re-enter controlled execution once for a repaired failed apply."""
+
+    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
+                    apply.op_id,
+                    apply.parent_op_id,
+                    coalesce(
+                        apply.input ->> 'incident_id',
+                        apply.incident_id::text
+                    ) AS incident_id,
+                    apply.input
+                FROM automation_operation_log apply
+                WHERE apply.operation_type = 'ansible_apply_executed'
+                  AND apply.status = 'failed'
+                  AND apply.created_at >= NOW() - (
+                      :candidate_max_age_hours * INTERVAL '1 hour'
+                  )
+                ORDER BY apply.created_at DESC
+                LIMIT :scan_limit
+                FOR UPDATE OF apply 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():
+            canonical_claim = _claim_from_failed_apply_catalog_repair_row(
+                dict(row_value)
+            )
+            if canonical_claim is None:
+                continue
+            replay_input = canonical_claim.input_payload
+            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
+                    ) SELECT
+                        '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
+                    WHERE NOT EXISTS (
+                        SELECT 1
+                        FROM automation_operation_log existing_replay
+                        WHERE existing_replay.operation_type
+                                = 'ansible_check_mode_executed'
+                          AND existing_replay.input
+                                ->> 'catalog_repair_of_apply_op_id'
+                                = :failed_apply_op_id
+                          AND existing_replay.input
+                                ->> 'catalog_replay_revision'
+                                = :catalog_replay_revision
+                    )
+                    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": "failed_apply_catalog_repair",
+                            "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
+                    ),
+                    "failed_apply_op_id": replay_input[
+                        "catalog_repair_of_apply_op_id"
+                    ],
+                    "catalog_replay_revision": replay_input[
+                        "catalog_replay_revision"
+                    ],
+                    "tags": [
+                        "ansible",
+                        "check_mode",
+                        "pending",
+                        "failed_apply_catalog_repair",
+                    ],
+                },
+            )
+            op_id_value = inserted.scalar_one_or_none()
+            if op_id_value is None:
+                continue
+            claims.append(
+                replace(canonical_claim, op_id=str(op_id_value))
+            )
+            if len(claims) >= max(1, limit):
+                break
+    return claims
+
+
 async def recent_ansible_transport_blockers(
     *,
     project_id: str = "awoooi",
@@ -8492,6 +8716,31 @@ async def run_pending_check_modes_once(
         0,
         remaining_limit - len(stdin_boundary_replay_claims),
     )
+    failed_apply_catalog_replay_claims: list[AnsibleCheckModeClaim] = []
+    failed_apply_catalog_replay_error: str | None = None
+    if remaining_limit:
+        try:
+            failed_apply_catalog_replay_claims = (
+                await claim_catalog_drift_failed_applies(
+                    project_id=project_id,
+                    limit=remaining_limit,
+                    candidate_max_age_hours=(
+                        settings.AWOOOP_ANSIBLE_CHECK_MODE_CANDIDATE_MAX_AGE_HOURS
+                    ),
+                )
+            )
+        except Exception as exc:
+            failed_apply_catalog_replay_error = type(exc).__name__
+            logger.warning(
+                "ansible_failed_apply_catalog_repair_claim_failed",
+                project_id=project_id,
+                error_type=failed_apply_catalog_replay_error,
+                fresh_candidate_claim_continues=True,
+            )
+    remaining_limit = max(
+        0,
+        remaining_limit - len(failed_apply_catalog_replay_claims),
+    )
     catalog_replay_claims: list[AnsibleCheckModeClaim] = []
     catalog_replay_error: str | None = None
     if remaining_limit:
@@ -8526,6 +8775,7 @@ async def run_pending_check_modes_once(
     claims = [
         *reclaimed_claims,
         *stdin_boundary_replay_claims,
+        *failed_apply_catalog_replay_claims,
         *catalog_replay_claims,
         *fresh_claims,
     ]
@@ -8674,6 +8924,12 @@ async def run_pending_check_modes_once(
             stdin_boundary_replay_terminal_written
         ),
         "stdin_boundary_replay_error": stdin_boundary_replay_error,
+        "failed_apply_catalog_replayed": len(
+            failed_apply_catalog_replay_claims
+        ),
+        "failed_apply_catalog_replay_error": (
+            failed_apply_catalog_replay_error
+        ),
         "catalog_replayed": len(catalog_replay_claims),
         "completed": completed,
         "failed": failed,