fix(telegram): CI/CD 告警簡化 + 心跳台北時區

2026-03-30 ogt: 告警格式修正

CI/CD 告警:
- 新增 CICDProgressMessage 簡潔格式
- webhooks.py 偵測 CD_*/CI_*/E2E_* 前綴
- 跳過 AI 仲裁，直接發送簡潔通知

心跳訊息:
- 修正 UTC → 台北時區 (feedback_timezone_taipei.md)
- 簡化格式，移除冗餘資訊

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

apps/api/src/api/v1/webhooks.py

@@ -1158,6 +1158,63 @@ async def alertmanager_webhook(
     # ==========================================================================
     alertname = alert.labels.get("alertname", "UnknownAlert")
 
+    # ==========================================================================
+    # 2026-03-30 ogt: CI/CD 告警偵測 - 跳過 AI 仲裁，使用簡潔格式
+    # CI/CD 告警 alertname 模式: CD_*, CI_*, E2E_*, SmokeTest, *_Build, *_Test
+    # ==========================================================================
+    cicd_prefixes = ("CD_", "CI_", "E2E_", "SmokeTest", "Build_", "Test_", "Deploy_")
+    cicd_suffixes = ("_Build", "_Test", "_Deploy", "_E2E")
+    is_cicd_alert = (
+        alertname.startswith(cicd_prefixes) or
+        alertname.endswith(cicd_suffixes) or
+        "CI/CD" in alertname or
+        "cicd" in alertname.lower()
+    )
+
+    if is_cicd_alert:
+        # CI/CD 告警 - 使用簡潔格式，不走 AI 仲裁
+        logger.info(
+            "alertmanager_cicd_detected",
+            alert_id=alert_id,
+            alertname=alertname,
+            message="Bypassing AI arbitration for CI/CD alert",
+        )
+
+        try:
+            telegram = get_telegram_gateway()
+            # 解析 CI/CD 狀態
+            stage = alert.labels.get("stage", "")
+            job_status = "success" if alert.labels.get("severity") == "info" else "running"
+            commit_sha = alert.labels.get("commit", "")
+            triggered_by = alert.labels.get("triggered_by", "CI")
+            workflow_url = alert.annotations.get("workflow_url", "")
+            summary = alert.annotations.get("summary", alertname)
+
+            await telegram.send_cicd_progress(
+                job_name=summary,
+                status=job_status,
+                stage=stage,
+                commit_sha=commit_sha,
+                triggered_by=triggered_by,
+                workflow_url=workflow_url,
+            )
+
+            return AlertResponse(
+                success=True,
+                message=f"CI/CD alert processed (simple format): {alertname}",
+                alert_id=alert_id,
+                approval_created=False,
+            )
+        except Exception as e:
+            logger.error("cicd_telegram_failed", error=str(e), alertname=alertname)
+            # CI/CD 通知失敗不阻擋流程
+            return AlertResponse(
+                success=True,
+                message=f"CI/CD alert logged (telegram failed): {alertname}",
+                alert_id=alert_id,
+                approval_created=False,
+            )
+
     # 映射 alertname → alert_type
     alertname_to_type = {
         "KubePodCrashLooping": "k8s_pod_crash",

apps/api/src/services/telegram_gateway.py

@@ -518,6 +518,66 @@ class DailySummaryMessage:
         return message[:900]
 
 
+@dataclass
+class CICDProgressMessage:
+    """
+    CI/CD 進度訊息 (CICD_PROGRESS)
+
+    2026-03-30 ogt: 新增，用於 CI/CD 流程進度通知
+    特性: 簡潔、不走 AI 仲裁、無按鈕
+    """
+    job_name: str               # Job 名稱 (e.g., Build, Test, Deploy)
+    status: str                 # running, success, failed
+    stage: str = ""             # CI/CD 階段 (e.g., build, test, deploy)
+    commit_sha: str = ""        # Git commit SHA
+    triggered_by: str = ""      # 觸發者
+    duration_seconds: int = 0   # 執行時間
+    message: str = ""           # 額外訊息
+    workflow_url: str = ""      # Workflow 連結
+
+    def format(self) -> str:
+        """格式化為 Telegram HTML (簡潔版)"""
+        # 狀態 emoji
+        status_emoji = {
+            "running": "🔄",
+            "success": "✅",
+            "failed": "❌",
+            "pending": "⏳",
+        }.get(self.status.lower(), "📦")
+
+        safe_job = html.escape(self.job_name[:40])
+        safe_stage = html.escape(self.stage[:20]) if self.stage else ""
+
+        # 時間格式化
+        duration_str = ""
+        if self.duration_seconds > 0:
+            minutes = self.duration_seconds // 60
+            seconds = self.duration_seconds % 60
+            duration_str = f" ({minutes}m {seconds}s)" if minutes > 0 else f" ({seconds}s)"
+
+        # Commit 資訊
+        commit_info = ""
+        if self.commit_sha:
+            commit_info = f"\n📋 <code>{html.escape(self.commit_sha[:8])}</code>"
+
+        # Workflow 連結
+        workflow_link = ""
+        if self.workflow_url:
+            safe_url = html.escape(self.workflow_url, quote=True)
+            workflow_link = f"\n🔗 <a href='{safe_url}'>Workflow</a>"
+
+        # 簡潔訊息
+        stage_label = f" | {safe_stage}" if safe_stage else ""
+        message = (
+            f"{status_emoji} <b>[AWOOOI CI/CD]</b>{stage_label}\n"
+            f"📦 {safe_job}{duration_str}"
+            f"{commit_info}"
+            f"{workflow_link}"
+        )
+
+        return message[:900]
+
+
 @dataclass
 class DeploySuccessMessage:
     """
@@ -1220,6 +1280,49 @@ class TelegramGateway:
 
         return result
 
+    async def send_cicd_progress(
+        self,
+        job_name: str,
+        status: str,
+        stage: str = "",
+        commit_sha: str = "",
+        triggered_by: str = "",
+        duration_seconds: int = 0,
+        message: str = "",
+        workflow_url: str = "",
+    ) -> dict:
+        """
+        發送 CI/CD 進度通知 (簡潔版，不走 AI 仲裁)
+
+        2026-03-30 ogt: 新增，解決 CI/CD 告警被當成事件處理的問題
+
+        Returns:
+            dict: Telegram API 回應
+        """
+        msg = CICDProgressMessage(
+            job_name=job_name,
+            status=status,
+            stage=stage,
+            commit_sha=commit_sha,
+            triggered_by=triggered_by,
+            duration_seconds=duration_seconds,
+            message=message,
+            workflow_url=workflow_url,
+        )
+
+        payload = {
+            "chat_id": self.chat_id,
+            "text": msg.format(),
+            "parse_mode": "HTML",
+            "disable_web_page_preview": True,
+        }
+
+        logger.info("telegram_cicd_progress_sending", job=job_name, status=status)
+        result = await self._send_request("sendMessage", payload)
+        logger.info("telegram_cicd_progress_sent", job=job_name, status=status)
+
+        return result
+
     async def send_deploy_success(
         self,
         commit_sha: str,
@@ -2110,30 +2213,15 @@ class TelegramGateway:
             if not self._initialized:
                 await self.initialize()
 
-            now = datetime.now(UTC)
-
-            # 計算上次訊息時間
-            last_msg_ago = "N/A"
-            if self._last_message_time:
-                delta = now - self._last_message_time
-                minutes = int(delta.total_seconds() // 60)
-                if minutes < 60:
-                    last_msg_ago = f"{minutes} 分鐘前"
-                else:
-                    hours = minutes // 60
-                    last_msg_ago = f"{hours} 小時前"
-
-            # 心跳訊息
+            # 心跳訊息 (2026-03-30 ogt: 改用台北時區，符合 feedback_timezone_taipei.md)
+            from src.utils.timezone import now_taipei
+            taipei_now = now_taipei()
             text = f"""💓 <b>AWOOOI 心跳</b>
-━━━━━━━━━━━━━━━━
-⏰ {now.strftime('%Y-%m-%d %H:%M:%S')} UTC
-📡 告警鏈路: ✅ 正常
-📨 上次訊息: {last_msg_ago}
-━━━━━━━━━━━━━━━━
-<i>每 30 分鐘自動發送</i>"""
+⏰ {taipei_now.strftime('%Y-%m-%d %H:%M:%S')} (台北)
+📡 告警鏈路: ✅ 正常"""
 
             await self.send_notification(text)
-            self._last_message_time = now
+            self._last_message_time = datetime.now(UTC)
 
             logger.info("telegram_heartbeat_sent")
             return True