在 Day 18 中，我們成功利用 GitHub Actions 排程實現了 24/7 的無人值守巡航。然而，在真正的生產維運（Production Ops）中，完全無人干預的系統就像一艘靜音潛水艇——你不知道它正順暢前進，還是已經在海底拋錨。

當遇到 Google API 速率限制（Quota Exceeded）、來源網址無效、或是 LLM 產生不在 Schema 約束內的畸變輸出時，若缺乏即時回饋，系統故障可能數天都未被發現。

本篇將建構雷達系統的巡航哨兵（Sentinel Monitor）：

1. 定義健康度心跳與日誌聚合金鑰。
2. 實作輕量、零第三方依賴的 Webhook 通知轉接器（以 Discord / Slack 格式為例）。
3. 在 GitHub Actions 失敗或管線異常時，自動觸發即時推播到行動端。

一、核心架構：告警通知配接器（src/adapters/notifier.py）

依循六角架構，通知管道屬於系統的主動驅動轉接器（Driven Adapter）。我們封裝 WebhookNotifier，只需透過標準函式庫 requests，就能以 Markdown Rich Embed 格式推送情報統計或報錯日誌：

# src/adapters/notifier.py
import logging
import os
import requests

logger = logging.getLogger(__name__)

class WebhookNotifier:
    """輕量級 Webhook 告警轉接器 (相容 Discord / Slack Webhook 格式)"""

    def __init__(self):
        self.webhook_url = os.getenv("ALERT_WEBHOOK_URL")

    def send_pipeline_report(self, processed_count: int, archived_count: int, error_count: int):
        """推送每日巡航成功總結報告"""
        if not self.webhook_url:
            logger.info("未設定 ALERT_WEBHOOK_URL，跳過推播。")
            return

        payload = {
            "embeds": [
                {
                    "title": "📡 Intelligence Radar 巡航巡邏完畢",
                    "color": 3066993 if error_count == 0 else 15158332, # 綠色或紅色
                    "fields": [
                        {"name": "通過精煉 (PROCESSED)", "value": f"**{processed_count}** 則", "inline": True},
                        {"name": "雜訊熔斷 (ARCHIVED)", "value": f"**{archived_count}** 則", "inline": True},
                        {"name": "異常中斷 (ERROR)", "value": f"**{error_count}** 則", "inline": True},
                    ],
                    "footer": {"text": "Google ADK x Hexagonal Intelligence Radar 2026"}
                }
            ]
        }
        self._dispatch(payload)

    def send_critical_alert(self, error_message: str, context: str = "Pipeline Failure"):
        """管線發生未預期崩潰時的緊急告警"""
        if not self.webhook_url:
            return

        payload = {
            "content": "🚨 **【緊急告警】情報雷達管線發生崩潰異常！**",
            "embeds": [
                {
                    "title": f"系統錯誤情境: {context}",
                    "color": 15158332, # 醒目紅
                    "description": f"```python\n{error_message[:1500]}\n```",
                    "footer": {"text": "請盡速檢查 GitHub Actions 執行記錄"}
                }
            ]
        }
        self._dispatch(payload)

    def _dispatch(self, payload: dict):
        try:
            resp = requests.post(self.webhook_url, json=payload, timeout=10)
            resp.raise_for_status()
        except Exception as e:
            logger.error(f"發送 Webhook 告警失敗: {e}")

二、主程式聚合與統計回填（run_pipeline.py）

我們微調 Day 16 的進入點 run_pipeline.py，加入健康度計數器與全域崩潰攔截器：

# run_pipeline.py
import logging
import os
import sys
from google import genai

from src.adapters.sheets_adapter import SheetsAdapter
from src.adapters.notifier import WebhookNotifier
from src.core.models import ItemStatus
from src.services.pipeline import IntelligencePipeline

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

def main():
    notifier = WebhookNotifier()
    try:
        api_key = os.environ.get("GEMINI_API_KEY")
        if not api_key:
            raise ValueError("找不到 GEMINI_API_KEY 環境變數")

        client = genai.Client(api_key=api_key)
        sheets_adapter = SheetsAdapter()
        pipeline = IntelligencePipeline(client)

        pending_items = sheets_adapter.fetch_pending_items(auto_lock=True)
        logger.info(f"成功鎖定待處理情報數量: {len(pending_items)}")

        stats = {ItemStatus.PROCESSED: 0, ItemStatus.ARCHIVED: 0, ItemStatus.ERROR: 0}

        for item in pending_items:
            try:
                processed_item = pipeline.process_item(item)
                sheets_adapter.update_item_status(processed_item)
                stats[processed_item.status] = stats.get(processed_item.status, 0) + 1
            except Exception as item_err:
                logger.error(f"條目 {item.entry_id} 處理異常: {item_err}")
                item.status = ItemStatus.ERROR
                item.metadata["pipeline_error"] = str(item_err)
                sheets_adapter.update_item_status(item)
                stats[ItemStatus.ERROR] += 1

        # 若有處理到項目，發布巡航總結推播
        if pending_items:
            notifier.send_pipeline_report(
                processed_count=stats.get(ItemStatus.PROCESSED, 0),
                archived_count=stats.get(ItemStatus.ARCHIVED, 0),
                error_count=stats.get(ItemStatus.ERROR, 0),
            )

    except Exception as fatal_err:
        logger.critical(f"管線遭遇致命崩潰: {fatal_err}", exc_info=True)
        notifier.send_critical_alert(str(fatal_err), context="Fatal Unhandled Exception")
        sys.exit(1)

