fix(iwooos): isolate control plane source reads

apps/api/tests/test_iwooos_security_asset_control_plane.py

@@ -16,6 +16,7 @@ from fastapi.testclient import TestClient
 from src.api.v1.iwooos import router
 from src.services.iwooos_security_asset_control_plane import (
     IwoooSSecurityAssetControlPlaneService,
+    _read_public_source,
     build_iwooos_security_asset_control_plane,
     build_unavailable_iwooos_security_asset_control_plane,
 )
@@ -25,7 +26,7 @@ def _row(**values):
     return SimpleNamespace(**values)
 
 
-def _ready_payload() -> dict:
+def _ready_payload(source_health: list[dict] | None = None) -> dict:
     now = datetime(2026, 7, 11, 4, 0, tzinfo=UTC)
     discovery = _row(
         latest_status="success",
@@ -111,6 +112,7 @@ def _ready_payload() -> dict:
             asset_change_count_24h=4,
             asset_change_ai_analysis_count_24h=3,
         ),
+        source_health=source_health,
         generated_at=now,
     )
 
@@ -201,6 +203,60 @@ def test_unknown_coverage_and_compliance_create_p0_work_items() -> None:
     } <= work_item_ids
 
 
+def test_partial_source_failure_is_visible_without_erasing_core_inventory() -> None:
+    source_health = [
+        {"source_id": "asset_discovery", "status": "ready", "reason_code": None},
+        {"source_id": "asset_inventory", "status": "ready", "reason_code": None},
+        {
+            "source_id": "audit_runtime",
+            "status": "unavailable",
+            "reason_code": "audit_runtime_query_failed",
+        },
+    ]
+    payload = _ready_payload(source_health)
+
+    assert payload["status"] == "degraded"
+    assert payload["source_status"] == "live_database_partial"
+    assert payload["summary"]["managed_asset_count"] == 24
+    assert payload["summary"]["ready_source_count"] == 2
+    assert payload["summary"]["unavailable_source_count"] == 1
+    assert payload["source_health"] == source_health
+    assert payload["work_items"][0]["work_item_id"] == "AIA-P0-006-00"
+
+
+def test_read_public_source_returns_redacted_reason_code_on_query_failure() -> None:
+    class NestedContext:
+        async def __aenter__(self):
+            return None
+
+        async def __aexit__(self, exc_type, exc, traceback):
+            return False
+
+    class FailingDb:
+        def begin_nested(self):
+            return NestedContext()
+
+        async def execute(self, statement):
+            raise RuntimeError("private table detail")
+
+    value, health = asyncio.run(
+        _read_public_source(
+            FailingDb(),
+            source_id="audit_runtime",
+            statement="SELECT 1",
+            result_mode="one",
+            default={},
+        )
+    )
+
+    assert value == {}
+    assert health == {
+        "source_id": "audit_runtime",
+        "status": "unavailable",
+        "reason_code": "audit_runtime_query_failed",
+    }
+
+
 def test_unavailable_payload_is_fixed_degraded_contract_without_raw_error() -> None:
     payload = build_unavailable_iwooos_security_asset_control_plane(
         "database_query_failed"