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 dda2f292b..27549f312 100644 --- a/apps/api/src/services/ai_automation_asset_capability_matrix.py +++ b/apps/api/src/services/ai_automation_asset_capability_matrix.py @@ -30,8 +30,14 @@ 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 +LIVE_READBACK_TIMEOUT_SECONDS = 7.25 LIVE_READBACK_DB_STATEMENT_TIMEOUT_MILLISECONDS = 6_000 +PUBLIC_MATRIX_FRESH_CACHE_TTL_SECONDS = 45 +PUBLIC_MATRIX_FALLBACK_CACHE_TTL_SECONDS = 15 * 60 +PUBLIC_MATRIX_CACHE_IO_TIMEOUT_SECONDS = 0.2 + +_PUBLIC_MATRIX_FRESH_CACHE_NAMESPACE = "ai_asset_capability_matrix_fresh" +_PUBLIC_MATRIX_FALLBACK_CACHE_NAMESPACE = "ai_asset_capability_matrix_fallback" STAGE_IDS = ( "source_truth", @@ -1174,6 +1180,133 @@ async def _load_live_rows( return latest_run, list(assets_by_id.values()), receipt +def _public_matrix_cache_key(project_id: str) -> dict[str, str]: + return { + "project_id": project_id, + "projection": "declared_assets_only", + "schema_version": SCHEMA_VERSION, + } + + +async def _read_public_matrix_cache( + project_id: str, + *, + namespace: str, + ttl_seconds: int, +) -> dict[str, Any] | None: + from src.core.redis_client import get_redis + from src.services.operator_summary_cache import ( + get_cached_operator_summary, + get_cached_operator_summary_async, + ) + + key_parts = _public_matrix_cache_key(project_id) + try: + get_redis() + except RuntimeError: + return get_cached_operator_summary( + namespace, + key_parts, + ttl_seconds=ttl_seconds, + ) + try: + return await asyncio.wait_for( + get_cached_operator_summary_async( + namespace, + key_parts, + ttl_seconds=ttl_seconds, + ), + timeout=PUBLIC_MATRIX_CACHE_IO_TIMEOUT_SECONDS, + ) + except Exception as exc: + logger.debug( + "asset_capability_matrix_cache_read_failed", + namespace=namespace, + error_type=type(exc).__name__, + ) + return get_cached_operator_summary( + namespace, + key_parts, + ttl_seconds=ttl_seconds, + ) + + +async def _store_public_matrix_caches( + project_id: str, + payload: dict[str, Any], +) -> None: + from src.core.redis_client import get_redis + from src.services.operator_summary_cache import ( + store_operator_summary, + store_operator_summary_async, + ) + + key_parts = _public_matrix_cache_key(project_id) + cache_targets = ( + ( + _PUBLIC_MATRIX_FRESH_CACHE_NAMESPACE, + PUBLIC_MATRIX_FRESH_CACHE_TTL_SECONDS, + ), + ( + _PUBLIC_MATRIX_FALLBACK_CACHE_NAMESPACE, + PUBLIC_MATRIX_FALLBACK_CACHE_TTL_SECONDS, + ), + ) + for namespace, ttl_seconds in cache_targets: + store_operator_summary( + namespace, + key_parts, + payload, + ttl_seconds=ttl_seconds, + ) + + try: + get_redis() + except RuntimeError: + return + + try: + await asyncio.wait_for( + asyncio.gather( + *( + store_operator_summary_async( + namespace, + key_parts, + payload, + ttl_seconds=ttl_seconds, + ) + for namespace, ttl_seconds in cache_targets + ) + ), + timeout=PUBLIC_MATRIX_CACHE_IO_TIMEOUT_SECONDS, + ) + except Exception as exc: + logger.debug( + "asset_capability_matrix_cache_write_failed", + error_type=type(exc).__name__, + ) + + +def _annotate_cached_matrix( + payload: dict[str, Any], + *, + mode: str, + timeout_seconds: float, + error_type: str = "", +) -> dict[str, Any]: + cache_meta = _dict(payload.pop("cache", {})) + payload["live_readback_timeout_seconds"] = timeout_seconds + payload["live_readback_cache"] = { + **cache_meta, + "mode": mode, + "fallback_used": mode == "stale_if_error", + } + if error_type: + payload["live_readback_status"] = "degraded_cached" + payload["live_readback_error_type"] = error_type + return payload + + async def build_asset_capability_matrix_with_live_readback( *, project_id: str = DEFAULT_PROJECT_ID, @@ -1187,8 +1320,23 @@ async def build_asset_capability_matrix_with_live_readback( get_ai_automation_program_scope_classes, ) - scope_classes = get_ai_automation_program_scope_classes() bounded_timeout_seconds = max(0.001, float(timeout_seconds)) + use_public_cache = repo_root is None and not include_live_only_assets + + if use_public_cache: + cached_payload = await _read_public_matrix_cache( + project_id, + namespace=_PUBLIC_MATRIX_FRESH_CACHE_NAMESPACE, + ttl_seconds=PUBLIC_MATRIX_FRESH_CACHE_TTL_SECONDS, + ) + if cached_payload is not None: + return _annotate_cached_matrix( + cached_payload, + mode="fresh", + timeout_seconds=bounded_timeout_seconds, + ) + + scope_classes = get_ai_automation_program_scope_classes() async def _load_and_build() -> dict[str, Any]: latest_run, live_assets, receipt = await _load_live_rows(project_id) @@ -1214,6 +1362,19 @@ async def build_asset_capability_matrix_with_live_readback( error_type=type(exc).__name__, timeout_seconds=bounded_timeout_seconds, ) + if use_public_cache: + cached_payload = await _read_public_matrix_cache( + project_id, + namespace=_PUBLIC_MATRIX_FALLBACK_CACHE_NAMESPACE, + ttl_seconds=PUBLIC_MATRIX_FALLBACK_CACHE_TTL_SECONDS, + ) + if cached_payload is not None: + return _annotate_cached_matrix( + cached_payload, + mode="stale_if_error", + timeout_seconds=bounded_timeout_seconds, + error_type=type(exc).__name__, + ) payload = build_asset_capability_matrix( scope_classes, repo_root=repo_root, @@ -1223,5 +1384,7 @@ async def build_asset_capability_matrix_with_live_readback( payload["live_readback_error_type"] = type(exc).__name__ payload["live_readback_timeout_seconds"] = bounded_timeout_seconds return payload + if use_public_cache: + await _store_public_matrix_caches(project_id, payload) 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 542701c8c..e36e7d95f 100644 --- a/apps/api/tests/test_ai_automation_asset_capability_matrix.py +++ b/apps/api/tests/test_ai_automation_asset_capability_matrix.py @@ -534,3 +534,107 @@ def test_matrix_build_phase_is_inside_the_total_timeout(monkeypatch: Any) -> Non assert payload["live_readback_status"] == "degraded" assert payload["live_readback_error_type"] == "TimeoutError" assert payload["live_readback_timeout_seconds"] == 0.01 + + +def test_public_matrix_returns_fresh_cache_without_live_db_read( + monkeypatch: Any, +) -> None: + async def _cached_read( + _project_id: str, + *, + namespace: str, + ttl_seconds: int, + ) -> dict[str, Any] | None: + assert namespace == matrix_service._PUBLIC_MATRIX_FRESH_CACHE_NAMESPACE + assert ttl_seconds == matrix_service.PUBLIC_MATRIX_FRESH_CACHE_TTL_SECONDS + return { + "live_readback_status": "ready", + "summary": {"live_observed_asset_count": 193}, + "cache": {"status": "hit", "age_seconds": 2.5}, + } + + async def _unexpected_live_read( + _project_id: str, + ) -> tuple[dict[str, Any], list[dict[str, Any]], dict[str, Any]]: + raise AssertionError("fresh cache must bypass the live DB read") + + monkeypatch.setattr(matrix_service, "_read_public_matrix_cache", _cached_read) + monkeypatch.setattr(matrix_service, "_load_live_rows", _unexpected_live_read) + + payload = asyncio.run(build_asset_capability_matrix_with_live_readback()) + + assert payload["live_readback_status"] == "ready" + assert payload["summary"]["live_observed_asset_count"] == 193 + assert payload["live_readback_cache"]["mode"] == "fresh" + assert payload["live_readback_cache"]["fallback_used"] is False + + +def test_public_matrix_success_warms_cache_for_next_request( + monkeypatch: Any, +) -> None: + from src.services.operator_summary_cache import clear_operator_summary_cache + + live_read_count = 0 + + async def _live_readback( + _project_id: str, + ) -> tuple[dict[str, Any], list[dict[str, Any]], dict[str, Any]]: + nonlocal live_read_count + live_read_count += 1 + return {}, [], {} + + clear_operator_summary_cache() + monkeypatch.setattr(matrix_service, "_load_live_rows", _live_readback) + try: + first = asyncio.run(build_asset_capability_matrix_with_live_readback()) + second = asyncio.run(build_asset_capability_matrix_with_live_readback()) + finally: + clear_operator_summary_cache() + + assert first["live_readback_status"] == "ready" + assert second["live_readback_status"] == "ready" + assert second["live_readback_cache"]["mode"] == "fresh" + assert live_read_count == 1 + + +def test_public_matrix_timeout_uses_last_ready_cache(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 {}, [], {} + + async def _cached_read( + _project_id: str, + *, + namespace: str, + ttl_seconds: int, + ) -> dict[str, Any] | None: + if namespace == matrix_service._PUBLIC_MATRIX_FRESH_CACHE_NAMESPACE: + return None + assert namespace == matrix_service._PUBLIC_MATRIX_FALLBACK_CACHE_NAMESPACE + assert ttl_seconds == matrix_service.PUBLIC_MATRIX_FALLBACK_CACHE_TTL_SECONDS + return { + "live_readback_status": "ready", + "summary": { + "live_inventory_asset_count": 9595, + "live_observed_asset_count": 193, + "stage_coverage_percent": 18, + }, + "cache": {"status": "hit", "age_seconds": 51.2}, + } + + monkeypatch.setattr(matrix_service, "_read_public_matrix_cache", _cached_read) + monkeypatch.setattr(matrix_service, "_load_live_rows", _slow_live_readback) + + payload = asyncio.run( + build_asset_capability_matrix_with_live_readback(timeout_seconds=0.01) + ) + + assert payload["live_readback_status"] == "degraded_cached" + assert payload["live_readback_error_type"] == "TimeoutError" + assert payload["summary"]["live_inventory_asset_count"] == 9595 + assert payload["summary"]["live_observed_asset_count"] == 193 + assert payload["summary"]["stage_coverage_percent"] == 18 + assert payload["live_readback_cache"]["mode"] == "stale_if_error" + assert payload["live_readback_cache"]["fallback_used"] is True