if __name__ == "__main__":
    main()

三、GitHub Actions CI/CD 失敗救援推播（.github/workflows/patrol.yml）

除了 Python 內部捕獲的異常，若虛擬機在「安裝依賴失敗」或「Pytest 測試沒過」時中斷，Python 腳本根本還沒執行，這時該怎麼辦？

我們利用 GitHub Actions 的 if: failure() 條件，在工作流最後掛上一道「守護神告警 Step」：

# 在 .github/workflows/patrol.yml 的 jobs.patrol.steps 最後追加：

      - name: Notify CI/CD Pipeline Crash
        if: failure() # 只有當前面任何一個步驟失敗亮紅燈時才觸發
        env:
          ALERT_WEBHOOK_URL: ${{ secrets.ALERT_WEBHOOK_URL }}
        run: |
          if [ -n "$ALERT_WEBHOOK_URL" ]; then
            curl -H "Content-Type: application/json" \
                 -X POST \
                 -d '{"content": "🚨 **[GitHub Actions 巡航失敗]** 情報雷達在測試或環境建置階段中斷，請立即前往檢視！"}' \
                 $ALERT_WEBHOOK_URL
          fi

四、實戰踩坑指南：告警架構的防護邊界

在實作雲端哨兵時，有兩大容易引發「次生災難」的經典陷阱：

1. 警報風暴（Alert Fatigue / Notification Storm）

• 反面教材：在 for item in pending_items: 迴圈裡面，每發生一次錯誤就發送一則 Webhook。若一次批次有 50 筆錯誤，手機會在 3 秒內被震動癱瘓，甚至觸發 Discord / Slack 的發信頻率限制（Rate Limit 429）。
• 正確解法：在迴圈內只更新記憶體內的計數器 stats**，整個批次跑完後只彙總推送一次單一 Embed 報告**。

2. 避免通知失敗掩蓋業務錯誤

• 在 _dispatch 方法中，發送 HTTP 請求必須加上 try...except 隔離，不能因為網路抖動導致通知發不出去而把原本正常回填到 Sheets 的流程中斷。

總結與成效驗證

完成本篇設定後，只要在 GitHub Secrets 填入 ALERT_WEBHOOK_URL：

• 平日靜默運行：無人值守排程每日巡航。若試算表內有新進資料，巡邏結束後手機會優雅地收到一張綠色儀表板卡片，清楚列出幾則通過初審、幾則被雜訊熔斷。
• 異常秒級捕獲：若憑證失效、模型 Quota 耗盡或 CI 測試破裂，手機第一時間便會收到紅色警報與 Traceback 摘要。

明天 Day 20，我們將正式進入系列文章的第五階段：戰情室視覺化與情資落地，探討如何運用 Google Sheets 的樞紐分析與前端輕量儀表板，將這些 PROCESSED 的多維情報轉化為一目了然的科研與投資戰情雷達！

本地到遠端推進提示：

1. 建立 src/adapters/notifier.py。
2. 更新 run_pipeline.py 與 .github/workflows/patrol.yml。
3. （可選）在 Discord 隨便開一個私人頻道 ➔ 頻道設定 ➔ 整合 ➔ 建立 Webhook ➔ 複製網址，貼到 GitHub Secrets 的 ALERT_WEBHOOK_URL。
4. 執行 ruff check . 與 pytest tests/ -v 確保全綠後 git push！

🛠️ 實戰踩坑指南（增補版）：從分佈式鎖到隔離驗證

在實作哨兵監控與驗證過程中，有兩個關鍵現象值得深入剖析：

1. 狀態機的鎖定效應（PENDING ➔ ANALYZING）

在手動除錯時，讀者常發現把資料設為 PENDING 後，工作流一旦跑過，該資料在試算表上會自動變成 ANALYZING。
這證明了 fetch_pending_items(auto_lock=True) 的原子性搶佔鎖（Pessimistic Lock） 正在發揮作用：

• 只要條目被 Worker 捕獲，立即上鎖為 ANALYZING，防止多節點重複處理。
• 若管線在環境建置或上游階段崩潰，該條目會停留在鎖定態，避免髒資料反覆被未修復的 Worker 空轉拉取。除錯時只需手動復位為 PENDING 即可重新排隊。

2. 告警隔離性驗證（Chaos Testing 實測）

當我們刻意將 ALERT_WEBHOOK_URL 設為無效的 404 網址時，在 GitHub Actions 執行細節中可觀察到：

• 系統內部拋出 發送 Webhook 告警失敗: 404 Client Error，並被 _dispatch 的 try...except 吞吐吸收。
• 主管線不受通訊軟體網路狀態或 API 錯誤影響，Google Sheets 上的狀態依舊順利回填為 PROCESSED，整體 GitHub Actions 依舊順暢綠燈收尾。

3. 警報風暴（Alert Fatigue）防禦

• 在迴圈內只做記憶體內的 stats 計數，整個批次跑完後只發送一次單一彙總卡片，杜絕短時間內轟炸 Discord 伺服器引發 429 限流。