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,