fix(iwooos): expose security runtime gaps truthfully

apps/api/tests/test_iwooos_security_asset_control_plane.py

@@ -71,6 +71,8 @@ def _ready_payload(
     *,
     discovery_status: str = "success",
     discovery_error: str | None = None,
+    coverage_rows: list[SimpleNamespace] | None = None,
+    compliance_rows: list[SimpleNamespace] | None = None,
     automation_rows: list[SimpleNamespace] | None = None,
     runtime_receipt_counts: dict[str, int] | None = None,
     strict_runtime: dict | None = None,
@@ -98,30 +100,38 @@ def _ready_payload(
         _row(asset_type="monitoring_target", cnt=3, unowned_count=0, stale_count=0),
         _row(asset_type="ai_agent", cnt=1, unowned_count=0, stale_count=0),
     ]
-    coverage = [
-        _row(dimension=dimension, status="green", cnt=20)
-        for dimension in (
-            "auto_monitoring",
-            "auto_alerting",
-            "auto_rule_creation",
-            "auto_rule_matching",
-            "auto_playbook",
-            "auto_remediation",
-            "auto_km_creation",
-        )
-    ]
-    compliance = [
-        _row(dimension=dimension, status="compliant", cnt=20)
-        for dimension in (
-            "ssl_cert_valid",
-            "cve_scan",
-            "secret_rotated",
-            "backup_tested",
-            "audit_log_enabled",
-            "access_reviewed",
-            "encryption_at_rest",
-        )
-    ]
+    coverage = (
+        coverage_rows
+        if coverage_rows is not None
+        else [
+            _row(dimension=dimension, status="green", cnt=20)
+            for dimension in (
+                "auto_monitoring",
+                "auto_alerting",
+                "auto_rule_creation",
+                "auto_rule_matching",
+                "auto_playbook",
+                "auto_remediation",
+                "auto_km_creation",
+            )
+        ]
+    )
+    compliance = (
+        compliance_rows
+        if compliance_rows is not None
+        else [
+            _row(dimension=dimension, status="compliant", cnt=20)
+            for dimension in (
+                "ssl_cert_valid",
+                "cve_scan",
+                "secret_rotated",
+                "backup_tested",
+                "audit_log_enabled",
+                "access_reviewed",
+                "encryption_at_rest",
+            )
+        ]
+    )
     automation = (
         automation_rows
         if automation_rows is not None
@@ -226,6 +236,11 @@ def test_builds_live_asset_siem_audit_and_ai_control_plane_without_false_closure
     }
     assert domain_by_id["endpoint_xdr_wazuh"]["status"] == "not_evidenced"
     assert domain_by_id["siem_soc_threat_intel"]["controlled_percent"] == 50
+    work_item_by_control = {item["control_id"]: item for item in payload["work_items"]}
+    assert work_item_by_control["endpoint_xdr_wazuh"]["priority"] == "P0"
+    assert work_item_by_control["endpoint_xdr_wazuh"]["gap_count"] == 1
+    assert work_item_by_control["endpoint_xdr_wazuh"]["gap_percent"] == 100
+    assert payload["work_items"][0]["order"] <= payload["work_items"][1]["order"]
     assert len(payload["siem"]["pipeline"]) == 8
     assert payload["security_audit"]["mcp_tool_audit_count_24h"] == 11
     assert payload["summary"]["package_digest_unpinned_count"] == 0
@@ -256,6 +271,36 @@ def test_builds_live_asset_siem_audit_and_ai_control_plane_without_false_closure
     assert 'secret_value_collection_allowed": true' not in public_text.lower()
 
 
+def test_uncovered_security_domains_are_explicit_ordered_p0_work_items() -> None:
+    payload = _ready_payload(coverage_rows=[], compliance_rows=[])
+
+    work_item_by_control = {item["control_id"]: item for item in payload["work_items"]}
+    for control_id in (
+        "endpoint_xdr_wazuh",
+        "incident_response_soar",
+        "identity_zero_trust",
+        "data_secrets_privacy",
+        "vulnerability_exposure",
+        "telemetry_detection",
+        "backup_dr_resilience",
+        "siem_soc_threat_intel",
+    ):
+        assert work_item_by_control[control_id]["priority"] == "P0"
+
+    p0_orders = [
+        item["order"] for item in payload["work_items"] if item["priority"] == "P0"
+    ]
+    assert p0_orders == sorted(p0_orders)
+    assert all(item["order"] < 999 for item in payload["work_items"])
+    assert [item["control_id"] for item in payload["work_items"][:5]] == [
+        "asset_scope",
+        "asset_ownership",
+        "automation_coverage",
+        "telemetry_detection",
+        "endpoint_xdr_wazuh",
+    ]
+
+
 def test_projects_canonical_runtime_receipts_without_false_same_run_closure() -> None:
     payload = _ready_payload(
         automation_rows=[
@@ -318,8 +363,7 @@ def test_projects_exact_authoritative_same_run_runtime_closure() -> None:
     assert payload["ai_automation"]["same_run_closed_loop_proven"] is True
     assert payload["ai_automation"]["same_run_closed_loop_count"] == 1
     assert all(
-        item["work_item_id"] != "AIA-P0-008-01"
-        for item in payload["work_items"]
+        item["work_item_id"] != "AIA-P0-008-01" for item in payload["work_items"]
     )
     assert payload["completion"]["overall_percent"] == round(
         sum(
@@ -372,8 +416,7 @@ def test_strict_runtime_projection_fails_closed_on_contract_or_receipt_drift() -
         assert payload["completion"]["strict_runtime_closure_percent"] == 0
         assert payload["ai_automation"]["same_run_closed_loop_proven"] is False
         assert any(
-            item["work_item_id"] == "AIA-P0-008-01"
-            for item in payload["work_items"]
+            item["work_item_id"] == "AIA-P0-008-01" for item in payload["work_items"]
         )
 
 
@@ -886,8 +929,14 @@ def test_service_bounds_source_concurrency_for_database_budget(monkeypatch) -> N
         for statement in statements
         if "asset_compliance_snapshot" in statement
     )
-    assert "ORDER BY snapshot_id DESC" in compliance_query
-    assert "LIMIT 100000" in compliance_query
+    assert "CROSS JOIN dimensions" in compliance_query
+    assert "LEFT JOIN LATERAL" in compliance_query
+    assert (
+        "ORDER BY snapshot.detected_at DESC, snapshot.snapshot_id DESC"
+        in compliance_query
+    )
+    assert "latest.detected_at < NOW() - INTERVAL '36 hours'" in compliance_query
+    assert "THEN 'unknown'" in compliance_query
     inventory_query = next(
         statement for statement in statements if "FROM asset_inventory" in statement
     )
@@ -908,6 +957,12 @@ def test_service_bounds_source_concurrency_for_database_budget(monkeypatch) -> N
     assert "executor_returncode_trusted" in siem_query
     assert "learning_writeback_count_24h" in siem_query
     assert "ansible_learning_writeback_recorded" in siem_query
+    audit_query = next(
+        statement for statement in statements if "WITH k8s AS" in statement
+    )
+    assert "count(*) FILTER (WHERE success IS FALSE)" in audit_query
+    assert "CROSS JOIN mcp" in audit_query
+    assert "CROSS JOIN asset_changes" in audit_query
     assert "repository_readback_verified" in siem_query