feat(Phase 3.5 + Phase 4): AI 學習成果持久化到 PostgreSQL — 修正「AI 失憶」架構缺陷

ADR-085: AI 學習成果不可存在 Cache

架構鐵律確立：
- PostgreSQL = System of Record（AI 的永久記憶）
- Redis = Warm Cache（加速讀取，TTL 到期從 PG 復原）

核心變更：
1. models.py: 新增 PlaybookRecord / DynamicBaselineRecord / LogClusterRecord ORM
2. base.py: ALTER TABLE playbooks 補加 trust_score / requires_approval_level 等欄位
3. playbook_repository.py: 完整雙寫實作（PG upsert + Redis cache）
4. dynamic_baseline_service.py: Holt-Winters 訓練結果寫入 PG，Redis 只作 24h warm cache
5. log_anomaly_detector.py: Drain3 cluster template 寫入 PG（UPSERT on cluster_id）
6. main.py: 啟動時執行 backfill_redis_to_pg()（Redis → PG 冪等補救）

修正的問題：
- Playbook 7天 Redis TTL 到期 → AI 失去所有修復知識
- trust_score EWMA 隨 Redis TTL 歸零 → AI 重新回到初始信任度 0.3
- Holt-Winters 基線 24h TTL → AI 每天重新學習「正常」的定義
- Drain3 cluster 沒有持久化 → AI 把已知 log pattern 反覆當新 pattern

Phase 4 新服務（requirements.txt 已加入 statsmodels + drain3 + numpy）

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

apps/api/src/repositories/playbook_repository.py

@@ -1,26 +1,31 @@
 """
-Playbook Repository - #7 Playbook 萃取
-======================================
-Playbook CRUD 操作 (Redis + PostgreSQL)
+Playbook Repository — Phase 3.5 雙寫持久化
+==========================================
+ADR-085: AI 學習成果必須儲存到 PostgreSQL
+
+儲存策略（Phase 3.5 架構升級）：
+- PostgreSQL = System of Record（永久保存，source of truth）
+- Redis = Warm Cache（7天 TTL，加速讀取，PG 為準）
+
+雙寫規則：
+- create/update → 先寫 PG，再更新 Redis cache
+- 讀取 → Redis-first（cache hit），miss 時從 PG 載入並回填 Redis
+- trust_score / EWMA 統計 → PG 永久保存，Redis TTL 到期後從 PG 復原
 
 Phase 7.2: Repository 實作
-建立時間: 2026-03-26 (台北時區)
-建立者: Claude Code (#7 Playbook 萃取)
-
-遵循 leWOOOgo 積木化原則:
-- 實作 IPlaybookRepository Protocol
-- Redis 為 Working Memory (7天 TTL)
-- PostgreSQL 為 Episodic Memory
-
 Phase 22 P2: 相似度計算邏輯移至 utils
-2026-03-31 Claude Code (首席架構師技術債修復)
+Phase 3.5 (ADR-085): PostgreSQL dual-write + Redis warm cache
+2026-04-15 ogt + Claude Sonnet 4.6（亞太）: Phase 3.5 重寫，修正「AI 失憶」架構問題
 """
 
 import json
 
 import structlog
+from sqlalchemy import select
 
 from src.core.redis_client import get_redis
