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 0267ca114..0b20c0255 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -3317,10 +3317,44 @@ async def backfill_missing_auto_repair_execution_receipts_once( "runtime_stage_receipts_written": 0, "incident_closure_written": 0, "telegram_receipt_acknowledged": 0, + "retry_terminal_projection_scanned": 0, + "retry_terminal_projection_written": 0, + "retry_terminal_projection_incident_receipt_written": 0, + "retry_terminal_projection_lifecycle_written": 0, + "retry_terminal_projection_verified": 0, + "retry_terminal_projection_runtime_apply_executed": False, "skipped": 0, "error": None, } try: + projection = await backfill_missing_retry_terminal_projections_once( + project_id=project_id, + window_hours=window_hours, + limit=max(1, min(limit, 5)), + ) + stats["retry_terminal_projection_scanned"] = int( + projection.get("scanned") or 0 + ) + stats["retry_terminal_projection_written"] = int( + projection.get("written") or 0 + ) + stats["retry_terminal_projection_incident_receipt_written"] = int( + projection.get("incident_receipt_written") or 0 + ) + stats["retry_terminal_projection_lifecycle_written"] = int( + projection.get("lifecycle_written") or 0 + ) + stats["retry_terminal_projection_verified"] = int( + projection.get("verified") or 0 + ) + projection_error = str(projection.get("error") or "")[:500] + if stats["retry_terminal_projection_scanned"] or projection_error: + stats["scanned"] = stats["retry_terminal_projection_scanned"] + stats["incident_closure_written"] = stats[ + "retry_terminal_projection_written" + ] + stats["error"] = projection_error or None + return stats async with get_db_context(project_id) as db: await db.execute(text("SET LOCAL statement_timeout = '5000ms'")) result = await db.execute( @@ -3654,6 +3688,327 @@ async def preflight_failed_apply_retry_queue_once( } +async def _load_missing_retry_terminal_projection_rows( + *, + project_id: str, + window_hours: int, + limit: int, +) -> list[dict[str, Any]]: + """Load verified no-write retries whose incident projection is incomplete.""" + + async with get_db_context(project_id) as db: + await db.execute(text("SET LOCAL statement_timeout = '5000ms'")) + result = await db.execute( + text(""" + SELECT + apply.op_id::text AS op_id, + apply.parent_op_id::text AS parent_op_id, + coalesce( + apply.incident_id::text, + apply.input ->> 'incident_id' + ) AS incident_id, + apply.input, + apply.output, + apply.dry_run_result, + apply.error, + apply.duration_ms, + apply.status, + apply.created_at, + replay.op_id::text AS replay_op_id, + retry_receipt.receipt #>> '{detail,terminal_type}' + AS terminal_type, + 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 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}' + ) = coalesce( + apply.incident_id::text, + apply.input ->> 'incident_id' + ) + ) AS telegram_receipt_acknowledged + FROM automation_operation_log apply + JOIN automation_operation_log replay + ON replay.parent_op_id = apply.op_id + AND replay.operation_type = 'ansible_execution_skipped' + AND replay.input ->> 'execution_mode' + = 'controlled_retry_check_mode_replay' + AND replay.status IN ('success', 'failed') + AND replay.output ->> 'runtime_apply_executed' = 'false' + JOIN incidents incident + ON incident.incident_id = coalesce( + apply.incident_id::text, + apply.input ->> 'incident_id' + ) + AND incident.project_id = :project_id + JOIN LATERAL ( + SELECT receipt.value AS receipt + FROM jsonb_array_elements( + coalesce( + apply.input -> 'runtime_stage_receipts', + '[]'::jsonb + ) + ) receipt(value) + WHERE receipt.value ->> 'stage_id' = 'retry_or_rollback' + AND receipt.value #>> + '{detail,retry_check_mode_op_id}' = replay.op_id::text + AND receipt.value #>> + '{detail,failed_apply_op_id}' = apply.op_id::text + AND receipt.value #>> + '{detail,verified_no_write_terminal}' = 'true' + AND receipt.value #>> + '{detail,runtime_apply_executed}' = 'false' + LIMIT 1 + ) retry_receipt ON TRUE + WHERE apply.operation_type = 'ansible_apply_executed' + AND apply.status = 'failed' + AND apply.created_at >= NOW() - ( + :window_hours * INTERVAL '1 hour' + ) + AND EXISTS ( + SELECT 1 + FROM incident_evidence verifier + WHERE verifier.incident_id = incident.incident_id + AND verifier.post_execution_state ->> 'apply_op_id' + = apply.op_id::text + AND verifier.verification_result IN ('failed', 'timeout') + ) + 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' + OR NOT EXISTS ( + SELECT 1 + FROM alert_operation_log lifecycle + WHERE lifecycle.incident_id = incident.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' + = apply.input ->> 'automation_run_id' + ) + ) + ORDER BY replay.created_at DESC + LIMIT :limit + """), + { + "project_id": project_id, + "window_hours": max(1, window_hours), + "limit": max(1, min(limit, 5)), + }, + ) + return [dict(row) for row in result.mappings().all()] + + +async def _verify_retry_terminal_projection_readback( + claim: AnsibleCheckModeClaim, + *, + apply_op_id: str, + replay_op_id: str, + terminal_type: str, + project_id: str, +) -> bool: + automation_run_id = _automation_run_id_for_claim(claim) + async with get_db_context(project_id) as db: + await db.execute(text("SET LOCAL statement_timeout = '5000ms'")) + result = await db.execute( + text(""" + SELECT + upper(incident.status::text) = 'MITIGATING' + AND cast(incident.outcome AS jsonb) #>> + '{automation_terminal,schema_version}' + = 'ansible_incident_terminal_disposition_v1' + AND cast(incident.outcome AS jsonb) #>> + '{automation_terminal,automation_run_id}' + = :automation_run_id + AND cast(incident.outcome AS jsonb) #>> + '{automation_terminal,apply_op_id}' = :apply_op_id + AND cast(incident.outcome AS jsonb) #>> + '{automation_terminal,retry_op_id}' = :replay_op_id + AND cast(incident.outcome AS jsonb) #>> + '{automation_terminal,terminal_type}' = :terminal_type + AND cast(incident.outcome AS jsonb) #>> + '{automation_terminal,no_write_terminal}' = 'true' + AND 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 #>> '{detail,apply_op_id}' + = :apply_op_id + AND receipt.value #>> '{detail,retry_op_id}' + = :replay_op_id + AND receipt.value #>> + '{detail,repository_readback_verified}' = 'true' + AND receipt.value ->> 'durable_receipt' = 'true' + ) + AND EXISTS ( + SELECT 1 + FROM alert_operation_log lifecycle + WHERE lifecycle.incident_id = incident.incident_id + AND lifecycle.event_type::text = 'EXECUTION_COMPLETED' + AND lifecycle.success IS FALSE + AND lifecycle.context ->> 'apply_op_id' = :apply_op_id + AND lifecycle.context ->> 'automation_run_id' + = :automation_run_id + ) AS projection_verified + 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 + """), + { + "automation_run_id": automation_run_id, + "apply_op_id": apply_op_id, + "replay_op_id": replay_op_id, + "terminal_type": terminal_type, + "incident_id": claim.incident_id, + "project_id": project_id, + }, + ) + return result.scalar() is True + + +async def backfill_missing_retry_terminal_projections_once( + *, + project_id: str = "awoooi", + window_hours: int = 24, + limit: int = 1, +) -> dict[str, Any]: + """Repair missing incident terminals from durable no-write retry receipts.""" + + stats: dict[str, Any] = { + "scanned": 0, + "written": 0, + "incident_receipt_written": 0, + "lifecycle_written": 0, + "verified": 0, + "runtime_apply_executed": False, + "error": None, + } + try: + rows = await _load_missing_retry_terminal_projection_rows( + project_id=project_id, + window_hours=window_hours, + limit=limit, + ) + stats["scanned"] = len(rows) + for row in rows: + reconstructed = _claim_from_apply_operation_row(row) + if reconstructed is None: + continue + 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 "") + if not ( + apply_op_id + and replay_op_id + and terminal_type.startswith("no_write_replay_") + and row.get("telegram_receipt_acknowledged") is True + ): + continue + terminal = await _record_incident_terminal_disposition( + claim, + apply_op_id=apply_op_id, + project_id=project_id, + terminal_type=terminal_type, + success_terminal=False, + retry_op_id=replay_op_id, + telegram_receipt_acknowledged=True, + ) + if terminal is None: + continue + incident_receipt_written = ( + await _append_runtime_stage_receipts_to_apply( + apply_op_id=apply_op_id, + receipts=( + _runtime_stage_receipt( + claim, + stage_id="incident_closure", + evidence_ref=( + f"incidents:{claim.incident_id}:automation_terminal" + ), + detail=terminal, + derived_from_durable_chain=True, + ), + ), + project_id=project_id, + ) + ) + lifecycle_written = await _append_alert_lifecycle_receipt( + claim, + "EXECUTION_COMPLETED", + apply_op_id=apply_op_id, + success=False, + action_detail="controlled_retry_no_write_terminal_backfilled", + project_id=project_id, + error_message="failed_apply_retry_terminal", + post_verifier_passed=False, + ) + stats["incident_receipt_written"] += int( + incident_receipt_written + ) + stats["lifecycle_written"] += int(lifecycle_written) + if not incident_receipt_written or not lifecycle_written: + continue + verified = await _verify_retry_terminal_projection_readback( + claim, + apply_op_id=apply_op_id, + replay_op_id=replay_op_id, + terminal_type=terminal_type, + project_id=project_id, + ) + stats["verified"] += int(verified) + stats["written"] += int(verified) + except Exception as exc: + stats["error"] = f"{type(exc).__name__}: {exc}"[:500] + logger.warning( + "ansible_retry_terminal_projection_backfill_failed", + project_id=project_id, + **stats, + ) + return stats + + async def run_failed_apply_check_mode_replay_once( *, project_id: str = "awoooi", @@ -4857,6 +5212,16 @@ async def run_pending_check_modes_once( "repair_receipt_telegram_acknowledged": int( receipt_stats.get("telegram_receipt_acknowledged") or 0 ), + "retry_terminal_projection_scanned": int( + receipt_stats.get("retry_terminal_projection_scanned") or 0 + ), + "retry_terminal_projection_written": int( + receipt_stats.get("retry_terminal_projection_written") or 0 + ), + "retry_terminal_projection_verified": int( + receipt_stats.get("retry_terminal_projection_verified") or 0 + ), + "retry_terminal_projection_runtime_apply_executed": False, "repair_receipt_backfill_error": ( str(receipt_stats.get("error") or "")[:500] or None ), diff --git a/apps/api/tests/test_ansible_verified_closure.py b/apps/api/tests/test_ansible_verified_closure.py index 89514e054..316d688ed 100644 --- a/apps/api/tests/test_ansible_verified_closure.py +++ b/apps/api/tests/test_ansible_verified_closure.py @@ -236,6 +236,21 @@ async def test_broker_reconciles_receipts_before_claiming_fresh_work( "_expire_stale_ansible_execution_capabilities", AsyncMock(return_value=0), ) + monkeypatch.setattr( + service, + "backfill_missing_retry_terminal_projections_once", + AsyncMock( + return_value={ + "scanned": 0, + "written": 0, + "incident_receipt_written": 0, + "lifecycle_written": 0, + "verified": 0, + "runtime_apply_executed": False, + "error": None, + } + ), + ) monkeypatch.setattr( service, "backfill_missing_auto_repair_execution_receipts_once", @@ -263,6 +278,116 @@ async def test_broker_reconciles_receipts_before_claiming_fresh_work( retry_replayer.assert_not_awaited() +@pytest.mark.asyncio +async def test_retry_terminal_projection_backfill_writes_and_verifies_no_apply( + monkeypatch: pytest.MonkeyPatch, +) -> None: + row = { + "op_id": "00000000-0000-0000-0000-000000000104", + "replay_op_id": "00000000-0000-0000-0000-000000000105", + "terminal_type": "no_write_replay_passed_waiting_repair_candidate", + "telegram_receipt_acknowledged": True, + } + monkeypatch.setattr( + service, + "_load_missing_retry_terminal_projection_rows", + AsyncMock(return_value=[row]), + ) + monkeypatch.setattr( + service, + "_claim_from_apply_operation_row", + lambda _row: ( + _claim(), + service.AnsibleRunResult( + returncode=1, + stdout="", + stderr="", + duration_ms=0, + ), + ), + ) + terminal_writer = AsyncMock( + return_value={ + "automation_run_id": "00000000-0000-0000-0000-000000000102", + "apply_op_id": row["op_id"], + "retry_op_id": row["replay_op_id"], + "terminal_type": row["terminal_type"], + "repository_readback_verified": True, + } + ) + receipt_writer = AsyncMock(return_value=True) + lifecycle_writer = AsyncMock(return_value=True) + verifier = AsyncMock(return_value=True) + monkeypatch.setattr( + service, "_record_incident_terminal_disposition", terminal_writer + ) + monkeypatch.setattr( + service, "_append_runtime_stage_receipts_to_apply", receipt_writer + ) + monkeypatch.setattr( + service, "_append_alert_lifecycle_receipt", lifecycle_writer + ) + monkeypatch.setattr( + service, "_verify_retry_terminal_projection_readback", verifier + ) + + result = await service.backfill_missing_retry_terminal_projections_once() + + assert result == { + "scanned": 1, + "written": 1, + "incident_receipt_written": 1, + "lifecycle_written": 1, + "verified": 1, + "runtime_apply_executed": False, + "error": None, + } + terminal_writer.assert_awaited_once() + assert terminal_writer.await_args.kwargs["success_terminal"] is False + receipt = receipt_writer.await_args.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 + + +@pytest.mark.asyncio +async def test_broker_prioritizes_retry_terminal_projection_backfill( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + service, + "_expire_stale_ansible_execution_capabilities", + AsyncMock(return_value=0), + ) + receipt_backfill = AsyncMock( + return_value={ + "scanned": 1, + "written": 0, + "incident_closure_written": 1, + "telegram_receipt_acknowledged": 0, + "retry_terminal_projection_scanned": 1, + "retry_terminal_projection_written": 1, + "retry_terminal_projection_verified": 1, + "retry_terminal_projection_runtime_apply_executed": False, + "error": None, + } + ) + monkeypatch.setattr( + service, + "backfill_missing_auto_repair_execution_receipts_once", + receipt_backfill, + ) + + result = await service.run_pending_check_modes_once(limit=1) + + assert result["claimed"] == 0 + assert result["retry_terminal_projection_verified"] == 1 + assert result["retry_terminal_projection_runtime_apply_executed"] is False + assert result["repair_receipt_backfill_priority_tick"] is True + receipt_backfill.assert_awaited_once() + + @pytest.mark.asyncio async def test_signal_worker_retry_preflight_is_query_only( monkeypatch: pytest.MonkeyPatch,