From bd27d4e759042940ec34605894d7741808dd807c Mon Sep 17 00:00:00 2001 From: ogt Date: Sat, 11 Jul 2026 16:09:31 +0800 Subject: [PATCH] fix(agent): keep closure lanes moving safely --- .../awooop_ansible_candidate_backfill_job.py | 31 +- .../services/awooop_ansible_audit_service.py | 173 ++- .../awooop_ansible_check_mode_service.py | 1287 ++++++++++++++--- .../tests/test_ansible_verified_closure.py | 443 +++++- ...t_awooop_ansible_candidate_backfill_job.py | 18 +- .../tests/test_awooop_truth_chain_service.py | 42 +- 6 files changed, 1675 insertions(+), 319 deletions(-) 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 6dcaa4b93..c34e260c4 100644 --- a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py +++ b/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py @@ -25,6 +25,7 @@ from src.db.base import get_db_context from src.services.awooop_ansible_audit_service import ( build_ansible_decision_audit_payload, record_ansible_decision_audit, + verified_ansible_candidate_terminal_sql, ) from src.services.awooop_ansible_check_mode_service import ( preflight_failed_apply_retry_queue_once, @@ -63,10 +64,13 @@ async def _fetch_missing_candidate_incidents( window_hours: int, scan_limit: int, ) -> list[dict[str, Any]]: + terminal_candidate_sql = verified_ansible_candidate_terminal_sql( + "existing" + ) async with get_db_context(project_id) as db: await db.execute(text("SET LOCAL statement_timeout = '5000ms'")) result = await db.execute( - text(""" + text(f""" SELECT incident_id, project_id, @@ -107,13 +111,25 @@ 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' + AND existing.op_id = ( + SELECT latest.op_id + FROM automation_operation_log latest + WHERE latest.operation_type + = 'ansible_candidate_matched' + AND latest.created_at >= NOW() - ( + :window_hours * INTERVAL '1 hour' + ) + AND latest.input ->> 'executor' = 'ansible' + AND latest.input ->> 'decision_path' + = 'repair_candidate_controlled_queue' + AND coalesce( + latest.incident_id::text, + latest.input ->> 'incident_id' + ) = incidents.incident_id::text + ORDER BY latest.created_at DESC, latest.op_id DESC + LIMIT 1 ) + AND NOT ({terminal_candidate_sql}) ) ORDER BY created_at DESC LIMIT :scan_limit @@ -843,7 +859,6 @@ async def enqueue_missing_ansible_candidates_once( ) if stats["failed_apply_retry_priority_tick"]: logger.info("awooop_ansible_failed_apply_retry_priority_tick", **stats) - return stats if receipt_backfiller is not None: receipt_stats = await receipt_backfiller( diff --git a/apps/api/src/services/awooop_ansible_audit_service.py b/apps/api/src/services/awooop_ansible_audit_service.py index f842b2812..41336b060 100644 --- a/apps/api/src/services/awooop_ansible_audit_service.py +++ b/apps/api/src/services/awooop_ansible_audit_service.py @@ -9,6 +9,7 @@ from __future__ import annotations import json import os +import re from typing import Any from uuid import UUID, uuid4 @@ -35,6 +36,152 @@ ANSIBLE_OPERATION_TYPES = frozenset({ "ansible_execution_skipped", }) + +def verified_ansible_candidate_terminal_sql(candidate_alias: str) -> str: + """Return the shared SQL predicate for a safely terminal candidate.""" + + if re.fullmatch(r"[a-z_][a-z0-9_]*", candidate_alias) is None: + raise ValueError("invalid_candidate_sql_alias") + candidate = candidate_alias + incident_id = ( + f"coalesce({candidate}.incident_id::text, " + f"{candidate}.input ->> 'incident_id')" + ) + automation_run_id = ( + f"coalesce(nullif({candidate}.input ->> 'automation_run_id', ''), " + f"{candidate}.op_id::text)" + ) + template = """ + EXISTS ( + SELECT 1 + FROM automation_operation_log direct_terminal + WHERE direct_terminal.parent_op_id = __CANDIDATE__.op_id + AND direct_terminal.operation_type = 'ansible_execution_skipped' + AND direct_terminal.status = 'dry_run' + AND direct_terminal.dry_run_result ->> 'skipped' = 'true' + AND coalesce( + direct_terminal.dry_run_result ->> 'apply_executed', + 'false' + ) = 'false' + ) + OR EXISTS ( + SELECT 1 + FROM automation_operation_log check_mode + JOIN automation_operation_log apply + ON apply.parent_op_id = check_mode.op_id + AND apply.operation_type = 'ansible_apply_executed' + AND apply.status = 'failed' + JOIN automation_operation_log replay + ON replay.parent_op_id = apply.op_id + AND replay.operation_type = 'ansible_execution_skipped' + AND replay.status IN ('success', 'failed') + AND replay.input ->> 'execution_mode' + = 'controlled_retry_check_mode_replay' + AND replay.output ->> 'runtime_apply_executed' = 'false' + WHERE check_mode.parent_op_id = __CANDIDATE__.op_id + AND check_mode.operation_type = 'ansible_check_mode_executed' + AND check_mode.status = 'success' + AND check_mode.input ->> 'automation_run_id' + = __AUTOMATION_RUN_ID__ + AND coalesce( + check_mode.incident_id::text, + check_mode.input ->> 'incident_id' + ) = __INCIDENT_ID__ + AND apply.input ->> 'automation_run_id' = __AUTOMATION_RUN_ID__ + AND coalesce( + apply.incident_id::text, + apply.input ->> 'incident_id' + ) = __INCIDENT_ID__ + AND replay.input ->> 'automation_run_id' = __AUTOMATION_RUN_ID__ + AND coalesce( + replay.incident_id::text, + replay.input ->> 'incident_id' + ) = __INCIDENT_ID__ + AND replay.output ->> 'terminal_disposition' = CASE + WHEN replay.status = 'success' + THEN 'no_write_replay_passed_waiting_repair_candidate' + ELSE 'no_write_replay_failed_waiting_playbook_or_transport_repair' + END + AND EXISTS ( + SELECT 1 + FROM incident_evidence verifier + WHERE verifier.incident_id = __INCIDENT_ID__ + AND verifier.post_execution_state ->> 'apply_op_id' + = apply.op_id::text + AND verifier.verification_result IN ('failed', 'timeout') + ) + AND EXISTS ( + SELECT 1 + FROM jsonb_array_elements( + coalesce( + apply.input -> 'runtime_stage_receipts', + '[]'::jsonb + ) + ) receipt(value) + WHERE receipt.value ->> 'stage_id' = 'retry_or_rollback' + AND receipt.value ->> 'automation_run_id' + = __AUTOMATION_RUN_ID__ + AND receipt.value #>> '{detail,failed_apply_op_id}' + = apply.op_id::text + AND receipt.value #>> '{detail,retry_check_mode_op_id}' + = replay.op_id::text + AND receipt.value #>> '{detail,terminal_type}' + = replay.output ->> 'terminal_disposition' + AND receipt.value #>> '{detail,verified_no_write_terminal}' + = 'true' + AND receipt.value #>> '{detail,runtime_apply_executed}' + = 'false' + AND receipt.value ->> 'durable_receipt' = 'true' + ) + AND EXISTS ( + SELECT 1 + FROM jsonb_array_elements( + coalesce( + apply.input -> 'runtime_stage_receipts', + '[]'::jsonb + ) + ) closure_receipt(value) + WHERE closure_receipt.value ->> 'stage_id' + = 'incident_closure' + AND closure_receipt.value ->> 'automation_run_id' + = __AUTOMATION_RUN_ID__ + AND closure_receipt.value #>> '{detail,apply_op_id}' + = apply.op_id::text + AND closure_receipt.value #>> '{detail,retry_op_id}' + = replay.op_id::text + AND closure_receipt.value #>> '{detail,terminal_type}' + = replay.output ->> 'terminal_disposition' + AND closure_receipt.value #>> + '{detail,no_write_terminal}' = 'true' + AND closure_receipt.value #>> + '{detail,proposal_executed}' = 'true' + AND closure_receipt.value #>> + '{detail,execution_success}' = 'false' + AND closure_receipt.value #>> + '{detail,telegram_receipt_acknowledged}' = 'true' + AND closure_receipt.value #>> + '{detail,repository_readback_verified}' = 'true' + AND closure_receipt.value ->> 'durable_receipt' = 'true' + ) + AND 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 FALSE + AND lifecycle.context ->> 'apply_op_id' + = apply.op_id::text + AND lifecycle.context ->> 'automation_run_id' + = __AUTOMATION_RUN_ID__ + ) + ) + """ + return ( + template.replace("__CANDIDATE__", candidate) + .replace("__INCIDENT_ID__", incident_id) + .replace("__AUTOMATION_RUN_ID__", automation_run_id) + ) + _CATALOG: tuple[dict[str, Any], ...] = ( { "catalog_id": "ansible:awoooi-auto-repair-canary", @@ -730,6 +877,9 @@ async def record_ansible_decision_audit( except (TypeError, ValueError): candidate_op_id = str(uuid4()) payload["input"]["automation_run_id"] = candidate_op_id + terminal_candidate_sql = verified_ansible_candidate_terminal_sql( + "candidate" + ) try: async with get_db_context(str(project_id)) as db: await db.execute( @@ -741,7 +891,7 @@ async def record_ansible_decision_audit( {"idempotency_key": payload["input"]["idempotency_key"]}, ) existing = await db.execute( - text(""" + text(f""" SELECT candidate.op_id FROM automation_operation_log candidate WHERE candidate.operation_type = 'ansible_candidate_matched' @@ -752,13 +902,22 @@ async def record_ansible_decision_audit( AND candidate.input ->> 'executor' = 'ansible' AND candidate.input ->> 'decision_path' = 'repair_candidate_controlled_queue' - AND NOT EXISTS ( - SELECT 1 - FROM automation_operation_log terminal - WHERE terminal.parent_op_id = candidate.op_id - AND terminal.operation_type - = 'ansible_execution_skipped' + AND candidate.op_id = ( + SELECT latest.op_id + FROM automation_operation_log latest + WHERE latest.operation_type + = 'ansible_candidate_matched' + AND coalesce( + latest.incident_id::text, + latest.input ->> 'incident_id' + ) = :incident_id + AND latest.input ->> 'executor' = 'ansible' + AND latest.input ->> 'decision_path' + = 'repair_candidate_controlled_queue' + ORDER BY latest.created_at DESC, latest.op_id DESC + LIMIT 1 ) + AND NOT ({terminal_candidate_sql}) LIMIT 1 """), {"incident_id": incident_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 de29d0e59..88ed2eed7 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -18,7 +18,7 @@ from dataclasses import dataclass, replace from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any -from uuid import UUID +from uuid import UUID, uuid4 import structlog from sqlalchemy import select, text @@ -2128,6 +2128,9 @@ async def _read_verified_apply_closure_prerequisites( 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 outbound.source_envelope #>> '{callback_reply,action}' = 'controlled_apply_result' @@ -2336,6 +2339,317 @@ async def _read_incident_closure_readback( } +def _incident_terminal_disposition_payload( + claim: AnsibleCheckModeClaim, + *, + apply_op_id: str, + terminal_type: str, + success_terminal: bool, + retry_op_id: str | None, + telegram_receipt_acknowledged: bool, +) -> dict[str, Any]: + return { + "schema_version": "ansible_incident_terminal_disposition_v1", + "automation_run_id": _automation_run_id_for_claim(claim), + "apply_op_id": apply_op_id, + "retry_op_id": retry_op_id, + "terminal_type": terminal_type, + "incident_status": "RESOLVED" if success_terminal else "MITIGATING", + "incident_resolved": success_terminal, + "no_write_terminal": not success_terminal, + "safe_next_action": ( + "keep_verified_repair_and_monitor_recurrence" + if success_terminal + else "queue_ai_playbook_or_transport_repair_candidate" + ), + "telegram_receipt_acknowledged": telegram_receipt_acknowledged, + "raw_log_payload_stored": False, + "secret_value_stored": False, + } + + +async def _commit_verified_incident_closure( + claim: AnsibleCheckModeClaim, + *, + apply_op_id: str, + project_id: str, + closure_prerequisite_count: int, +) -> dict[str, Any] | None: + """Atomically persist the success terminal, receipt, and lifecycle row.""" + + terminal = { + **_incident_terminal_disposition_payload( + claim, + apply_op_id=apply_op_id, + terminal_type="controlled_apply_verified_closed", + success_terminal=True, + retry_op_id=None, + telegram_receipt_acknowledged=True, + ), + "incident_id": claim.incident_id, + "proposal_executed": True, + "execution_success": True, + "repository_write_acknowledged": True, + "repository_readback_verified": True, + "durable_write_acknowledged": True, + "writer_source_sha": os.getenv( + "AWOOOI_BUILD_COMMIT_SHA", + "", + ).strip().lower() + or None, + } + closure_receipt = _runtime_stage_receipt( + claim, + stage_id="incident_closure", + evidence_ref=f"incidents:{claim.incident_id}:automation_terminal", + detail={ + **terminal, + "closure_prerequisite_count": closure_prerequisite_count, + "closure_prerequisites_verified": True, + }, + ) + automation_run_id = _automation_run_id_for_claim(claim) + lifecycle_context = { + "automation_run_id": automation_run_id, + "catalog_id": claim.catalog_id, + "check_mode_op_id": claim.op_id, + "apply_op_id": apply_op_id, + "single_writer_executor": "awoooi-ansible-executor-broker", + "post_verifier_passed": True, + } + try: + async with get_db_context(project_id) as db: + await db.execute( + text(""" + SELECT pg_advisory_xact_lock( + hashtextextended(:closure_lock_key, 0) + ) + """), + {"closure_lock_key": f"ansible-closure:{apply_op_id}"}, + ) + committed = await db.execute( + text(""" + WITH target_apply AS ( + SELECT apply.op_id + FROM automation_operation_log apply + WHERE apply.op_id = CAST(:apply_op_id AS uuid) + AND apply.operation_type = 'ansible_apply_executed' + AND apply.status = 'success' + AND coalesce( + apply.incident_id::text, + apply.input ->> 'incident_id' + ) = :incident_id + AND apply.input ->> 'automation_run_id' + = :automation_run_id + AND coalesce( + apply.output + ->> 'independent_post_verifier_passed', + 'false' + ) = 'true' + FOR UPDATE + ), + lifecycle_existing AS ( + SELECT lifecycle.id + 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 + ), + lifecycle_inserted AS ( + INSERT INTO alert_operation_log ( + incident_id, + approval_id, + event_type, + actor, + action_detail, + success, + context + ) + SELECT + CAST(:incident_id AS varchar(30)), + CAST(:approval_id AS varchar(36)), + 'RESOLVED'::alert_event_type, + 'awoooi-ansible-executor-broker', + 'controlled_apply_verified_closed', + true, + CAST(:lifecycle_context AS jsonb) + FROM target_apply + WHERE NOT EXISTS ( + SELECT 1 FROM lifecycle_existing + ) + RETURNING id + ), + lifecycle_ready AS ( + SELECT id FROM lifecycle_existing + UNION ALL + SELECT id FROM lifecycle_inserted + ), + apply_updated AS ( + UPDATE automation_operation_log apply + SET input = jsonb_set( + coalesce(apply.input, '{}'::jsonb), + '{runtime_stage_receipts}', + coalesce( + ( + SELECT jsonb_agg( + deduplicated.receipt + ORDER BY deduplicated.stage_id + ) + FROM ( + SELECT DISTINCT ON ( + merged.receipt ->> 'stage_id' + ) + merged.receipt, + merged.receipt ->> 'stage_id' + AS stage_id + FROM jsonb_array_elements( + coalesce( + apply.input + -> 'runtime_stage_receipts', + '[]'::jsonb + ) || CAST(:receipts AS jsonb) + ) WITH ORDINALITY + AS merged(receipt, position) + WHERE merged.receipt ->> 'stage_id' + IS NOT NULL + ORDER BY + merged.receipt ->> 'stage_id', + merged.position DESC + ) deduplicated + ), + '[]'::jsonb + ), + true + ) + FROM target_apply + WHERE apply.op_id = target_apply.op_id + AND EXISTS (SELECT 1 FROM lifecycle_ready) + RETURNING apply.op_id + ), + incident_updated AS ( + UPDATE incidents incident + SET status = CASE + WHEN upper(incident.status::text) = 'CLOSED' + THEN incident.status + ELSE 'RESOLVED'::incidentstatus + END, + outcome = CAST( + coalesce( + CASE + WHEN incident.outcome IS NULL + THEN '{}'::jsonb + WHEN jsonb_typeof( + CAST(incident.outcome AS jsonb) + ) = 'object' + THEN CAST(incident.outcome AS jsonb) + ELSE jsonb_build_object( + 'legacy_outcome', + CAST(incident.outcome AS jsonb) + ) + END, + '{}'::jsonb + ) + || jsonb_build_object( + 'automation_terminal', + CAST(:terminal_payload AS jsonb), + 'proposal_executed', + true, + 'execution_success', + true + ) + AS json + ), + resolved_at = coalesce(incident.resolved_at, NOW()), + updated_at = NOW() + WHERE incident.incident_id = :incident_id + AND incident.project_id = :project_id + AND EXISTS (SELECT 1 FROM apply_updated) + AND ( + upper(incident.status::text) + NOT IN ('RESOLVED', 'CLOSED') + OR cast(incident.outcome AS jsonb) #>> + '{automation_terminal,automation_run_id}' + = :automation_run_id + ) + RETURNING + incident.status::text AS incident_status, + incident.updated_at, + incident.resolved_at, + incident.outcome + ) + SELECT + incident_status, + updated_at, + resolved_at, + outcome, + EXISTS (SELECT 1 FROM apply_updated) + AS closure_receipt_written, + EXISTS (SELECT 1 FROM lifecycle_ready) + AS resolved_lifecycle_written + FROM incident_updated + """), + { + "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_for_claim(claim) or None, + "lifecycle_context": json.dumps( + lifecycle_context, + ensure_ascii=False, + ), + "receipts": json.dumps( + [closure_receipt], + ensure_ascii=False, + ), + "terminal_payload": json.dumps( + terminal, + ensure_ascii=False, + ), + }, + ) + row = committed.mappings().one_or_none() + if row is None: + raise RuntimeError("verified_incident_closure_atomic_write_pending") + outcome = _json_loads(row.get("outcome")) + readback = outcome.get("automation_terminal") + if not ( + isinstance(readback, Mapping) + and readback.get("automation_run_id") == automation_run_id + and readback.get("apply_op_id") == apply_op_id + and outcome.get("proposal_executed") is True + and outcome.get("execution_success") is True + and row.get("closure_receipt_written") is True + and row.get("resolved_lifecycle_written") is True + and row.get("resolved_at") is not None + ): + raise RuntimeError("verified_incident_closure_atomic_readback_failed") + return { + **terminal, + "incident_status": str(row.get("incident_status") or ""), + "incident_row_version": ( + row["updated_at"].isoformat() + if row.get("updated_at") is not None + else None + ), + "closure_receipt_written": True, + "resolved_lifecycle_written": True, + } + except Exception as exc: + logger.warning( + "ansible_verified_incident_closure_atomic_write_failed", + automation_run_id=automation_run_id, + incident_id=claim.incident_id, + apply_op_id=apply_op_id, + error_type=type(exc).__name__, + ) + return None + + async def _finalize_verified_apply_closure( claim: AnsibleCheckModeClaim, *, @@ -2354,56 +2668,20 @@ async def _finalize_verified_apply_closure( "missing": list(prerequisites.get("missing") or []), } - terminal = await _record_incident_terminal_disposition( + terminal = await _commit_verified_incident_closure( claim, apply_op_id=apply_op_id, project_id=project_id, - terminal_type="controlled_apply_verified_closed", - success_terminal=True, - telegram_receipt_acknowledged=True, + closure_prerequisite_count=len(prerequisites["receipts"]), ) 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", + "status": "atomic_closure_write_pending", "closed": False, "missing": [ - name - for name, ready in { - "incident_closure_receipt": receipt_written, - "resolved_lifecycle": resolved_lifecycle, - }.items() - if not ready + "incident_terminal_readback", + "incident_closure_receipt", + "resolved_lifecycle", ], } @@ -2434,56 +2712,66 @@ async def _record_incident_terminal_disposition( ) -> dict[str, Any] | None: automation_run_id = _automation_run_id_for_claim(claim) incident_status = "RESOLVED" if success_terminal else "MITIGATING" - safe_next_action = ( - "keep_verified_repair_and_monitor_recurrence" - if success_terminal - else "queue_ai_playbook_or_transport_repair_candidate" + terminal_payload = _incident_terminal_disposition_payload( + claim, + apply_op_id=apply_op_id, + terminal_type=terminal_type, + success_terminal=success_terminal, + retry_op_id=retry_op_id, + telegram_receipt_acknowledged=telegram_receipt_acknowledged, ) - terminal_payload = { - "schema_version": "ansible_incident_terminal_disposition_v1", - "automation_run_id": automation_run_id, - "apply_op_id": apply_op_id, - "retry_op_id": retry_op_id, - "terminal_type": terminal_type, - "incident_status": incident_status, - "incident_resolved": success_terminal, - "no_write_terminal": not success_terminal, - "safe_next_action": safe_next_action, - "telegram_receipt_acknowledged": telegram_receipt_acknowledged, - "raw_log_payload_stored": False, - "secret_value_stored": False, - } try: async with get_db_context(project_id) as db: + execution = await db.execute( + text(""" + SELECT coalesce( + (apply.dry_run_result ->> 'apply_executed')::boolean, + (apply.output + ->> 'controlled_apply_executed')::boolean, + false + ) AS proposal_executed + FROM automation_operation_log apply + 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}, + ) + execution_row = execution.mappings().one_or_none() + if execution_row is None: + return None + proposal_executed = execution_row.get("proposal_executed") is True + execution_success: bool | None = ( + success_terminal if proposal_executed else None + ) + if success_terminal and not proposal_executed: + return None updated = await db.execute( text(""" UPDATE incidents SET status = CAST(:incident_status AS incidentstatus), outcome = CAST( - jsonb_set( + coalesce( CASE WHEN outcome IS NULL THEN '{}'::jsonb WHEN jsonb_typeof(CAST(outcome AS jsonb)) - = 'object' + = 'object' THEN CAST(outcome AS jsonb) ELSE jsonb_build_object( 'legacy_outcome', CAST(outcome AS jsonb) ) END, - '{automation_terminal}', - CAST(:terminal_payload AS jsonb), - true + '{}'::jsonb + ) + || jsonb_build_object( + 'automation_terminal', + CAST(:terminal_payload AS jsonb), + 'proposal_executed', + CAST(:proposal_executed AS boolean), + 'execution_success', + CAST(:execution_success AS boolean) ) - || CASE - WHEN :success_terminal THEN jsonb_build_object( - 'proposal_executed', - true, - 'execution_success', - true - ) - ELSE '{}'::jsonb - END AS json ), resolved_at = CASE @@ -2510,6 +2798,8 @@ async def _record_incident_terminal_disposition( ensure_ascii=False, ), "success_terminal": success_terminal, + "proposal_executed": proposal_executed, + "execution_success": execution_success, "incident_id": claim.incident_id, "project_id": project_id, }, @@ -2521,9 +2811,9 @@ async def _record_incident_terminal_disposition( readback = outcome.get("automation_terminal") if not isinstance(readback, Mapping): return None - if success_terminal and not ( - outcome.get("proposal_executed") is True - and outcome.get("execution_success") is True + if ( + outcome.get("proposal_executed") is not proposal_executed + or outcome.get("execution_success") is not execution_success ): return None if ( @@ -2536,8 +2826,8 @@ async def _record_incident_terminal_disposition( **terminal_payload, "incident_id": claim.incident_id, "incident_status": str(row.get("incident_status") or ""), - "proposal_executed": outcome.get("proposal_executed") is True, - "execution_success": outcome.get("execution_success") is True, + "proposal_executed": proposal_executed, + "execution_success": execution_success, "incident_row_version": ( row["updated_at"].isoformat() if row.get("updated_at") is not None @@ -2563,6 +2853,118 @@ async def _record_incident_terminal_disposition( return None +def _build_retry_runtime_stage_receipt( + claim: AnsibleCheckModeClaim, + result: AnsibleRunResult, + *, + apply_op_id: str, + replay_op_id: str, + derived_from_durable_chain: bool = False, + check_mode_replay_performed: bool = True, +) -> dict[str, Any]: + terminal_type = ( + "no_write_replay_passed_waiting_repair_candidate" + if result.returncode == 0 + else "no_write_replay_failed_waiting_playbook_or_transport_repair" + ) + return _runtime_stage_receipt( + claim, + stage_id="retry_or_rollback", + evidence_ref=f"automation_operation_log:{replay_op_id}:dry_run_result", + derived_from_durable_chain=derived_from_durable_chain, + detail={ + "schema_version": "ansible_controlled_retry_terminal_v2", + "failed_apply_op_id": apply_op_id, + "retry_check_mode_op_id": replay_op_id, + "terminal_type": terminal_type, + "check_mode_replay_performed": check_mode_replay_performed, + "check_mode_replay_returncode": result.returncode, + "runtime_apply_executed": False, + "rollback_performed": False, + "verified_no_write_terminal": True, + "retry_operation_readback_verified": True, + "retry_attempt": 1, + "max_retry_attempts": _CONTROLLED_RETRY_MAX_ATTEMPTS, + "retry_idempotency_key": ( + f"ansible-retry:{apply_op_id}:check-mode" + ), + "retry_error_backoff_min_seconds": ( + _CONTROLLED_RETRY_ERROR_BACKOFF_MIN_SECONDS + ), + "retry_error_backoff_max_seconds": ( + _CONTROLLED_RETRY_ERROR_BACKOFF_MAX_SECONDS + ), + "bounded_retry": True, + "repair_candidate_required": True, + "safe_next_action": ( + "queue_ai_playbook_or_transport_repair_candidate" + ), + "repository_readback_verified": True, + "durable_write_acknowledged": True, + "writer_source_sha": os.getenv( + "AWOOOI_BUILD_COMMIT_SHA", + "", + ).strip().lower() + or None, + "raw_log_payload_stored": False, + "secret_value_stored": False, + }, + ) + + +async def _retry_telegram_receipt_acknowledged( + claim: AnsibleCheckModeClaim, + *, + project_id: str, +) -> bool: + automation_run_id = _automation_run_id_for_claim(claim) + try: + async with get_db_context(project_id) as db: + result = await db.execute( + text(""" + SELECT 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}', + outbound.source_envelope #>> + '{source_refs,automation_run_ids,0}' + ) = :automation_run_id + AND coalesce( + outbound.source_envelope #>> + '{callback_reply,incident_id}', + outbound.source_envelope #>> + '{source_refs,incident_ids,0}' + ) = :incident_id + ) + """), + { + "project_id": project_id, + "automation_run_id": automation_run_id, + "incident_id": claim.incident_id, + }, + ) + return result.scalar() is True + except Exception as exc: + logger.warning( + "ansible_retry_telegram_receipt_readback_failed", + automation_run_id=automation_run_id, + incident_id=claim.incident_id, + error_type=type(exc).__name__, + ) + return False + + async def _record_retry_runtime_stage_receipt( claim: AnsibleCheckModeClaim, result: AnsibleRunResult, @@ -2570,6 +2972,7 @@ async def _record_retry_runtime_stage_receipt( apply_op_id: str, replay_op_id: str, project_id: str, + check_mode_replay_performed: bool = True, ) -> bool: """Append the verified no-write retry terminal to the original apply run.""" @@ -2579,7 +2982,6 @@ async def _record_retry_runtime_stage_receipt( if result.returncode == 0 else "no_write_replay_failed_waiting_playbook_or_transport_repair" ) - retry_idempotency_key = f"ansible-retry:{apply_op_id}:check-mode" replay_readback_verified = False telegram_receipt_acknowledged = False try: @@ -2622,6 +3024,7 @@ async def _record_retry_runtime_stage_receipt( WHERE project_id = :project_id AND channel_type = 'telegram' AND send_status = 'sent' + AND provider_message_id IS NOT NULL AND source_envelope #>> '{callback_reply,action}' = 'controlled_apply_result' AND COALESCE( @@ -2631,11 +3034,18 @@ async def _record_retry_runtime_stage_receipt( source_envelope #>> '{source_refs,automation_run_ids,0}' ) = :automation_run_id + AND COALESCE( + source_envelope #>> + '{callback_reply,incident_id}', + source_envelope #>> + '{source_refs,incident_ids,0}' + ) = :incident_id ) """), { "project_id": project_id, "automation_run_id": automation_run_id, + "incident_id": claim.incident_id, }, ) telegram_receipt_acknowledged = telegram_readback.scalar() is True @@ -2657,45 +3067,12 @@ async def _record_retry_runtime_stage_receipt( ) return False - retry_receipt = _runtime_stage_receipt( + retry_receipt = _build_retry_runtime_stage_receipt( claim, - stage_id="retry_or_rollback", - evidence_ref=f"automation_operation_log:{replay_op_id}:dry_run_result", - detail={ - "schema_version": "ansible_controlled_retry_terminal_v2", - "failed_apply_op_id": apply_op_id, - "retry_check_mode_op_id": replay_op_id, - "terminal_type": terminal_type, - "check_mode_replay_performed": True, - "check_mode_replay_returncode": result.returncode, - "runtime_apply_executed": False, - "rollback_performed": False, - "verified_no_write_terminal": replay_readback_verified, - "retry_operation_readback_verified": replay_readback_verified, - "retry_attempt": 1, - "max_retry_attempts": _CONTROLLED_RETRY_MAX_ATTEMPTS, - "retry_idempotency_key": retry_idempotency_key, - "retry_error_backoff_min_seconds": ( - _CONTROLLED_RETRY_ERROR_BACKOFF_MIN_SECONDS - ), - "retry_error_backoff_max_seconds": ( - _CONTROLLED_RETRY_ERROR_BACKOFF_MAX_SECONDS - ), - "bounded_retry": True, - "repair_candidate_required": True, - "safe_next_action": ( - "queue_ai_playbook_or_transport_repair_candidate" - ), - "repository_readback_verified": replay_readback_verified, - "durable_write_acknowledged": replay_readback_verified, - "writer_source_sha": os.getenv( - "AWOOOI_BUILD_COMMIT_SHA", - "", - ).strip().lower() - or None, - "raw_log_payload_stored": False, - "secret_value_stored": False, - }, + result, + apply_op_id=apply_op_id, + replay_op_id=replay_op_id, + check_mode_replay_performed=check_mode_replay_performed, ) incident_detail = await _record_incident_terminal_disposition( claim, @@ -3345,6 +3722,8 @@ async def backfill_missing_auto_repair_execution_receipts_once( "telegram_receipt_acknowledged": 0, "retry_terminal_projection_scanned": 0, "retry_terminal_projection_written": 0, + "retry_terminal_projection_retry_receipt_written": 0, + "retry_terminal_projection_telegram_receipt_acknowledged": 0, "retry_terminal_projection_incident_receipt_written": 0, "retry_terminal_projection_lifecycle_written": 0, "retry_terminal_projection_verified": 0, @@ -3364,6 +3743,12 @@ async def backfill_missing_auto_repair_execution_receipts_once( stats["retry_terminal_projection_written"] = int( projection.get("written") or 0 ) + stats["retry_terminal_projection_retry_receipt_written"] = int( + projection.get("retry_receipt_written") or 0 + ) + stats[ + "retry_terminal_projection_telegram_receipt_acknowledged" + ] = int(projection.get("telegram_receipt_acknowledged") or 0) stats["retry_terminal_projection_incident_receipt_written"] = int( projection.get("incident_receipt_written") or 0 ) @@ -3619,6 +4004,7 @@ async def _load_open_failed_apply_retry_row( *, project_id: str, window_hours: int, + stale_after_seconds: int | None = None, ) -> dict[str, Any] | None: """Read one durable failed-apply retry candidate before transport checks.""" @@ -3638,8 +4024,30 @@ async def _load_open_failed_apply_retry_row( apply.error, apply.duration_ms, apply.status, - apply.created_at + apply.created_at, + replay.op_id::text AS stale_replay_op_id, + replay.input AS stale_replay_input, + replay.created_at AS stale_replay_created_at, + CASE + WHEN replay.input ->> 'retry_recovery_count' + ~ '^[0-9]+$' + THEN (replay.input + ->> 'retry_recovery_count')::integer + ELSE 0 + END AS stale_replay_recovery_count FROM automation_operation_log apply + LEFT JOIN LATERAL ( + SELECT pending.op_id, pending.input, pending.status, + pending.created_at + FROM automation_operation_log pending + WHERE pending.operation_type + = 'ansible_execution_skipped' + AND pending.parent_op_id = apply.op_id + AND pending.input ->> 'execution_mode' + = 'controlled_retry_check_mode_replay' + ORDER BY pending.created_at DESC, pending.op_id DESC + LIMIT 1 + ) replay ON TRUE WHERE apply.operation_type = 'ansible_apply_executed' AND apply.status = 'failed' AND apply.created_at >= NOW() - ( @@ -3654,21 +4062,38 @@ async def _load_open_failed_apply_retry_row( ) AND verifier.post_execution_state ->> 'apply_op_id' = apply.op_id::text - AND verifier.verification_result = 'failed' + AND verifier.verification_result IN ('failed', 'timeout') ) - 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' + AND ( + replay.op_id IS NULL + OR ( + replay.status = 'pending' + AND coalesce( + nullif( + replay.input ->> 'retry_claimed_at', + '' + )::timestamptz, + replay.created_at + ) <= NOW() - ( + :stale_after_seconds * INTERVAL '1 second' + ) + ) ) ORDER BY apply.created_at DESC LIMIT 1 - FOR UPDATE SKIP LOCKED + FOR UPDATE OF apply SKIP LOCKED """), - {"window_hours": max(1, window_hours)}, + { + "window_hours": max(1, window_hours), + "stale_after_seconds": max( + 300, + stale_after_seconds + or ( + settings.AWOOOP_ANSIBLE_CHECK_MODE_TIMEOUT_SECONDS + + 120 + ), + ), + }, ) row = result.mappings().one_or_none() return dict(row) if row is not None else None @@ -3741,7 +4166,17 @@ async def _load_missing_retry_terminal_projection_rows( apply.status, apply.created_at, replay.op_id::text AS replay_op_id, - retry_receipt.receipt #>> '{detail,terminal_type}' + replay.status AS replay_status, + replay.output AS replay_output, + replay.dry_run_result AS replay_dry_run_result, + replay.error AS replay_error, + replay.duration_ms AS replay_duration_ms, + retry_receipt.receipt IS NOT NULL + AS retry_receipt_present, + coalesce( + retry_receipt.receipt #>> '{detail,terminal_type}', + replay.output ->> 'terminal_disposition' + ) AS terminal_type, EXISTS ( SELECT 1 @@ -3750,6 +4185,9 @@ async def _load_missing_retry_terminal_projection_rows( 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', @@ -3776,13 +4214,27 @@ async def _load_missing_retry_terminal_projection_rows( = 'controlled_retry_check_mode_replay' AND replay.status IN ('success', 'failed') AND replay.output ->> 'runtime_apply_executed' = 'false' + AND replay.input ->> 'automation_run_id' + = apply.input ->> 'automation_run_id' + AND coalesce( + replay.incident_id::text, + replay.input ->> 'incident_id' + ) = coalesce( + apply.incident_id::text, + apply.input ->> 'incident_id' + ) + AND replay.output ->> 'terminal_disposition' = CASE + WHEN replay.status = 'success' + THEN 'no_write_replay_passed_waiting_repair_candidate' + ELSE 'no_write_replay_failed_waiting_playbook_or_transport_repair' + END JOIN incidents incident ON incident.incident_id = coalesce( apply.incident_id::text, apply.input ->> 'incident_id' ) AND incident.project_id = :project_id - JOIN LATERAL ( + LEFT JOIN LATERAL ( SELECT receipt.value AS receipt FROM jsonb_array_elements( coalesce( @@ -3791,6 +4243,8 @@ async def _load_missing_retry_terminal_projection_rows( ) ) receipt(value) WHERE receipt.value ->> 'stage_id' = 'retry_or_rollback' + AND receipt.value ->> 'automation_run_id' + = apply.input ->> 'automation_run_id' AND receipt.value #>> '{detail,retry_check_mode_op_id}' = replay.op_id::text AND receipt.value #>> @@ -3799,6 +4253,11 @@ async def _load_missing_retry_terminal_projection_rows( '{detail,verified_no_write_terminal}' = 'true' AND receipt.value #>> '{detail,runtime_apply_executed}' = 'false' + AND receipt.value #>> + '{detail,repository_readback_verified}' = 'true' + AND receipt.value #>> '{detail,terminal_type}' + = replay.output ->> 'terminal_disposition' + AND receipt.value ->> 'durable_receipt' = 'true' LIMIT 1 ) retry_receipt ON TRUE WHERE apply.operation_type = 'ansible_apply_executed' @@ -3814,23 +4273,106 @@ async def _load_missing_retry_terminal_projection_rows( = apply.op_id::text AND verifier.verification_result IN ('failed', 'timeout') ) + AND EXISTS ( + SELECT 1 + FROM automation_operation_log source_check + JOIN automation_operation_log source_candidate + ON source_candidate.op_id = source_check.parent_op_id + AND source_candidate.operation_type + = 'ansible_candidate_matched' + WHERE source_check.op_id = apply.parent_op_id + AND source_check.operation_type + = 'ansible_check_mode_executed' + ) + AND NOT EXISTS ( + SELECT 1 + FROM automation_operation_log source_check + JOIN automation_operation_log source_candidate + ON source_candidate.op_id = source_check.parent_op_id + AND source_candidate.operation_type + = 'ansible_candidate_matched' + JOIN automation_operation_log newer_candidate + ON newer_candidate.operation_type + = 'ansible_candidate_matched' + AND newer_candidate.input ->> 'executor' = 'ansible' + AND newer_candidate.input ->> 'decision_path' + = 'repair_candidate_controlled_queue' + AND coalesce( + newer_candidate.incident_id::text, + newer_candidate.input ->> 'incident_id' + ) = incident.incident_id + AND ( + newer_candidate.created_at, + newer_candidate.op_id + ) > ( + source_candidate.created_at, + source_candidate.op_id + ) + WHERE source_check.op_id = apply.parent_op_id + AND source_check.operation_type + = 'ansible_check_mode_executed' + ) AND ( - cast(incident.outcome AS jsonb) #>> - '{automation_terminal,automation_run_id}' - IS DISTINCT FROM apply.input ->> 'automation_run_id' - OR cast(incident.outcome AS jsonb) #>> - '{automation_terminal,apply_op_id}' - IS DISTINCT FROM apply.op_id::text - OR cast(incident.outcome AS jsonb) #>> - '{automation_terminal,retry_op_id}' - IS DISTINCT FROM replay.op_id::text - OR cast(incident.outcome AS jsonb) #>> - '{automation_terminal,terminal_type}' - IS DISTINCT FROM retry_receipt.receipt #>> - '{detail,terminal_type}' - OR cast(incident.outcome AS jsonb) #>> - '{automation_terminal,no_write_terminal}' - IS DISTINCT FROM 'true' + retry_receipt.receipt IS NULL + OR NOT 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}', + outbound.source_envelope #>> + '{source_refs,automation_run_ids,0}' + ) = apply.input ->> 'automation_run_id' + AND coalesce( + outbound.source_envelope #>> + '{callback_reply,incident_id}', + outbound.source_envelope #>> + '{source_refs,incident_ids,0}' + ) = incident.incident_id + ) + OR NOT EXISTS ( + SELECT 1 + FROM jsonb_array_elements( + coalesce( + apply.input -> 'runtime_stage_receipts', + '[]'::jsonb + ) + ) closure_receipt(value) + WHERE closure_receipt.value ->> 'stage_id' + = 'incident_closure' + AND closure_receipt.value ->> 'automation_run_id' + = apply.input ->> 'automation_run_id' + AND closure_receipt.value #>> + '{detail,apply_op_id}' = apply.op_id::text + AND closure_receipt.value #>> + '{detail,retry_op_id}' = replay.op_id::text + AND closure_receipt.value #>> + '{detail,terminal_type}' + = replay.output ->> 'terminal_disposition' + AND closure_receipt.value #>> + '{detail,no_write_terminal}' = 'true' + AND closure_receipt.value #>> + '{detail,proposal_executed}' = 'true' + AND closure_receipt.value #>> + '{detail,execution_success}' = 'false' + AND closure_receipt.value #>> + '{detail,telegram_receipt_acknowledged}' + = 'true' + AND closure_receipt.value #>> + '{detail,repository_readback_verified}' + = 'true' + AND closure_receipt.value ->> 'durable_receipt' + = 'true' + ) OR NOT EXISTS ( SELECT 1 FROM alert_operation_log lifecycle @@ -3843,6 +4385,41 @@ async def _load_missing_retry_terminal_projection_rows( AND lifecycle.context ->> 'automation_run_id' = apply.input ->> 'automation_run_id' ) + OR ( + upper(incident.status::text) + IS DISTINCT FROM 'MITIGATING' + OR cast(incident.outcome AS jsonb) #>> + '{automation_terminal,schema_version}' + IS DISTINCT FROM + 'ansible_incident_terminal_disposition_v1' + OR cast(incident.outcome AS jsonb) ->> + 'proposal_executed' + IS DISTINCT FROM 'true' + OR cast(incident.outcome AS jsonb) ->> + 'execution_success' + IS DISTINCT FROM 'false' + OR + cast(incident.outcome AS jsonb) #>> + '{automation_terminal,automation_run_id}' + IS DISTINCT FROM apply.input + ->> 'automation_run_id' + OR cast(incident.outcome AS jsonb) #>> + '{automation_terminal,apply_op_id}' + IS DISTINCT FROM apply.op_id::text + OR cast(incident.outcome AS jsonb) #>> + '{automation_terminal,retry_op_id}' + IS DISTINCT FROM replay.op_id::text + OR cast(incident.outcome AS jsonb) #>> + '{automation_terminal,terminal_type}' + IS DISTINCT FROM replay.output + ->> 'terminal_disposition' + OR cast(incident.outcome AS jsonb) #>> + '{automation_terminal,no_write_terminal}' + IS DISTINCT FROM 'true' + OR cast(incident.outcome AS jsonb) #>> + '{automation_terminal,telegram_receipt_acknowledged}' + IS DISTINCT FROM 'true' + ) ) ORDER BY replay.created_at DESC LIMIT :limit @@ -3885,6 +4462,39 @@ async def _verify_retry_terminal_projection_readback( '{automation_terminal,terminal_type}' = :terminal_type AND cast(incident.outcome AS jsonb) #>> '{automation_terminal,no_write_terminal}' = 'true' + AND cast(incident.outcome AS jsonb) #>> + '{automation_terminal,telegram_receipt_acknowledged}' + = 'true' + AND cast(incident.outcome AS jsonb) ->> + 'proposal_executed' = 'true' + AND cast(incident.outcome AS jsonb) ->> + 'execution_success' = 'false' + AND EXISTS ( + SELECT 1 + FROM jsonb_array_elements( + coalesce( + apply.input -> 'runtime_stage_receipts', + '[]'::jsonb + ) + ) retry_receipt(value) + WHERE retry_receipt.value ->> 'stage_id' + = 'retry_or_rollback' + AND retry_receipt.value ->> 'automation_run_id' + = :automation_run_id + AND retry_receipt.value #>> + '{detail,failed_apply_op_id}' = :apply_op_id + AND retry_receipt.value #>> + '{detail,retry_check_mode_op_id}' = :replay_op_id + AND retry_receipt.value #>> + '{detail,terminal_type}' = :terminal_type + AND retry_receipt.value #>> + '{detail,verified_no_write_terminal}' = 'true' + AND retry_receipt.value #>> + '{detail,runtime_apply_executed}' = 'false' + AND retry_receipt.value #>> + '{detail,repository_readback_verified}' = 'true' + AND retry_receipt.value ->> 'durable_receipt' = 'true' + ) AND EXISTS ( SELECT 1 FROM jsonb_array_elements( @@ -3900,6 +4510,12 @@ async def _verify_retry_terminal_projection_readback( = :apply_op_id AND receipt.value #>> '{detail,retry_op_id}' = :replay_op_id + AND receipt.value #>> + '{detail,telegram_receipt_acknowledged}' = 'true' + AND receipt.value #>> + '{detail,proposal_executed}' = 'true' + AND receipt.value #>> + '{detail,execution_success}' = 'false' AND receipt.value #>> '{detail,repository_readback_verified}' = 'true' AND receipt.value ->> 'durable_receipt' = 'true' @@ -3917,6 +4533,25 @@ async def _verify_retry_terminal_projection_readback( FROM incidents incident JOIN automation_operation_log apply ON apply.op_id = CAST(:apply_op_id AS uuid) + AND apply.operation_type = 'ansible_apply_executed' + AND apply.status = 'failed' + AND apply.input ->> 'automation_run_id' + = :automation_run_id + JOIN automation_operation_log replay + ON replay.op_id = CAST(:replay_op_id AS uuid) + AND replay.parent_op_id = apply.op_id + AND replay.operation_type = 'ansible_execution_skipped' + AND replay.status IN ('success', 'failed') + AND replay.input ->> 'execution_mode' + = 'controlled_retry_check_mode_replay' + AND replay.input ->> 'automation_run_id' + = :automation_run_id + AND coalesce( + replay.incident_id::text, + replay.input ->> 'incident_id' + ) = incident.incident_id + AND replay.output ->> 'runtime_apply_executed' = 'false' + AND replay.output ->> 'terminal_disposition' = :terminal_type WHERE incident.incident_id = :incident_id AND incident.project_id = :project_id LIMIT 1 @@ -3944,6 +4579,8 @@ async def backfill_missing_retry_terminal_projections_once( stats: dict[str, Any] = { "scanned": 0, "written": 0, + "retry_receipt_written": 0, + "telegram_receipt_acknowledged": 0, "incident_receipt_written": 0, "lifecycle_written": 0, "verified": 0, @@ -3961,17 +4598,101 @@ async def backfill_missing_retry_terminal_projections_once( reconstructed = _claim_from_apply_operation_row(row) if reconstructed is None: continue - claim, _failed_apply_result = reconstructed + claim, failed_apply_result = reconstructed apply_op_id = str(row.get("op_id") or "") replay_op_id = str(row.get("replay_op_id") or "") terminal_type = str(row.get("terminal_type") or "") + replay_status = str(row.get("replay_status") or "") + expected_terminal_type = ( + "no_write_replay_passed_waiting_repair_candidate" + if replay_status == "success" + else ( + "no_write_replay_failed_waiting_playbook_or_transport_repair" + if replay_status == "failed" + else "" + ) + ) if not ( apply_op_id and replay_op_id - and terminal_type.startswith("no_write_replay_") - and row.get("telegram_receipt_acknowledged") is True + and terminal_type == expected_terminal_type ): continue + telegram_acknowledged = ( + row.get("telegram_receipt_acknowledged") is True + ) + if not telegram_acknowledged: + apply_output = _json_loads(row.get("output")) + receipt_sent = await _send_controlled_apply_telegram_receipt( + claim, + failed_apply_result, + apply_op_id=apply_op_id, + writeback={ + "verification_result": str( + apply_output.get( + "independent_post_verifier_result" + ) + or "failed" + ), + "verification": True, + "learning": False, + }, + project_id=project_id, + ) + if receipt_sent: + telegram_acknowledged = ( + await _retry_telegram_receipt_acknowledged( + claim, + project_id=project_id, + ) + ) + if not telegram_acknowledged: + continue + stats["telegram_receipt_acknowledged"] += 1 + if row.get("retry_receipt_present") is not True: + replay_output = _json_loads(row.get("replay_output")) + replay_dry_run = _json_loads( + row.get("replay_dry_run_result") + ) + raw_returncode = replay_output.get("returncode") + try: + replay_returncode = int(raw_returncode) + except (TypeError, ValueError): + replay_returncode = 0 if replay_status == "success" else 1 + replay_result = AnsibleRunResult( + returncode=replay_returncode, + stdout="", + stderr=str(row.get("replay_error") or ""), + duration_ms=int(row.get("replay_duration_ms") or 0), + timed_out=replay_dry_run.get("timed_out") is True, + post_verifier_passed=False, + ) + retry_receipt_written = ( + await _append_runtime_stage_receipts_to_apply( + apply_op_id=apply_op_id, + receipts=( + _build_retry_runtime_stage_receipt( + claim, + replay_result, + apply_op_id=apply_op_id, + replay_op_id=replay_op_id, + derived_from_durable_chain=True, + check_mode_replay_performed=( + replay_output.get( + "check_mode_replay_performed" + ) + is not False + ), + ), + ), + project_id=project_id, + ) + ) + stats["retry_receipt_written"] += int( + retry_receipt_written + ) + if not retry_receipt_written: + continue terminal = await _record_incident_terminal_disposition( claim, apply_op_id=apply_op_id, @@ -3979,7 +4700,7 @@ async def backfill_missing_retry_terminal_projections_once( terminal_type=terminal_type, success_terminal=False, retry_op_id=replay_op_id, - telegram_receipt_acknowledged=True, + telegram_receipt_acknowledged=telegram_acknowledged, ) if terminal is None: continue @@ -4050,13 +4771,20 @@ async def run_failed_apply_check_mode_replay_once( "check_mode_failed": 0, "runtime_apply_executed": 0, "runtime_stage_receipt_written": 0, + "stale_retry_reclaimed": 0, + "stale_retry_terminalized": 0, "blockers": [], "error": None, } try: + effective_timeout = ( + timeout_seconds or settings.AWOOOP_ANSIBLE_CHECK_MODE_TIMEOUT_SECONDS + ) + stale_after_seconds = max(300, effective_timeout + 120) row = await _load_open_failed_apply_retry_row( project_id=project_id, window_hours=window_hours, + stale_after_seconds=stale_after_seconds, ) if row is None: return stats @@ -4083,6 +4811,8 @@ async def run_failed_apply_check_mode_replay_once( stats["blockers"] = transport_blockers return stats + retry_claim_id = str(uuid4()) + retry_claimed_at = datetime.now(UTC).isoformat() input_payload = { **claim.input_payload, "automation_run_id": automation_run_id, @@ -4096,13 +4826,92 @@ async def run_failed_apply_check_mode_replay_once( "controlled_apply_allowed": True, "controlled_apply_blocker": None, "bounded_execution_scope": "retry_check_mode_only", + "retry_claim_id": retry_claim_id, + "retry_claimed_at": retry_claimed_at, + "retry_recovery_count": 0, "next_controlled_action": ( "queue_ai_playbook_or_transport_repair_candidate" ), } async with get_db_context(project_id) as db: - claimed = await db.execute( + stale_replay_op_id = str( + row.get("stale_replay_op_id") or "" + ) + await db.execute( text(""" + SELECT pg_advisory_xact_lock( + hashtextextended(:retry_claim_lock_key, 0) + ) + """), + { + "retry_claim_lock_key": ( + f"ansible-retry-claim:{apply_op_id}" + ) + }, + ) + if stale_replay_op_id: + claimed = await db.execute( + text(""" + UPDATE automation_operation_log replay + SET input = coalesce(replay.input, '{}'::jsonb) + || jsonb_build_object( + 'retry_claim_id', + CAST(:retry_claim_id AS text), + 'retry_claimed_at', + CAST(:retry_claimed_at AS text), + 'retry_claim_recovered', + true, + 'retry_recovery_count', + CASE + WHEN replay.input + ->> 'retry_recovery_count' + ~ '^[0-9]+$' + THEN (replay.input + ->> 'retry_recovery_count')::integer + + 1 + ELSE 1 + END + ), + dry_run_result = coalesce( + replay.dry_run_result, + '{}'::jsonb + ) || jsonb_build_object( + 'claim_state', + 'reclaimed_after_stale_pending' + ) + WHERE replay.op_id = CAST(:replay_op_id AS uuid) + AND replay.operation_type + = 'ansible_execution_skipped' + AND replay.status = 'pending' + AND replay.parent_op_id + = CAST(:parent_op_id AS uuid) + AND replay.input ->> 'automation_run_id' + = :automation_run_id + AND replay.input ->> 'execution_mode' + = 'controlled_retry_check_mode_replay' + AND coalesce( + nullif( + replay.input ->> 'retry_claimed_at', + '' + )::timestamptz, + replay.created_at + ) <= NOW() - ( + :stale_after_seconds * INTERVAL '1 second' + ) + RETURNING replay.op_id + """), + { + "replay_op_id": stale_replay_op_id, + "retry_claim_id": retry_claim_id, + "retry_claimed_at": retry_claimed_at, + "stale_after_seconds": stale_after_seconds, + "parent_op_id": apply_op_id, + "automation_run_id": automation_run_id, + }, + ) + else: + claimed = await db.execute( + text(""" INSERT INTO automation_operation_log ( operation_type, actor, status, incident_id, input, output, dry_run_result, @@ -4127,58 +4936,72 @@ async def run_failed_apply_check_mode_replay_once( = 'controlled_retry_check_mode_replay' ) 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}", - ], - }, - ) + """), + { + "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_value = claimed.scalar_one_or_none() if replay_op_id_value is None: - stats["blockers"] = ["failed_apply_retry_already_claimed"] + stats["blockers"] = ["failed_apply_retry_claim_raced"] return stats replay_op_id = str(replay_op_id_value) - effective_timeout = ( - timeout_seconds or settings.AWOOOP_ANSIBLE_CHECK_MODE_TIMEOUT_SECONDS + stale_recovery_count = int( + row.get("stale_replay_recovery_count") or 0 ) - 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: + stale_retry_terminalized = bool( + stale_replay_op_id + and stale_recovery_count >= _CONTROLLED_RETRY_MAX_ATTEMPTS + ) + stats["stale_retry_reclaimed"] = int(bool(stale_replay_op_id)) + stats["stale_retry_terminalized"] = int(stale_retry_terminalized) + if stale_retry_terminalized: replay_result = AnsibleRunResult( returncode=1, stdout="", - stderr=f"ansible_retry_check_mode_runtime_error: {exc}", + stderr="stale_pending_retry_recovery_budget_exhausted", duration_ms=0, ) + else: + 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=True, @@ -4191,6 +5014,8 @@ async def run_failed_apply_check_mode_replay_once( "approval_required_before_apply": False, "owner_review_required": False, "runtime_apply_executed": False, + "check_mode_replay_performed": not stale_retry_terminalized, + "stale_pending_retry_terminalized": stale_retry_terminalized, "bounded_execution_scope": "retry_check_mode_only", "next_required_step": ( "queue_ai_playbook_or_transport_repair_candidate" @@ -4208,13 +5033,15 @@ async def run_failed_apply_check_mode_replay_once( "approval_required_before_apply": False, "owner_review_required": False, "runtime_apply_executed": False, + "check_mode_replay_performed": not stale_retry_terminalized, + "stale_pending_retry_terminalized": stale_retry_terminalized, "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: - await db.execute( + updated = await db.execute( text(""" UPDATE automation_operation_log SET status = :status, @@ -4224,6 +5051,9 @@ async def run_failed_apply_check_mode_replay_once( duration_ms = :duration_ms, stderr_feed_back = :stderr WHERE op_id = CAST(:op_id AS uuid) + AND status = 'pending' + AND input ->> 'retry_claim_id' = :retry_claim_id + RETURNING op_id """), { "status": status, @@ -4233,9 +5063,13 @@ async def run_failed_apply_check_mode_replay_once( "duration_ms": replay_result.duration_ms, "stderr": _tail(replay_result.stderr, _STDERR_LIMIT), "op_id": replay_op_id, + "retry_claim_id": retry_claim_id, }, ) - stats["replayed"] = 1 + if updated.scalar_one_or_none() is None: + stats["error"] = "failed_apply_retry_terminal_update_lost_lease" + return stats + stats["replayed"] = 0 if stale_retry_terminalized else 1 stats[ "check_mode_passed" if replay_result.returncode == 0 else "check_mode_failed" ] = 1 @@ -4245,6 +5079,7 @@ async def run_failed_apply_check_mode_replay_once( apply_op_id=apply_op_id, replay_op_id=replay_op_id, project_id=project_id, + check_mode_replay_performed=not stale_retry_terminalized, ): stats["runtime_stage_receipt_written"] = 1 logger.info( @@ -5244,6 +6079,18 @@ async def run_pending_check_modes_once( "retry_terminal_projection_written": int( receipt_stats.get("retry_terminal_projection_written") or 0 ), + "retry_terminal_projection_retry_receipt_written": int( + receipt_stats.get( + "retry_terminal_projection_retry_receipt_written" + ) + or 0 + ), + "retry_terminal_projection_telegram_receipt_acknowledged": int( + receipt_stats.get( + "retry_terminal_projection_telegram_receipt_acknowledged" + ) + or 0 + ), "retry_terminal_projection_verified": int( receipt_stats.get("retry_terminal_projection_verified") or 0 ), @@ -5265,27 +6112,6 @@ async def run_pending_check_modes_once( project_id=project_id, **receipt_backfill_summary, ) - if receipt_backfill_summary["repair_receipt_backfill_error"]: - 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, @@ -5310,6 +6136,12 @@ async def run_pending_check_modes_once( "failed_apply_retry_stage_receipt_written": int( retry_stats.get("runtime_stage_receipt_written") or 0 ), + "failed_apply_retry_stale_reclaimed": int( + retry_stats.get("stale_retry_reclaimed") or 0 + ), + "failed_apply_retry_stale_terminalized": int( + retry_stats.get("stale_retry_terminalized") or 0 + ), "failed_apply_retry_blockers": failed_apply_retry_blockers, "failed_apply_retry_error": failed_apply_retry_error, } @@ -5328,25 +6160,6 @@ async def run_pending_check_modes_once( 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, - **receipt_backfill_summary, - **failed_apply_retry_summary, - } blockers = _runtime_blockers() if blockers: logger.warning("ansible_check_mode_runtime_blocked", blockers=blockers) diff --git a/apps/api/tests/test_ansible_verified_closure.py b/apps/api/tests/test_ansible_verified_closure.py index 91488d411..70f227c94 100644 --- a/apps/api/tests/test_ansible_verified_closure.py +++ b/apps/api/tests/test_ansible_verified_closure.py @@ -1,6 +1,7 @@ from __future__ import annotations import inspect +from contextlib import asynccontextmanager from unittest.mock import AsyncMock import pytest @@ -9,6 +10,36 @@ from src.jobs import awooop_ansible_candidate_backfill_job as candidate_job from src.services import awooop_ansible_check_mode_service as service +class _MappingResult: + def __init__(self, row: dict | None = None, scalar=None) -> None: + self._row = row + self._scalar = scalar + + def mappings(self) -> _MappingResult: + return self + + def one_or_none(self) -> dict | None: + return self._row + + def scalar_one_or_none(self): + return self._scalar + + def scalar(self): + return self._scalar + + +class _SequenceDB: + def __init__(self, *results: _MappingResult) -> None: + self._results = list(results) + self.statements: list[str] = [] + self.parameters: list[dict] = [] + + async def execute(self, statement, parameters=None): + self.statements.append(str(statement)) + self.parameters.append(dict(parameters or {})) + return self._results.pop(0) if self._results else _MappingResult() + + def _claim() -> service.AnsibleCheckModeClaim: return service.AnsibleCheckModeClaim( op_id="00000000-0000-0000-0000-000000000101", @@ -61,25 +92,69 @@ def test_incident_terminal_writer_normalizes_legacy_non_object_outcome() -> None source = inspect.getsource(service._record_incident_terminal_disposition) assert "jsonb_typeof(CAST(outcome AS jsonb))" in source - assert "jsonb_set(" in source + assert "jsonb_build_object(" in source assert "'legacy_outcome'" in source - assert "'{automation_terminal}'" in source -def test_incident_terminal_persists_automated_outcome_truth() -> None: - source = inspect.getsource(service._record_incident_terminal_disposition) +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("proposal_executed", "expected_success"), + [(True, False), (False, None)], +) +async def test_incident_terminal_round_trips_negative_outcome_truth( + monkeypatch: pytest.MonkeyPatch, + proposal_executed: bool, + expected_success: bool | None, +) -> None: + terminal = { + "schema_version": "ansible_incident_terminal_disposition_v1", + "automation_run_id": "00000000-0000-0000-0000-000000000102", + "apply_op_id": "00000000-0000-0000-0000-000000000104", + "terminal_type": "no_write_replay_failed_waiting_repair_candidate", + } + db = _SequenceDB( + _MappingResult({"proposal_executed": proposal_executed}), + _MappingResult( + { + "incident_status": "MITIGATING", + "updated_at": None, + "resolved_at": None, + "outcome": { + "automation_terminal": terminal, + "proposal_executed": proposal_executed, + "execution_success": expected_success, + }, + } + ), + ) - assert "'proposal_executed'" in source - assert "'execution_success'" in source - assert 'outcome.get("proposal_executed") is True' in source - assert 'outcome.get("execution_success") is True' in source + @asynccontextmanager + async def fake_db_context(_project_id: str): + yield db + + monkeypatch.setattr(service, "get_db_context", fake_db_context) + + result = await service._record_incident_terminal_disposition( + _claim(), + apply_op_id="00000000-0000-0000-0000-000000000104", + project_id="awoooi", + terminal_type=terminal["terminal_type"], + success_terminal=False, + retry_op_id=None, + telegram_receipt_acknowledged=False, + ) + + assert result is not None + assert result["proposal_executed"] is proposal_executed + assert result["execution_success"] is expected_success + assert db.parameters[1]["execution_success"] is expected_success @pytest.mark.asyncio async def test_closure_refuses_to_resolve_when_any_receipt_is_missing( monkeypatch: pytest.MonkeyPatch, ) -> None: - terminal_writer = AsyncMock() + atomic_writer = AsyncMock() monkeypatch.setattr( service, "_read_verified_apply_closure_prerequisites", @@ -93,8 +168,8 @@ async def test_closure_refuses_to_resolve_when_any_receipt_is_missing( ) monkeypatch.setattr( service, - "_record_incident_terminal_disposition", - terminal_writer, + "_commit_verified_incident_closure", + atomic_writer, ) result = await service._finalize_verified_apply_closure( @@ -108,7 +183,7 @@ async def test_closure_refuses_to_resolve_when_any_receipt_is_missing( "closed": False, "missing": ["telegram_receipt"], } - terminal_writer.assert_not_awaited() + atomic_writer.assert_not_awaited() @pytest.mark.asyncio @@ -120,30 +195,19 @@ async def test_closure_resolves_only_after_durable_readback( "_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, + atomic_writer = AsyncMock( + return_value={ + "automation_run_id": "00000000-0000-0000-0000-000000000102", + "apply_op_id": "00000000-0000-0000-0000-000000000104", + "incident_resolved": True, + "closure_receipt_written": True, + "resolved_lifecycle_written": True, + } ) monkeypatch.setattr( service, - "_append_alert_lifecycle_receipt", - lifecycle_writer, + "_commit_verified_incident_closure", + atomic_writer, ) monkeypatch.setattr( service, @@ -165,10 +229,65 @@ async def test_closure_resolves_only_after_durable_readback( 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" + atomic_writer.assert_awaited_once() + assert atomic_writer.await_args.kwargs["closure_prerequisite_count"] == 1 + + +@pytest.mark.asyncio +async def test_atomic_closure_rolls_back_when_bundle_readback_is_missing( + monkeypatch: pytest.MonkeyPatch, +) -> None: + db = _SequenceDB(_MappingResult(), _MappingResult(None)) + transaction_rolled_back = False + + @asynccontextmanager + async def fake_db_context(_project_id: str): + nonlocal transaction_rolled_back + try: + yield db + except Exception: + transaction_rolled_back = True + raise + + monkeypatch.setattr(service, "get_db_context", fake_db_context) + + result = await service._commit_verified_incident_closure( + _claim(), + apply_op_id="00000000-0000-0000-0000-000000000104", + project_id="awoooi", + closure_prerequisite_count=12, + ) + + assert result is None + assert transaction_rolled_back is True + atomic_statement = db.statements[1] + assert "INSERT INTO alert_operation_log" in atomic_statement + assert "UPDATE automation_operation_log apply" in atomic_statement + assert "UPDATE incidents incident" in atomic_statement + assert "WITH target_apply AS" in atomic_statement + assert "WHEN upper(incident.status::text) = 'CLOSED'" in atomic_statement + + +def test_terminal_candidate_contract_requires_verified_four_node_chain() -> None: + predicate = candidate_job.verified_ansible_candidate_terminal_sql( + "candidate" + ) + recorder_source = inspect.getsource( + candidate_job.record_ansible_decision_audit + ) + + assert "direct_terminal.status = 'dry_run'" in predicate + assert "check_mode.parent_op_id = candidate.op_id" in predicate + assert "apply.parent_op_id = check_mode.op_id" in predicate + assert "replay.parent_op_id = apply.op_id" in predicate + assert "replay.status IN ('success', 'failed')" in predicate + assert "verified_no_write_terminal" in predicate + assert "repository_readback_verified" in predicate + assert "{detail,execution_success}" in predicate + assert "{detail,telegram_receipt_acknowledged}" in predicate + assert "ORDER BY latest.created_at DESC, latest.op_id DESC" in ( + recorder_source + ) @pytest.mark.asyncio @@ -234,15 +353,35 @@ def test_backfill_preserves_approval_and_replaces_terminal_skipped_candidate() - "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 + query_source = inspect.getsource( + candidate_job._fetch_missing_candidate_incidents ) 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 + assert "verified_ansible_candidate_terminal_sql" in query_source + assert "AND NOT ({terminal_candidate_sql})" in query_source + assert "ORDER BY latest.created_at DESC, latest.op_id DESC" in query_source + + +def test_retry_projection_only_repairs_latest_candidate_run() -> None: + source = inspect.getsource( + service._load_missing_retry_terminal_projection_rows + ) + + assert "source_candidate.op_id = source_check.parent_op_id" in source + assert "newer_candidate.created_at," in source + assert "source_candidate.created_at," in source + assert "repair_candidate_controlled_queue" in source + assert "retry_receipt.receipt IS NULL" in source + assert "controlled_apply_result" in source + assert "LEFT JOIN LATERAL" in source + assert "receipt.value ->> 'automation_run_id'" in source + assert "upper(incident.status::text)" in source + assert "ansible_incident_terminal_disposition_v1" in source + assert "'proposal_executed'" in source + assert "'execution_success'" in source + assert "verification_result IN ('failed', 'timeout')" in source @pytest.mark.asyncio @@ -326,6 +465,72 @@ async def test_broker_reconciles_receipts_then_continues_claiming_fresh_work( fresh_claimer.assert_awaited_once() +@pytest.mark.asyncio +async def test_broker_errors_do_not_starve_unrelated_fresh_claims( + monkeypatch: pytest.MonkeyPatch, +) -> None: + 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": 3, + "written": 0, + "incident_closure_written": 0, + "error": "legacy_projection_write_failed", + } + ), + ) + monkeypatch.setattr( + service, + "run_failed_apply_check_mode_replay_once", + AsyncMock( + return_value={ + "scanned": 1, + "replayed": 0, + "blockers": ["foreign_retry_already_claimed"], + "error": "foreign_retry_reconstruction_failed", + } + ), + ) + monkeypatch.setattr(service, "_runtime_blockers", lambda: []) + monkeypatch.setattr( + service, + "recent_ansible_transport_blockers", + AsyncMock(return_value=[]), + ) + monkeypatch.setattr( + service, + "claim_stale_pending_check_modes", + AsyncMock(return_value=[]), + ) + monkeypatch.setattr( + service, + "claim_catalog_drift_failed_check_modes", + AsyncMock(return_value=[]), + ) + fresh_claimer = AsyncMock(return_value=[]) + monkeypatch.setattr(service, "claim_pending_check_modes", fresh_claimer) + + result = await service.run_pending_check_modes_once(limit=1) + + assert result["repair_receipt_backfill_error"] == ( + "legacy_projection_write_failed" + ) + assert result["failed_apply_retry_error"] == ( + "foreign_retry_reconstruction_failed" + ) + assert result["failed_apply_retry_blockers"] == [ + "foreign_retry_already_claimed" + ] + fresh_claimer.assert_awaited_once() + + @pytest.mark.asyncio async def test_retry_terminal_projection_backfill_writes_and_verifies_no_apply( monkeypatch: pytest.MonkeyPatch, @@ -333,8 +538,13 @@ async def test_retry_terminal_projection_backfill_writes_and_verifies_no_apply( row = { "op_id": "00000000-0000-0000-0000-000000000104", "replay_op_id": "00000000-0000-0000-0000-000000000105", + "replay_status": "success", + "replay_output": {"returncode": 0}, + "replay_dry_run_result": {"timed_out": False}, + "replay_duration_ms": 18, + "retry_receipt_present": False, "terminal_type": "no_write_replay_passed_waiting_repair_candidate", - "telegram_receipt_acknowledged": True, + "telegram_receipt_acknowledged": False, } monkeypatch.setattr( service, @@ -366,6 +576,8 @@ async def test_retry_terminal_projection_backfill_writes_and_verifies_no_apply( receipt_writer = AsyncMock(return_value=True) lifecycle_writer = AsyncMock(return_value=True) verifier = AsyncMock(return_value=True) + telegram_sender = AsyncMock(return_value=True) + telegram_readback = AsyncMock(return_value=True) monkeypatch.setattr( service, "_record_incident_terminal_disposition", terminal_writer ) @@ -378,12 +590,24 @@ async def test_retry_terminal_projection_backfill_writes_and_verifies_no_apply( monkeypatch.setattr( service, "_verify_retry_terminal_projection_readback", verifier ) + monkeypatch.setattr( + service, + "_send_controlled_apply_telegram_receipt", + telegram_sender, + ) + monkeypatch.setattr( + service, + "_retry_telegram_receipt_acknowledged", + telegram_readback, + ) result = await service.backfill_missing_retry_terminal_projections_once() assert result == { "scanned": 1, "written": 1, + "retry_receipt_written": 1, + "telegram_receipt_acknowledged": 1, "incident_receipt_written": 1, "lifecycle_written": 1, "verified": 1, @@ -392,11 +616,16 @@ async def test_retry_terminal_projection_backfill_writes_and_verifies_no_apply( } terminal_writer.assert_awaited_once() assert terminal_writer.await_args.kwargs["success_terminal"] is False - receipt = receipt_writer.await_args.kwargs["receipts"][0] + retry_receipt = receipt_writer.await_args_list[0].kwargs["receipts"][0] + assert retry_receipt["stage_id"] == "retry_or_rollback" + assert retry_receipt["derived_from_durable_chain"] is True + receipt = receipt_writer.await_args_list[1].kwargs["receipts"][0] assert receipt["stage_id"] == "incident_closure" assert receipt["derived_from_durable_chain"] is True assert lifecycle_writer.await_args.args[1] == "EXECUTION_COMPLETED" assert lifecycle_writer.await_args.kwargs["success"] is False + telegram_sender.assert_awaited_once() + telegram_readback.assert_awaited_once() @pytest.mark.asyncio @@ -426,6 +655,29 @@ async def test_broker_prioritizes_retry_terminal_projection_backfill( "backfill_missing_auto_repair_execution_receipts_once", receipt_backfill, ) + monkeypatch.setattr( + service, + "run_failed_apply_check_mode_replay_once", + AsyncMock(return_value={"scanned": 0, "replayed": 0}), + ) + monkeypatch.setattr(service, "_runtime_blockers", lambda: []) + monkeypatch.setattr( + service, + "recent_ansible_transport_blockers", + AsyncMock(return_value=[]), + ) + monkeypatch.setattr( + service, + "claim_stale_pending_check_modes", + AsyncMock(return_value=[]), + ) + monkeypatch.setattr( + service, + "claim_catalog_drift_failed_check_modes", + AsyncMock(return_value=[]), + ) + fresh_claimer = AsyncMock(return_value=[]) + monkeypatch.setattr(service, "claim_pending_check_modes", fresh_claimer) result = await service.run_pending_check_modes_once(limit=1) @@ -434,6 +686,7 @@ async def test_broker_prioritizes_retry_terminal_projection_backfill( assert result["retry_terminal_projection_runtime_apply_executed"] is False assert result["repair_receipt_backfill_priority_tick"] is True receipt_backfill.assert_awaited_once() + fresh_claimer.assert_awaited_once() @pytest.mark.asyncio @@ -458,6 +711,108 @@ async def test_signal_worker_retry_preflight_is_query_only( assert result["execution_owner"] == "awoooi-ansible-executor-broker" +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("recovery_count", "expected_replayed", "expected_terminalized"), + [(0, 1, 0), (1, 0, 1)], +) +async def test_stale_pending_retry_is_reclaimed_without_runtime_apply( + monkeypatch: pytest.MonkeyPatch, + recovery_count: int, + expected_replayed: int, + expected_terminalized: int, +) -> None: + replay_op_id = "00000000-0000-0000-0000-000000000105" + apply_op_id = "00000000-0000-0000-0000-000000000104" + db = _SequenceDB( + _MappingResult(), + _MappingResult(scalar=replay_op_id), + _MappingResult(scalar=replay_op_id), + ) + + @asynccontextmanager + async def fake_db_context(_project_id: str): + yield db + + monkeypatch.setattr(service, "get_db_context", fake_db_context) + monkeypatch.setattr( + service, + "_load_open_failed_apply_retry_row", + AsyncMock( + return_value={ + "op_id": apply_op_id, + "stale_replay_op_id": replay_op_id, + "stale_replay_recovery_count": recovery_count, + } + ), + ) + monkeypatch.setattr( + service, + "_claim_from_apply_operation_row", + lambda _row: ( + _claim(), + service.AnsibleRunResult( + returncode=1, + stdout="", + stderr="failed apply", + duration_ms=10, + ), + ), + ) + monkeypatch.setattr(service, "_runtime_blockers", lambda: []) + monkeypatch.setattr( + service, + "recent_ansible_transport_blockers", + AsyncMock(return_value=[]), + ) + monkeypatch.setattr( + service, + "build_ansible_check_mode_command", + lambda **_kwargs: object(), + ) + replay_runner = AsyncMock( + return_value=service.AnsibleRunResult( + returncode=0, + stdout="check ok", + stderr="", + duration_ms=15, + ) + ) + monkeypatch.setattr(service, "_run_ansible_command", replay_runner) + receipt_writer = AsyncMock(return_value=True) + monkeypatch.setattr( + service, + "_record_retry_runtime_stage_receipt", + receipt_writer, + ) + + result = await service.run_failed_apply_check_mode_replay_once( + timeout_seconds=30, + ) + + assert result["replayed"] == expected_replayed + if expected_terminalized: + assert result["check_mode_passed"] == 0 + assert result["check_mode_failed"] == 1 + else: + assert result["check_mode_passed"] == 1 + assert result["check_mode_failed"] == 0 + assert result["stale_retry_terminalized"] == expected_terminalized + assert result["runtime_apply_executed"] == 0 + assert "pg_advisory_xact_lock" in db.statements[0] + assert "UPDATE automation_operation_log replay" in db.statements[1] + assert "reclaimed_after_stale_pending" in db.statements[1] + assert "input ->> 'retry_claim_id'" in db.statements[2] + if expected_terminalized: + replay_runner.assert_not_awaited() + else: + replay_runner.assert_awaited_once() + receipt_writer.assert_awaited_once() + assert receipt_writer.await_args.kwargs[ + "check_mode_replay_performed" + ] is (not bool(expected_terminalized)) + + def test_candidate_worker_defaults_to_query_only_retry_preflight() -> None: defaults = candidate_job.enqueue_missing_ansible_candidates_once.__kwdefaults__ 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 1d7ebf1a3..a1b9f6eed 100644 --- a/apps/api/tests/test_awooop_ansible_candidate_backfill_job.py +++ b/apps/api/tests/test_awooop_ansible_candidate_backfill_job.py @@ -130,7 +130,7 @@ async def test_backfill_enqueues_catalog_matched_incident(monkeypatch: pytest.Mo @pytest.mark.asyncio -async def test_backfill_prioritizes_open_failed_apply_retry_before_fresh_candidates( +async def test_backfill_runs_retry_then_continues_fresh_candidates( monkeypatch: pytest.MonkeyPatch, ) -> None: from src.jobs import awooop_ansible_candidate_backfill_job as job @@ -147,11 +147,10 @@ async def test_backfill_prioritizes_open_failed_apply_retry_before_fresh_candida "error": None, } - async def fresh_scan_must_not_run(**_kwargs): - raise AssertionError("fresh candidates must wait for the open retry tick") + fresh_scan = AsyncMock(return_value=[]) monkeypatch.setattr(job.settings, "ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER", True) - monkeypatch.setattr(job, "_fetch_missing_candidate_incidents", fresh_scan_must_not_run) + monkeypatch.setattr(job, "_fetch_missing_candidate_incidents", fresh_scan) receipt_backfiller = AsyncMock(return_value={"written": 0, "error": None}) recorder = AsyncMock(return_value=True) @@ -167,12 +166,13 @@ async def test_backfill_prioritizes_open_failed_apply_retry_before_fresh_candida assert result["failed_apply_retry_stage_receipt_written"] == 1 assert result["failed_apply_retry_priority_tick"] is True assert result["queued"] == 0 - receipt_backfiller.assert_not_awaited() + receipt_backfiller.assert_awaited_once() + fresh_scan.assert_awaited_once() recorder.assert_not_awaited() @pytest.mark.asyncio -async def test_backfill_surfaces_retry_blockers_without_queueing_fresh_work( +async def test_backfill_surfaces_retry_blockers_and_continues_fresh_scan( monkeypatch: pytest.MonkeyPatch, ) -> None: from src.jobs import awooop_ansible_candidate_backfill_job as job @@ -185,11 +185,10 @@ async def test_backfill_surfaces_retry_blockers_without_queueing_fresh_work( "error": None, } - async def fresh_scan_must_not_run(**_kwargs): - raise AssertionError("fresh candidates must not grow behind retry blockers") + fresh_scan = AsyncMock(return_value=[]) monkeypatch.setattr(job.settings, "ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER", True) - monkeypatch.setattr(job, "_fetch_missing_candidate_incidents", fresh_scan_must_not_run) + monkeypatch.setattr(job, "_fetch_missing_candidate_incidents", fresh_scan) result = await job.enqueue_missing_ansible_candidates_once( retry_replayer=fake_retry_replayer, @@ -201,6 +200,7 @@ async def test_backfill_surfaces_retry_blockers_without_queueing_fresh_work( ] assert result["failed_apply_retry_priority_tick"] is True assert result["queued"] == 0 + fresh_scan.assert_awaited_once() @pytest.mark.asyncio diff --git a/apps/api/tests/test_awooop_truth_chain_service.py b/apps/api/tests/test_awooop_truth_chain_service.py index 45da1ccdf..5035b449c 100644 --- a/apps/api/tests/test_awooop_truth_chain_service.py +++ b/apps/api/tests/test_awooop_truth_chain_service.py @@ -17,6 +17,7 @@ from src.services.awooop_ansible_audit_service import ( build_ansible_decision_audit_payload, build_ansible_truth, record_ansible_decision_audit, + verified_ansible_candidate_terminal_sql, ) from src.services.awooop_ansible_check_mode_service import ( _EXECUTION_CAPABILITY_FALLBACK_OPERATION_TYPE, @@ -26,6 +27,7 @@ from src.services.awooop_ansible_check_mode_service import ( _append_runtime_stage_receipts_to_apply, _automation_operation_log_incident_id, _build_auto_repair_execution_receipt, + _build_retry_runtime_stage_receipt, _claim_from_apply_operation_row, _claim_from_stale_check_mode_row, _execution_capability_timeout_seconds, @@ -147,13 +149,14 @@ def test_quality_summary_uses_batched_truth_chain_inputs() -> None: def test_ansible_audit_keeps_external_incident_id_in_json_not_bigint_column() -> None: decision_source = inspect.getsource(record_ansible_decision_audit) + terminal_source = verified_ansible_candidate_terminal_sql("candidate") claim_source = inspect.getsource(claim_pending_check_modes) assert "operation_type, actor, status, incident_id" in decision_source assert "candidate.incident_id::text" in decision_source assert "candidate.input ->> 'incident_id'" in decision_source - assert "terminal.parent_op_id = candidate.op_id" in decision_source - assert "ansible_execution_skipped" in decision_source + assert "verified_ansible_candidate_terminal_sql" in decision_source + assert "ansible_execution_skipped" in terminal_source assert "operation_type, actor, status, incident_id" in claim_source assert "incident_db_id" in decision_source assert "incident_db_id" in claim_source @@ -2108,7 +2111,7 @@ async def test_catalog_drift_query_failure_does_not_block_fresh_candidate_claims @pytest.mark.asyncio -async def test_execution_broker_runs_failed_apply_retry_before_all_candidate_claims( +async def test_execution_broker_runs_retry_then_keeps_fresh_capacity_moving( monkeypatch: pytest.MonkeyPatch, ) -> None: from src.services import awooop_ansible_check_mode_service as service @@ -2132,7 +2135,7 @@ async def test_execution_broker_runs_failed_apply_retry_before_all_candidate_cla "error": None, } - async def candidate_claim_must_not_run(**_kwargs): + async def candidate_claim_continues(**_kwargs): nonlocal claim_attempted claim_attempted = True return [] @@ -2161,22 +2164,22 @@ async def test_execution_broker_runs_failed_apply_retry_before_all_candidate_cla monkeypatch.setattr( service, "claim_stale_pending_check_modes", - candidate_claim_must_not_run, + candidate_claim_continues, ) monkeypatch.setattr( service, "claim_catalog_drift_failed_check_modes", - candidate_claim_must_not_run, + candidate_claim_continues, ) monkeypatch.setattr( service, "claim_pending_check_modes", - candidate_claim_must_not_run, + candidate_claim_continues, ) result = await service.run_pending_check_modes_once(limit=1) - assert claim_attempted is False + assert claim_attempted is True assert result["claimed"] == 0 assert result["failed_apply_retry_scanned"] == 1 assert result["failed_apply_retry_replayed"] == 1 @@ -2424,10 +2427,11 @@ 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 preflight_source + assert "FOR UPDATE OF apply 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 preflight_source + assert "verification_result IN ('failed', 'timeout')" 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 @@ -2436,16 +2440,26 @@ 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 preflight_source - assert "failed_apply_retry_already_claimed" in source + assert "LEFT JOIN LATERAL" in preflight_source + assert "replay.op_id IS NULL" in preflight_source + assert "replay.status = 'pending'" in preflight_source + assert "failed_apply_retry_claim_raced" in source + assert "reclaimed_after_stale_pending" in source + assert "retry_claim_id" in source + assert "retry_recovery_count" in source + assert "pg_advisory_xact_lock" in source assert "run_controlled_apply_for_claim" not in source receipt_source = inspect.getsource(_record_retry_runtime_stage_receipt) + receipt_builder_source = inspect.getsource( + _build_retry_runtime_stage_receipt + ) append_source = inspect.getsource(_append_runtime_stage_receipts_to_apply) - assert "_runtime_stage_receipt" in receipt_source - assert 'stage_id="retry_or_rollback"' in receipt_source + assert "_build_retry_runtime_stage_receipt" in receipt_source + assert "_runtime_stage_receipt" in receipt_builder_source + assert 'stage_id="retry_or_rollback"' in receipt_builder_source assert "jsonb_array_elements" in append_source - assert "ansible_controlled_retry_terminal_v2" in receipt_source + assert "ansible_controlled_retry_terminal_v2" in receipt_builder_source assert "ansible_retry_terminal_receipt_not_written_unverified" in receipt_source assert "runtime_apply_executed" in receipt_source