From ac1a30109b1b68e4d80f3252c30593b64bbc4cbb Mon Sep 17 00:00:00 2001 From: Your Name Date: Wed, 22 Jul 2026 18:02:45 +0800 Subject: [PATCH] fix(iwooos): retry expired Wazuh posture claim --- .../awooop_ansible_candidate_backfill_job.py | 20 ++++ ...wooos_wazuh_controlled_executor_runtime.py | 105 ++++++++++++++++++ 2 files changed, 125 insertions(+) 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 dd34d686f..f07232782 100644 --- a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py +++ b/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py @@ -70,6 +70,7 @@ _WAZUH_INCOMPLETE_RUNTIME_RETRY_REASONS = frozenset( "controlled_apply_failed", "post_apply_learning_incomplete", "recipient_visible_transition_missing", + "scheduled_candidate_recurrence_expired_before_claim", } ) _WAZUH_INGRESS_CATALOG_ID = "ansible:wazuh-alertmanager-integration" @@ -1219,6 +1220,8 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( telegram.send_status AS latest_telegram_send_status, telegram.provider_message_id AS latest_telegram_provider_message_id, + execution_skip.input ->> 'not_used_reason' + AS latest_execution_skip_reason, EXISTS ( SELECT 1 FROM incident_evidence evidence @@ -1290,6 +1293,14 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( ORDER BY packet.created_at DESC LIMIT 1 ) capability_packet ON TRUE + LEFT JOIN LATERAL ( + SELECT skipped.input + FROM automation_operation_log skipped + WHERE skipped.parent_op_id = candidate.op_id + AND skipped.operation_type = 'ansible_execution_skipped' + ORDER BY skipped.created_at DESC, skipped.op_id DESC + LIMIT 1 + ) execution_skip ON TRUE LEFT JOIN LATERAL ( SELECT outbound.send_status, outbound.provider_message_id @@ -1681,6 +1692,10 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( retry_attempt_ref = f"{deploy_ref}-learning" elif retry_reason == "recipient_visible_transition_missing": retry_attempt_ref = f"{deploy_ref}-receipt" + elif retry_reason == ( + "scheduled_candidate_recurrence_expired_before_claim" + ): + retry_attempt_ref = f"{deploy_ref}-recurrence" elif latest_check_status == "success": retry_attempt_ref = f"{deploy_ref}-context" retry_reason = "predecision_context" @@ -2001,6 +2016,11 @@ def _wazuh_posture_retry_reason( 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 "") + execution_skip_reason = str( + existing_row.get("latest_execution_skip_reason") or "" + ) + if execution_skip_reason == "scheduled_candidate_identity_chain_not_verified": + return "scheduled_candidate_recurrence_expired_before_claim" if check_status == "failed": return "check_mode_failed" if apply_status == "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 d5967fa1a..c27133f42 100644 --- a/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py +++ b/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py @@ -2764,6 +2764,111 @@ async def test_wazuh_posture_scheduler_queues_one_recipient_transition_per_deplo duplicate_collector.assert_not_awaited() +@pytest.mark.asyncio +async def test_wazuh_posture_scheduler_retries_expired_preclaim_recurrence_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-000000000327", + "candidate_status": "dry_run", + "automation_run_id": "00000000-0000-0000-0000-000000000327", + "source_receipt_ref": "scheduled-wazuh-manager-posture:2026071100", + "latest_check_status": None, + "latest_apply_status": None, + "latest_learning_status": None, + "latest_execution_skip_reason": ( + "scheduled_candidate_identity_chain_not_verified" + ), + "predecision_evidence_ready": True, + } + + class _Mappings: + def first(self): + return existing_row + + class _Result: + def mappings(self): + return _Mappings() + + observed_sql: list[str] = [] + + class _Db: + async def execute(self, statement, _params): + observed_sql.append(str(statement)) + 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", + "abcdef1234567890abcdef1234567890abcdef12", + ) + + result = await job.enqueue_iwooos_wazuh_manager_posture_candidate_once( + recorder=recorder, + evidence_collector=AsyncMock(return_value=MagicMock(snapshot_id="wazuh-6")), + evidence_verifier=AsyncMock(return_value=True), + source_recurrence=_current_wazuh_source_recurrence( + "2026-07-11T05:59:58+08:00" + ), + ) + + assert result["status"] == "queued_for_controlled_executor_retry" + assert result["retry_reason"] == ( + "scheduled_candidate_recurrence_expired_before_claim" + ) + assert result["retry_of_incomplete_runtime_closure"] is True + assert "ansible_execution_skipped" in observed_sql[0] + assert "latest_execution_skip_reason" in observed_sql[0] + recorded = recorder.await_args.kwargs + assert recorded["incident"]["source_receipt_ref"] == ( + "scheduled-wazuh-manager-posture:2026071100-ABCDEF123456-RECURRENCE" + ) + + 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, + source_recurrence=_current_wazuh_source_recurrence( + "2026-07-11T05:59:58+08:00" + ), + ) + + assert duplicate["status"] == ( + "incomplete_runtime_retry_already_attempted_for_deploy" + ) + assert duplicate["queued"] == 0 + assert duplicate["retry_reason"] == ( + "scheduled_candidate_recurrence_expired_before_claim" + ) + 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,