From 0fe6bd3ba064b6eb367347e24602ddb524c7660e Mon Sep 17 00:00:00 2001 From: ogt Date: Sat, 11 Jul 2026 11:57:38 +0800 Subject: [PATCH] fix(agent): fail closed on incomplete Wazuh context --- .../awooop_ansible_candidate_backfill_job.py | 136 ++++++++++++--- ...wooos_wazuh_controlled_executor_runtime.py | 163 +++++++++++++++++- 2 files changed, 270 insertions(+), 29 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 f51580162..3bccde124 100644 --- a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py +++ b/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py @@ -359,8 +359,17 @@ def _wazuh_posture_incident( receipt_ref = ( f"{bucket_ref}-{attempt_ref.upper()}" if attempt_ref else bucket_ref ) + incident_id = f"IWZ-POSTURE-{bucket_ref}" + if attempt_ref: + attempt_token = uuid5( + NAMESPACE_URL, + f"awoooi:iwooos:wazuh-manager-posture-attempt:{attempt_ref}", + ).hex[:12].upper() + incident_id = f"IWZ-P-{bucket_ref}-{attempt_token}" return { - "incident_id": f"IWZ-POSTURE-{receipt_ref}", + # IncidentEvidence.incident_id is VARCHAR(30). Keep retry identities + # deterministic and bounded while preserving the full ref separately. + "incident_id": incident_id, "project_id": "awoooi", "status": "INVESTIGATING", "severity": "low", @@ -422,7 +431,29 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( candidate.status AS candidate_status, 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 + check_mode.status AS latest_check_status, + EXISTS ( + SELECT 1 + FROM incident_evidence evidence + WHERE evidence.incident_id = candidate.input ->> 'incident_id' + AND evidence.collected_at <= candidate.created_at + AND ( + evidence.post_execution_state IS NULL + OR CAST(evidence.post_execution_state AS jsonb) = 'null'::jsonb + ) + AND evidence.sensors_succeeded > 0 + AND NULLIF(btrim(evidence.recent_logs), '') IS NOT NULL + AND EXISTS ( + SELECT 1 + FROM jsonb_each( + coalesce( + CAST(evidence.mcp_health AS jsonb), + '{}'::jsonb + ) + ) health + WHERE health.value = 'true'::jsonb + ) + ) AS predecision_evidence_ready FROM automation_operation_log candidate LEFT JOIN LATERAL ( SELECT check_receipt.status @@ -446,8 +477,20 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( }, ) existing_row = existing.mappings().first() + existing_evidence_ready = bool( + existing_row + and existing_row.get("predecision_evidence_ready") is True + ) + latest_check_status = str( + existing_row.get("latest_check_status") or "" + ) if existing_row else "" retry_attempt_ref: str | None = None - if existing_row and existing_row.get("latest_check_status") != "failed": + retry_reason: str | None = None + if ( + existing_row + and latest_check_status != "failed" + and existing_evidence_ready + ): return { "status": "fresh_candidate_already_present", "queued": 0, @@ -463,30 +506,63 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( } if existing_row: source_sha = os.getenv("AWOOOI_BUILD_COMMIT_SHA", "").strip().lower() - retry_attempt_ref = ( + deploy_ref = ( source_sha[:12] if len(source_sha) >= 12 and all(character in "0123456789abcdef" for character in source_sha) else "runtime-retry" ) - existing_source_ref = str( - existing_row.get("source_receipt_ref") or "" - ).lower() - if existing_source_ref.endswith(f"-{retry_attempt_ref.lower()}"): + if latest_check_status == "failed": + retry_attempt_ref = deploy_ref + retry_reason = "failed_check_mode" + elif latest_check_status == "success": + retry_attempt_ref = f"{deploy_ref}-context" + retry_reason = "predecision_context" + else: return { - "status": "failed_check_retry_already_attempted_for_deploy", + "status": "fresh_candidate_predecision_evidence_not_verified", "queued": 0, "existing": 1, - "evidence_ready": 1, + "evidence_ready": 0, "automation_run_id": str( existing_row.get("automation_run_id") or existing_row.get("op_id") or "" ), "work_item_id": _WAZUH_POSTURE_WORK_ITEM_ID, - "retry_of_failed_check_mode": True, + "active_blockers": [ + "predecision_mcp_or_log_evidence_missing" + ], + } + existing_source_ref = str( + existing_row.get("source_receipt_ref") or "" + ).lower() + if existing_source_ref.endswith(f"-{retry_attempt_ref.lower()}"): + return { + "status": ( + "failed_check_retry_already_attempted_for_deploy" + if retry_reason == "failed_check_mode" + else "context_retry_already_attempted_for_deploy" + ), + "queued": 0, + "existing": 1, + "evidence_ready": 1 if existing_evidence_ready else 0, + "automation_run_id": str( + existing_row.get("automation_run_id") + or existing_row.get("op_id") + or "" + ), + "work_item_id": _WAZUH_POSTURE_WORK_ITEM_ID, + "retry_of_failed_check_mode": ( + retry_reason == "failed_check_mode" + ), + "retry_of_incomplete_predecision_context": ( + retry_reason == "predecision_context" + ), "active_blockers": [ "bounded_retry_already_attempted_for_deploy" + if retry_reason == "failed_check_mode" + else "bounded_context_retry_already_attempted_for_deploy" ], } @@ -522,6 +598,23 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( error_type=type(exc).__name__, ) + if not evidence_ready: + return { + "status": "blocked_predecision_evidence_not_verified", + "queued": 0, + "existing": 1 if existing_row else 0, + "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_of_incomplete_predecision_context": ( + retry_reason == "predecision_context" + ), + "active_blockers": [ + "predecision_mcp_or_log_evidence_missing" + ], + } + queued = await recorder( incident=incident, proposal_data={ @@ -540,26 +633,23 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( return { "status": ( "queued_for_controlled_executor_retry" - if queued and evidence_ready and retry_attempt_ref + if queued and retry_reason == "failed_check_mode" + else "queued_for_controlled_executor_context_retry" + if queued and retry_reason == "predecision_context" else "queued_for_controlled_executor" - if queued and evidence_ready - else ( - "queued_for_controlled_executor_retry_with_degraded_context" - if queued and retry_attempt_ref - else "queued_for_controlled_executor_with_degraded_context" - if queued - else "candidate_race_deduplicated" - ) + if queued + else "candidate_race_deduplicated" ), "queued": 1 if queued else 0, "existing": 0 if queued else 1, "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": bool(retry_attempt_ref), - "active_blockers": ( - [] if evidence_ready else ["predecision_mcp_or_log_evidence_missing"] + "retry_of_failed_check_mode": retry_reason == "failed_check_mode", + "retry_of_incomplete_predecision_context": ( + retry_reason == "predecision_context" ), + "active_blockers": [] if queued else ["candidate_write_not_acknowledged"], } except Exception as exc: logger.warning( 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 9094f0a11..c1dbfbed5 100644 --- a/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py +++ b/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py @@ -338,15 +338,106 @@ async def test_wazuh_posture_scheduler_retries_failed_check_once_per_deploy( assert result["queued"] == 1 assert result["retry_of_failed_check_mode"] is True recorded = recorder.await_args.kwargs - assert recorded["incident"]["incident_id"] == ( - "IWZ-POSTURE-2026071100-ABCDEF123456" + 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-ABCDEF123456" ) assert recorded["automation_run_id"] == result["automation_run_id"] +@pytest.mark.asyncio +async def test_wazuh_posture_scheduler_queues_one_context_retry_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-000000000322", + "candidate_status": "success", + "automation_run_id": "00000000-0000-0000-0000-000000000322", + "source_receipt_ref": "scheduled-wazuh-manager-posture:2026071100", + "latest_check_status": "success", + "predecision_evidence_ready": False, + } + + 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): + 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-3")), + evidence_verifier=AsyncMock(return_value=True), + ) + + assert result["status"] == "queued_for_controlled_executor_context_retry" + assert result["queued"] == 1 + assert result["retry_of_failed_check_mode"] is False + assert result["retry_of_incomplete_predecision_context"] is True + recorded = recorder.await_args.kwargs + assert recorded["incident"]["source_receipt_ref"] == ( + "scheduled-wazuh-manager-posture:2026071100-ABCDEF123456-CONTEXT" + ) + assert recorded["incident"]["incident_id"].startswith( + "IWZ-P-2026071100-" + ) + assert len(recorded["incident"]["incident_id"]) == 29 + + 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"] == "context_retry_already_attempted_for_deploy" + assert duplicate["queued"] == 0 + assert duplicate["retry_of_incomplete_predecision_context"] is True + assert duplicate["active_blockers"] == [ + "bounded_context_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, @@ -360,6 +451,7 @@ async def test_wazuh_posture_scheduler_does_not_duplicate_active_candidate( "candidate_status": "dry_run", "automation_run_id": "00000000-0000-0000-0000-000000000331", "latest_check_status": "pending", + "predecision_evidence_ready": True, } class _Result: @@ -396,6 +488,65 @@ async def test_wazuh_posture_scheduler_does_not_duplicate_active_candidate( recorder.assert_not_awaited() +@pytest.mark.asyncio +async def test_wazuh_posture_scheduler_does_not_retry_pending_candidate_without_context( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from src.jobs import awooop_ansible_candidate_backfill_job as job + + class _Mappings: + def first(self): + return { + "op_id": "00000000-0000-0000-0000-000000000332", + "candidate_status": "dry_run", + "automation_run_id": "00000000-0000-0000-0000-000000000332", + "latest_check_status": "pending", + "predecision_evidence_ready": False, + } + + 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): + yield _Db() + + recorder = AsyncMock(return_value=True) + collector = AsyncMock() + 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, + ) + + result = await job.enqueue_iwooos_wazuh_manager_posture_candidate_once( + recorder=recorder, + evidence_collector=collector, + ) + + assert result["status"] == ( + "fresh_candidate_predecision_evidence_not_verified" + ) + assert result["queued"] == 0 + assert result["evidence_ready"] == 0 + assert result["active_blockers"] == [ + "predecision_mcp_or_log_evidence_missing" + ] + recorder.assert_not_awaited() + collector.assert_not_awaited() + + @pytest.mark.asyncio async def test_wazuh_posture_scheduler_limits_failed_retry_to_once_per_deploy( monkeypatch: pytest.MonkeyPatch, @@ -463,7 +614,7 @@ async def test_wazuh_posture_scheduler_limits_failed_retry_to_once_per_deploy( @pytest.mark.asyncio -async def test_wazuh_posture_scheduler_queues_bounded_probe_with_degraded_context( +async def test_wazuh_posture_scheduler_fails_closed_with_degraded_context( monkeypatch: pytest.MonkeyPatch, ) -> None: from src.jobs import awooop_ansible_candidate_backfill_job as job @@ -503,11 +654,11 @@ async def test_wazuh_posture_scheduler_queues_bounded_probe_with_degraded_contex evidence_verifier=AsyncMock(return_value=False), ) - assert result["status"] == ("queued_for_controlled_executor_with_degraded_context") - assert result["queued"] == 1 + assert result["status"] == "blocked_predecision_evidence_not_verified" + assert result["queued"] == 0 assert result["evidence_ready"] == 0 assert result["active_blockers"] == ["predecision_mcp_or_log_evidence_missing"] - recorder.assert_awaited_once() + recorder.assert_not_awaited() def test_wazuh_runtime_api_is_public_safe(monkeypatch: pytest.MonkeyPatch) -> None: