feat(telegram): surface awooop status chain

apps/api/src/services/telegram_gateway.py

@@ -295,6 +295,144 @@ def _format_remediation_evidence_block(history: dict[str, object] | None) -> str
     )
 
 
+def _safe_int(value: object) -> int:
+    try:
+        return int(value or 0)
+    except (TypeError, ValueError):
+        return 0
+
+
+def _bool_code(value: object, *, unknown_when_none: bool = False) -> str:
+    if value is None and unknown_when_none:
+        return "unknown"
+    return "yes" if bool(value) else "no"
+
+
+def _format_awooop_status_chain_lines(
+    *,
+    truth_chain: dict[str, object] | None = None,
+    remediation_history: dict[str, object] | None = None,
+) -> list[str]:
+    """Unified Telegram detail/history summary for the AwoooP automation stage."""
+    if not truth_chain and not remediation_history:
+        return []
+
+    truth_status = (
+        truth_chain.get("truth_status")
+        if isinstance(truth_chain, dict) and isinstance(truth_chain.get("truth_status"), dict)
+        else {}
+    )
+    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 {}
+    quality_blockers = quality.get("blockers") if isinstance(quality.get("blockers"), list) else []
+    truth_blockers = truth_status.get("blockers") if isinstance(truth_status.get("blockers"), list) else []
+
+    latest = _latest_remediation_history_item(remediation_history)
+    remediation_state = _remediation_evidence_state(remediation_history) or "missing"
+    remediation_total = (
+        _safe_int(remediation_history.get("total"))
+        if isinstance(remediation_history, dict)
+        else 0
+    )
+    latest_route = "none"
+    if latest:
+        latest_route = (
+            f"{latest.get('agent_id') or 'unknown_agent'}/"
+            f"{latest.get('tool_name') or 'current_state'}/"
+            f"{latest.get('required_scope') or 'unknown'}"
+        )
+
+    current_stage = str(truth_status.get("current_stage") or "unknown")
+    stage_status = str(truth_status.get("stage_status") or "unknown")
+    verdict = str(quality.get("verdict") or "unknown")
+    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"))
+    gateway_total = _safe_int(facts.get("mcp_gateway_total"))
+    km_entries = _safe_int(facts.get("knowledge_entries"))
+    needs_human = bool(truth_status.get("needs_human"))
+
+    if verdict == "auto_repaired_verified":
+        repair_state = "auto_repaired_verified"
+        next_step = "monitor_for_regression"
+    elif auto_repair_records > 0 or operation_records > 0:
+        repair_state = "executed_pending_verification" if verification == "missing" else "executed"
+        next_step = "verify_execution_result"
+    elif remediation_state == "read_only":
+        repair_state = "read_only_dry_run"
+        next_step = "approve_or_escalate_from_awooop"
+    elif remediation_state == "write_observed":
+        repair_state = "write_observed_manual_review"
+        next_step = "review_write_evidence"
+    elif remediation_state == "blocked":
+        repair_state = "blocked_manual_required"
+        next_step = "manual_investigation"
+    elif needs_human:
+        repair_state = "manual_required"
+        next_step = "manual_investigation"
+    else:
+        repair_state = "no_execution_evidence"
+        next_step = "collect_evidence_or_wait"
+
+    if remediation_state in {"blocked", "fetch_failed"}:
+        needs_human = True
+    if (
+        remediation_state == "write_observed"
+        and repair_state != "auto_repaired_verified"
+    ):
+        needs_human = True
+
+    lines = [
+        "",
+        "🧭 <b>AwoooP 狀態鏈</b>",
+        "來源: <code>DB Truth-chain</code> + <code>ADR-100 history</code>",
+        (
+            "階段: "
+            f"<code>{html.escape(current_stage)}</code> / "
+            f"<code>{html.escape(stage_status)}</code> | "
+            f"判定: <code>{html.escape(verdict)}</code>"
+        ),
+        (
+            "AI 修復: "
+            f"<code>{html.escape(repair_state)}</code> | "
+            f"驗證: <code>{html.escape(str(verification))}</code>"
+        ),
+        (
+            "證據: "
+            f"auto-repair <code>{auto_repair_records}</code> / "
+            f"ops <code>{operation_records}</code> / "
+            f"MCP <code>{gateway_total}</code> / "
+            f"KM <code>{km_entries}</code>"
+        ),
+        (
+            "ADR-100: "
+            f"<code>{html.escape(remediation_state)}</code> "
+            f"<code>{remediation_total}</code> 次 | "
+            f"<code>{html.escape(str(latest_route))}</code>"
+        ),
+        (
+            "寫入: "
+            f"incident <code>{html.escape(_bool_code(latest.get('writes_incident_state'), unknown_when_none=True))}</code> / "
+            f"auto-repair <code>{html.escape(_bool_code(latest.get('writes_auto_repair_result'), unknown_when_none=True))}</code> | "
+            f"人工: <code>{'yes' if needs_human else 'no'}</code>"
+        ),
+        f"下一步: <code>{html.escape(next_step)}</code>",
+    ]
+
+    blockers = [str(item) for item in [*truth_blockers, *quality_blockers] if item]
+    if blockers:
+        lines.append("卡點: " + html.escape(", ".join(blockers[:4])))
+    return lines
+
+
 async def _fetch_remediation_summary_for_card(
     *,
     approval_id: str,
@@ -5700,6 +5838,7 @@ class TelegramGateway:
                             + html.escape(", ".join(mismatch_codes[:4]))
                         )
 
