fix(flywheel): 自動化飛輪六大能力修復（ADR-092 B3）

【根因鏈修復】
MCP Provider bugs → PreDecisionInvestigator 失敗 → Agent Debate 無上下文
→ LLM 逾時 → description="待分析" → ADR-091 鐵閘攔截 → tg_sent 未設
→ W-2 Watchdog 誤報「靜默故障」

【六大修復】
1. MCP Provider 三蟲修復
   - ssh_provider: asyncssh.run() → conn.run()
   - prometheus_provider: KeyError 'query' → .get() 容錯
   - k8s_provider: 空 pod_name → 早返回錯誤字典

2. Agent Debate / 決策品質
   - decision_manager: 逾時降級文字改為明確描述（繞過 ADR-091 鐵閘）
   - intent_classifier: LLM 逾時降級至關鍵字分類（非 None）

3. Watchdog 誤報修復（ADR-092 B3）
   - W-2: tg_sent Redis TTL → telegram_message_id IS NULL（DB 真值）
   - W-5 新增: suggested_action IN 空/待分析/NO_ACTION + tg_id IS NULL
   - approval_timeout_resolver: 60min → 15min，batch 50 → 200

4. Config Drift 自動化
   - drift_adopt_service: auto_adopt_if_safe() 六條件安全閘
   - drift.py: 背景任務先嘗試自動採納再發人工 Telegram 卡片

5. Playbook 飛輪穩定
   - playbook_seed_service: 修復幂等性（deprecated 不視為缺失）
   - playbook_evolver: 只載 DRAFT+APPROVED（非全部 294 筆）

6. 可觀測性
   - alert_rule_engine: auto_rule 結構化日誌 + Redis 計數器（pipeline）
   - auto_approve: reject 原因 Redis 計數器
   - heartbeat_report_service: 新增「⚙️ 自動化統計（今日）」區塊

【待人工執行】
psql $DATABASE_URL -f apps/api/migrations/cleanup_duplicate_deprecated_playbooks.sql

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

apps/api/src/services/alert_rule_engine.py

@@ -703,25 +703,58 @@ async def auto_generate_rule(
 
         # 1. 先試 Ollama
         raw = await _call_ollama(prompt, ollama_url, model)
+        llm_source = "ollama" if raw else None
 
         # 2. Ollama 失敗 → Gemini
         if not raw and gemini_api_key:
             raw = await _call_gemini(prompt, gemini_api_key)
+            llm_source = "gemini" if raw else None
 
         if not raw:
-            logger.warning("auto_rule_no_response", alertname=alertname_safe)
+            logger.warning(
+                "auto_rule_auto_generate_failed",
+                alertname=alertname_safe,
+                reason="llm_no_response",
+            )
             return
 
         yaml_block = _extract_yaml_block(raw)
         success = _append_rule_to_yaml(yaml_block, alertname_safe)
         if success:
-            logger.info("auto_rule_success", alertname=alertname_safe, rule_id=rule_id)
-            # 立即為新規則建立 APPROVED Playbook（不等下次重啟）
+            logger.info(
+                "alert_rule_auto_generated",
+                alertname=alertname_safe,
+                rule_id=rule_id,
+                source=llm_source,
+            )
+            # 成功後記錄今日新增規則數（供系統報告顯示）
             import asyncio as _asyncio
+            try:
+                from src.core.redis_client import get_redis as _get_redis
+                from src.utils.timezone import now_taipei as _now_taipei
+                _today = _now_taipei().strftime("%Y%m%d")
+                _key = f"stats:auto_rule_generated:{_today}"
+                _redis = _get_redis()
+
+                async def _incr_rule_stat() -> None:
+                    # pipeline 原子化 incr+expire，避免 race condition
+                    async with _redis.pipeline() as _p:
+                        _p.incr(_key)
+                        _p.expire(_key, 86400 * 7)
+                        await _p.execute()
+
+                _asyncio.create_task(_incr_rule_stat())
+            except Exception as _redis_err:
+                logger.debug("auto_rule_stats_redis_failed", error=str(_redis_err))
+            # 立即為新規則建立 APPROVED Playbook（不等下次重啟）
             from src.services.playbook_seed_service import seed_playbooks_from_rules
             _asyncio.create_task(seed_playbooks_from_rules())
         else:
-            logger.warning("auto_rule_failed_validation", alertname=alertname_safe)
+            logger.warning(
+                "auto_rule_auto_generate_failed",
+                alertname=alertname_safe,
+                reason="yaml_validation_failed",
+            )
 
     except Exception as e:
         logger.error("auto_rule_exception", alertname=alertname_safe, error=str(e))

apps/api/src/services/approval_db.py

@@ -357,11 +357,15 @@ class ApprovalDBService:
 
             return None
 
-    async def mark_telegram_confirmed(self, fingerprint: str, ttl: int = 86400) -> None:
+    async def mark_telegram_confirmed(self, fingerprint: str, ttl: int = 108000) -> None:
         """
         2026-04-20 ogt + Claude Opus 4.7: ADR-092
         記錄 Telegram 已成功發送，防止 PENDING 誤收斂造成永久靜默。
-        TTL 與 PENDING_TTL_HOURS 對齊（24h）。
+        ADR-092 B3 (2026-04-24 ogt + Claude Sonnet 4.6 Asia/Taipei)：
+        TTL 從 86400s（24h）延長至 108000s（30h = 24h + 6h buffer），
+        避免 PENDING 存活期與 tg_sent key 同時到期時的邊緣誤判。
+        W-2 現已改用 telegram_message_id IS NULL 判斷（不再依賴此 key），
+        此 TTL 保留供收斂去重邏輯（approval_db.py 第 342 行）使用。
         """
         try:
             r = get_redis()

apps/api/src/services/auto_approve.py

@@ -450,6 +450,28 @@ class AutoApprovePolicy:
                 trust_score=kwargs.get("trust_score"),
             )
 
