feat(telegram): show automation flow progress

apps/api/src/services/telegram_gateway.py

@@ -412,6 +412,7 @@ class TelegramMessage:
     alert_category: str = ""   # host/k8s/database/service/external_site/secops 等
     playbook_name: str = ""    # 匹配到的 Playbook 名稱（空字串=規則匹配）
     automation_state: str = ""  # diagnosis_collected_manual_required / diagnosis_failed_manual_required
+    automation_quality: dict | None = None  # truth-chain automation_quality 摘要
 
     # ==========================================================================
     # Phase 22: Nemotron 協作欄位 (ADR-044)
@@ -471,6 +472,23 @@ class TelegramMessage:
         action = (self.suggested_action or "").upper()
         text = f"{self.root_cause} {self.suggested_action}".lower()
         state = (self.automation_state or "").lower()
+        quality = self.automation_quality or {}
+        facts = quality.get("facts") if isinstance(quality.get("facts"), dict) else {}
+        verdict = str(quality.get("verdict") or "")
+        auto_repair_records = int(facts.get("auto_repair_execution_records") or 0)
+        operation_records = int(facts.get("automation_operation_records") or 0)
+        verification = str(facts.get("verification_result") or "missing")
+
+        if verdict == "auto_repaired_verified":
+            return "✅ 已驗證自動修復完成"
+        if auto_repair_records > 0 or operation_records > 0:
+            if verification == "missing":
+                return "🔄 已自動執行，等待驗證證據"
+            return f"🔄 已自動執行，驗證結果：{verification}"
+        if verdict == "approval_required":
+            return "🟡 需要審批後才會執行"
+        if verdict.startswith("manual_required"):
+            return "🟠 未自動修復，需人工判斷"
 
         if state == "diagnosis_collected_manual_required":
             return "🔎 AI 已完成只讀診斷，需人工判斷"
@@ -518,6 +536,75 @@ class TelegramMessage:
             f"└ Flow：<code>{flow}</code>\n"
         )
 
