fix(agents): keep alert log readback green under pool pressure

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(