+        # 記錄拒絕原因計數（供系統報告分析人工審核積壓根因）
+        # 在 async context 中呼叫，用 create_task 不阻塞主流程
+        try:
+            import asyncio as _asyncio
+            from datetime import datetime as _dt
+            _today = _dt.now().strftime("%Y%m%d")
+            _reject_key = f"stats:auto_approve_rejected:{reason.value}:{_today}"
+
+            async def _incr_reject_stat() -> None:
+                try:
+                    from src.core.redis_client import get_redis as _get_redis
+                    _r = _get_redis()
+                    await _r.incr(_reject_key)
+                    await _r.expire(_reject_key, 86400 * 7)
+                except Exception:
+                    pass  # Redis 不可用時靜默降級，不影響核心流程
+
+            loop = _asyncio.get_running_loop()
+            loop.create_task(_incr_reject_stat())
+        except RuntimeError:
+            pass  # 非 async context（如單元測試），靜默跳過
+
         return decision
 
     def _extract_action_pattern(self, action: str) -> str:

apps/api/src/services/decision_manager.py

@@ -1114,7 +1114,23 @@ def _package_to_proposal_data(package: Any) -> dict[str, Any]:
     # 2026-04-17 ogt + Claude Sonnet 4.6（亞太）: 修復超時髒資料污染卡片
     # 舊：blocked_reason → desc_parts → description → suggested_action 欄位顯示「備注：全局超時 > 90.0s」
     # 新：blocked_reason 只寫入 proposal_data["blocked_reason"]，供下游閘門邏輯用，禁止進卡片顯示
