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 3bccde124..bacd412de 100644 --- a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py +++ b/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py @@ -12,7 +12,7 @@ import asyncio import json import os import random -from collections.abc import Awaitable, Callable +from collections.abc import Awaitable, Callable, Mapping from types import SimpleNamespace from typing import Any from uuid import NAMESPACE_URL, uuid4, uuid5 @@ -432,6 +432,8 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( candidate.input ->> 'automation_run_id' AS automation_run_id, candidate.input ->> 'source_receipt_ref' AS source_receipt_ref, check_mode.status AS latest_check_status, + apply.status AS latest_apply_status, + learning.status AS latest_learning_status, EXISTS ( SELECT 1 FROM incident_evidence evidence @@ -456,13 +458,29 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( ) AS predecision_evidence_ready FROM automation_operation_log candidate LEFT JOIN LATERAL ( - SELECT check_receipt.status + SELECT check_receipt.op_id, check_receipt.status FROM automation_operation_log check_receipt WHERE check_receipt.parent_op_id = candidate.op_id AND check_receipt.operation_type = 'ansible_check_mode_executed' ORDER BY check_receipt.created_at DESC LIMIT 1 ) check_mode ON TRUE + LEFT JOIN LATERAL ( + SELECT apply_receipt.op_id, apply_receipt.status + FROM automation_operation_log apply_receipt + WHERE apply_receipt.parent_op_id = check_mode.op_id + AND apply_receipt.operation_type = 'ansible_apply_executed' + ORDER BY apply_receipt.created_at DESC + LIMIT 1 + ) apply ON TRUE + LEFT JOIN LATERAL ( + SELECT learning_receipt.status + FROM automation_operation_log learning_receipt + WHERE learning_receipt.parent_op_id = apply.op_id + AND learning_receipt.operation_type = 'ansible_learning_writeback_recorded' + ORDER BY learning_receipt.created_at DESC + LIMIT 1 + ) learning ON TRUE WHERE candidate.operation_type = 'ansible_candidate_matched' AND candidate.input -> 'executor_candidates' @> CAST(:catalog_match AS jsonb) AND candidate.created_at >= NOW() - (:freshness_hours * INTERVAL '1 hour') @@ -485,9 +503,10 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( existing_row.get("latest_check_status") or "" ) if existing_row else "" retry_attempt_ref: str | None = None - retry_reason: str | None = None + retry_reason = _wazuh_posture_retry_reason(existing_row) if ( existing_row + and retry_reason is None and latest_check_status != "failed" and existing_evidence_ready ): @@ -512,9 +531,12 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( and all(character in "0123456789abcdef" for character in source_sha) else "runtime-retry" ) - if latest_check_status == "failed": + if retry_reason == "check_mode_failed": retry_attempt_ref = deploy_ref - retry_reason = "failed_check_mode" + elif retry_reason == "controlled_apply_failed": + retry_attempt_ref = f"{deploy_ref}-apply" + elif retry_reason == "post_apply_learning_incomplete": + retry_attempt_ref = f"{deploy_ref}-learning" elif latest_check_status == "success": retry_attempt_ref = f"{deploy_ref}-context" retry_reason = "predecision_context" @@ -530,6 +552,7 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( or "" ), "work_item_id": _WAZUH_POSTURE_WORK_ITEM_ID, + "retry_reason": retry_reason, "active_blockers": [ "predecision_mcp_or_log_evidence_missing" ], @@ -541,8 +564,10 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( return { "status": ( "failed_check_retry_already_attempted_for_deploy" - if retry_reason == "failed_check_mode" + if retry_reason == "check_mode_failed" else "context_retry_already_attempted_for_deploy" + if retry_reason == "predecision_context" + else "incomplete_runtime_retry_already_attempted_for_deploy" ), "queued": 0, "existing": 1, @@ -553,15 +578,20 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( or "" ), "work_item_id": _WAZUH_POSTURE_WORK_ITEM_ID, + "retry_reason": retry_reason, "retry_of_failed_check_mode": ( - retry_reason == "failed_check_mode" + retry_reason == "check_mode_failed" ), "retry_of_incomplete_predecision_context": ( retry_reason == "predecision_context" ), + "retry_of_incomplete_runtime_closure": retry_reason in { + "controlled_apply_failed", + "post_apply_learning_incomplete", + }, "active_blockers": [ "bounded_retry_already_attempted_for_deploy" - if retry_reason == "failed_check_mode" + if retry_reason != "predecision_context" else "bounded_context_retry_already_attempted_for_deploy" ], } @@ -606,10 +636,15 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( "evidence_ready": 0, "automation_run_id": automation_run_id, "work_item_id": _WAZUH_POSTURE_WORK_ITEM_ID, - "retry_of_failed_check_mode": retry_reason == "failed_check_mode", + "retry_reason": retry_reason, + "retry_of_failed_check_mode": retry_reason == "check_mode_failed", "retry_of_incomplete_predecision_context": ( retry_reason == "predecision_context" ), + "retry_of_incomplete_runtime_closure": retry_reason in { + "controlled_apply_failed", + "post_apply_learning_incomplete", + }, "active_blockers": [ "predecision_mcp_or_log_evidence_missing" ], @@ -632,10 +667,10 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( ) return { "status": ( - "queued_for_controlled_executor_retry" - if queued and retry_reason == "failed_check_mode" - else "queued_for_controlled_executor_context_retry" + "queued_for_controlled_executor_context_retry" if queued and retry_reason == "predecision_context" + else "queued_for_controlled_executor_retry" + if queued and retry_reason else "queued_for_controlled_executor" if queued else "candidate_race_deduplicated" @@ -645,10 +680,15 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( "evidence_ready": 1 if evidence_ready else 0, "automation_run_id": automation_run_id, "work_item_id": _WAZUH_POSTURE_WORK_ITEM_ID, - "retry_of_failed_check_mode": retry_reason == "failed_check_mode", + "retry_reason": retry_reason, + "retry_of_failed_check_mode": retry_reason == "check_mode_failed", "retry_of_incomplete_predecision_context": ( retry_reason == "predecision_context" ), + "retry_of_incomplete_runtime_closure": retry_reason in { + "controlled_apply_failed", + "post_apply_learning_incomplete", + }, "active_blockers": [] if queued else ["candidate_write_not_acknowledged"], } except Exception as exc: @@ -666,6 +706,23 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( } +def _wazuh_posture_retry_reason( + existing_row: Mapping[str, Any] | None, +) -> str | None: + if not existing_row: + return None + check_status = str(existing_row.get("latest_check_status") or "") + apply_status = str(existing_row.get("latest_apply_status") or "") + learning_status = str(existing_row.get("latest_learning_status") or "") + if check_status == "failed": + return "check_mode_failed" + if apply_status == "failed": + return "controlled_apply_failed" + if apply_status == "success" and learning_status != "success": + return "post_apply_learning_incomplete" + return None + + async def enqueue_missing_ansible_candidates_once( *, project_id: str = "awoooi", diff --git a/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py b/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py index 61141b289..7f886bc91 100644 --- a/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py +++ b/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py @@ -238,10 +238,10 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback( check_failed = data.get("check_status") == "failed" check_failure_class = _check_failure_class(data) apply_terminal = data.get("apply_status") in {"success", "failed"} - apply_passed = ( - data.get("apply_status") == "success" - and str(data.get("apply_returncode") or "") == "0" - ) + apply_passed = apply_terminal and str( + data.get("apply_returncode") or "" + ) == "0" + apply_failed = apply_terminal and not apply_passed verifier_passed = data.get("verification_result") == "success" and str( data.get("verifier_run_id") or "" ) == str(data.get("candidate_op_id") or "") @@ -302,7 +302,7 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback( active_blockers.append("wazuh_manager_posture_check_mode_failed") if check_failure_class: active_blockers.append(check_failure_class) - if data.get("apply_status") == "failed": + if apply_failed: active_blockers.append("wazuh_manager_posture_controlled_probe_failed") if apply_passed and not verifier_passed: active_blockers.append("independent_post_verifier_missing_or_failed") diff --git a/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py b/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py index fbba0981f..19027bc93 100644 --- a/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py +++ b/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py @@ -114,6 +114,32 @@ def test_wazuh_runtime_readback_keeps_partial_execution_open() -> None: assert "same_run_mcp_context_missing" in payload["active_blockers"] +def test_wazuh_runtime_readback_separates_probe_success_from_writeback_failure() -> None: + row = _complete_row() + row["apply_status"] = "failed" + row["apply_returncode"] = "0" + row["km_entry_id"] = None + row["stage_ids"] = [ + stage for stage in row["stage_ids"] if stage != "rag_writeback" + ] + + payload = build_iwooos_wazuh_controlled_executor_runtime_readback( + row, + executor_enabled=True, + ) + + assert payload["status"] == ( + "wazuh_manager_posture_executed_runtime_writeback_open" + ) + assert payload["summary"]["bounded_posture_execution_passed_count"] == 1 + assert payload["summary"]["runtime_closed_count"] == 0 + assert "km_rag_playbook_writeback_incomplete" in payload["active_blockers"] + assert ( + "wazuh_manager_posture_controlled_probe_failed" + not in payload["active_blockers"] + ) + + def test_wazuh_runtime_readback_classifies_failed_check_for_bounded_retry() -> None: row = { "candidate_op_id": "00000000-0000-0000-0000-000000000311", @@ -289,6 +315,7 @@ async def test_wazuh_posture_scheduler_enqueues_after_mcp_and_log_evidence( assert recorded["incident"]["asset_scope_aliases"] == ["security_manager_primary"] assert recorded["proposal_data"]["execution_priority"] == 0 assert recorded["incident"]["incident_id"] == "IWZ-POSTURE-2026071100" + assert len(recorded["incident"]["incident_id"]) <= 30 assert ( recorded["incident"]["source_receipt_ref"] == "scheduled-wazuh-manager-posture:2026071100" @@ -380,6 +407,8 @@ async def test_wazuh_posture_scheduler_queues_one_context_retry_per_deploy( "automation_run_id": "00000000-0000-0000-0000-000000000322", "source_receipt_ref": "scheduled-wazuh-manager-posture:2026071100", "latest_check_status": "success", + "latest_apply_status": None, + "latest_learning_status": None, "predecision_evidence_ready": False, } @@ -396,7 +425,8 @@ async def test_wazuh_posture_scheduler_queues_one_context_retry_per_deploy( return _Result() @asynccontextmanager - async def fake_db_context(_project_id: str): + async def fake_db_context(project_id: str): + assert project_id == "awoooi" yield _Db() recorder = AsyncMock(return_value=True) @@ -458,6 +488,105 @@ async def test_wazuh_posture_scheduler_queues_one_context_retry_per_deploy( duplicate_collector.assert_not_awaited() +@pytest.mark.asyncio +async def test_wazuh_posture_scheduler_retries_incomplete_learning_once_per_deploy( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from src.jobs import awooop_ansible_candidate_backfill_job as job + + existing_row = { + "op_id": "00000000-0000-0000-0000-000000000325", + "candidate_status": "success", + "automation_run_id": "00000000-0000-0000-0000-000000000325", + "source_receipt_ref": ( + "scheduled-wazuh-manager-posture:2026071100-69E55718123C" + ), + "latest_check_status": "success", + "latest_apply_status": "success", + "latest_learning_status": None, + "predecision_evidence_ready": True, + } + + class _Mappings: + def first(self): + return existing_row + + class _Result: + def mappings(self): + return _Mappings() + + class _Db: + async def execute(self, _statement, _params): + return _Result() + + @asynccontextmanager + async def fake_db_context(project_id: str): + assert project_id == "awoooi" + yield _Db() + + recorder = AsyncMock(return_value=True) + monkeypatch.setattr(job, "get_db_context", fake_db_context) + monkeypatch.setattr( + job.settings, + "ENABLE_IWOOOS_WAZUH_MANAGER_POSTURE_EXECUTOR", + True, + ) + monkeypatch.setattr( + job.settings, + "IWOOOS_WAZUH_MANAGER_POSTURE_FRESHNESS_HOURS", + 6, + ) + monkeypatch.setattr( + job, + "now_taipei", + MagicMock(return_value=datetime.fromisoformat("2026-07-11T05:59:59+08:00")), + ) + monkeypatch.setenv( + "AWOOOI_BUILD_COMMIT_SHA", + "4bc97290b9ee109c333e11bfdac2f75eb07df567", + ) + + result = await job.enqueue_iwooos_wazuh_manager_posture_candidate_once( + recorder=recorder, + evidence_collector=AsyncMock(return_value=MagicMock(snapshot_id="wazuh-4")), + evidence_verifier=AsyncMock(return_value=True), + ) + + assert result["status"] == "queued_for_controlled_executor_retry" + assert result["queued"] == 1 + assert result["retry_reason"] == "post_apply_learning_incomplete" + assert result["retry_of_failed_check_mode"] is False + assert result["retry_of_incomplete_runtime_closure"] is True + recorded = recorder.await_args.kwargs + assert recorded["incident"]["incident_id"].startswith( + "IWZ-P-2026071100-" + ) + assert len(recorded["incident"]["incident_id"]) == 29 + assert recorded["incident"]["source_receipt_ref"] == ( + "scheduled-wazuh-manager-posture:2026071100-4BC97290B9EE-LEARNING" + ) + + existing_row["automation_run_id"] = result["automation_run_id"] + existing_row["source_receipt_ref"] = recorded["incident"]["source_receipt_ref"] + duplicate_collector = AsyncMock() + duplicate = await job.enqueue_iwooos_wazuh_manager_posture_candidate_once( + recorder=recorder, + evidence_collector=duplicate_collector, + ) + + assert duplicate["status"] == ( + "incomplete_runtime_retry_already_attempted_for_deploy" + ) + assert duplicate["queued"] == 0 + assert duplicate["retry_reason"] == "post_apply_learning_incomplete" + assert duplicate["retry_of_incomplete_runtime_closure"] is True + assert duplicate["active_blockers"] == [ + "bounded_retry_already_attempted_for_deploy" + ] + recorder.assert_awaited_once() + duplicate_collector.assert_not_awaited() + + @pytest.mark.asyncio async def test_wazuh_posture_scheduler_does_not_duplicate_active_candidate( monkeypatch: pytest.MonkeyPatch,