fix(ops): harden heartbeat and momo alert noise

apps/api/src/services/telegram_gateway.py

@@ -71,6 +71,40 @@ INCIDENT_UPDATE_GLOBAL_FAILURE_DEDUP_TTL_SECONDS = 10 * 60  # 相同失敗摘要
 GROUPED_ALERT_DIGEST_DEDUP_PREFIX = "awoooi:tg_group_digest:"  # {group_key}
 GROUPED_ALERT_DIGEST_DEDUP_TTL_SECONDS = 5 * 60  # 同一告警群組 5 分鐘只推一則 digest
 
+_HEARTBEAT_WARNING_FINGERPRINT_RULES: tuple[tuple[re.Pattern[str], str], ...] = (
+    # 2026-06-24 Codex + ogt: Telegram heartbeat dedupe must track the
+    # actionable condition, not volatile probe details.  HTTP status, timeout,
+    # latency and count text can vary every 30 minutes while the operator
+    # action stays exactly the same, which caused repeated "heartbeat" noise.
+    (re.compile(r"^(Ollama\s+[^:]+)\s+異常:.*$"), r"\1 異常"),
+    (re.compile(r"^(.+?)\s+服務異常:.*$"), r"\1 服務異常"),
+    (re.compile(r"^(MCP\s+[^:]+):.*$"), r"\1"),
+    (re.compile(r"^(ArgoCD):.*$"), r"\1"),
+    (re.compile(r"^(PostgreSQL):.*$"), r"\1"),
+    (re.compile(r"^(Redis):.*$"), r"\1"),
+    (re.compile(r"^KM\s+向量化率偏低:.*$"), "KM 向量化率偏低"),
+    (re.compile(r"^系統沉默\s+.*$"), "系統沉默（無學習活動）"),
+    (re.compile(r"^待人工審核\s+.*$"), "待人工審核積壓"),
+)
+
+
+def _normalize_heartbeat_warning_for_fingerprint(warning: str) -> str:
+    """Return the stable actionable identity used for heartbeat dedupe."""
+    normalized = " ".join(str(warning).split())
+    for pattern, replacement in _HEARTBEAT_WARNING_FINGERPRINT_RULES:
+        replaced = pattern.sub(replacement, normalized)
+        if replaced != normalized:
+            return replaced
+    return normalized
+
+
+def _heartbeat_warnings_hash(warnings: list[str]) -> str:
+    """Hash heartbeat warnings after stripping volatile probe details."""
+    warnings_str = "|".join(
+        sorted(_normalize_heartbeat_warning_for_fingerprint(item) for item in warnings)
+    )
+    return hashlib.md5(warnings_str.encode()).hexdigest()[:12]
+
 # 2026-04-01 Claude Code: Long Polling 分散式 Leader Election
 # 防止多 Pod 同時 getUpdates → 409 Conflict 互搶問題
 POLLING_LEADER_KEY = "telegram:polling:leader"
@@ -10118,13 +10152,11 @@ class TelegramGateway:
             #   有 warnings 跟上次相同 → 跳過（hash 對比）
             #   有 warnings 跟上次不同 → 立即推送（新狀況不漏）
             #   warnings 消失 → 只推一次恢復通知，之後回到安靜
-            import hashlib
             WARNINGS_HASH_TTL = 24 * 3600
             healthy_suppressed_key = "heartbeat:healthy_suppressed_last_seen"
             warnings_hash_key = "heartbeat:warnings_hash"
 
-            warnings_str = "|".join(sorted(report.warnings))
-            warnings_hash = hashlib.md5(warnings_str.encode()).hexdigest()[:12]
+            warnings_hash = _heartbeat_warnings_hash(report.warnings)
 
             if not report.warnings:
                 # 健康狀態：沒有上一輪 warning 時不送 Telegram，避免成功心跳洗版。