From aaec2bfb4f67b89686809e1b6c6aa2d4359316ed Mon Sep 17 00:00:00 2001 From: Your Name Date: Wed, 22 Jul 2026 18:12:39 +0800 Subject: [PATCH] fix(iwooos): expose closed Wazuh runtime lane --- apps/api/src/api/v1/iwooos.py | 5 +- ...wooos_wazuh_controlled_executor_runtime.py | 83 ++++++++- ...est_iwooos_wazuh_runtime_lane_selection.py | 176 ++++++++++++++++++ 3 files changed, 258 insertions(+), 6 deletions(-) create mode 100644 apps/api/tests/test_iwooos_wazuh_runtime_lane_selection.py diff --git a/apps/api/src/api/v1/iwooos.py b/apps/api/src/api/v1/iwooos.py index e1f01a88c..a63176a33 100644 --- a/apps/api/src/api/v1/iwooos.py +++ b/apps/api/src/api/v1/iwooos.py @@ -839,8 +839,9 @@ async def preview_iwooos_wazuh_controlled_executor_handoff( response_model=dict[str, Any], summary="取得 Wazuh controlled executor live runtime receipt 讀回", description=( - "優先讀取內部 worker 排程的 Wazuh sanitized alert-ingress convergence，" - "尚無 ingress candidate 時才回退到 manager bounded no-write posture executor；" + "同時讀取內部 worker 排程的 Wazuh sanitized alert-ingress convergence 與 manager " + "bounded no-write posture executor；已閉環 lane 優先，兩者皆未閉環時維持 ingress " + "candidate 優先，並公開兩 lane 的脫敏摘要，避免 blocked lane 遮住 closure truth。" "candidate、check-mode、execution、independent verifier、KM / RAG / PlayBook、" "MCP context 與 Telegram durable receipts。此端點不執行命令、不回傳 command output、" "inventory identity、主機位址、raw Wazuh payload 或機密值。" 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 7a805000a..19d99dc86 100644 --- a/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py +++ b/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py @@ -646,15 +646,90 @@ _LIVE_RUNTIME_SQL = """ async def load_latest_iwooos_wazuh_controlled_executor_runtime( *, project_id: str = "awoooi" ) -> dict[str, Any]: - """Read the latest Wazuh posture executor chain from durable receipts.""" + """Read both Wazuh lanes without letting a blocked lane hide closure.""" lanes = await load_iwooos_wazuh_controlled_executor_runtime_lanes( project_id=project_id ) + selected_lane_id, selection_reason = _select_wazuh_runtime_lane(lanes) + payload = dict(lanes[selected_lane_id]) + payload["lane_selection"] = { + "schema_version": "iwooos_wazuh_runtime_lane_selection_v1", + "selected_lane_id": selected_lane_id, + "selection_reason": selection_reason, + "lanes": { + lane_id: _wazuh_runtime_lane_summary(lane_id, lane) + for lane_id, lane in lanes.items() + }, + } + return payload + + +def _select_wazuh_runtime_lane( + lanes: Mapping[str, Mapping[str, Any]], +) -> tuple[str, str]: + """Choose closed truth first, then retain ingress convergence priority.""" + ingress = lanes["ingress"] - if ingress["summary"]["dispatch_candidate_count"] > 0: - return ingress - return lanes["posture"] + posture = lanes["posture"] + ingress_closed = _wazuh_runtime_lane_closed(ingress) + posture_closed = _wazuh_runtime_lane_closed(posture) + if ingress_closed: + return "ingress", "ingress_runtime_closed" + if posture_closed: + return "posture", "posture_runtime_closed_ingress_open" + ingress_summary = ingress.get("summary") + ingress_candidate_count = ( + _nonnegative_int(ingress_summary.get("dispatch_candidate_count")) + if isinstance(ingress_summary, Mapping) + else 0 + ) + if ingress_candidate_count > 0: + return "ingress", "ingress_candidate_active" + return "posture", "posture_fallback_no_ingress_candidate" + + +def _wazuh_runtime_lane_closed(lane: Mapping[str, Any]) -> bool: + summary = lane.get("summary") + return bool( + isinstance(summary, Mapping) + and _nonnegative_int(summary.get("runtime_closed_count")) > 0 + ) + + +def _wazuh_runtime_lane_summary( + lane_id: str, + lane: Mapping[str, Any], +) -> dict[str, Any]: + summary = lane.get("summary") + if not isinstance(summary, Mapping): + summary = {} + return { + "lane_id": lane_id, + "work_item_id": str(lane.get("work_item_id") or "") or None, + "status": str(lane.get("status") or "") or None, + "latest_receipt_at": ( + str(lane.get("latest_receipt_at") or "") or None + ), + "selected_catalog_id": ( + str(summary.get("selected_catalog_id") or "") or None + ), + "dispatch_candidate_count": _nonnegative_int( + summary.get("dispatch_candidate_count") + ), + "check_mode_passed_count": _nonnegative_int( + summary.get("check_mode_passed_count") + ), + "check_mode_failed_count": _nonnegative_int( + summary.get("check_mode_failed_count") + ), + "same_run_completion_percent": _nonnegative_int( + summary.get("same_run_completion_percent") + ), + "runtime_closed": _wazuh_runtime_lane_closed(lane), + "missing_stage_ids": list(lane.get("missing_stage_ids") or []), + "active_blockers": list(lane.get("active_blockers") or []), + } async def load_iwooos_wazuh_controlled_executor_runtime_lanes( diff --git a/apps/api/tests/test_iwooos_wazuh_runtime_lane_selection.py b/apps/api/tests/test_iwooos_wazuh_runtime_lane_selection.py new file mode 100644 index 000000000..d955611ca --- /dev/null +++ b/apps/api/tests/test_iwooos_wazuh_runtime_lane_selection.py @@ -0,0 +1,176 @@ +from __future__ import annotations + +from typing import Any + +import pytest + +from src.services import iwooos_wazuh_controlled_executor_runtime as runtime + + +def _lane( + lane_id: str, + *, + candidate_count: int, + completion_percent: int, + closed: bool, + failed: bool = False, +) -> dict[str, Any]: + return { + "schema_version": "iwooos_wazuh_controlled_executor_runtime_readback_v3", + "status": f"{lane_id}_{'closed' if closed else 'open'}", + "work_item_id": f"wazuh:{lane_id}", + "latest_receipt_at": f"2026-07-22T0{1 if lane_id == 'ingress' else 2}:00:00Z", + "summary": { + "selected_catalog_id": f"ansible:wazuh-{lane_id}", + "dispatch_candidate_count": candidate_count, + "check_mode_passed_count": 1 if closed else 0, + "check_mode_failed_count": 1 if failed else 0, + "same_run_completion_percent": completion_percent, + "runtime_closed_count": 1 if closed else 0, + }, + "missing_stage_ids": [] if closed else ["controlled_apply"], + "active_blockers": ["check_mode_failed"] if failed else [], + } + + +@pytest.mark.asyncio +async def test_closed_posture_lane_is_not_hidden_by_blocked_ingress( + monkeypatch: pytest.MonkeyPatch, +) -> None: + lanes = { + "ingress": _lane( + "ingress", + candidate_count=1, + completion_percent=6, + closed=False, + failed=True, + ), + "posture": _lane( + "posture", + candidate_count=1, + completion_percent=100, + closed=True, + ), + } + + async def fake_lanes(*, project_id: str) -> dict[str, dict[str, Any]]: + assert project_id == "awoooi" + return lanes + + monkeypatch.setattr( + runtime, + "load_iwooos_wazuh_controlled_executor_runtime_lanes", + fake_lanes, + ) + + payload = await runtime.load_latest_iwooos_wazuh_controlled_executor_runtime() + + assert payload["work_item_id"] == "wazuh:posture" + selection = payload["lane_selection"] + assert selection["selected_lane_id"] == "posture" + assert selection["selection_reason"] == ( + "posture_runtime_closed_ingress_open" + ) + assert selection["lanes"]["posture"]["runtime_closed"] is True + assert selection["lanes"]["ingress"]["runtime_closed"] is False + assert selection["lanes"]["ingress"]["check_mode_failed_count"] == 1 + assert selection["lanes"]["ingress"]["latest_receipt_at"] == ( + "2026-07-22T01:00:00Z" + ) + + +@pytest.mark.asyncio +async def test_closed_ingress_lane_keeps_convergence_priority( + monkeypatch: pytest.MonkeyPatch, +) -> None: + lanes = { + "ingress": _lane( + "ingress", + candidate_count=1, + completion_percent=100, + closed=True, + ), + "posture": _lane( + "posture", + candidate_count=1, + completion_percent=100, + closed=True, + ), + } + + async def fake_lanes(*, project_id: str) -> dict[str, dict[str, Any]]: + assert project_id == "awoooi" + return lanes + + monkeypatch.setattr( + runtime, + "load_iwooos_wazuh_controlled_executor_runtime_lanes", + fake_lanes, + ) + + payload = await runtime.load_latest_iwooos_wazuh_controlled_executor_runtime() + + assert payload["work_item_id"] == "wazuh:ingress" + assert payload["lane_selection"]["selected_lane_id"] == "ingress" + assert payload["lane_selection"]["selection_reason"] == ( + "ingress_runtime_closed" + ) + + +@pytest.mark.asyncio +async def test_active_ingress_candidate_wins_when_neither_lane_is_closed( + monkeypatch: pytest.MonkeyPatch, +) -> None: + lanes = { + "ingress": _lane( + "ingress", + candidate_count=1, + completion_percent=44, + closed=False, + ), + "posture": _lane( + "posture", + candidate_count=1, + completion_percent=89, + closed=False, + ), + } + + async def fake_lanes(*, project_id: str) -> dict[str, dict[str, Any]]: + assert project_id == "awoooi" + return lanes + + monkeypatch.setattr( + runtime, + "load_iwooos_wazuh_controlled_executor_runtime_lanes", + fake_lanes, + ) + + payload = await runtime.load_latest_iwooos_wazuh_controlled_executor_runtime() + + assert payload["work_item_id"] == "wazuh:ingress" + assert payload["lane_selection"]["selection_reason"] == ( + "ingress_candidate_active" + ) + + +def test_posture_lane_is_fallback_without_ingress_candidate() -> None: + lanes = { + "ingress": _lane( + "ingress", + candidate_count=0, + completion_percent=0, + closed=False, + ), + "posture": _lane( + "posture", + candidate_count=1, + completion_percent=50, + closed=False, + ), + } + + assert runtime._select_wazuh_runtime_lane(lanes) == ( + "posture", + "posture_fallback_no_ingress_candidate", + )