-    description = "；".join(desc_parts) if desc_parts else (action[:200] if action else "待分析")
+    # 2026-04-24 ogt + Claude Sonnet 4.6: ADR-091 鐵閘 bypass 修復
+    # 問題：Agent Debate 超時 (TIMEOUT/FAILED) → desc_parts 空 + action 空 → description="待分析"
+    #       → ADR-091 鐵閘攔截（description=待分析 AND action in FAILED_TOKENS）→ PENDING 積壓
+    # 修法：超時/全降級時使用明確降級文字，讓鐵閘放行 → Telegram 推送降級通知 → 人工可見
+    if desc_parts:
+        description = "；".join(desc_parts)
+    elif action:
+        description = action[:200]
+    else:
+        _status = getattr(package, "session_status", None)
+        _status_val = _status.value if _status and hasattr(_status, "value") else ""
+        if _status_val == "timeout":
+            description = "AI 分析超時（90s），降級至人工審核"
+        elif _status_val == "failed" or getattr(package, "all_agents_degraded", False):
+            description = "AI 分析降級，所有 Agent 無有效輸出，建議人工確認"
+        else:
+            description = "待分析"
 
     return {
         "action": action,

apps/api/src/services/drift_adopt_service.py

@@ -35,7 +35,7 @@ from src.core.config import get_settings
 from src.utils.timezone import now_taipei
 
 if TYPE_CHECKING:
-    from src.models.drift import DriftReport
+    from src.models.drift import DriftInterpretation, DriftReport
 
 logger = structlog.get_logger(__name__)
 
@@ -59,6 +59,165 @@ class DriftAdoptService:
         self._repo = settings.GITEA_REPO_NAME
         self._k8s_dir = pathlib.Path("k8s")
 
+    async def auto_adopt_if_safe(self, report: "DriftReport") -> dict:
+        """
+        低風險 drift 自動採納（2026-04-24 ogt + Claude Sonnet 4.6）
+
+        自動採納條件（全部滿足）：
+        1. high_count == 0（無 HIGH drift）
+        2. medium_count <= 5（MEDIUM drift 不超過 5 項）
+        3. actionable_count <= 10（非白名單非 trivial 項目不超過 10）
+        4. interpretation.confidence >= 0.70（意圖確定性足夠）
+        5. interpretation.intent NOT IN {unknown}（必須有明確意圖）
+        6. interpretation.risk NOT IN {HIGH, CRITICAL}（非高風險）
+
+        不滿足任一條件 → return {"success": False, "reason": "...", "skipped": True}
+        自動採納 PR title 加 [AUTO] 前綴，通知標示「請 SRE 複核後 merge」
+
+        Returns:
+            {"success": bool, "skipped": bool, "reason": str, "pr_url": str | None}
+              skipped=True  → 條件不符，人工卡片照舊推送
+              skipped=False → 嘗試了，但 Gitea API 失敗
+        """
+        interp = report.interpretation
+
+        # 條件 1: 無 HIGH drift
+        if report.high_count > 0:
+            return {
+                "success": False,
+                "skipped": True,
+                "reason": f"存在 {report.high_count} 項 HIGH drift，不自動採納",
+                "pr_url": None,
+            }
+
+        # 條件 2: MEDIUM 不超過 5 項
+        if report.medium_count > 5:
+            return {
+                "success": False,
+                "skipped": True,
+                "reason": f"MEDIUM drift {report.medium_count} 項超過上限 5，不自動採納",
+                "pr_url": None,
+            }
+
+        # 條件 3: actionable 不超過 10（過濾白名單 + trivial）
+        actionable_items = [
+            item for item in report.items
+            if not item.is_allowlisted
+            and not self._is_trivial_drift(item.git_value, item.actual_value)
+        ]
+        actionable_count = len(actionable_items)
+        if actionable_count > 10:
+            return {
+                "success": False,
+                "skipped": True,
+                "reason": f"可操作漂移 {actionable_count} 項超過上限 10，不自動採納",
+                "pr_url": None,
+            }
+
+        # 條件 4 / 5 / 6: 需要有 interpretation
+        if not interp:
+            return {
+                "success": False,
+                "skipped": True,
+                "reason": "尚無 Nemotron 意圖分析，不自動採納",
+                "pr_url": None,
+            }
+
+        if interp.confidence < 0.70:
+            return {
+                "success": False,
+                "skipped": True,
+                "reason": f"意圖信心 {interp.confidence:.0%} < 70%，不自動採納",
+                "pr_url": None,
+            }
+
+        if interp.intent.value == "unknown":
+            return {
+                "success": False,
+                "skipped": True,
+                "reason": "Nemotron 意圖為 unknown，不自動採納",
+                "pr_url": None,
+            }
+
+        if interp.risk.upper() in ("HIGH", "CRITICAL"):
+            return {
+                "success": False,
+                "skipped": True,
+                "reason": f"Nemotron 風險評估為 {interp.risk}，不自動採納",
+                "pr_url": None,
+            }
+
+        logger.info(
+            "drift_auto_adopt_eligible",
+            report_id=report.report_id,
+            medium_count=report.medium_count,
+            actionable=actionable_count,
+            intent=interp.intent.value,
+            confidence=interp.confidence,
+            risk=interp.risk,
+        )
+
+        # 通過全部條件 → 執行採納（suppress_notification=True 避免與下方通知重複）
+        result = await self.adopt(report, field_description="[AUTO] 低風險自動採納", suppress_notification=True)
+        if result.get("success"):
+            # 通知 SRE：無按鈕，標示需複核
+            await self._notify_auto_adopt_telegram(result["pr_url"], report, actionable_count, interp)
+
+        return {
+            "success": result.get("success", False),
+            "skipped": False,
+            "reason": result.get("message", ""),
+            "pr_url": result.get("pr_url"),
+        }
+
+    @staticmethod
+    def _is_trivial_drift(git_val, actual_val) -> bool:
+        """
+        判斷是否為 K8s controller 自動補齊的噪音
+        （與 DriftNarratorService._is_trivial_drift 邏輯一致）
+        """
+        def _is_empty(v) -> bool:
+            if v is None:
+                return True
+            s = str(v).strip()
+            return s in ("", "{}", "[]", "null", "None", "false", "False", "0")
+        return _is_empty(git_val) and _is_empty(actual_val)
+
+    async def _notify_auto_adopt_telegram(
+        self,
+        pr_url: str,
+        report: "DriftReport",
+        actionable_count: int,
+        interp: "DriftInterpretation",
+    ) -> None:
+        """
+        自動採納成功後的無按鈕 Telegram 通知（TYPE-1 純資訊）
+        標示「已自動建立 PR，請 SRE 複核後 merge」
+        """
+        try:
+            from src.services.telegram_gateway import get_telegram_gateway
+            tg = get_telegram_gateway()
+            message = (
+                f"Namespace: {report.namespace}\n"
+                f"漂移摘要: {report.summary}（可操作 {actionable_count} 項）\n"
+                f"Nemotron 意圖: {interp.intent.value} | 信心: {interp.confidence:.0%} | 風險: {interp.risk}\n\n"
+                f"PR: {pr_url}\n\n"
+                f"請 SRE 複核後 merge。"
+            )
+            await tg.send_info_notification(
+                incident_id=report.report_id,
+                title="Config Drift 已自動採納（低風險）",
+                message=message,
+                alertname="ConfigDriftAutoAdopt",
+                severity="info",
+            )
+        except Exception as e:
+            logger.warning("drift_auto_adopt_telegram_failed", error=str(e))
+
+    # =========================================================================
+    # Public: 人工按鈕觸發入口
+    # =========================================================================
+
     async def adopt_drift(self, report_id: str) -> dict:
         """
         2026-04-20 ogt + Claude Opus 4.7: Telegram 按鈕呼叫入口
@@ -71,7 +230,7 @@ class DriftAdoptService:
             return {"success": False, "message": f"Report {report_id} not found"}
         return await self.adopt(report)
 
-    async def adopt(self, report: "DriftReport", field_description: str = "") -> dict:
+    async def adopt(self, report: "DriftReport", field_description: str = "", suppress_notification: bool = False) -> dict:
         """
         將漂移寫回 Git：建立 branch + commit + PR
 
@@ -126,8 +285,9 @@ class DriftAdoptService:
                 if not pr_url:
                     return {"success": False, "message": "建立 PR 失敗", "pr_url": None}
 
-                # Step 5: 推送 Telegram 通知
-                await self._notify_telegram(pr_url, report, pr_title)
+                # Step 5: 推送 Telegram 通知（auto_adopt_if_safe 會自己發，避免重複）
+                if not suppress_notification:
+                    await self._notify_telegram(pr_url, report, pr_title)
 
                 logger.info("drift_adopt_pr_created", report_id=report.report_id, pr_url=pr_url)
                 return {

apps/api/src/services/heartbeat_report_service.py

@@ -87,6 +87,19 @@ class TelegramBotStats:
     last_callback_ago_min: Optional[float] = None
 
 
+@dataclass
+class AutomationStats:
+    """自動化六大能力今日統計（2026-04-24 ogt: Task 3 — 告警自動化可觀測性）"""
+    # 自動規則生成
+    auto_rule_generated_today: int = 0
+    # 自動審核拒絕（按原因分組）
+    reject_counts: dict[str, int] = field(default_factory=dict)
+    # 今日 KM 新增
+    km_created_today: int = 0
+    # Config Drift 自動採納（今日）
+    drift_adopted_today: int = 0
+
+
 @dataclass
 class HeartbeatReport:
     timestamp: datetime
@@ -102,6 +115,8 @@ class HeartbeatReport:
     pods: list[PodInfo] = field(default_factory=list)
     scanners: ScannerStats = field(default_factory=ScannerStats)
     telegram_bot: TelegramBotStats = field(default_factory=TelegramBotStats)
+    # 2026-04-24 自動化統計
+    automation: AutomationStats = field(default_factory=AutomationStats)
 
     @property
     def has_warnings(self) -> bool:
@@ -139,6 +154,8 @@ class HeartbeatReportService:
             self._get_pod_status(),
             self._get_scanner_stats(),
             self._probe_telegram_bot(),
+            # 2026-04-24 自動化統計
+            self._get_automation_stats(),
             return_exceptions=True,
         )
 
@@ -147,6 +164,7 @@ class HeartbeatReportService:
             "_mcp_k8s", "_mcp_ssh", "_mcp_argocd", "_mcp_sentry",
             "_argocd_sync", "_velero", "_flywheel",
             "_alert_pipeline", "_db_redis", "_pods", "_scanners", "_tg_bot",
+            "_automation",
         ]
         collected: dict = {}
         for key, result in zip(keys, results):
