fix(telegram): Redis Leader Election 解決多 Pod 409 Conflict

問題: 2 個 API Pod 同時 getUpdates → 互相 409 → 兩個都失敗
根本原因: explicit env TELEGRAM_ENABLE_POLLING=false 被 kubectl patch 設入
  deployment，覆蓋 ConfigMap 的 true (feedback_k8s_env_precedence.md 違規)

修復步驟:
1. kubectl patch 移除 deployment 的 explicit env override
2. 實作 Redis Leader Election 防止多 Pod 競爭
   - 使用 SET NX EX=45 取得 Leader Lock
   - _leader_renewer(): 每 20s 續約，確保 Leader 持有 Lock
   - _leader_watcher(): 非 Leader Pod 每 30s 嘗試接管
   - 409 時主動釋放 Lock，Watcher 競爭接管

結果: 一個 Pod 正常 polling，另一個 Pod 進入 Watcher 待命模式

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

apps/api/src/services/telegram_gateway.py

@@ -24,6 +24,7 @@ SOUL.md 鐵律 (4.1 Telegram 訊息壓縮原則):
 
 import asyncio
 import html
+import os
 from dataclasses import dataclass
 from datetime import UTC, datetime
 
@@ -48,6 +49,13 @@ SILENCE_KEY_PREFIX = "telegram_silence:"  # {resource_name} -> 靜默
 SNOOZE_TTL_SECONDS = 30 * 60  # 30 分鐘
 SILENCE_TTL_SECONDS = 60 * 60  # 1 小時
 
+# 2026-04-01 Claude Code: Long Polling 分散式 Leader Election
+# 防止多 Pod 同時 getUpdates → 409 Conflict 互搶問題
+POLLING_LEADER_KEY = "telegram:polling:leader"
+POLLING_LEADER_TTL = 45   # seconds - Pod 宕掉後 45s 自動轉移
+POLLING_LEADER_RENEW = 20  # seconds - 每 20s 續約
+POLLING_LEADER_WATCH = 30  # seconds - 非 Leader Pod 每 30s 嘗試接管
+
 logger = structlog.get_logger(__name__)
 
 # =============================================================================
@@ -1018,6 +1026,9 @@ class TelegramGateway:
         self._polling_active = False
         self._polling_task: asyncio.Task | None = None
         self._last_update_id = 0
+        # 2026-04-01 Claude Code: 分散式 Leader Election (防 2-Pod 409 互搶)
+        self._pod_id = os.environ.get("POD_NAME", os.urandom(8).hex())
+        self._leader_task: asyncio.Task | None = None
         # Phase 6.5: 心跳監控 (防止沉默盲點)
         self._last_message_time: datetime | None = None
         self._heartbeat_task: asyncio.Task | None = None
@@ -2232,15 +2243,17 @@ class TelegramGateway:
 
     async def close(self) -> None:
         """關閉 Gateway"""
-        # 停止 Long Polling
+        # 停止 Long Polling 與 Leader 相關 Tasks
         self._polling_active = False
-        if self._polling_task and not self._polling_task.done():
-            self._polling_task.cancel()
-            try:
-                await self._polling_task
-            except asyncio.CancelledError:
-                pass
-            self._polling_task = None
+        for task in (self._polling_task, self._leader_task):
+            if task and not task.done():
+                task.cancel()
+                try:
+                    await task
+                except asyncio.CancelledError:
+                    pass
+        self._polling_task = None
+        self._leader_task = None
 
         if self._http_client:
             await self._http_client.aclose()
@@ -2258,6 +2271,9 @@ class TelegramGateway:
 
         取代 Webhook 模式，適用於內網環境
         統帥鐵律: 內網無法接收外部 Webhook，必須主動輪詢
