diff --git a/apps/api/src/models/playbook.py b/apps/api/src/models/playbook.py
index 883f5299..34b02186 100644
--- a/apps/api/src/models/playbook.py
+++ b/apps/api/src/models/playbook.py
@@ -39,6 +39,7 @@ class PlaybookSource(str, Enum):
 
     EXTRACTED = "extracted"  # 從 Incident 自動萃取
     MANUAL = "manual"  # 人工建立
+    YAML_RULE = "yaml_rule"  # 從 alert_rules.yaml 匯入（2026-04-15 ogt）
 
 
 class ActionType(str, Enum):
diff --git a/apps/api/src/repositories/playbook_repository.py b/apps/api/src/repositories/playbook_repository.py
index e0b6d22a..650aa215 100644
--- a/apps/api/src/repositories/playbook_repository.py
+++ b/apps/api/src/repositories/playbook_repository.py
@@ -79,8 +79,24 @@ def _pydantic_to_orm(playbook: Playbook) -> PlaybookRecord:
     )
 
 
+def _normalize_repair_step(step: dict) -> dict:
+    """舊格式 repair_steps 補齊必填欄位（2026-04-15 ogt: 向下相容）"""
+    step = dict(step)
+    if "step_number" not in step:
+        step["step_number"] = 1
+    if "action_type" not in step:
+        cmd = step.get("command", "")
+        step["action_type"] = "kubectl" if cmd and not cmd.startswith("ssh") else "manual"
+    return step
+
+
 def _orm_to_pydantic(record: PlaybookRecord) -> Playbook:
     """PlaybookRecord ORM → Pydantic Playbook"""
+    raw_steps = record.repair_steps or []
+    normalized_steps = [
+        _normalize_repair_step(s) if isinstance(s, dict) else s
+        for s in raw_steps
+    ]
     return Playbook.model_validate({
         "playbook_id": record.playbook_id,
         "name": record.name,
@@ -88,7 +104,7 @@ def _orm_to_pydantic(record: PlaybookRecord) -> Playbook:
         "status": record.status,
         "source": record.source,
         "symptom_pattern": record.symptom_pattern,
-        "repair_steps": record.repair_steps,
+        "repair_steps": normalized_steps,
         "estimated_duration_minutes": record.estimated_duration_minutes,
         "source_incident_ids": record.source_incident_ids,
         "ai_confidence": float(record.ai_confidence),
diff --git a/apps/api/src/services/flywheel_stats_service.py b/apps/api/src/services/flywheel_stats_service.py
index 14353223..7bc5914a 100644
--- a/apps/api/src/services/flywheel_stats_service.py
+++ b/apps/api/src/services/flywheel_stats_service.py
@@ -225,8 +225,10 @@ class FlywheelStatsService:
 
             async with get_db_context() as db:
                 # 未向量化數量 (embedding IS NULL = 未向量化)
+                # 2026-04-15 ogt: KnowledgeEntryRecord ORM 不宣告 embedding 欄位（pgvector），
+                #                  改用 raw SQL 避免 AttributeError
                 unvectorized_q = await db.execute(
-                    select(func.count()).where(KnowledgeEntryRecord.embedding.is_(None))
+                    text("SELECT COUNT(*) FROM knowledge_entries WHERE embedding IS NULL")
                 )
                 unvectorized = unvectorized_q.scalar_one_or_none() or 0
 
diff --git a/apps/api/src/services/playbook_seed_service.py b/apps/api/src/services/playbook_seed_service.py
index 437afdab..da36c1a9 100644
--- a/apps/api/src/services/playbook_seed_service.py
+++ b/apps/api/src/services/playbook_seed_service.py
@@ -42,15 +42,22 @@ async def seed_playbooks_from_rules() -> None:
 
         repo = get_playbook_repository()
 
-        # 取得現有 playbook source_ids，避免重複建立
-        existing, _ = await repo.list_playbooks(status=PlaybookStatus.APPROVED, limit=500)
-        existing_sources = {p.source for p in existing if p.source}
+        # 取得現有 YAML_RULE playbook，依 name 去重避免重複建立
+        # 2026-04-15 ogt: 不再用 list_playbooks（舊格式 repair_steps 會 validation error）
+        #                  改用 raw SQL 只撈 name 欄位，更穩健
+        from src.db.session 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'")
+            )
+            existing_names = {r[0] for r in rows.fetchall()}
 
         seeded = 0
         for rule in rules:
             rule_id = rule.get("id", "")
-            source_key = f"alert_rule:{rule_id}"
-            if source_key in existing_sources:
+            rule_name = rule.get("description", rule_id)
+            if rule_name in existing_names:
                 continue
 
             resp = rule.get("response", {})
@@ -71,7 +78,7 @@ async def seed_playbooks_from_rules() -> None:
                 name=rule.get("description", rule_id),
                 description=resp.get("description", rule.get("description", "")),
                 status=PlaybookStatus.APPROVED,
-                source=source_key,
+                source=PlaybookSource.YAML_RULE,
                 symptom_pattern=SymptomPattern(
                     alert_names=alertnames,
                     affected_services=[],