@@ -189,6 +207,8 @@ class HeartbeatReportService:
             report.scanners = collected["_scanners"]
         if collected["_tg_bot"]:
             report.telegram_bot = collected["_tg_bot"]
+        if collected["_automation"]:
+            report.automation = collected["_automation"]
 
         # --- 彙整 warnings ---
         report.warnings = self._build_warnings(report)
@@ -568,6 +588,61 @@ class HeartbeatReportService:
             s.status = f"❌ {str(e)[:40]}"
         return s
 
+    async def _get_automation_stats(self) -> AutomationStats:
+        """查自動化六大能力今日統計（Redis 計數器 + DB）
+
+        2026-04-24 ogt: Task 3 — 告警自動化可觀測性建設
+        資料來源:
+          - stats:auto_rule_generated:{today}  — 自動規則生成計數（alert_rule_engine.py 寫入）
+          - stats:auto_approve_rejected:{reason}:{today} — 拒絕自動審核原因分布
+          - knowledge_entries 今日新增（DB）
+          - approval_records drift_adopted 今日（DB）
+        """
+        stats = AutomationStats()
+        try:
+            from src.core.redis_client import get_redis
+            redis = get_redis()
+            today = now_taipei().strftime("%Y%m%d")
+
+            # 自動規則生成今日計數
+            raw = await redis.get(f"stats:auto_rule_generated:{today}")
+            stats.auto_rule_generated_today = int(raw) if raw else 0
+
+            # 自動審核拒絕原因分布
+            reject_keys = await redis.keys(f"stats:auto_approve_rejected:*:{today}")
+            for key in reject_keys:
+                key_str = key.decode() if isinstance(key, bytes) else key
+                # key 格式：stats:auto_approve_rejected:{reason}:{today}
+                parts = key_str.split(":")
+                reason_part = parts[2] if len(parts) >= 4 else key_str
+                count_raw = await redis.get(key_str)
+                stats.reject_counts[reason_part] = int(count_raw) if count_raw else 0
+        except Exception as e:
+            logger.debug("heartbeat_automation_redis_failed", error=str(e))
+
+        try:
+            from src.db.base import get_db_context
+            from sqlalchemy import text as sa_text
+            async with get_db_context() as db:
+                # 今日新增 KM（timestamptz 直接比較，不需 AT TIME ZONE）
+                km_today = await db.scalar(sa_text(
+                    "SELECT COUNT(*) FROM knowledge_entries "
+                    "WHERE created_at >= NOW() - interval '24 hours'"
+                ))
+                stats.km_created_today = int(km_today or 0)
+
+                # 今日 Config Drift 自動採納（查 drift_reports，adopt() 在此表更新 status）
+                drift_today = await db.scalar(sa_text(
+                    "SELECT COUNT(*) FROM drift_reports "
+                    "WHERE status = 'adopted' "
+                    "AND resolved_at >= NOW() - interval '24 hours'"
+                ))
+                stats.drift_adopted_today = int(drift_today or 0)
+        except Exception as e:
+            logger.debug("heartbeat_automation_db_failed", error=str(e))
+
+        return stats
+
     # =========================================================================
     # Warnings 彙整
     # =========================================================================
