diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py
index 077fd1872..d81bf2de9 100644
--- a/apps/api/src/services/telegram_gateway.py
+++ b/apps/api/src/services/telegram_gateway.py
@@ -1009,12 +1009,122 @@ def _format_awooop_status_chain_lines(
if outcome_lines:
lines.extend(["", *outcome_lines])
+ lines.extend(
+ _format_awooop_asset_deposition_lines(
+ truth_chain=truth_chain,
+ remediation_history=remediation_history,
+ )
+ )
+
blockers = [str(item) for item in [*truth_blockers, *quality_blockers] if item]
if blockers:
lines.append("卡點: " + html.escape(", ".join(blockers[:4])))
return lines
+def _asset_value(value: int) -> str:
+ return str(value) if value > 0 else "--"
+
+
+def _asset_tone(value: int, *, blocked: bool = False) -> str:
+ if value > 0:
+ return "ready"
+ return "blocked" if blocked else "pending"
+
+
+def _format_awooop_asset_deposition_lines(
+ *,
+ truth_chain: dict[str, object] | None = None,
+ automation_quality: dict[str, object] | None = None,
+ remediation_history: dict[str, object] | None = None,
+ playbook_name: str = "",
+) -> list[str]:
+ """Summarize KM / PlayBook / script / schedule / verifier visibility for TG cards."""
+ quality = (
+ automation_quality
+ if isinstance(automation_quality, dict)
+ else (
+ truth_chain.get("automation_quality")
+ if isinstance(truth_chain, dict)
+ and isinstance(truth_chain.get("automation_quality"), dict)
+ else {}
+ )
+ )
+ facts = quality.get("facts") if isinstance(quality.get("facts"), dict) else {}
+ execution = _callback_reply_awooop_execution_snapshot(truth_chain)
+ ansible = execution.get("ansible") if isinstance(execution.get("ansible"), dict) else {}
+ channel = truth_chain.get("channel") if isinstance(truth_chain, dict) else {}
+ if not isinstance(channel, dict):
+ channel = {}
+
+ inbound_events = channel.get("inbound_events") if isinstance(channel.get("inbound_events"), list) else []
+ outbound_messages = channel.get("outbound_messages") if isinstance(channel.get("outbound_messages"), list) else []
+ km_count = _safe_int(facts.get("knowledge_entries"))
+ playbook_count = (
+ len(execution.get("playbook_ids") or [])
+ + len(execution.get("playbook_paths") or [])
+ + (1 if playbook_name else 0)
+ + (1 if ansible.get("latest_catalog_id") or ansible.get("latest_playbook_path") else 0)
+ )
+ script_count = (
+ _safe_int(ansible.get("candidate_count"))
+ + _safe_int(ansible.get("check_mode_total"))
+ + _safe_int(ansible.get("apply_total"))
+ )
+ schedule_count = len(inbound_events) + len(outbound_messages)
+ remediation_state = _remediation_evidence_state(remediation_history)
+ verifier = str(
+ facts.get("verification_result")
+ or _latest_remediation_history_item(remediation_history).get("verification_result_preview")
+ or "missing"
+ )
+ verifier_lower = verifier.lower()
+ verifier_ready = any(token in verifier_lower for token in ("healthy", "verified", "success", "passed"))
+ verifier_blocked = any(token in verifier_lower for token in ("degraded", "failed", "error", "timeout"))
+ quality_blockers = quality.get("blockers")
+ blocker_items = quality_blockers if isinstance(quality_blockers, list) else []
+ no_repair_candidate = "candidate" in " ".join(str(item) for item in blocker_items if item).lower()
+ no_repair_candidate = no_repair_candidate or remediation_state in {"blocked", "fetch_failed"}
+
+ asset_states = [
+ _asset_tone(km_count, blocked=bool(facts) and km_count <= 0),
+ _asset_tone(playbook_count, blocked=no_repair_candidate),
+ _asset_tone(script_count, blocked=remediation_state in {"missing", "fetch_failed"}),
+ _asset_tone(schedule_count),
+ "ready" if verifier_ready else "blocked" if verifier_blocked else "pending",
+ ]
+ ready_count = sum(1 for item in asset_states if item == "ready")
+ blocked_count = sum(1 for item in asset_states if item == "blocked")
+ verifier_value = "ready" if verifier_ready else "blocked" if verifier_blocked else "pending"
+
+ return [
+ "",
+ "🧾 自動化資產沉澱",
+ (
+ "├ KM:"
+ f"{html.escape(_asset_value(km_count))} | "
+ "PlayBook:"
+ f"{html.escape(_asset_value(playbook_count))}"
+ ),
+ (
+ "├ 腳本/Ansible:"
+ f"{html.escape(_asset_value(script_count))} | "
+ "排程/來源:"
+ f"{html.escape(_asset_value(schedule_count))}"
+ ),
+ (
+ "├ Verifier:"
+ f"{html.escape(verifier_value)} "
+ f"({html.escape(verifier[:80])})"
+ ),
+ (
+ "└ 判讀:"
+ f"完成 {ready_count} / 卡點 {blocked_count};"
+ "只讀顯示,不代表已授權執行"
+ ),
+ ]
+
+
def _first_mapping_item(values: object) -> dict[str, object]:
if not isinstance(values, list):
return {}
@@ -2588,6 +2698,14 @@ class TelegramMessage:
def _format_remediation_evidence_block(self) -> str:
return _format_remediation_evidence_block(self.remediation_summary)
+ def _format_asset_deposition_block(self) -> str:
+ lines = _format_awooop_asset_deposition_lines(
+ automation_quality=self.automation_quality,
+ remediation_history=self.remediation_summary,
+ playbook_name=self.playbook_name,
+ )
+ return "\n".join(lines) + "\n" if lines else ""
+
def format(self) -> str:
"""
格式化為 SOUL.md 規範的訊息 (含 AI 仲裁 + SignOz)
@@ -2737,6 +2855,7 @@ class TelegramMessage:
playbook_line = f"📖 Playbook:{html.escape(self.playbook_name)}\n"
remediation_evidence_block = self._format_remediation_evidence_block()
flow_progress_block = self._format_flow_progress_block()
+ asset_deposition_block = self._format_asset_deposition_block()
operator_outcome_block = self._format_operator_outcome_block()
automation_block = self._format_automation_block()
manual_handoff_block = self._format_manual_handoff_package_block()
@@ -2751,6 +2870,7 @@ class TelegramMessage:
f"🧭 處置狀態:{safe_automation_summary}\n"
f"{remediation_evidence_block}\n"
f"{flow_progress_block}\n"
+ f"{asset_deposition_block}"
f"{operator_outcome_block}"
f"{automation_block}"
f"{manual_handoff_block}"
diff --git a/apps/api/tests/test_telegram_message_templates.py b/apps/api/tests/test_telegram_message_templates.py
index 0ece7672c..6472fcb5d 100644
--- a/apps/api/tests/test_telegram_message_templates.py
+++ b/apps/api/tests/test_telegram_message_templates.py
@@ -374,6 +374,9 @@ def test_awooop_status_chain_lines_show_verified_auto_repair_stage() -> None:
assert "monitor_for_regression" in joined
assert "處置結論" in joined
assert "已驗證自動修復完成" in joined
+ assert "自動化資產沉澱" in joined
+ assert "KM:2" in joined
+ assert "Verifier:ready" in joined
def test_awooop_status_chain_lines_show_read_only_manual_gate() -> None:
@@ -419,6 +422,8 @@ def test_awooop_status_chain_lines_show_read_only_manual_gate() -> None:
assert "人工: yes" in joined
assert "approve_or_escalate_from_awooop" in joined
assert "pending_human_approval" in joined
+ assert "自動化資產沉澱" in joined
+ assert "PlayBook:--" in joined
def test_awooop_status_chain_lines_do_not_treat_audit_ops_as_repair() -> None:
@@ -1392,6 +1397,12 @@ class TestTelegramMessageFormat:
assert "OpenClaw" in result
assert "NemoTron" in result
assert "ElephantAlpha" in result
+ assert "自動化資產沉澱" in result
+ assert "KM" in result
+ assert "PlayBook" in result
+ assert "腳本/Ansible" in result
+ assert "排程/來源" in result
+ assert "Verifier" in result
assert len(result) <= 4096 # Telegram HTML message limit
def test_telegram_message_ai_proposal_marks_approval_wait(self):
@@ -1511,6 +1522,51 @@ class TestTelegramMessageFormat:
assert "auto_repair_executor/ssh_diagnose/read" in result
assert "incident false" in result
assert "auto-repair false" in result
+ assert "自動化資產沉澱" in result
+ assert "Verifier:blocked" in result
+
+ def test_telegram_message_asset_deposition_shows_km_playbook_verifier(self):
+ """主告警卡必須把 KM / PlayBook / Verifier 沉澱結果前移到可掃描區。"""
+ msg = TelegramMessage(
+ status_emoji="⚠️",
+ risk_level="MEDIUM",
+ resource_name="awoooi-api",
+ root_cause="已完成診斷並產生修復候選",
+ suggested_action="等待人工批准",
+ estimated_downtime="unknown",
+ approval_id="INC-20260618-ASSET",
+ playbook_name="infra/ansible/playbooks/110-devops.yml",
+ automation_quality={
+ "verdict": "approval_required",
+ "facts": {
+ "knowledge_entries": 2,
+ "verification_result": "healthy",
+ "mcp_gateway_total": 7,
+ "auto_repair_execution_records": 0,
+ "automation_operation_records": 0,
+ },
+ "blockers": [],
+ },
+ remediation_summary={
+ "total": 1,
+ "items": [
+ {
+ "agent_id": "investigator",
+ "tool_name": "ssh_diagnose",
+ "required_scope": "read",
+ "verification_result_preview": "healthy",
+ }
+ ],
+ },
+ )
+
+ result = msg.format()
+
+ assert "自動化資產沉澱" in result
+ assert "KM:2" in result
+ assert "PlayBook:1" in result
+ assert "Verifier:ready" in result
+ assert "完成 2 / 卡點 0" in result
def test_telegram_message_surfaces_missing_remediation_evidence(self):
"""沒有補救試跑紀錄時,主卡要明確說明,不讓值班者猜。"""