fix(agents): stabilize runtime readback under transient pressure

apps/api/src/services/ai_agent_autonomous_runtime_control.py

@@ -53,6 +53,7 @@ _RUNTIME_RECEIPT_QUERY_BUDGET_SECONDS = 2.0
 _RUNTIME_RECEIPT_SINGLE_QUERY_TIMEOUT_SECONDS = 0.45
 _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
 _AUXILIARY_RUNTIME_RECEIPT_QUERY_NAMES = {
     "alert_operation_counts",
@@ -60,6 +61,7 @@ _AUXILIARY_RUNTIME_RECEIPT_QUERY_NAMES = {
     "grouped_alert_event_counts",
     "db_context_exit",
     "legacy_mcp_counts",
+    "playbook_trust_counts",
 }
 _CONSUMER_RECEIPT_FALLBACK_ERROR_TYPES = {
     "RuntimeReceiptDbContextTimeout",
@@ -102,6 +104,10 @@ _runtime_receipt_readback_cache: dict[
     tuple[str, int, int],
     tuple[float, dict[str, Any]],
 ] = {}
+_log_controlled_writeback_consumer_readback_cache: dict[
+    str,
+    tuple[float, dict[str, Any]],
+] = {}
 _runtime_receipt_readback_lock: asyncio.Lock | None = None
 
 
@@ -213,6 +219,7 @@ def _runtime_receipt_readback_cache_store(
 
 def _clear_runtime_receipt_readback_cache() -> None:
     _runtime_receipt_readback_cache.clear()
+    _log_controlled_writeback_consumer_readback_cache.clear()
 
 
 def _sanitize_latest_rows(
@@ -471,18 +478,33 @@ async def _load_log_controlled_writeback_consumer_readback(
     """Attach LOG consumer bindings without writing KM/RAG/PlayBook/MCP targets."""
 
     try:
-        return await asyncio.wait_for(
+        readback = await asyncio.wait_for(
             load_latest_ai_agent_log_controlled_writeback_consumer_readback(
                 project_id=project_id,
             ),
             timeout=_LOG_CONTROLLED_WRITEBACK_CONSUMER_TIMEOUT_SECONDS,
         )
+        _log_controlled_writeback_consumer_readback_cache_store(
+            project_id=project_id,
+            readback=readback,
+        )
+        return readback
     except Exception as exc:  # pragma: no cover - keeps runtime control API visible
         logger.warning(
             "log_controlled_writeback_consumer_readback_failed",
             project_id=project_id,
             error_type=type(exc).__name__,
         )
+        cached_readback = _log_controlled_writeback_consumer_readback_cache_get(
+            project_id=project_id,
+        )
+        if cached_readback is not None:
+            cached_readback["cache_fallback_active"] = True
+            cached_readback["record_quality"] = "cached_live_consumer_readback"
+            readback_info = cached_readback.get("readback")
+            if isinstance(readback_info, dict):
+                readback_info["cache_fallback_error_type"] = type(exc).__name__
+            return cached_readback
         return _fallback_log_controlled_writeback_consumer_readback(
             error_type=type(exc).__name__,
         )
@@ -602,6 +624,35 @@ def _log_controlled_writeback_consumer_ready(
     )
 
 
+def _log_controlled_writeback_consumer_readback_cache_get(
+    *,
+    project_id: str,
+) -> dict[str, Any] | None:
+    cached = _log_controlled_writeback_consumer_readback_cache.get(project_id)
+    if cached is None:
+        return None
+    stored_at, readback = cached
+    if (
+        time.monotonic() - stored_at
+        > _LOG_CONTROLLED_WRITEBACK_CONSUMER_READBACK_CACHE_TTL_SECONDS
+    ):
+        _log_controlled_writeback_consumer_readback_cache.pop(project_id, None)
+        return None
+    return copy.deepcopy(readback)
+
+
+def _log_controlled_writeback_consumer_readback_cache_store(
+    *,
+    project_id: str,
+    readback: Mapping[str, Any],
+) -> None:
+    if _log_controlled_writeback_consumer_ready(readback):
+        _log_controlled_writeback_consumer_readback_cache[project_id] = (
+            time.monotonic(),
+            copy.deepcopy(dict(readback)),
+        )
+
+
 def _consumer_receipt_fallback_active(
     *,
     db_read_status: str,

apps/api/tests/test_ai_agent_autonomous_runtime_control.py

@@ -532,6 +532,8 @@ async def test_live_runtime_receipt_readback_returns_partial_when_query_budget_e
 async def test_log_controlled_writeback_consumer_timeout_returns_fallback(
     monkeypatch,
 ):
+    runtime_control_module._clear_runtime_receipt_readback_cache()
+
     async def _slow_consumer_readback(*, project_id: str):
         assert project_id == "awoooi"
         await asyncio.sleep(1)
@@ -548,9 +550,12 @@ async def test_log_controlled_writeback_consumer_timeout_returns_fallback(
         _slow_consumer_readback,
     )
 
-    readback = await runtime_control_module._load_log_controlled_writeback_consumer_readback(
-        project_id="awoooi",
-    )
+    try:
+        readback = await runtime_control_module._load_log_controlled_writeback_consumer_readback(
+            project_id="awoooi",
+        )
+    finally:
+        runtime_control_module._clear_runtime_receipt_readback_cache()
 
     assert readback["status"] == "blocked_waiting_controlled_writeback_consumer_receipts"
     assert "log_controlled_writeback_consumer_readback_unavailable" in readback[
@@ -559,6 +564,50 @@ async def test_log_controlled_writeback_consumer_timeout_returns_fallback(
     assert readback["readback"]["error_type"] == "TimeoutError"
 
 
+@pytest.mark.asyncio
+async def test_log_controlled_writeback_consumer_timeout_uses_cached_ready_readback(
+    monkeypatch,
+):
+    runtime_control_module._clear_runtime_receipt_readback_cache()
+    calls = 0
+
+    async def _sometimes_slow_consumer_readback(*, project_id: str):
+        nonlocal calls
+        assert project_id == "awoooi"
+        calls += 1
+        if calls == 1:
+            return _log_controlled_writeback_consumer_readback()
+        await asyncio.sleep(1)
+        return _log_controlled_writeback_consumer_readback()
+
+    monkeypatch.setattr(
+        runtime_control_module,
+        "_LOG_CONTROLLED_WRITEBACK_CONSUMER_TIMEOUT_SECONDS",
+        0.01,
+    )
+    monkeypatch.setattr(
+        runtime_control_module,
+        "load_latest_ai_agent_log_controlled_writeback_consumer_readback",
+        _sometimes_slow_consumer_readback,
+    )
+
+    try:
+        first = await runtime_control_module._load_log_controlled_writeback_consumer_readback(
+            project_id="awoooi",
+        )
+        second = await runtime_control_module._load_log_controlled_writeback_consumer_readback(
+            project_id="awoooi",
+        )
+    finally:
+        runtime_control_module._clear_runtime_receipt_readback_cache()
+
+    assert first["status"] == "controlled_writeback_consumer_readback_ready"
+    assert second["status"] == "controlled_writeback_consumer_readback_ready"
+    assert second["cache_fallback_active"] is True
+    assert second["record_quality"] == "cached_live_consumer_readback"
+    assert second["readback"]["cache_fallback_error_type"] == "TimeoutError"
+
+
 @pytest.mark.asyncio
 async def test_live_runtime_receipt_readback_uses_short_cache_to_protect_db_pool(
     monkeypatch,
@@ -1676,6 +1725,10 @@ def test_runtime_receipt_auxiliary_alert_query_partial_does_not_block_runtime_tr
         ],
         partial_query_failures=[
             {"query_name": "legacy_mcp_counts", "error_type": "DBAPIError"},
+            {
+                "query_name": "playbook_trust_counts",
+                "error_type": "RuntimeReceiptQueryBudgetExceeded",
+            },
             {"query_name": "alert_operation_counts", "error_type": "DBAPIError"},
             {
                 "query_name": "alertmanager_event_counts",