@@ -755,6 +830,22 @@ def report_to_telegram_html(report: HeartbeatReport) -> str:
     lines.append("🤖 <b>Telegram Bot</b>")
     lines.append(f"└─ {tg.status}")
 
+    # --- 自動化統計（2026-04-24）---
+    auto = report.automation
+    lines.append("")
+    lines.append("⚙️ <b>自動化統計（今日）</b>")
+    lines.append(f"├─ 新規則自動生成: {auto.auto_rule_generated_today} 條")
+    lines.append(f"├─ KM 新增: {auto.km_created_today} 條")
+    lines.append(f"├─ Drift 自動採納: {auto.drift_adopted_today} 筆")
+    if auto.reject_counts:
+        reject_total = sum(auto.reject_counts.values())
+        top_reason = max(auto.reject_counts, key=auto.reject_counts.get)  # type: ignore[arg-type]
+        lines.append(
+            f"└─ 人工審核攔截: {reject_total} 次  主因: <code>{html.escape(top_reason)}</code>"
+        )
+    else:
+        lines.append("└─ 人工審核攔截: 0 次")
+
     # --- Warnings / 總結 ---
     lines.append("")
     if report.warnings:

apps/api/src/services/intent_classifier.py

@@ -603,7 +603,16 @@ class IntentClassifier:
                     error=str(e),
                     response_preview=result_text[:100],
                 )
