From 47ab6613ddf78dfcd7e6189ef9b4de4cb1b78085 Mon Sep 17 00:00:00 2001 From: ogt Date: Sat, 11 Jul 2026 13:18:16 +0800 Subject: [PATCH] fix(agent): close verified alert lifecycle --- .../awooop_ansible_candidate_backfill_job.py | 47 +- .../awooop_ansible_check_mode_service.py | 713 +++++++++++++++++- .../tests/test_ansible_verified_closure.py | 292 +++++++ .../tests/test_awooop_truth_chain_service.py | 11 + 4 files changed, 1048 insertions(+), 15 deletions(-) create mode 100644 apps/api/tests/test_ansible_verified_closure.py 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 bacd412de..6dcaa4b93 100644 --- a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py +++ b/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py @@ -27,8 +27,7 @@ from src.services.awooop_ansible_audit_service import ( record_ansible_decision_audit, ) from src.services.awooop_ansible_check_mode_service import ( - backfill_missing_auto_repair_execution_receipts_once, - run_failed_apply_check_mode_replay_once, + preflight_failed_apply_retry_queue_once, ) from src.services.evidence_snapshot import EvidenceSnapshot from src.services.pre_decision_investigator import get_pre_decision_investigator @@ -83,7 +82,18 @@ async def _fetch_missing_candidate_incidents( frequency_snapshot, affected_services, signals, - decision_chain + decision_chain, + ( + SELECT approval.id + FROM approval_records approval + WHERE approval.incident_id = incidents.incident_id + AND lower(approval.status::text) IN ( + 'approved', + 'execution_failed' + ) + ORDER BY approval.updated_at DESC + LIMIT 1 + ) AS approval_id FROM incidents WHERE (project_id = :project_id OR project_id IS NULL) AND created_at >= NOW() - (:window_hours * INTERVAL '1 hour') @@ -97,6 +107,13 @@ async def _fetch_missing_candidate_incidents( AND existing.input ->> 'executor' = 'ansible' AND existing.input ->> 'decision_path' = 'repair_candidate_controlled_queue' AND coalesce(existing.incident_id::text, existing.input ->> 'incident_id') = incidents.incident_id::text + AND NOT EXISTS ( + SELECT 1 + FROM automation_operation_log terminal + WHERE terminal.parent_op_id = existing.op_id + AND terminal.operation_type + = 'ansible_execution_skipped' + ) ) ORDER BY created_at DESC LIMIT :scan_limit @@ -116,6 +133,7 @@ def _build_backfill_proposal(incident: dict[str, Any]) -> dict[str, Any]: "risk_level": str(incident.get("severity") or ""), "action": "enqueue_allowlisted_ansible_check_mode", "alertname": incident.get("alertname"), + "approval_id": str(incident.get("approval_id") or ""), } @@ -731,10 +749,10 @@ async def enqueue_missing_ansible_candidates_once( recorder: Recorder = record_ansible_decision_audit, receipt_backfiller: Callable[ ..., Awaitable[dict[str, Any]] - ] = backfill_missing_auto_repair_execution_receipts_once, + ] | None = None, retry_replayer: Callable[ ..., Awaitable[dict[str, Any]] - ] = run_failed_apply_check_mode_replay_once, + ] = preflight_failed_apply_retry_queue_once, evidence_collector: EvidenceCollector = ( collect_ansible_candidate_pre_decision_evidence ), @@ -827,14 +845,17 @@ async def enqueue_missing_ansible_candidates_once( logger.info("awooop_ansible_failed_apply_retry_priority_tick", **stats) return stats - receipt_stats = await receipt_backfiller( - project_id=project_id, - window_hours=bounded_window_hours, - limit=bounded_limit, - ) - 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"] + if receipt_backfiller is not None: + receipt_stats = await receipt_backfiller( + project_id=project_id, + window_hours=bounded_window_hours, + limit=bounded_limit, + ) + 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"] incidents = await _fetch_missing_candidate_incidents( project_id=project_id, 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 41ba82b5b..15c90d848 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -61,6 +61,21 @@ _EXECUTION_CAPABILITY_OPERATION_TYPES = frozenset( "ansible_executor_capability_expired", } ) +_VERIFIED_CLOSURE_REQUIRED_STAGE_IDS = frozenset( + { + "mcp_context", + "service_log_evidence", + "normalized_asset_identity", + "source_truth_diff", + "risk_policy_decision", + "executor_log_projection", + "retry_or_rollback", + "km_playbook_writeback", + "rag_writeback", + "playbook_trust", + "timeline_projection", + } +) FORCED_COMMAND_BLOCKER = "ansible_repair_ssh_forced_command_denies_ansible_bootstrap" REPAIR_FORCED_COMMAND_KEY_PATH = Path("/etc/repair-ssh/id_ed25519") REPAIR_FORCED_COMMAND_KNOWN_HOSTS_PATH = Path("/etc/repair-known-hosts/known_hosts") @@ -747,6 +762,38 @@ async def _append_alert_lifecycle_receipt( get_alert_operation_log_repository, ) + automation_run_id = _automation_run_id_for_claim(claim) + try: + async with get_db_context(project_id) as db: + existing = await db.execute( + text(""" + SELECT EXISTS ( + SELECT 1 + FROM alert_operation_log + WHERE incident_id = :incident_id + AND event_type::text = :event_type + AND context ->> 'apply_op_id' = :apply_op_id + AND context ->> 'automation_run_id' = :automation_run_id + ) + """), + { + "incident_id": claim.incident_id, + "event_type": event_type, + "apply_op_id": apply_op_id, + "automation_run_id": automation_run_id, + }, + ) + if existing.scalar() is True: + return True + except Exception as exc: + logger.warning( + "ansible_alert_lifecycle_receipt_readback_failed", + incident_id=claim.incident_id, + apply_op_id=apply_op_id, + event_type=event_type, + error_type=type(exc).__name__, + ) + record = await get_alert_operation_log_repository().append( event_type, incident_id=claim.incident_id, @@ -756,7 +803,7 @@ async def _append_alert_lifecycle_receipt( success=success, error_message=error_message, context={ - "automation_run_id": _automation_run_id_for_claim(claim), + "automation_run_id": automation_run_id, "catalog_id": claim.catalog_id, "check_mode_op_id": claim.op_id, "apply_op_id": apply_op_id, @@ -1988,6 +2035,393 @@ async def _append_runtime_stage_receipts_to_apply( return False +def _runtime_stage_ids(value: Any) -> set[str]: + payload = _json_loads(value) + receipts = payload.get("runtime_stage_receipts") + if not isinstance(receipts, list): + return set() + return { + str(receipt.get("stage_id") or "") + for receipt in receipts + if isinstance(receipt, Mapping) and receipt.get("durable_receipt") is True + } + + +async def _read_verified_apply_closure_prerequisites( + claim: AnsibleCheckModeClaim, + *, + apply_op_id: str, + project_id: str, +) -> dict[str, Any]: + """Read back every durable prerequisite before resolving an incident.""" + + automation_run_id = _automation_run_id_for_claim(claim) + approval_id = _approval_id_for_claim(claim) + canonical_playbook_id = canonical_ansible_playbook_id(claim.catalog_id) + async with get_db_context(project_id) as db: + result = await db.execute( + text(""" + SELECT + apply.status = 'success' AS apply_success, + apply.input AS apply_input, + coalesce( + apply.output ->> 'independent_post_verifier_passed', + 'false' + ) = 'true' AS apply_verifier_terminal, + check_mode.status = 'success' AS check_mode_success, + candidate.op_id IS NOT NULL AS candidate_present, + EXISTS ( + SELECT 1 + FROM auto_repair_executions receipt + WHERE receipt.incident_id = :incident_id + AND receipt.triggered_by = 'ansible_controlled_apply' + AND receipt.success IS TRUE + AND receipt.executed_steps::text LIKE + '%' || :apply_op_id || '%' + ) AS auto_repair_receipt, + EXISTS ( + SELECT 1 + FROM incident_evidence evidence + WHERE evidence.incident_id = :incident_id + AND evidence.post_execution_state ->> 'apply_op_id' + = :apply_op_id + AND evidence.verification_result = 'success' + ) AS post_verifier_receipt, + EXISTS ( + SELECT 1 + FROM knowledge_entries km + WHERE km.related_incident_id = :incident_id + AND km.path_type = :km_path_type + AND km.related_playbook_id = :canonical_playbook_id + AND km.updated_at IS NOT NULL + ) AS km_writeback_receipt, + EXISTS ( + SELECT 1 + FROM automation_operation_log learning + WHERE learning.operation_type + = 'ansible_learning_writeback_recorded' + AND learning.parent_op_id = apply.op_id + AND learning.status = 'success' + AND learning.output ->> 'learning_recorded' = 'true' + AND learning.output ->> 'trust_updated' = 'true' + ) AS playbook_trust_receipt, + EXISTS ( + SELECT 1 + FROM playbooks playbook + WHERE playbook.playbook_id = :canonical_playbook_id + AND playbook.success_count > 0 + AND playbook.last_used_at IS NOT NULL + ) AS playbook_readback, + EXISTS ( + SELECT 1 + FROM timeline_events timeline + WHERE timeline.incident_id = :incident_id + AND timeline.actor = 'ansible_controlled_apply_worker' + AND timeline.status = 'success' + AND timeline.description LIKE + '%apply_op_id=' || :apply_op_id || ';%' + ) AS timeline_readback, + EXISTS ( + SELECT 1 + FROM awooop_outbound_message outbound + WHERE outbound.project_id = :project_id + AND outbound.channel_type = 'telegram' + AND outbound.send_status = 'sent' + AND outbound.provider_message_id IS NOT NULL + AND outbound.source_envelope + #>> '{callback_reply,action}' + = 'controlled_apply_result' + AND coalesce( + outbound.source_envelope + ->> 'automation_run_id', + outbound.source_envelope + #>> '{callback_reply,automation_run_id}' + ) = :automation_run_id + AND coalesce( + outbound.source_envelope + #>> '{callback_reply,incident_id}', + outbound.source_envelope + #>> '{source_refs,incident_ids,0}' + ) = :incident_id + ) AS telegram_receipt, + ( + :approval_id = '' + OR EXISTS ( + SELECT 1 + FROM approval_records approval + WHERE approval.id = :approval_id + AND lower(approval.status::text) + = 'execution_success' + AND approval.extra_metadata ->> 'apply_op_id' + = :apply_op_id + AND approval.extra_metadata + ->> 'post_verifier_passed' = 'true' + ) + ) AS approval_projection, + EXISTS ( + SELECT 1 + FROM alert_operation_log lifecycle + WHERE lifecycle.incident_id = :incident_id + AND lifecycle.event_type::text + = 'EXECUTION_COMPLETED' + AND lifecycle.success IS TRUE + AND lifecycle.context ->> 'apply_op_id' + = :apply_op_id + AND lifecycle.context ->> 'automation_run_id' + = :automation_run_id + ) AS execution_lifecycle, + EXISTS ( + SELECT 1 + FROM alert_operation_log lifecycle + WHERE lifecycle.incident_id = :incident_id + AND lifecycle.event_type::text + = 'TELEGRAM_RESULT_SENT' + AND lifecycle.success IS TRUE + AND lifecycle.context ->> 'apply_op_id' + = :apply_op_id + AND lifecycle.context ->> 'automation_run_id' + = :automation_run_id + ) AS telegram_lifecycle + FROM automation_operation_log apply + LEFT JOIN automation_operation_log check_mode + ON check_mode.op_id = apply.parent_op_id + AND check_mode.operation_type = 'ansible_check_mode_executed' + LEFT JOIN automation_operation_log candidate + ON candidate.op_id = check_mode.parent_op_id + AND candidate.operation_type = 'ansible_candidate_matched' + WHERE apply.op_id = CAST(:apply_op_id AS uuid) + AND apply.operation_type = 'ansible_apply_executed' + LIMIT 1 + """), + { + "apply_op_id": apply_op_id, + "incident_id": claim.incident_id, + "project_id": project_id, + "automation_run_id": automation_run_id, + "approval_id": approval_id, + "canonical_playbook_id": canonical_playbook_id, + "km_path_type": f"ansible_apply_receipt:{apply_op_id[:8]}", + }, + ) + row = result.mappings().one_or_none() + + if row is None: + return { + "ready": False, + "missing": ["apply_operation_readback"], + "receipts": {}, + "stage_ids": [], + } + + stage_ids = _runtime_stage_ids(row.get("apply_input")) + receipts = { + "candidate": row.get("candidate_present") is True, + "check_mode": row.get("check_mode_success") is True, + "controlled_apply": row.get("apply_success") is True, + "post_apply_verifier": ( + row.get("apply_verifier_terminal") is True + and row.get("post_verifier_receipt") is True + ), + "auto_repair_execution_receipt": row.get("auto_repair_receipt") is True, + "km_playbook_writeback": row.get("km_writeback_receipt") is True, + "rag_writeback": "rag_writeback" in stage_ids, + "playbook_trust": ( + row.get("playbook_trust_receipt") is True + and row.get("playbook_readback") is True + ), + "timeline_projection": ( + "timeline_projection" in stage_ids + and row.get("timeline_readback") is True + ), + "telegram_receipt": row.get("telegram_receipt") is True, + "approval_projection": row.get("approval_projection") is True, + "execution_lifecycle": row.get("execution_lifecycle") is True, + "telegram_lifecycle": row.get("telegram_lifecycle") is True, + "required_runtime_stage_receipts": ( + _VERIFIED_CLOSURE_REQUIRED_STAGE_IDS.issubset(stage_ids) + ), + } + missing = sorted(name for name, ready in receipts.items() if not ready) + return { + "ready": not missing, + "missing": missing, + "receipts": receipts, + "stage_ids": sorted(stage_ids), + "required_stage_ids": sorted(_VERIFIED_CLOSURE_REQUIRED_STAGE_IDS), + "automation_run_id": automation_run_id, + "canonical_playbook_id": canonical_playbook_id, + } + + +async def _read_incident_closure_readback( + claim: AnsibleCheckModeClaim, + *, + apply_op_id: str, + project_id: str, +) -> dict[str, Any]: + automation_run_id = _automation_run_id_for_claim(claim) + async with get_db_context(project_id) as db: + result = await db.execute( + text(""" + SELECT + incident.status::text AS incident_status, + incident.resolved_at, + incident.outcome, + EXISTS ( + SELECT 1 + FROM jsonb_array_elements( + coalesce( + apply.input -> 'runtime_stage_receipts', + '[]'::jsonb + ) + ) receipt(value) + WHERE receipt.value ->> 'stage_id' = 'incident_closure' + AND receipt.value ->> 'automation_run_id' + = :automation_run_id + AND receipt.value ->> 'durable_receipt' = 'true' + ) AS closure_receipt, + EXISTS ( + SELECT 1 + FROM alert_operation_log lifecycle + WHERE lifecycle.incident_id = :incident_id + AND lifecycle.event_type::text = 'RESOLVED' + AND lifecycle.success IS TRUE + AND lifecycle.context ->> 'apply_op_id' + = :apply_op_id + AND lifecycle.context ->> 'automation_run_id' + = :automation_run_id + ) AS resolved_lifecycle + FROM incidents incident + JOIN automation_operation_log apply + ON apply.op_id = CAST(:apply_op_id AS uuid) + WHERE incident.incident_id = :incident_id + AND incident.project_id = :project_id + LIMIT 1 + """), + { + "incident_id": claim.incident_id, + "project_id": project_id, + "apply_op_id": apply_op_id, + "automation_run_id": automation_run_id, + }, + ) + row = result.mappings().one_or_none() + + if row is None: + return {"closed": False, "missing": ["incident_readback"]} + outcome = _json_loads(row.get("outcome")) + terminal = outcome.get("automation_terminal") + terminal_ready = bool( + isinstance(terminal, Mapping) + and terminal.get("automation_run_id") == automation_run_id + and terminal.get("apply_op_id") == apply_op_id + and terminal.get("incident_resolved") is True + ) + checks = { + "incident_status": str(row.get("incident_status") or "").upper() + in {"RESOLVED", "CLOSED"}, + "incident_resolved_at": row.get("resolved_at") is not None, + "incident_terminal_readback": terminal_ready, + "incident_closure_receipt": row.get("closure_receipt") is True, + "resolved_lifecycle": row.get("resolved_lifecycle") is True, + } + missing = sorted(name for name, ready in checks.items() if not ready) + return { + "closed": not missing, + "missing": missing, + "checks": checks, + "incident_status": str(row.get("incident_status") or ""), + "automation_run_id": automation_run_id, + "apply_op_id": apply_op_id, + } + + +async def _finalize_verified_apply_closure( + claim: AnsibleCheckModeClaim, + *, + apply_op_id: str, + project_id: str, +) -> dict[str, Any]: + prerequisites = await _read_verified_apply_closure_prerequisites( + claim, + apply_op_id=apply_op_id, + project_id=project_id, + ) + if prerequisites.get("ready") is not True: + return { + "status": "closure_receipts_pending", + "closed": False, + "missing": list(prerequisites.get("missing") or []), + } + + terminal = await _record_incident_terminal_disposition( + claim, + apply_op_id=apply_op_id, + project_id=project_id, + terminal_type="controlled_apply_verified_closed", + success_terminal=True, + telegram_receipt_acknowledged=True, + ) + if terminal is None: + return { + "status": "incident_terminal_write_pending", + "closed": False, + "missing": ["incident_terminal_readback"], + } + + closure_receipt = _runtime_stage_receipt( + claim, + stage_id="incident_closure", + evidence_ref=f"incidents:{claim.incident_id}:automation_terminal", + detail={ + **terminal, + "closure_prerequisite_count": len(prerequisites["receipts"]), + "closure_prerequisites_verified": True, + }, + ) + receipt_written = await _append_runtime_stage_receipts_to_apply( + apply_op_id=apply_op_id, + receipts=(closure_receipt,), + project_id=project_id, + ) + resolved_lifecycle = await _append_alert_lifecycle_receipt( + claim, + "RESOLVED", + apply_op_id=apply_op_id, + success=True, + action_detail="controlled_apply_verified_closed", + project_id=project_id, + post_verifier_passed=True, + ) + if not receipt_written or not resolved_lifecycle: + return { + "status": "closure_projection_pending", + "closed": False, + "missing": [ + name + for name, ready in { + "incident_closure_receipt": receipt_written, + "resolved_lifecycle": resolved_lifecycle, + }.items() + if not ready + ], + } + + readback = await _read_incident_closure_readback( + claim, + apply_op_id=apply_op_id, + project_id=project_id, + ) + return { + "status": ( + "controlled_apply_closed" + if readback.get("closed") is True + else "closure_readback_pending" + ), + **readback, + } + + async def _record_incident_terminal_disposition( claim: AnsibleCheckModeClaim, *, @@ -2772,6 +3206,97 @@ async def _send_controlled_apply_telegram_receipt( return False +async def _reconcile_verified_apply_closure_projections( + claim: AnsibleCheckModeClaim, + result: AnsibleRunResult, + *, + apply_op_id: str, + writeback: dict[str, Any], + project_id: str, +) -> dict[str, Any]: + """Repair terminal projections without ever executing the apply again.""" + + if not ( + result.returncode == 0 + and result.post_verifier_passed is True + and writeback.get("verification_passed") is True + ): + return { + "status": "not_verified_success", + "closed": False, + "runtime_apply_executed": False, + } + + approval_projection = await _finalize_controlled_approval_projection( + claim, + result, + apply_op_id=apply_op_id, + project_id=project_id, + ) + execution_lifecycle = await _append_alert_lifecycle_receipt( + claim, + "EXECUTION_COMPLETED", + apply_op_id=apply_op_id, + success=True, + action_detail=f"controlled_apply_completed:{claim.catalog_id}", + project_id=project_id, + post_verifier_passed=True, + ) + + readback = await _read_verified_apply_closure_prerequisites( + claim, + apply_op_id=apply_op_id, + project_id=project_id, + ) + telegram_receipt = bool( + (readback.get("receipts") or {}).get("telegram_receipt") + ) + if not telegram_receipt: + await _send_controlled_apply_telegram_receipt( + claim, + result, + apply_op_id=apply_op_id, + writeback=writeback, + project_id=project_id, + ) + readback = await _read_verified_apply_closure_prerequisites( + claim, + apply_op_id=apply_op_id, + project_id=project_id, + ) + telegram_receipt = bool( + (readback.get("receipts") or {}).get("telegram_receipt") + ) + + telegram_lifecycle = False + if telegram_receipt: + telegram_lifecycle = await _append_alert_lifecycle_receipt( + claim, + "TELEGRAM_RESULT_SENT", + apply_op_id=apply_op_id, + success=True, + action_detail="controlled_apply_result_receipt_sent", + project_id=project_id, + post_verifier_passed=True, + ) + + closure = await _finalize_verified_apply_closure( + claim, + apply_op_id=apply_op_id, + project_id=project_id, + ) + return { + **closure, + "approval_projection_written": ( + approval_projection or not _approval_id_for_claim(claim) + ), + "execution_lifecycle_written": execution_lifecycle, + "telegram_receipt_acknowledged": telegram_receipt, + "telegram_lifecycle_written": telegram_lifecycle, + "runtime_apply_executed": False, + } + + async def backfill_missing_auto_repair_execution_receipts_once( *, project_id: str = "awoooi", @@ -2790,6 +3315,8 @@ async def backfill_missing_auto_repair_execution_receipts_once( "playbook_trust_written": 0, "timeline_projection_written": 0, "runtime_stage_receipts_written": 0, + "incident_closure_written": 0, + "telegram_receipt_acknowledged": 0, "skipped": 0, "error": None, } @@ -2893,11 +3420,58 @@ async def backfill_missing_auto_repair_execution_receipts_once( )::timestamptz <= NOW() - INTERVAL '5 minutes' ) ) + OR ( + apply.status = 'success' + AND ( + NOT EXISTS ( + SELECT 1 + FROM jsonb_array_elements( + coalesce( + apply.input + -> 'runtime_stage_receipts', + '[]'::jsonb + ) + ) AS receipt(value) + WHERE receipt.value ->> 'stage_id' + = 'incident_closure' + ) + OR EXISTS ( + SELECT 1 + FROM incidents incident + WHERE incident.incident_id = coalesce( + apply.incident_id::text, + apply.input ->> 'incident_id' + ) + AND incident.project_id = :project_id + AND ( + incident.resolved_at IS NULL + OR upper(incident.status::text) + NOT IN ('RESOLVED', 'CLOSED') + ) + ) + OR NOT EXISTS ( + SELECT 1 + FROM alert_operation_log lifecycle + WHERE lifecycle.incident_id = coalesce( + apply.incident_id::text, + apply.input ->> 'incident_id' + ) + AND lifecycle.event_type::text + = 'RESOLVED' + AND lifecycle.context ->> 'apply_op_id' + = apply.op_id::text + ) + ) + ) ) ORDER BY apply.created_at DESC LIMIT :limit """), - {"window_hours": max(1, window_hours), "limit": max(1, limit)}, + { + "project_id": project_id, + "window_hours": max(1, window_hours), + "limit": max(1, limit), + }, ) rows = [dict(row) for row in result.mappings().all()] stats["scanned"] = len(rows) @@ -2964,6 +3538,17 @@ async def backfill_missing_auto_repair_execution_receipts_once( ), ): stats["runtime_stage_receipts_written"] += 1 + closure = await _reconcile_verified_apply_closure_projections( + claim, + verified_result, + apply_op_id=str(row.get("op_id") or ""), + writeback=writeback, + project_id=project_id, + ) + if closure.get("telegram_receipt_acknowledged") is True: + stats["telegram_receipt_acknowledged"] += 1 + if closure.get("closed") is True: + stats["incident_closure_written"] += 1 except Exception as exc: stats["error"] = f"{type(exc).__name__}: {exc}"[:500] logger.warning("ansible_auto_repair_execution_receipt_backfill_failed", **stats) @@ -3029,6 +3614,46 @@ async def _load_open_failed_apply_retry_row( return dict(row) if row is not None else None +async def preflight_failed_apply_retry_queue_once( + *, + project_id: str = "awoooi", + window_hours: int = 24, + **_kwargs: Any, +) -> dict[str, Any]: + """Query retry backlog without using executor transport or writing state.""" + + try: + row = await _load_open_failed_apply_retry_row( + project_id=project_id, + window_hours=window_hours, + ) + return { + "scanned": 1 if row is not None else 0, + "replayed": 0, + "check_mode_passed": 0, + "check_mode_failed": 0, + "runtime_stage_receipt_written": 0, + "blockers": [], + "error": None, + "query_only": True, + "runtime_apply_executed": False, + "execution_owner": "awoooi-ansible-executor-broker", + } + except Exception as exc: + return { + "scanned": 0, + "replayed": 0, + "check_mode_passed": 0, + "check_mode_failed": 0, + "runtime_stage_receipt_written": 0, + "blockers": [], + "error": type(exc).__name__, + "query_only": True, + "runtime_apply_executed": False, + "execution_owner": "awoooi-ansible-executor-broker", + } + + async def run_failed_apply_check_mode_replay_once( *, project_id: str = "awoooi", @@ -4138,6 +4763,28 @@ async def run_controlled_apply_for_claim( project_id=project_id, post_verifier_passed=verified_result.post_verifier_passed, ) + try: + closure = await _reconcile_verified_apply_closure_projections( + claim, + verified_result, + apply_op_id=apply_op_id, + writeback=writeback, + project_id=project_id, + ) + except Exception as exc: + closure = { + "status": "closure_reconcile_retry_required", + "closed": False, + "missing": [f"closure_reconcile:{type(exc).__name__}"], + } + logger.warning( + "ansible_verified_apply_closure_reconcile_failed", + automation_run_id=_automation_run_id_for_claim(claim), + incident_id=claim.incident_id, + apply_op_id=apply_op_id, + error_type=type(exc).__name__, + runtime_apply_replay_required=False, + ) logger.info( "ansible_controlled_apply_completed", @@ -4162,6 +4809,9 @@ async def run_controlled_apply_for_claim( alert_lifecycle_completed=lifecycle_completed, telegram_receipt_sent=telegram_receipt_sent, alert_lifecycle_telegram_written=lifecycle_telegram_written, + incident_closure_status=closure.get("status"), + incident_closure_readback=closure.get("closed") is True, + incident_closure_missing=closure.get("missing") or [], ) return verified_result @@ -4178,6 +4828,61 @@ async def run_pending_check_modes_once( effective_timeout_seconds = ( timeout_seconds or settings.AWOOOP_ANSIBLE_CHECK_MODE_TIMEOUT_SECONDS ) + receipt_stats = await backfill_missing_auto_repair_execution_receipts_once( + project_id=project_id, + window_hours=24, + limit=max(1, min(int(limit), 5)), + ) + receipt_backfill_summary = { + "repair_receipt_backfill_scanned": int( + receipt_stats.get("scanned") or 0 + ), + "repair_receipt_backfill_written": int( + receipt_stats.get("written") or 0 + ), + "repair_receipt_closure_written": int( + receipt_stats.get("incident_closure_written") or 0 + ), + "repair_receipt_telegram_acknowledged": int( + receipt_stats.get("telegram_receipt_acknowledged") or 0 + ), + "repair_receipt_backfill_error": ( + str(receipt_stats.get("error") or "")[:500] or None + ), + } + receipt_backfill_priority_tick = bool( + receipt_backfill_summary["repair_receipt_backfill_scanned"] + or receipt_backfill_summary["repair_receipt_backfill_error"] + ) + receipt_backfill_summary["repair_receipt_backfill_priority_tick"] = ( + receipt_backfill_priority_tick + ) + if receipt_backfill_priority_tick: + logger.info( + "ansible_execution_broker_receipt_backfill_priority_tick", + project_id=project_id, + **receipt_backfill_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": receipt_backfill_summary[ + "repair_receipt_backfill_error" + ], + "blockers": [], + **receipt_backfill_summary, + } retry_stats = await run_failed_apply_check_mode_replay_once( project_id=project_id, window_hours=24, @@ -4236,6 +4941,7 @@ async def run_pending_check_modes_once( "catalog_replay_error": None, "error": failed_apply_retry_error, "blockers": failed_apply_retry_blockers, + **receipt_backfill_summary, **failed_apply_retry_summary, } blockers = _runtime_blockers() @@ -4247,6 +4953,7 @@ async def run_pending_check_modes_once( "failed": 0, "capability_expired": expired_capability_count, "blockers": blockers, + **receipt_backfill_summary, **failed_apply_retry_summary, } transport_blockers = await recent_ansible_transport_blockers(project_id=project_id) @@ -4258,6 +4965,7 @@ async def run_pending_check_modes_once( "failed": 0, "capability_expired": expired_capability_count, "blockers": transport_blockers, + **receipt_backfill_summary, **failed_apply_retry_summary, } reclaimed_claims = await claim_stale_pending_check_modes( @@ -4421,6 +5129,7 @@ async def run_pending_check_modes_once( "capability_expired": expired_capability_count, "capability_revoke_failed": capability_revoke_failed, "catalog_replay_error": catalog_replay_error, + **receipt_backfill_summary, **failed_apply_retry_summary, "blockers": [], } diff --git a/apps/api/tests/test_ansible_verified_closure.py b/apps/api/tests/test_ansible_verified_closure.py new file mode 100644 index 000000000..89514e054 --- /dev/null +++ b/apps/api/tests/test_ansible_verified_closure.py @@ -0,0 +1,292 @@ +from __future__ import annotations + +from unittest.mock import AsyncMock + +import pytest + +from src.jobs import awooop_ansible_candidate_backfill_job as candidate_job +from src.services import awooop_ansible_check_mode_service as service + + +def _claim() -> service.AnsibleCheckModeClaim: + return service.AnsibleCheckModeClaim( + op_id="00000000-0000-0000-0000-000000000101", + source_candidate_op_id="00000000-0000-0000-0000-000000000100", + incident_id="INC-20260711-D037E5", + catalog_id="ansible:awoooi-auto-repair-canary", + playbook_path="infra/ansible/playbooks/awoooi-auto-repair-canary.yml", + apply_playbook_path=( + "infra/ansible/playbooks/awoooi-auto-repair-canary.yml" + ), + inventory_hosts=("host_121",), + risk_level="medium", + input_payload={ + "automation_run_id": "00000000-0000-0000-0000-000000000102", + "approval_id": "00000000-0000-0000-0000-000000000103", + }, + ) + + +def _verified_result() -> service.AnsibleRunResult: + return service.AnsibleRunResult( + returncode=0, + stdout="", + stderr="", + duration_ms=25, + post_verifier_passed=True, + ) + + +def test_runtime_stage_ids_only_accepts_durable_receipts() -> None: + stage_ids = service._runtime_stage_ids( + { + "runtime_stage_receipts": [ + { + "stage_id": "mcp_context", + "durable_receipt": True, + }, + { + "stage_id": "telegram_receipt", + "durable_receipt": False, + }, + ] + } + ) + + assert stage_ids == {"mcp_context"} + + +@pytest.mark.asyncio +async def test_closure_refuses_to_resolve_when_any_receipt_is_missing( + monkeypatch: pytest.MonkeyPatch, +) -> None: + terminal_writer = AsyncMock() + monkeypatch.setattr( + service, + "_read_verified_apply_closure_prerequisites", + AsyncMock( + return_value={ + "ready": False, + "missing": ["telegram_receipt"], + "receipts": {}, + } + ), + ) + monkeypatch.setattr( + service, + "_record_incident_terminal_disposition", + terminal_writer, + ) + + result = await service._finalize_verified_apply_closure( + _claim(), + apply_op_id="00000000-0000-0000-0000-000000000104", + project_id="awoooi", + ) + + assert result == { + "status": "closure_receipts_pending", + "closed": False, + "missing": ["telegram_receipt"], + } + terminal_writer.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_closure_resolves_only_after_durable_readback( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + service, + "_read_verified_apply_closure_prerequisites", + AsyncMock(return_value={"ready": True, "receipts": {"all": True}}), + ) + monkeypatch.setattr( + service, + "_record_incident_terminal_disposition", + AsyncMock( + return_value={ + "automation_run_id": ( + "00000000-0000-0000-0000-000000000102" + ), + "apply_op_id": "00000000-0000-0000-0000-000000000104", + "incident_resolved": True, + } + ), + ) + receipt_writer = AsyncMock(return_value=True) + lifecycle_writer = AsyncMock(return_value=True) + monkeypatch.setattr( + service, + "_append_runtime_stage_receipts_to_apply", + receipt_writer, + ) + monkeypatch.setattr( + service, + "_append_alert_lifecycle_receipt", + lifecycle_writer, + ) + monkeypatch.setattr( + service, + "_read_incident_closure_readback", + AsyncMock( + return_value={ + "closed": True, + "missing": [], + "incident_status": "RESOLVED", + } + ), + ) + + result = await service._finalize_verified_apply_closure( + _claim(), + apply_op_id="00000000-0000-0000-0000-000000000104", + project_id="awoooi", + ) + + assert result["status"] == "controlled_apply_closed" + assert result["closed"] is True + closure_receipt = receipt_writer.await_args.kwargs["receipts"][0] + assert closure_receipt["stage_id"] == "incident_closure" + assert closure_receipt["durable_receipt"] is True + assert lifecycle_writer.await_args.args[1] == "RESOLVED" + + +@pytest.mark.asyncio +async def test_projection_replay_sends_missing_receipt_without_reapplying( + monkeypatch: pytest.MonkeyPatch, +) -> None: + readback = AsyncMock( + side_effect=[ + {"receipts": {"telegram_receipt": False}}, + {"receipts": {"telegram_receipt": True}}, + ] + ) + telegram_sender = AsyncMock(return_value={"ok": True}) + monkeypatch.setattr( + service, + "_finalize_controlled_approval_projection", + AsyncMock(return_value=True), + ) + monkeypatch.setattr( + service, + "_append_alert_lifecycle_receipt", + AsyncMock(return_value=True), + ) + monkeypatch.setattr( + service, + "_read_verified_apply_closure_prerequisites", + readback, + ) + monkeypatch.setattr( + service, + "_send_controlled_apply_telegram_receipt", + telegram_sender, + ) + monkeypatch.setattr( + service, + "_finalize_verified_apply_closure", + AsyncMock(return_value={"status": "controlled_apply_closed", "closed": True}), + ) + + result = await service._reconcile_verified_apply_closure_projections( + _claim(), + _verified_result(), + apply_op_id="00000000-0000-0000-0000-000000000104", + writeback={ + "verification_passed": True, + "verification_result": "success", + "verification": True, + "learning": True, + }, + project_id="awoooi", + ) + + assert result["closed"] is True + assert result["runtime_apply_executed"] is False + telegram_sender.assert_awaited_once() + + +def test_backfill_preserves_approval_and_replaces_terminal_skipped_candidate() -> None: + proposal = candidate_job._build_backfill_proposal( + { + "alertname": "AwoooPAutoRepairCanaryT16", + "severity": "medium", + "approval_id": "00000000-0000-0000-0000-000000000103", + } + ) + query_source = "\n".join( + str(value) + for value in candidate_job._fetch_missing_candidate_incidents.__code__.co_consts + ) + + assert proposal["approval_id"] == "00000000-0000-0000-0000-000000000103" + assert "approval_records" in query_source + assert "ansible_execution_skipped" in query_source + assert "terminal.parent_op_id = existing.op_id" in query_source + + +@pytest.mark.asyncio +async def test_broker_reconciles_receipts_before_claiming_fresh_work( + monkeypatch: pytest.MonkeyPatch, +) -> None: + retry_replayer = AsyncMock() + monkeypatch.setattr( + service, + "_expire_stale_ansible_execution_capabilities", + AsyncMock(return_value=0), + ) + monkeypatch.setattr( + service, + "backfill_missing_auto_repair_execution_receipts_once", + AsyncMock( + return_value={ + "scanned": 1, + "written": 1, + "incident_closure_written": 1, + "telegram_receipt_acknowledged": 1, + "error": None, + } + ), + ) + monkeypatch.setattr( + service, + "run_failed_apply_check_mode_replay_once", + retry_replayer, + ) + + result = await service.run_pending_check_modes_once(limit=1) + + assert result["claimed"] == 0 + assert result["repair_receipt_backfill_priority_tick"] is True + assert result["repair_receipt_closure_written"] == 1 + retry_replayer.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_signal_worker_retry_preflight_is_query_only( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + service, + "_load_open_failed_apply_retry_row", + AsyncMock(return_value={"op_id": "apply-op"}), + ) + + result = await service.preflight_failed_apply_retry_queue_once( + project_id="awoooi", + window_hours=24, + ) + + assert result["scanned"] == 1 + assert result["replayed"] == 0 + assert result["query_only"] is True + assert result["runtime_apply_executed"] is False + assert result["execution_owner"] == "awoooi-ansible-executor-broker" + + +def test_candidate_worker_defaults_to_query_only_retry_preflight() -> None: + defaults = candidate_job.enqueue_missing_ansible_candidates_once.__kwdefaults__ + + assert defaults is not None + assert defaults["retry_replayer"] is service.preflight_failed_apply_retry_queue_once diff --git a/apps/api/tests/test_awooop_truth_chain_service.py b/apps/api/tests/test_awooop_truth_chain_service.py index fc75a8cb2..1b063f0c9 100644 --- a/apps/api/tests/test_awooop_truth_chain_service.py +++ b/apps/api/tests/test_awooop_truth_chain_service.py @@ -5,6 +5,7 @@ import os from datetime import UTC, datetime, timedelta from pathlib import Path from types import SimpleNamespace +from unittest.mock import AsyncMock import pytest @@ -2071,6 +2072,11 @@ async def test_catalog_drift_query_failure_does_not_block_fresh_candidate_claims "_expire_stale_ansible_execution_capabilities", no_expired_capabilities, ) + monkeypatch.setattr( + service, + "backfill_missing_auto_repair_execution_receipts_once", + AsyncMock(return_value={"scanned": 0, "written": 0, "error": None}), + ) monkeypatch.setattr(service, "_runtime_blockers", lambda: []) monkeypatch.setattr( service, @@ -2133,6 +2139,11 @@ async def test_execution_broker_runs_failed_apply_retry_before_all_candidate_cla "_expire_stale_ansible_execution_capabilities", no_expired_capabilities, ) + monkeypatch.setattr( + service, + "backfill_missing_auto_repair_execution_receipts_once", + AsyncMock(return_value={"scanned": 0, "written": 0, "error": None}), + ) monkeypatch.setattr(service, "_runtime_blockers", lambda: []) monkeypatch.setattr( service,