diff --git a/apps/api/src/services/awooop_ansible_audit_service.py b/apps/api/src/services/awooop_ansible_audit_service.py index ea0311778..bbe4a6c3a 100644 --- a/apps/api/src/services/awooop_ansible_audit_service.py +++ b/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", 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 e6f9cdbef..4363333e5 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/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, diff --git a/apps/api/tests/test_awooop_ansible_post_verifier.py b/apps/api/tests/test_awooop_ansible_post_verifier.py index a85ad6310..b7f89bd73 100644 --- a/apps/api/tests/test_awooop_ansible_post_verifier.py +++ b/apps/api/tests/test_awooop_ansible_post_verifier.py @@ -69,7 +69,7 @@ def test_host110_node_exporter_postconditions_are_bounded() -> None: assert catalog["risk_level"] == "low" assert catalog["auto_apply_enabled"] is True assert catalog["catalog_revision"] == ( - "2026-07-14-node-exporter-bounded-v4" + "2026-07-14-node-exporter-bounded-v5" ) diff --git a/apps/api/tests/test_host110_node_exporter_ansible_catalog.py b/apps/api/tests/test_host110_node_exporter_ansible_catalog.py index b4be166a8..3999364ce 100644 --- a/apps/api/tests/test_host110_node_exporter_ansible_catalog.py +++ b/apps/api/tests/test_host110_node_exporter_ansible_catalog.py @@ -1,5 +1,7 @@ from __future__ import annotations +import inspect +import subprocess from pathlib import Path import yaml @@ -9,7 +11,10 @@ from src.services.awooop_ansible_audit_service import ( get_ansible_catalog_item, ) from src.services.awooop_ansible_check_mode_service import ( + _claim_from_failed_apply_catalog_repair_row, _claim_from_stale_check_mode_row, + claim_catalog_drift_failed_applies, + run_pending_check_modes_once, ) ROOT = Path(__file__).resolve().parents[3] @@ -48,7 +53,7 @@ def test_node_exporter_playbook_only_reconciles_one_container() -> None: assert catalog is not None assert catalog["playbook_path"].endswith("110-node-exporter-repair.yml") assert catalog["catalog_revision"] == ( - "2026-07-14-node-exporter-bounded-v4" + "2026-07-14-node-exporter-bounded-v5" ) assert "docker\n - compose\n - up" in text assert "node-exporter" in text @@ -63,6 +68,27 @@ def test_node_exporter_playbook_only_reconciles_one_container() -> None: assert forbidden not in text.lower() +def test_node_exporter_post_apply_prometheus_probe_is_valid_bash() -> None: + payload = yaml.safe_load(PLAYBOOK.read_text(encoding="utf-8")) + verifier = next( + task + for task in payload[0]["tasks"] + if task["name"] + == "Verify Prometheus sees node-exporter after controlled apply" + ) + command = verifier["ansible.builtin.shell"]["cmd"] + + result = subprocess.run( + ["bash", "-n"], + input=command, + text=True, + capture_output=True, + check=False, + ) + + assert result.returncode == 0, result.stderr + + def test_existing_110_devops_lineage_migrates_to_bounded_revision() -> None: claim = _claim_from_stale_check_mode_row( { @@ -90,6 +116,80 @@ def test_existing_110_devops_lineage_migrates_to_bounded_revision() -> None: assert claim.playbook_path.endswith("110-node-exporter-repair.yml") assert claim.apply_playbook_path.endswith("110-node-exporter-repair.yml") assert claim.input_payload["catalog_revision"] == ( - "2026-07-14-node-exporter-bounded-v4" + "2026-07-14-node-exporter-bounded-v5" ) assert claim.input_payload["controlled_apply_allowed"] is True + + +def test_failed_apply_catalog_repair_reenters_same_automation_run() -> None: + failed_apply_op_id = "00000000-0000-0000-0000-000000000201" + source_candidate_op_id = "00000000-0000-0000-0000-000000000110" + claim = _claim_from_failed_apply_catalog_repair_row( + { + "op_id": failed_apply_op_id, + "incident_id": "INC-NODE-EXPORTER-110", + "input": { + "automation_run_id": source_candidate_op_id, + "source_candidate_op_id": source_candidate_op_id, + "incident_id": "INC-NODE-EXPORTER-110", + "catalog_id": "ansible:110-devops", + "catalog_revision": ( + "2026-07-14-node-exporter-bounded-v4" + ), + "playbook_path": ( + "infra/ansible/playbooks/110-node-exporter-repair.yml" + ), + "inventory_hosts": ["host_110"], + "risk_level": "low", + }, + } + ) + + assert claim is not None + assert claim.source_candidate_op_id == source_candidate_op_id + assert claim.input_payload["automation_run_id"] == source_candidate_op_id + assert claim.input_payload["catalog_revision"] == ( + "2026-07-14-node-exporter-bounded-v5" + ) + assert claim.input_payload["catalog_repair_of_apply_op_id"] == ( + failed_apply_op_id + ) + assert claim.input_payload["controlled_apply_allowed"] is True + assert claim.input_payload["bounded_execution_scope"] == ( + "catalog_repair_single_apply" + ) + + +def test_failed_apply_current_revision_is_not_replayed() -> None: + claim = _claim_from_failed_apply_catalog_repair_row( + { + "op_id": "00000000-0000-0000-0000-000000000201", + "incident_id": "INC-NODE-EXPORTER-110", + "input": { + "automation_run_id": ( + "00000000-0000-0000-0000-000000000110" + ), + "source_candidate_op_id": ( + "00000000-0000-0000-0000-000000000110" + ), + "incident_id": "INC-NODE-EXPORTER-110", + "catalog_id": "ansible:110-devops", + "catalog_revision": ( + "2026-07-14-node-exporter-bounded-v5" + ), + }, + } + ) + + assert claim is None + + +def test_failed_apply_catalog_repair_is_deduped_and_prioritized() -> None: + claim_source = inspect.getsource(claim_catalog_drift_failed_applies) + worker_source = inspect.getsource(run_pending_check_modes_once) + + assert "catalog_repair_of_apply_op_id" in claim_source + assert "catalog_replay_revision" in claim_source + assert worker_source.index("claim_catalog_drift_failed_applies") < ( + worker_source.index("claim_catalog_drift_failed_check_modes") + ) diff --git a/infra/ansible/playbooks/110-node-exporter-repair.yml b/infra/ansible/playbooks/110-node-exporter-repair.yml index c4e607101..774e25497 100644 --- a/infra/ansible/playbooks/110-node-exporter-repair.yml +++ b/infra/ansible/playbooks/110-node-exporter-repair.yml @@ -88,11 +88,11 @@ - name: Verify Prometheus sees node-exporter after controlled apply ansible.builtin.shell: - cmd: >- + cmd: | for attempt in 1 2 3 4 5 6; do - curl -fsS --max-time 5 - 'http://127.0.0.1:9090/api/v1/query?query=up%7Bjob%3D%22node-exporter-110%22%7D' - | grep -Eq '"value":\[[^]]+,"1"\]' && exit 0; + if curl -fsS --max-time 5 'http://127.0.0.1:9090/api/v1/query?query=up%7Bjob%3D%22node-exporter-110%22%7D' | grep -Eq '"value":\[[^]]+,"1"\]'; then + exit 0 + fi sleep 5; done; exit 1