fix(telegram): harden detail history html fallback

apps/api/src/services/telegram_gateway.py

@@ -342,7 +342,7 @@ def _telegram_html_chunks(lines: list[str], limit: int = _TELEGRAM_HTML_CHUNK_LI
             current = []
             current_len = 0
         if line_len > limit:
-            chunks.append(line[:limit])
+            chunks.append(_html_safe_plain_chunk(line, limit=limit))
             continue
         current.append(line)
         current_len += line_len
@@ -357,6 +357,17 @@ def _plain_text_from_html(text: str, limit: int = 3900) -> str:
     return html.unescape(plain)[:limit]
 
 
+def _html_safe_plain_chunk(text: str, limit: int) -> str:
+    """Render one overlong HTML line as parse-safe text for HTML mode chunks."""
+    plain = _plain_text_from_html(text, limit=limit)
+    escaped = html.escape(plain)
+    if len(escaped) <= limit:
+        return escaped
+    # Escaping may expand &, <, >. Trim once more after escaping; a partial HTML
+    # entity is still plain text to Telegram, while a partial tag is not.
+    return escaped[:limit]
+
+
 def _sanitize_telegram_error(text: str) -> str:
     """遮蔽 Telegram Bot URL 中的 token，避免例外字串污染 log / trace。"""
     return _TELEGRAM_BOT_URL_RE.sub(r"\1<redacted>", text)
@@ -5920,13 +5931,29 @@ class TelegramGateway:
         Returns:
             dict: API 回應
         """
+        payload_text = text[:500]
+        payload_parse_mode = parse_mode
+        if parse_mode and parse_mode.upper() == "HTML" and len(text) > 500:
+            payload_text = _plain_text_from_html(text, limit=500)
+            payload_parse_mode = None
+
         payload = {
             "chat_id": chat_id or self.alert_chat_id,
-            "text": text[:500],  # SOUL.md 字數限制
-            "parse_mode": parse_mode,
+            "text": payload_text,  # SOUL.md 字數限制
         }
+        if payload_parse_mode:
+            payload["parse_mode"] = payload_parse_mode
 
-        return await self._send_request("sendMessage", payload)
+        try:
+            return await self._send_request("sendMessage", payload)
+        except TelegramGatewayError as exc:
+            if payload_parse_mode and payload_parse_mode.upper() == "HTML" and "HTTP error: 400" in str(exc):
+                fallback_payload = {
+                    "chat_id": chat_id or self.alert_chat_id,
+                    "text": _plain_text_from_html(text, limit=500),
+                }
+                return await self._send_request("sendMessage", fallback_payload)
+            raise
 
     async def _send_html_line_message(
         self,

apps/api/tests/test_telegram_message_templates.py

@@ -71,6 +71,20 @@ def test_telegram_html_chunks_preserve_complete_lines() -> None:
     assert all(chunk.count("<b>") == chunk.count("</b>") for chunk in chunks)
 
 
+def test_telegram_html_chunks_render_single_overlong_html_line_as_safe_text() -> None:
+    """單行過長時不得切出未閉合 <code>，否則 Telegram 會 400。"""
+    line = "告警鍵: <code>" + ("node&lt;188&gt;&amp;" * 80) + "</code>"
+
+    chunks = telegram_gateway_module._telegram_html_chunks([line], limit=120)
+
+    assert len(chunks) == 1
+    assert len(chunks[0]) <= 120
+    assert "<code>" not in chunks[0]
+    assert "</code>" not in chunks[0]
+    assert "告警鍵" in chunks[0]
+    assert "&lt;" in chunks[0]
+
+
 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")
@@ -224,6 +238,55 @@ async def test_info_callback_sends_history_when_answer_callback_is_stale(monkeyp
     assert sent_history == ["INC-20260513-79ED5E"]
 
 
+@pytest.mark.asyncio
+async def test_send_notification_retries_plain_text_on_html_parse_400(monkeypatch):
+    """簡短 HTML 若被 Telegram 拒收，要用純文字重送，不回報成 HTTP 400。"""
+    gateway = TelegramGateway()
+    sent_requests = []
+
+    async def fake_send_request(method, payload):
+        sent_requests.append((method, payload))
+        if len(sent_requests) == 1:
+            raise telegram_gateway_module.TelegramGatewayError("HTTP error: 400")
+        return {"ok": True}
+
+    monkeypatch.setattr(gateway, "_send_request", fake_send_request)
+
+    result = await gateway.send_notification(
+        "⚠️ 無法取得歷史統計: <b>broken",
+        chat_id="chat",
+    )
+
+    assert result == {"ok": True}
+    assert sent_requests[0][1]["parse_mode"] == "HTML"
+    assert "parse_mode" not in sent_requests[1][1]
+    assert sent_requests[1][1]["text"] == "⚠️ 無法取得歷史統計: broken"
+
+
+@pytest.mark.asyncio
+async def test_send_notification_long_html_uses_plain_text_without_cutting_tags(monkeypatch):
+    """send_notification 的 500 字限制不可切壞 HTML tag。"""
+    gateway = TelegramGateway()
+    sent_requests = []
+
+    async def fake_send_request(method, payload):
+        sent_requests.append((method, payload))
+        return {"ok": True}
+
+    monkeypatch.setattr(gateway, "_send_request", fake_send_request)
+    long_text = "📊 <b>事件歷史統計</b>\n告警鍵: <code>" + ("node<188>&" * 90) + "</code>"
+
+    await gateway.send_notification(long_text, chat_id="chat")
+
+    assert len(sent_requests) == 1
+    payload = sent_requests[0][1]
+    assert "parse_mode" not in payload
+    assert len(payload["text"]) <= 500
+    assert "<code>" not in payload["text"]
+    assert "</code>" not in payload["text"]
+    assert "事件歷史統計" in payload["text"]
+
+
 class TestTelegramMessageFormat:
     """測試現有 TelegramMessage 格式化"""