fix(awooop): expose telegram automation coverage gaps

apps/api/src/services/ai_agent_report_truth_actionability_review.py

@@ -79,6 +79,13 @@ def _with_live_telegram_egress_guard(payload: dict[str, Any]) -> dict[str, Any]:
     summary = guard.get("summary") if isinstance(guard.get("summary"), dict) else {}
     live_files = sorted({str(item.get("path")) for item in live_calls if item.get("path")})
     new_bypass_count = int(summary.get("new_bypass_count") or 0)
+    surface_counts = _telegram_direct_surface_counts(live_calls)
+    receipt_missing_count = sum(
+        1 for item in live_calls if item.get("receipt_state") != "awooop_outbound_receipt_recorded"
+    )
+    ai_route_missing_count = sum(
+        1 for item in live_calls if item.get("ai_controlled_route_state") != "ai_controlled_gateway_route_ready"
+    )
 
     updated["telegram_egress_guard"] = {
         "schema_version": "telegram_notification_egress_no_new_bypass_guard_v1",
@@ -94,6 +101,45 @@ def _with_live_telegram_egress_guard(payload: dict[str, Any]) -> dict[str, Any]:
             "live_scan_matches_committed_guard_count": (
                 len(live_calls) == int(summary.get("current_direct_bot_api_call_count") or -1)
             ),
+            "gitea_workflow_direct_bot_api_call_count": surface_counts.get(
+                "gitea_workflow_direct_bot_api", 0
+            ),
+            "ops_script_direct_bot_api_call_count": surface_counts.get(
+                "ops_script_direct_bot_api", 0
+            ),
+            "ci_script_direct_bot_api_call_count": surface_counts.get("ci_script_direct_bot_api", 0),
+            "api_direct_bot_api_call_count": surface_counts.get("api_direct_bot_api", 0),
+            "direct_bot_api_awooop_db_receipt_count": len(live_calls) - receipt_missing_count,
+            "direct_bot_api_awooop_db_receipt_missing_count": receipt_missing_count,
+            "direct_bot_api_ai_controlled_route_count": len(live_calls) - ai_route_missing_count,
+            "direct_bot_api_ai_controlled_route_missing_count": ai_route_missing_count,
+        },
+        "telegram_receipt_coverage": {
+            "coverage_status": "partial_gateway_receipts_direct_routes_missing"
+            if live_calls
+            else "complete_no_direct_routes_detected",
+            "gateway_alert_notification_db_receipt_enabled": True,
+            "gateway_alert_notification_receipt_source": "TelegramGateway.send_alert_notification",
+            "direct_bot_api_total_count": len(live_calls),
+            "direct_bot_api_awooop_db_receipt_count": len(live_calls) - receipt_missing_count,
+            "direct_bot_api_awooop_db_receipt_missing_count": receipt_missing_count,
+            "direct_bot_api_ai_controlled_route_count": len(live_calls) - ai_route_missing_count,
+            "direct_bot_api_ai_controlled_route_missing_count": ai_route_missing_count,
+            "all_direct_routes_have_awooop_db_receipt": receipt_missing_count == 0,
+            "all_direct_routes_have_ai_controlled_route": ai_route_missing_count == 0,
+            "all_telegram_alerts_have_db_or_log_receipts": receipt_missing_count == 0,
+            "all_telegram_alerts_ai_controlled": ai_route_missing_count == 0,
+            "evidence_persistence_states": sorted(
+                {
+                    str(item.get("evidence_persistence_state"))
+                    for item in live_calls
+                    if item.get("evidence_persistence_state")
+                }
+            ),
+            "ai_agent_next_action": (
+                "create_controlled_migration_work_items_for_direct_bot_api_routes_with_db_receipt_"
+                "verifier_and_km_playbook_writeback"
+            ),
         },
         "current_direct_bot_api_calls": live_calls,
         "ai_agent_interpretation": {
@@ -101,6 +147,8 @@ def _with_live_telegram_egress_guard(payload: dict[str, Any]) -> dict[str, Any]:
             "existing_direct_routes_need_migration": len(live_calls) > 0,
             "db_receipt_required": True,
             "ai_controlled_route_required": True,
+            "all_direct_routes_have_awooop_db_receipt": receipt_missing_count == 0,
+            "all_direct_routes_have_ai_controlled_route": ai_route_missing_count == 0,
             "manual_default_outcome_allowed": False,
             "next_action": "migrate_direct_bot_api_calls_to_awoooi_api_or_telegram_gateway_receipts",
         },
