diff --git a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py b/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py index 7307b8fee..c49ea52a4 100644 --- a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py +++ b/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py @@ -1,9 +1,9 @@ """AwoooP Ansible candidate backfill worker. This worker closes the gap between "AI found an allowlisted PlayBook candidate" -and "the check-mode worker has a durable AOL row to claim". It does not execute -host writes by itself; it only writes ``ansible_candidate_matched`` rows for -recent unresolved incidents that already match the static Ansible catalog. +and "the check-mode worker has a durable AOL row to claim". It also runs a +bounded ``--check --diff`` replay for a failed controlled apply. It never retries +the runtime apply by itself. """ from __future__ import annotations @@ -23,6 +23,7 @@ from src.services.awooop_ansible_audit_service import ( ) from src.services.awooop_ansible_check_mode_service import ( backfill_missing_auto_repair_execution_receipts_once, + run_failed_apply_check_mode_replay_once, ) logger = structlog.get_logger(__name__) @@ -101,7 +102,12 @@ async def enqueue_missing_ansible_candidates_once( limit: int | None = None, window_hours: int | None = None, recorder: Recorder = record_ansible_decision_audit, - receipt_backfiller: Callable[..., Awaitable[dict[str, Any]]] = backfill_missing_auto_repair_execution_receipts_once, + receipt_backfiller: Callable[ + ..., Awaitable[dict[str, Any]] + ] = backfill_missing_auto_repair_execution_receipts_once, + retry_replayer: Callable[ + ..., Awaitable[dict[str, Any]] + ] = run_failed_apply_check_mode_replay_once, ) -> dict[str, Any]: """Backfill missing Ansible candidate rows for recent unresolved incidents.""" @@ -112,6 +118,10 @@ async def enqueue_missing_ansible_candidates_once( "queued": 0, "already_existing_or_write_skipped": 0, "no_catalog_candidate": 0, + "repair_receipts_backfilled": 0, + "failed_apply_retry_replayed": 0, + "failed_apply_retry_check_mode_passed": 0, + "failed_apply_retry_check_mode_failed": 0, "error": None, } @@ -128,6 +138,9 @@ async def enqueue_missing_ansible_candidates_once( "already_existing_or_write_skipped": 0, "no_catalog_candidate": 0, "repair_receipts_backfilled": 0, + "failed_apply_retry_replayed": 0, + "failed_apply_retry_check_mode_passed": 0, + "failed_apply_retry_check_mode_failed": 0, "error": None, } @@ -168,6 +181,21 @@ async def enqueue_missing_ansible_candidates_once( stats["repair_receipts_backfilled"] = int(receipt_stats.get("written") or 0) if receipt_stats.get("error") and not stats["error"]: stats["error"] = receipt_stats["error"] + retry_stats = await retry_replayer( + project_id=project_id, + window_hours=bounded_window_hours, + ) + stats["failed_apply_retry_replayed"] = int( + retry_stats.get("replayed") or 0 + ) + stats["failed_apply_retry_check_mode_passed"] = int( + retry_stats.get("check_mode_passed") or 0 + ) + stats["failed_apply_retry_check_mode_failed"] = int( + retry_stats.get("check_mode_failed") or 0 + ) + if retry_stats.get("error") and not stats["error"]: + stats["error"] = retry_stats["error"] except Exception as exc: stats["error"] = f"{type(exc).__name__}: {exc}"[:500] logger.warning("awooop_ansible_candidate_backfill_once_failed", **stats) 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 77f06f0d3..f4e9a9ce4 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -1175,6 +1175,231 @@ async def backfill_missing_auto_repair_execution_receipts_once( return stats +async def run_failed_apply_check_mode_replay_once( + *, + project_id: str = "awoooi", + window_hours: int = 24, + timeout_seconds: int | None = None, +) -> dict[str, Any]: + """Replay one failed apply in no-write check mode and persist its terminal receipt.""" + + stats: dict[str, Any] = { + "scanned": 0, + "replayed": 0, + "check_mode_passed": 0, + "check_mode_failed": 0, + "runtime_apply_executed": 0, + "blockers": [], + "error": None, + } + blockers = _runtime_blockers() + if blockers: + stats["blockers"] = blockers + return stats + transport_blockers = await recent_ansible_transport_blockers(project_id=project_id) + if transport_blockers: + stats["blockers"] = transport_blockers + return stats + + try: + async with get_db_context(project_id) as db: + result = await db.execute( + text(""" + SELECT + apply.op_id::text AS op_id, + apply.parent_op_id::text AS parent_op_id, + coalesce( + apply.incident_id::text, + apply.input ->> 'incident_id' + ) AS incident_id, + apply.input, + apply.output, + apply.dry_run_result, + apply.error, + apply.duration_ms, + apply.status, + apply.created_at + FROM automation_operation_log apply + WHERE apply.operation_type = 'ansible_apply_executed' + AND apply.status = 'failed' + AND apply.created_at >= NOW() - ( + :window_hours * INTERVAL '1 hour' + ) + AND EXISTS ( + SELECT 1 + FROM incident_evidence verifier + WHERE verifier.post_execution_state ->> 'apply_op_id' + = apply.op_id::text + AND verifier.verification_result = 'failed' + ) + AND NOT EXISTS ( + SELECT 1 + FROM automation_operation_log replay + WHERE replay.operation_type = 'ansible_execution_skipped' + AND replay.parent_op_id = apply.op_id + AND replay.input ->> 'execution_mode' + = 'controlled_retry_check_mode_replay' + ) + ORDER BY apply.created_at DESC + LIMIT 1 + FOR UPDATE SKIP LOCKED + """), + {"window_hours": max(1, window_hours)}, + ) + row = result.mappings().one_or_none() + if row is None: + return stats + stats["scanned"] = 1 + reconstructed = _claim_from_apply_operation_row(dict(row)) + if reconstructed is None: + stats["error"] = "failed_apply_receipt_could_not_be_reconstructed" + return stats + claim, _failed_apply_result = reconstructed + apply_op_id = str(row.get("op_id") or "") + automation_run_id = str( + claim.input_payload.get("automation_run_id") + or claim.source_candidate_op_id + ) + input_payload = { + **claim.input_payload, + "automation_run_id": automation_run_id, + "execution_mode": "controlled_retry_check_mode_replay", + "retry_of_apply_op_id": apply_op_id, + "check_mode": True, + "diff": True, + "apply_enabled": False, + "approval_required_before_apply": True, + "controlled_apply_allowed": False, + "controlled_apply_blocker": ( + "retry_requires_repair_and_new_apply_gate" + ), + } + claimed = 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_execution_skipped', + 'ansible_controlled_retry_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( + claim.incident_id + ), + "input": json.dumps(input_payload, ensure_ascii=False), + "dry_run_result": json.dumps( + { + "execution_mode": ( + "controlled_retry_check_mode_replay" + ), + "retry_of_apply_op_id": apply_op_id, + "check_mode_executed": False, + "runtime_apply_executed": False, + "claim_state": "claimed", + }, + ensure_ascii=False, + ), + "parent_op_id": apply_op_id, + "tags": [ + "ansible", + "controlled_retry", + "check_mode_replay", + "no_runtime_apply", + f"automation_run_id:{automation_run_id}", + ], + }, + ) + replay_op_id = str(claimed.scalar_one()) + effective_timeout = ( + timeout_seconds or settings.AWOOOP_ANSIBLE_CHECK_MODE_TIMEOUT_SECONDS + ) + try: + spec = build_ansible_check_mode_command( + playbook_path=claim.playbook_path, + inventory_hosts=claim.inventory_hosts, + ) + replay_result = await _run_ansible_command( + spec, + timeout_seconds=effective_timeout, + ) + except Exception as exc: + replay_result = AnsibleRunResult( + returncode=1, + stdout="", + stderr=f"ansible_retry_check_mode_runtime_error: {exc}", + duration_ms=0, + ) + status, output, dry_run_result, error = _build_result_payload( + replay_result, + controlled_apply_allowed=False, + controlled_apply_blocker="retry_requires_repair_and_new_apply_gate", + ) + output.update({ + "execution_mode": "controlled_retry_check_mode_replay", + "retry_of_apply_op_id": apply_op_id, + "runtime_apply_executed": False, + "terminal_disposition": ( + "no_write_replay_passed_waiting_repair_candidate" + if replay_result.returncode == 0 + else "no_write_replay_failed_waiting_playbook_or_transport_repair" + ), + }) + dry_run_result.update({ + "execution_mode": "controlled_retry_check_mode_replay", + "retry_of_apply_op_id": apply_op_id, + "runtime_apply_executed": False, + }) + async with get_db_context(project_id) as db: + await db.execute( + text(""" + UPDATE automation_operation_log + SET status = :status, + output = CAST(:output AS jsonb), + dry_run_result = CAST(:dry_run_result AS jsonb), + error = :error, + duration_ms = :duration_ms, + stderr_feed_back = :stderr + WHERE op_id = CAST(:op_id AS uuid) + """), + { + "status": status, + "output": json.dumps(output, ensure_ascii=False), + "dry_run_result": json.dumps(dry_run_result, ensure_ascii=False), + "error": _tail(error or "", 2000) or None, + "duration_ms": replay_result.duration_ms, + "stderr": _tail(replay_result.stderr, _STDERR_LIMIT), + "op_id": replay_op_id, + }, + ) + stats["replayed"] = 1 + stats[ + "check_mode_passed" if replay_result.returncode == 0 else "check_mode_failed" + ] = 1 + logger.info( + "ansible_failed_apply_check_mode_replay_completed", + automation_run_id=automation_run_id, + apply_op_id=apply_op_id, + replay_op_id=str(replay_op_id), + returncode=replay_result.returncode, + runtime_apply_executed=False, + ) + except Exception as exc: + stats["error"] = f"{type(exc).__name__}: {exc}"[:500] + logger.warning("ansible_failed_apply_check_mode_replay_failed", **stats) + return stats + + async def claim_pending_check_modes( *, project_id: str = "awoooi", diff --git a/apps/api/tests/test_awooop_ansible_candidate_backfill_job.py b/apps/api/tests/test_awooop_ansible_candidate_backfill_job.py index 9ce5f198c..fdcf8455b 100644 --- a/apps/api/tests/test_awooop_ansible_candidate_backfill_job.py +++ b/apps/api/tests/test_awooop_ansible_candidate_backfill_job.py @@ -66,6 +66,15 @@ async def test_backfill_enqueues_catalog_matched_incident(monkeypatch: pytest.Mo assert kwargs["project_id"] == "awoooi" return {"written": 2, "error": None} + async def fake_retry_replayer(**kwargs): + assert kwargs["project_id"] == "awoooi" + return { + "replayed": 1, + "check_mode_passed": 1, + "check_mode_failed": 0, + "error": None, + } + monkeypatch.setattr(job, "get_db_context", fake_db_context) monkeypatch.setattr(job.settings, "ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER", True) @@ -75,11 +84,15 @@ async def test_backfill_enqueues_catalog_matched_incident(monkeypatch: pytest.Mo window_hours=24, recorder=fake_recorder, receipt_backfiller=fake_receipt_backfiller, + retry_replayer=fake_retry_replayer, ) assert result["queued"] == 1 assert result["no_catalog_candidate"] == 0 assert result["repair_receipts_backfilled"] == 2 + assert result["failed_apply_retry_replayed"] == 1 + assert result["failed_apply_retry_check_mode_passed"] == 1 + assert result["failed_apply_retry_check_mode_failed"] == 0 assert recorded[0]["decision_path"] == "repair_candidate_controlled_queue" assert recorded[0]["incident"]["incident_id"] == "INC-20260627-NODE110" diff --git a/apps/api/tests/test_awooop_truth_chain_service.py b/apps/api/tests/test_awooop_truth_chain_service.py index 17616f4ab..bcd244ddd 100644 --- a/apps/api/tests/test_awooop_truth_chain_service.py +++ b/apps/api/tests/test_awooop_truth_chain_service.py @@ -35,6 +35,7 @@ from src.services.awooop_ansible_check_mode_service import ( finalize_check_mode_claim, recent_ansible_transport_blockers, run_controlled_apply_for_claim, + run_failed_apply_check_mode_replay_once, run_pending_check_modes_once, ) from src.services.awooop_truth_chain_service import ( @@ -1778,6 +1779,19 @@ def test_ansible_live_controlled_apply_sends_telegram_receipt_but_backfill_does_ assert inspect.iscoroutinefunction(_send_controlled_apply_telegram_receipt) +def test_failed_apply_retry_replay_is_no_write_and_idempotent() -> None: + source = inspect.getsource(run_failed_apply_check_mode_replay_once) + + assert "FOR UPDATE SKIP LOCKED" in source + assert "controlled_retry_check_mode_replay" in source + assert "build_ansible_check_mode_command" in source + assert "controlled_apply_allowed=False" in source + assert '"runtime_apply_executed": False' in source + assert "retry_requires_repair_and_new_apply_gate" in source + assert "NOT EXISTS" in source + assert "run_controlled_apply_for_claim" not in source + + def test_ansible_check_and_apply_rows_persist_canonical_automation_run_id() -> None: finalize_source = inspect.getsource(finalize_check_mode_claim) apply_source = inspect.getsource(run_controlled_apply_for_claim)