diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py
index 2df5c590..d82ffbec 100644
--- a/apps/api/src/services/telegram_gateway.py
+++ b/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 = [
+ "",
+ "🧭 AwoooP 狀態鏈",
+ "來源: DB Truth-chain + ADR-100 history",
+ (
+ "階段: "
+ f"{html.escape(current_stage)} / "
+ f"{html.escape(stage_status)} | "
+ f"判定: {html.escape(verdict)}"
+ ),
+ (
+ "AI 修復: "
+ f"{html.escape(repair_state)} | "
+ f"驗證: {html.escape(str(verification))}"
+ ),
+ (
+ "證據: "
+ f"auto-repair {auto_repair_records} / "
+ f"ops {operation_records} / "
+ f"MCP {gateway_total} / "
+ f"KM {km_entries}"
+ ),
+ (
+ "ADR-100: "
+ f"{html.escape(remediation_state)} "
+ f"{remediation_total} 次 | "
+ f"{html.escape(str(latest_route))}"
+ ),
+ (
+ "寫入: "
+ f"incident {html.escape(_bool_code(latest.get('writes_incident_state'), unknown_when_none=True))} / "
+ f"auto-repair {html.escape(_bool_code(latest.get('writes_auto_repair_result'), unknown_when_none=True))} | "
+ f"人工: {'yes' if needs_human else 'no'}"
+ ),
+ f"下一步: {html.escape(next_step)}",
+ ]
+
+ 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 += [
- "",
- "🧭 DB Truth-chain",
- (
- "階段: "
- f"{html.escape(str(truth_status.get('current_stage') or 'unknown'))}"
- " / "
- f"{html.escape(str(truth_status.get('stage_status') or 'unknown'))}"
- ),
- (
- "人工介入: "
- f"{'yes' if truth_status.get('needs_human') else 'no'}"
- ),
- ]
- 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,
diff --git a/apps/api/tests/test_telegram_message_templates.py b/apps/api/tests/test_telegram_message_templates.py
index 18748e86..63e094d0 100644
--- a/apps/api/tests/test_telegram_message_templates.py
+++ b/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 "驗證: healthy" in joined
+ assert "auto-repair 1" in joined
+ assert "auto_repair_executor/rollout_restart/write" in joined
+ assert "人工: no" 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 "人工: yes" 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")