fix(agent): preserve verified runtime closure readback

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,

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,