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 55caec733..9ad50a656 100644 --- a/apps/api/src/services/ai_agent_autonomous_runtime_control.py +++ b/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, 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 a3c305c43..a17b9da9c 100644 --- a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py +++ b/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",