From 9f862bd25466c80729d971145990fe42056cee5a Mon Sep 17 00:00:00 2001 From: ogt Date: Fri, 10 Jul 2026 00:11:39 +0800 Subject: [PATCH 1/2] fix(agents): keep alert log readback green under pool pressure --- ...gram_alert_monitoring_coverage_readback.py | 119 +++++++++++++++++- ..._alert_monitoring_coverage_readback_api.py | 77 ++++++++++++ 2 files changed, 195 insertions(+), 1 deletion(-) diff --git a/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py b/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py index 86995bf37..1845a5783 100644 --- a/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py +++ b/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py @@ -19,6 +19,7 @@ from typing import Any from sqlalchemy import text +from src.core.config import settings from src.core.logging import get_logger from src.db.base import get_db_context from src.services.ai_agent_log_controlled_writeback_consumer_readback import ( @@ -36,6 +37,10 @@ from src.services.telegram_alert_learning_context_post_apply_verifier import ( SCHEMA_VERSION = "telegram_alert_monitoring_coverage_readback_v1" DEFAULT_PROJECT_ID = "awoooi" _LIVE_READBACK_TIMEOUT_SECONDS = 3.0 +_RUNTIME_DIRECT_CONNECT_TIMEOUT_SECONDS = 1.5 +_RUNTIME_DIRECT_QUERY_TIMEOUT_SECONDS = 0.75 +_RUNTIME_DIRECT_CLOSE_TIMEOUT_SECONDS = 0.25 +_RUNTIME_DIRECT_STATEMENT_TIMEOUT_MS = 650 _MONITORING_INVENTORY = "monitoring-alerting-observability-inventory.snapshot.json" _MONITORING_POST_INCIDENT_READBACK = "monitoring-post-incident-readback-plan.snapshot.json" _ALERT_LOG_EVENT_TYPES = ( @@ -382,6 +387,10 @@ def build_telegram_alert_monitoring_coverage_readback( "all_verified_ai_alert_context_receipts_reusable_by_ai_agent": ( effective_verifier_ready ), + "km_rag_mcp_playbook_ai_agent_context_ready": effective_verifier_ready, + "km_rag_mcp_playbook_ai_agent_context_source": str( + ai_loop_context["source"] + ), "all_monitoring_inventory_surfaces_have_accepted_live_receipt": ( monitoring_live_gap_count == 0 ), @@ -453,6 +462,10 @@ def build_telegram_alert_monitoring_coverage_readback( "ai_loop_ai_agent_context_receipt_count": _int( ai_loop_context["ai_agent_context_receipt_count"] ), + "km_rag_mcp_playbook_ai_agent_context_ready": effective_verifier_ready, + "km_rag_mcp_playbook_ai_agent_context_source": str( + ai_loop_context["source"] + ), "log_controlled_consumer_readback_ready": bool( consumer_rollups.get("controlled_consumer_readback_ready") ), @@ -675,6 +688,9 @@ async def _load_runtime_log_readback(*, project_id: str) -> dict[str, Any]: timeout=_LIVE_READBACK_TIMEOUT_SECONDS, ) except Exception as exc: # pragma: no cover - live DB pressure + direct_payload = await _load_runtime_log_readback_direct(project_id=project_id) + if direct_payload is not None: + return direct_payload logger.warning( "telegram_alert_monitoring_runtime_log_readback_unavailable", project_id=project_id, @@ -732,16 +748,117 @@ async def _load_runtime_log_readback_once(*, project_id: str) -> dict[str, Any]: ) event_rows = [_dict(row) for row in event_result.mappings().all()] + outbound_row = _dict(outbound_result.mappings().first() or {}) + return _runtime_log_readback_from_rows( + event_rows=event_rows, + outbound_row=outbound_row, + readback_source="db_session", + ) + + +async def _load_runtime_log_readback_direct( + *, + project_id: str, +) -> dict[str, Any] | None: + """Read alert log rollups through a short-lived connection when the pool is busy.""" + + db_url = settings.DATABASE_URL.replace("postgresql+asyncpg://", "postgresql://") + event_types_sql = ", ".join(f"'{event_type}'" for event_type in _ALERT_LOG_EVENT_TYPES) + try: + import asyncpg + + conn = await asyncio.wait_for( + asyncpg.connect(db_url), + timeout=_RUNTIME_DIRECT_CONNECT_TIMEOUT_SECONDS, + ) + try: + await asyncio.wait_for( + conn.execute("SELECT set_config('app.project_id', $1, TRUE)", project_id), + timeout=_RUNTIME_DIRECT_QUERY_TIMEOUT_SECONDS, + ) + await asyncio.wait_for( + conn.execute( + "SET statement_timeout = " + f"'{int(_RUNTIME_DIRECT_STATEMENT_TIMEOUT_MS)}ms'" + ), + timeout=_RUNTIME_DIRECT_QUERY_TIMEOUT_SECONDS, + ) + event_rows = await asyncio.wait_for( + conn.fetch(f""" + SELECT + event_type, + COUNT(*) AS count, + MAX(created_at) AS latest_at + FROM alert_operation_log + WHERE created_at >= NOW() - INTERVAL '7 days' + AND event_type IN ({event_types_sql}) + GROUP BY event_type + """), + timeout=_RUNTIME_DIRECT_QUERY_TIMEOUT_SECONDS, + ) + outbound_row = await asyncio.wait_for( + conn.fetchrow(""" + SELECT + COUNT(*) AS outbound_telegram_total, + COUNT(*) FILTER (WHERE send_status = 'sent') AS outbound_sent_total, + COUNT(*) FILTER ( + WHERE COALESCE(source_envelope::jsonb, '{}'::jsonb) + ? 'ai_automation_alert_card' + ) AS outbound_ai_alert_card_total, + COUNT(*) FILTER ( + WHERE COALESCE(source_envelope::jsonb, '{}'::jsonb) + ? 'source_refs' + ) AS outbound_source_refs_total, + MAX(COALESCE(sent_at, queued_at)) AS latest_outbound_at + FROM awooop_outbound_message + WHERE project_id = $1 + AND channel_type = 'telegram' + AND queued_at >= NOW() - INTERVAL '7 days' + """, project_id), + timeout=_RUNTIME_DIRECT_QUERY_TIMEOUT_SECONDS, + ) + return _runtime_log_readback_from_rows( + event_rows=[dict(row) for row in event_rows], + outbound_row=dict(outbound_row) if outbound_row is not None else {}, + readback_source="direct_connection", + ) + finally: + try: + await asyncio.wait_for( + conn.close(), + timeout=_RUNTIME_DIRECT_CLOSE_TIMEOUT_SECONDS, + ) + except Exception as close_exc: # pragma: no cover - live cleanup + logger.warning( + "telegram_alert_monitoring_runtime_log_direct_close_failed", + project_id=project_id, + error_type=type(close_exc).__name__, + ) + except Exception as exc: # pragma: no cover - live DB pressure + logger.warning( + "telegram_alert_monitoring_runtime_log_direct_readback_failed", + project_id=project_id, + error_type=type(exc).__name__, + ) + return None + + +def _runtime_log_readback_from_rows( + *, + event_rows: list[dict[str, Any]], + outbound_row: dict[str, Any], + readback_source: str, +) -> dict[str, Any]: event_counts = {str(row["event_type"]): _int(row["count"]) for row in event_rows} latest_event_at = max( [str(row.get("latest_at")) for row in event_rows if row.get("latest_at")], default=None, ) - outbound_row = _dict(outbound_result.mappings().first() or {}) total = sum(event_counts.values()) return { "status": "ok", "summary": { + "readback_source": readback_source, "alert_operation_log_total_7d": total, "telegram_sent_event_count_7d": event_counts.get("TELEGRAM_SENT", 0), "telegram_result_sent_event_count_7d": event_counts.get( diff --git a/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py b/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py index 379d994b1..8b203e032 100644 --- a/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py +++ b/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py @@ -216,6 +216,14 @@ def test_telegram_alert_monitoring_coverage_readback_surfaces_live_gaps(): ] is True ) + assert ( + payload["operator_answer"]["km_rag_mcp_playbook_ai_agent_context_ready"] + is True + ) + assert ( + payload["operator_answer"]["km_rag_mcp_playbook_ai_agent_context_source"] + == "post_apply_verifier" + ) assert ( payload["operator_answer"][ "all_monitoring_inventory_surfaces_have_accepted_live_receipt" @@ -351,6 +359,11 @@ def test_telegram_alert_monitoring_coverage_uses_consumer_context_fallback(): assert summary["ai_loop_context_fallback_used"] is True assert summary["ai_loop_context_receipt_count"] == 6 assert summary["ai_loop_ai_agent_context_receipt_count"] == 1 + assert summary["km_rag_mcp_playbook_ai_agent_context_ready"] is True + assert ( + summary["km_rag_mcp_playbook_ai_agent_context_source"] + == "log_controlled_writeback_consumer" + ) assert summary["effective_ai_alert_context_receipt_total"] == 6 assert summary["effective_ai_alert_context_ready_total"] == 6 assert summary["log_controlled_consumer_readback_ready"] is True @@ -362,6 +375,14 @@ def test_telegram_alert_monitoring_coverage_uses_consumer_context_fallback(): ] is True ) + assert ( + payload["operator_answer"]["km_rag_mcp_playbook_ai_agent_context_ready"] + is True + ) + assert ( + payload["operator_answer"]["km_rag_mcp_playbook_ai_agent_context_source"] + == "log_controlled_writeback_consumer" + ) assert ( "awooop_ai_alert_card_delivery_db_readback_unavailable" in payload["active_blockers"] @@ -446,6 +467,16 @@ async def test_telegram_alert_monitoring_live_readbacks_are_bounded( slow_runtime_log, ) + async def no_direct_runtime_log(*, project_id: str): + assert project_id == "awoooi" + return None + + monkeypatch.setattr( + coverage_service, + "_load_runtime_log_readback_direct", + no_direct_runtime_log, + ) + verifier = await coverage_service._load_post_apply_verifier_readback( project_id="awoooi" ) @@ -464,6 +495,52 @@ async def test_telegram_alert_monitoring_live_readbacks_are_bounded( assert runtime_log["event_type_counts_7d"] == {} +@pytest.mark.asyncio +async def test_telegram_alert_monitoring_runtime_log_uses_direct_fallback( + monkeypatch: pytest.MonkeyPatch, +): + async def failed_runtime_log(*, project_id: str): + assert project_id == "awoooi" + raise TimeoutError + + async def direct_runtime_log(*, project_id: str): + assert project_id == "awoooi" + return { + "status": "ok", + "summary": { + "readback_source": "direct_connection", + "alert_operation_log_total_7d": 3, + "telegram_sent_event_count_7d": 1, + "km_converted_event_count_7d": 1, + "playbook_draft_event_count_7d": 1, + "outbound_telegram_total_7d": 1, + }, + "event_type_counts_7d": { + "TELEGRAM_SENT": 1, + "KM_CONVERTED": 1, + "PLAYBOOK_DRAFT_CREATED": 1, + }, + } + + monkeypatch.setattr( + coverage_service, + "_load_runtime_log_readback_once", + failed_runtime_log, + ) + monkeypatch.setattr( + coverage_service, + "_load_runtime_log_readback_direct", + direct_runtime_log, + ) + + runtime_log = await coverage_service._load_runtime_log_readback(project_id="awoooi") + + assert runtime_log["status"] == "ok" + assert runtime_log["summary"]["readback_source"] == "direct_connection" + assert runtime_log["summary"]["alert_operation_log_total_7d"] == 3 + assert runtime_log["event_type_counts_7d"]["TELEGRAM_SENT"] == 1 + + @pytest.mark.asyncio async def test_telegram_alert_monitoring_marks_ai_alert_cards_source_unavailable( monkeypatch: pytest.MonkeyPatch, From a1ec18b9bad909bb62722b0a2c8a54714d609e8c Mon Sep 17 00:00:00 2001 From: ogt Date: Fri, 10 Jul 2026 00:12:51 +0800 Subject: [PATCH 2/2] fix(agents): stabilize runtime readback under transient pressure --- .../ai_agent_autonomous_runtime_control.py | 53 ++++++++++++++++- ...est_ai_agent_autonomous_runtime_control.py | 59 ++++++++++++++++++- 2 files changed, 108 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 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",