fix(telegram): clarify automation notification state

apps/api/src/services/telegram_gateway.py

@@ -23,6 +23,7 @@ SOUL.md 鐵律 (4.1 Telegram 訊息壓縮原則):
 """
 
 import asyncio
+import hashlib
 import html
 import os
 import re
@@ -50,6 +51,8 @@ SNOOZE_KEY_PREFIX = "telegram_snooze:"  # {approval_id} -> 稍後提醒
 SILENCE_KEY_PREFIX = "telegram_silence:"  # {resource_name} -> 靜默
 SNOOZE_TTL_SECONDS = 30 * 60  # 30 分鐘
 SILENCE_TTL_SECONDS = 60 * 60  # 1 小時
+INCIDENT_UPDATE_DEDUP_PREFIX = "awoooi:tg_update_dedup:"  # {incident_id}:{status_hash}
+INCIDENT_UPDATE_DEDUP_TTL_SECONDS = 5 * 60  # 5 分鐘內相同狀態不重複洗版
 
 # 2026-04-01 Claude Code: Long Polling 分散式 Leader Election
 # 防止多 Pod 同時 getUpdates → 409 Conflict 互搶問題
@@ -261,6 +264,28 @@ class TelegramMessage:
             return "analysis_degraded"
         return "safe_gate_pending"
 
+    def _automation_status_summary(self) -> str:
+        """Telegram 首屏的人類可讀處置狀態。
+
+        這行是值班判斷入口：先讓人知道這張卡是「AI 已有建議待審批」、
+        「AI 無法修復需人工」或「純觀察」，細節才放到後面的鏈路區塊。
+        """
+        mode = self._automation_mode()
+        action = (self.suggested_action or "").upper()
+        text = f"{self.root_cause} {self.suggested_action}".lower()
+
+        if mode == "llm_timeout_manual_gate":
+            return "🔴 AI 分析超時，需人工排查"
+        if action in {"NO_ACTION", "待分析", ""} or "invalid_target" in text:
+            return "🟠 AI 無可安全執行動作，需人工判斷"
+        if self.confidence <= 0:
+            return "🟡 規則建議待審批"
+        if mode == "analysis_degraded":
+            return "🟠 AI 降級分析，需人工判斷"
+        if mode == "ai_proposal_ready":
+            return "🟡 AI 已提出修復建議，等待人工批准"
+        return "🟡 安全閘門待審批"
+
     def _format_automation_block(self) -> str:
         """Visible AI automation chain for every ACTION REQUIRED card.
         2026-05-04 ogt: 加入 Token 用量 + 具體 Ollama 伺服器顯示
@@ -344,6 +369,7 @@ class TelegramMessage:
         safe_root_cause = html.escape(self.root_cause)
         safe_action = html.escape(self.suggested_action)
         safe_downtime = html.escape(self.estimated_downtime)
+        safe_automation_summary = html.escape(self._automation_status_summary())
 
         # 2026-03-29 ogt: AI Token/Cost 顯示
         ai_cost_display = ""
@@ -441,6 +467,7 @@ class TelegramMessage:
             f"📋 <code>{html.escape(incident_id)}</code>\n"
             f"🎯 資源：<code>{safe_resource}</code>\n"
             f"{category_line}"
+            f"🧭 處置狀態：<b>{safe_automation_summary}</b>\n"
             f"\n"
             f"{automation_block}"
             f"\n"
