fix(agents): batch monitoring live receipt gaps

apps/api/src/services/telegram_alert_monitoring_coverage_readback.py

@@ -322,6 +322,9 @@ def build_telegram_alert_monitoring_coverage_readback(
                 1 for stage in alert_receipt_pipeline if stage["ready"]
             ),
             "ai_controlled_gap_queue_count": len(ai_controlled_gap_queue),
+            "ai_controlled_live_receipt_batch_count": _int(
+                monitoring_asset_rollups.get("live_receipt_work_batch_count")
+            ),
             "required_tag_dimension_count": len(_REQUIRED_TAG_DIMENSIONS),
             "monitoring_asset_config_kind_count": len(
                 monitoring_asset_rollups["by_config_kind"]
@@ -333,6 +336,9 @@ def build_telegram_alert_monitoring_coverage_readback(
         },
         "required_tag_dimensions": _required_tag_dimensions(),
         "monitoring_asset_rollups": monitoring_asset_rollups,
+        "ai_controlled_live_receipt_batches": _list_of_dicts(
+            monitoring_asset_rollups.get("live_receipt_work_batches")
+        ),
         "runtime_event_type_counts_7d": runtime_event_type_counts,
         "alert_receipt_pipeline": alert_receipt_pipeline,
         "work_item_progress": work_item_progress,
@@ -555,6 +561,7 @@ def _monitoring_asset_rollups(
         and not bool(item.get("post_incident_readback_accepted"))
         and not bool(item.get("live_evidence_received"))
     ][:12]
+    live_receipt_work_batches = _monitoring_live_receipt_work_batches(inventory_source)
     return {
         "surface_count": len(surfaces) or len(inventory_source),
         "readback_candidate_count": len(readback_candidates),
@@ -575,6 +582,8 @@ def _monitoring_asset_rollups(
         "by_reviewer_outcome": _count_by_key(inventory_source, "reviewer_outcome"),
         "by_expected_scope_prefix": _count_by_expected_scope_prefix(inventory_source),
         "live_receipt_gap_samples": live_receipt_gap_samples,
+        "live_receipt_work_batch_count": len(live_receipt_work_batches),
+        "live_receipt_work_batches": live_receipt_work_batches,
     }
 
 
@@ -609,6 +618,170 @@ def _monitoring_gap_sample(item: Mapping[str, Any]) -> dict[str, Any]:
     }
 
 
+def _monitoring_live_receipt_work_batches(
+    inventory_source: list[dict[str, Any]],
+) -> list[dict[str, Any]]:
+    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"))
+    ]
+    grouped: dict[str, list[dict[str, Any]]] = {}
+    for item in gap_items:
+        grouped.setdefault(_monitoring_receipt_domain(item), []).append(item)
+
+    batches: list[dict[str, Any]] = []
+    for index, (domain, items) in enumerate(sorted(grouped.items()), start=1):
+        missing_fields = sorted(
+            {
+                field
+                for item in items
+                for field in _monitoring_gap_sample(item)["missing_receipt_fields"]
+            }
+        )
+        source_paths = sorted(
+            {
+                str(item.get("repo_source_path") or item.get("source_path") or "")
+                for item in items
+                if item.get("repo_source_path") or item.get("source_path")
+            }
+        )
+        write_capable_count = sum(
+            1 for item in items if bool(item.get("write_capable_surface"))
+        )
+        batches.append({
+            "batch_id": f"tg_live_receipt_{index:02d}_{domain}",
+            "domain": domain,
+            "priority": "P0" if write_capable_count or index <= 6 else "P1",
+            "surface_count": len(items),
+            "write_capable_surface_count": write_capable_count,
+            "requires_live_evidence_count": sum(
+                1 for item in items if bool(item.get("requires_live_evidence"))
+            ),
+            "missing_receipt_fields": missing_fields,
+            "tag_hints": _monitoring_batch_tag_hints(domain, items),
+            "target_selector": {
+                "surface_ids": [str(item.get("surface_id") or "") for item in items],
+                "config_kinds": sorted(
+                    {str(item.get("config_kind") or "unknown") for item in items}
+                ),
+                "control_tiers": sorted(
+                    {str(item.get("control_tier") or "unknown") for item in items}
+                ),
+                "expected_scope_prefixes": sorted(
+                    {
+                        str(item.get("expected_scope") or "unknown").split("_", 1)[0]
+                        for item in items
+                    }
+                ),
+                "repo_source_paths": source_paths[:12],
+            },
+            "controlled_next_action": (
+                "ingest_metadata_only_live_receipt_batch_then_verify_gap_delta"
+            ),
+            "post_verifier": "/api/v1/agents/telegram-alert-monitoring-coverage-readback",
+            "sample_surfaces": [_monitoring_gap_sample(item) for item in items[:4]],
+        })
+    return batches
+
+
+def _monitoring_receipt_domain(item: Mapping[str, Any]) -> str:
+    config_kind = str(item.get("config_kind") or "").lower()
+    expected_scope = str(item.get("expected_scope") or "").lower()
+    source_path = str(
+        item.get("repo_source_path") or item.get("source_path") or ""
+    ).lower()
+    haystack = f"{config_kind} {expected_scope} {source_path}"
+    for domain, needles in (
+        ("telegram", ("telegram",)),
+        ("alertmanager", ("alertmanager", "webhook_receiver")),
+        ("grafana", ("grafana",)),
+        ("signoz", ("signoz",)),
+        ("sentry", ("sentry",)),
+        ("langfuse", ("langfuse",)),
+        ("otel", ("otel", "opentelemetry")),
+        ("prometheus", ("prometheus", "scrape")),
+        ("notification", ("notification", "router", "policy")),
+        ("runtime_service", ("runtime", "compose", "service", "host", "k3s")),
+        ("smoke", ("smoke", "health", "readiness")),
+        ("drift_guard", ("drift", "recurrence")),
+        ("ai_agent", ("ai_", "agent", "openclaw")),
+    ):
+        if any(needle in haystack for needle in needles):
+            return domain
+    return "monitoring_general"
+
+
+def _monitoring_batch_tag_hints(
+    domain: str,
+    items: list[dict[str, Any]],
+) -> dict[str, Any]:
+    config_kinds = sorted({str(item.get("config_kind") or "unknown") for item in items})
+    expected_scope_prefixes = sorted(
+        {str(item.get("expected_scope") or "unknown").split("_", 1)[0] for item in items}
+    )
+    return {
+        "project_id": DEFAULT_PROJECT_ID,
+        "product_id": _monitoring_product_hint(domain, config_kinds),
+        "website_id": "awoooi.wooo.work",
+        "service_name": _monitoring_service_hint(domain),
+        "package_name": _monitoring_package_hint(domain),
+        "tool_id": domain,
+        "log_source": "monitoring_live_receipt_metadata",
+        "alertname": "monitoring_surface_live_receipt_gap",
+        "playbook_id": f"playbook://awoooi/monitoring-live-receipt/{domain}",
+        "rag_context_ref": f"rag://awoooi/monitoring/{domain}",
+        "mcp_evidence_ref": f"mcp://awoooi/monitoring/{domain}",
+        "schedule_id": f"schedule://awoooi/monitoring-live-receipt/{domain}",
+        "config_kinds": config_kinds,
+        "expected_scope_prefixes": expected_scope_prefixes,
+    }
+
+
+def _monitoring_product_hint(domain: str, config_kinds: list[str]) -> str:
+    if domain in {"telegram", "notification", "alertmanager"}:
+        return "awooop"
+    if any("langfuse" in kind for kind in config_kinds):
+        return "ai-observability"
+    if domain in {"sentry", "signoz", "grafana", "prometheus", "otel"}:
+        return "awoooi-observability"
+    return "awoooi"
+
+
+def _monitoring_service_hint(domain: str) -> str:
+    return {
+        "telegram": "telegram-gateway",
+        "notification": "notification-router",
+        "alertmanager": "alertmanager-webhook",
+        "prometheus": "prometheus",
+        "grafana": "grafana",
+        "signoz": "signoz",
+        "sentry": "sentry",
+        "langfuse": "langfuse",
+        "otel": "otel-collector",
+        "runtime_service": "runtime-service-monitoring",
+        "smoke": "smoke-verifier",
+        "drift_guard": "recurrence-guard",
+        "ai_agent": "ai-agent-loop",
+    }.get(domain, "monitoring")
+
+
+def _monitoring_package_hint(domain: str) -> str:
+    return {
+        "prometheus": "prometheus-config",
+        "grafana": "grafana-dashboard",
+        "signoz": "signoz-rules",
+        "sentry": "sentry-runtime",
+        "langfuse": "langfuse-runtime",
+        "otel": "otel-collector-config",
+        "telegram": "telegram-alert-delivery",
+        "notification": "notification-policy",
+        "alertmanager": "alertmanager-receiver",
+    }.get(domain, "monitoring-receipt")
+
+
 def _alert_receipt_pipeline(
     *,
     monitoring_surface_count: int,
@@ -730,8 +903,43 @@ def _ai_controlled_gap_queue(
     monitoring_asset_rollups: Mapping[str, Any],
 ) -> list[dict[str, Any]]:
     gap_samples = _list_of_dicts(monitoring_asset_rollups.get("live_receipt_gap_samples"))
+    work_batches = _list_of_dicts(monitoring_asset_rollups.get("live_receipt_work_batches"))
     queue: list[dict[str, Any]] = []
     for index, blocker in enumerate(active_blockers, start=1):
+        if blocker.startswith("monitoring_live_receipt_gap") and work_batches:
+            for batch_index, batch in enumerate(work_batches, start=1):
+                queue.append({
+                    "work_item_id": f"CIR-P0-TG-001-LR-{batch_index:02d}",
+                    "parent_work_item_id": "CIR-P0-TG-001",
+                    "priority": str(batch.get("priority") or "P1"),
+                    "blocker": blocker,
+                    "status": "queued_ai_controlled_apply",
+                    "owner_review_required_for_low_medium_high": False,
+                    "critical_break_glass_required": True,
+                    "batch_id": str(batch.get("batch_id") or ""),
+                    "domain": str(batch.get("domain") or "monitoring_general"),
+                    "surface_count": _int(batch.get("surface_count")),
+                    "write_capable_surface_count": _int(
+                        batch.get("write_capable_surface_count")
+                    ),
+                    "tag_hints": _dict(batch.get("tag_hints")),
+                    "target_selector": _dict(batch.get("target_selector")),
+                    "controlled_next_action": str(
+                        batch.get("controlled_next_action")
+                        or "ingest_metadata_only_live_receipt_batch_then_verify_gap_delta"
+                    ),
+                    "post_verifier": (
+                        "/api/v1/agents/telegram-alert-monitoring-coverage-readback"
+                    ),
+                    "sample_surfaces": _list_of_dicts(batch.get("sample_surfaces")),
+                    "operation_boundaries": {
+                        "requires_secret": False,
+                        "requires_runtime_send": False,
+                        "stores_raw_payload": False,
+                        "destructive_action": False,
+                    },
+                })
+            continue
         queue.append({
             "work_item_id": f"CIR-P0-TG-001-{index:02d}",
             "parent_work_item_id": "CIR-P0-TG-001",