+            remediation_history: dict[str, object] | None = None
             try:
                 from src.services.adr100_remediation_service import (
                     get_adr100_remediation_service,
@@ -5709,7 +5848,6 @@ class TelegramGateway:
                     limit=5,
                     incident_id=incident_id,
                 )
-                lines += _format_remediation_history_lines(remediation_history)
             except Exception as remediation_exc:
                 logger.warning(
                     "incident_detail_remediation_history_summary_failed",
@@ -5724,6 +5862,11 @@ class TelegramGateway:
                     source_id=incident_id,
                     project_id=getattr(incident, "project_id", None) or "awoooi",
                 )
+                lines += _format_awooop_status_chain_lines(
+                    truth_chain=truth_chain,
+                    remediation_history=remediation_history,
+                )
+                lines += _format_remediation_history_lines(remediation_history)
                 gateway_summary = (
                     (truth_chain.get("mcp") or {})
                     .get("awooop_gateway")
@@ -5738,6 +5881,10 @@ class TelegramGateway:
                     incident_id=incident_id,
                     error=str(truth_exc),
                 )
+                lines += _format_awooop_status_chain_lines(
+                    remediation_history=remediation_history,
+                )
+                lines += _format_remediation_history_lines(remediation_history)
 
             await self._send_html_line_message(
                 lines,
@@ -5841,6 +5988,23 @@ class TelegramGateway:
                     lines += ["", "⚠️ Redis 統計暫時無法取得"]
 
             # === Layer 3: DB truth-chain（避免 Redis TTL / frequency_snapshot 缺口造成誤判） ===
+            remediation_history: dict[str, object] | None = None
+            try:
+                from src.services.adr100_remediation_service import (
+                    get_adr100_remediation_service,
+                )
+
+                remediation_history = await get_adr100_remediation_service().history(
+                    limit=5,
+                    incident_id=incident_id,
+                )
+            except Exception as remediation_exc:
+                logger.warning(
+                    "incident_history_remediation_history_summary_failed",
+                    incident_id=incident_id,
+                    error=str(remediation_exc),
+                )
+
             try:
                 from src.services.awooop_truth_chain_service import fetch_truth_chain
 
@@ -5848,28 +6012,11 @@ class TelegramGateway:
                     source_id=incident_id,
                     project_id=getattr(incident, "project_id", None) or "awoooi",
                 )
-                truth_status = truth_chain.get("truth_status") or {}
-                if truth_status:
-                    lines += [
-                        "",
-                        "🧭 <b>DB Truth-chain</b>",
-                        (
-                            "階段: "
-                            f"<code>{html.escape(str(truth_status.get('current_stage') or 'unknown'))}</code>"
-                            " / "
-                            f"<code>{html.escape(str(truth_status.get('stage_status') or 'unknown'))}</code>"
-                        ),
-                        (
-                            "人工介入: "
-                            f"<code>{'yes' if truth_status.get('needs_human') else 'no'}</code>"
-                        ),
-                    ]
-                    blockers = truth_status.get("blockers")
-                    if isinstance(blockers, list) and blockers:
-                        lines.append(
-                            "卡點: "
-                            + html.escape(", ".join(str(item) for item in blockers[:4]))
-                        )
+                lines += _format_awooop_status_chain_lines(
+                    truth_chain=truth_chain,
+                    remediation_history=remediation_history,
+                )
+                lines += _format_remediation_history_lines(remediation_history)
                 lines += _format_automation_quality_lines(
                     truth_chain.get("automation_quality")
                 )