@@ -4462,8 +4489,6 @@ class TelegramGateway:
 
         2026-04-09 Claude Sonnet 4.6 Asia/Taipei (統帥要求: 狀態變更在原訊息延續)
         """
-        from src.core.redis_client import get_redis
-
         redis = get_redis()
         redis_key = f"tg_msg:{incident_id}"
         stored = await redis.get(redis_key)
@@ -4481,6 +4506,31 @@ class TelegramGateway:
             logger.warning("append_incident_update_invalid_message_id", stored=stored)
             return False
 
+        # Telegram 只適合放決策摘要；同一 incident 的相同狀態 5 分鐘內不重複回覆，
+        # 詳細執行紀錄應進 timeline / AwoooP Run Monitor，避免群組被 auto-failure 洗版。
+        status_hash = hashlib.sha1(status_line.encode("utf-8")).hexdigest()[:16]
+        dedup_key = f"{INCIDENT_UPDATE_DEDUP_PREFIX}{incident_id}:{status_hash}"
+        try:
+            was_set = await redis.set(
+                dedup_key,
+                "1",
+                ex=INCIDENT_UPDATE_DEDUP_TTL_SECONDS,
+                nx=True,
+            )
+            if not was_set:
+                logger.info(
+                    "append_incident_update_dedup_suppressed",
+                    incident_id=incident_id,
+                    dedup_key=dedup_key,
+                )
+                return True
+        except Exception as exc:
+            logger.warning(
+                "append_incident_update_dedup_failed",
+                incident_id=incident_id,
+                error=str(exc),
+            )
+
         # Step 1: 取得原始訊息文字（Telegram Bot API 不提供讀取原文，只能在 editMessageText 裡重建）
         # 策略: 只追加 status_line，不讀取原文（Telegram edit 要傳完整新文字）
         # 所以先用 editMessageReplyMarkup 換按鈕，再 sendMessage 同 chat 以 reply 方式追加狀態

apps/api/tests/test_telegram_message_templates.py

@@ -7,6 +7,7 @@ test_telegram_message_templates.py - Telegram 訊息模板測試
 
 import pytest
 
+import src.services.telegram_gateway as telegram_gateway_module
 from src.services.telegram_gateway import (
     DailySummaryMessage,
     DeploySuccessMessage,
@@ -15,6 +16,7 @@ from src.services.telegram_gateway import (
     ResourceWarnMessage,
     SentryErrorMessage,
     TelegramMessage,
+    TelegramGateway,
 )
 
 
@@ -38,12 +40,50 @@ class TestTelegramMessageFormat:
         assert "🚨" in result
         assert "嚴重" in result
         assert "test-pod-123" in result
+        assert "處置狀態" in result
+        assert "規則建議待審批" in result
         assert "AI 自動化鏈路" in result
         assert "OpenClaw" in result
         assert "NemoTron" in result
         assert "ElephantAlpha" in result
         assert len(result) <= 4096  # Telegram HTML message limit
 
+    def test_telegram_message_ai_proposal_marks_approval_wait(self):
+        """有 AI 信心分數的修復建議必須標示為 AI 待審批。"""
+        msg = TelegramMessage(
+            status_emoji="⚠️",
+            risk_level="MEDIUM",
+            resource_name="awoooi-api",
+            root_cause="CPU sustained high",
+            suggested_action="kubectl rollout restart deployment/awoooi-api",
+            estimated_downtime="~30s",
+            approval_id="INC-20260506-0000",
+            confidence=0.82,
+            ai_provider="ollama_gcp_a",
+        )
+
+        result = msg.format()
+
+        assert "處置狀態" in result
+        assert "AI 已提出修復建議，等待人工批准" in result
+
+    def test_telegram_message_no_action_marks_manual_judgement(self):
+        """NO_ACTION 卡片必須一眼看得出需要人工判斷。"""
+        msg = TelegramMessage(
+            status_emoji="ℹ️",
+            risk_level="LOW",
+            resource_name="node-exporter-110",
+            root_cause="規則命中但沒有安全可執行動作",
+            suggested_action="NO_ACTION",
+            estimated_downtime="unknown",
+            approval_id="INC-20260506-0001",
+        )
+
+        result = msg.format()
+
+        assert "處置狀態" in result
+        assert "AI 無可安全執行動作，需人工判斷" in result
+
     def test_telegram_message_with_token_cost(self):
         """測試含 Token/Cost 的訊息"""
         msg = TelegramMessage(
@@ -63,6 +103,46 @@ class TestTelegramMessageFormat:
         assert "💰 Tokens: 1,500 / $0.0015" in result
 
 
+@pytest.mark.asyncio
+async def test_append_incident_update_deduplicates_same_status(monkeypatch):
+    """同一 Incident 的相同狀態更新 5 分鐘內不可重複洗版。"""
+
+    class FakeRedis:
+        def __init__(self):
+            self.set_calls = 0
+
+        async def get(self, key):
+            assert key == "tg_msg:INC-DEDUP"
+            return "12345"
+
+        async def set(self, *args, **kwargs):
+            self.set_calls += 1
+            assert kwargs["nx"] is True
+            assert kwargs["ex"] > 0
+            return self.set_calls == 1
+
+    fake_redis = FakeRedis()
+    sent_requests = []
+    gateway = TelegramGateway()
+
+    async def fake_send_request(method, payload):
+        sent_requests.append((method, payload))
+        return {"ok": True}
+
+    monkeypatch.setattr(telegram_gateway_module, "get_redis", lambda: fake_redis)
+    monkeypatch.setattr(gateway, "_send_request", fake_send_request)
+
+    status_line = "🤖❌ <b>[AUTO] AI 自動修復失敗，已升級人工介入</b>"
+
+    assert await gateway.append_incident_update("INC-DEDUP", status_line) is True
+    assert await gateway.append_incident_update("INC-DEDUP", status_line) is True
+
+    assert [method for method, _ in sent_requests] == [
+        "editMessageReplyMarkup",
+        "sendMessage",
+    ]
+
+
 class TestSentryErrorMessage:
     """測試 Sentry 錯誤訊息"""