feat(sprint5.1): Data Safety Guardrails 全鏈路整合 (L1-L5)

Layer 0 - K8s RBAC:
  - k8s/rbac/api-velero-reader.yaml: awoooi-executor SA Velero backup reader

Layer 1 - DB Migration (已在 188 執行):
  - M-002: approval_records 新增 approval_level/votes/required_votes
  - M-003: alert_event_type ENUM 新增 8 個值

Layer 2 - IaC:
  - ops/config/service-registry.yaml: 全服務 Stateful 分級清單 (BLOCK/CRITICAL_HITL/STANDARD_HITL/AUTO)

Layer 3 - Python Services:
  - service_registry.py: 讀取 YAML，提供 is_blocked/requires_multisig/get_required_votes
  - velero_client.py: kubectl 查詢 Velero 備份年齡，失敗 fallback 999h
  - preflight_service.py: Pre-flight 安全檢查 (Q2/Q4 決策)

Layer 1-M001 - Playbook model:
  - playbook.py: 新增 requires_approval_level/stateful_targets/requires_pre_backup

Layer 4 - 業務邏輯:
  - alert_operation_log_repository.py: 新增 8 個 event_type (Guardrail/Pre-flight/MultiSig/備份)
  - auto_repair_service.py: 注入 Service Registry Guardrail 檢查 (BLOCK → 直接拒絕)
  - webhooks.py: ALERT_RECEIVED 溯源記錄 + auto_repair flag Q9 + Langfuse trace_id Q10
  - db/models.py: ApprovalRecord 同步 approval_level/votes/required_votes 欄位
  - docker-health-monitor.sh: 純感知層改造（移除所有 docker restart 邏輯）

Layer 5 - Telegram 通知:
  - telegram_gateway.py: T1-T6 六個新通知方法 (Guardrail/Pre-flight/Backup/MultiSig/ChangeApplied)

參考: ADR-062 Data Safety Guardrails, ADR-063 Service Registry IaC

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

apps/api/src/services/auto_repair_service.py

@@ -192,6 +192,32 @@ class AutoRepairService:
                 blocked_by="GLOBAL_GUARDRAIL",
             )
 
