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 c4fb4525f..73d62c5b3 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -2146,6 +2146,65 @@ async def backfill_missing_auto_repair_execution_receipts_once( return stats +async def _load_open_failed_apply_retry_row( + *, + project_id: str, + window_hours: int, +) -> dict[str, Any] | None: + """Read one durable failed-apply retry candidate before transport checks.""" + + 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.incident_id = coalesce( + apply.incident_id::text, + apply.input ->> 'incident_id' + ) + AND 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() + return dict(row) if row is not None else None + + async def run_failed_apply_check_mode_replay_once( *, project_id: str = "awoooi", @@ -2164,102 +2223,62 @@ async def run_failed_apply_check_mode_replay_once( "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: + row = await _load_open_failed_apply_retry_row( + project_id=project_id, + window_hours=window_hours, + ) + if row is None: + return stats + stats["scanned"] = 1 + reconstructed = _claim_from_apply_operation_row(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 + ) + + 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 + + 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": False, + "owner_review_required": False, + "controlled_apply_allowed": True, + "controlled_apply_blocker": None, + "bounded_execution_scope": "retry_check_mode_only", + "next_controlled_action": ( + "queue_ai_playbook_or_transport_repair_candidate" + ), + } 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.incident_id = coalesce( - apply.incident_id::text, - apply.input ->> 'incident_id' - ) - AND 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": False, - "owner_review_required": False, - "controlled_apply_allowed": True, - "controlled_apply_blocker": None, - "bounded_execution_scope": "retry_check_mode_only", - "next_controlled_action": ( - "queue_ai_playbook_or_transport_repair_candidate" - ), - } 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 ( + ) + SELECT 'ansible_execution_skipped', 'ansible_controlled_retry_worker', 'pending', @@ -2269,6 +2288,13 @@ async def run_failed_apply_check_mode_replay_once( CAST(:dry_run_result AS jsonb), CAST(:parent_op_id AS uuid), :tags + WHERE NOT EXISTS ( + SELECT 1 + FROM automation_operation_log replay + WHERE replay.operation_type = 'ansible_execution_skipped' + AND replay.parent_op_id = CAST(:parent_op_id AS uuid) + AND replay.input ->> 'execution_mode' + = 'controlled_retry_check_mode_replay' ) RETURNING op_id """), @@ -2299,7 +2325,11 @@ async def run_failed_apply_check_mode_replay_once( ], }, ) - replay_op_id = str(claimed.scalar_one()) + replay_op_id_value = claimed.scalar_one_or_none() + if replay_op_id_value is None: + stats["blockers"] = ["failed_apply_retry_already_claimed"] + return stats + replay_op_id = str(replay_op_id_value) effective_timeout = ( timeout_seconds or settings.AWOOOP_ANSIBLE_CHECK_MODE_TIMEOUT_SECONDS ) @@ -3164,6 +3194,69 @@ async def run_pending_check_modes_once( expired_capability_count = await _expire_stale_ansible_execution_capabilities( project_id=project_id, ) + effective_timeout_seconds = ( + timeout_seconds or settings.AWOOOP_ANSIBLE_CHECK_MODE_TIMEOUT_SECONDS + ) + retry_stats = await run_failed_apply_check_mode_replay_once( + project_id=project_id, + window_hours=24, + timeout_seconds=effective_timeout_seconds, + ) + failed_apply_retry_blockers = [ + str(item)[:200] + for item in retry_stats.get("blockers") or [] + ] + failed_apply_retry_error = ( + str(retry_stats.get("error") or "")[:500] or None + ) + failed_apply_retry_summary = { + "failed_apply_retry_scanned": int(retry_stats.get("scanned") or 0), + "failed_apply_retry_replayed": int(retry_stats.get("replayed") or 0), + "failed_apply_retry_check_mode_passed": int( + retry_stats.get("check_mode_passed") or 0 + ), + "failed_apply_retry_check_mode_failed": int( + retry_stats.get("check_mode_failed") or 0 + ), + "failed_apply_retry_stage_receipt_written": int( + retry_stats.get("runtime_stage_receipt_written") or 0 + ), + "failed_apply_retry_blockers": failed_apply_retry_blockers, + "failed_apply_retry_error": failed_apply_retry_error, + } + failed_apply_retry_priority_tick = bool( + failed_apply_retry_summary["failed_apply_retry_scanned"] + or failed_apply_retry_summary["failed_apply_retry_replayed"] + or failed_apply_retry_blockers + or failed_apply_retry_error + ) + failed_apply_retry_summary["failed_apply_retry_priority_tick"] = ( + failed_apply_retry_priority_tick + ) + if failed_apply_retry_priority_tick: + logger.info( + "ansible_execution_broker_failed_apply_retry_priority_tick", + project_id=project_id, + **failed_apply_retry_summary, + ) + return { + "claimed": 0, + "reclaimed": 0, + "catalog_replayed": 0, + "completed": 0, + "failed": 0, + "apply_completed": 0, + "apply_failed": 0, + "apply_blocked": 0, + "capability_issued": 0, + "capability_revoked": 0, + "capability_expired": expired_capability_count, + "capability_revoke_failed": 0, + "catalog_replay_error": None, + "error": failed_apply_retry_error, + "blockers": failed_apply_retry_blockers, + **failed_apply_retry_summary, + } blockers = _runtime_blockers() if blockers: logger.warning("ansible_check_mode_runtime_blocked", blockers=blockers) @@ -3173,6 +3266,7 @@ async def run_pending_check_modes_once( "failed": 0, "capability_expired": expired_capability_count, "blockers": blockers, + **failed_apply_retry_summary, } transport_blockers = await recent_ansible_transport_blockers(project_id=project_id) if transport_blockers: @@ -3183,11 +3277,8 @@ async def run_pending_check_modes_once( "failed": 0, "capability_expired": expired_capability_count, "blockers": transport_blockers, + **failed_apply_retry_summary, } - - effective_timeout_seconds = ( - timeout_seconds or settings.AWOOOP_ANSIBLE_CHECK_MODE_TIMEOUT_SECONDS - ) reclaimed_claims = await claim_stale_pending_check_modes( project_id=project_id, limit=limit, @@ -3340,5 +3431,6 @@ async def run_pending_check_modes_once( "capability_expired": expired_capability_count, "capability_revoke_failed": capability_revoke_failed, "catalog_replay_error": catalog_replay_error, + **failed_apply_retry_summary, "blockers": [], } diff --git a/apps/api/tests/test_awooop_truth_chain_service.py b/apps/api/tests/test_awooop_truth_chain_service.py index ac98891c0..49315047a 100644 --- a/apps/api/tests/test_awooop_truth_chain_service.py +++ b/apps/api/tests/test_awooop_truth_chain_service.py @@ -25,6 +25,7 @@ from src.services.awooop_ansible_check_mode_service import ( _execution_capability_timeout_seconds, _expire_stale_ansible_execution_capabilities, _issue_ansible_execution_capability, + _load_open_failed_apply_retry_row, _load_pre_decision_context_runtime_stage_receipts, _post_apply_km_path_type, _post_apply_verification_result, @@ -1871,6 +1872,17 @@ async def test_catalog_drift_query_failure_does_not_block_fresh_candidate_claims async def no_stale_claims(**_kwargs): return [] + async def no_failed_apply_retry(**_kwargs): + return { + "scanned": 0, + "replayed": 0, + "check_mode_passed": 0, + "check_mode_failed": 0, + "runtime_stage_receipt_written": 0, + "blockers": [], + "error": None, + } + async def failed_catalog_replay(**_kwargs): raise RuntimeError("catalog replay query canceled") @@ -1891,6 +1903,11 @@ async def test_catalog_drift_query_failure_does_not_block_fresh_candidate_claims no_transport_blockers, ) monkeypatch.setattr(service, "claim_stale_pending_check_modes", no_stale_claims) + monkeypatch.setattr( + service, + "run_failed_apply_check_mode_replay_once", + no_failed_apply_retry, + ) monkeypatch.setattr( service, "claim_catalog_drift_failed_check_modes", @@ -1906,6 +1923,79 @@ async def test_catalog_drift_query_failure_does_not_block_fresh_candidate_claims assert result["catalog_replay_error"] == "RuntimeError" +@pytest.mark.asyncio +async def test_execution_broker_runs_failed_apply_retry_before_all_candidate_claims( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from src.services import awooop_ansible_check_mode_service as service + + claim_attempted = False + + async def no_expired_capabilities(**_kwargs): + return 0 + + async def no_transport_blockers(**_kwargs): + return [] + + async def replay_open_retry(**_kwargs): + return { + "scanned": 1, + "replayed": 1, + "check_mode_passed": 1, + "check_mode_failed": 0, + "runtime_stage_receipt_written": 1, + "blockers": [], + "error": None, + } + + async def candidate_claim_must_not_run(**_kwargs): + nonlocal claim_attempted + claim_attempted = True + return [] + + monkeypatch.setattr( + service, + "_expire_stale_ansible_execution_capabilities", + no_expired_capabilities, + ) + monkeypatch.setattr(service, "_runtime_blockers", lambda: []) + monkeypatch.setattr( + service, + "recent_ansible_transport_blockers", + no_transport_blockers, + ) + monkeypatch.setattr( + service, + "run_failed_apply_check_mode_replay_once", + replay_open_retry, + ) + monkeypatch.setattr( + service, + "claim_stale_pending_check_modes", + candidate_claim_must_not_run, + ) + monkeypatch.setattr( + service, + "claim_catalog_drift_failed_check_modes", + candidate_claim_must_not_run, + ) + monkeypatch.setattr( + service, + "claim_pending_check_modes", + candidate_claim_must_not_run, + ) + + result = await service.run_pending_check_modes_once(limit=1) + + assert claim_attempted is False + assert result["claimed"] == 0 + assert result["failed_apply_retry_scanned"] == 1 + assert result["failed_apply_retry_replayed"] == 1 + assert result["failed_apply_retry_check_mode_passed"] == 1 + assert result["failed_apply_retry_stage_receipt_written"] == 1 + assert result["failed_apply_retry_priority_tick"] is True + + def test_check_mode_worker_error_backoff_is_short_and_bounded() -> None: from src.jobs import awooop_ansible_check_mode_job as job @@ -2108,12 +2198,13 @@ def test_ansible_live_controlled_apply_sends_telegram_receipt_but_backfill_does_ def test_failed_apply_retry_replay_is_no_write_and_idempotent() -> None: source = inspect.getsource(run_failed_apply_check_mode_replay_once) + preflight_source = inspect.getsource(_load_open_failed_apply_retry_row) - assert "FOR UPDATE SKIP LOCKED" in source + assert "FOR UPDATE SKIP LOCKED" in preflight_source assert "controlled_retry_check_mode_replay" in source assert "build_ansible_check_mode_command" in source - assert "verifier.incident_id = coalesce(" in source - assert "apply.input ->> 'incident_id'" in source + assert "verifier.incident_id = coalesce(" in preflight_source + assert "apply.input ->> 'incident_id'" in preflight_source assert "controlled_apply_allowed=True" in source assert '"approval_required_before_apply": False' in source assert '"owner_review_required": False' in source @@ -2121,7 +2212,8 @@ def test_failed_apply_retry_replay_is_no_write_and_idempotent() -> None: assert "queue_ai_playbook_or_transport_repair_candidate" in source assert "retry_requires_repair_and_new_apply_gate" not in source assert "_record_retry_runtime_stage_receipt" in source - assert "NOT EXISTS" in source + assert "NOT EXISTS" in preflight_source + assert "failed_apply_retry_already_claimed" in source assert "run_controlled_apply_for_claim" not in source receipt_source = inspect.getsource(_record_retry_runtime_stage_receipt)