diff --git a/apps/api/src/api/v1/webhooks.py b/apps/api/src/api/v1/webhooks.py
index 2ad9887e3..356fef925 100644
--- a/apps/api/src/api/v1/webhooks.py
+++ b/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",
diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py
index 9d223b2e5..6753b5564 100644
--- a/apps/api/src/services/telegram_gateway.py
+++ b/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📋 {html.escape(self.commit_sha[:8])}"
+
+ # Workflow 連結
+ workflow_link = ""
+ if self.workflow_url:
+ safe_url = html.escape(self.workflow_url, quote=True)
+ workflow_link = f"\n🔗 Workflow"
+
+ # 簡潔訊息
+ stage_label = f" | {safe_stage}" if safe_stage else ""
+ message = (
+ f"{status_emoji} [AWOOOI CI/CD]{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"""💓 AWOOOI 心跳
-━━━━━━━━━━━━━━━━
-⏰ {now.strftime('%Y-%m-%d %H:%M:%S')} UTC
-📡 告警鏈路: ✅ 正常
-📨 上次訊息: {last_msg_ago}
-━━━━━━━━━━━━━━━━
-每 30 分鐘自動發送"""
+⏰ {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