fix(api): bound asset capability live readback

apps/api/src/services/ai_automation_asset_capability_matrix.py

@@ -8,6 +8,7 @@ the database boundary.
 
 from __future__ import annotations
 
+import asyncio
 import hashlib
 import json
 import re
@@ -29,6 +30,7 @@ RECONCILIATION_SUMMARY_OPERATION = "asset_capability_reconciliation_completed"
 RECONCILIATION_WORK_ITEM_OPERATION = "asset_capability_reconciliation_work_item_created"
 FRESHNESS_SLO_SECONDS = 2 * 60 * 60
 RECONCILIATION_FRESHNESS_SLO_SECONDS = 36 * 60 * 60
+LIVE_READBACK_TIMEOUT_SECONDS = 8.0
 
 STAGE_IDS = (
     "source_truth",
@@ -1044,6 +1046,7 @@ async def _load_live_rows(
                     "created_at": item.get("coverage_created_at"),
                 }
 
+        asset_ids = list(assets_by_id) or [0]
         compliance_result = await db.execute(
             text(
                 """
@@ -1054,7 +1057,7 @@ async def _load_live_rows(
                 ORDER BY asset_id, dimension, detected_at DESC
                 """
             ),
-            {"asset_ids": list(assets_by_id) or [0]},
+            {"asset_ids": asset_ids},
         )
         for row in compliance_result.mappings().all():
             item = dict(row)
@@ -1072,9 +1075,11 @@ async def _load_live_rows(
                 FROM asset_change_event
                 WHERE detected_at >= NOW() - INTERVAL '36 hours'
                   AND asset_id IS NOT NULL
+                  AND asset_id = ANY(:asset_ids)
                 ORDER BY detected_at DESC
                 """
-            )
+            ),
+            {"asset_ids": asset_ids},
         )
         for row in changes_result.mappings().all():
             item = dict(row)
@@ -1112,6 +1117,7 @@ async def build_asset_capability_matrix_with_live_readback(
     *,
     project_id: str = DEFAULT_PROJECT_ID,
     repo_root: Path | None = None,
+    timeout_seconds: float = LIVE_READBACK_TIMEOUT_SECONDS,
 ) -> dict[str, Any]:
     """Build the matrix from live DB truth, failing closed to declared scope."""
 
@@ -1120,12 +1126,17 @@ async def build_asset_capability_matrix_with_live_readback(
     )
 
     scope_classes = get_ai_automation_program_scope_classes()
+    bounded_timeout_seconds = max(0.001, float(timeout_seconds))
     try:
-        latest_run, live_assets, receipt = await _load_live_rows(project_id)
+        latest_run, live_assets, receipt = await asyncio.wait_for(
+            _load_live_rows(project_id),
+            timeout=bounded_timeout_seconds,
+        )
     except Exception as exc:
         logger.warning(
             "asset_capability_matrix_live_readback_degraded",
             error_type=type(exc).__name__,
+            timeout_seconds=bounded_timeout_seconds,
         )
         payload = build_asset_capability_matrix(
             scope_classes,
@@ -1133,8 +1144,9 @@ async def build_asset_capability_matrix_with_live_readback(
             live_readback_status="degraded",
         )
         payload["live_readback_error_type"] = type(exc).__name__
+        payload["live_readback_timeout_seconds"] = bounded_timeout_seconds
         return payload
-    return build_asset_capability_matrix(
+    payload = build_asset_capability_matrix(
         scope_classes,
         live_assets=live_assets,
         latest_run=latest_run,
@@ -1142,3 +1154,5 @@ async def build_asset_capability_matrix_with_live_readback(
         repo_root=repo_root,
         live_readback_status="ready",
     )
+    payload["live_readback_timeout_seconds"] = bounded_timeout_seconds
+    return payload

apps/api/tests/test_ai_automation_asset_capability_matrix.py

@@ -1,8 +1,11 @@
 from __future__ import annotations
 
+import asyncio
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
+from typing import Any
 
+import src.services.ai_automation_asset_capability_matrix as matrix_service
 from src.jobs.asset_capability_reconciliation_job import (
     build_reconciliation_summary_payload,
     build_reconciliation_work_item_input,
@@ -10,7 +13,9 @@ from src.jobs.asset_capability_reconciliation_job import (
 from src.services.ai_automation_asset_capability_matrix import (
     REQUIRED_CONTROL_IDS,
     STAGE_IDS,
+    _load_live_rows,
     build_asset_capability_matrix,
+    build_asset_capability_matrix_with_live_readback,
     build_reconciliation_candidates,
 )
 from src.services.awoooi_priority_work_order_readback import (
@@ -279,3 +284,132 @@ def test_priority_overlay_projects_aia_p0_006_runtime_controls() -> None:
     assert work_item["runtime_progress"]["required_control_count"] == len(
         REQUIRED_CONTROL_IDS
     )
+
+
+class _FakeMappings:
+    def __init__(self, rows: list[dict[str, Any]]) -> None:
+        self._rows = rows
+
+    def first(self) -> dict[str, Any] | None:
+        return self._rows[0] if self._rows else None
+
+    def all(self) -> list[dict[str, Any]]:
+        return self._rows
+
+
+class _FakeResult:
+    def __init__(self, rows: list[dict[str, Any]]) -> None:
+        self._rows = rows
+
+    def mappings(self) -> _FakeMappings:
+        return _FakeMappings(self._rows)
+
+
+class _FakeLiveReadbackDb:
+    def __init__(self, now: datetime) -> None:
+        self.now = now
+        self.calls: list[tuple[str, dict[str, Any]]] = []
+
+    async def execute(
+        self,
+        statement: Any,
+        params: dict[str, Any] | None = None,
+    ) -> _FakeResult:
+        sql = str(statement)
+        bound = params or {}
+        self.calls.append((sql, bound))
+        if "FROM asset_discovery_run" in sql:
+            return _FakeResult(
+                [
+                    {
+                        "run_id": "00000000-0000-0000-0000-000000000001",
+                        "status": "success",
+                        "ended_at": self.now,
+                    }
+                ]
+            )
+        if "FROM asset_inventory" in sql:
+            return _FakeResult(
+                [
+                    {
+                        "asset_id": 42,
+                        "asset_key": "host:99",
+                        "asset_type": "host",
+                        "environment": "prod",
+                        "namespace": "",
+                        "name": "99",
+                        "owner_team": "platform_sre",
+                        "source_repo": "wooo/awoooi",
+                        "source_commit_sha": "abc123",
+                        "lifecycle_state": "active",
+                        "last_seen_at": self.now,
+                        "dimension": "auto_monitoring",
+                        "coverage_status": "green",
+                        "coverage_created_at": self.now,
+                    }
+                ]
+            )
+        if "FROM asset_compliance_snapshot" in sql:
+            return _FakeResult([])
+        if "FROM asset_change_event" in sql:
+            return _FakeResult([{"asset_id": 42, "change_type": "asset_modified"}])
+        if "FROM automation_operation_log" in sql:
+            return _FakeResult([])
+        raise AssertionError(f"unexpected SQL: {sql}")
+
+
+class _FakeDbContext:
+    def __init__(self, db: _FakeLiveReadbackDb) -> None:
+        self.db = db
+
+    async def __aenter__(self) -> _FakeLiveReadbackDb:
+        return self.db
+
+    async def __aexit__(self, *_args: object) -> None:
+        return None
+
+
+def test_live_change_event_readback_is_scoped_to_selected_asset_ids(
+    monkeypatch: Any,
+) -> None:
+    now = datetime(2026, 7, 15, 1, 0, tzinfo=UTC)
+    db = _FakeLiveReadbackDb(now)
+
+    monkeypatch.setattr(
+        "src.db.base.get_db_context",
+        lambda _project_id: _FakeDbContext(db),
+    )
+
+    _latest_run, assets, _receipt = asyncio.run(_load_live_rows("awoooi"))
+
+    change_sql, change_params = next(
+        (sql, params) for sql, params in db.calls if "FROM asset_change_event" in sql
+    )
+    assert "asset_id = ANY(:asset_ids)" in change_sql
+    assert change_params == {"asset_ids": [42]}
+    assert assets[0]["change_types"] == ["asset_modified"]
+
+
+def test_live_readback_timeout_fails_closed_without_hanging(
+    monkeypatch: Any,
+) -> None:
+    async def _slow_live_readback(
+        _project_id: str,
+    ) -> tuple[dict[str, Any], list[dict[str, Any]], dict[str, Any]]:
+        await asyncio.sleep(1)
+        return {}, [], {}
+
+    monkeypatch.setattr(matrix_service, "_load_live_rows", _slow_live_readback)
+
+    payload = asyncio.run(
+        build_asset_capability_matrix_with_live_readback(
+            repo_root=_REPO_ROOT,
+            timeout_seconds=0.01,
+        )
+    )
+
+    assert payload["live_readback_status"] == "degraded"
+    assert payload["live_readback_error_type"] == "TimeoutError"
+    assert payload["live_readback_timeout_seconds"] == 0.01
+    assert payload["summary"]["declared_asset_count"] == 193
+    assert payload["operation_boundaries"]["host_write_performed"] is False