+
+        2026-04-01 Claude Code: 加入 Redis Leader Election
+        多 Pod 環境下，只有 Leader 執行 getUpdates，其餘 Pod 進入 Watcher 模式
         """
         if not self._initialized:
             success = await self.initialize()
@@ -2269,15 +2285,33 @@ class TelegramGateway:
             logger.warning("telegram_long_polling_already_running")
             return
 
-        # 先刪除 Webhook，確保可以使用 getUpdates
+        # 嘗試取得 Leader Lock (NX = 僅在不存在時設定)
+        redis = await get_redis()
+        acquired = await redis.set(POLLING_LEADER_KEY, self._pod_id, nx=True, ex=POLLING_LEADER_TTL)
+
+        if not acquired:
+            current_leader = await redis.get(POLLING_LEADER_KEY)
+            logger.info(
+                "telegram_polling_not_leader",
+                pod_id=self._pod_id,
+                current_leader=current_leader,
+                action="watcher_mode",
+            )
+            # 啟動 Watcher：定期嘗試接管
+            self._leader_task = asyncio.create_task(self._leader_watcher())
+            return
+
+        # 取得 Leader Lock，開始 Polling
         await self._delete_webhook()
 
         self._polling_active = True
         self._last_update_id = 0
         self._polling_task = asyncio.create_task(self._polling_loop())
+        self._leader_task = asyncio.create_task(self._leader_renewer())
 
         logger.info(
             "telegram_long_polling_started",
+            pod_id=self._pod_id,
             timeout=LONG_POLLING_TIMEOUT,
             chat_id=self.chat_id[:10] + "..." if self.chat_id else "N/A",
         )
@@ -2354,15 +2388,23 @@ class TelegramGateway:
 
             except httpx.HTTPStatusError as e:
                 if e.response.status_code == 409:
-                    # 409 Conflict: 另一個 getUpdates session 殘留
-                    # 可能是舊 pod/host 的連線未清除，等待後重試
+                    # 409 Conflict: 另一個 Pod 正在 polling，主動釋放 Leader Lock
+                    # 2026-04-01 Claude Code: 改為釋放 Lock 讓 Watcher 競爭
+                    # (舊: 侵略性搶佔 2s，已不適用 - 現在是多 Pod 場景而非 .188 搶佔)
                     logger.warning(
                         "telegram_polling_conflict",
                         status=409,
-                        message="409 衝突，等待殘留 session 過期...",
+                        pod_id=self._pod_id,
+                        action="releasing_leader_lock",
                     )
-                    # 2026-03-31 ogt: 侵略性搶佔模式 - 縮短等待以從 .188 手中奪回主權
-                    await asyncio.sleep(2)
+                    redis = await get_redis()
+                    current = await redis.get(POLLING_LEADER_KEY)
+                    if current == self._pod_id:
+                        await redis.delete(POLLING_LEADER_KEY)
+                    self._polling_active = False
+                    # Watcher 會在 POLLING_LEADER_WATCH 秒後重新競爭
+                    self._leader_task = asyncio.create_task(self._leader_watcher())
+                    break
                 else:
                     logger.error("telegram_polling_http_error", status=e.response.status_code)
                     await asyncio.sleep(LONG_POLLING_RETRY_DELAY)
@@ -2374,6 +2416,68 @@ class TelegramGateway:
 
         logger.info("telegram_long_polling_stopped")
 
+    async def _leader_renewer(self) -> None:
+        """
+        Leader Lock 續約背景任務
+
+        每 POLLING_LEADER_RENEW 秒更新 Redis TTL，
+        確保 Leader 在 Poll 期間持續持有 Lock。
+        若 Lock 被搶走，停止 Polling。
+
+        2026-04-01 Claude Code: 分散式 Leader Election
+        """
+        while self._polling_active:
+            await asyncio.sleep(POLLING_LEADER_RENEW)
+            if not self._polling_active:
+                break
+            try:
+                redis = await get_redis()
+                current = await redis.get(POLLING_LEADER_KEY)
+                if current != self._pod_id:
+                    logger.warning(
+                        "telegram_leader_lock_lost",
+                        pod_id=self._pod_id,
+                        current_leader=current,
+                    )
+                    self._polling_active = False
+                    break
+                await redis.expire(POLLING_LEADER_KEY, POLLING_LEADER_TTL)
+            except Exception as e:
+                logger.error("telegram_leader_renew_error", error=str(e))
+
+    async def _leader_watcher(self) -> None:
+        """
+        非 Leader Pod 的接管監控任務
+
+        每 POLLING_LEADER_WATCH 秒嘗試取得 Leader Lock。
+        若原 Leader 宕掉（TTL 過期），此 Pod 接管 Polling。
+
+        2026-04-01 Claude Code: 分散式 Leader Election
+        """
+        while not self._polling_active:
+            await asyncio.sleep(POLLING_LEADER_WATCH)
+            try:
+                redis = await get_redis()
+                acquired = await redis.set(
+                    POLLING_LEADER_KEY, self._pod_id, nx=True, ex=POLLING_LEADER_TTL
+                )
+                if acquired:
+                    logger.info(
+                        "telegram_leader_acquired",
+                        pod_id=self._pod_id,
+                        action="starting_polling",
+                    )
+                    await self._delete_webhook()
+                    self._polling_active = True
+                    self._last_update_id = 0
+                    self._polling_task = asyncio.create_task(self._polling_loop())
+                    self._leader_task = asyncio.create_task(self._leader_renewer())
+                    break
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                logger.error("telegram_leader_watch_error", error=str(e))
+
     async def _get_updates(self) -> list[dict]:
         """
         呼叫 Telegram getUpdates API