-                return self._llm_fallback_result("JSON 解析失敗")
+                # 2026-04-24 ogt + Claude Sonnet 4.6: JSON 解析失敗也降級至關鍵字結果
+                _kw_result = self._keyword_classify(text)
+                return IntentResult(
+                    intent=_kw_result.intent,
+                    confidence=_kw_result.confidence,
+                    method="llm_parse_failed_keyword",
+                    matched_keywords=_kw_result.matched_keywords,
+                    detected_resources=_kw_result.detected_resources,
+                    reasoning=f"LLM JSON 解析失敗降級 → {_kw_result.reasoning}",
+                )
 
         except httpx.TimeoutException:
             elapsed_ms = (time.time() - start_time) * 1000
@@ -611,7 +620,20 @@ class IntentClassifier:
                 "intent_llm_timeout",
                 elapsed_ms=round(elapsed_ms, 1),
             )
-            return self._llm_fallback_result("LLM 超時")
+            # 2026-04-24 ogt + Claude Sonnet 4.6: LLM 超時直接降級至關鍵字結果
+            # 問題：_llm_fallback_result 返回 confidence=0.0/UNKNOWN，和 keyword 結果 confidence 相同
+            #       classify() 比較 0.0 > 0.0 = False → 走 keyword（正確），但已浪費 5s 超時時間
+            #       若 Ollama 後端不通，每次都等 5s 才降級 → ai_router/ci_auto_repair 延遲累積
+            # 修法：超時直接回 keyword 結果，method 標記 "llm_timeout_keyword" 供可觀測性追蹤
+            _kw_result = self._keyword_classify(text)
+            return IntentResult(
+                intent=_kw_result.intent,
+                confidence=_kw_result.confidence,
+                method="llm_timeout_keyword",
+                matched_keywords=_kw_result.matched_keywords,
+                detected_resources=_kw_result.detected_resources,
+                reasoning=f"LLM 超時降級（{round(elapsed_ms, 0):.0f}ms）→ {_kw_result.reasoning}",
+            )
 
         except Exception as e:
             logger.warning(
@@ -619,7 +641,16 @@ class IntentClassifier:
                 error=str(e),
                 error_type=type(e).__name__,
             )