+from src.db.base import get_session_factory
+from src.db.models import PlaybookRecord
 from src.models.playbook import (
     Playbook,
     PlaybookStatus,
@@ -32,7 +37,7 @@ from src.utils.timezone import now_taipei
 
 logger = structlog.get_logger(__name__)
 
-# Redis TTL: 7 天
+# Redis TTL: 7 天（warm cache only，PG 為 source of truth）
 PLAYBOOK_TTL_SECONDS = 7 * 24 * 60 * 60
 
 # Redis Key 前綴
@@ -41,129 +46,175 @@ PLAYBOOK_INDEX_ALERT_PREFIX = "playbook:index:alert:"
 PLAYBOOK_INDEX_SERVICE_PREFIX = "playbook:index:service:"
 
 
+# =============================================================================
+# ORM ↔ Pydantic 轉換 Helpers
+# =============================================================================
+
+def _pydantic_to_orm(playbook: Playbook) -> PlaybookRecord:
+    """Pydantic Playbook → PlaybookRecord ORM"""
+    return PlaybookRecord(
+        playbook_id=playbook.playbook_id,
+        name=playbook.name,
+        description=playbook.description,
+        status=playbook.status.value,
+        source=playbook.source.value,
+        symptom_pattern=playbook.symptom_pattern.model_dump(),
+        repair_steps=[s.model_dump() for s in playbook.repair_steps],
+        estimated_duration_minutes=playbook.estimated_duration_minutes,
+        source_incident_ids=playbook.source_incident_ids,
+        ai_confidence=playbook.ai_confidence,
+        success_count=playbook.success_count,
+        failure_count=playbook.failure_count,
+        last_used_at=playbook.last_used_at,
+        trust_score=playbook.trust_score,
+        approved_by=playbook.approved_by,
+        approved_at=playbook.approved_at,
+        tags=playbook.tags,
+        notes=playbook.notes,
+        requires_approval_level=playbook.requires_approval_level,
+        stateful_targets=playbook.stateful_targets,
+        requires_pre_backup=playbook.requires_pre_backup,
+        created_at=playbook.created_at,
+        updated_at=playbook.updated_at,
+    )
+
+
+def _orm_to_pydantic(record: PlaybookRecord) -> Playbook:
+    """PlaybookRecord ORM → Pydantic Playbook"""
+    return Playbook.model_validate({
+        "playbook_id": record.playbook_id,
+        "name": record.name,
+        "description": record.description,
+        "status": record.status,
+        "source": record.source,
+        "symptom_pattern": record.symptom_pattern,
+        "repair_steps": record.repair_steps,
+        "estimated_duration_minutes": record.estimated_duration_minutes,
+        "source_incident_ids": record.source_incident_ids,
+        "ai_confidence": float(record.ai_confidence),
+        "success_count": record.success_count,
+        "failure_count": record.failure_count,
+        "last_used_at": record.last_used_at,
+        "trust_score": float(record.trust_score),
+        "approved_by": record.approved_by,
+        "approved_at": record.approved_at,
+        "tags": record.tags,
+        "notes": record.notes,
+        "requires_approval_level": record.requires_approval_level,
+        "stateful_targets": record.stateful_targets,
+        "requires_pre_backup": record.requires_pre_backup,
+        "created_at": record.created_at,
+        "updated_at": record.updated_at,
+    })
+
+
 class PlaybookRepository:
     """
-    Playbook Repository 實作
+    Playbook Repository — Phase 3.5 雙寫實作
 
     儲存策略:
-    - Redis: Working Memory (快速讀取，7天 TTL)
-    - PostgreSQL: Episodic Memory (持久化，待實作)
+    - PostgreSQL: source of truth（所有 create/update 必須先寫 PG）
+    - Redis: warm cache（加速讀取，7d TTL，到期後從 PG 復原）
 
-    Phase 7.2 先實作 Redis 層，PostgreSQL 待 #7.5 整合
+    這保證：
+    - trust_score / EWMA 永久不丟失
+    - Pod 重啟後 AI 學習成果完整保留
+    - Redis 失效只影響讀取速度，不影響資料完整性
     """
 
-    # === CRUD Operations ===
+    # =========================================================================
+    # CRUD Operations
+    # =========================================================================
 
     async def create(self, playbook: Playbook) -> Playbook:
         """
         建立新的 Playbook
 
-        1. 儲存到 Redis
-        2. 建立索引 (alert_names, services)
+        順序：
+        1. 寫入 PostgreSQL（source of truth）
+        2. 寫入 Redis cache（warm cache）
+        3. 更新索引
         """
-        try:
-            redis_client = get_redis()
+        if not playbook.created_at:
+            playbook.created_at = now_taipei()
+        playbook.updated_at = now_taipei()
 
-            # 確保有建立時間
-            if not playbook.created_at:
-                playbook.created_at = now_taipei()
-            playbook.updated_at = now_taipei()
+        # 1. 寫入 PostgreSQL（先寫，確保資料不丟失）
+        await self._pg_upsert(playbook)
 
-            # 儲存 Playbook
-            key = f"{PLAYBOOK_KEY_PREFIX}{playbook.playbook_id}"
-            await redis_client.set(
-                key,
-                json.dumps(playbook.to_redis_dict(), ensure_ascii=False),
-                ex=PLAYBOOK_TTL_SECONDS,
-            )
+        # 2. 寫入 Redis（warm cache）
+        await self._redis_set(playbook)
 
-            # 建立索引
-            await self._update_indexes(playbook)
+        # 3. 建立索引
+        await self._update_indexes(playbook)
 
-            logger.info(
-                "playbook_created",
-                playbook_id=playbook.playbook_id,
-                name=playbook.name,
-            )
-            return playbook
-
-        except Exception as e:
-            logger.error("playbook_create_failed", error=str(e))
-            raise
+        logger.info("playbook_created", playbook_id=playbook.playbook_id, name=playbook.name)
+        return playbook
 
     async def get_by_id(self, playbook_id: str) -> Playbook | None:
-        """根據 ID 取得 Playbook"""
-        try:
-            redis_client = get_redis()
-            key = f"{PLAYBOOK_KEY_PREFIX}{playbook_id}"
-            data = await redis_client.get(key)
+        """
+        根據 ID 取得 Playbook
 
-            if data:
-                return Playbook.from_redis_dict(json.loads(data))
-            return None
+        Redis-first → miss 時從 PG 載入並回填 Redis
+        """
+        # 1. Redis cache hit
+        cached = await self._redis_get(playbook_id)
+        if cached is not None:
+            return cached
 
-        except Exception as e:
-            logger.error("playbook_get_failed", playbook_id=playbook_id, error=str(e))
-            return None
+        # 2. PG fallback（cache miss）
+        playbook = await self._pg_get(playbook_id)
+        if playbook is not None:
+            # 回填 Redis cache
+            await self._redis_set(playbook)
+        return playbook
 
     async def update(self, playbook: Playbook) -> Playbook | None:
-        """更新 Playbook"""
-        try:
-            existing = await self.get_by_id(playbook.playbook_id)
-            if not existing:
-                return None
+        """
+        更新 Playbook
 
-            playbook.updated_at = now_taipei()
-
-            redis_client = get_redis()
-            key = f"{PLAYBOOK_KEY_PREFIX}{playbook.playbook_id}"
-            await redis_client.set(
-                key,
-                json.dumps(playbook.to_redis_dict(), ensure_ascii=False),
-                ex=PLAYBOOK_TTL_SECONDS,
-            )
-
-            # 更新索引
-            await self._update_indexes(playbook)
-
-            logger.info("playbook_updated", playbook_id=playbook.playbook_id)
-            return playbook
-
-        except Exception as e:
-            logger.error(
-                "playbook_update_failed",
+        先確認 PG 存在 → 寫入 PG → 更新 Redis
+        """
+        existing = await self._pg_get(playbook.playbook_id)
+        if existing is None:
+            # PG 找不到，可能是老資料只在 Redis：先建立 PG 記錄
+            logger.warning(
+                "playbook_pg_not_found_creating",
                 playbook_id=playbook.playbook_id,
-                error=str(e),
             )
-            return None
+
+        playbook.updated_at = now_taipei()
+
+        # 1. 寫入 PG
+        await self._pg_upsert(playbook)
+
+        # 2. 更新 Redis
+        await self._redis_set(playbook)
+
+        # 3. 更新索引
+        await self._update_indexes(playbook)
+
+        logger.info("playbook_updated", playbook_id=playbook.playbook_id)
+        return playbook
 
     async def delete(self, playbook_id: str) -> bool:
         """
-        刪除 Playbook (軟刪除 → DEPRECATED)
-
-        不真正刪除，而是將狀態改為 DEPRECATED
+        軟刪除 Playbook（狀態改為 DEPRECATED）
         """
-        try:
-            playbook = await self.get_by_id(playbook_id)
-            if not playbook:
-                return False
-
-            playbook.status = PlaybookStatus.DEPRECATED
-            playbook.updated_at = now_taipei()
-            await self.update(playbook)
-
-            logger.info("playbook_deprecated", playbook_id=playbook_id)
-            return True
-
-        except Exception as e:
-            logger.error(
-                "playbook_delete_failed",
-                playbook_id=playbook_id,
-                error=str(e),
-            )
+        playbook = await self.get_by_id(playbook_id)
+        if not playbook:
             return False
 
-    # === Query Operations ===
+        playbook.status = PlaybookStatus.DEPRECATED
+        playbook.updated_at = now_taipei()
+        await self.update(playbook)
+
+        logger.info("playbook_deprecated", playbook_id=playbook_id)
+        return True
+
+    # =========================================================================
+    # Query Operations
+    # =========================================================================
 
     async def list_playbooks(
         self,
@@ -173,42 +224,32 @@ class PlaybookRepository:
         offset: int = 0,
     ) -> tuple[list[Playbook], int]:
         """
-        列出 Playbooks
+        列出 Playbooks（從 PostgreSQL 查詢，不走 Redis scan）
 
-        注意: Redis 實作效率較低，後續需遷移到 PostgreSQL
+        Phase 3.5：改用 PG 查詢，效率更高，資料更完整
         """
         try:
-            redis_client = get_redis()
+            factory = get_session_factory()
+            async with factory() as session:
+                stmt = select(PlaybookRecord)
+                if status is not None:
+                    stmt = stmt.where(PlaybookRecord.status == status.value)
+                stmt = stmt.order_by(PlaybookRecord.updated_at.desc())
 
-            # 掃描所有 Playbook keys
-            pattern = f"{PLAYBOOK_KEY_PREFIX}PB-*"
-            keys = []
-            async for key in redis_client.scan_iter(match=pattern, count=100):
-                keys.append(key)
+                result = await session.execute(stmt)
+                all_records = result.scalars().all()
 
-            # 讀取並過濾
-            all_playbooks: list[Playbook] = []
-            for key in keys:
-                data = await redis_client.get(key)
-                if data:
-                    playbook = Playbook.from_redis_dict(json.loads(data))
+            all_playbooks = [_orm_to_pydantic(r) for r in all_records]
 
-                    # 狀態過濾
-                    if status and playbook.status != status:
-                        continue
-
-                    # 標籤過濾
-                    if tags and not set(tags).intersection(set(playbook.tags)):
-                        continue
-
-                    all_playbooks.append(playbook)
-
-            # 排序: 按 updated_at 降序
-            all_playbooks.sort(key=lambda p: p.updated_at, reverse=True)
+            # 標籤過濾（PG JSONB 過濾後續可移到 SQL，目前 Python 過濾）
+            if tags:
+                all_playbooks = [
+                    p for p in all_playbooks
+                    if set(tags).intersection(set(p.tags))
+                ]
 
             total = len(all_playbooks)
-            items = all_playbooks[offset : offset + limit]
-
+            items = all_playbooks[offset: offset + limit]
             return items, total
 
         except Exception as e:
@@ -225,7 +266,7 @@ class PlaybookRepository:
         根據症狀模式找相似 Playbook
 
         策略:
-        1. 從索引快速過濾候選
+        1. 從 Redis 索引快速過濾候選
         2. 計算詳細相似度
         3. 返回 Top K
         """
@@ -235,24 +276,19 @@ class PlaybookRepository:
             # 1. 使用索引找候選 Playbook IDs
             candidate_ids: set[str] = set()
 
-            # 從 alert_names 索引查詢
             for alert_name in symptoms.alert_names:
                 index_key = f"{PLAYBOOK_INDEX_ALERT_PREFIX}{alert_name}"
                 members = await redis_client.smembers(index_key)
                 candidate_ids.update(m.decode() if isinstance(m, bytes) else m for m in members)
 
-            # 從 services 索引查詢
             for service in symptoms.affected_services:
                 index_key = f"{PLAYBOOK_INDEX_SERVICE_PREFIX}{service}"
                 members = await redis_client.smembers(index_key)
                 candidate_ids.update(m.decode() if isinstance(m, bytes) else m for m in members)
 
-            # 如果沒有索引命中，掃描所有 APPROVED Playbooks
+            # 索引無命中 → 從 PG 掃 APPROVED
             if not candidate_ids:
-                playbooks, _ = await self.list_playbooks(
-                    status=PlaybookStatus.APPROVED,
-                    limit=100,
-                )
+                playbooks, _ = await self.list_playbooks(status=PlaybookStatus.APPROVED, limit=100)
                 candidate_ids = {p.playbook_id for p in playbooks}
 
             # 2. 計算相似度
@@ -260,26 +296,16 @@ class PlaybookRepository:
 
             for playbook_id in candidate_ids:
                 playbook = await self.get_by_id(playbook_id)
-                if not playbook:
+                if not playbook or playbook.status != PlaybookStatus.APPROVED:
                     continue
 
-                # 只考慮 APPROVED 狀態
-                if playbook.status != PlaybookStatus.APPROVED:
-                    continue
-
-                similarity = calculate_symptom_similarity(
-                    symptoms,
-                    playbook.symptom_pattern,
-                )
-
-                # alert_names 完全匹配時，保證通過（不因其他維度拉低分數）
+                similarity = calculate_symptom_similarity(symptoms, playbook.symptom_pattern)
                 alert_exact_match = bool(
                     set(symptoms.alert_names) & set(playbook.symptom_pattern.alert_names)
                 )
                 if alert_exact_match or similarity >= min_similarity:
                     results.append((playbook, similarity))
 
-            # 3. 排序並返回 Top K
             results.sort(key=lambda x: x[1], reverse=True)
             return results[:top_k]
 
@@ -293,13 +319,13 @@ class PlaybookRepository:
         success: bool,
     ) -> bool:
         """
-        更新執行統計 + EWMA 信任度
+        更新執行統計 + EWMA 信任度（雙寫到 PG + Redis）
 
         ADR-083 Phase 3: 負向 2x 強化 EWMA 公式
         成功: trust_new = 0.9 * trust_old + 0.1 * 1.0
         失敗: trust_new = 0.8 * trust_old + 0.2 * 0.0（衰減速度 2x）
         trust < 0.1 → 記錄警告，由 Evolver Agent 封存
-        2026-04-15 ogt + Claude Sonnet 4.6（亞太）: Phase 3 EWMA 實裝
+        2026-04-15 ogt + Claude Sonnet 4.6（亞太）: Phase 3.5 EWMA 雙寫 PG
         """
         try:
             playbook = await self.get_by_id(playbook_id)
@@ -308,16 +334,15 @@ class PlaybookRepository:
 
             if success:
                 playbook.success_count += 1
-                # 正向 EWMA：alpha=0.1，正向結果權重較小（保守更新）
                 playbook.trust_score = 0.9 * playbook.trust_score + 0.1 * 1.0
             else:
                 playbook.failure_count += 1
-                # 負向 EWMA：alpha=0.2，失敗懲罰 2x（快速衰退）
                 playbook.trust_score = 0.8 * playbook.trust_score + 0.2 * 0.0
 
-            # 邊界保護
             playbook.trust_score = max(0.0, min(1.0, playbook.trust_score))
             playbook.last_used_at = now_taipei()
+
+            # 雙寫（PG + Redis）
             await self.update(playbook)
 
             if playbook.trust_score < 0.1:
@@ -338,70 +363,25 @@ class PlaybookRepository:
             return True
 
         except Exception as e:
-            logger.error(
-                "playbook_stats_update_failed",
-                playbook_id=playbook_id,
-                error=str(e),
-            )
+            logger.error("playbook_stats_update_failed", playbook_id=playbook_id, error=str(e))
             return False
 
-    # === Index Management ===
-
-    async def _update_indexes(self, playbook: Playbook) -> None:
-        """更新索引"""
+    async def find_by_source_incident(self, incident_id: str) -> list[Playbook]:
+        """根據來源 Incident ID 找 Playbook（從 PG 查詢）"""
         try:
-            redis_client = get_redis()
-
-            # Alert names 索引
-            for alert_name in playbook.symptom_pattern.alert_names:
-                index_key = f"{PLAYBOOK_INDEX_ALERT_PREFIX}{alert_name}"
-                await redis_client.sadd(index_key, playbook.playbook_id)
-                await redis_client.expire(index_key, PLAYBOOK_TTL_SECONDS)
-
-            # Services 索引
-            for service in playbook.symptom_pattern.affected_services:
-                index_key = f"{PLAYBOOK_INDEX_SERVICE_PREFIX}{service}"
-                await redis_client.sadd(index_key, playbook.playbook_id)
-                await redis_client.expire(index_key, PLAYBOOK_TTL_SECONDS)
-
+            factory = get_session_factory()
+            async with factory() as session:
+                # PG JSONB contains 查詢
+                stmt = select(PlaybookRecord).where(
+                    PlaybookRecord.source_incident_ids.contains([incident_id])
+                )
+                result = await session.execute(stmt)
+                records = result.scalars().all()
+            return [_orm_to_pydantic(r) for r in records]
         except Exception as e:
-            logger.warning("playbook_index_update_failed", error=str(e))
-
-    # === Learning Service 信心度調整 (2026-03-30 Claude Code) ===
-
-    async def find_by_source_incident(
-        self,
-        incident_id: str,
-    ) -> list[Playbook]:
-        """
-        根據來源 Incident ID 找 Playbook
-
-        2026-03-30 Claude Code: Learning Service 信心度調整用
-        尋找 source_incident_ids 包含此 incident_id 的 Playbooks
-        """
-        try:
-            redis_client = get_redis()
-
-            # 掃描所有 Playbook keys
-            pattern = f"{PLAYBOOK_KEY_PREFIX}PB-*"
-            results: list[Playbook] = []
-
-            async for key in redis_client.scan_iter(match=pattern, count=100):
-                data = await redis_client.get(key)
-                if data:
-                    playbook = Playbook.from_redis_dict(json.loads(data))
-                    if incident_id in playbook.source_incident_ids:
-                        results.append(playbook)
-
-            return results
-
-        except Exception as e:
-            logger.error(
-                "playbook_find_by_incident_failed",
-                incident_id=incident_id,
-                error=str(e),
-            )
-            return []
+            logger.error("playbook_find_by_incident_failed", incident_id=incident_id, error=str(e))
+            # Fallback: Redis scan（維持向下相容）
+            return await self._redis_find_by_incident(incident_id)
 
     async def adjust_confidence(
         self,
@@ -410,9 +390,7 @@ class PlaybookRepository:
         reason: str,
     ) -> Playbook | None:
         """
-        調整 Playbook 信心度
-
-        2026-03-30 Claude Code: Learning Service 信心度調整用
+        調整 Playbook 信心度（雙寫到 PG + Redis）
 
         邏輯:
         - ai_confidence += delta (clamp 到 0.0~1.0)
@@ -425,25 +403,16 @@ class PlaybookRepository:
                 return None
 
             old_confidence = playbook.ai_confidence
-
-            # 調整信心度 (clamp 到 0.0~1.0)
             playbook.ai_confidence = max(0.0, min(1.0, playbook.ai_confidence + delta))
-
-            # 狀態自動轉換
             old_status = playbook.status
 
-            # 高信心度自動升級
-            if (
-                playbook.ai_confidence >= 0.9
-                and playbook.status == PlaybookStatus.DRAFT
-            ):
+            if playbook.ai_confidence >= 0.9 and playbook.status == PlaybookStatus.DRAFT:
                 playbook.status = PlaybookStatus.APPROVED
                 playbook.approved_by = "auto_learning"
                 playbook.approved_at = now_taipei()
                 note = f"\n[Auto-approved: confidence {playbook.ai_confidence:.2f}]"
                 playbook.notes = (playbook.notes or "") + note
 
-            # 低信心度 + 高失敗率 → 棄用
             elif (
                 playbook.ai_confidence < 0.3
                 and playbook.total_executions >= 5
@@ -455,7 +424,6 @@ class PlaybookRepository:
                 note = f"\n[Auto-deprecated: conf={conf:.2f}, fail={fail:.0%}]"
                 playbook.notes = (playbook.notes or "") + note
 
-            # 儲存
             updated = await self.update(playbook)
 
             logger.info(
@@ -467,18 +435,190 @@ class PlaybookRepository:
                 new_status=playbook.status.value,
                 reason=reason,
             )
-
             return updated
 
         except Exception as e:
-            logger.error(
-                "playbook_confidence_adjust_failed",
-                playbook_id=playbook_id,
-                delta=delta,
-                error=str(e),
-            )
+            logger.error("playbook_confidence_adjust_failed", playbook_id=playbook_id, delta=delta, error=str(e))
             return None
 
+    # =========================================================================
+    # Startup Backfill — Redis → PG 遷移
+    # =========================================================================
+
+    async def backfill_redis_to_pg(self) -> int:
+        """
+        啟動時將 Redis 中存在但 PG 中沒有的 Playbook 寫入 PG。
+
+        用途：Phase 3.5 上線前後，確保舊 Redis 資料不丟失。
+        返回：補寫的 Playbook 數量。
+        """
+        count = 0
+        try:
+            redis_client = get_redis()
+            pattern = f"{PLAYBOOK_KEY_PREFIX}PB-*"
+
+            async for key in redis_client.scan_iter(match=pattern, count=100):
+                data = await redis_client.get(key)
+                if not data:
+                    continue
+
+                try:
+                    playbook = Playbook.from_redis_dict(json.loads(data))
+                    existing = await self._pg_get(playbook.playbook_id)
+                    if existing is None:
+                        await self._pg_upsert(playbook)
+                        count += 1
+                        logger.info("playbook_backfilled_to_pg", playbook_id=playbook.playbook_id)
+                except Exception as e:
+                    logger.warning("playbook_backfill_item_failed", key=str(key), error=str(e))
+
+        except Exception as e:
+            logger.warning("playbook_backfill_failed", error=str(e))
+
+        logger.info("playbook_backfill_complete", count=count)
+        return count
+
+    # =========================================================================
+    # Private — PostgreSQL Helpers
+    # =========================================================================
+
+    async def _pg_upsert(self, playbook: Playbook) -> None:
+        """Upsert Playbook 到 PostgreSQL（INSERT ON CONFLICT UPDATE）"""
+        try:
+            from sqlalchemy.dialects.postgresql import insert as pg_insert
+
+            factory = get_session_factory()
+            async with factory() as session:
+                stmt = pg_insert(PlaybookRecord).values(
+                    playbook_id=playbook.playbook_id,
+                    name=playbook.name,
+                    description=playbook.description,
+                    status=playbook.status.value,
+                    source=playbook.source.value,
+                    symptom_pattern=playbook.symptom_pattern.model_dump(),
+                    repair_steps=[s.model_dump() for s in playbook.repair_steps],
+                    estimated_duration_minutes=playbook.estimated_duration_minutes,
+                    source_incident_ids=playbook.source_incident_ids,
+                    ai_confidence=playbook.ai_confidence,
+                    success_count=playbook.success_count,
+                    failure_count=playbook.failure_count,
+                    last_used_at=playbook.last_used_at,
+                    trust_score=playbook.trust_score,
+                    approved_by=playbook.approved_by,
+                    approved_at=playbook.approved_at,
+                    tags=playbook.tags,
+                    notes=playbook.notes,
+                    requires_approval_level=playbook.requires_approval_level,
+                    stateful_targets=playbook.stateful_targets,
+                    requires_pre_backup=playbook.requires_pre_backup,
+                    created_at=playbook.created_at,
+                    updated_at=playbook.updated_at,
+                ).on_conflict_do_update(
+                    index_elements=["playbook_id"],
+                    set_={
+                        "name": playbook.name,
+                        "description": playbook.description,
+                        "status": playbook.status.value,
+                        "source": playbook.source.value,
+                        "symptom_pattern": playbook.symptom_pattern.model_dump(),
+                        "repair_steps": [s.model_dump() for s in playbook.repair_steps],
+                        "estimated_duration_minutes": playbook.estimated_duration_minutes,
+                        "source_incident_ids": playbook.source_incident_ids,
+                        "ai_confidence": playbook.ai_confidence,
+                        "success_count": playbook.success_count,
+                        "failure_count": playbook.failure_count,
+                        "last_used_at": playbook.last_used_at,
+                        "trust_score": playbook.trust_score,
+                        "approved_by": playbook.approved_by,
+                        "approved_at": playbook.approved_at,
+                        "tags": playbook.tags,
+                        "notes": playbook.notes,
+                        "requires_approval_level": playbook.requires_approval_level,
+                        "stateful_targets": playbook.stateful_targets,
+                        "requires_pre_backup": playbook.requires_pre_backup,
+                        "updated_at": playbook.updated_at,
+                    },
+                )
+                await session.execute(stmt)
+                await session.commit()
+        except Exception as e:
+            logger.error("playbook_pg_upsert_failed", playbook_id=playbook.playbook_id, error=str(e))
+            raise
+
+    async def _pg_get(self, playbook_id: str) -> Playbook | None:
+        """從 PostgreSQL 載入 Playbook"""
+        try:
+            factory = get_session_factory()
+            async with factory() as session:
+                result = await session.get(PlaybookRecord, playbook_id)
+                if result is None:
+                    return None
+                return _orm_to_pydantic(result)
+        except Exception as e:
+            logger.warning("playbook_pg_get_failed", playbook_id=playbook_id, error=str(e))
+            return None
+
+    # =========================================================================
+    # Private — Redis Cache Helpers
+    # =========================================================================
+
+    async def _redis_set(self, playbook: Playbook) -> None:
+        """寫入 Redis warm cache（TTL 7天）"""
+        try:
+            redis_client = get_redis()
+            key = f"{PLAYBOOK_KEY_PREFIX}{playbook.playbook_id}"
+            await redis_client.set(
+                key,
+                json.dumps(playbook.to_redis_dict(), ensure_ascii=False),
+                ex=PLAYBOOK_TTL_SECONDS,
+            )
+        except Exception as e:
+            # Redis 寫入失敗不 raise（PG 已成功為主）
+            logger.warning("playbook_redis_set_failed", playbook_id=playbook.playbook_id, error=str(e))
+
+    async def _redis_get(self, playbook_id: str) -> Playbook | None:
+        """從 Redis cache 讀取 Playbook"""
+        try:
+            redis_client = get_redis()
+            key = f"{PLAYBOOK_KEY_PREFIX}{playbook_id}"
+            data = await redis_client.get(key)
+            if data:
+                return Playbook.from_redis_dict(json.loads(data))
+            return None
+        except Exception as e:
+            logger.warning("playbook_redis_get_failed", playbook_id=playbook_id, error=str(e))
+            return None
+
+    async def _redis_find_by_incident(self, incident_id: str) -> list[Playbook]:
+        """Redis fallback: 掃描所有 Playbook 找包含 incident_id 的"""
+        try:
+            redis_client = get_redis()
+            results: list[Playbook] = []
+            async for key in redis_client.scan_iter(match=f"{PLAYBOOK_KEY_PREFIX}PB-*", count=100):
+                data = await redis_client.get(key)
+                if data:
+                    playbook = Playbook.from_redis_dict(json.loads(data))
+                    if incident_id in playbook.source_incident_ids:
+                        results.append(playbook)
+            return results
+        except Exception:
+            return []
+
+    async def _update_indexes(self, playbook: Playbook) -> None:
+        """更新 Redis 索引（供快速 symptom 過濾）"""
+        try:
+            redis_client = get_redis()
+            for alert_name in playbook.symptom_pattern.alert_names:
+                index_key = f"{PLAYBOOK_INDEX_ALERT_PREFIX}{alert_name}"
+                await redis_client.sadd(index_key, playbook.playbook_id)
+                await redis_client.expire(index_key, PLAYBOOK_TTL_SECONDS)
+            for service in playbook.symptom_pattern.affected_services:
+                index_key = f"{PLAYBOOK_INDEX_SERVICE_PREFIX}{service}"
+                await redis_client.sadd(index_key, playbook.playbook_id)
+                await redis_client.expire(index_key, PLAYBOOK_TTL_SECONDS)
+        except Exception as e:
+            logger.warning("playbook_index_update_failed", error=str(e))
+
 
 # =============================================================================
 # Singleton