fix(telegram): route channel hub interim via gateway

apps/api/src/services/awoooi_priority_work_order_readback.py

@@ -335,11 +335,12 @@ _COMMANDER_INSERTED_REQUIREMENT_WORK_ITEMS: list[dict[str, Any]] = [
             "post verifier 與 learning writeback；direct send 或 manual default 都列缺口。"
         ),
         "current_state": (
-            "Telegram alert AI automation matrix endpoint 已實作；目前明確讀回 workflow direct send 0、"
-            "ops script direct gap 4、API direct gap 1，下一步收斂 direct Bot API 旁路。"
+            "Telegram alert AI automation matrix endpoint 已實作；channel_hub API direct sender 已改走 "
+            "TelegramGateway / AwoooP outbound receipt path；目前明確讀回 workflow direct send 0、"
+            "ops script direct gap 4、API direct gap 0。"
         ),
         "acceptance": "Telegram alert matrix 顯示 total、DB receipt、AI route、controlled queue、manual/default gap 與 direct send gap。",
-        "next_action": "先把 apps/api/src/services/channel_hub.py direct sender 改走 TelegramGateway / AwoooP outbound receipt path。",
+        "next_action": "收斂 scripts/ops 內剩餘 4 條 direct Bot API sendMessage 到 AWOOI API / TelegramGateway receipt path。",
         "mapped_workplan_id": "P0-006",
     },
     {

apps/api/src/services/channel_hub.py

@@ -1122,26 +1122,16 @@ async def _send_telegram_interim(
     content: str,
     run_id: UUID,
 ) -> None:
-    """實際發送 Telegram interim 訊息（non-shadow 專用）"""
+    """透過 TelegramGateway 發送 interim 訊息（non-shadow 專用）。"""
     try:
-        import os
+        from src.services.telegram_gateway import get_telegram_gateway
 
-        import httpx
-
-        bot_token = os.environ.get("TELEGRAM_BOT_TOKEN")
-        if not bot_token:
-            logger.warning("interim_telegram_no_token", run_id=str(run_id))
-            return
-
-        async with httpx.AsyncClient(timeout=10) as client:
-            await client.post(
-                f"https://api.telegram.org/bot{bot_token}/sendMessage",
-                json={
-                    "chat_id": channel_chat_id,
-                    "text": content,
-                    "parse_mode": "HTML",
-                },
-            )
+        await get_telegram_gateway().send_text(
+            content,
+            chat_id=channel_chat_id,
+            parse_mode="HTML",
+            disable_web_page_preview=True,
+        )
     except Exception as exc:
         logger.warning(
             "interim_telegram_send_failed",

apps/api/src/services/telegram_alert_ai_automation_matrix.py

@@ -84,6 +84,8 @@ def load_latest_telegram_alert_ai_automation_matrix(
 
     direct_gap_count = int(egress_summary["direct_bot_api_call_count"])
     direct_gap_file_count = int(egress_summary["direct_bot_api_file_count"])
+    api_direct_gap_count = int(egress_summary["api_direct_bot_api_call_count"])
+    ops_script_gap_count = int(egress_summary["ops_script_direct_bot_api_call_count"])
     direct_gap_refs = [
         {
             "surface_id": item["egress_surface_id"],
@@ -189,6 +191,12 @@ def load_latest_telegram_alert_ai_automation_matrix(
         },
     ]
 
+    next_priority_action = _next_priority_action(
+        api_direct_gap_count=api_direct_gap_count,
+        ops_script_gap_count=ops_script_gap_count,
+        direct_gap_count=direct_gap_count,
+    )
+    next_controlled_actions = _ordered_controlled_actions(next_priority_action)
     summary = _build_summary(
         matrix=matrix,
         direct_gap_count=direct_gap_count,
@@ -199,6 +207,7 @@ def load_latest_telegram_alert_ai_automation_matrix(
         readability_summary=readability_summary,
         digest_rollups=digest_rollups,
         source_proof=source_proof,
+        next_priority_action=next_priority_action,
     )
 
     return {
@@ -232,43 +241,57 @@ def load_latest_telegram_alert_ai_automation_matrix(
         "summary": summary,
         "matrix": matrix,
         "direct_send_gap_refs": direct_gap_refs,
-        "next_controlled_actions": [
-            {
-                "action_id": "migrate_api_direct_sender_to_gateway",
-                "scope": "apps/api/src/services/channel_hub.py",
-                "priority": "P0",
-                "why": "API direct Bot API path is the remaining high-blast-radius bypass.",
-                "requires_secret": False,
-                "requires_runtime_send": False,
-                "post_verifier": "telegram-notification-egress-inventory direct gap count decreases and TelegramGateway mirror tests pass.",
-            },
-            {
-                "action_id": "migrate_ops_script_direct_senders_to_receipt_path",
-                "scope": "scripts/ops direct Bot API sendMessage surfaces",
-                "priority": "P0",
-                "why": "Ops script direct sends can bypass DB/log receipt and AI automation card metadata.",
-                "requires_secret": False,
-                "requires_runtime_send": False,
-                "post_verifier": "egress inventory and owner acceptance matrix show no unreviewed direct sends.",
-            },
-            {
-                "action_id": "bind_ai_alert_cards_to_learning_writeback",
-                "scope": "ai_automation_alert_card_v1 outbound mirror metadata",
-                "priority": "P0",
-                "why": "Telegram must become an AI automation learning input, not a manual terminal state.",
-                "requires_secret": False,
-                "requires_runtime_send": False,
-                "post_verifier": "/api/v1/platform/runs/ai-alert-cards readback shows delivery_receipt_readback_required.",
-            },
-        ],
+        "next_controlled_actions": next_controlled_actions,
         "operator_interpretation": [
             "TelegramGateway AI alert cards already expose AI route, controlled queue, final-exit formatter, outbound mirror metadata, and readback endpoint support.",
-            "Known direct Bot API surfaces remain the concrete AI automation gap; they must not be treated as completed delivery receipts.",
+            "Known direct Bot API surfaces remain the concrete AI automation gap; API direct senders are clear when api_direct_send_gap_count is 0.",
             "Manual/default semantics are only acceptable for critical/break-glass or historical evidence; direct send gaps remain controlled migration candidates.",
         ],
     }
 
 
+def _ordered_controlled_actions(next_priority_action: dict[str, Any]) -> list[dict[str, Any]]:
+    actions = [
+        next_priority_action,
+        {
+            "action_id": "migrate_api_direct_sender_to_gateway",
+            "scope": "apps/api/src/services/channel_hub.py",
+            "priority": "P0",
+            "why": "API direct Bot API path is the remaining high-blast-radius bypass.",
+            "requires_secret": False,
+            "requires_runtime_send": False,
+            "post_verifier": "telegram-notification-egress-inventory direct gap count decreases and TelegramGateway mirror tests pass.",
+        },
+        {
+            "action_id": "migrate_ops_script_direct_senders_to_receipt_path",
+            "scope": "scripts/ops direct Bot API sendMessage surfaces",
+            "priority": "P0",
+            "why": "Ops script direct sends can bypass DB/log receipt and AI automation card metadata.",
+            "requires_secret": False,
+            "requires_runtime_send": False,
+            "post_verifier": "egress inventory and owner acceptance matrix show no unreviewed direct sends.",
+        },
+        {
+            "action_id": "bind_ai_alert_cards_to_learning_writeback",
+            "scope": "ai_automation_alert_card_v1 outbound mirror metadata",
+            "priority": "P0",
+            "why": "Telegram must become an AI automation learning input, not a manual terminal state.",
+            "requires_secret": False,
+            "requires_runtime_send": False,
+            "post_verifier": "/api/v1/platform/runs/ai-alert-cards readback shows delivery_receipt_readback_required.",
+        },
+    ]
+    ordered: list[dict[str, Any]] = []
+    seen: set[str] = set()
+    for action in actions:
+        action_id = str(action["action_id"])
+        if action_id in seen:
+            continue
+        seen.add(action_id)
+        ordered.append(action)
+    return ordered
+
+
 def _load_required_json(path: Path) -> dict[str, Any]:
     if not path.exists():
         raise FileNotFoundError(f"required Telegram matrix source missing: {path}")
@@ -366,6 +389,53 @@ def _inspect_source_contract(repo_root: Path) -> dict[str, int | bool]:
     return result
 
 
+def _next_priority_action(
+    *,
+    api_direct_gap_count: int,
+    ops_script_gap_count: int,
+    direct_gap_count: int,
+) -> dict[str, Any]:
+    if api_direct_gap_count > 0:
+        return {
+            "action_id": "migrate_api_direct_sender_to_gateway",
+            "scope": "apps/api/src/services/channel_hub.py",
+            "priority": "P0",
+            "why": "API direct Bot API path is the remaining high-blast-radius bypass.",
+            "requires_secret": False,
+            "requires_runtime_send": False,
+            "post_verifier": "telegram-notification-egress-inventory direct gap count decreases and TelegramGateway mirror tests pass.",
+        }
+    if ops_script_gap_count > 0:
+        return {
+            "action_id": "migrate_ops_script_direct_senders_to_receipt_path",
+            "scope": "scripts/ops direct Bot API sendMessage surfaces",
+            "priority": "P0",
+            "why": "Only ops script direct Bot API paths remain after API sender convergence.",
+            "requires_secret": False,
+            "requires_runtime_send": False,
+            "post_verifier": "telegram-notification-egress-inventory reports ops=0 and owner acceptance candidates=0 for direct Bot API surfaces.",
+        }
+    if direct_gap_count > 0:
+        return {
+            "action_id": "migrate_remaining_direct_senders_to_receipt_path",
+            "scope": "remaining direct Bot API sendMessage surfaces",
+            "priority": "P0",
+            "why": "Direct Bot API paths still bypass the controlled receipt path.",
+            "requires_secret": False,
+            "requires_runtime_send": False,
+            "post_verifier": "telegram-notification-egress-inventory direct gap count reaches 0.",
+        }
+    return {
+        "action_id": "bind_ai_alert_cards_to_learning_writeback",
+        "scope": "ai_automation_alert_card_v1 outbound mirror metadata",
+        "priority": "P0",
+        "why": "All direct Bot API gaps are clear; close the learning/writeback loop.",
+        "requires_secret": False,
+        "requires_runtime_send": False,
+        "post_verifier": "/api/v1/platform/runs/ai-alert-cards readback shows delivery_receipt_readback_required and learning writeback refs.",
+    }
+
+
 def _build_summary(
     *,
     matrix: list[dict[str, Any]],
@@ -377,6 +447,7 @@ def _build_summary(
     readability_summary: dict[str, Any],
     digest_rollups: dict[str, Any],
     source_proof: dict[str, int | bool],
+    next_priority_action: dict[str, Any],
 ) -> dict[str, Any]:
     ready = lambda key, prefix: sum(1 for item in matrix if str(item[key]).startswith(prefix))
     return {
@@ -409,6 +480,6 @@ def _build_summary(
         "ai_alert_card_delivery_readback_endpoint_count": source_proof[
             "ai_alert_card_delivery_readback_endpoint_count"
         ],
-        "next_priority_action_id": "migrate_api_direct_sender_to_gateway",
+        "next_priority_action_id": next_priority_action["action_id"],
         "next_priority_work_item_id": "CIR-P0-TG-001",
     }