fix(agents): apply telegram monitoring live receipts

apps/api/src/api/v1/agents.py

@@ -479,6 +479,10 @@ from src.services.telegram_alert_learning_context_post_apply_verifier import (
 from src.services.telegram_alert_monitoring_coverage_readback import (
     load_latest_telegram_alert_monitoring_coverage_readback,
 )
+from src.services.telegram_alert_monitoring_live_receipt_apply import (
+    apply_latest_telegram_alert_monitoring_live_receipts,
+    load_latest_telegram_alert_monitoring_live_receipt_apply_readback,
+)
 
 router = APIRouter(prefix="/agents", tags=["Agent Teams"])
 logger = get_logger("awoooi.agents")
@@ -2548,6 +2552,62 @@ async def get_telegram_alert_monitoring_coverage_readback() -> dict[str, Any]:
         ) from exc
 
 
+@router.post(
+    "/telegram-alert-monitoring-live-receipt-apply",
+    response_model=dict[str, Any],
+    summary="執行 Telegram monitoring live receipt metadata apply",
+    description=(
+        "把 Telegram monitoring coverage readback 中的 AI controlled live receipt "
+        "batches 寫入 automation_operation_log，形成 metadata-only receipt，"
+        "供 KM / RAG / MCP / PlayBook / AI Agent loop 消費。此端點不送 Telegram、"
+        "不呼叫 Bot API、不改 receiver、不 reload Alertmanager、不觸發 workflow、"
+        "不保存 raw payload、不讀 secret、不呼叫 GitHub；critical break-glass 仍維持硬阻擋。"
+    ),
+)
+async def post_telegram_alert_monitoring_live_receipt_apply() -> dict[str, Any]:
+    """Persist metadata-only monitoring live receipt apply rows."""
+    try:
+        payload = await apply_latest_telegram_alert_monitoring_live_receipts()
+        return redact_public_lan_topology(payload)
+    except (json.JSONDecodeError, ValueError) as exc:
+        logger.error(
+            "telegram_alert_monitoring_live_receipt_apply_invalid",
+            error=str(exc),
+        )
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Telegram monitoring live receipt apply 無效",
+        ) from exc
+
+
+@router.get(
+    "/telegram-alert-monitoring-live-receipt-apply-readback",
+    response_model=dict[str, Any],
+    summary="取得 Telegram monitoring live receipt metadata apply readback",
+    description=(
+        "讀取 automation_operation_log 中的 Telegram monitoring live receipt "
+        "metadata apply rows，並映射回 coverage batches 與 surface counts。"
+        "此端點只讀 metadata，不送 Telegram、不呼叫 Bot API、不改 receiver、"
+        "不 reload Alertmanager、不觸發 workflow、不保存 raw payload、不讀 secret、"
+        "不呼叫 GitHub，也不把 source live gap 假綠成 accepted receipt。"
+    ),
+)
+async def get_telegram_alert_monitoring_live_receipt_apply_readback() -> dict[str, Any]:
+    """Read metadata-only monitoring live receipt apply rows."""
+    try:
+        payload = await load_latest_telegram_alert_monitoring_live_receipt_apply_readback()
+        return redact_public_lan_topology(payload)
+    except (json.JSONDecodeError, ValueError) as exc:
+        logger.error(
+            "telegram_alert_monitoring_live_receipt_apply_readback_invalid",
+            error=str(exc),
+        )
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Telegram monitoring live receipt apply readback 無效",
+        ) from exc
+
+
 @router.post(
     "/agent-log-controlled-writeback-consumer-apply",
     response_model=dict[str, Any],

apps/api/src/services/telegram_alert_monitoring_live_receipt_apply.py

@@ -0,0 +1,935 @@
+"""Telegram monitoring live receipt metadata apply.
+
+This service consumes the AI-controlled monitoring live receipt batches from the
+coverage readback and writes metadata-only receipts to automation_operation_log.
+It does not send Telegram messages, call Bot API, change Alertmanager routes,
+store raw alert payloads, call MCP tools, write KM/RAG/PlayBook targets, or read
+secrets.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from collections.abc import Mapping
+from typing import Any
+
+from sqlalchemy import text
+
+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 (
+    FALLBACK_LEDGER_OPERATION_TYPE,
+)
+from src.services.telegram_alert_monitoring_coverage_readback import (
+    DEFAULT_PROJECT_ID,
+    load_latest_telegram_alert_monitoring_coverage_readback,
+)
+
+APPLY_SCHEMA_VERSION = "telegram_alert_monitoring_live_receipt_apply_receipt_v1"
+READBACK_SCHEMA_VERSION = "telegram_alert_monitoring_live_receipt_apply_readback_v1"
+OPERATION_TYPE = "telegram_monitoring_live_receipt_applied"
+EXECUTOR_ROUTE = "ai_agent_monitoring_live_receipt_consumer"
+POST_VERIFIER_ROUTE = "/api/v1/agents/telegram-alert-monitoring-coverage-readback"
+APPLY_ROUTE = "/api/v1/agents/telegram-alert-monitoring-live-receipt-apply"
+READBACK_ROUTE = "/api/v1/agents/telegram-alert-monitoring-live-receipt-apply-readback"
+_DB_RETRY_DELAYS_SECONDS = (0.0, 0.25, 1.0)
+logger = get_logger(__name__)
+
+
+async def apply_latest_telegram_alert_monitoring_live_receipts(
+    *,
+    project_id: str = DEFAULT_PROJECT_ID,
+) -> dict[str, Any]:
+    """Persist idempotent metadata live receipt rows for coverage batches."""
+
+    coverage = await load_latest_telegram_alert_monitoring_coverage_readback(
+        project_id=project_id,
+    )
+    receipt = build_telegram_alert_monitoring_live_receipt_apply_receipt(coverage)
+    if receipt["active_blockers"]:
+        receipt["apply_result"] = {
+            "operation_type": OPERATION_TYPE,
+            "ledger_operation_type": None,
+            "semantic_operation_type": OPERATION_TYPE,
+            "db_write_status": "not_attempted",
+            "live_receipt_apply_row_count": 0,
+            "reason": "active_blockers_present",
+        }
+        return receipt
+
+    try:
+        ledger_operation_type, rows = await _write_live_receipts_with_retry(
+            project_id=project_id,
+            receipt=receipt,
+        )
+    except Exception as exc:  # pragma: no cover - live DB pool pressure
+        logger.warning(
+            "telegram_monitoring_live_receipt_apply_db_unavailable",
+            project_id=project_id,
+            error_type=type(exc).__name__,
+        )
+        receipt["status"] = "blocked_waiting_telegram_monitoring_live_receipt_apply_db"
+        receipt["active_blockers"] = _unique([
+            *receipt.get("active_blockers", []),
+            "telegram_monitoring_live_receipt_apply_db_write_unavailable",
+        ])
+        receipt["apply_result"] = {
+            "operation_type": OPERATION_TYPE,
+            "ledger_operation_type": None,
+            "semantic_operation_type": OPERATION_TYPE,
+            "db_write_status": "unavailable",
+            "error_type": type(exc).__name__,
+            "live_receipt_apply_row_count": 0,
+            "rows": [],
+        }
+        return receipt
+
+    inserted_count = sum(1 for row in rows if row["created"] is True)
+    existing_count = sum(1 for row in rows if row["created"] is False)
+    receipt["apply_result"] = {
+        "operation_type": OPERATION_TYPE,
+        "ledger_operation_type": ledger_operation_type,
+        "semantic_operation_type": OPERATION_TYPE,
+        "inserted_count": inserted_count,
+        "existing_count": existing_count,
+        "live_receipt_apply_row_count": len(rows),
+        "rows": rows,
+    }
+    receipt["rollups"]["runtime_target_write_performed"] = True
+    receipt["rollups"]["live_receipt_apply_row_count"] = len(rows)
+    receipt["rollups"]["inserted_live_receipt_apply_row_count"] = inserted_count
+    receipt["rollups"]["existing_live_receipt_apply_row_count"] = existing_count
+    receipt["operation_boundaries"]["monitoring_live_receipt_metadata_write_performed"] = True
+    receipt["operation_boundaries"]["runtime_target_write_performed"] = True
+    return receipt
+
+
+async def load_latest_telegram_alert_monitoring_live_receipt_apply_readback(
+    *,
+    project_id: str = DEFAULT_PROJECT_ID,
+) -> dict[str, Any]:
+    """Read back metadata live receipt rows and map them to coverage batches."""
+
+    coverage = await load_latest_telegram_alert_monitoring_coverage_readback(
+        project_id=project_id,
+    )
+    try:
+        rows = await _load_live_receipt_rows_with_retry(project_id=project_id)
+    except Exception as exc:  # pragma: no cover - live DB pool pressure
+        logger.warning(
+            "telegram_monitoring_live_receipt_apply_readback_db_unavailable",
+            project_id=project_id,
+            error_type=type(exc).__name__,
+        )
+        return _fallback_apply_readback(
+            coverage=coverage,
+            project_id=project_id,
+            error_type=type(exc).__name__,
+        )
+    return build_telegram_alert_monitoring_live_receipt_apply_readback(
+        coverage=coverage,
+        rows=rows,
+    )
+
+
+def build_telegram_alert_monitoring_live_receipt_apply_receipt(
+    coverage: Mapping[str, Any],
+) -> dict[str, Any]:
+    """Build a deterministic controlled-apply envelope from coverage readback."""
+
+    batches = _coverage_batches(coverage)
+    apply_receipts = [_live_receipt_apply_receipt(batch) for batch in batches]
+    active_blockers = _apply_active_blockers(
+        coverage=coverage,
+        batches=batches,
+        apply_receipts=apply_receipts,
+    )
+    return {
+        "schema_version": APPLY_SCHEMA_VERSION,
+        "priority": "P0-TELEGRAM-MONITORING-AI-AUTOMATION-COVERAGE",
+        "scope": "telegram_alert_monitoring_live_receipt_apply",
+        "status": (
+            "telegram_monitoring_live_receipt_apply_ready"
+            if not active_blockers
+            else "blocked_waiting_telegram_monitoring_live_receipt_apply_inputs"
+        ),
+        "readback": {
+            "workplan_id": "CIR-P0-TG-001-LIVE-RECEIPT-APPLY",
+            "workplan_title": (
+                "Telegram monitoring live receipt batches consumed into AI "
+                "metadata ledger for KM / RAG / MCP / PlayBook / Agent loop"
+            ),
+            "source_schema_version": coverage.get("schema_version"),
+            "source_status": coverage.get("status"),
+            "operation_type": OPERATION_TYPE,
+            "executor_route": EXECUTOR_ROUTE,
+            "safe_next_step": (
+                "readback_live_receipt_apply_then_verify_coverage_gap_delta"
+            ),
+        },
+        "live_receipt_apply_receipts": apply_receipts,
+        "rollups": {
+            "source_live_receipt_batch_count": len(batches),
+            "ready_source_live_receipt_batch_count": sum(
+                1 for item in apply_receipts if item["status"] == "ready_for_metadata_live_receipt_write"
+            ),
+            "source_surface_count": sum(_int(batch.get("surface_count")) for batch in batches),
+            "source_write_capable_surface_count": sum(
+                _int(batch.get("write_capable_surface_count")) for batch in batches
+            ),
+            "metadata_only_receipt_count": sum(
+                1 for item in apply_receipts if item["raw_payload_included"] is False
+            ),
+            "post_apply_verifier_ref_count": sum(
+                len(item["post_apply_verifier_refs"]) for item in apply_receipts
+            ),
+            "controlled_live_receipt_apply_ready": not active_blockers,
+            "runtime_target_write_performed": False,
+            "live_receipt_apply_row_count": 0,
+            "inserted_live_receipt_apply_row_count": 0,
+            "existing_live_receipt_apply_row_count": 0,
+        },
+        "active_blockers": active_blockers,
+        "operation_boundaries": {
+            "metadata_ledger_write_only": True,
+            "monitoring_live_receipt_metadata_write_performed": False,
+            "runtime_target_write_performed": False,
+            "monitoring_inventory_source_mutated": False,
+            "monitoring_live_receipt_acceptance_mutated": False,
+            "km_write_performed": False,
+            "rag_index_write_performed": False,
+            "playbook_trust_write_performed": False,
+            "mcp_tool_call_performed": False,
+            "agent_runtime_action_performed": False,
+            "telegram_send_performed": False,
+            "bot_api_call_performed": False,
+            "alertmanager_reload_performed": False,
+            "workflow_trigger_performed": False,
+            "raw_alert_payload_stored": False,
+            "raw_payload_included": False,
+            "secret_value_read": False,
+            "github_api_used": False,
+        },
+    }
+
+
+def build_telegram_alert_monitoring_live_receipt_apply_readback(
+    *,
+    coverage: Mapping[str, Any],
+    rows: list[dict[str, Any]],
+) -> dict[str, Any]:
+    """Build live receipt apply readback from coverage batches and ledger rows."""
+
+    batches = _coverage_batches(coverage)
+    rows_by_batch = _rows_by_batch_id(rows)
+    bindings = [
+        _live_receipt_binding(batch=batch, row=rows_by_batch.get(str(batch.get("batch_id") or "")))
+        for batch in batches
+    ]
+    active_blockers = _readback_active_blockers(batches=batches, bindings=bindings)
+    applied_bindings = [
+        binding for binding in bindings if binding["target_write_performed"] is True
+    ]
+    coverage_summary = _dict(coverage.get("summary"))
+    expected_surface_count = sum(_int(batch.get("surface_count")) for batch in batches)
+    applied_surface_count = sum(
+        _int(binding.get("surface_count")) for binding in applied_bindings
+    )
+    ready = not active_blockers and bool(batches)
+    return {
+        "schema_version": READBACK_SCHEMA_VERSION,
+        "priority": "P0-TELEGRAM-MONITORING-AI-AUTOMATION-COVERAGE",
+        "scope": "telegram_alert_monitoring_live_receipt_apply_readback",
+        "status": (
+            "telegram_monitoring_live_receipt_apply_readback_ready"
+            if ready
+            else "blocked_waiting_telegram_monitoring_live_receipt_apply_receipts"
+        ),
+        "readback": {
+            "workplan_id": "CIR-P0-TG-001-LIVE-RECEIPT-APPLY-READBACK",
+            "workplan_title": (
+                "Telegram monitoring live receipt metadata apply rows mapped "
+                "back to AI controlled batches"
+            ),
+            "source_schema_version": coverage.get("schema_version"),
+            "source_status": coverage.get("status"),
+            "source_coverage_route": POST_VERIFIER_ROUTE,
+            "operation_type": OPERATION_TYPE,
+            "executor_route": EXECUTOR_ROUTE,
+            "apply_route": APPLY_ROUTE,
+            "safe_next_step": (
+                "keep_live_gap_source_truth_and_consume_metadata_receipts_in_ai_loop"
+            ),
+        },
+        "operator_answer": {
+            "all_ai_controlled_live_receipt_batches_have_apply_receipt": ready,
+            "metadata_live_receipt_apply_performed": bool(applied_bindings),
+            "metadata_apply_does_not_mark_monitoring_live_receipt_accepted": True,
+            "monitoring_live_receipt_gap_count_source_truth": _int(
+                coverage_summary.get("monitoring_live_receipt_gap_count")
+            ),
+            "manual_default_terminal_state_allowed": False,
+            "critical_break_glass_exception_only": True,
+        },
+        "live_receipt_bindings": bindings,
+        "rollups": {
+            "expected_live_receipt_batch_count": len(batches),
+            "applied_live_receipt_batch_count": len(applied_bindings),
+            "ready_live_receipt_binding_count": sum(
+                1 for binding in bindings if binding["status"] == "metadata_live_receipt_applied"
+            ),
+            "expected_surface_count": expected_surface_count,
+            "applied_surface_count": applied_surface_count,
+            "expected_write_capable_surface_count": sum(
+                _int(batch.get("write_capable_surface_count")) for batch in batches
+            ),
+            "applied_write_capable_surface_count": sum(
+                _int(binding.get("write_capable_surface_count"))
+                for binding in applied_bindings
+            ),
+            "metadata_only_receipt_count": sum(
+                1 for binding in bindings if binding["raw_payload_included"] is False
+            ),
+            "post_apply_verifier_ref_count": sum(
+                len(binding["post_apply_verifier_refs"]) for binding in bindings
+            ),
+            "monitoring_live_receipt_gap_count_source_truth": _int(
+                coverage_summary.get("monitoring_live_receipt_gap_count")
+            ),
+            "metadata_live_receipt_apply_readback_ready": ready,
+            "runtime_target_write_performed": bool(applied_bindings),
+        },
+        "active_blockers": active_blockers,
+        "operation_boundaries": {
+            "readback_only": True,
+            "metadata_ledger_read_performed": True,
+            "metadata_live_receipt_apply_read_performed": True,
+            "runtime_target_write_performed": bool(applied_bindings),
+            "monitoring_inventory_source_mutated": False,
+            "monitoring_live_receipt_acceptance_mutated": False,
+            "km_write_performed": False,
+            "rag_index_write_performed": False,
+            "playbook_trust_write_performed": False,
+            "mcp_tool_call_performed": False,
+            "agent_runtime_action_performed": False,
+            "telegram_send_performed": False,
+            "bot_api_call_performed": False,
+            "alertmanager_reload_performed": False,
+            "workflow_trigger_performed": False,
+            "raw_alert_payload_stored": False,
+            "raw_payload_included": False,
+            "secret_value_read": False,
+            "github_api_used": False,
+        },
+    }
+
+
+async def _write_live_receipts_with_retry(
+    *,
+    project_id: str,
+    receipt: dict[str, Any],
+) -> tuple[str, list[dict[str, Any]]]:
+    last_error: Exception | None = None
+    for attempt, delay_seconds in enumerate(_DB_RETRY_DELAYS_SECONDS, start=1):
+        if delay_seconds:
+            await asyncio.sleep(delay_seconds)
+        try:
+            return await _write_live_receipts_once(
+                project_id=project_id,
+                receipt=receipt,
+            )
+        except Exception as exc:  # pragma: no cover - exercised by live pressure
+            last_error = exc
+            logger.warning(
+                "telegram_monitoring_live_receipt_apply_attempt_failed",
+                project_id=project_id,
+                attempt=attempt,
+                error_type=type(exc).__name__,
+            )
+    if last_error is not None:
+        raise last_error
+    return FALLBACK_LEDGER_OPERATION_TYPE, []
+
+
+async def _write_live_receipts_once(
+    *,
+    project_id: str,
+    receipt: dict[str, Any],
+) -> tuple[str, list[dict[str, Any]]]:
+    rows = []
+    async with get_db_context(project_id) as db:
+        await db.execute(text("SET LOCAL statement_timeout = '5000ms'"), {})
+        ledger_operation_type = await _resolve_ledger_operation_type(db)
+        for item in receipt["live_receipt_apply_receipts"]:
+            rows.append(
+                await _insert_live_receipt_row(
+                    db,
+                    project_id=project_id,
+                    item=item,
+                    ledger_operation_type=ledger_operation_type,
+                )
+            )
+    return ledger_operation_type, rows
+
+
+async def _insert_live_receipt_row(
+    db: Any,
+    *,
+    project_id: str,
+    item: dict[str, Any],
+    ledger_operation_type: str,
+) -> dict[str, Any]:
+    result = await db.execute(
+        text("""
+            INSERT INTO automation_operation_log (
+                operation_type, actor, status,
+                input, output, dry_run_result, tags
+            )
+            SELECT
+                :ledger_operation_type,
+                :actor,
+                'success',
+                CAST(:input AS jsonb),
+                CAST(:output AS jsonb),
+                CAST(:dry_run_result AS jsonb),
+                :tags
+            WHERE NOT EXISTS (
+                SELECT 1
+                FROM automation_operation_log existing
+                WHERE coalesce(
+                    existing.input ->> 'semantic_operation_type',
+                    existing.operation_type
+                ) = :operation_type
+                  AND existing.input ->> 'live_receipt_id' = :live_receipt_id
+            )
+            RETURNING op_id::text
+        """),
+        {
+            "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()
+    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 = await db.execute(
+        text("""
+            SELECT op_id::text
+            FROM automation_operation_log
+            WHERE coalesce(input ->> 'semantic_operation_type', operation_type)
+                = :operation_type
+              AND input ->> 'live_receipt_id' = :live_receipt_id
+            ORDER BY created_at DESC
+            LIMIT 1
+        """),
+        {
+            "operation_type": OPERATION_TYPE,
+            "live_receipt_id": item["live_receipt_id"],
+        },
+    )
+    return {
+        "live_receipt_id": item["live_receipt_id"],
+        "batch_id": item["batch_id"],
+        "domain": item["domain"],
+        "created": False,
+        "op_id": str(existing.scalar() or ""),
+    }
+
+
+async def _load_live_receipt_rows_with_retry(
+    *,
+    project_id: str,
+) -> list[dict[str, Any]]:
+    last_error: Exception | None = None
+    for attempt, delay_seconds in enumerate(_DB_RETRY_DELAYS_SECONDS, start=1):
+        if delay_seconds:
+            await asyncio.sleep(delay_seconds)
+        try:
+            return await _load_live_receipt_rows_once(project_id=project_id)
+        except Exception as exc:  # pragma: no cover - exercised by live pressure
+            last_error = exc
+            logger.warning(
+                "telegram_monitoring_live_receipt_apply_readback_attempt_failed",
+                project_id=project_id,
+                attempt=attempt,
+                error_type=type(exc).__name__,
+            )
+    if last_error is not None:
+        raise last_error
+    return []
+
+
+async def _load_live_receipt_rows_once(
+    *,
+    project_id: str,
+) -> list[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 ->> '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)
+                    = :operation_type
+                  AND input ->> 'project_id' = :project_id
+                ORDER BY created_at DESC, op_id DESC
+                LIMIT 100
+            """),
+            {
+                "operation_type": OPERATION_TYPE,
+                "project_id": project_id,
+            },
+        )
+    return _result_rows(result)
+
+
+async def _resolve_ledger_operation_type(db: Any) -> str:
+    result = await db.execute(
+        text("""
+            SELECT CASE
+                WHEN EXISTS (
+                    SELECT 1
+                    FROM pg_constraint
+                    WHERE conname = 'automation_operation_log_type_valid'
+                      AND pg_get_constraintdef(oid) LIKE '%' || :operation_type || '%'
+                )
+                THEN :operation_type
+                ELSE :fallback_operation_type
+            END
+        """),
+        {
+            "operation_type": OPERATION_TYPE,
+            "fallback_operation_type": FALLBACK_LEDGER_OPERATION_TYPE,
+        },
+    )
+    resolved = str(result.scalar() or "")
+    if resolved == OPERATION_TYPE:
+        return OPERATION_TYPE
+    return FALLBACK_LEDGER_OPERATION_TYPE
+
+
+def _live_receipt_apply_receipt(batch: Mapping[str, Any]) -> dict[str, Any]:
+    batch_id = str(batch.get("batch_id") or "")
+    domain = str(batch.get("domain") or "monitoring_general")
+    verifier_ref = str(batch.get("post_verifier") or POST_VERIFIER_ROUTE)
+    return {
+        "live_receipt_id": f"{OPERATION_TYPE}::{batch_id}",
+        "batch_id": batch_id,
+        "domain": domain,
+        "priority": str(batch.get("priority") or "P1"),
+        "surface_count": _int(batch.get("surface_count")),
+        "write_capable_surface_count": _int(batch.get("write_capable_surface_count")),
+        "target_selector": _dict(batch.get("target_selector")),
+        "tag_hints": _dict(batch.get("tag_hints")),
+        "executor_route": EXECUTOR_ROUTE,
+        "status": (
+            "ready_for_metadata_live_receipt_write"
+            if _batch_ready_for_apply(batch)
+            else "blocked_waiting_live_receipt_controls"
+        ),
+        "post_apply_verifier_refs": [f"{verifier_ref}#batch={batch_id}"],
+        "raw_payload_included": False,
+        "source_of_truth_diff": {
+            "current_state": "monitoring_live_receipt_gap_present",
+            "desired_state": "metadata_live_receipt_available_to_ai_loop",
+            "delta_kind": f"telegram_monitoring_live_receipt::{domain}",
+            "surface_count": _int(batch.get("surface_count")),
+            "raw_payload_included": False,
+        },
+        "check_mode": {
+            "enabled": True,
+            "checks": [
+                "batch_id_present",
+                "target_selector_surface_ids_present",
+                "tag_hints_project_product_service_tool_present",
+                "metadata_only_raw_payload_absent",
+                "post_apply_verifier_present",
+                "critical_break_glass_not_required",
+            ],
+        },
+        "rollback": {
+            "required": True,
+            "rollback_ref": (
+                "rollback://telegram-monitoring-live-receipt-apply/"
+                f"{OPERATION_TYPE}::{batch_id}"
+            ),
+            "strategy": "mark_metadata_live_receipt_superseded_and_ignore",
+        },
+    }
+
+
+def _batch_ready_for_apply(batch: Mapping[str, Any]) -> bool:
+    target_selector = _dict(batch.get("target_selector"))
+    tag_hints = _dict(batch.get("tag_hints"))
+    if not str(batch.get("batch_id") or ""):
+        return False
+    if not str(batch.get("domain") or ""):
+        return False
+    if _int(batch.get("surface_count")) <= 0:
+        return False
+    if not _strings(target_selector.get("surface_ids")):
+        return False
+    for key in ("project_id", "product_id", "service_name", "tool_id"):
+        if not str(tag_hints.get(key) or ""):
+            return False
+    if not str(batch.get("controlled_next_action") or ""):
+        return False
+    if not str(batch.get("post_verifier") or ""):
+        return False
+    return True
+
+
+def _apply_active_blockers(
+    *,
+    coverage: Mapping[str, Any],
+    batches: list[dict[str, Any]],
+    apply_receipts: list[dict[str, Any]],
+) -> list[str]:
+    blockers: list[str] = []
+    if not batches:
+        blockers.append("telegram_monitoring_live_receipt_batches_missing")
+    boundaries = _dict(coverage.get("operation_boundaries"))
+    unsafe_boundary_keys = (
+        "telegram_send_performed",
+        "bot_api_call_performed",
+        "gateway_queue_write_performed",
+        "receiver_route_changed",
+        "alertmanager_reload_performed",
+        "raw_alert_payload_stored",
+        "secret_value_read",
+        "github_api_used",
+    )
+    for key in unsafe_boundary_keys:
+        if boundaries.get(key) is True:
+            blockers.append(f"coverage_boundary_{key}_unexpected")
+    for item in apply_receipts:
+        batch_id = item["batch_id"] or "unknown_batch"
+        if item["status"] != "ready_for_metadata_live_receipt_write":
+            blockers.append(f"{batch_id}_metadata_live_receipt_not_ready")
+        if item["raw_payload_included"] is not False:
+            blockers.append(f"{batch_id}_raw_payload_included")
+    return _unique(blockers)
+
+
+def _live_receipt_binding(
+    *,
+    batch: Mapping[str, Any],
+    row: Mapping[str, Any] | None,
+) -> dict[str, Any]:
+    batch_id = str(batch.get("batch_id") or "")
+    domain = str(batch.get("domain") or "monitoring_general")
+    raw_payload_included = (
+        str((row or {}).get("raw_payload_included") or "").lower() == "true"
+    )
+    row_ready = _row_ready(row)
+    return {
+        "live_receipt_id": f"{OPERATION_TYPE}::{batch_id}",
+        "batch_id": batch_id,
+        "domain": domain,
+        "priority": str(batch.get("priority") or "P1"),
+        "surface_count": _int(batch.get("surface_count")),
+        "write_capable_surface_count": _int(batch.get("write_capable_surface_count")),
+        "target_selector": _dict(batch.get("target_selector")),
+        "tag_hints": _dict(batch.get("tag_hints")),
+        "status": (
+            "metadata_live_receipt_applied"
+            if row_ready
+            else "waiting_metadata_live_receipt_apply"
+        ),
+        "ledger_op_id": str((row or {}).get("op_id") or ""),
+        "ledger_status": str((row or {}).get("status") or ""),
+        "semantic_operation_type": str((row or {}).get("semantic_operation_type") or ""),
+        "ledger_operation_type": str((row or {}).get("ledger_operation_type") or ""),
+        "executor_route": str((row or {}).get("actor") or EXECUTOR_ROUTE),
+        "target_write_performed": row_ready,
+        "runtime_target_write_performed": (
+            str((row or {}).get("runtime_target_write_performed") or "").lower()
+            == "true"
+        ),
+        "target_context_receipt_write_performed": (
+            str((row or {}).get("target_context_receipt_write_performed") or "").lower()
+            == "true"
+        ),
+        "post_apply_verifier_refs": _strings((row or {}).get("post_apply_verifier_refs"))
+        or [f"{POST_VERIFIER_ROUTE}#batch={batch_id}"],
+        "next_action": str((row or {}).get("next_action") or ""),
+        "raw_payload_included": raw_payload_included,
+    }
+
+
+def _row_ready(row: Mapping[str, Any] | None) -> bool:
+    if not row:
+        return False
+    if str(row.get("semantic_operation_type") or "") != OPERATION_TYPE:
+        return False
+    if str(row.get("actor") or "") != EXECUTOR_ROUTE:
+        return False
+    if str(row.get("status") or "") != "success":
+        return False
+    if not str(row.get("live_receipt_id") or ""):
+        return False
+    if not str(row.get("batch_id") or ""):
+        return False
+    if str(row.get("raw_payload_included") or "").lower() == "true":
+        return False
+    return bool(_strings(row.get("post_apply_verifier_refs")))
+
+
+def _readback_active_blockers(
+    *,
+    batches: list[dict[str, Any]],
+    bindings: list[dict[str, Any]],
+) -> list[str]:
+    blockers: list[str] = []
+    if not batches:
+        blockers.append("telegram_monitoring_live_receipt_batches_missing")
+    for binding in bindings:
+        batch_id = binding["batch_id"] or "unknown_batch"
+        if binding["status"] != "metadata_live_receipt_applied":
+            blockers.append(f"{batch_id}_metadata_live_receipt_missing")
+        if binding["raw_payload_included"] is not False:
+            blockers.append(f"{batch_id}_raw_payload_included")
+    return _unique(blockers)
+
+
+def _fallback_apply_readback(
+    *,
+    coverage: Mapping[str, Any],
+    project_id: str,
+    error_type: str,
+) -> dict[str, Any]:
+    batches = _coverage_batches(coverage)
+    coverage_summary = _dict(coverage.get("summary"))
+    return {
+        "schema_version": READBACK_SCHEMA_VERSION,
+        "priority": "P0-TELEGRAM-MONITORING-AI-AUTOMATION-COVERAGE",
+        "scope": "telegram_alert_monitoring_live_receipt_apply_readback",
+        "project_id": project_id,
+        "status": "blocked_waiting_telegram_monitoring_live_receipt_apply_db_readback",
+        "readback": {
+            "workplan_id": "CIR-P0-TG-001-LIVE-RECEIPT-APPLY-READBACK",
+            "source_schema_version": coverage.get("schema_version"),
+            "source_status": coverage.get("status"),
+            "db_read_status": "unavailable",
+            "error_type": error_type,
+            "operation_type": OPERATION_TYPE,
+            "executor_route": EXECUTOR_ROUTE,
+            "apply_route": APPLY_ROUTE,
+            "safe_next_step": "retry_live_receipt_apply_readback_after_db_pool_pressure_clears",
+        },
+        "operator_answer": {
+            "all_ai_controlled_live_receipt_batches_have_apply_receipt": False,
+            "metadata_live_receipt_apply_performed": False,
+            "metadata_apply_does_not_mark_monitoring_live_receipt_accepted": True,
+            "monitoring_live_receipt_gap_count_source_truth": _int(
+                coverage_summary.get("monitoring_live_receipt_gap_count")
+            ),
+            "manual_default_terminal_state_allowed": False,
+            "critical_break_glass_exception_only": True,
+        },
+        "live_receipt_bindings": [],
+        "rollups": {
+            "expected_live_receipt_batch_count": len(batches),
+            "applied_live_receipt_batch_count": 0,
+            "ready_live_receipt_binding_count": 0,
+            "expected_surface_count": sum(_int(batch.get("surface_count")) for batch in batches),
+            "applied_surface_count": 0,
+            "expected_write_capable_surface_count": sum(
+                _int(batch.get("write_capable_surface_count")) for batch in batches
+            ),
+            "applied_write_capable_surface_count": 0,
+            "metadata_only_receipt_count": 0,
+            "post_apply_verifier_ref_count": 0,
+            "monitoring_live_receipt_gap_count_source_truth": _int(
+                coverage_summary.get("monitoring_live_receipt_gap_count")
+            ),
+            "metadata_live_receipt_apply_readback_ready": False,
+            "runtime_target_write_performed": False,
+        },
+        "active_blockers": [
+            "telegram_monitoring_live_receipt_apply_db_readback_unavailable"
+        ],
+        "operation_boundaries": {
+            "readback_only": True,
+            "metadata_ledger_read_performed": False,
+            "metadata_live_receipt_apply_read_performed": False,
+            "runtime_target_write_performed": False,
+            "monitoring_inventory_source_mutated": False,
+            "monitoring_live_receipt_acceptance_mutated": False,
+            "km_write_performed": False,
+            "rag_index_write_performed": False,
+            "playbook_trust_write_performed": False,
+            "mcp_tool_call_performed": False,
+            "agent_runtime_action_performed": False,
+            "telegram_send_performed": False,
+            "bot_api_call_performed": False,
+            "alertmanager_reload_performed": False,
+            "workflow_trigger_performed": False,
+            "raw_alert_payload_stored": False,
+            "raw_payload_included": False,
+            "secret_value_read": False,
+            "github_api_used": False,
+        },
+    }
+
+
+def _coverage_batches(coverage: Mapping[str, Any]) -> list[dict[str, Any]]:
+    batches = coverage.get("ai_controlled_live_receipt_batches")
+    if not isinstance(batches, list):
+        return []
+    return [dict(batch) for batch in batches if isinstance(batch, dict)]
+
+
+def _rows_by_batch_id(rows: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
+    mapped: dict[str, dict[str, Any]] = {}
+    for row in rows:
+        batch_id = str(row.get("batch_id") or "")
+        if not batch_id or batch_id in mapped:
+            continue
+        mapped[batch_id] = row
+    return mapped
+
+
+def _receipt_tags(item: Mapping[str, Any]) -> list[str]:
+    tag_hints = _dict(item.get("tag_hints"))
+    tags = [
+        "ai_agent",
+        "telegram_monitoring",
+        "controlled_live_receipt",
+        str(item.get("domain") or "monitoring_general"),
+    ]
+    for key in ("product_id", "service_name", "tool_id"):
+        value = str(tag_hints.get(key) or "")
+        if value:
+            tags.append(value)
+    return _unique(tags)
+
+
+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 _dict(value: Any) -> dict[str, Any]:
+    return dict(value) if isinstance(value, Mapping) else {}
+
+
+def _strings(value: Any) -> list[str]:
+    if isinstance(value, list):
+        return [str(item) for item in value]
+    if isinstance(value, tuple):
+        return [str(item) for item in value]
+    return []
+
+
+def _int(value: Any) -> int:
+    try:
+        return int(value or 0)
+    except (TypeError, ValueError):
+        return 0
+
+
+def _unique(values: list[str]) -> list[str]:
+    seen = set()
+    result = []
+    for value in values:
+        if value in seen:
+            continue
+        seen.add(value)
+        result.append(value)
+    return result