diff --git a/apps/api/src/services/awooop_ansible_check_mode_service.py b/apps/api/src/services/awooop_ansible_check_mode_service.py index 1c2a03143..79a0dcc52 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/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, diff --git a/apps/api/tests/test_awooop_truth_chain_service.py b/apps/api/tests/test_awooop_truth_chain_service.py index e99b61f11..ede54cd0f 100644 --- a/apps/api/tests/test_awooop_truth_chain_service.py +++ b/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)