fix(api): index asset capability projection

apps/api/src/jobs/asset_capability_reconciliation_job.py

@@ -216,7 +216,8 @@ async def reconcile_once(
 
     started = time.monotonic()
     matrix = await build_asset_capability_matrix_with_live_readback(
-        project_id=project_id
+        project_id=project_id,
+        include_live_only_assets=True,
     )
     if matrix.get("live_readback_status") != "ready":
         logger.warning(

apps/api/src/services/ai_automation_asset_capability_matrix.py

@@ -12,7 +12,7 @@ import asyncio
 import hashlib
 import json
 import re
-from collections import Counter
+from collections import Counter, defaultdict
 from collections.abc import Iterable, Mapping, Sequence
 from datetime import UTC, datetime
 from pathlib import Path
@@ -31,6 +31,7 @@ RECONCILIATION_WORK_ITEM_OPERATION = "asset_capability_reconciliation_work_item_
 FRESHNESS_SLO_SECONDS = 2 * 60 * 60
 RECONCILIATION_FRESHNESS_SLO_SECONDS = 36 * 60 * 60
 LIVE_READBACK_TIMEOUT_SECONDS = 8.0
+LIVE_READBACK_DB_STATEMENT_TIMEOUT_MILLISECONDS = 6_000
 
 STAGE_IDS = (
     "source_truth",
@@ -354,25 +355,67 @@ def build_declared_asset_catalog(
     return catalog
 
 
-def _matching_score(declared: Mapping[str, Any], live: Mapping[str, Any]) -> int:
-    declared_values = (
-        declared.get("declared_asset_id"),
-        declared.get("display_name"),
-        declared.get("canonical_id"),
+def _identity_components(values: Iterable[Any]) -> tuple[set[str], set[str]]:
+    present_values = [value for value in values if value]
+    slugs = {_slug(value) for value in present_values}
+    tokens = set().union(*(_identity_tokens(value) for value in present_values))
+    return slugs, tokens
+
+
+def _build_live_identity_indexes(
+    live_rows: Sequence[Mapping[str, Any]],
+) -> tuple[dict[str, list[int]], dict[str, list[int]]]:
+    slug_index: defaultdict[str, list[int]] = defaultdict(list)
+    token_index: defaultdict[str, list[int]] = defaultdict(list)
+    for index, live in enumerate(live_rows):
+        slugs, tokens = _identity_components(
+            (live.get("name"), live.get("asset_key"), live.get("source_repo"))
+        )
+        for slug in slugs:
+            slug_index[slug].append(index)
+        for token in tokens:
+            token_index[token].append(index)
+    return dict(slug_index), dict(token_index)
+
+
+def _best_live_match_index(
+    declared: Mapping[str, Any],
+    *,
+    slug_index: Mapping[str, Sequence[int]],
+    token_index: Mapping[str, Sequence[int]],
+    used_live_indexes: set[int],
+) -> int | None:
+    declared_slugs, declared_tokens = _identity_components(
+        (
+            declared.get("declared_asset_id"),
+            declared.get("display_name"),
+            declared.get("canonical_id"),
+        )
     )
-    live_values = (live.get("name"), live.get("asset_key"), live.get("source_repo"))
-    declared_slugs = {_slug(value) for value in declared_values if value}
-    live_slugs = {_slug(value) for value in live_values if value}
-    if declared_slugs & live_slugs:
-        return 100
-    declared_tokens = set().union(
-        *(_identity_tokens(value) for value in declared_values)
+    exact_candidates = {
+        index
+        for slug in declared_slugs
+        for index in slug_index.get(slug, ())
+        if index not in used_live_indexes
+    }
+    if exact_candidates:
+        return min(exact_candidates)
+
+    overlap_counts: Counter[int] = Counter()
+    for token in declared_tokens:
+        overlap_counts.update(
+            index
+            for index in token_index.get(token, ())
+            if index not in used_live_indexes
+        )
+    if not overlap_counts:
+        return None
+
+    # Scores cap at six overlapping tokens; original live order breaks ties.
+    return min(
+        overlap_counts,
+        key=lambda index: (-min(overlap_counts[index], 6), index),
     )
-    live_tokens = set().union(*(_identity_tokens(value) for value in live_values))
-    overlap = declared_tokens & live_tokens
-    if overlap:
-        return min(80, 20 + (len(overlap) * 10))
-    return 0
 
 
 def _is_fresh(value: Any, *, now: datetime, slo_seconds: int) -> bool:
@@ -601,6 +644,21 @@ def _receipt_controls(
     }
 
 
+def _matrix_fingerprint_row(asset: Mapping[str, Any]) -> dict[str, Any]:
+    return {
+        "canonical_id": asset["canonical_id"],
+        "declared": asset["declared"],
+        "runtime_identity": _dict(asset.get("runtime_identity")).get(
+            "asset_key_fingerprint"
+        ),
+        "stages": {
+            stage_id: _dict(_dict(asset.get("stages")).get(stage_id)).get("status")
+            for stage_id in STAGE_IDS
+        },
+        "change_types": asset.get("change_types"),
+    }
+
+
 def build_asset_capability_matrix(
     scope_classes: Sequence[Mapping[str, Any]],
     *,
@@ -610,6 +668,7 @@ def build_asset_capability_matrix(
     repo_root: Path | None = None,
     generated_at: datetime | None = None,
     live_readback_status: str = "ready",
+    include_live_only_assets: bool = True,
 ) -> dict[str, Any]:
     """Build a public-safe per-asset matrix from declared and live truth."""
 
@@ -626,19 +685,20 @@ def build_asset_capability_matrix(
         now=now,
         slo_seconds=FRESHNESS_SLO_SECONDS,
     )
-    used_live_ids: set[int] = set()
+    slug_index, token_index = _build_live_identity_indexes(live_rows)
+    used_live_indexes: set[int] = set()
     assets: list[dict[str, Any]] = []
 
     for declared in declared_assets:
-        candidates = [
-            (live, _matching_score(declared, live))
-            for live in live_rows
-            if int(live.get("asset_id") or 0) not in used_live_ids
-        ]
-        live, score = max(candidates, key=lambda item: item[1], default=(None, 0))
-        matched_live = live if score >= 30 else None
-        if matched_live is not None:
-            used_live_ids.add(int(matched_live.get("asset_id") or 0))
+        matched_index = _best_live_match_index(
+            declared,
+            slug_index=slug_index,
+            token_index=token_index,
+            used_live_indexes=used_live_indexes,
+        )
+        matched_live = live_rows[matched_index] if matched_index is not None else None
+        if matched_index is not None:
+            used_live_indexes.add(matched_index)
         stages = _build_stages(
             declared,
             matched_live,
@@ -669,52 +729,41 @@ def build_asset_capability_matrix(
             }
         )
 
-    for live in live_rows:
-        live_id = int(live.get("asset_id") or 0)
-        if live_id in used_live_ids:
-            continue
+    fingerprint_rows = [_matrix_fingerprint_row(asset) for asset in assets]
+    unmatched_live_indexes = [
+        index for index in range(len(live_rows)) if index not in used_live_indexes
+    ]
+    for live_index in unmatched_live_indexes:
+        live = live_rows[live_index]
         discovered = _live_only_asset(live, scope_labels)
         stages = _build_stages(discovered, live, now=now, run_fresh=run_fresh)
-        assets.append(
-            {
-                **discovered,
-                "runtime_identity": _public_runtime_identity(
-                    str(discovered["canonical_id"]),
-                    live,
-                ),
-                "stages": stages,
-                "missing_stage_ids": [
-                    stage_id
-                    for stage_id, stage in stages.items()
-                    if stage.get("status") == "missing"
-                ],
-                "stale_stage_ids": [
-                    stage_id
-                    for stage_id, stage in stages.items()
-                    if stage.get("status") == "stale"
-                ],
-                "change_types": list(
-                    dict.fromkeys(["asset_added", *_strings(live.get("change_types"))])
-                ),
-            }
-        )
+        live_only_asset = {
+            **discovered,
+            "runtime_identity": _public_runtime_identity(
+                str(discovered["canonical_id"]),
+                live,
+            ),
+            "stages": stages,
+            "missing_stage_ids": [
+                stage_id
+                for stage_id, stage in stages.items()
+                if stage.get("status") == "missing"
+            ],
+            "stale_stage_ids": [
+                stage_id
+                for stage_id, stage in stages.items()
+                if stage.get("status") == "stale"
+            ],
+            "change_types": list(
+                dict.fromkeys(["asset_added", *_strings(live.get("change_types"))])
+            ),
+        }
+        fingerprint_rows.append(_matrix_fingerprint_row(live_only_asset))
+        if include_live_only_assets:
+            assets.append(live_only_asset)
 
     assets.sort(key=lambda row: (str(row["scope_id"]), str(row["canonical_id"])))
-    fingerprint_rows = [
-        {
-            "canonical_id": asset["canonical_id"],
-            "declared": asset["declared"],
-            "runtime_identity": _dict(asset.get("runtime_identity")).get(
-                "asset_key_fingerprint"
-            ),
-            "stages": {
-                stage_id: _dict(_dict(asset.get("stages")).get(stage_id)).get("status")
-                for stage_id in STAGE_IDS
-            },
-            "change_types": asset.get("change_types"),
-        }
-        for asset in assets
-    ]
+    fingerprint_rows.sort(key=lambda row: str(row["canonical_id"]))
     matrix_fingerprint = _fingerprint(fingerprint_rows)
     canonical_ids = [str(asset.get("canonical_id") or "") for asset in assets]
     expected_scope_counts = {
@@ -819,6 +868,12 @@ def build_asset_capability_matrix(
             "declared_asset_count": len(declared_assets),
             "live_inventory_asset_count": len(live_rows),
             "matrix_asset_count": len(assets),
+            "reconciled_asset_count": len(declared_assets)
+            + len(unmatched_live_indexes),
+            "live_only_assets_embedded": include_live_only_assets,
+            "embedded_live_only_asset_count": (
+                len(unmatched_live_indexes) if include_live_only_assets else 0
+            ),
             "live_observed_asset_count": live_observed_count,
             "declared_only_asset_count": len(declared_assets)
             - sum(
@@ -828,9 +883,7 @@ def build_asset_capability_matrix(
                 and _dict(asset.get("runtime_identity")).get("observation_status")
                 == "live_observed"
             ),
-            "discovered_not_declared_count": sum(
-                1 for asset in assets if asset.get("declared") is False
-            ),
+            "discovered_not_declared_count": len(unmatched_live_indexes),
             "gap_asset_count": len(gap_assets),
             "covered_stage_cell_count": stage_counts["covered"],
             "missing_stage_cell_count": stage_counts["missing"],
@@ -972,6 +1025,14 @@ async def _load_live_rows(
     from src.db.base import get_db_context
 
     async with get_db_context(project_id) as db:
+        await db.execute(
+            text("SELECT set_config('statement_timeout', :statement_timeout, true)"),
+            {
+                "statement_timeout": (
+                    f"{LIVE_READBACK_DB_STATEMENT_TIMEOUT_MILLISECONDS}ms"
+                )
+            },
+        )
         latest_result = await db.execute(
             text(
                 """
@@ -1118,8 +1179,9 @@ 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,
+    include_live_only_assets: bool = False,
 ) -> dict[str, Any]:
-    """Build the matrix from live DB truth, failing closed to declared scope."""
+    """Build a bounded matrix from live DB truth, failing closed to declared scope."""
 
     from src.services.awoooi_priority_work_order_readback import (
         get_ai_automation_program_scope_classes,
@@ -1127,9 +1189,23 @@ 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))
+
+    async def _load_and_build() -> dict[str, Any]:
+        latest_run, live_assets, receipt = await _load_live_rows(project_id)
+        return await asyncio.to_thread(
+            build_asset_capability_matrix,
+            scope_classes,
+            live_assets=live_assets,
+            latest_run=latest_run,
+            reconciliation_receipt=receipt,
+            repo_root=repo_root,
+            live_readback_status="ready",
+            include_live_only_assets=include_live_only_assets,
+        )
+
     try:
-        latest_run, live_assets, receipt = await asyncio.wait_for(
-            _load_live_rows(project_id),
+        payload = await asyncio.wait_for(
+            _load_and_build(),
             timeout=bounded_timeout_seconds,
         )
     except Exception as exc:
@@ -1142,17 +1218,10 @@ async def build_asset_capability_matrix_with_live_readback(
             scope_classes,
             repo_root=repo_root,
             live_readback_status="degraded",
+            include_live_only_assets=include_live_only_assets,
         )
         payload["live_readback_error_type"] = type(exc).__name__
         payload["live_readback_timeout_seconds"] = bounded_timeout_seconds
         return payload
-    payload = build_asset_capability_matrix(
-        scope_classes,
-        live_assets=live_assets,
-        latest_run=latest_run,
-        reconciliation_receipt=receipt,
-        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

@@ -102,6 +102,93 @@ def test_declared_catalog_expands_to_193_unique_assets() -> None:
     assert matrix["operation_boundaries"]["raw_asset_key_exposed"] is False
 
 
+def test_public_projection_counts_live_only_assets_without_embedding_them() -> None:
+    now = datetime(2026, 7, 15, 1, 0, tzinfo=UTC)
+    live_assets = [
+        {
+            "asset_id": index + 1,
+            "asset_key": f"zxq{index:x}v",
+            "asset_type": "host",
+            "name": f"zxq{index:x}v",
+            "last_seen_at": now,
+            "coverage": {},
+            "compliance": {},
+            "change_types": [],
+        }
+        for index in range(1_000)
+    ]
+
+    public_matrix = build_asset_capability_matrix(
+        _single_host_scope(),
+        live_assets=live_assets,
+        repo_root=_REPO_ROOT,
+        generated_at=now,
+        include_live_only_assets=False,
+    )
+    reconciliation_matrix = build_asset_capability_matrix(
+        _single_host_scope(),
+        live_assets=live_assets,
+        repo_root=_REPO_ROOT,
+        generated_at=now,
+        include_live_only_assets=True,
+    )
+
+    assert len(public_matrix["assets"]) == 1
+    public_summary = public_matrix["summary"]
+    assert public_summary["live_inventory_asset_count"] == 1_000
+    assert public_summary["matrix_asset_count"] == 1
+    assert public_summary["reconciled_asset_count"] == 1_001
+    assert public_summary["live_only_assets_embedded"] is False
+    assert public_summary["embedded_live_only_asset_count"] == 0
+    assert public_summary["discovered_not_declared_count"] == 1_000
+    assert len(reconciliation_matrix["assets"]) == 1_001
+    assert reconciliation_matrix["summary"]["live_only_assets_embedded"] is True
+    assert reconciliation_matrix["summary"]["embedded_live_only_asset_count"] == 1_000
+    assert (
+        public_matrix["matrix_fingerprint"]
+        == (reconciliation_matrix["matrix_fingerprint"])
+    )
+
+
+def test_indexed_matching_preserves_score_cap_and_original_live_order() -> None:
+    now = datetime(2026, 7, 15, 1, 0, tzinfo=UTC)
+    scope = [
+        {
+            "id": "services",
+            "label": "Services",
+            "asset_count": 1,
+            "asset_ids": ["one two three four five six seven"],
+            "inventory_sources": ["service-inventory.json"],
+        }
+    ]
+    live_assets = [
+        {
+            "asset_id": 1,
+            "asset_key": "candidate:first",
+            "asset_type": "container",
+            "name": "one two three four five six alpha",
+            "last_seen_at": now,
+        },
+        {
+            "asset_id": 2,
+            "asset_key": "candidate:second",
+            "asset_type": "container",
+            "name": "one two three four five six seven beta",
+            "last_seen_at": now,
+        },
+    ]
+
+    matrix = build_asset_capability_matrix(
+        scope,
+        live_assets=live_assets,
+        repo_root=_REPO_ROOT,
+        generated_at=now,
+        include_live_only_assets=False,
+    )
+
+    assert matrix["assets"][0]["runtime_identity"]["asset_inventory_id"] == 1
+
+
 def test_live_asset_has_per_stage_covered_status_and_closes_after_receipt() -> None:
     now = datetime(2026, 7, 11, 12, 0, tzinfo=UTC)
     initial = build_asset_capability_matrix(
@@ -318,6 +405,8 @@ class _FakeLiveReadbackDb:
         sql = str(statement)
         bound = params or {}
         self.calls.append((sql, bound))
+        if "set_config('statement_timeout'" in sql:
+            return _FakeResult([])
         if "FROM asset_discovery_run" in sql:
             return _FakeResult(
                 [
@@ -382,6 +471,13 @@ def test_live_change_event_readback_is_scoped_to_selected_asset_ids(
 
     _latest_run, assets, _receipt = asyncio.run(_load_live_rows("awoooi"))
 
+    timeout_sql, timeout_params = next(
+        (sql, params)
+        for sql, params in db.calls
+        if "set_config('statement_timeout'" in sql
+    )
+    assert "statement_timeout" in timeout_sql
+    assert timeout_params == {"statement_timeout": "6000ms"}
     change_sql, change_params = next(
         (sql, params) for sql, params in db.calls if "FROM asset_change_event" in sql
     )
@@ -413,3 +509,28 @@ def test_live_readback_timeout_fails_closed_without_hanging(
     assert payload["live_readback_timeout_seconds"] == 0.01
     assert payload["summary"]["declared_asset_count"] == 193
     assert payload["operation_boundaries"]["host_write_performed"] is False
+
+
+def test_matrix_build_phase_is_inside_the_total_timeout(monkeypatch: Any) -> None:
+    async def _fast_live_readback(
+        _project_id: str,
+    ) -> tuple[dict[str, Any], list[dict[str, Any]], dict[str, Any]]:
+        return {}, [], {}
+
+    async def _slow_to_thread(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
+        await asyncio.sleep(1)
+        return {}
+
+    monkeypatch.setattr(matrix_service, "_load_live_rows", _fast_live_readback)
+    monkeypatch.setattr(matrix_service.asyncio, "to_thread", _slow_to_thread)
+
+    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