diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py
index 22aa60f1..6aa7bc1b 100644
--- a/apps/api/src/services/telegram_gateway.py
+++ b/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:{flow}\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 (
+ "🧭 流程進度\n"
+ f"├ 收件:received | 診斷:{html.escape(diagnose_state)}\n"
+ f"├ 匹配:{html.escape(str(match_state)[:60])} | "
+ f"執行:{html.escape(execute_state)}\n"
+ f"├ 驗證:{html.escape(verify_state)} | "
+ f"KM:{km_entries} | MCP:{gateway_total}\n"
+ f"└ 判定:{html.escape(verdict)} — {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:{html.escape(self.playbook_name)}\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"🧭 處置狀態:{safe_automation_summary}\n"
f"\n"
+ f"{flow_progress_block}\n"
f"{automation_block}"
f"\n"
f"🧠 AI 深度診斷\n"
@@ -816,6 +905,7 @@ class TelegramMessage:
playbook_line = ""
if self.playbook_name:
playbook_line = f"📖 {html.escape(self.playbook_name)}\n"
+ flow_progress_block = self._format_flow_progress_block()
# 組裝訊息
message = (
@@ -823,6 +913,7 @@ class TelegramMessage:
f"{safe_resource}\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 啟用,使用雙軌格式
diff --git a/apps/api/tests/test_telegram_ai_automation_block.py b/apps/api/tests/test_telegram_ai_automation_block.py
index 5ac5b24c..8cb9cd1b 100644
--- a/apps/api/tests/test_telegram_ai_automation_block.py
+++ b/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 "執行:no_action_or_observe" 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 "執行:auto_repair_recorded:1" in body
+ assert "驗證:success" in body
+ assert "KM:2" in body
+ assert "MCP:5" in body
+ assert "已驗證自動修復" in body
+ assert "已驗證自動修復完成" in body
+ assert "等待人工批准" not in body