From 485a423dea8a2e04a0f8b15a0acba5a5c2bae222 Mon Sep 17 00:00:00 2001 From: ogt Date: Sat, 11 Jul 2026 00:44:51 +0800 Subject: [PATCH] fix(agent): preserve verified runtime closure readback --- .../ai_agent_autonomous_runtime_control.py | 75 ++++++++++- ...est_ai_agent_autonomous_runtime_control.py | 127 ++++++++++++++++++ 2 files changed, 198 insertions(+), 4 deletions(-) diff --git a/apps/api/src/services/ai_agent_autonomous_runtime_control.py b/apps/api/src/services/ai_agent_autonomous_runtime_control.py index 01c5a3da8..19472e4a3 100644 --- a/apps/api/src/services/ai_agent_autonomous_runtime_control.py +++ b/apps/api/src/services/ai_agent_autonomous_runtime_control.py @@ -62,6 +62,7 @@ _RUNTIME_RECEIPT_STATEMENT_TIMEOUT_MS = 400 _LOG_CONTROLLED_WRITEBACK_CONSUMER_TIMEOUT_SECONDS = 4.0 _LOG_CONTROLLED_WRITEBACK_CONSUMER_READBACK_CACHE_TTL_SECONDS = 30.0 _RUNTIME_RECEIPT_READBACK_CACHE_TTL_SECONDS = 20.0 +_RUNTIME_RECEIPT_READBACK_STALE_FALLBACK_TTL_SECONDS = 300.0 _AUXILIARY_RUNTIME_RECEIPT_QUERY_NAMES = { "alert_operation_counts", "alertmanager_event_counts", @@ -208,18 +209,61 @@ def _runtime_receipt_readback_cache_get( return None stored_at, readback = cached if time.monotonic() - stored_at > _RUNTIME_RECEIPT_READBACK_CACHE_TTL_SECONDS: - _runtime_receipt_readback_cache.pop(key, None) return None return copy.deepcopy(readback) +def _runtime_receipt_readback_cache_get_stale( + key: tuple[str, int, int], + *, + fallback_reason: str, + live_db_read_status: str, + live_error_type: str | None, +) -> dict[str, Any] | None: + cached = _runtime_receipt_readback_cache.get(key) + if cached is None: + return None + stored_at, readback = cached + cache_age_seconds = max(0.0, time.monotonic() - stored_at) + if cache_age_seconds > _RUNTIME_RECEIPT_READBACK_STALE_FALLBACK_TTL_SECONDS: + _runtime_receipt_readback_cache.pop(key, None) + return None + cached_readback = copy.deepcopy(readback) + cached_readback["cache_fallback_active"] = True + cached_readback["record_quality"] = "cached_verified_runtime_receipt_readback" + cached_readback["runtime_receipt_cache_fallback"] = { + "active": True, + "reason": fallback_reason, + "live_db_read_status": live_db_read_status, + "live_error_type": live_error_type, + "cache_age_seconds": round(cache_age_seconds, 3), + "max_stale_seconds": _RUNTIME_RECEIPT_READBACK_STALE_FALLBACK_TTL_SECONDS, + "source": "in_process_last_verified_db_receipt_readback", + "durable_receipts_reused": True, + "writes_on_read": False, + } + return cached_readback + + +def _runtime_receipt_readback_is_cacheable( + readback: Mapping[str, Any], +) -> bool: + db_read_status = str(readback.get("db_read_status") or "") + if db_read_status == "ok": + return True + partial_query_failures = readback.get("partial_query_failures") + return bool( + db_read_status == "partial" + and isinstance(partial_query_failures, list) + and _has_only_auxiliary_runtime_receipt_failures(partial_query_failures) + ) + + def _runtime_receipt_readback_cache_store( key: tuple[str, int, int], readback: Mapping[str, Any], ) -> None: - fallback = readback.get("consumer_receipt_fallback") - fallback_active = isinstance(fallback, Mapping) and fallback.get("active") is True - if readback.get("db_read_status") == "unavailable" and not fallback_active: + if not _runtime_receipt_readback_is_cacheable(readback): return _runtime_receipt_readback_cache[key] = (time.monotonic(), copy.deepcopy(dict(readback))) @@ -5475,6 +5519,21 @@ async def load_ai_agent_autonomous_runtime_receipt_readback( lookback_hours=normalized_lookback_hours, limit=normalized_limit, ) + if not _runtime_receipt_readback_is_cacheable(readback): + error = readback.get("error") + live_error_type = ( + str(error.get("type") or "") + if isinstance(error, Mapping) + else None + ) + stale_readback = _runtime_receipt_readback_cache_get_stale( + cache_key, + fallback_reason="live_runtime_receipt_readback_degraded", + live_db_read_status=str(readback.get("db_read_status") or "unavailable"), + live_error_type=live_error_type, + ) + if stale_readback is not None: + return stale_readback _runtime_receipt_readback_cache_store(cache_key, readback) return readback except TimeoutError: @@ -5483,6 +5542,14 @@ async def load_ai_agent_autonomous_runtime_receipt_readback( project_id=project_id, timeout_seconds=_RUNTIME_RECEIPT_LOCK_TIMEOUT_SECONDS, ) + stale_readback = _runtime_receipt_readback_cache_get_stale( + cache_key, + fallback_reason="runtime_receipt_readback_lock_timeout", + live_db_read_status="unavailable", + live_error_type="RuntimeReceiptReadbackLockTimeout", + ) + if stale_readback is not None: + return stale_readback return build_runtime_receipt_readback_from_rows( project_id=project_id, lookback_hours=normalized_lookback_hours, diff --git a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py index 8711bcd24..3eb2f6704 100644 --- a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py +++ b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py @@ -781,6 +781,133 @@ async def test_live_runtime_receipt_readback_does_not_cache_pool_unavailable( assert second["db_read_status"] == "ok" +@pytest.mark.asyncio +async def test_live_runtime_receipt_readback_uses_last_verified_cache_when_db_degrades( + monkeypatch, +): + runtime_control_module._clear_runtime_receipt_readback_cache() + calls = 0 + + async def _fake_uncached_loader( + *, + project_id: str, + lookback_hours: int, + limit: int, + ) -> dict: + nonlocal calls + calls += 1 + if calls == 1: + return build_runtime_receipt_readback_from_rows( + project_id=project_id, + lookback_hours=lookback_hours, + db_read_status="ok", + service_log_count_rows=[ + {"status": "sanitized_recent_logs", "total": 3, "recent": 1}, + ], + ) + return build_runtime_receipt_readback_from_rows( + project_id=project_id, + lookback_hours=lookback_hours, + db_read_status="unavailable", + error_type="RuntimeReceiptDbContextTimeout", + log_controlled_writeback_consumer=( + _log_controlled_writeback_consumer_readback() + ), + ) + + monkeypatch.setattr( + runtime_control_module, + "_load_ai_agent_autonomous_runtime_receipt_readback_uncached", + _fake_uncached_loader, + ) + + try: + first = await runtime_control_module.load_ai_agent_autonomous_runtime_receipt_readback() + cache_key = ("awoooi", 24, 20) + stored_at, cached_readback = ( + runtime_control_module._runtime_receipt_readback_cache[cache_key] + ) + runtime_control_module._runtime_receipt_readback_cache[cache_key] = ( + stored_at - 21.0, + cached_readback, + ) + second = await runtime_control_module.load_ai_agent_autonomous_runtime_receipt_readback() + finally: + runtime_control_module._clear_runtime_receipt_readback_cache() + + assert calls == 2 + assert first["db_read_status"] == "ok" + assert second["db_read_status"] == "ok" + assert second["cache_fallback_active"] is True + assert second["record_quality"] == "cached_verified_runtime_receipt_readback" + fallback = second["runtime_receipt_cache_fallback"] + assert fallback["active"] is True + assert fallback["reason"] == "live_runtime_receipt_readback_degraded" + assert fallback["live_db_read_status"] == "unavailable" + assert fallback["live_error_type"] == "RuntimeReceiptDbContextTimeout" + assert fallback["cache_age_seconds"] >= 21.0 + assert fallback["max_stale_seconds"] == 300.0 + assert fallback["source"] == "in_process_last_verified_db_receipt_readback" + assert fallback["durable_receipts_reused"] is True + assert fallback["writes_on_read"] is False + + +@pytest.mark.asyncio +async def test_live_runtime_receipt_lock_timeout_uses_last_verified_cache( + monkeypatch, +): + runtime_control_module._clear_runtime_receipt_readback_cache() + + async def _fake_uncached_loader( + *, + project_id: str, + lookback_hours: int, + limit: int, + ) -> dict: + return build_runtime_receipt_readback_from_rows( + project_id=project_id, + lookback_hours=lookback_hours, + db_read_status="ok", + ) + + monkeypatch.setattr( + runtime_control_module, + "_load_ai_agent_autonomous_runtime_receipt_readback_uncached", + _fake_uncached_loader, + ) + first = await runtime_control_module.load_ai_agent_autonomous_runtime_receipt_readback() + assert first["db_read_status"] == "ok" + + monkeypatch.setattr( + runtime_control_module, + "_RUNTIME_RECEIPT_READBACK_CACHE_TTL_SECONDS", + -1.0, + ) + lock = asyncio.Lock() + await lock.acquire() + monkeypatch.setattr(runtime_control_module, "_runtime_receipt_readback_lock", lock) + monkeypatch.setattr( + runtime_control_module, + "_RUNTIME_RECEIPT_LOCK_TIMEOUT_SECONDS", + 0.01, + ) + + try: + second = await runtime_control_module.load_ai_agent_autonomous_runtime_receipt_readback() + finally: + lock.release() + runtime_control_module._clear_runtime_receipt_readback_cache() + + assert second["db_read_status"] == "ok" + assert second["cache_fallback_active"] is True + assert second["runtime_receipt_cache_fallback"]["reason"] == ( + "runtime_receipt_readback_lock_timeout" + ) + assert second["runtime_receipt_cache_fallback"]["live_error_type"] == ( + "RuntimeReceiptReadbackLockTimeout" + ) + + @pytest.mark.asyncio async def test_runtime_control_live_readback_timeout_returns_degraded_payload( monkeypatch,