fix(iwooos): retry incomplete wazuh closure

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",

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")

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,