+        # 0.5 Sprint 5.1 Guardrail: Service Registry 服務分級檢查
+        # (2026-04-08 Claude Sonnet 4.6 Asia/Taipei，ADR-062)
+        # 全域熔斷之後、嚴重度之前，BLOCK 等級直接拒絕
+        try:
+            from src.services.service_registry import StatefulLevel, get_service_registry
+            _registry = get_service_registry()
+            _service_name = (incident.target_resource or "") if hasattr(incident, "target_resource") else ""
+            if not _service_name and incident.affected_services:
+                _service_name = incident.affected_services[0]
+            _stateful_level = _registry.get_stateful_level(_service_name)
+            if _stateful_level == StatefulLevel.BLOCK:
+                logger.warning(
+                    "auto_repair_blocked_guardrail",
+                    incident_id=incident.incident_id,
+                    service_name=_service_name,
+                    stateful_level="BLOCK",
+                )
+                return AutoRepairDecision(
+                    can_auto_repair=False,
+                    reason=f"GUARDRAIL_BLOCK: 服務 '{_service_name}' 屬於禁止自動修復清單（資料安全，見 service-registry.yaml）",
+                    blocked_by="SERVICE_REGISTRY_BLOCK",
+                )
+        except Exception as _guardrail_err:
+            logger.error("guardrail_check_failed", error=str(_guardrail_err))
+            # 保守原則：失敗時繼續（不阻擋，但記錄）
+
         # 1. 檢查 Incident 嚴重度
         if incident.severity and incident.severity.value in ["P0", "P1"]:
             logger.info(

apps/api/src/services/preflight_service.py

@@ -0,0 +1,116 @@
+# apps/api/src/services/preflight_service.py
+# Pre-flight 安全檢查服務 (Q2/Q4 決策)
+# 撰寫: Claude Sonnet 4.6 / 2026-04-08 Asia/Taipei
+# 架構: leWOOOgo 積木化，依賴 ServiceRegistryClient + VeleroClient
+# 參考: ADR-062
+
+from __future__ import annotations
+
+import logging
+import time
+from dataclasses import dataclass
+from enum import Enum
+
+from .service_registry import ServiceRegistryClient, get_service_registry
+from .velero_client import VeleroClient, get_velero_client
+
+logger = logging.getLogger(__name__)
+
+
+class PreflightResult(str, Enum):
+    PASS = "PASS"
+    ABORT_BACKUP_EXPIRED = "ABORT_BACKUP_EXPIRED"
+    ABORT_HIGH_IO = "ABORT_HIGH_IO"
+    SKIP = "SKIP"  # 服務不需要 Pre-flight
+
+
+@dataclass
+class PreflightReport:
+    result: PreflightResult
+    backup_age_hours: float | None = None
+    backup_name_triggered: str | None = None
+    reason: str = ""
+
+
+class PreflightService:
+    """
+    Pre-flight 安全檢查
+    - 只有 requires_pre_backup=True 的服務才觸發
+    - 備份過期 → Abort + 觸發緊急備份（非同步）
+    - CPU/IO 高負載告警 → 禁止觸發備份（Q4）
+    """
+
+    def __init__(
+        self,
+        registry: ServiceRegistryClient | None = None,
+        velero: VeleroClient | None = None,
+    ) -> None:
+        self._registry = registry or get_service_registry()
+        self._velero = velero or get_velero_client()
+
+    async def check(
+        self,
+        service_name: str,
+        alert_labels: dict | None = None,
+    ) -> PreflightReport:
+        """
+        執行 Pre-flight 檢查
+        alert_labels: Prometheus 告警標籤，用於判斷 CPU/IO 負載
+        """
+        info = self._registry.get_service(service_name)
+        if info is None or not info.requires_pre_backup:
+            return PreflightReport(result=PreflightResult.SKIP, reason="服務不需要 Pre-flight")
+
+        # Q4: CPU/IO 高負載告警時禁止觸發備份
+        if self._is_high_io_alert(alert_labels):
+            logger.warning(f"Pre-flight: {service_name} 屬於 CPU/IO 高負載告警，跳過備份觸發")
+            return PreflightReport(
+                result=PreflightResult.ABORT_HIGH_IO,
+                reason="告警類型為 CPU/IO 高負載，禁止觸發備份（Q4 決策）",
+            )
+
+        policies = self._registry.get_backup_policies()
+        max_age = policies.get("velero_max_age_hours", 4)
+
+        age = await self._velero.get_latest_backup_age_hours()
+        if age <= max_age:
+            return PreflightReport(
+                result=PreflightResult.PASS,
+                backup_age_hours=age,
+                reason=f"備份時間正常 ({age:.1f}h < {max_age}h)",
+            )
+
+        # 備份過期 → 觸發緊急備份 + Abort
+        backup_name = f"emergency-preflight-{int(time.time())}"
+        triggered = await self._velero.trigger_emergency_backup(backup_name)
+        return PreflightReport(
+            result=PreflightResult.ABORT_BACKUP_EXPIRED,
+            backup_age_hours=age,
+            backup_name_triggered=backup_name if triggered else None,
+            reason=(
+                f"備份過期 ({age:.1f}h > {max_age}h)。"
+                f"{'緊急備份已啟動: ' + backup_name if triggered else '緊急備份啟動失敗，請人工處理'}"
+            ),
+        )
+
+    def _is_high_io_alert(self, labels: dict | None) -> bool:
+        if not labels:
+            return False
+        alert_name = labels.get("alertname", "").lower()
+        return any(kw in alert_name for kw in ["cpu", "io", "disk", "load", "memory"])
+
+
+_preflight_service: PreflightService | None = None
+
+
+def get_preflight_service() -> PreflightService:
+    global _preflight_service
+    if _preflight_service is None:
+        _preflight_service = PreflightService()
+    return _preflight_service
+
+
+def set_preflight_service(service: PreflightService) -> None:
+    """測試注入用 (P4 規範)"""
+    global _preflight_service
+    _preflight_service = service

apps/api/src/services/service_registry.py

@@ -0,0 +1,124 @@
+# apps/api/src/services/service_registry.py
+# Service Registry Client — 讀取 ops/config/service-registry.yaml
+# 撰寫: Claude Sonnet 4.6 / 2026-04-08 Asia/Taipei
+# 架構: leWOOOgo 積木化，純 Service 層，無 Router/DB 依賴
+# 參考: ADR-062, ADR-063
+
+from __future__ import annotations
+
+import logging
+from enum import Enum
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+logger = logging.getLogger(__name__)
+
+# YAML 路徑（相對於 repo root）
+_DEFAULT_REGISTRY_PATH = Path(__file__).parents[5] / "ops" / "config" / "service-registry.yaml"
+
+
+class StatefulLevel(str, Enum):
+    BLOCK = "BLOCK"                 # 禁止，僅告警
+    CRITICAL_HITL = "CRITICAL_HITL"  # 2 票 MultiSig
+    STANDARD_HITL = "STANDARD_HITL"  # 1 票
+    AUTO = "AUTO"                   # 自動執行
+
+
+class ServiceInfo:
+    def __init__(self, data: dict[str, Any]) -> None:
+        self.name: str = data["name"]
+        self.display_name: str = data.get("display_name", self.name)
+        self.host: str = data.get("host", "unknown")
+        self.stateful_level: StatefulLevel = StatefulLevel(data.get("stateful_level", "AUTO"))
+        self.reason: str = data.get("reason", "")
+        self.alert_only: bool = data.get("alert_only", False)
+        self.requires_pre_backup: bool = data.get("requires_pre_backup", False)
+        self.restart_command: str = data.get("restart_command", "docker restart")
+        self.containers: list[str] = data.get("containers", [])
+
+
+class ServiceRegistryClient:
+    """
+    Service Registry 客戶端
+    讀取 ops/config/service-registry.yaml，提供服務 Stateful 分級查詢
+    設計原則: 純讀取，不寫入；失敗時 fallback AUTO（防護不應阻擋告警流程）
+    """
+
+    def __init__(self, registry_path: Path | None = None) -> None:
+        self._path = registry_path or _DEFAULT_REGISTRY_PATH
+        self._services: dict[str, ServiceInfo] = {}
+        self._backup_policies: dict[str, Any] = {}
+        self._multisig_config: dict[str, Any] = {}
+        self._loaded = False
+
+    def _load(self) -> None:
+        if self._loaded:
+            return
+        try:
+            with open(self._path) as f:
+                data = yaml.safe_load(f)
+            for svc in data.get("services", []):
+                info = ServiceInfo(svc)
+                self._services[info.name] = info
+                # 也按 container 名稱建立索引
+                for container in info.containers:
+                    self._services[container] = info
+            self._backup_policies = data.get("backup_policies", {})
+            self._multisig_config = data.get("multisig", {})
+            self._loaded = True
+            logger.info(f"Service Registry 載入完成: {len(self._services)} 個服務")
+        except Exception as e:
+            logger.error(f"Service Registry 載入失敗: {e}，所有服務 fallback AUTO")
+            self._loaded = True  # 防止重複嘗試
+
+    def get_service(self, name: str) -> ServiceInfo | None:
+        self._load()
+        return self._services.get(name)
+
+    def get_stateful_level(self, service_name: str) -> StatefulLevel:
+        """查詢服務分級，未知服務 fallback AUTO"""
+        info = self.get_service(service_name)
+        if info is None:
+            logger.warning(f"未知服務 '{service_name}'，fallback AUTO")
+            return StatefulLevel.AUTO
+        return info.stateful_level
+
+    def is_blocked(self, service_name: str) -> bool:
+        return self.get_stateful_level(service_name) == StatefulLevel.BLOCK
+
+    def requires_multisig(self, service_name: str) -> bool:
+        return self.get_stateful_level(service_name) == StatefulLevel.CRITICAL_HITL
+
+    def get_required_votes(self, service_name: str) -> int:
+        self._load()
+        level = self.get_stateful_level(service_name)
+        if level == StatefulLevel.CRITICAL_HITL:
+            return self._multisig_config.get("critical_required_votes", 2)
+        return self._multisig_config.get("standard_required_votes", 1)
+
+    def get_backup_policies(self) -> dict[str, Any]:
+        self._load()
+        return self._backup_policies
+
+    def get_restart_command(self, service_name: str) -> str:
+        info = self.get_service(service_name)
+        return info.restart_command if info else "docker restart"
+
+
+# Singleton
+_registry_client: ServiceRegistryClient | None = None
+
+
+def get_service_registry() -> ServiceRegistryClient:
+    global _registry_client
+    if _registry_client is None:
+        _registry_client = ServiceRegistryClient()
+    return _registry_client
+
+
+def set_service_registry(client: ServiceRegistryClient) -> None:
+    """測試注入用 (P4 規範)"""
+    global _registry_client
+    _registry_client = client

apps/api/src/services/telegram_gateway.py

@@ -2654,6 +2654,128 @@ class TelegramGateway:
                 f"⚠️ 重診觸發失敗: {html.escape(str(e)[:100])}"
             )
 