@@ -146,6 +194,38 @@ def _sanitize_telegram_egress_excerpt(line: str) -> str:
     return excerpt[:180]
 
 
+def _telegram_direct_surface_kind(relative_path: str) -> str:
+    if relative_path.startswith(".gitea/workflows/"):
+        return "gitea_workflow_direct_bot_api"
+    if relative_path.startswith("scripts/ops/"):
+        return "ops_script_direct_bot_api"
+    if relative_path.startswith("scripts/ci/"):
+        return "ci_script_direct_bot_api"
+    if relative_path.startswith("apps/api/src/"):
+        return "api_direct_bot_api"
+    return "unknown_direct_bot_api"
+
+
+def _telegram_direct_persistence_state(surface_kind: str) -> str:
+    if surface_kind == "gitea_workflow_direct_bot_api":
+        return "gitea_actions_log_only_no_awooop_db_receipt"
+    if surface_kind == "ops_script_direct_bot_api":
+        return "host_script_stdout_only_no_awooop_db_receipt"
+    if surface_kind == "ci_script_direct_bot_api":
+        return "ci_script_log_only_no_awooop_db_receipt"
+    if surface_kind == "api_direct_bot_api":
+        return "api_direct_send_no_awooop_outbound_receipt_contract"
+    return "direct_send_no_awooop_outbound_receipt_contract"
+
+
+def _telegram_direct_surface_counts(calls: list[dict[str, Any]]) -> dict[str, int]:
+    counts: dict[str, int] = {}
+    for item in calls:
+        surface_kind = str(item.get("surface_kind") or "unknown_direct_bot_api")
+        counts[surface_kind] = counts.get(surface_kind, 0) + 1
+    return counts
+
+
 def _scan_current_direct_telegram_endpoints(root: Path) -> list[dict[str, Any]]:
     findings: list[dict[str, Any]] = []
     for scan_root in _telegram_egress_scan_roots(root):
@@ -158,13 +238,19 @@ def _scan_current_direct_telegram_endpoints(root: Path) -> list[dict[str, Any]]:
             text = path.read_text(encoding="utf-8", errors="replace")
             for line_number, line in enumerate(text.splitlines(), start=1):
                 for match in _BOT_ENDPOINT_RE.finditer(line):
+                    surface_kind = _telegram_direct_surface_kind(relative_path)
                     findings.append(
                         {
                             "path": relative_path,
                             "line": line_number,
                             "method": match.group("method"),
+                            "surface_kind": surface_kind,
                             "sanitized_excerpt": _sanitize_telegram_egress_excerpt(line),
                             "receipt_state": "missing_awooop_outbound_receipt_contract",
+                            "evidence_persistence_state": _telegram_direct_persistence_state(
+                                surface_kind
+                            ),
+                            "ai_controlled_route_state": "missing_ai_controlled_gateway_route",
                         }
                     )
     return findings
@@ -174,6 +260,7 @@ def _telegram_direct_call_route_finding(item: dict[str, Any]) -> dict[str, str]:
     path = str(item.get("path") or "unknown")
     line = int(item.get("line") or 0)
     method = str(item.get("method") or "sendMessage")
+    surface_kind = str(item.get("surface_kind") or "unknown_direct_bot_api")
     route_key = re.sub(r"[^a-z0-9]+", "_", f"{path}_{line}_{method}".lower()).strip("_")
     return {
         "route_id": f"telegram_direct_bot_api_{route_key}",
@@ -182,7 +269,7 @@ def _telegram_direct_call_route_finding(item: dict[str, Any]) -> dict[str, str]:
         "current_state": "direct_send_path_present",
         "target_state": "gateway_db_receipt_required",
         "risk": (
-            "這條 Telegram 直送路徑不保證寫入 awooop_outbound_message；"
+            f"{surface_kind} 不保證寫入 awooop_outbound_message 或 AI controlled route；"
             f"sanitized excerpt: {item.get('sanitized_excerpt', '')}"
         ),
         "required_fix": (