From 76da612d372401c5f8fe734ac7d70a55e0c1a9b2 Mon Sep 17 00:00:00 2001 From: ogt Date: Wed, 15 Jul 2026 00:41:34 +0800 Subject: [PATCH] fix(api): bound asset capability live readback --- .../ai_automation_asset_capability_matrix.py | 22 ++- ...t_ai_automation_asset_capability_matrix.py | 134 ++++++++++++++++++ 2 files changed, 152 insertions(+), 4 deletions(-) diff --git a/apps/api/src/services/ai_automation_asset_capability_matrix.py b/apps/api/src/services/ai_automation_asset_capability_matrix.py index 891cd9b41..6df7a1934 100644 --- a/apps/api/src/services/ai_automation_asset_capability_matrix.py +++ b/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 diff --git a/apps/api/tests/test_ai_automation_asset_capability_matrix.py b/apps/api/tests/test_ai_automation_asset_capability_matrix.py index 47b3bd417..cdb6acbbe 100644 --- a/apps/api/tests/test_ai_automation_asset_capability_matrix.py +++ b/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