+    def _format_flow_progress_block(self) -> str:
+        """Operator-facing state of where the alert is in the automation loop."""
+        quality = self.automation_quality or {}
+        facts = quality.get("facts") if isinstance(quality.get("facts"), dict) else {}
+        verdict = str(quality.get("verdict") or self._automation_mode())
+
+        action_upper = (self.suggested_action or "").upper()
+        is_noop = (
+            "NO_ACTION" in action_upper
+            or action_upper.startswith("OBSERVE")
+            or action_upper.startswith("INVESTIGATE")
+            or not action_upper.strip()
+            or action_upper == "待分析"
+        )
+        auto_repair_records = int(facts.get("auto_repair_execution_records") or 0)
+        operation_records = int(facts.get("automation_operation_records") or 0)
+        verification = str(facts.get("verification_result") or "missing")
+        gateway_total = int(facts.get("mcp_gateway_total") or 0)
+        km_entries = int(facts.get("knowledge_entries") or 0)
+
+        if self.confidence > 0:
+            diagnose_state = "ai_ready"
+        elif self.automation_state == "diagnosis_failed_manual_required":
+            diagnose_state = "failed"
+        else:
+            diagnose_state = "rule_or_degraded"
+
+        match_state = self.playbook_name or "rule_catalog"
+        if auto_repair_records > 0:
+            execute_state = f"auto_repair_recorded:{auto_repair_records}"
+        elif operation_records > 0:
+            execute_state = f"operation_recorded:{operation_records}"
+        elif is_noop:
+            execute_state = "no_action_or_observe"
+        elif "approval" in verdict or self._automation_mode() == "ai_proposal_ready":
+            execute_state = "awaiting_approval"
+        else:
+            execute_state = "not_started"
+
+        if verification != "missing":
+            verify_state = verification
+        elif auto_repair_records > 0 or operation_records > 0:
+            verify_state = "pending_or_missing"
+        else:
+            verify_state = "not_started"
+
+        if verdict == "auto_repaired_verified":
+            conclusion = "已驗證自動修復"
+        elif auto_repair_records > 0 or operation_records > 0:
+            conclusion = "已記錄執行，等待或缺少驗證"
+        elif is_noop:
+            conclusion = "未自動修復，需人工判斷"
+        elif "approval" in verdict:
+            conclusion = "等待審批後才會執行"
+        elif "manual" in verdict:
+            conclusion = "轉人工處理"
+        else:
+            conclusion = "尚未形成可宣稱自動修復的證據鏈"
+
+        return (
+            "🧭 <b>流程進度</b>\n"
+            f"├ 收件：<code>received</code> | 診斷：<code>{html.escape(diagnose_state)}</code>\n"
+            f"├ 匹配：<code>{html.escape(str(match_state)[:60])}</code> | "
+            f"執行：<code>{html.escape(execute_state)}</code>\n"
+            f"├ 驗證：<code>{html.escape(verify_state)}</code> | "
+            f"KM：<code>{km_entries}</code> | MCP：<code>{gateway_total}</code>\n"
+            f"└ 判定：<code>{html.escape(verdict)}</code> — {html.escape(conclusion)}\n"
+        )
+
     def format(self) -> str:
         """
         格式化為 SOUL.md 規範的訊息 (含 AI 仲裁 + SignOz)
@@ -660,6 +747,7 @@ class TelegramMessage:
         playbook_line = ""
         if self.playbook_name:
             playbook_line = f"📖 Playbook：<code>{html.escape(self.playbook_name)}</code>\n"
+        flow_progress_block = self._format_flow_progress_block()
         automation_block = self._format_automation_block()
 
         # ADR-075 TYPE-3 格式組裝
@@ -671,6 +759,7 @@ class TelegramMessage:
             f"{category_line}"
             f"🧭 處置狀態：<b>{safe_automation_summary}</b>\n"
             f"\n"
+            f"{flow_progress_block}\n"
             f"{automation_block}"
             f"\n"
             f"🧠 <b>AI 深度診斷</b>\n"
@@ -816,6 +905,7 @@ class TelegramMessage:
         playbook_line = ""
         if self.playbook_name:
             playbook_line = f"📖 <code>{html.escape(self.playbook_name)}</code>\n"
+        flow_progress_block = self._format_flow_progress_block()
 
         # 組裝訊息
         message = (
@@ -823,6 +913,7 @@ class TelegramMessage:
             f"<b>{safe_resource}</b>\n"
             f"{category_line}"
             f"\n"
+            f"{flow_progress_block}\n"
             f"{self._format_automation_block()}\n"
             f"{conf_line}\n"
             f"👥 {resp_display}\n"
@@ -2229,6 +2320,29 @@ class TelegramGateway:
                 trace_url=signoz_trace_url,
             )
 
+        automation_quality: dict | None = None
+        if incident_id:
+            try:
+                from src.services.awooop_truth_chain_service import fetch_truth_chain
+
+                truth_chain = await asyncio.wait_for(
+                    fetch_truth_chain(
+                        source_id=incident_id,
+                        project_id="awoooi",
+                    ),
+                    timeout=2.5,
+                )
+                quality = truth_chain.get("automation_quality")
+                if isinstance(quality, dict):
+                    automation_quality = quality
+            except Exception as truth_exc:
+                logger.debug(
+                    "telegram_approval_card_truth_chain_fetch_failed",
+                    approval_id=approval_id,
+                    incident_id=incident_id,
+                    error=str(truth_exc),
+                )
+
         # 建立訊息結構 (含 AI 仲裁 + SignOz + Token/Cost + 頻率統計)
         message = TelegramMessage(
             status_emoji=emoji,
@@ -2266,6 +2380,7 @@ class TelegramGateway:
             alert_category=alert_category,
             playbook_name=playbook_name,
             automation_state=automation_state,
+            automation_quality=automation_quality,
         )
 
         # 格式化訊息 — Phase 22: 如果 Nemotron 啟用，使用雙軌格式

apps/api/tests/test_telegram_ai_automation_block.py

@@ -24,6 +24,8 @@ def test_action_required_card_exposes_ai_automation_on_fallback() -> None:
     assert "NemoTron" in body
     assert "Hermes" in body
     assert "ElephantAlpha" in body
+    assert "流程進度" in body
+    assert "執行：<code>no_action_or_observe</code>" in body
 
 
 def test_nemotron_card_exposes_same_ai_automation_chain() -> None:
@@ -50,3 +52,41 @@ def test_nemotron_card_exposes_same_ai_automation_chain() -> None:
     assert "OpenClaw Nemo" in body
     assert "tool_ready" in body
     assert "restart_deployment" in body
+    assert "流程進度" in body
+
+
+def test_action_required_card_exposes_truth_chain_progress() -> None:
+    message = TelegramMessage(
+        status_emoji="⚠️",
+        risk_level="LOW",
+        resource_name="awoooi-api",
+        root_cause="restart spike",
+        suggested_action="kubectl rollout restart deployment/awoooi-api",
+        estimated_downtime="30s",
+        approval_id="approval-id",
+        incident_id="INC-20260513-TEST03",
+        primary_responsibility="INFRA",
+        confidence=0.91,
+        playbook_name="restart_deployment",
+        automation_quality={
+            "verdict": "auto_repaired_verified",
+            "facts": {
+                "auto_repair_execution_records": 1,
+                "automation_operation_records": 1,
+                "verification_result": "success",
+                "mcp_gateway_total": 5,
+                "knowledge_entries": 2,
+            },
+        },
+    )
+
+    body = message.format()
+
+    assert "流程進度" in body
+    assert "執行：<code>auto_repair_recorded:1</code>" in body
+    assert "驗證：<code>success</code>" in body
+    assert "KM：<code>2</code>" in body
+    assert "MCP：<code>5</code>" in body
+    assert "已驗證自動修復" in body
+    assert "已驗證自動修復完成" in body
+    assert "等待人工批准" not in body