fix(automation): bound asset reconciliation readback

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",

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,

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,