fix(api): 在 Telegram 告警顯示自動化資產沉澱

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 [
+        "",
+        "🧾 <b>自動化資產沉澱</b>",
+        (
+            "├ KM："
+            f"<code>{html.escape(_asset_value(km_count))}</code> | "
+            "PlayBook："
+            f"<code>{html.escape(_asset_value(playbook_count))}</code>"
+        ),
+        (
+            "├ 腳本/Ansible："
+            f"<code>{html.escape(_asset_value(script_count))}</code> | "
+            "排程/來源："
+            f"<code>{html.escape(_asset_value(schedule_count))}</code>"
+        ),
+        (
+            "├ Verifier："
+            f"<code>{html.escape(verifier_value)}</code> "
+            f"(<code>{html.escape(verifier[:80])}</code>)"
+        ),
+        (
+            "└ 判讀："
+            f"<code>完成 {ready_count} / 卡點 {blocked_count}</code>；"
+            "只讀顯示，不代表已授權執行"
+        ),
+    ]
+
+
 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：<code>{html.escape(self.playbook_name)}</code>\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"🧭 處置狀態：<b>{safe_automation_summary}</b>\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}"