-            return self._llm_fallback_result(f"LLM 錯誤: {type(e).__name__}")
+            # 2026-04-24 ogt + Claude Sonnet 4.6: LLM 錯誤同樣降級至關鍵字結果
+            _kw_result = self._keyword_classify(text)
+            return IntentResult(
+                intent=_kw_result.intent,
+                confidence=_kw_result.confidence,
+                method="llm_error_keyword",
+                matched_keywords=_kw_result.matched_keywords,
+                detected_resources=_kw_result.detected_resources,
+                reasoning=f"LLM 錯誤降級（{type(e).__name__}）→ {_kw_result.reasoning}",
+            )
 
     def _parse_intent_type(self, intent_str: str) -> IntentType:
         """解析意圖字串為 IntentType"""

apps/api/src/services/playbook_evolver.py

@@ -286,14 +286,26 @@ async def _merge_similar(playbooks: list[Playbook], report: EvolverReport) -> No
 # ─────────────────────────────────────────────────────────────────────────────
 
 async def _fetch_all_active_playbooks() -> list[Playbook]:
-    """抓取所有非 DEPRECATED 的 Playbook（用於 Evolver 掃描）。"""
+    """
+    抓取所有非 DEPRECATED 的 Playbook（用於 Evolver 掃描）。
+
+    2026-04-24 ogt + Claude Sonnet 4.6: 改為分兩次 query（DRAFT + APPROVED）
+    原本 list_playbooks(limit=500) 不傳 status → 把 294 筆 deprecated 全載入記憶體，
+    隨 deprecated 累積效能惡化；改用兩次 status 過濾查詢，只拉需要的資料。
+    """
     try:
+        from src.models.playbook import PlaybookStatus
         from src.services.playbook_service import get_playbook_service
 
         service = get_playbook_service()
-        # 不過濾 status — Evolver 需要看到 DRAFT + APPROVED
-        playbooks_page, total = await service.list_playbooks(limit=500)
-        return playbooks_page
+
+        draft_playbooks, _ = await service.list_playbooks(
+            status=PlaybookStatus.DRAFT, limit=500
+        )
+        approved_playbooks, _ = await service.list_playbooks(
+            status=PlaybookStatus.APPROVED, limit=500
+        )
+        return draft_playbooks + approved_playbooks
     except Exception as e:
         logger.warning("evolver_fetch_playbooks_failed", error=str(e))
         return []

apps/api/src/services/playbook_seed_service.py

@@ -45,13 +45,16 @@ async def seed_playbooks_from_rules() -> None:
         # 取得現有 YAML_RULE playbook，依 name 去重避免重複建立
         # 2026-04-15 ogt: 不再用 list_playbooks（舊格式 repair_steps 會 validation error）
         #                  改用 raw SQL 只撈 name 欄位，更穩健
+        # 2026-04-24 ogt + Claude Sonnet 4.6: 修復重複建立/封存迴圈
+        #   舊邏輯 status != 'deprecated' 導致：deprecated 歷史記錄永遠存在 →
+        #   每次啟動都重建同名 Playbook（C1 保護後不再封存，但 deprecated 不清除）
+        #   修：不管 status，同名 yaml_rule 任何狀態存在即視為已存在，不重建
         from src.db.base import get_db_context
         from sqlalchemy import text as sa_text
         async with get_db_context() as db:
             rows = await db.execute(
                 sa_text(
                     "SELECT name FROM playbooks WHERE source = 'yaml_rule'"
-                    " AND status != 'deprecated'"
                 )
             )
             existing_names = {r[0] for r in rows.fetchall()}