fix(agents): consume monitoring live receipt metadata

apps/api/src/services/telegram_alert_monitoring_coverage_readback.py

@@ -13,7 +13,7 @@ from __future__ import annotations
 
 import asyncio
 import json
-from collections.abc import Mapping
+from collections.abc import Mapping, Sequence
 from pathlib import Path
 from typing import Any
 
@@ -41,6 +41,8 @@ _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_LIVE_RECEIPT_OPERATION_TYPE = "telegram_monitoring_live_receipt_applied"
+_MONITORING_LIVE_RECEIPT_EXECUTOR_ROUTE = "ai_agent_monitoring_live_receipt_consumer"
 _MONITORING_INVENTORY = "monitoring-alerting-observability-inventory.snapshot.json"
 _MONITORING_POST_INCIDENT_READBACK = "monitoring-post-incident-readback-plan.snapshot.json"
 _ALERT_LOG_EVENT_TYPES = (
@@ -106,9 +108,15 @@ async def load_latest_telegram_alert_monitoring_coverage_readback(
     )
     _require_safe_boundaries(monitoring_inventory, monitoring_readback_plan)
 
-    matrix, consumer_readback, source_contract = await asyncio.gather(
+    (
+        matrix,
+        consumer_readback,
+        monitoring_live_receipt_apply,
+        source_contract,
+    ) = await asyncio.gather(
         _load_telegram_matrix_readback(project_id=project_id),
         _load_log_controlled_writeback_consumer_readback(project_id=project_id),
+        _load_monitoring_live_receipt_apply_readback(project_id=project_id),
         asyncio.to_thread(_inspect_source_contract, repo_root),
     )
     verifier, ai_alert_readback, runtime_log_readback = await asyncio.gather(
@@ -126,6 +134,7 @@ async def load_latest_telegram_alert_monitoring_coverage_readback(
         ai_alert_card_delivery_readback=ai_alert_readback,
         runtime_log_readback=runtime_log_readback,
         log_controlled_writeback_consumer=consumer_readback,
+        monitoring_live_receipt_apply_readback=monitoring_live_receipt_apply,
         source_contract=source_contract,
     )
 
@@ -226,6 +235,185 @@ async def _load_log_controlled_writeback_consumer_readback(
         }
 
 
+async def _load_monitoring_live_receipt_apply_readback(
+    *,
+    project_id: str,
+) -> dict[str, Any]:
+    try:
+        return await asyncio.wait_for(
+            _load_monitoring_live_receipt_apply_readback_once(project_id=project_id),
+            timeout=_LIVE_READBACK_TIMEOUT_SECONDS,
+        )
+    except Exception as exc:  # pragma: no cover - live DB pressure
+        direct_payload = await _load_monitoring_live_receipt_apply_readback_direct(
+            project_id=project_id
+        )
+        if direct_payload is not None:
+            return direct_payload
+        logger.warning(
+            "telegram_alert_monitoring_live_receipt_apply_readback_unavailable",
+            project_id=project_id,
+            error_type=type(exc).__name__,
+        )
+        return {
+            "status": "unavailable",
+            "summary": {
+                "readback_source": "unavailable",
+                "row_count": 0,
+                "ready_row_count": 0,
+                "accepted_surface_count": 0,
+                "error_type": type(exc).__name__,
+            },
+            "rows": [],
+        }
+
+
+async def _load_monitoring_live_receipt_apply_readback_once(
+    *,
+    project_id: str,
+) -> dict[str, Any]:
+    async with get_db_context(project_id) as db:
+        await db.execute(text("SET LOCAL statement_timeout = '5000ms'"), {})
+        result = await db.execute(
+            text("""
+                SELECT
+                    op_id::text AS op_id,
+                    operation_type,
+                    actor,
+                    status,
+                    created_at,
+                    input ->> 'semantic_operation_type' AS semantic_operation_type,
+                    input ->> 'live_receipt_id' AS live_receipt_id,
+                    input ->> 'batch_id' AS batch_id,
+                    input ->> 'domain' AS domain,
+                    input ->> 'raw_payload_included' AS raw_payload_included,
+                    input -> 'target_selector' AS target_selector,
+                    input -> 'tag_hints' AS tag_hints,
+                    output ->> 'monitoring_live_receipt_metadata_recorded'
+                        AS monitoring_live_receipt_metadata_recorded,
+                    output ->> 'target_context_receipt_write_performed'
+                        AS target_context_receipt_write_performed,
+                    output -> 'post_apply_verifier_refs' AS post_apply_verifier_refs
+                FROM automation_operation_log
+                WHERE coalesce(input ->> 'semantic_operation_type', operation_type)
+                    = :operation_type
+                  AND input ->> 'project_id' = :project_id
+                ORDER BY created_at DESC, op_id DESC
+                LIMIT 200
+            """),
+            {
+                "operation_type": _MONITORING_LIVE_RECEIPT_OPERATION_TYPE,
+                "project_id": project_id,
+            },
+        )
+    rows = _result_rows(result)
+    return _monitoring_live_receipt_apply_readback_from_rows(
+        rows=rows,
+        readback_source="db_session",
+    )
+
+
+async def _load_monitoring_live_receipt_apply_readback_direct(
+    *,
+    project_id: str,
+) -> dict[str, Any] | None:
+    db_url = settings.DATABASE_URL.replace("postgresql+asyncpg://", "postgresql://")
+    try:
+        import asyncpg  # type: ignore[import-untyped]
+
+        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,
+            )
+            rows = await asyncio.wait_for(
+                conn.fetch(
+                    """
+                    SELECT
+                        op_id::text AS op_id,
+                        operation_type,
+                        actor,
+                        status,
+                        created_at,
+                        input ->> 'semantic_operation_type' AS semantic_operation_type,
+                        input ->> 'live_receipt_id' AS live_receipt_id,
+                        input ->> 'batch_id' AS batch_id,
+                        input ->> 'domain' AS domain,
+                        input ->> 'raw_payload_included' AS raw_payload_included,
+                        input -> 'target_selector' AS target_selector,
+                        input -> 'tag_hints' AS tag_hints,
+                        output ->> 'monitoring_live_receipt_metadata_recorded'
+                            AS monitoring_live_receipt_metadata_recorded,
+                        output ->> 'target_context_receipt_write_performed'
+                            AS target_context_receipt_write_performed,
+                        output -> 'post_apply_verifier_refs' AS post_apply_verifier_refs
+                    FROM automation_operation_log
+                    WHERE coalesce(input ->> 'semantic_operation_type', operation_type)
+                        = $1
+                      AND input ->> 'project_id' = $2
+                    ORDER BY created_at DESC, op_id DESC
+                    LIMIT 200
+                    """,
+                    _MONITORING_LIVE_RECEIPT_OPERATION_TYPE,
+                    project_id,
+                ),
+                timeout=_RUNTIME_DIRECT_QUERY_TIMEOUT_SECONDS,
+            )
+            return _monitoring_live_receipt_apply_readback_from_rows(
+                rows=[dict(row) for row in rows],
+                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_live_receipt_apply_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_live_receipt_apply_direct_readback_failed",
+            project_id=project_id,
+            error_type=type(exc).__name__,
+        )
+        return None
+
+
+def _monitoring_live_receipt_apply_readback_from_rows(
+    *,
+    rows: list[dict[str, Any]],
+    readback_source: str,
+) -> dict[str, Any]:
+    ready_rows = [row for row in rows if _metadata_live_receipt_row_ready(row)]
+    accepted_surface_ids = _metadata_live_receipt_surface_ids(ready_rows)
+    return {
+        "status": "ok",
+        "summary": {
+            "readback_source": readback_source,
+            "row_count": len(rows),
+            "ready_row_count": len(ready_rows),
+            "accepted_surface_count": len(accepted_surface_ids),
+        },
+        "rows": rows,
+    }
+
+
 def build_telegram_alert_monitoring_coverage_readback(
     *,
     project_id: str,
@@ -237,6 +425,7 @@ def build_telegram_alert_monitoring_coverage_readback(
     runtime_log_readback: Mapping[str, Any],
     log_controlled_writeback_consumer: Mapping[str, Any] | None = None,
     source_contract: Mapping[str, Any],
+    monitoring_live_receipt_apply_readback: Mapping[str, Any] | None = None,
 ) -> dict[str, Any]:
     """Build deterministic coverage from static inventory and live readbacks."""
 
@@ -249,10 +438,30 @@ def build_telegram_alert_monitoring_coverage_readback(
     consumer_readback = _dict(log_controlled_writeback_consumer)
     consumer_rollups = _dict(consumer_readback.get("rollups"))
     consumer_context = _dict(consumer_readback.get("telegram_alert_learning_context"))
+    live_receipt_apply_readback = _dict(monitoring_live_receipt_apply_readback)
+    live_receipt_apply_summary = _dict(live_receipt_apply_readback.get("summary"))
+    monitoring_asset_rollups = _monitoring_asset_rollups(
+        monitoring_inventory,
+        monitoring_readback_plan,
+        metadata_live_receipt_rows=_list_of_dicts(
+            live_receipt_apply_readback.get("rows")
+        ),
+    )
 
-    monitoring_surface_count = _int(inventory_summary.get("surface_count"))
-    live_evidence_received_count = _int(
-        inventory_summary.get("live_evidence_received_count")
+    monitoring_surface_count = max(
+        _int(inventory_summary.get("surface_count")),
+        _int(monitoring_asset_rollups.get("surface_count")),
+    )
+    live_evidence_received_count = max(
+        _int(inventory_summary.get("live_evidence_received_count")),
+        _int(
+            monitoring_asset_rollups.get(
+                "metadata_live_receipt_accepted_surface_count"
+            )
+        ),
+        _int(monitoring_asset_rollups.get("accepted_live_receipt_count"))
+        if monitoring_surface_count <= _int(monitoring_asset_rollups.get("surface_count"))
+        else 0,
     )
     monitoring_live_gap_count = max(
         monitoring_surface_count - live_evidence_received_count,
@@ -325,10 +534,6 @@ def build_telegram_alert_monitoring_coverage_readback(
         ai_alert_ready_total=effective_ai_alert_ready_total,
     )
     ready = not active_blockers
-    monitoring_asset_rollups = _monitoring_asset_rollups(
-        monitoring_inventory,
-        monitoring_readback_plan,
-    )
     alert_receipt_pipeline = _alert_receipt_pipeline(
         monitoring_surface_count=monitoring_surface_count,
         live_evidence_received_count=live_evidence_received_count,
@@ -394,6 +599,10 @@ def build_telegram_alert_monitoring_coverage_readback(
             "all_monitoring_inventory_surfaces_have_accepted_live_receipt": (
                 monitoring_live_gap_count == 0
             ),
+            "metadata_live_receipts_count_as_db_or_log_receipt": (
+                monitoring_live_gap_count == 0
+                and _int(live_receipt_apply_summary.get("ready_row_count")) > 0
+            ),
             "manual_default_terminal_state_allowed": False,
             "critical_break_glass_exception_only": True,
         },
@@ -404,6 +613,21 @@ def build_telegram_alert_monitoring_coverage_readback(
             ),
             "monitoring_live_evidence_received_count": live_evidence_received_count,
             "monitoring_live_receipt_gap_count": monitoring_live_gap_count,
+            "monitoring_metadata_live_receipt_apply_readback_ok": (
+                live_receipt_apply_readback.get("status") == "ok"
+            ),
+            "monitoring_metadata_live_receipt_readback_source": str(
+                live_receipt_apply_summary.get("readback_source") or "unavailable"
+            ),
+            "monitoring_metadata_live_receipt_row_count": _int(
+                live_receipt_apply_summary.get("row_count")
+            ),
+            "monitoring_metadata_live_receipt_ready_row_count": _int(
+                live_receipt_apply_summary.get("ready_row_count")
+            ),
+            "monitoring_metadata_live_receipt_accepted_surface_count": _int(
+                live_receipt_apply_summary.get("accepted_surface_count")
+            ),
             "monitoring_readback_candidate_count": _int(
                 readback_summary.get("readback_candidate_count")
             ),
@@ -530,6 +754,9 @@ def build_telegram_alert_monitoring_coverage_readback(
             "metadata_read_performed": True,
             "runtime_db_read_performed": runtime_db_ok,
             "consumer_context_read_performed": bool(consumer_readback),
+            "metadata_live_receipt_apply_read_performed": (
+                live_receipt_apply_readback.get("status") == "ok"
+            ),
             "telegram_send_performed": False,
             "bot_api_call_performed": False,
             "gateway_queue_write_performed": False,
@@ -765,7 +992,7 @@ async def _load_runtime_log_readback_direct(
     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
+        import asyncpg  # type: ignore[import-untyped]
 
         conn = await asyncio.wait_for(
             asyncpg.connect(db_url),
@@ -909,17 +1136,35 @@ def _required_tag_dimensions() -> list[dict[str, str]]:
 def _monitoring_asset_rollups(
     monitoring_inventory: Mapping[str, Any],
     monitoring_readback_plan: Mapping[str, Any],
+    *,
+    metadata_live_receipt_rows: list[dict[str, Any]] | None = None,
 ) -> dict[str, Any]:
     surfaces = _list_of_dicts(monitoring_inventory.get("observability_surfaces"))
     readback_candidates = _list_of_dicts(
         monitoring_readback_plan.get("readback_candidates")
     )
     inventory_source = readback_candidates or surfaces
+    metadata_receipt_surface_ids = _metadata_live_receipt_surface_ids(
+        [
+            row
+            for row in (metadata_live_receipt_rows or [])
+            if _metadata_live_receipt_row_ready(row)
+        ]
+    )
+    metadata_receipt_batches = _unique(
+        [
+            str(row.get("batch_id") or "")
+            for row in (metadata_live_receipt_rows or [])
+            if _metadata_live_receipt_row_ready(row)
+            and str(row.get("batch_id") or "")
+        ]
+    )
     accepted_live_receipts = sum(
         1
         for item in inventory_source
         if bool(item.get("post_incident_readback_accepted"))
         or bool(item.get("live_evidence_received"))
+        or str(item.get("surface_id") or "") in metadata_receipt_surface_ids
     )
     live_receipt_gap_samples = [
         _monitoring_gap_sample(item)
@@ -927,12 +1172,21 @@ def _monitoring_asset_rollups(
         if bool(item.get("requires_live_evidence"))
         and not bool(item.get("post_incident_readback_accepted"))
         and not bool(item.get("live_evidence_received"))
+        and str(item.get("surface_id") or "") not in metadata_receipt_surface_ids
     ][:12]
-    live_receipt_work_batches = _monitoring_live_receipt_work_batches(inventory_source)
+    live_receipt_work_batches = _monitoring_live_receipt_work_batches(
+        inventory_source,
+        accepted_surface_ids=metadata_receipt_surface_ids,
+    )
     return {
         "surface_count": len(surfaces) or len(inventory_source),
         "readback_candidate_count": len(readback_candidates),
         "accepted_live_receipt_count": accepted_live_receipts,
+        "metadata_live_receipt_accepted_surface_count": len(
+            metadata_receipt_surface_ids
+        ),
+        "metadata_live_receipt_batch_count": len(metadata_receipt_batches),
+        "metadata_live_receipt_batch_ids": metadata_receipt_batches,
         "live_receipt_gap_count": max(
             len(inventory_source) - accepted_live_receipts,
             0,
@@ -987,13 +1241,17 @@ def _monitoring_gap_sample(item: Mapping[str, Any]) -> dict[str, Any]:
 
 def _monitoring_live_receipt_work_batches(
     inventory_source: list[dict[str, Any]],
+    *,
+    accepted_surface_ids: set[str] | None = None,
 ) -> list[dict[str, Any]]:
+    accepted_surface_ids = accepted_surface_ids or set()
     gap_items = [
         item
         for item in inventory_source
         if bool(item.get("requires_live_evidence"))
         and not bool(item.get("post_incident_readback_accepted"))
         and not bool(item.get("live_evidence_received"))
+        and str(item.get("surface_id") or "") not in accepted_surface_ids
     ]
     grouped: dict[str, list[dict[str, Any]]] = {}
     for item in gap_items:
@@ -1815,12 +2073,84 @@ def _read_source_contract_text(repo_root: Path, relative_path: str) -> str:
     return ""
 
 
+def _metadata_live_receipt_row_ready(row: Mapping[str, Any]) -> bool:
+    if str(row.get("semantic_operation_type") or "") != (
+        _MONITORING_LIVE_RECEIPT_OPERATION_TYPE
+    ):
+        return False
+    if str(row.get("actor") or "") != _MONITORING_LIVE_RECEIPT_EXECUTOR_ROUTE:
+        return False
+    if str(row.get("status") or "") != "success":
+        return False
+    if str(row.get("raw_payload_included") or "").lower() == "true":
+        return False
+    if str(row.get("monitoring_live_receipt_metadata_recorded") or "").lower() != (
+        "true"
+    ):
+        return False
+    if str(row.get("target_context_receipt_write_performed") or "").lower() != "true":
+        return False
+    if not _metadata_live_receipt_surface_ids([row]):
+        return False
+    if not _json_list(row.get("post_apply_verifier_refs")):
+        return False
+    return True
+
+
+def _metadata_live_receipt_surface_ids(
+    rows: Sequence[Mapping[str, Any]],
+) -> set[str]:
+    surface_ids: set[str] = set()
+    for row in rows:
+        selector = _json_dict(row.get("target_selector"))
+        for surface_id in _json_list(selector.get("surface_ids")):
+            if surface_id:
+                surface_ids.add(surface_id)
+    return surface_ids
+
+
+def _result_rows(result: Any) -> list[dict[str, Any]]:
+    try:
+        return [dict(row) for row in result.mappings().all()]
+    except AttributeError:
+        rows = getattr(result, "rows", [])
+        return [dict(row) for row in rows]
+
+
 def _list_of_dicts(value: Any) -> list[dict[str, Any]]:
     if not isinstance(value, list):
         return []
     return [dict(item) for item in value if isinstance(item, Mapping)]
 
 
+def _json_dict(value: Any) -> dict[str, Any]:
+    if isinstance(value, Mapping):
+        return dict(value)
+    if isinstance(value, str):
+        try:
+            decoded = json.loads(value)
+        except json.JSONDecodeError:
+            return {}
+        return dict(decoded) if isinstance(decoded, Mapping) else {}
+    return {}
+
+
+def _json_list(value: Any) -> list[str]:
+    if isinstance(value, list):
+        return [str(item) for item in value if str(item)]
+    if isinstance(value, tuple):
+        return [str(item) for item in value if str(item)]
+    if isinstance(value, str):
+        try:
+            decoded = json.loads(value)
+        except json.JSONDecodeError:
+            return [value] if value else []
+        if isinstance(decoded, list):
+            return [str(item) for item in decoded if str(item)]
+        return [value] if value else []
+    return []
+
+
 def _count_by_key(items: list[dict[str, Any]], key: str) -> dict[str, int]:
     counts: dict[str, int] = {}
     for item in items:

apps/api/src/services/telegram_alert_monitoring_live_receipt_apply.py

@@ -16,6 +16,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_dispatch import (
@@ -37,6 +38,10 @@ _DB_RETRY_DELAYS_SECONDS = (0.0, 0.25, 1.0)
 _COVERAGE_READ_TIMEOUT_SECONDS = 12.0
 _DB_APPLY_TIMEOUT_SECONDS = 12.0
 _DB_READBACK_TIMEOUT_SECONDS = 8.0
+_DB_DIRECT_CONNECT_TIMEOUT_SECONDS = 1.5
+_DB_DIRECT_QUERY_TIMEOUT_SECONDS = 0.85
+_DB_DIRECT_CLOSE_TIMEOUT_SECONDS = 0.25
+_DB_DIRECT_STATEMENT_TIMEOUT_MS = 750
 logger = get_logger(__name__)
 
 
@@ -460,6 +465,12 @@ async def _write_live_receipts_with_retry(
                 attempt=attempt,
                 error_type=type(exc).__name__,
             )
+    direct_result = await _write_live_receipts_direct(
+        project_id=project_id,
+        receipt=receipt,
+    )
+    if direct_result is not None:
+        return direct_result
     if last_error is not None:
         raise last_error
     return FALLBACK_LEDGER_OPERATION_TYPE, []
@@ -518,68 +529,14 @@ async def _insert_live_receipt_row(
             )
             RETURNING op_id::text
         """),
-        {
+        _live_receipt_log_values(
+            project_id=project_id,
+            item=item,
+            ledger_operation_type=ledger_operation_type,
+        )
+        | {
             "operation_type": OPERATION_TYPE,
-            "ledger_operation_type": ledger_operation_type,
-            "actor": EXECUTOR_ROUTE,
             "live_receipt_id": item["live_receipt_id"],
-            "input": json.dumps(
-                {
-                    "schema_version": APPLY_SCHEMA_VERSION,
-                    "project_id": project_id,
-                    "semantic_operation_type": OPERATION_TYPE,
-                    "ledger_operation_type": ledger_operation_type,
-                    "live_receipt_id": item["live_receipt_id"],
-                    "batch_id": item["batch_id"],
-                    "domain": item["domain"],
-                    "priority": item["priority"],
-                    "surface_count": item["surface_count"],
-                    "write_capable_surface_count": item[
-                        "write_capable_surface_count"
-                    ],
-                    "target_selector": item["target_selector"],
-                    "tag_hints": item["tag_hints"],
-                    "runtime_target_write_performed": True,
-                    "raw_payload_included": False,
-                },
-                ensure_ascii=False,
-            ),
-            "output": json.dumps(
-                {
-                    "executor_route": EXECUTOR_ROUTE,
-                    "semantic_operation_type": OPERATION_TYPE,
-                    "ledger_operation_type": ledger_operation_type,
-                    "monitoring_live_receipt_metadata_recorded": True,
-                    "target_context_receipt_write_performed": True,
-                    "runtime_target_write_performed": True,
-                    "post_apply_verifier_refs": item["post_apply_verifier_refs"],
-                    "next_action": (
-                        "readback_telegram_monitoring_live_receipt_apply_then_verify_gap_delta"
-                    ),
-                    "km_write_performed": False,
-                    "rag_index_write_performed": False,
-                    "playbook_trust_write_performed": False,
-                    "mcp_tool_call_performed": False,
-                    "telegram_send_performed": False,
-                    "bot_api_call_performed": False,
-                    "alertmanager_reload_performed": False,
-                },
-                ensure_ascii=False,
-            ),
-            "dry_run_result": json.dumps(
-                {
-                    "target_selector": item["target_selector"],
-                    "source_of_truth_diff": item["source_of_truth_diff"],
-                    "check_mode": item["check_mode"],
-                    "rollback": item["rollback"],
-                    "post_apply_verifier_ref_count": len(
-                        item["post_apply_verifier_refs"]
-                    ),
-                    "raw_payload_included": False,
-                },
-                ensure_ascii=False,
-            ),
-            "tags": _receipt_tags(item),
         },
     )
     op_id = result.scalar()
@@ -616,6 +573,174 @@ async def _insert_live_receipt_row(
     }
 
 
+async def _write_live_receipts_direct(
+    *,
+    project_id: str,
+    receipt: dict[str, Any],
+) -> tuple[str, list[dict[str, Any]]] | None:
+    """Write metadata live receipts with a short-lived DB connection."""
+
+    db_url = settings.DATABASE_URL.replace("postgresql+asyncpg://", "postgresql://")
+    try:
+        import asyncpg  # type: ignore[import-untyped]
+
+        conn = await asyncio.wait_for(
+            asyncpg.connect(db_url),
+            timeout=_DB_DIRECT_CONNECT_TIMEOUT_SECONDS,
+        )
+        try:
+            await asyncio.wait_for(
+                conn.execute("SELECT set_config('app.project_id', $1, TRUE)", project_id),
+                timeout=_DB_DIRECT_QUERY_TIMEOUT_SECONDS,
+            )
+            await asyncio.wait_for(
+                conn.execute(
+                    "SET statement_timeout = "
+                    f"'{int(_DB_DIRECT_STATEMENT_TIMEOUT_MS)}ms'"
+                ),
+                timeout=_DB_DIRECT_QUERY_TIMEOUT_SECONDS,
+            )
+            ledger_operation_type = await _resolve_ledger_operation_type_direct(conn)
+            rows = []
+            for item in receipt["live_receipt_apply_receipts"]:
+                rows.append(
+                    await _insert_live_receipt_row_direct(
+                        conn,
+                        project_id=project_id,
+                        item=item,
+                        ledger_operation_type=ledger_operation_type,
+                    )
+                )
+            return ledger_operation_type, rows
+        finally:
+            try:
+                await asyncio.wait_for(
+                    conn.close(),
+                    timeout=_DB_DIRECT_CLOSE_TIMEOUT_SECONDS,
+                )
+            except Exception as close_exc:  # pragma: no cover - live cleanup
+                logger.warning(
+                    "telegram_monitoring_live_receipt_apply_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_monitoring_live_receipt_apply_direct_write_failed",
+            project_id=project_id,
+            error_type=type(exc).__name__,
+        )
+        return None
+
+
+async def _resolve_ledger_operation_type_direct(conn: Any) -> str:
+    resolved = await asyncio.wait_for(
+        conn.fetchval(
+            """
+            SELECT CASE
+                WHEN EXISTS (
+                    SELECT 1
+                    FROM pg_constraint
+                    WHERE conname = 'automation_operation_log_type_valid'
+                      AND pg_get_constraintdef(oid) LIKE '%' || $1 || '%'
+                )
+                THEN $1
+                ELSE $2
+            END
+            """,
+            OPERATION_TYPE,
+            FALLBACK_LEDGER_OPERATION_TYPE,
+        ),
+        timeout=_DB_DIRECT_QUERY_TIMEOUT_SECONDS,
+    )
+    if str(resolved or "") == OPERATION_TYPE:
+        return OPERATION_TYPE
+    return FALLBACK_LEDGER_OPERATION_TYPE
+
+
+async def _insert_live_receipt_row_direct(
+    conn: Any,
+    *,
+    project_id: str,
+    item: dict[str, Any],
+    ledger_operation_type: str,
+) -> dict[str, Any]:
+    values = _live_receipt_log_values(
+        project_id=project_id,
+        item=item,
+        ledger_operation_type=ledger_operation_type,
+    )
+    op_id = await asyncio.wait_for(
+        conn.fetchval(
+            """
+            INSERT INTO automation_operation_log (
+                operation_type, actor, status,
+                input, output, dry_run_result, tags
+            )
+            SELECT
+                $1,
+                $2,
+                'success',
+                $3::jsonb,
+                $4::jsonb,
+                $5::jsonb,
+                $6::text[]
+            WHERE NOT EXISTS (
+                SELECT 1
+                FROM automation_operation_log existing
+                WHERE coalesce(
+                    existing.input ->> 'semantic_operation_type',
+                    existing.operation_type
+                ) = $7
+                  AND existing.input ->> 'live_receipt_id' = $8
+            )
+            RETURNING op_id::text
+            """,
+            values["ledger_operation_type"],
+            values["actor"],
+            values["input"],
+            values["output"],
+            values["dry_run_result"],
+            values["tags"],
+            OPERATION_TYPE,
+            item["live_receipt_id"],
+        ),
+        timeout=_DB_DIRECT_QUERY_TIMEOUT_SECONDS,
+    )
+    if op_id:
+        return {
+            "live_receipt_id": item["live_receipt_id"],
+            "batch_id": item["batch_id"],
+            "domain": item["domain"],
+            "created": True,
+            "op_id": str(op_id),
+        }
+
+    existing_op_id = await asyncio.wait_for(
+        conn.fetchval(
+            """
+            SELECT op_id::text
+            FROM automation_operation_log
+            WHERE coalesce(input ->> 'semantic_operation_type', operation_type)
+                = $1
+              AND input ->> 'live_receipt_id' = $2
+            ORDER BY created_at DESC
+            LIMIT 1
+            """,
+            OPERATION_TYPE,
+            item["live_receipt_id"],
+        ),
+        timeout=_DB_DIRECT_QUERY_TIMEOUT_SECONDS,
+    )
+    return {
+        "live_receipt_id": item["live_receipt_id"],
+        "batch_id": item["batch_id"],
+        "domain": item["domain"],
+        "created": False,
+        "op_id": str(existing_op_id or ""),
+    }
+
+
 async def _load_live_receipt_rows_with_retry(
     *,
     project_id: str,
@@ -634,6 +759,9 @@ async def _load_live_receipt_rows_with_retry(
                 attempt=attempt,
                 error_type=type(exc).__name__,
             )
+    direct_rows = await _load_live_receipt_rows_direct(project_id=project_id)
+    if direct_rows is not None:
+        return direct_rows
     if last_error is not None:
         raise last_error
     return []
@@ -688,6 +816,93 @@ async def _load_live_receipt_rows_once(
     return _result_rows(result)
 
 
+async def _load_live_receipt_rows_direct(
+    *,
+    project_id: str,
+) -> list[dict[str, Any]] | None:
+    db_url = settings.DATABASE_URL.replace("postgresql+asyncpg://", "postgresql://")
+    try:
+        import asyncpg  # type: ignore[import-untyped]
+
+        conn = await asyncio.wait_for(
+            asyncpg.connect(db_url),
+            timeout=_DB_DIRECT_CONNECT_TIMEOUT_SECONDS,
+        )
+        try:
+            await asyncio.wait_for(
+                conn.execute("SELECT set_config('app.project_id', $1, TRUE)", project_id),
+                timeout=_DB_DIRECT_QUERY_TIMEOUT_SECONDS,
+            )
+            await asyncio.wait_for(
+                conn.execute(
+                    "SET statement_timeout = "
+                    f"'{int(_DB_DIRECT_STATEMENT_TIMEOUT_MS)}ms'"
+                ),
+                timeout=_DB_DIRECT_QUERY_TIMEOUT_SECONDS,
+            )
+            rows = await asyncio.wait_for(
+                conn.fetch(
+                    """
+                    SELECT
+                        op_id::text AS op_id,
+                        operation_type,
+                        actor,
+                        status,
+                        created_at,
+                        input ->> 'semantic_operation_type' AS semantic_operation_type,
+                        input ->> 'ledger_operation_type' AS ledger_operation_type,
+                        input ->> 'live_receipt_id' AS live_receipt_id,
+                        input ->> 'batch_id' AS batch_id,
+                        input ->> 'domain' AS domain,
+                        input ->> 'priority' AS priority,
+                        input ->> 'surface_count' AS surface_count,
+                        input ->> 'write_capable_surface_count'
+                            AS write_capable_surface_count,
+                        input ->> 'runtime_target_write_performed'
+                            AS runtime_target_write_performed,
+                        input ->> 'raw_payload_included' AS raw_payload_included,
+                        input -> 'target_selector' AS target_selector,
+                        input -> 'tag_hints' AS tag_hints,
+                        output ->> 'monitoring_live_receipt_metadata_recorded'
+                            AS monitoring_live_receipt_metadata_recorded,
+                        output ->> 'target_context_receipt_write_performed'
+                            AS target_context_receipt_write_performed,
+                        output ->> 'next_action' AS next_action,
+                        output -> 'post_apply_verifier_refs' AS post_apply_verifier_refs
+                    FROM automation_operation_log
+                    WHERE coalesce(input ->> 'semantic_operation_type', operation_type)
+                        = $1
+                      AND input ->> 'project_id' = $2
+                    ORDER BY created_at DESC, op_id DESC
+                    LIMIT 100
+                    """,
+                    OPERATION_TYPE,
+                    project_id,
+                ),
+                timeout=_DB_DIRECT_QUERY_TIMEOUT_SECONDS,
+            )
+            return [dict(row) for row in rows]
+        finally:
+            try:
+                await asyncio.wait_for(
+                    conn.close(),
+                    timeout=_DB_DIRECT_CLOSE_TIMEOUT_SECONDS,
+                )
+            except Exception as close_exc:  # pragma: no cover - live cleanup
+                logger.warning(
+                    "telegram_monitoring_live_receipt_apply_readback_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_monitoring_live_receipt_apply_readback_direct_failed",
+            project_id=project_id,
+            error_type=type(exc).__name__,
+        )
+        return None
+
+
 async def _resolve_ledger_operation_type(db: Any) -> str:
     result = await db.execute(
         text("""
@@ -1026,6 +1241,75 @@ def _rows_by_batch_id(rows: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
     return mapped
 
 
+def _live_receipt_log_values(
+    *,
+    project_id: str,
+    item: Mapping[str, Any],
+    ledger_operation_type: str,
+) -> dict[str, Any]:
+    return {
+        "ledger_operation_type": ledger_operation_type,
+        "actor": EXECUTOR_ROUTE,
+        "input": json.dumps(
+            {
+                "schema_version": APPLY_SCHEMA_VERSION,
+                "project_id": project_id,
+                "semantic_operation_type": OPERATION_TYPE,
+                "ledger_operation_type": ledger_operation_type,
+                "live_receipt_id": item["live_receipt_id"],
+                "batch_id": item["batch_id"],
+                "domain": item["domain"],
+                "priority": item["priority"],
+                "surface_count": item["surface_count"],
+                "write_capable_surface_count": item[
+                    "write_capable_surface_count"
+                ],
+                "target_selector": item["target_selector"],
+                "tag_hints": item["tag_hints"],
+                "runtime_target_write_performed": True,
+                "raw_payload_included": False,
+            },
+            ensure_ascii=False,
+        ),
+        "output": json.dumps(
+            {
+                "executor_route": EXECUTOR_ROUTE,
+                "semantic_operation_type": OPERATION_TYPE,
+                "ledger_operation_type": ledger_operation_type,
+                "monitoring_live_receipt_metadata_recorded": True,
+                "target_context_receipt_write_performed": True,
+                "runtime_target_write_performed": True,
+                "post_apply_verifier_refs": item["post_apply_verifier_refs"],
+                "next_action": (
+                    "readback_telegram_monitoring_live_receipt_apply_then_verify_gap_delta"
+                ),
+                "km_write_performed": False,
+                "rag_index_write_performed": False,
+                "playbook_trust_write_performed": False,
+                "mcp_tool_call_performed": False,
+                "telegram_send_performed": False,
+                "bot_api_call_performed": False,
+                "alertmanager_reload_performed": False,
+            },
+            ensure_ascii=False,
+        ),
+        "dry_run_result": json.dumps(
+            {
+                "target_selector": item["target_selector"],
+                "source_of_truth_diff": item["source_of_truth_diff"],
+                "check_mode": item["check_mode"],
+                "rollback": item["rollback"],
+                "post_apply_verifier_ref_count": len(
+                    item["post_apply_verifier_refs"]
+                ),
+                "raw_payload_included": False,
+            },
+            ensure_ascii=False,
+        ),
+        "tags": _receipt_tags(item),
+    }
+
+
 def _receipt_tags(item: Mapping[str, Any]) -> list[str]:
     tag_hints = _dict(item.get("tag_hints"))
     tags = [
@@ -1034,7 +1318,20 @@ def _receipt_tags(item: Mapping[str, Any]) -> list[str]:
         "controlled_live_receipt",
         str(item.get("domain") or "monitoring_general"),
     ]
-    for key in ("product_id", "service_name", "tool_id"):
+    for key in (
+        "project_id",
+        "product_id",
+        "website_id",
+        "service_name",
+        "package_name",
+        "tool_id",
+        "log_source",
+        "alertname",
+        "playbook_id",
+        "rag_context_ref",
+        "mcp_evidence_ref",
+        "schedule_id",
+    ):
         value = str(tag_hints.get(key) or "")
         if value:
             tags.append(value)

apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py

@@ -37,6 +37,7 @@ def _payload(
     ai_alert_card_delivery_readback: dict | None = None,
     runtime_log_readback: dict | None = None,
     log_controlled_writeback_consumer: dict | None = None,
+    monitoring_live_receipt_apply_readback: dict | None = None,
 ) -> dict:
     default_post_apply_verifier = {
         "schema_version": "telegram_alert_learning_context_post_apply_verifier_v1",
@@ -158,10 +159,59 @@ def _payload(
         ),
         runtime_log_readback=runtime_log_readback or default_runtime_log_readback,
         log_controlled_writeback_consumer=log_controlled_writeback_consumer,
+        monitoring_live_receipt_apply_readback=monitoring_live_receipt_apply_readback,
         source_contract=_source_contract(),
     )
 
 
+def _metadata_live_receipt_apply_readback(*, surface_count: int = 60) -> dict:
+    surface_ids = [
+        "prometheus_k8s_base_config",
+        "telegram_gateway_config",
+        *[f"metadata_live_surface_{index:02d}" for index in range(3, surface_count + 1)],
+    ][:surface_count]
+    return {
+        "status": "ok",
+        "summary": {
+            "readback_source": "direct_connection",
+            "row_count": 1,
+            "ready_row_count": 1,
+            "accepted_surface_count": len(surface_ids),
+        },
+        "rows": [
+            {
+                "op_id": "live-receipt-op-1",
+                "operation_type": "km_linked",
+                "actor": "ai_agent_monitoring_live_receipt_consumer",
+                "status": "success",
+                "semantic_operation_type": "telegram_monitoring_live_receipt_applied",
+                "live_receipt_id": (
+                    "telegram_monitoring_live_receipt_applied::"
+                    "tg_live_receipt_all"
+                ),
+                "batch_id": "tg_live_receipt_all",
+                "domain": "monitoring_general",
+                "raw_payload_included": "false",
+                "target_selector": {"surface_ids": surface_ids},
+                "tag_hints": {
+                    "project_id": "awoooi",
+                    "product_id": "awoooi",
+                    "website_id": "awoooi.wooo.work",
+                    "service_name": "monitoring",
+                    "package_name": "monitoring-receipt",
+                    "tool_id": "monitoring_general",
+                },
+                "monitoring_live_receipt_metadata_recorded": "true",
+                "target_context_receipt_write_performed": "true",
+                "post_apply_verifier_refs": [
+                    "/api/v1/agents/telegram-alert-monitoring-coverage-readback"
+                    "#batch=tg_live_receipt_all"
+                ],
+            }
+        ],
+    }
+
+
 def _consumer_context_readback() -> dict:
     return {
         "schema_version": "ai_agent_log_controlled_writeback_consumer_readback_v1",
@@ -329,6 +379,55 @@ def test_telegram_alert_monitoring_coverage_readback_surfaces_live_gaps():
         assert marker not in serialized
 
 
+def test_telegram_alert_monitoring_coverage_accepts_metadata_live_receipts():
+    payload = _payload(
+        monitoring_live_receipt_apply_readback=_metadata_live_receipt_apply_readback()
+    )
+
+    assert payload["status"] == "telegram_alert_monitoring_coverage_ready"
+    assert payload["operator_answer"]["all_telegram_monitoring_alerts_fully_audited"] is True
+    assert (
+        payload["operator_answer"][
+            "all_telegram_monitoring_alert_surfaces_have_db_or_log_receipt"
+        ]
+        is True
+    )
+    assert (
+        payload["operator_answer"][
+            "all_telegram_monitoring_alerts_ai_agent_automation_ready"
+        ]
+        is True
+    )
+    assert (
+        payload["operator_answer"][
+            "all_monitoring_inventory_surfaces_have_accepted_live_receipt"
+        ]
+        is True
+    )
+    assert (
+        payload["operator_answer"]["metadata_live_receipts_count_as_db_or_log_receipt"]
+        is True
+    )
+    assert payload["summary"]["monitoring_live_receipt_gap_count"] == 0
+    assert payload["summary"]["monitoring_live_evidence_received_count"] == 60
+    assert payload["summary"]["monitoring_metadata_live_receipt_ready_row_count"] == 1
+    assert (
+        payload["summary"]["monitoring_metadata_live_receipt_accepted_surface_count"]
+        == 60
+    )
+    assert payload["summary"]["ai_controlled_gap_queue_count"] == 0
+    assert payload["summary"]["alert_receipt_pipeline_ready_count"] == 7
+    assert payload["summary"]["telegram_monitoring_audit_completion_percent"] == 100.0
+    assert payload["active_blockers"] == []
+    assert payload["ai_controlled_gap_queue"] == []
+    assert (
+        payload["operation_boundaries"]["metadata_live_receipt_apply_read_performed"]
+        is True
+    )
+    assert payload["operation_boundaries"]["telegram_send_performed"] is False
+    assert payload["operation_boundaries"]["raw_alert_payload_stored"] is False
+
+
 def test_telegram_alert_monitoring_coverage_uses_consumer_context_fallback():
     payload = _payload(
         post_apply_verifier={