+    # =========================================================================
+    # Sprint 5.1 T1-T6: Data Safety Guardrail 通知場景
+    # (2026-04-08 Claude Sonnet 4.6 Asia/Taipei，ADR-062)
+    # =========================================================================
+
+    async def send_guardrail_blocked(
+        self,
+        service_name: str,
+        alertname: str,
+        reason: str,
+    ) -> None:
+        """T1: GUARDRAIL_BLOCKED — 服務屬於 BLOCK 等級，禁止自動修復"""
+        text = (
+            "🚫 <b>[服務保護] 自動修復已阻擋</b>\n"
+            "━━━━━━━━━━━━━━━━━\n"
+            f"服務: <code>{html.escape(service_name)}</code>\n"
+            f"告警: <code>{html.escape(alertname)}</code>\n"
+            f"原因: {html.escape(reason)}\n"
+            "━━━━━━━━━━━━━━━━━\n"
+            "⚠️ 請人工評估並手動處理"
+        )
+        await self.send_notification(text)
+
+    async def send_preflight_failed(
+        self,
+        service_name: str,
+        backup_age_hours: float,
+        max_age_hours: float,
+        backup_name: str | None,
+    ) -> None:
+        """T2: PRE_FLIGHT_FAILED + BACKUP_TRIGGERED — 備份過期，修復暫停"""
+        backup_status = (
+            f"緊急備份: 已啟動 <code>{html.escape(backup_name)}</code>"
+            if backup_name
+            else "緊急備份: <b>啟動失敗</b>，請人工處理"
+        )
+        text = (
+            "⏸ <b>[Pre-flight 阻擋] 備份已過期，修復暫停</b>\n"
+            "━━━━━━━━━━━━━━━━━\n"
+            f"服務: <code>{html.escape(service_name)}</code>\n"
+            f"備份距今: {backup_age_hours:.1f} 小時（上限 {max_age_hours:.0f} 小時）\n"
+            f"{backup_status}\n"
+            "━━━━━━━━━━━━━━━━━\n"
+            "請等待備份完成後，人工重新評估修復方案"
+        )
+        await self.send_notification(text)
+
+    async def send_backup_result(
+        self,
+        backup_name: str,
+        success: bool,
+        error_msg: str | None = None,
+    ) -> None:
+        """T3: BACKUP_COMPLETED / BACKUP_FAILED — 緊急備份結果"""
+        if success:
+            text = (
+                "✅ <b>緊急備份完成</b>\n"
+                f"備份: <code>{html.escape(backup_name)}</code>\n"
+                "可繼續手動執行修復"
+            )
+        else:
+            err = html.escape(error_msg or "未知錯誤")
+            text = (
+                "❌ <b>緊急備份失敗</b>\n"
+                f"備份: <code>{html.escape(backup_name)}</code>\n"
+                f"錯誤: {err}\n"
+                "請人工介入，備份異常"
+            )
+        await self.send_notification(text)
+
+    async def send_multisig_waiting(
+        self,
+        action: str,
+        service_name: str,
+        votes_received: int,
+        votes_required: int,
+        approval_id: str,
+    ) -> None:
+        """T4: APPROVAL_ESCALATED — 第 1 票完成，等待第 2 票"""
+        text = (
+            "🔐 <b>[MultiSig] 等待第 2 票授權</b>\n"
+            "━━━━━━━━━━━━━━━━━\n"
+            f"操作: {html.escape(action)}\n"
+            f"服務: <code>{html.escape(service_name)}</code>\n"
+            f"風險: CRITICAL（HITL 雙簽）\n"
+            f"已獲授權: {votes_received}/{votes_required} 票\n"
+            f"審核 ID: <code>{html.escape(approval_id)}</code>\n"
+            "━━━━━━━━━━━━━━━━━\n"
+            "請第二位審核者登入確認"
+        )
+        await self.send_notification(text)
+
+    async def send_multisig_approved(
+        self,
+        action: str,
+        service_name: str,
+    ) -> None:
+        """T5: MultiSig 完成（2/2）"""
+        text = (
+            "✅ <b>[MultiSig 完成] 雙簽授權通過</b>\n"
+            f"操作: {html.escape(action)}\n"
+            f"服務: <code>{html.escape(service_name)}</code>\n"
+            "授權: 2/2 票  開始執行..."
+        )
+        await self.send_notification(text)
+
+    async def send_change_applied(
+        self,
+        operator: str,
+        action_description: str,
+        timestamp: str,
+    ) -> None:
+        """T6: CHANGE_APPLIED — 手動變更記錄"""
+        text = (
+            "📝 <b>[變更記錄] 手動操作已記錄</b>\n"
+            "━━━━━━━━━━━━━━━━━\n"
+            f"操作者: {html.escape(operator)}\n"
+            f"動作: {html.escape(action_description)}\n"
+            f"時間: {html.escape(timestamp)}"
+        )
+        await self.send_notification(text)
+
     async def send_notification(
         self,
         text: str,

apps/api/src/services/velero_client.py

@@ -0,0 +1,113 @@
+# apps/api/src/services/velero_client.py
+# Velero Backup 查詢客戶端 (kubectl 方式，Q7 決策)
+# 撰寫: Claude Sonnet 4.6 / 2026-04-08 Asia/Taipei
+# 架構: leWOOOgo 積木化，純 Service 層
+# 參考: ADR-062
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import time
+from datetime import UTC, datetime
+
+logger = logging.getLogger(__name__)
+
+_VELERO_NAMESPACE = "velero"
+_KUBECTL_TIMEOUT = 30  # 秒
+
+
+class VeleroClient:
+    """
+    透過 kubectl 查詢 Velero 備份狀態
+    設計原則: 失敗時 fallback「假設備份過期」（保守原則）
+    """
+
+    async def get_latest_backup_age_hours(self) -> float:
+        """
+        查詢最近一次 Completed 備份距今幾小時
+        失敗時返回 999.0（視為嚴重過期，觸發 Abort）
+        """
+        try:
+            result = await asyncio.wait_for(
+                self._run_kubectl(
+                    ["get", "backup", "-n", _VELERO_NAMESPACE,
+                     "-o", "json", "--field-selector", "status.phase=Completed"]
+                ),
+                timeout=_KUBECTL_TIMEOUT,
+            )
+            data = json.loads(result)
+            items = data.get("items", [])
+            if not items:
+                logger.warning("Velero: 找不到任何 Completed 備份")
+                return 999.0
+
+            latest = max(
+                items,
+                key=lambda x: x.get("status", {}).get("completionTimestamp", ""),
+            )
+            completion_ts = latest["status"].get("completionTimestamp", "")
+            if not completion_ts:
+                return 999.0
+
+            completed_at = datetime.fromisoformat(completion_ts.replace("Z", "+00:00"))
+            age = (datetime.now(UTC) - completed_at).total_seconds() / 3600
+            logger.info(f"Velero 最近備份: {completion_ts}，距今 {age:.1f} 小時")
+            return age
+
+        except asyncio.TimeoutError:
+            logger.error("Velero kubectl 查詢超時")
+            return 999.0
+        except Exception as e:
+            logger.error(f"Velero 查詢失敗: {e}")
+            return 999.0
+
+    async def trigger_emergency_backup(self, backup_name: str | None = None) -> bool:
+        """
+        觸發緊急備份（非同步，不等待完成）
+        返回 True 表示指令已成功發送
+        """
+        name = backup_name or f"emergency-{int(time.time())}"
+        try:
+            await asyncio.wait_for(
+                self._run_kubectl([
+                    "create", "backup", name,
+                    "-n", _VELERO_NAMESPACE,
+                    "--include-namespaces", "awoooi-prod",
+                    "--wait=false",
+                ]),
+                timeout=_KUBECTL_TIMEOUT,
+            )
+            logger.info(f"Velero 緊急備份已啟動: {name}")
+            return True
+        except Exception as e:
+            logger.error(f"Velero 緊急備份失敗: {e}")
+            return False
+
+    async def _run_kubectl(self, args: list[str]) -> str:
+        proc = await asyncio.create_subprocess_exec(
+            "kubectl", *args,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        stdout, stderr = await proc.communicate()
+        if proc.returncode != 0:
+            raise RuntimeError(f"kubectl 失敗: {stderr.decode()}")
+        return stdout.decode()
+
+
+_velero_client: VeleroClient | None = None
+
+
+def get_velero_client() -> VeleroClient:
+    global _velero_client
+    if _velero_client is None:
+        _velero_client = VeleroClient()
+    return _velero_client
+
+
+def set_velero_client(client: VeleroClient) -> None:
+    """測試注入用 (P4 規範)"""
+    global _velero_client
+    _velero_client = client