@@ -5879,6 +6026,10 @@ class TelegramGateway:
                     incident_id=incident_id,
                     error=str(truth_exc),
                 )
+                lines += _format_awooop_status_chain_lines(
+                    remediation_history=remediation_history,
+                )
+                lines += _format_remediation_history_lines(remediation_history)
 
             await self._send_html_line_message(
                 lines,

apps/api/tests/test_telegram_message_templates.py

@@ -85,6 +85,98 @@ def test_telegram_html_chunks_render_single_overlong_html_line_as_safe_text() ->
     assert "<" in chunks[0]
 
 
+def test_awooop_status_chain_lines_show_verified_auto_repair_stage() -> None:
+    """詳情/歷史要直接說清楚是否已 AI 自動修復、驗證到哪裡。"""
+    lines = telegram_gateway_module._format_awooop_status_chain_lines(
+        truth_chain={
+            "truth_status": {
+                "current_stage": "execution_succeeded",
+                "stage_status": "success",
+                "needs_human": False,
+                "blockers": [],
+            },
+            "automation_quality": {
+                "verdict": "auto_repaired_verified",
+                "facts": {
+                    "auto_repair_execution_records": 1,
+                    "automation_operation_records": 1,
+                    "verification_result": "healthy",
+                    "mcp_gateway_total": 3,
+                    "knowledge_entries": 2,
+                },
+                "blockers": [],
+            },
+        },
+        remediation_history={
+            "total": 1,
+            "items": [
+                {
+                    "agent_id": "auto_repair_executor",
+                    "tool_name": "rollout_restart",
+                    "required_scope": "write",
+                    "verification_result_preview": "healthy",
+                    "writes_incident_state": True,
+                    "writes_auto_repair_result": True,
+                }
+            ],
+        },
+    )
+
+    joined = "\n".join(lines)
+    assert "AwoooP 狀態鏈" in joined
+    assert "auto_repaired_verified" in joined
+    assert "驗證: <code>healthy</code>" in joined
+    assert "auto-repair <code>1</code>" in joined
+    assert "auto_repair_executor/rollout_restart/write" in joined
+    assert "人工: <code>no</code>" in joined
+    assert "monitor_for_regression" in joined
+
+
+def test_awooop_status_chain_lines_show_read_only_manual_gate() -> None:
+    """只讀試跑不能被說成已自動修復，必須顯示等待審批/人工。"""
+    lines = telegram_gateway_module._format_awooop_status_chain_lines(
+        truth_chain={
+            "truth_status": {
+                "current_stage": "approval_required",
+                "stage_status": "waiting",
+                "needs_human": True,
+                "blockers": ["pending_human_approval"],
+            },
+            "automation_quality": {
+                "verdict": "approval_required",
+                "facts": {
+                    "auto_repair_execution_records": 0,
+                    "automation_operation_records": 0,
+                    "verification_result": "missing",
+                    "mcp_gateway_total": 1,
+                    "knowledge_entries": 0,
+                },
+                "blockers": [],
+            },
+        },
+        remediation_history={
+            "total": 2,
+            "items": [
+                {
+                    "agent_id": "investigator",
+                    "tool_name": "ssh_diagnose",
+                    "required_scope": "read",
+                    "verification_result_preview": "degraded",
+                    "writes_incident_state": False,
+                    "writes_auto_repair_result": False,
+                }
+            ],
+        },
+    )
+
+    joined = "\n".join(lines)
+    assert "read_only_dry_run" in joined
+    assert "investigator/ssh_diagnose/read" in joined
+    assert "人工: <code>yes</code>" in joined
+    assert "approve_or_escalate_from_awooop" in joined
+    assert "pending_human_approval" in joined
+
+
 def test_awooop_runs_url_for_incident_uses_public_incident_filter() -> None:
     """Telegram URL button 必須導到公開 AwoooP Run list，並帶 incident filter。"""
     url = telegram_gateway_module._awooop_runs_url_for_incident("INC-20260514-F85F21")