feat(telegram): surface awooop agent evidence chain

apps/api/src/services/telegram_gateway.py

@@ -433,6 +433,164 @@ def _format_awooop_status_chain_lines(
     return lines
 
 
+def _first_mapping_item(values: object) -> dict[str, object]:
+    if not isinstance(values, list):
+        return {}
+    for item in values:
+        if isinstance(item, dict):
+            return item
+    return {}
+
+
+def _provider_correlation_summary(source_correlation: dict[str, object] | None) -> str:
+    providers = (
+        source_correlation.get("providers")
+        if isinstance(source_correlation, dict)
+        and isinstance(source_correlation.get("providers"), dict)
+        else {}
+    )
+    parts: list[str] = []
+    for provider in ("sentry", "signoz"):
+        item = providers.get(provider) if isinstance(providers, dict) else {}
+        if not isinstance(item, dict):
+            item = {}
+        parts.append(
+            f"{provider} "
+            f"{_safe_int(item.get('direct_ref_total'))}/"
+            f"{_safe_int(item.get('candidate_total'))}/"
+            f"{_safe_int(item.get('applied_link_total'))}"
+        )
+    return " · ".join(parts)
+
+
+def _format_awooop_agent_evidence_lines(
+    *,
+    truth_chain: dict[str, object] | None = None,
+    remediation_history: dict[str, object] | None = None,
+    source_correlation: dict[str, object] | None = None,
+) -> list[str]:
+    """Render the same agent evidence matrix used by Operator UI for Telegram."""
+    if not truth_chain and not remediation_history and not source_correlation:
+        return []
+
+    quality = (
+        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 {}
+    latest = _latest_remediation_history_item(remediation_history)
+    mcp = _callback_reply_awooop_mcp_snapshot(truth_chain)
+    execution = _callback_reply_awooop_execution_snapshot(truth_chain)
+
+    gateway = mcp.get("gateway") if isinstance(mcp.get("gateway"), dict) else {}
+    legacy = mcp.get("legacy") if isinstance(mcp.get("legacy"), dict) else {}
+    top_tool = _first_mapping_item(mcp.get("top_tools"))
+
+    ansible = (
+        execution.get("ansible")
+        if isinstance(execution.get("ansible"), dict)
+        else {}
+    )
+    candidate_playbook = _first_mapping_item(ansible.get("candidate_playbooks"))
+    playbook_paths = execution.get("playbook_paths")
+    playbook_ids = execution.get("playbook_ids")
+    selected_playbook = ""
+    if isinstance(playbook_paths, list) and playbook_paths:
+        selected_playbook = str(playbook_paths[0] or "")
+    elif isinstance(playbook_ids, list) and playbook_ids:
+        selected_playbook = str(playbook_ids[0] or "")
+    else:
+        selected_playbook = str(
+            ansible.get("latest_playbook_path")
+            or candidate_playbook.get("playbook_path")
+            or candidate_playbook.get("catalog_id")
+            or "--"
+        )
+
+    source_status = "missing"
+    direct_total = 0
+    candidate_total = 0
+    applied_total = 0
+    if isinstance(source_correlation, dict):
+        source_status = str(
+            source_correlation.get("verification_status")
+            or source_correlation.get("status")
+            or "missing"
+        )
+        direct_total = _safe_int(source_correlation.get("direct_ref_total"))
+        candidate_total = _safe_int(source_correlation.get("candidate_total"))
+        applied_total = _safe_int(source_correlation.get("applied_link_total"))
+
+    verification = (
+        facts.get("verification_result")
+        or latest.get("verification_result_preview")
+        or "missing"
+    )
+    auto_repair_records = _safe_int(facts.get("auto_repair_execution_records"))
+    operation_records = _safe_int(facts.get("automation_operation_records"))
+    km_entries = _safe_int(facts.get("knowledge_entries"))
+
+    latest_executor = str(execution.get("latest_executor") or "--")
+    latest_status = str(execution.get("latest_status") or "--")
+    latest_operation = str(execution.get("latest_operation_type") or "--")
+    latest_action = str(execution.get("latest_action") or "--")
+    top_tool_name = str(top_tool.get("tool_name") or "--")
+
+    return [
+        "",
+        "🧩 <b>AI Agent 證據鏈</b>",
+        (
+            "MCP / 自建 MCP: "
+            f"Gateway <code>{_safe_int(gateway.get('success'))}/"
+            f"{_safe_int(gateway.get('total'))}</code>，"
+            f"失敗 <code>{_safe_int(gateway.get('failed'))}</code>，"
+            f"阻擋 <code>{_safe_int(gateway.get('blocked'))}</code>"
+        ),
+        (
+            "MCP top: "
+            f"<code>{html.escape(top_tool_name)}</code> | "
+            f"first-class <code>{_safe_int(gateway.get('first_class_total'))}</code> / "
+            f"legacy <code>{_safe_int(legacy.get('total'))}</code> / "
+            f"policy <code>{_safe_int(gateway.get('policy_enforced_total'))}</code>"
+        ),
+        (
+            "Sentry/SigNoz: "
+            f"<code>{html.escape(source_status)}</code> | "
+            f"direct <code>{direct_total}</code> / "
+            f"candidate <code>{candidate_total}</code> / "
+            f"applied <code>{applied_total}</code>"
+        ),
+        (
+            "Provider: "
+            f"<code>{html.escape(_provider_correlation_summary(source_correlation))}</code>"
+        ),
+        (
+            "Executor: "
+            f"<code>{html.escape(latest_executor)}/{html.escape(latest_status)}</code> | "
+            f"op <code>{html.escape(latest_operation)}</code> / "
+            f"action <code>{html.escape(latest_action)}</code> / "
+            f"ops <code>{_safe_int(execution.get('operation_total'))}</code>"
+        ),
+        (
+            "PlayBook / Ansible: "
+            f"<code>{html.escape(selected_playbook)}</code> | "
+            f"ansible <code>{html.escape(_bool_code(ansible.get('considered'), unknown_when_none=True))}</code> / "
+            f"candidates <code>{_safe_int(ansible.get('candidate_count'))}</code> / "
+            f"check-mode <code>{html.escape(_bool_code(ansible.get('latest_check_mode'), unknown_when_none=True))}</code> / "
+            f"status <code>{html.escape(str(ansible.get('latest_status') or '--'))}</code>"
+        ),
+        (
+            "KM / Learning: "
+            f"KM <code>{km_entries}</code> / "
+            f"AutoRepair <code>{auto_repair_records}</code> / "
+            f"Ops <code>{operation_records}</code> | "
+            f"verify <code>{html.escape(str(verification))}</code>"
+        ),
+    ]
+
+
 def _format_km_stale_completion_lines(summary: dict[str, object] | None) -> list[str]:
     """Render KM owner-review completion state for Telegram detail/history replies."""
     if not summary:
@@ -6545,6 +6703,12 @@ class TelegramGateway:
                         error=str(source_exc),
                     )
 
+            lines += _format_awooop_agent_evidence_lines(
+                truth_chain=truth_chain,
+                remediation_history=remediation_history,
+                source_correlation=source_correlation,
+            )
+
             awooop_status_chain_snapshot = _callback_reply_awooop_status_chain_snapshot(
                 incident_id=incident_id,
                 truth_chain=truth_chain,
@@ -6724,6 +6888,12 @@ class TelegramGateway:
                         error=str(source_exc),
                     )
 
+            lines += _format_awooop_agent_evidence_lines(
+                truth_chain=truth_chain,
+                remediation_history=remediation_history,
+                source_correlation=source_correlation,
+            )
+
             awooop_status_chain_snapshot = _callback_reply_awooop_status_chain_snapshot(
                 incident_id=incident_id,
                 truth_chain=truth_chain,

apps/api/tests/test_telegram_adr050.py

@@ -96,11 +96,13 @@ class TestDetailMessageFormat:
         assert "incident.severity" in self._read_gateway()
 
     def test_detail_includes_truth_chain_gateway_summary(self):
-        """detail 顯示 AwoooP truth-chain / MCP Gateway / automation quality 摘要"""
+        """detail 顯示 AwoooP truth-chain / MCP Gateway / automation quality / agent evidence 摘要"""
         source = self._read_gateway()
         assert "fetch_truth_chain" in source
         assert "_format_gateway_summary_lines" in source
         assert "_format_automation_quality_lines" in source
+        assert "_format_awooop_agent_evidence_lines" in source
+        assert "AI Agent 證據鏈" in source
         assert "MCP Gateway" in source
         assert "自動化品質" in source
 
@@ -140,6 +142,7 @@ class TestHistoryMessageFormat:
         assert "DB Truth-chain" in source
         assert "incident_history_truth_chain_summary_failed" in source
         assert "_format_automation_quality_lines" in source
+        assert "_format_awooop_agent_evidence_lines" in source
 
 
 # =============================================================================

apps/api/tests/test_telegram_message_templates.py

@@ -177,6 +177,128 @@ def test_awooop_status_chain_lines_show_read_only_manual_gate() -> None:
     assert "pending_human_approval" in joined
 
 
+def test_awooop_agent_evidence_lines_show_mcp_source_execution_playbook_km() -> None:
+    """Telegram 詳情/歷史要像前端一樣顯示五段 AI Agent 證據鏈。"""
+    lines = telegram_gateway_module._format_awooop_agent_evidence_lines(
+        truth_chain={
+            "automation_quality": {
+                "verdict": "approval_required",
+                "facts": {
+                    "auto_repair_execution_records": 0,
+                    "automation_operation_records": 1,
+                    "verification_result": "degraded",
+                    "mcp_gateway_total": 2,
+                    "knowledge_entries": 1,
+                },
+            },
+            "mcp": {
+                "awooop_gateway": {
+                    "total": 2,
+                    "success": 1,
+                    "failed": 1,
+                    "blocked": 0,
+                    "first_class_total": 2,
+                    "legacy_bridge_total": 0,
+                    "policy_enforced_total": 2,
+                    "by_tool": [
+                        {
+                            "tool_name": "prometheus.query",
+                            "total": 2,
+                            "success": 1,
+                            "failed": 1,
+                            "blocked": 0,
+                        }
+                    ],
+                },
+                "legacy": {
+                    "total": 1,
+                    "success": 1,
+                    "failed": 0,
+                },
+            },
+            "execution": {
+                "automation_operation_log": [
+                    {
+                        "operation_type": "ansible_candidate_matched",
+                        "status": "dry_run",
+                        "actor": "Hermes",
+                        "input_executor": "ansible",
+                        "input_action": "check_mode",
+                    }
+                ],
+                "ansible": {
+                    "considered": True,
+                    "records": [
+                        {
+                            "operation_type": "ansible_candidate_matched",
+                            "status": "dry_run",
+                            "playbook_path": "infra/ansible/playbooks/188-ai-web.yml",
+                            "check_mode": True,
+                        }
+                    ],
+                    "candidate_catalog": {
+                        "candidates": [
+                            {
+                                "catalog_id": "ansible:188-ai-web",
+                                "playbook_path": "infra/ansible/playbooks/188-ai-web.yml",
+                                "risk_level": "medium",
+                                "match_score": 3,
+                            }
+                        ]
+                    },
+                },
+            },
+        },
+        remediation_history={
+            "total": 1,
+            "items": [
+                {
+                    "verification_result_preview": "degraded",
+                    "writes_incident_state": False,
+                    "writes_auto_repair_result": False,
+                }
+            ],
+        },
+        source_correlation={
+            "status": "candidate_found",
+            "verification_status": "candidate_only",
+            "direct_ref_total": 0,
+            "candidate_total": 1,
+            "applied_link_total": 0,
+            "providers": {
+                "sentry": {
+                    "direct_ref_total": 0,
+                    "candidate_total": 1,
+                    "applied_link_total": 0,
+                },
+                "signoz": {
+                    "direct_ref_total": 0,
+                    "candidate_total": 0,
+                    "applied_link_total": 0,
+                },
+            },
+        },
+    )
+
+    joined = "\n".join(lines)
+    assert "AI Agent 證據鏈" in joined
+    assert "MCP / 自建 MCP" in joined
+    assert "Gateway <code>1/2</code>" in joined
+    assert "prometheus.query" in joined
+    assert "Sentry/SigNoz" in joined
+    assert "candidate <code>1</code>" in joined
+    assert "sentry 0/1/0" in joined
+    assert "Executor: <code>ansible/dry_run</code>" in joined
+    assert "op <code>ansible_candidate_matched</code>" in joined
+    assert "PlayBook / Ansible" in joined
+    assert "infra/ansible/playbooks/188-ai-web.yml" in joined
+    assert "ansible <code>yes</code>" in joined
+    assert "check-mode <code>yes</code>" in joined
+    assert "KM / Learning" in joined
+    assert "KM <code>1</code>" in joined
+    assert "verify <code>degraded</code>" in joined
+
+
 def test_callback_reply_awooop_status_chain_snapshot_marks_manual_gate() -> None:
     """Callback evidence 要保存當下 AwoooP 狀態鏈，不只保存 live query 結果。"""
     snapshot = telegram_gateway_module._callback_reply_awooop_status_chain_snapshot(

docs/LOGBOOK.md

@@ -20156,3 +20156,49 @@ screenshots:
 - KM governance：約 84.5%。
 - AI Provider lane visibility：約 92%。
 - 完整 AI 自動化管理產品化：約 96.1%。
+
+---
+
+## 2026-05-25 T182 — Telegram 詳情 / 歷史補 AI Agent 證據鏈
+
+**背景**：
+
+- T181 已讓前端共用狀態鏈顯示 `MCP / 自建 MCP`、`Sentry / SigNoz`、`Executor`、`PlayBook / Ansible`、`KM / Learning` 五段證據。
+- 但 Telegram `詳情` / `歷史` callback 仍主要顯示 AwoooP 狀態鏈、MCP Gateway 摘要與 automation quality；操作者仍需要進前端才能完整看出是否有使用 MCP、自建 MCP、Sentry / SigNoz 關聯、Ansible check-mode、PlayBook 候選、KM / Learning 寫入。
+- 本輪延續「Telegram 不只告警，要能回答流程跑到哪裡」的要求，不新增假資料、不新增新的資料來源，只把既有 truth-chain / ADR-100 history / source correlation 轉成可讀摘要。
+
+**本輪修正**：
+
+- `telegram_gateway.py` 新增 `_format_awooop_agent_evidence_lines()`，把 detail/history message 補上五段 AI Agent 證據鏈：
+  - `MCP / 自建 MCP`：Gateway 成功/總數、failed、blocked、top tool、first-class / legacy / policy count。
+  - `Sentry/SigNoz`：source correlation status、direct / candidate / applied 數量、provider 摘要。
+  - `Executor`：latest executor/status、operation type、action、operation count。
+  - `PlayBook / Ansible`：selected/candidate playbook、ansible considered、candidate count、check-mode、status。
+  - `KM / Learning`：KM entries、AutoRepair records、Ops records、verification result。
+- `detail:{incident_id}` 與 `history:{incident_id}` 兩個 callback 都在送出 Telegram chunk 前渲染同一段證據鏈。
+- `check-mode` 顯示收斂為 `yes/no/unknown`，避免 Telegram 訊息暴露 Python `True/False` 內部值。
+- callback reply 的 source envelope 既有 `awooop_status_chain` snapshot 已包含 `mcp`、`execution`、`source_refs.correlation`，本輪保持 DB replayable 結構不變，只補訊息可讀性。
+
+**local validation（完成）**：
+
+```text
+python3 -m py_compile apps/api/src/services/telegram_gateway.py apps/api/tests/test_telegram_message_templates.py apps/api/tests/test_telegram_adr050.py
+git diff --check
+PYTHONPATH=. DATABASE_URL='postgresql+asyncpg://test:test@localhost/test' /Users/ogt/.pyenv/shims/pytest tests/test_telegram_message_templates.py tests/test_telegram_adr050.py -q
+  81 passed in 0.59s
+```
+
+**目前整體進度**：
+
+- AwoooP 告警可觀測鏈：約 99.45%。
+- 低風險自動修復閉環：約 95.8%。
+- 前端 AI 自動化管理介面同步：約 98.6%。
+- 首頁 KPI / 小龍蝦流程 truth alignment：約 96.5%。
+- Telegram 詳情 / 歷史可追溯：約 97.5%。
+- callback / DB replayability：約 96.5%。
+- MCP / 自建 MCP 可視化：約 94.5%。
+- Sentry / SigNoz source correlation：約 92.8%。
+- Ansible / PlayBook 可視化：約 91.8%。
+- KM governance：約 84.5%。
+- AI Provider lane visibility：約 92%。
+- 完整 AI 自動化管理產品化：約 96.6%。