fix(api): 修復全部 lint 錯誤 (ruff --fix)

- Import sorting (I001)
- Unused imports (F401)
- f-string without placeholders (F541)
- Loop variable unused (B007)
- zip() strict parameter (B905)
- Exception chaining (B904)
- collections.abc imports (UP035)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

apps/api/src/services/__init__.py

@@ -44,6 +44,14 @@ from .graph_rag import (
     create_mock_topology,
     topology_graph,
 )
+from .model_registry import (
+    IModelRegistry,
+    ModelRegistry,
+    get_model,
+    get_model_by_complexity,
+    get_model_registry,
+    reset_model_registry,
+)
 from .trust_engine import (
     RiskAdjustment,
     RiskLevel,
@@ -99,4 +107,11 @@ __all__ = [
     "ConsensusResult",
     "AgentOpinion",
     "AgentType",
+    # Model Registry (Phase 12 P1)
+    "ModelRegistry",
+    "IModelRegistry",
+    "get_model_registry",
+    "get_model",
+    "get_model_by_complexity",
+    "reset_model_registry",
 ]

apps/api/src/services/agent_service.py

@@ -14,7 +14,6 @@ Phase 17 R4: 從 agents.py Router 抽離 Redis 操作
 """
 
 import json
-from datetime import UTC, datetime
 from enum import Enum
 from typing import Any, Protocol, runtime_checkable
 from uuid import uuid4
@@ -24,6 +23,7 @@ from src.core.redis_client import get_redis
 from src.core.sse import EventType, SSEEvent, get_publisher
 from src.models.incident import Incident, IncidentStatus, Severity, Signal
 from src.services.consensus_engine import get_consensus_engine
+from src.utils.timezone import now_taipei, now_taipei_iso
 
 logger = get_logger("awoooi.agent_service")
 
@@ -140,7 +140,7 @@ class AgentTaskRedisRepository:
             "agents_completed": 0,
             "total_agents": 4,
             "incident_id": incident_id,
-            "started_at": datetime.now(UTC).isoformat(),
+            "started_at": now_taipei_iso(),
             "trigger": trigger,
         }
 
@@ -333,7 +333,7 @@ class AgentService:
 
     def generate_task_id(self) -> str:
         """產生新的 Task ID"""
-        return f"TASK-{datetime.now(UTC).strftime('%Y%m%d')}-{uuid4().hex[:8].upper()}"
+        return f"TASK-{now_taipei().strftime('%Y%m%d')}-{uuid4().hex[:8].upper()}"
 
     async def create_analysis_task(
         self,
@@ -488,7 +488,7 @@ class AgentService:
                 "final_reasoning": result.final_reasoning,
                 "opinions": [op.to_dict() for op in result.opinions],
                 "dissenting_opinions": result.dissenting_opinions,
-                "completed_at": datetime.now(UTC).isoformat(),
+                "completed_at": now_taipei_iso(),
             }
 
             await self._repository.save_task_result(task_id, task_data)
@@ -526,7 +526,7 @@ class AgentService:
                 "state": TaskState.FAILED.value,
                 "progress": 0,
                 "error": str(e),
-                "completed_at": datetime.now(UTC).isoformat(),
+                "completed_at": now_taipei_iso(),
             }
 
             await self._repository.save_task_result(task_id, task_data)
@@ -638,7 +638,7 @@ class AgentService:
                     alert_name=alert_name,
                     severity=Severity(severity),
                     source="manual",
-                    fired_at=datetime.now(UTC),
+                    fired_at=now_taipei(),
                 ))
 
         return Incident(

apps/api/src/services/auto_repair_service.py

@@ -0,0 +1,425 @@
+"""
+Auto Repair Service - #8 自動升級決策
+=====================================
+高品質 Playbook 自動修復執行
+
+Phase 8: 自動化層實作
+建立時間: 2026-03-26 17:30 (台北時區)
+建立者: Claude Code (#8 自動升級決策)
+
+遵循 leWOOOgo 積木化原則:
+- Service 層只依賴 Repository/Service Interface
+- 不直接存取 Redis/DB
+- 封裝所有自動修復邏輯
+
+觸發條件 (AND):
+1. 有匹配的高品質 Playbook (is_high_quality = True)
+2. Playbook 中的動作風險等級 <= MEDIUM
+3. Incident 嚴重度 <= P2
+
+安全邊界:
+- HIGH/CRITICAL 風險動作永遠需要人工審核
+- P0/P1 嚴重度 Incident 需要人工確認
+"""
+
+from dataclasses import dataclass
+from typing import Protocol
+
+import structlog
+
+from src.models.incident import Incident, Severity
+from src.models.playbook import (
+    ActionType,
+    Playbook,
+    RiskLevel,
+    SymptomPattern,
+)
+from src.services.executor import get_executor
+from src.services.playbook_service import IPlaybookService, get_playbook_service
+
+logger = structlog.get_logger(__name__)
+
+
+# =============================================================================
+# Types
+# =============================================================================
+
+
+@dataclass
+class AutoRepairDecision:
+    """自動修復決策結果"""
+
+    can_auto_repair: bool
+    playbook: Playbook | None = None
+    reason: str = ""
+    risk_level: RiskLevel = RiskLevel.MEDIUM
+    blocked_by: str | None = None  # 阻擋原因 (如 HIGH_RISK, P1_SEVERITY)
+
+
+@dataclass
+class AutoRepairResult:
+    """自動修復執行結果"""
+
+    success: bool
+    playbook_id: str
+    incident_id: str
+    executed_steps: list[str]
+    error: str | None = None
+    execution_time_ms: int = 0
+
+
+# =============================================================================
+# Auto Repair Service Interface
+# =============================================================================
+
+
+class IAutoRepairService(Protocol):
+    """自動修復服務介面"""
+
+    async def evaluate_auto_repair(
+        self,
+        incident: Incident,
+    ) -> AutoRepairDecision:
+        """
+        評估是否可自動修復
+
+        Args:
+            incident: 待處理的 Incident
+
+        Returns:
+            AutoRepairDecision: 決策結果
+        """
+        ...
+
+    async def execute_auto_repair(
+        self,
+        incident: Incident,
+        playbook: Playbook,
+    ) -> AutoRepairResult:
+        """
+        執行自動修復
+
+        Args:
+            incident: 待處理的 Incident
+            playbook: 要執行的 Playbook
+
+        Returns:
+            AutoRepairResult: 執行結果
+        """
+        ...
+
+
+# =============================================================================
+# Auto Repair Service Implementation
+# =============================================================================
+
+
+class AutoRepairService:
+    """
+    自動修復服務實作
+
+    職責:
+    - 評估 Incident 是否可自動修復
+    - 執行高品質 Playbook
+    - 更新執行統計
+    """
+
+    # === 安全邊界常數 ===
+    MAX_AUTO_REPAIR_RISK = RiskLevel.MEDIUM  # 最高允許自動修復的風險等級
+    MAX_AUTO_REPAIR_SEVERITY = Severity.P2  # 最高允許自動修復的嚴重度
+    MIN_SIMILARITY_SCORE = 0.7  # 最低相似度門檻
+
+    def __init__(
+        self,
+        playbook_service: IPlaybookService | None = None,
+    ):
+        self._playbook_service = playbook_service or get_playbook_service()
+
+    async def evaluate_auto_repair(
+        self,
+        incident: Incident,
+    ) -> AutoRepairDecision:
+        """
+        評估是否可自動修復
+
+        決策流程:
+        1. 檢查 Incident 嚴重度 (P0/P1 需人工)
+        2. 從 Playbook 找匹配項
+        3. 檢查 Playbook 是否為高品質
+        4. 檢查動作風險等級
+        """
+        logger.info(
+            "auto_repair_evaluate_start",
+            incident_id=incident.incident_id,
+            severity=incident.severity.value if incident.severity else None,
+        )
+
+        # 1. 檢查 Incident 嚴重度
+        if incident.severity and incident.severity.value in ["P0", "P1"]:
+            logger.info(
+                "auto_repair_blocked_severity",
+                incident_id=incident.incident_id,
+                severity=incident.severity.value,
+            )
+            return AutoRepairDecision(
+                can_auto_repair=False,
+                reason=f"Incident 嚴重度 {incident.severity.value} 需要人工審核",
+                blocked_by="HIGH_SEVERITY",
+            )
+
+        # 2. 提取症狀模式
+        symptoms = self._extract_symptoms(incident)
+
+        # 3. 找匹配的 Playbook
+        recommendations = await self._playbook_service.get_recommendations(
+            symptoms=symptoms,
+            top_k=3,
+        )
+
+        if not recommendations:
+            logger.info(
+                "auto_repair_no_playbook_match",
+                incident_id=incident.incident_id,
+            )
+            return AutoRepairDecision(
+                can_auto_repair=False,
+                reason="未找到匹配的 Playbook",
+                blocked_by="NO_MATCH",
+            )
+
+        # 4. 檢查最佳匹配
+        best_match = recommendations[0]
+
+        # 相似度檢查
+        if best_match.similarity_score < self.MIN_SIMILARITY_SCORE:
+            return AutoRepairDecision(
+                can_auto_repair=False,
+                playbook=best_match.playbook,
+                reason=f"相似度 {best_match.similarity_score:.0%} 低於門檻 {self.MIN_SIMILARITY_SCORE:.0%}",
+                blocked_by="LOW_SIMILARITY",
+            )
+
+        # 高品質檢查
+        if not best_match.playbook.is_high_quality:
+            return AutoRepairDecision(
+                can_auto_repair=False,
+                playbook=best_match.playbook,
+                reason=f"Playbook 尚未達到高品質標準 (成功率: {best_match.playbook.success_rate:.0%}, 執行次數: {best_match.playbook.total_executions})",
+                blocked_by="NOT_HIGH_QUALITY",
+            )
+
+        # 5. 檢查動作風險等級
+        max_risk = self._get_max_risk_level(best_match.playbook)
+
+        if self._risk_exceeds_threshold(max_risk):
+            return AutoRepairDecision(
+                can_auto_repair=False,
+                playbook=best_match.playbook,
+                reason=f"Playbook 包含 {max_risk.value} 風險動作，需要人工審核",
+                risk_level=max_risk,
+                blocked_by="HIGH_RISK",
+            )
+
+        # 6. 可以自動修復
+        logger.info(
+            "auto_repair_approved",
+            incident_id=incident.incident_id,
+            playbook_id=best_match.playbook.playbook_id,
+            similarity=best_match.similarity_score,
+            success_rate=best_match.playbook.success_rate,
+        )
+
+        return AutoRepairDecision(
+            can_auto_repair=True,
+            playbook=best_match.playbook,
+            reason=f"匹配高品質 Playbook: {best_match.playbook.name} (成功率 {best_match.playbook.success_rate:.0%})",
+            risk_level=max_risk,
+        )
+
+    async def execute_auto_repair(
+        self,
+        incident: Incident,
+        playbook: Playbook,
+    ) -> AutoRepairResult:
+        """
+        執行自動修復
+
+        流程:
+        1. 依序執行 Playbook 中的 repair_steps
+        2. 記錄執行結果
+        3. 更新 Playbook 統計
+        """
+        import time
+
+        start_time = time.perf_counter()
+        executed_steps: list[str] = []
+
+        logger.info(
+            "auto_repair_execute_start",
+            incident_id=incident.incident_id,
+            playbook_id=playbook.playbook_id,
+            steps_count=len(playbook.repair_steps),
+        )
+
+        try:
+            # 執行每個步驟
+            for step in playbook.repair_steps:
+                # 安全檢查: 跳過高風險步驟
+                if self._risk_exceeds_threshold(step.risk_level):
+                    logger.warning(
+                        "auto_repair_skip_high_risk_step",
+                        step_number=step.step_number,
+                        risk_level=step.risk_level.value,
+                    )
+                    continue
+
+                # 執行步驟
+                step_result = await self._execute_step(incident, step)
+                executed_steps.append(
+                    f"Step {step.step_number}: {step.command[:50]}... -> {step_result}"
+                )
+
+            # 更新 Playbook 統計
+            await self._playbook_service.record_execution(
+                playbook_id=playbook.playbook_id,
+                success=True,
+            )
+
+            execution_time = int((time.perf_counter() - start_time) * 1000)
+
+            logger.info(
+                "auto_repair_execute_success",
+                incident_id=incident.incident_id,
+                playbook_id=playbook.playbook_id,
+                executed_steps=len(executed_steps),
+                execution_time_ms=execution_time,
+            )
+
+            return AutoRepairResult(
+                success=True,
+                playbook_id=playbook.playbook_id,
+                incident_id=incident.incident_id,
+                executed_steps=executed_steps,
+                execution_time_ms=execution_time,
+            )
+
+        except Exception as e:
+            # 更新失敗統計
+            await self._playbook_service.record_execution(
+                playbook_id=playbook.playbook_id,
+                success=False,
+            )
+
+            execution_time = int((time.perf_counter() - start_time) * 1000)
+
+            logger.error(
+                "auto_repair_execute_failed",
+                incident_id=incident.incident_id,
+                playbook_id=playbook.playbook_id,
+                error=str(e),
+            )
+
+            return AutoRepairResult(
+                success=False,
+                playbook_id=playbook.playbook_id,
+                incident_id=incident.incident_id,
+                executed_steps=executed_steps,
+                error=str(e),
+                execution_time_ms=execution_time,
+            )
+
+    # === Private Helpers ===
+
+    def _extract_symptoms(self, incident: Incident) -> SymptomPattern:
+        """從 Incident 提取症狀模式"""
+        alert_names = []
+        keywords = []
+
+        if incident.signals:
+            for signal in incident.signals:
+                alert_names.append(signal.alert_name)
+                # 從 annotations 提取關鍵字
+                if signal.annotations:
+                    for value in signal.annotations.values():
+                        if isinstance(value, str) and len(value) < 50:
+                            keywords.append(value)
+
+        return SymptomPattern(
+            alert_names=alert_names,
+            affected_services=incident.affected_services or [],
+            severity_range=[incident.severity.value] if incident.severity else ["P2"],
+            keywords=keywords[:10],
+        )
+
+    def _get_max_risk_level(self, playbook: Playbook) -> RiskLevel:
+        """取得 Playbook 中最高的風險等級"""
+        risk_order = {
+            RiskLevel.LOW: 0,
+            RiskLevel.MEDIUM: 1,
+            RiskLevel.HIGH: 2,
+            RiskLevel.CRITICAL: 3,
+        }
+
+        max_risk = RiskLevel.LOW
+        for step in playbook.repair_steps:
+            if risk_order.get(step.risk_level, 0) > risk_order.get(max_risk, 0):
+                max_risk = step.risk_level
+
+        return max_risk
+
+    def _risk_exceeds_threshold(self, risk: RiskLevel) -> bool:
+        """檢查風險是否超過自動修復門檻"""
+        high_risks = {RiskLevel.HIGH, RiskLevel.CRITICAL}
+        return risk in high_risks
+
+    async def _execute_step(self, incident: Incident, step) -> str:
+        """
+        執行單一修復步驟
+
+        目前整合:
+        - kubectl 命令: 透過 ActionExecutor
+        - script: 透過 subprocess
+        - manual: 跳過 (需人工)
+        """
+        if step.action_type == ActionType.MANUAL:
+            return "SKIPPED (manual step)"
+
+        if step.action_type == ActionType.KUBECTL:
+            # 整合 ActionExecutor
+            try:
+                executor = get_executor()
+
+                # 替換 {target} 為實際目標
+                command = step.command
+                if incident.affected_services:
+                    command = command.replace("{target}", incident.affected_services[0])
+
+                result = await executor.execute_kubectl_command(command)
+                return "SUCCESS" if result.success else f"FAILED: {result.error}"
+
+            except ImportError:
+                logger.warning("action_executor_not_available")
+                return "SKIPPED (executor not available)"
+
+        return "UNKNOWN_ACTION_TYPE"
+
+
+# =============================================================================
+# Singleton
+# =============================================================================
+
+_service: AutoRepairService | None = None
+
+
+def get_auto_repair_service() -> IAutoRepairService:
+    """取得 AutoRepairService 單例"""
+    global _service
+    if _service is None:
+        _service = AutoRepairService()
+    return _service
+
+
+def set_auto_repair_service(service: AutoRepairService | None) -> None:
+    """注入 AutoRepairService 實例 (用於 DI 或測試)"""
+    global _service
+    _service = service

apps/api/src/services/ci_auto_repair.py

@@ -21,14 +21,13 @@ CI 失敗自動修復服務，根據風險分級決定執行策略
 
 from __future__ import annotations
 
-import asyncio
 from dataclasses import dataclass
 from enum import Enum
 
 import structlog
 
-from src.services.intent_classifier import IntentType, RiskLevel, get_intent_classifier
 from src.services.complexity_scorer import get_complexity_scorer
+from src.services.intent_classifier import IntentType, RiskLevel, get_intent_classifier
 
 logger = structlog.get_logger(__name__)
 

apps/api/src/services/error_analyzer_service.py

@@ -0,0 +1,369 @@
+"""
+Error Analyzer Service - #39 Sentry 錯誤 AI 分析
+=================================================
+Phase 10: Sentry + OpenClaw + UI 整合
+
+功能:
+1. 接收 Sentry Issue + Stacktrace 數據
+2. 使用 OpenClaw LLM 進行根因分析
+3. 生成修復建議與預防措施
+
+遵循 leWOOOgo 積木化原則:
+- Service 層負責業務邏輯
+- 不直接存取 Redis/DB
+- 使用 DI 支援測試
+
+版本: v1.0
+建立: 2026-03-26 18:45 (台北時區)
+建立者: Claude Code (#39 Error Analyzer Agent)
+"""
+
+import json
+from typing import Protocol, runtime_checkable
+
+from pydantic import BaseModel, Field
+
+from src.core.logging import get_logger
+from src.utils.timezone import now_taipei_iso
+
+logger = get_logger("awoooi.error_analyzer")
+
+
+# =============================================================================
+# Error Analysis Prompt
+# =============================================================================
+
+ERROR_ANALYZER_SYSTEM_PROMPT = """# OpenClaw Error Analyzer - AWOOOI 錯誤分析專家
+
+You are a senior Software Engineer specialized in debugging and error analysis.
+
+## 🌐 Language Requirement (CRITICAL)
+- You MUST respond in **Traditional Chinese (繁體中文/正體中文)** for all text fields
+- FORBIDDEN: Simplified Chinese characters (简体字)
+- Use Taiwan locale conventions (台灣用語)
+
+## 🎯 Your Mission
+Analyze the given error from Sentry and provide:
+1. **Root Cause Analysis** - Why did this error occur?
+2. **Impact Assessment** - How serious is this error?
+3. **Fix Recommendations** - How to fix this error?
+4. **Prevention Suggestions** - How to prevent recurrence?
+
+## 📊 Analysis Categories
+- **CODE_BUG**: Logic error, null pointer, type error
+- **DEPENDENCY**: Third-party library issue, version conflict
+- **CONFIGURATION**: Missing env var, wrong config
+- **INFRASTRUCTURE**: Network, timeout, resource exhaustion
+- **DATA_INTEGRITY**: Corrupt data, schema mismatch
+- **EXTERNAL_SERVICE**: API failure, rate limit
+- **UNKNOWN**: Cannot determine from available information
+
+## ⚠️ Output Rules
+- Respond with ONLY valid JSON
+- confidence MUST be between 0.0 and 1.0
+- severity MUST be one of: LOW, MEDIUM, HIGH, CRITICAL
+- All text fields in Traditional Chinese
+
+## 📋 JSON Schema (REQUIRED)
+```json
+{
+  "root_cause": "string - 根因分析 (繁體中文)",
+  "category": "CODE_BUG|DEPENDENCY|CONFIGURATION|INFRASTRUCTURE|DATA_INTEGRITY|EXTERNAL_SERVICE|UNKNOWN",
+  "severity": "LOW|MEDIUM|HIGH|CRITICAL",
+  "impact_assessment": "string - 影響評估 (繁體中文)",
+  "fix_recommendation": {
+    "summary": "string - 修復摘要",
+    "steps": ["array - 修復步驟"],
+    "code_suggestion": "string | null - 建議的代碼修改"
+  },
+  "prevention": [
+    {
+      "type": "CODE_REVIEW|UNIT_TEST|MONITORING|VALIDATION|ERROR_HANDLING",
+      "description": "string - 預防措施描述"
+    }
+  ],
+  "related_files": ["array - 可能相關的檔案路徑"],
+  "confidence": "number - 0.0 to 1.0",
+  "reasoning": "string - 分析推理過程 (繁體中文)"
+}
+```
+
+Now analyze the following error:
+"""
+
+
+# =============================================================================
+# Response Models
+# =============================================================================
+
+
+class FixRecommendation(BaseModel):
+    """修復建議"""
+
+    summary: str = Field(description="修復摘要")
+    steps: list[str] = Field(default_factory=list, description="修復步驟")
+    code_suggestion: str | None = Field(None, description="建議的代碼修改")
+
+
+class PreventionMeasure(BaseModel):
+    """預防措施"""
+
+    type: str = Field(description="類型 (CODE_REVIEW, UNIT_TEST, etc.)")
+    description: str = Field(description="描述")
+
+
+class ErrorAnalysisResult(BaseModel):
+    """錯誤分析結果"""
+
+    root_cause: str = Field(description="根因分析")
+    category: str = Field(description="分類")
+    severity: str = Field(description="嚴重度")
+    impact_assessment: str = Field(description="影響評估")
+    fix_recommendation: FixRecommendation = Field(description="修復建議")
+    prevention: list[PreventionMeasure] = Field(
+        default_factory=list, description="預防措施"
+    )
+    related_files: list[str] = Field(default_factory=list, description="相關檔案")
+    confidence: float = Field(description="信心度")
+    reasoning: str = Field(description="分析推理過程")
+
+
+# =============================================================================
+# Protocol Interface
+# =============================================================================
+
+
+@runtime_checkable
+class ILLMProvider(Protocol):
+    """LLM Provider Protocol"""
+
+    async def call(self, prompt: str) -> tuple[str, str, bool]:
+        """
+        呼叫 LLM
+
+        Returns:
+            (response, provider_name, success)
+        """
+        ...
+
+
+# =============================================================================
+# Error Analyzer Service
+# =============================================================================
+
+
+class ErrorAnalyzerService:
+    """
+    Error Analyzer Service - Sentry 錯誤 AI 分析
+
+    職責:
+    1. 組裝分析 Prompt
+    2. 呼叫 OpenClaw LLM
+    3. 解析並驗證分析結果
+    """
+
+    def __init__(self, llm_provider: ILLMProvider | None = None) -> None:
+        """
+        初始化 Error Analyzer Service
+
+        Args:
+            llm_provider: LLM 提供者 (預設使用 OpenClaw)
+        """
+        self._llm_provider = llm_provider
+
+    async def _get_llm_provider(self) -> ILLMProvider:
+        """取得 LLM Provider (lazy init)"""
+        if self._llm_provider is None:
+            from src.services.openclaw import get_openclaw
+
+            self._llm_provider = get_openclaw()
+        return self._llm_provider
+
+    async def analyze_error(
+        self,
+        issue_id: str,
+        title: str,
+        level: str,
+        culprit: str | None,
+        count: int,
+        stacktrace: str,
+        context: dict | None = None,
+    ) -> tuple[ErrorAnalysisResult | None, str, bool]:
+        """
+        分析 Sentry 錯誤
+
+        Args:
+            issue_id: Sentry Issue ID
+            title: 錯誤標題
+            level: 嚴重度 (error, warning, etc.)
+            culprit: 錯誤來源 (函數/檔案)
+            count: 發生次數
+            stacktrace: 堆疊追蹤
+            context: 額外上下文 (browser, os, tags, etc.)
+
+        Returns:
+            (analysis_result, provider, success)
+        """
+        # 組裝 Prompt
+        error_context = {
+            "issue_id": issue_id,
+            "title": title,
+            "level": level,
+            "culprit": culprit,
+            "occurrence_count": count,
+            "stacktrace": stacktrace,
+            "context": context or {},
+            "analyzed_at": now_taipei_iso(),
+        }
+
+        prompt = ERROR_ANALYZER_SYSTEM_PROMPT + "\n```json\n"
+        prompt += json.dumps(error_context, ensure_ascii=False, indent=2)
+        prompt += "\n```"
+
+        logger.info(
+            "error_analysis_start",
+            issue_id=issue_id,
+            title=title,
+            level=level,
+        )
+
+        # 呼叫 LLM
+        try:
+            llm = await self._get_llm_provider()
+            response, provider, success = await llm.call(prompt)
+
+            if not success:
+                logger.error(
+                    "error_analysis_llm_failed",
+                    issue_id=issue_id,
+                    provider=provider,
+                )
+                return None, provider, False
+
+            logger.info(
+                "error_analysis_llm_response",
+                issue_id=issue_id,
+                provider=provider,
+                response_length=len(response),
+            )
+
+            # 解析結果
+            result = self._parse_analysis_result(response)
+
+            if result:
+                logger.info(
+                    "error_analysis_complete",
+                    issue_id=issue_id,
+                    category=result.category,
+                    severity=result.severity,
+                    confidence=result.confidence,
+                )
+            else:
+                logger.warning(
+                    "error_analysis_parse_failed",
+                    issue_id=issue_id,
+                    raw_response=response[:300],
+                )
+
+            return result, provider, True
+
+        except Exception as e:
+            logger.exception(
+                "error_analysis_failed",
+                issue_id=issue_id,
+                error=str(e),
+            )
+            return None, "error", False
+
+    def _parse_analysis_result(self, raw_response: str) -> ErrorAnalysisResult | None:
+        """
+        解析 LLM 回應為結構化結果
+
+        Args:
+            raw_response: LLM 原始回應
+
+        Returns:
+            解析後的 ErrorAnalysisResult，解析失敗返回 None
+        """
+        try:
+            # 嘗試找到 JSON 區塊
+            json_str = raw_response
+
+            # 處理可能的 markdown 包裝
+            if "```json" in raw_response:
+                start = raw_response.find("```json") + 7
+                end = raw_response.find("```", start)
+                if end > start:
+                    json_str = raw_response[start:end]
+            elif "```" in raw_response:
+                start = raw_response.find("```") + 3
+                end = raw_response.find("```", start)
+                if end > start:
+                    json_str = raw_response[start:end]
+
+            # 解析 JSON
+            data = json.loads(json_str.strip())
+
+            # 建立 FixRecommendation
+            fix_data = data.get("fix_recommendation", {})
+            fix_recommendation = FixRecommendation(
+                summary=fix_data.get("summary", "無建議"),
+                steps=fix_data.get("steps", []),
+                code_suggestion=fix_data.get("code_suggestion"),
+            )
+
+            # 建立 PreventionMeasure 列表
+            prevention = []
+            for p in data.get("prevention", []):
+                prevention.append(PreventionMeasure(
+                    type=p.get("type", "UNKNOWN"),
+                    description=p.get("description", ""),
+                ))
+
+            # 建立最終結果
+            return ErrorAnalysisResult(
+                root_cause=data.get("root_cause", "無法判斷根因"),
+                category=data.get("category", "UNKNOWN"),
+                severity=data.get("severity", "MEDIUM"),
+                impact_assessment=data.get("impact_assessment", "影響評估中"),
+                fix_recommendation=fix_recommendation,
+                prevention=prevention,
+                related_files=data.get("related_files", []),
+                confidence=float(data.get("confidence", 0.5)),
+                reasoning=data.get("reasoning", ""),
+            )
+
+        except json.JSONDecodeError as e:
+            logger.warning(
+                "error_analysis_json_decode_failed",
+                error=str(e),
+                raw_response=raw_response[:200],
+            )
+            return None
+        except Exception as e:
+            logger.warning(
+                "error_analysis_parse_error",
+                error=str(e),
+            )
+            return None
+
+
+# =============================================================================
+# Singleton
+# =============================================================================
+
+_error_analyzer_service: ErrorAnalyzerService | None = None
+
+
+def get_error_analyzer_service() -> ErrorAnalyzerService:
+    """取得 Error Analyzer Service 實例 (Singleton)"""
+    global _error_analyzer_service
+    if _error_analyzer_service is None:
+        _error_analyzer_service = ErrorAnalyzerService()
+    return _error_analyzer_service
+
+
+def set_error_analyzer_service(service: ErrorAnalyzerService) -> None:
+    """設定 Error Analyzer Service 實例 (for testing)"""
+    global _error_analyzer_service
+    _error_analyzer_service = service

apps/api/src/services/playbook_service.py

@@ -13,7 +13,6 @@ Phase 7.3: Service 實作
 - 封裝所有業務邏輯
 """
 
-from datetime import UTC, datetime
 from typing import Protocol
 
 import structlog
@@ -31,6 +30,7 @@ from src.models.playbook import (
 )
 from src.repositories.interfaces import IPlaybookRepository
 from src.repositories.playbook_repository import get_playbook_repository
+from src.utils.timezone import now_taipei
 
 logger = structlog.get_logger(__name__)
 
@@ -246,7 +246,7 @@ class PlaybookService:
 
         playbook.status = PlaybookStatus.APPROVED
         playbook.approved_by = approved_by
-        playbook.approved_at = datetime.now(UTC)
+        playbook.approved_at = now_taipei()
         if notes:
             playbook.notes = notes
 
@@ -294,6 +294,83 @@ class PlaybookService:
         """更新 Playbook"""
         return await self._repository.update(playbook)
 
+    async def update_with_validation(
+        self,
+        playbook_id: str,
+        update_data: dict,
+    ) -> Playbook | None:
+        """
+        更新 Playbook (含驗證)
+
+        Phase 8 P1 修復: 從 Router 層移至 Service 層進行驗證
+
+        驗證規則:
+        - 禁止直接修改 playbook_id
+        - 禁止反向狀態轉換 (APPROVED → DRAFT)
+        - 統計欄位 (success_count, failure_count) 只能透過 record_execution 更新
+
+        Args:
+            playbook_id: Playbook ID
+            update_data: 要更新的欄位 (dict)
+
+        Returns:
+            更新後的 Playbook 或 None
+        """
+        playbook = await self._repository.get_by_id(playbook_id)
+        if not playbook:
+            return None
+
+        # 禁止修改的欄位
+        forbidden_fields = {
+            "playbook_id",
+            "created_at",
+            "success_count",
+            "failure_count",
+            "last_used_at",
+        }
+
+        for field in forbidden_fields:
+            if field in update_data:
+                logger.warning(
+                    "playbook_update_forbidden_field",
+                    playbook_id=playbook_id,
+                    field=field,
+                )
+                del update_data[field]
+
+        # 狀態轉換驗證
+        if "status" in update_data:
+            new_status = update_data["status"]
+            current_status = playbook.status
+
+            # 允許的轉換: DRAFT → APPROVED, APPROVED → DEPRECATED
+            # 禁止: APPROVED → DRAFT, DEPRECATED → 任何
+            if current_status == PlaybookStatus.DEPRECATED:
+                logger.warning(
+                    "playbook_update_deprecated_status",
+                    playbook_id=playbook_id,
+                )
+                return None
+
+            if (
+                current_status == PlaybookStatus.APPROVED
+                and new_status == PlaybookStatus.DRAFT
+            ):
+                logger.warning(
+                    "playbook_update_invalid_status_transition",
+                    playbook_id=playbook_id,
+                    from_status=current_status.value,
+                    to_status=new_status,
+                )
+                return None
+
+        # 應用更新
+        for field, value in update_data.items():
+            if value is not None and hasattr(playbook, field):
+                setattr(playbook, field, value)
+
+        return await self._repository.update(playbook)
+
     async def delete(self, playbook_id: str) -> bool:
         """刪除 Playbook (軟刪除)"""
         return await self._repository.delete(playbook_id)

apps/api/src/services/resource_resolver.py

@@ -26,7 +26,6 @@ from typing import Protocol, runtime_checkable
 import structlog
 
 from src.utils.k8s_naming import (
-    NormalizeResult,
     ResourceType,
     extract_resource_hints,
     normalize_resource_name,

apps/api/src/services/sentry_service.py

@@ -0,0 +1,218 @@
+"""
+Sentry Service - Sentry API 封裝
+================================
+Phase 10: Sentry + OpenClaw + UI 整合
+
+遵循 leWOOOgo 積木化原則:
+- Service 層負責外部 API 呼叫
+- Router 層只做 HTTP 轉發
+- 單一職責: 只處理 Sentry API 互動
+
+版本: v1.0
+建立: 2026-03-26 21:15 (台北時區)
+建立者: Claude Code (P2 架構改善)
+"""
+
+from typing import Any
+
+import httpx
+
+from src.core.config import settings
+from src.core.logging import get_logger
+
+logger = get_logger("awoooi.sentry")
+
+
+class SentryService:
+    """
+    Sentry API Service
+
+    職責:
+    1. 封裝 Sentry API 呼叫
+    2. 處理認證與錯誤
+    3. 提供類型化的資料存取
+    """
+
+    def __init__(
+        self,
+        base_url: str | None = None,
+        org: str | None = None,
+        project: str | None = None,
+        auth_token: str | None = None,
+        timeout: float = 10.0,
+    ) -> None:
+        """
+        初始化 Sentry Service
+
+        Args:
+            base_url: Sentry API URL (預設從 settings)
+            org: Sentry organization slug
+            project: Sentry project slug
+            auth_token: API auth token
+            timeout: 請求超時秒數
+        """
+        self.base_url = base_url or settings.SENTRY_SELF_HOSTED_URL
+        self.org = org or settings.SENTRY_ORG
+        self.project = project or settings.SENTRY_PROJECT
+        self.auth_token = auth_token or settings.SENTRY_AUTH_TOKEN
+        self.timeout = timeout
+
+    async def _request(
+        self,
+        endpoint: str,
+        params: dict[str, Any] | None = None,
+    ) -> dict | list | None:
+        """
+        發送 Sentry API 請求
+
+        Args:
+            endpoint: API 端點 (不含 /api/0/ 前綴)
+            params: 查詢參數
+
+        Returns:
+            JSON 回應，失敗返回 None
+        """
+        headers = {}
+        if self.auth_token:
+            headers["Authorization"] = f"Bearer {self.auth_token}"
+
+        url = f"{self.base_url}/api/0/{endpoint}"
+
+        try:
+            async with httpx.AsyncClient(timeout=self.timeout) as client:
+                response = await client.get(url, headers=headers, params=params)
+
+                if response.status_code == 200:
+                    return response.json()
+                elif response.status_code == 401:
+                    logger.warning("sentry_api_unauthorized", endpoint=endpoint)
+                    return None
+                else:
+                    logger.warning(
+                        "sentry_api_error",
+                        status_code=response.status_code,
+                        endpoint=endpoint,
+                    )
+                    return None
+
+        except httpx.TimeoutException:
+            logger.error("sentry_api_timeout", endpoint=endpoint)
+            return None
+        except Exception as e:
+            logger.error("sentry_api_failed", endpoint=endpoint, error=str(e))
+            return None
+
+    # =========================================================================
+    # Organization APIs
+    # =========================================================================
+
+    async def list_projects(self) -> list[dict] | None:
+        """取得組織內所有專案"""
+        return await self._request(f"organizations/{self.org}/projects/")
+
+    # =========================================================================
+    # Project APIs
+    # =========================================================================
+
+    async def list_issues(
+        self,
+        project: str | None = None,
+        query: str = "is:unresolved",
+        limit: int = 25,
+        cursor: str | None = None,
+    ) -> list[dict] | None:
+        """
+        列出專案 Issues
+
+        Args:
+            project: 專案 slug (預設使用設定值)
+            query: Sentry 搜尋語法
+            limit: 每頁數量
+            cursor: 分頁游標
+        """
+        project_slug = project or self.project
+        params: dict[str, Any] = {"query": query, "limit": limit}
+        if cursor:
+            params["cursor"] = cursor
+
+        return await self._request(
+            f"projects/{self.org}/{project_slug}/issues/",
+            params=params,
+        )
+
+    async def get_project_stats(
+        self,
+        project: str | None = None,
+        stat: str = "received",
+        resolution: str = "1h",
+    ) -> list | None:
+        """
+        取得專案統計數據
+
+        Args:
+            project: 專案 slug
+            stat: 統計類型 (received, rejected, blacklisted)
+            resolution: 時間解析度 (1h, 1d, etc.)
+        """
+        project_slug = project or self.project
+        return await self._request(
+            f"projects/{self.org}/{project_slug}/stats/",
+            params={"stat": stat, "resolution": resolution},
+        )
+
+    # =========================================================================
+    # Issue APIs
+    # =========================================================================
+
+    async def get_issue(self, issue_id: str) -> dict | None:
+        """取得 Issue 詳情"""
+        return await self._request(f"issues/{issue_id}/")
+
+    async def get_issue_events(
+        self,
+        issue_id: str,
+        limit: int = 1,
+        full: bool = False,
+    ) -> list[dict] | None:
+        """
+        取得 Issue 事件
+
+        Args:
+            issue_id: Issue ID
+            limit: 事件數量
+            full: 是否包含完整堆疊
+        """
+        params: dict[str, Any] = {"limit": limit}
+        if full:
+            params["full"] = "true"
+
+        return await self._request(f"issues/{issue_id}/events/", params=params)
+
+    # =========================================================================
+    # Helper Methods
+    # =========================================================================
+
+    def get_issue_url(self, issue_id: str) -> str:
+        """取得 Issue 在 Sentry UI 的連結"""
+        return f"{self.base_url}/organizations/{self.org}/issues/{issue_id}/"
+
+
+# =============================================================================
+# Singleton
+# =============================================================================
+
+_sentry_service: SentryService | None = None
+
+
+def get_sentry_service() -> SentryService:
+    """取得 Sentry Service 實例 (Singleton)"""
+    global _sentry_service
+    if _sentry_service is None:
+        _sentry_service = SentryService()
+    return _sentry_service
+
+
+def set_sentry_service(service: SentryService) -> None:
+    """設定 Sentry Service 實例 (for testing)"""
+    global _sentry_service
+    _sentry_service = service

apps/api/src/services/stats_service.py

@@ -13,7 +13,8 @@ Stats Service - Phase 17 P0 Router 層違規修復
 """
 
 import json
-from typing import Any, Callable, Coroutine
+from collections.abc import Callable, Coroutine
+from typing import Any
 
 import structlog
 

apps/api/src/services/token_counter.py

@@ -32,7 +32,7 @@ SignOz 指標:
 
 import time
 from dataclasses import dataclass, field
-from datetime import UTC, datetime, timedelta
+from datetime import UTC, datetime
 from typing import Protocol
 
 import structlog
@@ -40,7 +40,6 @@ from opentelemetry import metrics
 from opentelemetry.metrics import Counter, Histogram, Meter
 
 from src.core.config import settings
-from src.services.langfuse_client import get_langfuse
 
 logger = structlog.get_logger(__name__)
 
@@ -450,7 +449,7 @@ class TokenCounter:
         # 建議
         recommendation = ""
         if is_over_budget:
-            recommendation = f"建議切換到本地模型 (Ollama) 以節省成本"
+            recommendation = "建議切換到本地模型 (Ollama) 以節省成本"
         elif alert_triggered:
             recommendation = f"接近預算上限 ({max(daily_usage_percent, monthly_usage_percent):.1f}%)，考慮減少 {provider} 呼叫"