fix(iwooos): isolate control plane source reads

apps/api/src/services/iwooos_security_asset_control_plane.py

@@ -1069,6 +1069,9 @@ def build_unavailable_iwooos_security_asset_control_plane(
         "source_status": "live_database_unavailable",
         "reason_code": reason_code,
         "summary": {
+            "source_count": 9,
+            "ready_source_count": 0,
+            "unavailable_source_count": 9,
             "managed_asset_count": 0,
             "expected_asset_type_count": len(_ASSET_TYPES),
             "present_asset_type_count": 0,
@@ -1212,6 +1215,24 @@ def build_unavailable_iwooos_security_asset_control_plane(
             "accepted_ai_trace_count_24h": 0,
             "immutable_external_audit_store_evidenced": False,
         },
+        "source_health": [
+            {
+                "source_id": source_id,
+                "status": "unavailable",
+                "reason_code": reason_code,
+            }
+            for source_id in (
+                "asset_discovery",
+                "asset_inventory",
+                "automation_coverage",
+                "asset_relationship",
+                "security_compliance",
+                "automation_operations",
+                "ai_decision_trace",
+                "siem_runtime",
+                "audit_runtime",
+            )
+        ],
         "work_items": [
             _work_item(
                 "AIA-P0-006-01",
@@ -1250,6 +1271,44 @@ def build_unavailable_iwooos_security_asset_control_plane(
     }
 
 
+async def _read_public_source(
+    db: Any,
+    *,
+    source_id: str,
+    statement: Any,
+    result_mode: str,
+    default: Any,
+) -> tuple[Any, dict[str, Any]]:
+    """Isolate one aggregate query so an optional source cannot abort the readback."""
+    try:
+        async with db.begin_nested():
+            result = await db.execute(statement)
+            if result_mode == "all":
+                value = result.fetchall()
+            elif result_mode == "one":
+                value = result.one()
+            elif result_mode == "one_or_none":
+                value = result.one_or_none()
+            else:
+                raise ValueError("unsupported_result_mode")
+        return value, {
+            "source_id": source_id,
+            "status": "ready",
+            "reason_code": None,
+        }
+    except Exception as exc:  # noqa: BLE001 - isolate and redact source failures
+        logger.warning(
+            "iwooos_security_asset_control_plane_source_unavailable",
+            source_id=source_id,
+            error_type=type(exc).__name__,
+        )
+        return default, {
+            "source_id": source_id,
+            "status": "unavailable",
+            "reason_code": f"{source_id}_query_failed",
+        }
+
+
 class IwoooSSecurityAssetControlPlaneService:
     """Read aggregate production security state with a bounded DB timeout."""
 
@@ -1274,8 +1333,12 @@ class IwoooSSecurityAssetControlPlaneService:
 
     async def _load_snapshot(self) -> dict[str, Any]:
         async with get_db_context("awoooi") as db:
-            discovery_result = await db.execute(
+            source_health: list[dict[str, Any]] = []
-                _sql(
+
+            discovery_row, health = await _read_public_source(
+                db,
+                source_id="asset_discovery",
+                statement=_sql(
                     """
                     SELECT
                         latest.status AS latest_status,
@@ -1303,12 +1366,16 @@ class IwoooSSecurityAssetControlPlaneService:
                         LIMIT 1
                     ) successful ON true
                     """
+                ),
+                result_mode="one_or_none",
+                default=None,
             )
-            )
+            source_health.append(health)
-            discovery_row = discovery_result.one_or_none()
 
-            inventory_result = await db.execute(
+            inventory_rows, health = await _read_public_source(
-                _sql(
+                db,
+                source_id="asset_inventory",
+                statement=_sql(
                     """
                     SELECT
                         asset_type,
@@ -1324,12 +1391,16 @@ class IwoooSSecurityAssetControlPlaneService:
                     GROUP BY asset_type
                     ORDER BY asset_type
                     """
+                ),
+                result_mode="all",
+                default=[],
             )
-            )
+            source_health.append(health)
-            inventory_rows = inventory_result.fetchall()
 
-            coverage_result = await db.execute(
+            coverage_rows, health = await _read_public_source(
-                _sql(
+                db,
+                source_id="automation_coverage",
+                statement=_sql(
                     """
                     SELECT dimension, coverage_status AS status, count(*) AS cnt
                     FROM asset_coverage_snapshot
@@ -1343,12 +1414,16 @@ class IwoooSSecurityAssetControlPlaneService:
                     GROUP BY dimension, coverage_status
                     ORDER BY dimension, coverage_status
                     """
+                ),
+                result_mode="all",
+                default=[],
             )
-            )
+            source_health.append(health)
-            coverage_rows = coverage_result.fetchall()
 
-            relationship_result = await db.execute(
+            relationship_row, health = await _read_public_source(
-                _sql(
+                db,
+                source_id="asset_relationship",
+                statement=_sql(
                     """
                     SELECT
                         (SELECT count(*) FROM asset_relationship WHERE is_active) AS relationship_count,
@@ -1363,12 +1438,16 @@ class IwoooSSecurityAssetControlPlaneService:
                     FROM asset_inventory asset
                     WHERE asset.lifecycle_state = 'active'
                     """
+                ),
+                result_mode="one",
+                default={},
             )
-            )
+            source_health.append(health)
-            relationship_row = relationship_result.one()
 
-            compliance_result = await db.execute(
+            compliance_rows, health = await _read_public_source(
-                _sql(
+                db,
+                source_id="security_compliance",
+                statement=_sql(
                     """
                     WITH latest AS (
                         SELECT DISTINCT ON (asset_id, dimension)
@@ -1381,12 +1460,16 @@ class IwoooSSecurityAssetControlPlaneService:
                     GROUP BY dimension, status
                     ORDER BY dimension, status
                     """
+                ),
+                result_mode="all",
+                default=[],
             )
-            )
+            source_health.append(health)
-            compliance_rows = compliance_result.fetchall()
 
-            automation_result = await db.execute(
+            automation_rows, health = await _read_public_source(
-                _sql(
+                db,
+                source_id="automation_operations",
+                statement=_sql(
                     """
                     SELECT operation_type, status, count(*) AS cnt
                     FROM automation_operation_log
@@ -1394,12 +1477,16 @@ class IwoooSSecurityAssetControlPlaneService:
                     GROUP BY operation_type, status
                     ORDER BY operation_type, status
                     """
+                ),
+                result_mode="all",
+                default=[],
             )
-            )
+            source_health.append(health)
-            automation_rows = automation_result.fetchall()
 
-            ai_trace_result = await db.execute(
+            ai_trace_row, health = await _read_public_source(
-                _sql(
+                db,
+                source_id="ai_decision_trace",
+                statement=_sql(
                     """
                     SELECT
                         count(*) AS trace_count,
@@ -1408,12 +1495,16 @@ class IwoooSSecurityAssetControlPlaneService:
                     JOIN automation_operation_log operation ON operation.op_id = trace.op_id
                     WHERE operation.created_at >= NOW() - INTERVAL '24 hours'
                     """
+                ),
+                result_mode="one",
+                default={},
             )
-            )
+            source_health.append(health)
-            ai_trace_row = ai_trace_result.one()
 
-            siem_result = await db.execute(
+            siem_row, health = await _read_public_source(
-                _sql(
+                db,
+                source_id="siem_runtime",
+                statement=_sql(
                     """
                     SELECT
                         (SELECT count(*) FROM alert_rule_catalog) AS rule_total,
@@ -1454,12 +1545,16 @@ class IwoooSSecurityAssetControlPlaneService:
                            AND resolved_at >= NOW() - INTERVAL '24 hours')
                             AS mttr_minutes_24h
                     """
+                ),
+                result_mode="one",
+                default={},
             )
-            )
+            source_health.append(health)
-            siem_row = siem_result.one()
 
-            audit_result = await db.execute(
+            audit_row, health = await _read_public_source(
-                _sql(
+                db,
+                source_id="audit_runtime",
+                statement=_sql(
                     """
                     SELECT
                         (SELECT count(*) FROM audit_logs
@@ -1486,9 +1581,11 @@ class IwoooSSecurityAssetControlPlaneService:
                            AND ai_analysis IS NOT NULL)
                             AS asset_change_ai_analysis_count_24h
                     """
+                ),
+                result_mode="one",
+                default={},
             )
-            )
+            source_health.append(health)
-            audit_row = audit_result.one()
 
         return build_iwooos_security_asset_control_plane(
             discovery_row=discovery_row,
@@ -1500,6 +1597,7 @@ class IwoooSSecurityAssetControlPlaneService:
             ai_trace_row=ai_trace_row,
             siem_row=siem_row,
             audit_row=audit_row,
+            source_health=source_health,
         )
 
 

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"