fix(iwooos): expose closed Wazuh runtime lane

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