From f8b6b88866cac3e6c00c738772d65180c28e4071 Mon Sep 17 00:00:00 2001 From: ogt Date: Wed, 15 Jul 2026 09:28:01 +0800 Subject: [PATCH] fix(automation): bound asset reconciliation readback --- .../asset_capability_reconciliation_job.py | 26 +++- .../ai_automation_asset_capability_matrix.py | 3 +- ...t_ai_automation_asset_capability_matrix.py | 116 ++++++++++++++++++ 3 files changed, 143 insertions(+), 2 deletions(-) diff --git a/apps/api/src/jobs/asset_capability_reconciliation_job.py b/apps/api/src/jobs/asset_capability_reconciliation_job.py index 8073fbed3..c13545927 100644 --- a/apps/api/src/jobs/asset_capability_reconciliation_job.py +++ b/apps/api/src/jobs/asset_capability_reconciliation_job.py @@ -34,7 +34,12 @@ logger = structlog.get_logger(__name__) _FIRST_DELAY_SECONDS = 90 _LOOP_BACKOFF_SECONDS = 30 * 60 _MAX_RECONCILIATION_CANDIDATES = 20_000 -_RETRYABLE_RESULT_STATUSES = {"blocked_safe_no_write", "degraded_no_write"} +_RECONCILIATION_LIVE_READBACK_TIMEOUT_SECONDS = 30.0 +_RETRYABLE_RESULT_STATUSES = { + "blocked_safe_no_write", + "degraded_cache_refresh", + "degraded_no_write", +} _DAILY_TRIGGER_HOUR_TAIPEI = 3 _DAILY_TRIGGER_MINUTE_TAIPEI = 30 _TAIPEI = ZoneInfo("Asia/Taipei") @@ -232,6 +237,7 @@ async def reconcile_once( started = time.monotonic() matrix = await build_asset_capability_matrix_with_live_readback( project_id=project_id, + timeout_seconds=_RECONCILIATION_LIVE_READBACK_TIMEOUT_SECONDS, include_live_only_assets=True, ) if matrix.get("live_readback_status") != "ready": @@ -264,6 +270,24 @@ async def reconcile_once( candidates, project_id=project_id, ) + public_matrix = await build_asset_capability_matrix_with_live_readback( + project_id=project_id, + timeout_seconds=_RECONCILIATION_LIVE_READBACK_TIMEOUT_SECONDS, + include_live_only_assets=False, + bypass_public_cache=True, + ) + public_cache_refreshed = bool( + public_matrix.get("live_readback_status") == "ready" + and public_matrix.get("closed") is True + ) + receipt["public_cache_refreshed"] = public_cache_refreshed + if not public_cache_refreshed: + receipt["status"] = "degraded_cache_refresh" + logger.warning( + "asset_capability_reconciliation_public_cache_refresh_degraded", + live_readback_status=public_matrix.get("live_readback_status"), + closed=public_matrix.get("closed"), + ) receipt["duration_ms"] = int((time.monotonic() - started) * 1000) logger.info( "asset_capability_reconciliation_completed", 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 2f35dba01..30ab3ac40 100644 --- a/apps/api/src/services/ai_automation_asset_capability_matrix.py +++ b/apps/api/src/services/ai_automation_asset_capability_matrix.py @@ -1314,6 +1314,7 @@ async def build_asset_capability_matrix_with_live_readback( repo_root: Path | None = None, timeout_seconds: float = LIVE_READBACK_TIMEOUT_SECONDS, include_live_only_assets: bool = False, + bypass_public_cache: bool = False, ) -> dict[str, Any]: """Build a bounded matrix from live DB truth, failing closed to declared scope.""" @@ -1324,7 +1325,7 @@ async def build_asset_capability_matrix_with_live_readback( 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: + if use_public_cache and not bypass_public_cache: cached_payload = await _read_public_matrix_cache( project_id, namespace=_PUBLIC_MATRIX_FRESH_CACHE_NAMESPACE, 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 ad2680394..ac817d4c5 100644 --- a/apps/api/tests/test_ai_automation_asset_capability_matrix.py +++ b/apps/api/tests/test_ai_automation_asset_capability_matrix.py @@ -389,6 +389,91 @@ def test_reconciliation_persistence_batches_candidate_queries_and_inserts() -> N assert "AND output ->> 'closed' = 'true'" in source +def test_reconcile_once_uses_bounded_worker_live_readback_timeout( + monkeypatch: Any, +) -> None: + captured: dict[str, object] = {} + + async def _fake_live_readback(**kwargs: object) -> dict[str, object]: + captured.update(kwargs) + return {"live_readback_status": "degraded"} + + monkeypatch.setattr( + reconciliation_job, + "build_asset_capability_matrix_with_live_readback", + _fake_live_readback, + ) + + result = asyncio.run( + reconciliation_job.reconcile_once( + triggered_by="controlled_replay", + project_id="awoooi", + ) + ) + + assert captured == { + "project_id": "awoooi", + "timeout_seconds": 30.0, + "include_live_only_assets": True, + } + assert result["status"] == "degraded_no_write" + + +def test_reconcile_once_publishes_closed_public_matrix_cache( + monkeypatch: Any, +) -> None: + live_readback_calls: list[dict[str, object]] = [] + + async def _fake_live_readback(**kwargs: object) -> dict[str, object]: + live_readback_calls.append(dict(kwargs)) + if kwargs["include_live_only_assets"] is True: + return {"live_readback_status": "ready"} + return {"live_readback_status": "ready", "closed": True} + + async def _fake_persist( + _matrix: object, + _candidates: object, + *, + project_id: str, + ) -> dict[str, object]: + assert project_id == "awoooi" + return {"closed": True, "candidate_count": 0} + + monkeypatch.setattr( + reconciliation_job, + "build_asset_capability_matrix_with_live_readback", + _fake_live_readback, + ) + monkeypatch.setattr( + reconciliation_job, + "build_reconciliation_candidates", + lambda _matrix: [], + ) + monkeypatch.setattr(reconciliation_job, "_persist_reconciliation", _fake_persist) + + result = asyncio.run( + reconciliation_job.reconcile_once( + triggered_by="controlled_replay", + project_id="awoooi", + ) + ) + + assert live_readback_calls == [ + { + "project_id": "awoooi", + "timeout_seconds": 30.0, + "include_live_only_assets": True, + }, + { + "project_id": "awoooi", + "timeout_seconds": 30.0, + "include_live_only_assets": False, + "bypass_public_cache": True, + }, + ] + assert result["public_cache_refreshed"] is True + + def test_reconciliation_loop_retries_degraded_result_before_daily_wait( monkeypatch: Any, ) -> None: @@ -678,6 +763,37 @@ def test_public_matrix_success_warms_cache_for_next_request( assert live_read_count == 1 +def test_public_matrix_cache_bypass_forces_shared_refresh(monkeypatch: Any) -> None: + stored_payloads: list[dict[str, Any]] = [] + + async def _unexpected_cache_read(*_args: Any, **_kwargs: Any) -> dict[str, Any]: + raise AssertionError("cache bypass must not read a prior projection") + + async def _live_readback( + _project_id: str, + ) -> tuple[dict[str, Any], list[dict[str, Any]], dict[str, Any]]: + return {}, [], {} + + async def _store_cache(_project_id: str, payload: dict[str, Any]) -> None: + stored_payloads.append(payload) + + monkeypatch.setattr( + matrix_service, "_read_public_matrix_cache", _unexpected_cache_read + ) + monkeypatch.setattr(matrix_service, "_load_live_rows", _live_readback) + monkeypatch.setattr(matrix_service, "_store_public_matrix_caches", _store_cache) + + payload = asyncio.run( + build_asset_capability_matrix_with_live_readback( + bypass_public_cache=True, + ) + ) + + assert payload["live_readback_status"] == "ready" + assert len(stored_payloads) == 1 + assert stored_payloads[0]["matrix_fingerprint"] == payload["matrix_fingerprint"] + + def test_public_matrix_timeout_uses_last_ready_cache(monkeypatch: Any) -> None: async def _slow_live_readback( _project_id: str,