fix(lint): ruff auto-fix + lewooogo-core src 加入 git

- Python: ruff --fix 修復 280 個 lint 錯誤
- lewooogo-core: src/ 目錄未追蹤，導致 CI eslint 失敗

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

apps/api/scripts/demo_multisig.py

@@ -16,21 +16,21 @@ CISO-101 Multi-Sig Demo Script
 """
 
 import sys
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
-from datetime import datetime, timezone, timedelta
 
 # Add parent to path for imports
 sys.path.insert(0, str(Path(__file__).parent.parent))
 
+from src.core.trust_engine import TrustEngine, get_required_signatures
 from src.models.approval import (
     ApprovalRequestCreate,
     ApprovalStatus,
-    RiskLevel,
     BlastRadius,
     DataImpact,
     DryRunCheck,
+    RiskLevel,
 )
-from src.core.trust_engine import TrustEngine, get_required_signatures
 
 
 def print_header(title: str) -> None:
@@ -125,7 +125,7 @@ def main():
             DryRunCheck(name="Backup Available", passed=False, message="No recent backup!"),
         ],
         requested_by="ClawBot",
-        expires_at=datetime.now(timezone.utc) + timedelta(hours=1),
+        expires_at=datetime.now(UTC) + timedelta(hours=1),
     )
 
     approval = engine.create_approval(request)

apps/api/scripts/e2e_openclaw_test.py

@@ -15,7 +15,6 @@ Phase 5 E2E 點火測試 - OpenClaw 全鏈路驗證
 """
 
 import asyncio
-import json
 import sys
 from datetime import datetime
 
@@ -94,12 +93,10 @@ Traceback (most recent call last):
     print_step(2, "安全攔截器 (Security Interceptor)")
 
     try:
+        from src.core.config import settings
         from src.services.security_interceptor import (
             TelegramSecurityInterceptor,
-            UserNotWhitelistedError,
-            NonceReplayError,
         )
-        from src.core.config import settings
 
         interceptor = TelegramSecurityInterceptor()
 
@@ -139,7 +136,7 @@ Traceback (most recent call last):
     print_step(3, "Telegram Gateway (SOUL.md 訊息格式)")
 
     try:
-        from src.services.telegram_gateway import TelegramMessage, RISK_EMOJI_MAP
+        from src.services.telegram_gateway import RISK_EMOJI_MAP, TelegramMessage
 
         # 建立測試訊息
         message = TelegramMessage(
@@ -180,7 +177,7 @@ Traceback (most recent call last):
     print_step(4, "OpenClaw AI 模組載入")
 
     try:
-        from src.services.openclaw import get_openclaw, OpenClawService
+        from src.services.openclaw import OpenClawService, get_openclaw
 
         openclaw = get_openclaw()
         assert isinstance(openclaw, OpenClawService)

apps/api/scripts/fire_live_alert.py

@@ -25,13 +25,10 @@ import hashlib
 import hmac
 import json
 import os
-import sys
-from datetime import datetime, timezone
-from typing import Literal
+from datetime import UTC, datetime
 
 import httpx
 
-
 # =============================================================================
 # Configuration
 # =============================================================================
@@ -186,7 +183,7 @@ def fire_alert(
         dict: API 回應
     """
     print_header(f"AWOOOI 實彈射擊 - {alert_type.upper()}")
-    print(f"執行時間: {datetime.now(timezone.utc).isoformat()}")
+    print(f"執行時間: {datetime.now(UTC).isoformat()}")
     print(f"目標端點: {api_url}{WEBHOOK_ENDPOINT}")
 
     # 取得告警模板
@@ -334,7 +331,7 @@ def main():
     print_header("AWOOOI 實彈射擊系統")
     print(f"API 端點: {args.api_url}")
     print(f"HMAC 配置: {'已設定' if args.hmac_secret else '未設定 (dev mode)'}")
-    print(f"Shadow Mode: 已啟用 (K8s 操作將被安全攔截)")
+    print("Shadow Mode: 已啟用 (K8s 操作將被安全攔截)")
 
     if args.all:
         # 發射所有類型的告警

apps/api/scripts/test_phase63_aggregation.py

@@ -19,10 +19,9 @@ Phase 6.3 聚合測試腳本
 """
 
 import asyncio
-import json
-import httpx
 from datetime import datetime
-import time
+
+import httpx
 
 # API 端點
 API_BASE = "http://localhost:8000"
@@ -105,7 +104,7 @@ async def main():
     print("Phase 6.3 聚合測試")
     print("=" * 60)
     print(f"時間: {datetime.now().isoformat()}")
-    print(f"目標: 驗證 3 筆同源告警聚合到 1 個 Incident")
+    print("目標: 驗證 3 筆同源告警聚合到 1 個 Incident")
     print()
 
     async with httpx.AsyncClient() as client:

apps/api/scripts/test_phase64_proposal.py

@@ -182,7 +182,7 @@ async def main():
     proposal_result = await generate_proposal(incident_id)
 
     if not proposal_result or not proposal_result.get("success"):
-        print(f"    [FAIL] 無法生成 Proposal")
+        print("    [FAIL] 無法生成 Proposal")
         print(f"    message: {proposal_result.get('message') if proposal_result else 'N/A'}")
         return
 
@@ -212,7 +212,7 @@ async def main():
     for approval in pending.get("approvals", []):
         if approval.get("id") == proposal.get("id"):
             found = True
-            print(f"    [FOUND] Proposal 出現在待簽核清單中!")
+            print("    [FOUND] Proposal 出現在待簽核清單中!")
             print()
             print("    === PendingApprovalsResponse JSON ===")
             print(json.dumps({
@@ -223,7 +223,7 @@ async def main():
 
     if not found:
         print("    [WARN] Proposal 未出現在待簽核清單中")
-        print(f"    (可能因為 risk_level=LOW 已自動批准)")
+        print("    (可能因為 risk_level=LOW 已自動批准)")
 
     # 5. 最終驗證
     print("\n" + "=" * 70)

apps/api/scripts/test_race_condition.py

@@ -183,7 +183,7 @@ async def main():
     success_count = sum(1 for r in results if r["success"])
     fail_count = sum(1 for r in results if not r["success"])
 
-    print(f"\n發射結果:")
+    print("\n發射結果:")
     print(f"    成功: {success_count}/{CONCURRENT_SIGNALS}")
     print(f"    失敗: {fail_count}/{CONCURRENT_SIGNALS}")
     print(f"    耗時: {duration:.3f} 秒")
@@ -218,7 +218,7 @@ async def main():
     severity = incident.get("severity", "N/A")
     affected_services = incident.get("affected_services", [])
 
-    print(f"\n找到 Incident:")
+    print("\n找到 Incident:")
     print(f"    incident_id: {incident_id}")
     print(f"    signal_count: {signal_count}")
     print(f"    severity: {severity}")

apps/api/scripts/test_signal_stream.py

@@ -16,13 +16,11 @@ Phase 6.1 測試腳本: Redis Streams Signal 流程驗證
 """
 
 import asyncio
-import json
 import os
 import sys
 
 import httpx
 
-
 API_BASE_URL = os.getenv("API_BASE_URL", "http://localhost:8000")
 SIGNAL_ENDPOINT = f"{API_BASE_URL}/api/v1/webhooks/signals"
 
@@ -55,7 +53,7 @@ async def main():
     print(f"[1] 發送測試 Signal 到 {SIGNAL_ENDPOINT}")
     try:
         result = await send_test_signal()
-        print(f"    ✅ 成功!")
+        print("    ✅ 成功!")
         print(f"    Message ID: {result.get('message_id')}")
         print(f"    Stream: {result.get('stream')}")
     except httpx.HTTPStatusError as e:

apps/api/scripts/tracer_bullet_2.py

@@ -290,7 +290,7 @@ class TracerBullet2:
             "executedAt": datetime.utcnow().isoformat(),
         }
 
-        print(f"\n[EXECUTION RESULT]")
+        print("\n[EXECUTION RESULT]")
         print(f"  Status: {execution_result['status']}")
         print(f"  Output: {execution_result['output']['message']}")
         print(f"  Restart Time: {execution_result['output']['restartTime']}")

apps/api/src/agents/__init__.py

@@ -12,10 +12,10 @@ Agents:
 符合 leWOOOgo BRAIN 積木介面
 """
 
-from src.agents.base import BaseAgent, AgentResult
-from src.agents.security import SecurityAgent, SecurityResult
+from src.agents.action_planner import ActionPlan, ActionPlannerAgent
+from src.agents.base import AgentResult, BaseAgent
 from src.agents.blast_radius import BlastRadiusAgent, BlastRadiusResult
-from src.agents.action_planner import ActionPlannerAgent, ActionPlan
+from src.agents.security import SecurityAgent, SecurityResult
 
 __all__ = [
     "BaseAgent",

apps/api/src/agents/base.py

@@ -10,7 +10,7 @@ Base Agent - 專家 Agent 基礎類別
 
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from enum import Enum
 from typing import Any, Generic, TypeVar
 
@@ -52,7 +52,7 @@ class AgentResult:
     latency_ms: int
     error: str | None = None
     raw_response: dict[str, Any] = field(default_factory=dict)
-    timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
+    timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))
 
     def to_dict(self) -> dict[str, Any]:
         """轉換為 dict (API 回傳用)"""

apps/api/src/agents/security.py

@@ -11,7 +11,6 @@ Security Agent - 安全風險評估專家
 符合 ADR-009 SecurityAgent 規範
 """
 
-import asyncio
 import time
 from dataclasses import dataclass, field
 from typing import Any

apps/api/src/api/v1/agents.py

@@ -22,7 +22,7 @@ Phase 9.4-9.5 核心功能:
 
 import asyncio
 import json
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from enum import Enum
 from typing import Any
 from uuid import uuid4
@@ -33,12 +33,10 @@ from pydantic import BaseModel, Field
 
 from src.core.logging import get_logger
 from src.core.redis_client import get_redis
-from src.core.sse import SSEEvent, EventType, get_publisher
-from src.models.incident import Incident, Severity, Signal, IncidentStatus
+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,
-    ConsensusResult,
-    AgentType,
 )
 
 router = APIRouter(prefix="/agents", tags=["Agent Teams"])
@@ -232,7 +230,7 @@ async def run_agent_analysis(
             "final_reasoning": result.final_reasoning,
             "opinions": [op.to_dict() for op in result.opinions],
             "dissenting_opinions": result.dissenting_opinions,
-            "completed_at": datetime.now(timezone.utc).isoformat(),
+            "completed_at": datetime.now(UTC).isoformat(),
         }
 
         await redis_client.set(
@@ -274,7 +272,7 @@ async def run_agent_analysis(
             "state": TaskState.FAILED.value,
             "progress": 0,
             "error": str(e),
-            "completed_at": datetime.now(timezone.utc).isoformat(),
+            "completed_at": datetime.now(UTC).isoformat(),
         }
 
         await redis_client.set(
@@ -388,7 +386,7 @@ async def analyze(
                     alert_name=alert_name,
                     severity=Severity(request.severity),
                     source="manual",
-                    fired_at=datetime.now(timezone.utc),
+                    fired_at=datetime.now(UTC),
                 ))
 
         incident = Incident(
@@ -405,7 +403,7 @@ async def analyze(
         )
 
     # 建立任務
-    task_id = f"TASK-{datetime.now(timezone.utc).strftime('%Y%m%d')}-{uuid4().hex[:8].upper()}"
+    task_id = f"TASK-{datetime.now(UTC).strftime('%Y%m%d')}-{uuid4().hex[:8].upper()}"
 
     # 初始化任務狀態
     task_data = {
@@ -416,7 +414,7 @@ async def analyze(
         "agents_completed": 0,
         "total_agents": 4,
         "incident_id": incident.incident_id,
-        "started_at": datetime.now(timezone.utc).isoformat(),
+        "started_at": datetime.now(UTC).isoformat(),
     }
 
     await redis_client.set(
@@ -632,7 +630,7 @@ async def trigger_agent_analysis_for_incident(
         return None
 
     # 建立任務
-    task_id = f"TASK-{datetime.now(timezone.utc).strftime('%Y%m%d')}-{uuid4().hex[:8].upper()}"
+    task_id = f"TASK-{datetime.now(UTC).strftime('%Y%m%d')}-{uuid4().hex[:8].upper()}"
 
     task_data = {
         "task_id": task_id,
@@ -642,7 +640,7 @@ async def trigger_agent_analysis_for_incident(
         "agents_completed": 0,
         "total_agents": 4,
         "incident_id": incident_id,
-        "started_at": datetime.now(timezone.utc).isoformat(),
+        "started_at": datetime.now(UTC).isoformat(),
         "trigger": "auto",
     }
 

apps/api/src/api/v1/ai.py

@@ -31,8 +31,8 @@ from src.models.approval import (
     DryRunCheck,
     RiskLevel,
 )
-from src.services.openclaw import get_openclaw
 from src.services.host_aggregator import HostAggregator
+from src.services.openclaw import get_openclaw
 
 router = APIRouter(prefix="/ai", tags=["AI Decision"])
 logger = get_logger("awoooi.ai")

apps/api/src/api/v1/approvals.py

@@ -25,14 +25,13 @@ import re
 from typing import TYPE_CHECKING
 from uuid import UUID
 
-from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, status, Header
+from fastapi import APIRouter, BackgroundTasks, Depends, Header, HTTPException, status
 
 if TYPE_CHECKING:
     from src.services.notifications import ExecutionStatus
 
 from src.core.config import settings
 from src.core.logging import get_logger
-from src.services.approval_db import get_approval_service, get_timeline_service
 from src.models.approval import (
     ApprovalRequest,
     ApprovalRequestCreate,
@@ -42,6 +41,7 @@ from src.models.approval import (
     SignRequest,
     SignResponse,
 )
+from src.services.approval_db import get_approval_service, get_timeline_service
 from src.services.executor import OperationType, get_executor
 from src.services.proposal_service import get_proposal_service
 
@@ -279,7 +279,7 @@ async def execute_approved_action(approval: ApprovalRequest) -> None:
         await timeline.add_event(
             event_type="exec",
             status="error",
-            title=f"執行失敗: 無法解析操作類型",
+            title="執行失敗: 無法解析操作類型",
             description=f"Action: {approval.action}",
             actor="leWOOOgo",
             actor_role="executor",
@@ -380,11 +380,11 @@ async def _send_execution_notification(
 
     將執行結果發送至所有已配置的通知頻道 (Discord, Slack, etc.)
     """
-    from src.services.notifications import (
-        get_notification_manager,
-        NotificationMessage,
-    )
     from src.core.config import settings
+    from src.services.notifications import (
+        NotificationMessage,
+        get_notification_manager,
+    )
 
     if not settings.NOTIFICATION_ENABLED:
         logger.info("notification_disabled", approval_id=str(approval.id))

apps/api/src/api/v1/audit_logs.py

@@ -11,7 +11,7 @@ Endpoints:
 提供 K8s 操作執行的完整審計軌跡。
 """
 
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from typing import Any
 
 from fastapi import APIRouter, HTTPException, Query, status
@@ -233,7 +233,7 @@ async def get_audit_stats() -> AuditStatsResponse:
         by_namespace = {row[0]: row[1] for row in ns_result.fetchall()}
 
         # Last 24 hours
-        cutoff = datetime.now(timezone.utc) - timedelta(hours=24)
+        cutoff = datetime.now(UTC) - timedelta(hours=24)
         last24_result = await db.execute(
             select(func.count(AuditLog.id)).where(AuditLog.created_at >= cutoff)
         )

apps/api/src/api/v1/dashboard.py

@@ -10,7 +10,7 @@ Endpoints:
 """
 
 import asyncio
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from typing import Any
 
 from fastapi import APIRouter, Request
@@ -20,7 +20,7 @@ from pydantic import BaseModel
 from src.core.config import settings
 from src.core.logging import get_logger
 from src.core.sse import EventPublisher, EventType, SSEEvent, get_publisher
-from src.services.host_aggregator import HostAggregator, AggregatedStatus
+from src.services.host_aggregator import AggregatedStatus, HostAggregator
 
 router = APIRouter()
 logger = get_logger("awoooi.dashboard")
@@ -304,7 +304,7 @@ async def get_hosts() -> dict:
     """
     return {
         "hosts": settings.four_hosts,
-        "timestamp": datetime.now(timezone.utc).isoformat(),
+        "timestamp": datetime.now(UTC).isoformat(),
     }
 
 
@@ -317,7 +317,7 @@ async def get_stream_clients() -> dict:
     return {
         "client_count": pub.client_count,
         "is_running": pub.is_running,
-        "timestamp": datetime.now(timezone.utc).isoformat(),
+        "timestamp": datetime.now(UTC).isoformat(),
     }
 
 

apps/api/src/api/v1/health.py

@@ -17,7 +17,7 @@ Components Checked:
 """
 
 import asyncio
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from typing import Literal
 
 import httpx
@@ -103,7 +103,7 @@ async def check_postgresql() -> ComponentHealth:
         await writer.wait_closed()
         latency = (asyncio.get_event_loop().time() - start) * 1000
         return ComponentHealth(status="up", latency_ms=round(latency, 2))
-    except asyncio.TimeoutError:
+    except TimeoutError:
         logger.warning("postgresql_health_check_timeout")
         return ComponentHealth(status="down", error="timeout")
     except Exception as e:
@@ -127,7 +127,7 @@ async def check_redis() -> ComponentHealth:
         await writer.wait_closed()
         latency = (asyncio.get_event_loop().time() - start) * 1000
         return ComponentHealth(status="up", latency_ms=round(latency, 2))
-    except asyncio.TimeoutError:
+    except TimeoutError:
         logger.warning("redis_health_check_timeout")
         return ComponentHealth(status="down", error="timeout")
     except Exception as e:
@@ -213,7 +213,7 @@ async def get_health() -> HealthResponse:
         version=settings.VERSION,
         environment=settings.ENVIRONMENT,
         mock_mode=settings.MOCK_MODE,
-        timestamp=datetime.now(timezone.utc),
+        timestamp=datetime.now(UTC),
         components=components,
     )
 

apps/api/src/api/v1/incidents.py

@@ -17,16 +17,18 @@ Phase 6.4 核心功能:
 - Proposal 必須關聯到 Incident
 """
 
+from datetime import UTC
+from typing import Any
+
 from fastapi import APIRouter, HTTPException, status
 from pydantic import BaseModel
-from typing import Any
 
 from src.core.logging import get_logger
 from src.core.redis_client import get_redis
 from src.models.approval import ApprovalRequestResponse
 from src.models.incident import Incident, IncidentStatus, Severity
-from src.services.proposal_service import get_proposal_service
 from src.services.decision_manager import get_decision_manager
+from src.services.proposal_service import get_proposal_service
 
 router = APIRouter(prefix="/incidents", tags=["Incidents"])
 logger = get_logger("awoooi.incidents")
@@ -354,10 +356,12 @@ async def debug_resolve_incident(incident_id: str) -> dict[str, Any]:
     DEBUG: 直接更新 Incident 狀態為 RESOLVED
     用於測試 resolve_incident_after_approval 邏輯
     """
+    from datetime import datetime
+
+    from sqlalchemy import select
+
     from src.db.base import get_db_context
     from src.db.models import IncidentRecord
-    from sqlalchemy import select
-    from datetime import datetime, timezone
 
     redis_client = get_redis()
     error_msg = None
@@ -391,7 +395,7 @@ async def debug_resolve_incident(incident_id: str) -> dict[str, Any]:
         if data:
             incident = Incident.model_validate_json(data)
             incident.status = IncidentStatus.RESOLVED
-            incident.updated_at = datetime.now(timezone.utc)
+            incident.updated_at = datetime.now(UTC)
             await redis_client.set(
                 f"incident:{incident_id}",
                 incident.model_dump_json(),
@@ -410,7 +414,7 @@ async def debug_resolve_incident(incident_id: str) -> dict[str, Any]:
             record = result.scalar_one_or_none()
             if record:
                 record.status = "resolved"
-                record.updated_at = datetime.now(timezone.utc)
+                record.updated_at = datetime.now(UTC)
                 await db.commit()
                 db_updated = True
     except Exception as e:
@@ -423,7 +427,7 @@ async def debug_resolve_incident(incident_id: str) -> dict[str, Any]:
         if data:
             incident = Incident.model_validate_json(data)
             after_redis = incident.status.value
-    except Exception as e:
+    except Exception:
         pass
 
     after_db = None
@@ -434,7 +438,7 @@ async def debug_resolve_incident(incident_id: str) -> dict[str, Any]:
             record = result.scalar_one_or_none()
             if record:
                 after_db = record.status
-    except Exception as e:
+    except Exception:
         pass
 
     return {

apps/api/src/api/v1/metrics.py

@@ -11,15 +11,15 @@ Data Sources:
 - SQLite AuditLog: AI Success Rate (executed / total proposals)
 """
 
-from datetime import datetime, timezone, timedelta
+from datetime import UTC, datetime, timedelta
 from typing import Any
 
 from fastapi import APIRouter
 from pydantic import BaseModel
 
 from src.core.logging import get_logger
-from src.services.signoz_client import get_signoz_client
 from src.db.base import get_db_context
+from src.services.signoz_client import get_signoz_client
 
 logger = get_logger("awoooi.metrics")
 router = APIRouter()
@@ -73,7 +73,7 @@ async def calculate_ai_success_rate(hours: int = 24) -> tuple[float, list[float]
             from sqlalchemy import text
 
             # 時間範圍
-            cutoff = datetime.now(timezone.utc) - timedelta(hours=hours)
+            cutoff = datetime.now(UTC) - timedelta(hours=hours)
             cutoff_str = cutoff.isoformat()
 
             # Query: 統計 executed vs total (approved + executed + execution_failed)
@@ -245,7 +245,7 @@ async def get_gold_metrics(
     # Response
     # =========================================================================
     return GoldMetricsResponse(
-        timestamp=datetime.now(timezone.utc),
+        timestamp=datetime.now(UTC),
         service_name=service_name,
         metrics=metrics_list,
         raw_data=raw_data,
@@ -271,5 +271,5 @@ async def metrics_health() -> dict:
     return {
         "status": "healthy" if clickhouse_ok else "degraded",
         "clickhouse": "connected" if clickhouse_ok else "disconnected",
-        "timestamp": datetime.now(timezone.utc).isoformat(),
+        "timestamp": datetime.now(UTC).isoformat(),
     }

apps/api/src/api/v1/telegram.py

@@ -26,13 +26,13 @@ from pydantic import BaseModel
 
 from src.core.config import settings
 from src.core.logging import get_logger
-from src.services.telegram_gateway import get_telegram_gateway, TelegramGatewayError
-from src.services.security_interceptor import (
-    UserNotWhitelistedError,
-    NonceReplayError,
-)
-from src.services.approval_db import get_approval_service
 from src.models.approval import Signature, SignatureSource
+from src.services.approval_db import get_approval_service
+from src.services.security_interceptor import (
+    NonceReplayError,
+    UserNotWhitelistedError,
+)
+from src.services.telegram_gateway import TelegramGatewayError, get_telegram_gateway
 
 logger = get_logger("awoooi.telegram")
 router = APIRouter(prefix="/telegram", tags=["Telegram"])

apps/api/src/api/v1/webhooks.py

@@ -24,15 +24,17 @@ Endpoints:
 
 import hashlib
 import hmac
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from typing import Literal
 
-from fastapi import APIRouter, BackgroundTasks, HTTPException, status, Request, Header
+from fastapi import APIRouter, BackgroundTasks, Header, HTTPException, Request, status
 from pydantic import BaseModel, Field
 
 from src.core.config import settings
 from src.core.logging import get_logger
-from src.services.approval_db import get_approval_service
+
+# Phase 6.1: Event Bus (Redis Streams)
+from src.core.redis_client import get_redis
 from src.models.approval import (
     ApprovalRequestCreate,
     BlastRadius,
@@ -40,12 +42,13 @@ from src.models.approval import (
     DryRunCheck,
     RiskLevel,
 )
+from src.services.approval_db import get_approval_service
+
 # Phase 5: OpenClaw AI Engine
 from src.services.openclaw import get_openclaw
+
 # Phase 5: Telegram Gateway (行動戰情室)
-from src.services.telegram_gateway import get_telegram_gateway, TelegramGatewayError
-# Phase 6.1: Event Bus (Redis Streams)
-from src.core.redis_client import get_redis
+from src.services.telegram_gateway import TelegramGatewayError, get_telegram_gateway
 
 router = APIRouter(prefix="/webhooks", tags=["Webhooks"])
 logger = get_logger("awoooi.webhooks")
@@ -467,7 +470,7 @@ async def produce_signal_to_stream(signal: SignalPayload) -> str:
         "message": signal.message,
         "labels": str(signal.labels or {}),
         "annotations": str(signal.annotations or {}),
-        "received_at": datetime.now(timezone.utc).isoformat(),
+        "received_at": datetime.now(UTC).isoformat(),
     }
 
     # XADD 寫入 Stream

apps/api/src/config.py

@@ -1,4 +1,4 @@
 # Backward compatibility - re-export from core.config
-from src.core.config import Settings, settings, get_settings
+from src.core.config import Settings, get_settings, settings
 
 __all__ = ["Settings", "settings", "get_settings"]

apps/api/src/core/redis_client.py

@@ -15,8 +15,8 @@ Features:
 """
 
 import asyncio
+from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
-from typing import AsyncGenerator
 
 import redis.asyncio as redis
 import structlog

apps/api/src/core/sse.py

@@ -15,11 +15,11 @@ ADR-004: SSE 串流企業級實作模式 (Buffer + AbortController + Zustand)
 import asyncio
 import json
 import uuid
-from collections.abc import AsyncGenerator
+from collections.abc import AsyncGenerator, Callable
 from dataclasses import dataclass, field
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from enum import Enum
-from typing import Any, Callable
+from typing import Any
 
 from src.core.logging import get_logger
 
@@ -58,7 +58,7 @@ class SSEEvent:
     type: EventType
     data: dict[str, Any]
     id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
-    timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
+    timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))
     retry: int | None = None  # Client retry interval in ms
 
     def to_sse_format(self) -> str:
@@ -101,14 +101,14 @@ class SSEClient:
     """
     id: str = field(default_factory=lambda: str(uuid.uuid4()))
     queue: asyncio.Queue = field(default_factory=lambda: asyncio.Queue(maxsize=CLIENT_QUEUE_SIZE))
-    connected_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
-    last_activity: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
+    connected_at: datetime = field(default_factory=lambda: datetime.now(UTC))
+    last_activity: datetime = field(default_factory=lambda: datetime.now(UTC))
     is_active: bool = True
     metadata: dict[str, Any] = field(default_factory=dict)
 
     def touch(self) -> None:
         """Update last activity timestamp"""
-        self.last_activity = datetime.now(timezone.utc)
+        self.last_activity = datetime.now(UTC)
 
     async def send(self, event: SSEEvent) -> bool:
         """
@@ -357,7 +357,7 @@ class EventPublisher:
                         timeout=HEARTBEAT_INTERVAL + 5,
                     )
                     yield event.to_sse_format()
-                except asyncio.TimeoutError:
+                except TimeoutError:
                     # No event received, but connection might still be alive
                     # Heartbeat will be sent by background task
                     continue
@@ -404,7 +404,7 @@ class EventPublisher:
             try:
                 await asyncio.sleep(CLEANUP_INTERVAL)
 
-                now = datetime.now(timezone.utc)
+                now = datetime.now(UTC)
                 stale_threshold = HEARTBEAT_INTERVAL * 3  # 45 seconds
 
                 async with self._lock:

apps/api/src/core/telemetry.py

@@ -20,16 +20,15 @@ Traces → SigNoz (192.168.0.188:4317)
 """
 
 import logging
-from typing import Optional
 
 from opentelemetry import trace
-from opentelemetry.sdk.trace import TracerProvider
-from opentelemetry.sdk.trace.export import BatchSpanProcessor
-from opentelemetry.sdk.resources import Resource, SERVICE_NAME, SERVICE_VERSION
 from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
 from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
 from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
 from opentelemetry.instrumentation.logging import LoggingInstrumentor
+from opentelemetry.sdk.resources import SERVICE_NAME, SERVICE_VERSION, Resource
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
 
 from src.core.config import settings
 
@@ -37,7 +36,7 @@ from src.core.config import settings
 _logger = logging.getLogger("awoooi.telemetry")
 
 # Global state
-_tracer_provider: Optional[TracerProvider] = None
+_tracer_provider: TracerProvider | None = None
 _initialized: bool = False
 
 
@@ -204,7 +203,7 @@ def get_tracer(name: str = "awoooi"):
     return trace.get_tracer(name, settings.VERSION)
 
 
-def get_current_trace_id() -> Optional[str]:
+def get_current_trace_id() -> str | None:
     """
     Get current trace ID for log correlation
 

apps/api/src/core/trust_engine.py

@@ -14,8 +14,8 @@ Features:
 - 狀態轉換控制
 """
 
-from datetime import datetime, timezone
-from typing import Callable
+from collections.abc import Callable
+from datetime import UTC, datetime
 from uuid import UUID
 
 from src.models.approval import (
@@ -28,7 +28,6 @@ from src.models.approval import (
     Signature,
 )
 
-
 # =============================================================================
 # Risk Classification Rules
 # =============================================================================
@@ -249,7 +248,7 @@ class TrustEngine:
         # LOW 風險自動批准
         if risk_level == RiskLevel.LOW:
             approval.status = ApprovalStatus.APPROVED
-            approval.resolved_at = datetime.now(timezone.utc)
+            approval.resolved_at = datetime.now(UTC)
             if self._on_approved:
                 self._on_approved(approval)
 
@@ -263,7 +262,7 @@ class TrustEngine:
 
     def get_pending_approvals(self) -> list[ApprovalRequest]:
         """取得所有待簽核請求"""
-        now = datetime.now(timezone.utc)
+        now = datetime.now(UTC)
         pending = []
 
         for approval in self._approvals.values():
@@ -314,13 +313,13 @@ class TrustEngine:
             comment=comment,
         )
         approval.signatures.append(signature)
-        approval.updated_at = datetime.now(timezone.utc)
+        approval.updated_at = datetime.now(UTC)
 
         # 檢查是否滿足簽核數
         execution_triggered = False
         if approval.is_fully_signed:
             approval.status = ApprovalStatus.APPROVED
-            approval.resolved_at = datetime.now(timezone.utc)
+            approval.resolved_at = datetime.now(UTC)
             execution_triggered = True
 
             if self._on_approved:
@@ -355,8 +354,8 @@ class TrustEngine:
         # 更新狀態
         approval.status = ApprovalStatus.REJECTED
         approval.rejection_reason = f"[{rejector_name}] {reason}"
-        approval.resolved_at = datetime.now(timezone.utc)
-        approval.updated_at = datetime.now(timezone.utc)
+        approval.resolved_at = datetime.now(UTC)
+        approval.updated_at = datetime.now(UTC)
 
         if self._on_rejected:
             self._on_rejected(approval)
@@ -370,7 +369,7 @@ class TrustEngine:
         Returns:
             已過期的請求列表
         """
-        now = datetime.now(timezone.utc)
+        now = datetime.now(UTC)
         expired = []
 
         for approval in self._approvals.values():

apps/api/src/db/base.py

@@ -26,7 +26,6 @@ from sqlalchemy.orm import DeclarativeBase
 
 from src.core.config import settings
 
-
 # =============================================================================
 # Base Model
 # =============================================================================

apps/api/src/db/models.py

@@ -10,25 +10,26 @@ Schema 設計原則：
 - 索引優化查詢
 """
 
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from typing import Any
 from uuid import uuid4
 
 from sqlalchemy import (
+    JSON,
     DateTime,
-    Enum as SQLEnum,
     Index,
     Integer,
     String,
     Text,
-    JSON,
+)
+from sqlalchemy import (
+    Enum as SQLEnum,
 )
 from sqlalchemy.orm import Mapped, mapped_column
 
 from src.db.base import Base
 from src.models.approval import ApprovalStatus, RiskLevel
-from src.models.incident import Severity, IncidentStatus
-
+from src.models.incident import IncidentStatus, Severity
 
 # =============================================================================
 # Helper Functions
@@ -36,7 +37,7 @@ from src.models.incident import Severity, IncidentStatus
 
 def utc_now() -> datetime:
     """Get current UTC datetime"""
-    return datetime.now(timezone.utc)
+    return datetime.now(UTC)
 
 
 def generate_uuid() -> str:

apps/api/src/main.py

@@ -14,50 +14,51 @@ Version: 1.0.0
 Date: 2026-03-20
 """
 
+from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
-from typing import AsyncGenerator
 
 import structlog
 from fastapi import FastAPI, Request
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 
-from src.core.config import settings
-from src.core.logging import setup_logging, get_logger
-from src.core.sse import get_publisher
-from src.core.telemetry import setup_telemetry, shutdown_telemetry
-from src.core.http_client import init_all_http_clients, close_all_http_clients
-from src.core.redis_client import init_redis_pool, close_redis_pool
-
-# CTO-201: Database & Executor
-from src.db.base import init_db, close_db
-from src.services.executor import close_executor
-# Phase 5: OpenClaw AI Engine
-from src.services.openclaw import close_openclaw
-from src.services.telegram_gateway import get_telegram_gateway
-# Phase 6.1: Event Bus (Signal Worker)
-from src.workers import init_signal_worker, close_signal_worker
+from src.api.v1 import agents as agents_v1  # Phase 9.5: Agent Teams API
+from src.api.v1 import ai as ai_v1
+from src.api.v1 import approvals as approvals_v1
+from src.api.v1 import audit_logs as audit_logs_v1
+from src.api.v1 import dashboard as dashboard_v1
 
 # Import API routers
 from src.api.v1 import health as health_v1
-from src.api.v1 import dashboard as dashboard_v1
-from src.api.v1 import approvals as approvals_v1
-from src.api.v1 import ai as ai_v1
-from src.api.v1 import webhooks as webhooks_v1
-from src.api.v1 import timeline as timeline_v1
-from src.api.v1 import audit_logs as audit_logs_v1
-from src.api.v1 import telegram as telegram_v1  # Phase 5.4: Telegram Gateway
-from src.api.v1 import metrics as metrics_v1  # Phase 7: Gold Metrics (真實血脈)
 from src.api.v1 import incidents as incidents_v1  # Phase 6.4: Decision Proposal
+from src.api.v1 import metrics as metrics_v1  # Phase 7: Gold Metrics (真實血脈)
 from src.api.v1 import proposals as proposals_v1  # Phase 6.4h: Proposals CRUD API
-from src.api.v1 import agents as agents_v1  # Phase 9.5: Agent Teams API
+from src.api.v1 import telegram as telegram_v1  # Phase 5.4: Telegram Gateway
+from src.api.v1 import timeline as timeline_v1
+from src.api.v1 import webhooks as webhooks_v1
+from src.core.config import settings
+from src.core.http_client import close_all_http_clients, init_all_http_clients
+from src.core.logging import get_logger, setup_logging
+from src.core.redis_client import close_redis_pool, init_redis_pool
+from src.core.sse import get_publisher
+from src.core.telemetry import setup_telemetry, shutdown_telemetry
 
-# Legacy route imports (to be migrated)
-from src.routes import agent, plugins, pipelines, notifications
+# CTO-201: Database & Executor
+from src.db.base import close_db, init_db
 
 # Phase 6.4g: lewooogo-brain 積木路由
 from src.routers import proposals as proposals_router
 
+# Legacy route imports (to be migrated)
+from src.routes import agent, notifications, pipelines, plugins
+from src.services.executor import close_executor
+
+# Phase 5: OpenClaw AI Engine
+from src.services.openclaw import close_openclaw
+from src.services.telegram_gateway import get_telegram_gateway
+
+# Phase 6.1: Event Bus (Signal Worker)
+from src.workers import close_signal_worker, init_signal_worker
 
 # =============================================================================
 # Initialize Logging (MUST be first)

apps/api/src/models/__init__.py

@@ -20,10 +20,10 @@ from src.models.approval import (
     PendingApprovalsResponse,
     RejectRequest,
     RiskLevel,
-    SignRequest,
-    SignResponse,
     Signature,
     SignatureSource,
+    SignRequest,
+    SignResponse,
 )
 
 # Incident Models (Phase 6 - 認知覺醒)

apps/api/src/models/ai.py

@@ -10,6 +10,7 @@ CAI-101: ClawBot AI 結構化輸出模型
 """
 
 from enum import Enum
+
 from pydantic import BaseModel, Field, field_validator
 
 

apps/api/src/models/approval.py

@@ -10,13 +10,12 @@ Features:
 - Pydantic 強型別驗證
 """
 
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from enum import Enum
 from uuid import UUID, uuid4
 
 from pydantic import BaseModel, Field
 
-
 # =============================================================================
 # Enums
 # =============================================================================
@@ -102,7 +101,7 @@ class Signature(BaseModel):
     id: UUID = Field(default_factory=uuid4)
     signer_id: str = Field(..., description="簽核者 ID")
     signer_name: str = Field(..., description="簽核者名稱")
-    signed_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+    signed_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
     comment: str | None = None
 
     # Phase 5.4.5: Telegram 審計軌跡
@@ -153,14 +152,14 @@ class ApprovalRequest(ApprovalRequestBase):
     status: ApprovalStatus = Field(default=ApprovalStatus.PENDING)
     required_signatures: int = Field(..., description="所需簽核數")
     signatures: list[Signature] = Field(default_factory=list)
-    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
-    updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
+    updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
     resolved_at: datetime | None = Field(default=None, description="解決時間")
     rejection_reason: str | None = Field(default=None)
     # 戰略 B: 告警風暴收斂
     fingerprint: str | None = Field(default=None, description="告警指紋 Hash")
     hit_count: int = Field(default=1, description="聚合觸發次數")
-    last_seen_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc), description="最後觸發時間")
+    last_seen_at: datetime = Field(default_factory=lambda: datetime.now(UTC), description="最後觸發時間")
 
     @property
     def current_signatures(self) -> int:

apps/api/src/models/incident.py

@@ -20,7 +20,7 @@ C-Suite 戰略會議決議 (2026-03-22):
 - Semantic Memory (Vector DB): 向量化後的知識，供 RAG 檢索
 """
 
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from enum import Enum
 from typing import Literal
 from uuid import UUID, uuid4
@@ -30,7 +30,6 @@ from pydantic import BaseModel, Field
 # 復用現有模型 (避免重複定義)
 from src.models.approval import BlastRadius
 
-
 # =============================================================================
 # Incident 專用 Enums
 # =============================================================================
@@ -280,7 +279,7 @@ class Incident(BaseModel):
 
     # === 識別 ===
     incident_id: str = Field(
-        default_factory=lambda: f"INC-{datetime.now(timezone.utc).strftime('%Y%m%d')}-{str(uuid4())[:6].upper()}",
+        default_factory=lambda: f"INC-{datetime.now(UTC).strftime('%Y%m%d')}-{str(uuid4())[:6].upper()}",
         description="事件唯一識別碼 (如 INC-20260322-A1B2C3)",
     )
 
@@ -322,11 +321,11 @@ class Incident(BaseModel):
 
     # === 時間軸 ===
     created_at: datetime = Field(
-        default_factory=lambda: datetime.now(timezone.utc),
+        default_factory=lambda: datetime.now(UTC),
         description="事件建立時間",
     )
     updated_at: datetime = Field(
-        default_factory=lambda: datetime.now(timezone.utc),
+        default_factory=lambda: datetime.now(UTC),
         description="最後更新時間",
     )
     resolved_at: datetime | None = Field(

apps/api/src/plugins/finops/__init__.py

@@ -4,15 +4,15 @@ Phase 3.3: 閒置資源掃描與成本換算
 """
 
 from .cost_analyzer import (
-    IdleResourceScanner,
-    idle_scanner,
     CostReport,
-    WastedResource,
+    IdleResourceScanner,
+    PricingConfig,
     RecommendedAction,
     ResourceType,
-    PricingConfig,
     SavingsType,
+    WastedResource,
     WasteReason,
+    idle_scanner,
 )
 
 __all__ = [

apps/api/src/plugins/mcp/__init__.py

@@ -5,10 +5,10 @@ Phase 3: 企業功能 - AI 與外部工具橋樑
 
 from .mcp_bridge import (
     MCPBridge,
-    mcp_bridge,
+    MCPServer,
     MCPTool,
     MCPToolResult,
-    MCPServer,
+    mcp_bridge,
 )
 
 __all__ = [

apps/api/src/plugins/security/__init__.py

@@ -4,9 +4,9 @@ AWOOOI Security Plugins
 
 from .privacy_shield import (
     PrivacyShield,
-    privacy_shield,
-    SensitiveDataType,
     RedactionResult,
+    SensitiveDataType,
+    privacy_shield,
 )
 
 __all__ = [

apps/api/src/plugins/security/privacy_shield.py

@@ -14,10 +14,9 @@ Phase 2.4: 資料清理引擎
 """
 
 import re
+from collections.abc import Callable
 from dataclasses import dataclass, field
 from enum import Enum
-from typing import Callable
-
 
 # ==================== Types ====================
 

apps/api/src/routers/proposals.py

@@ -7,14 +7,13 @@ POST /api/v1/incidents/{incident_id}/propose
 整合真實 ProposalService + OpenClaw LLM 實現決策提案生成。
 """
 
-from fastapi import APIRouter, Depends, HTTPException, status
-from pydantic import BaseModel, Field
-from typing import List
 
 import structlog
+from fastapi import APIRouter, Depends, HTTPException, status
+from pydantic import BaseModel, Field
 
-from src.services.proposal_service import get_proposal_service, ProposalService
 from src.models.approval import RiskLevel as ApprovalRiskLevel
+from src.services.proposal_service import ProposalService, get_proposal_service
 
 logger = structlog.get_logger(__name__)
 
@@ -31,7 +30,7 @@ class ProposalCreateRequest(BaseModel):
 class ProposalResponse(BaseModel):
     proposal_id: str = Field(..., description="決策書唯一識別碼")
     incident_id: str = Field(..., description="關聯的事件 ID")
-    actions: List[str] = Field(..., description="生成的具體作戰指令清單")
+    actions: list[str] = Field(..., description="生成的具體作戰指令清單")
     tier: int = Field(..., description="判定之授權級別 (1: 自主, 2: 授權, 3: 親核)")
     guardrails_passed: bool = Field(..., description="是否完全通過防爆圈檢測")
     rejection_reason: str | None = Field(default=None, description="若未通過防爆圈，顯示阻擋原因")

apps/api/src/routes/approvals.py

@@ -11,16 +11,16 @@ from uuid import UUID, uuid4
 from fastapi import APIRouter, HTTPException
 from pydantic import BaseModel
 
-from src.services.dry_run import dry_run_engine
 from src.services.approval import (
-    multi_sig_engine,
     RISK_MATRIX,
-    InsufficientPermissionError,
-    DuplicateSignatureError,
-    TOCTOUConflictError,
-    ApprovalNotFoundError,
     ApprovalAlreadyDecidedError,
+    ApprovalNotFoundError,
+    DuplicateSignatureError,
+    InsufficientPermissionError,
+    TOCTOUConflictError,
+    multi_sig_engine,
 )
+from src.services.dry_run import dry_run_engine
 
 router = APIRouter()
 

apps/api/src/routes/health.py

@@ -16,7 +16,7 @@ Endpoints:
 
 import asyncio
 import time
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from typing import Literal
 
 import httpx
@@ -84,7 +84,7 @@ async def check_database() -> Literal["up", "down"]:
                 return "down"
         finally:
             await conn.close()
-    except asyncio.TimeoutError:
+    except TimeoutError:
         logger.warning("health_check_database", status="down", reason="timeout")
         return "down"
     except Exception as e:
@@ -122,7 +122,7 @@ async def check_redis() -> Literal["up", "down"]:
                 return "down"
         finally:
             await client.close()
-    except asyncio.TimeoutError:
+    except TimeoutError:
         logger.warning("health_check_redis", status="down", reason="timeout")
         return "down"
     except Exception as e:
@@ -243,7 +243,7 @@ async def get_health() -> HealthResponse:
         status=overall_status,
         version=settings.VERSION,
         environment=settings.ENVIRONMENT,
-        timestamp=datetime.now(timezone.utc),
+        timestamp=datetime.now(UTC),
         components=components,
     )
 

apps/api/src/services/__init__.py

@@ -2,51 +2,51 @@
 AWOOOI API Services
 """
 
-from .dry_run import DryRunEngine, DryRunResult, dry_run_engine
 from .approval import (
-    MultiSigEngine,
-    multi_sig_engine,
-    ApprovalState,
-    Signature,
-    UserRole,
-    ApprovalStatus,
     RISK_MATRIX,
+    ApprovalAlreadyDecidedError,
     # Exceptions
     ApprovalError,
-    InsufficientPermissionError,
-    DuplicateSignatureError,
-    TOCTOUConflictError,
     ApprovalNotFoundError,
-    ApprovalAlreadyDecidedError,
-)
-from .trust_engine import (
-    TrustScoreManager,
-    trust_engine,
-    TrustRecord,
-    RiskAdjustment,
-    RiskLevel,
-    TrustThresholds,
-    normalize_action_pattern,
-)
-from .graph_rag import (
-    TopologyGraph,
-    topology_graph,
-    ServiceNode,
-    DependencyEdge,
-    NodeType,
-    EdgeType,
-    HealthStatus,
-    BlastRadiusResult,
-    RootCauseResult,
-    FullAnalysisResult,
-    create_mock_topology,
+    ApprovalState,
+    ApprovalStatus,
+    DuplicateSignatureError,
+    InsufficientPermissionError,
+    MultiSigEngine,
+    Signature,
+    TOCTOUConflictError,
+    UserRole,
+    multi_sig_engine,
 )
 from .consensus_engine import (
-    ConsensusEngine,
-    get_consensus_engine,
-    ConsensusResult,
     AgentOpinion,
     AgentType,
+    ConsensusEngine,
+    ConsensusResult,
+    get_consensus_engine,
+)
+from .dry_run import DryRunEngine, DryRunResult, dry_run_engine
+from .graph_rag import (
+    BlastRadiusResult,
+    DependencyEdge,
+    EdgeType,
+    FullAnalysisResult,
+    HealthStatus,
+    NodeType,
+    RootCauseResult,
+    ServiceNode,
+    TopologyGraph,
+    create_mock_topology,
+    topology_graph,
+)
+from .trust_engine import (
+    RiskAdjustment,
+    RiskLevel,
+    TrustRecord,
+    TrustScoreManager,
+    TrustThresholds,
+    normalize_action_pattern,
+    trust_engine,
 )
 
 __all__ = [

apps/api/src/services/approval.py

@@ -19,8 +19,7 @@ from enum import Enum
 from typing import Literal
 from uuid import UUID
 
-from .dry_run import dry_run_engine, DryRunResult
-
+from .dry_run import DryRunResult, dry_run_engine
 
 # ==================== Types ====================
 

apps/api/src/services/approval_db.py

@@ -13,13 +13,14 @@ Features:
 - 與原有 API 契約相容
 """
 
-from datetime import datetime, timezone, timedelta
+from datetime import UTC, datetime, timedelta
 from typing import Any
 from uuid import UUID
 
 import structlog
-from sqlalchemy import select, update, and_, or_
+from sqlalchemy import and_, or_, select, update
 
+from src.core.trust_engine import classify_risk, get_required_signatures
 from src.db.base import get_db_context
 from src.db.models import ApprovalRecord, TimelineEvent
 from src.models.approval import (
@@ -32,7 +33,6 @@ from src.models.approval import (
     RiskLevel,
     Signature,
 )
-from src.core.trust_engine import classify_risk, get_required_signatures
 
 logger = structlog.get_logger(__name__)
 
@@ -82,7 +82,7 @@ def approval_record_to_request(record: ApprovalRecord) -> ApprovalRequest:
                     signer_name=sig.get("signer_name", ""),
                     timestamp=datetime.fromisoformat(sig["timestamp"])
                     if sig.get("timestamp")
-                    else datetime.now(timezone.utc),
+                    else datetime.now(UTC),
                     comment=sig.get("comment"),
                 )
             )
@@ -140,7 +140,7 @@ def approval_request_to_record_data(
                 "message": check.message,
             })
 
-    now = datetime.now(timezone.utc)
+    now = datetime.now(UTC)
     return {
         "action": request.action,
         "description": request.description,
@@ -261,7 +261,7 @@ class ApprovalDBService:
         Returns:
             ApprovalRequest if found, None otherwise
         """
-        now = datetime.now(timezone.utc)
+        now = datetime.now(UTC)
         cutoff_time = now - timedelta(minutes=debounce_minutes)
 
         async with get_db_context() as db:
@@ -304,7 +304,7 @@ class ApprovalDBService:
 
         這樣可以跳過 LLM 分析，節省 API 成本！
         """
-        now = datetime.now(timezone.utc)
+        now = datetime.now(UTC)
 
         async with get_db_context() as db:
             # 更新 hit_count 和 last_seen_at
@@ -358,7 +358,7 @@ class ApprovalDBService:
         """
         取得所有待簽核請求
         """
-        now = datetime.now(timezone.utc)
+        now = datetime.now(UTC)
 
         async with get_db_context() as db:
             # 先更新過期的請求
@@ -440,7 +440,7 @@ class ApprovalDBService:
             new_signature = {
                 "signer_id": signer_id,
                 "signer_name": signer_name,
-                "timestamp": datetime.now(timezone.utc).isoformat(),
+                "timestamp": datetime.now(UTC).isoformat(),
                 "comment": comment,
             }
             signatures.append(new_signature)
@@ -452,7 +452,7 @@ class ApprovalDBService:
             resolved_at = None
             if new_sig_count >= record.required_signatures:
                 new_status = ApprovalStatus.APPROVED
-                resolved_at = datetime.now(timezone.utc)
+                resolved_at = datetime.now(UTC)
                 execution_triggered = True
 
             # Phase 5: 樂觀鎖更新 - 使用 WHERE current_signatures = old_value
@@ -535,7 +535,7 @@ class ApprovalDBService:
 
             record.status = ApprovalStatus.REJECTED
             record.rejection_reason = f"{rejector_name}: {reason}"
-            record.resolved_at = datetime.now(timezone.utc)
+            record.resolved_at = datetime.now(UTC)
 
             await db.flush()
             await db.refresh(record)

apps/api/src/services/clawbot.py

@@ -16,10 +16,11 @@ Features:
 """
 
 import json
+import random
 import re
 import time
-import random
 from typing import Any
+
 import httpx
 import structlog
 

apps/api/src/services/consensus_engine.py

@@ -18,7 +18,7 @@ Features:
 
 import asyncio
 import json
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from enum import Enum
 from typing import Any
 from uuid import uuid4
@@ -63,7 +63,7 @@ class AgentOpinion(BaseModel):
     kubectl_command: str | None = None
     priority: int = Field(default=5, ge=1, le=10, description="優先度 1-10, 10 最高")
     estimated_impact: dict[str, Any] = Field(default_factory=dict)
-    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
 
     model_config = {"use_enum_values": False}
 
@@ -124,7 +124,7 @@ class ConsensusResult(BaseModel):
     final_reasoning: str
     risk_level: str
     dissenting_opinions: list[str] = Field(default_factory=list)
-    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
 
     model_config = {"use_enum_values": False}
 
@@ -382,7 +382,7 @@ class ConsensusEngine:
                     agent.analyze(incident),
                     timeout=timeout_sec / len(self._agents),
                 )
-            except asyncio.TimeoutError:
+            except TimeoutError:
                 logger.warning(
                     "agent_analyze_timeout",
                     agent_type=agent.agent_type.value,
@@ -511,7 +511,7 @@ class ConsensusEngine:
 
         整合所有專家意見，產生結構化的 ConsensusResult
         """
-        consensus_id = f"CON-{datetime.now(timezone.utc).strftime('%Y%m%d')}-{uuid4().hex[:8].upper()}"
+        consensus_id = f"CON-{datetime.now(UTC).strftime('%Y%m%d')}-{uuid4().hex[:8].upper()}"
 
         # 找出最佳的 kubectl 指令 (來自最高 priority + confidence 的意見)
         best_kubectl = None

apps/api/src/services/context_gatherer.py

@@ -234,9 +234,10 @@ class ContextGatherer:
     async def initialize(self) -> bool:
         """初始化 K8s 連線"""
         try:
+            from pathlib import Path
+
             from kubernetes_asyncio import client
             from kubernetes_asyncio.config import load_kube_config
-            from pathlib import Path
 
             kubeconfig_path = Path(settings.KUBECONFIG_PATH)
             if not kubeconfig_path.is_absolute():
@@ -432,36 +433,36 @@ class ContextGatherer:
             str: 格式化的上下文字串
         """
         parts = [
-            f"## K8s Context",
+            "## K8s Context",
             f"- **Resource**: {context.resource_type}/{context.resource_name}",
             f"- **Namespace**: {context.namespace}",
             f"- **Gathered At**: {context.gathered_at}",
         ]
 
         if context.pod_status:
-            parts.append(f"\n### Pod Status")
+            parts.append("\n### Pod Status")
             parts.append(f"- Phase: {context.pod_status.get('phase', 'Unknown')}")
             parts.append(f"- Restart Count: {context.pod_status.get('restart_count', 0)}")
             parts.append(f"- Conditions: {', '.join(context.pod_status.get('conditions', []))}")
 
         if context.deployment_status:
-            parts.append(f"\n### Deployment Status")
+            parts.append("\n### Deployment Status")
             parts.append(f"- Replicas: {context.deployment_status.get('replicas', 0)}")
             parts.append(f"- Ready: {context.deployment_status.get('ready_replicas', 0)}")
             parts.append(f"- Available: {context.deployment_status.get('available_replicas', 0)}")
 
         if context.recent_events:
-            parts.append(f"\n### Recent Events")
+            parts.append("\n### Recent Events")
             for event in context.recent_events:
                 parts.append(f"- [{event['type']}] {event['reason']}: {event['message']}")
 
         if context.filtered_logs:
-            parts.append(f"\n### Filtered Logs (ERROR Only)")
-            parts.append(f"```")
+            parts.append("\n### Filtered Logs (ERROR Only)")
+            parts.append("```")
             parts.append(context.filtered_logs[:2000])  # 限制長度
             if len(context.filtered_logs) > 2000:
-                parts.append(f"... (truncated)")
-            parts.append(f"```")
+                parts.append("... (truncated)")
+            parts.append("```")
 
             if context.log_filter_stats:
                 stats = context.log_filter_stats

apps/api/src/services/decision_manager.py

@@ -20,15 +20,15 @@ Decision Manager - Phase 6.5 非同步決策狀態機
 """
 
 import asyncio
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from enum import Enum
 from typing import Any
 from uuid import uuid4
 
 import structlog
 
-from src.core.redis_client import get_redis
 from src.core.config import settings
+from src.core.redis_client import get_redis
 from src.models.incident import Incident
 from src.services.openclaw import get_openclaw
 
@@ -50,7 +50,9 @@ async def _push_decision_to_telegram(
     """
     try:
         # 延遲導入避免循環依賴
-        from src.services.telegram_gateway import get_telegram_gateway, TelegramGatewayError
+        from src.services.telegram_gateway import (
+            get_telegram_gateway,
+        )
 
         # 檢查是否有設定 Bot Token
         if not settings.OPENCLAW_TG_BOT_TOKEN:
@@ -226,8 +228,8 @@ class DecisionToken:
         self.state = state
         self.proposal_data = proposal_data
         self.proposal_id = proposal_id
-        self.created_at = created_at or datetime.now(timezone.utc)
-        self.updated_at = updated_at or datetime.now(timezone.utc)
+        self.created_at = created_at or datetime.now(UTC)
+        self.updated_at = updated_at or datetime.now(UTC)
         self.error = error
 
     def to_dict(self) -> dict[str, Any]:
@@ -329,7 +331,7 @@ class DecisionManager:
 
             token.state = DecisionState.READY
             token.proposal_data = proposal_data
-            token.updated_at = datetime.now(timezone.utc)
+            token.updated_at = datetime.now(UTC)
 
             logger.info(
                 "decision_ready",
@@ -337,7 +339,7 @@ class DecisionManager:
                 source=proposal_data.get("source", "unknown"),
             )
 
-        except asyncio.TimeoutError:
+        except TimeoutError:
             # Timeout: 使用 Expert System 保底
             logger.warning(
                 "decision_timeout_using_expert",
@@ -348,7 +350,7 @@ class DecisionManager:
             expert_result = expert_analyze(incident)
             token.state = DecisionState.READY
             token.proposal_data = expert_result
-            token.updated_at = datetime.now(timezone.utc)
+            token.updated_at = datetime.now(UTC)
 
         except Exception as e:
             # 任何錯誤: 使用 Expert System 保底
@@ -362,7 +364,7 @@ class DecisionManager:
             token.state = DecisionState.READY
             token.proposal_data = expert_result
             token.error = str(e)
-            token.updated_at = datetime.now(timezone.utc)
+            token.updated_at = datetime.now(UTC)
 
         # 4. 儲存最終結果
         await self._save_token(token)
@@ -494,7 +496,7 @@ class DecisionManager:
             return None
 
         token.state = new_state
-        token.updated_at = datetime.now(timezone.utc)
+        token.updated_at = datetime.now(UTC)
         if proposal_id:
             token.proposal_id = proposal_id
 
@@ -583,7 +585,7 @@ class DecisionManager:
 
             token.state = DecisionState.READY
             token.proposal_data = proposal_data
-            token.updated_at = datetime.now(timezone.utc)
+            token.updated_at = datetime.now(UTC)
 
             logger.info(
                 "decision_ready_with_consensus",
@@ -592,7 +594,7 @@ class DecisionManager:
                 consensus_score=consensus_result.consensus_score,
             )
 
-        except asyncio.TimeoutError:
+        except TimeoutError:
             logger.warning(
                 "consensus_timeout_using_expert",
                 token=token.token,
@@ -602,7 +604,7 @@ class DecisionManager:
             expert_result = expert_analyze(incident)
             token.state = DecisionState.READY
             token.proposal_data = expert_result
-            token.updated_at = datetime.now(timezone.utc)
+            token.updated_at = datetime.now(UTC)
 
         except Exception as e:
             logger.exception(
@@ -614,7 +616,7 @@ class DecisionManager:
             token.state = DecisionState.READY
             token.proposal_data = expert_result
             token.error = str(e)
-            token.updated_at = datetime.now(timezone.utc)
+            token.updated_at = datetime.now(UTC)
 
         await self._save_token(token)
         return token

apps/api/src/services/executor.py

@@ -21,7 +21,7 @@ Supported Operations:
 import asyncio
 import time
 from dataclasses import dataclass
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from enum import Enum
 from pathlib import Path
 from typing import Any
@@ -111,9 +111,9 @@ class ActionExecutor:
         try:
             from kubernetes_asyncio import client
             from kubernetes_asyncio.config import (
+                ConfigException,
                 load_incluster_config,
                 load_kube_config,
-                ConfigException,
             )
 
             config_source = None
@@ -342,9 +342,7 @@ class ActionExecutor:
                 target=target,
                 namespace=namespace,
                 message="[SHADOW MODE] Operation blocked - dry-run only",
-                would_execute="kubectl rollout restart deployment/{name} -n {namespace}".format(
-                    name=name, namespace=namespace
-                ),
+                would_execute=f"kubectl rollout restart deployment/{name} -n {namespace}",
             )
             return ExecutionResult(
                 success=True,
@@ -378,7 +376,7 @@ class ActionExecutor:
                     "template": {
                         "metadata": {
                             "annotations": {
-                                "kubectl.kubernetes.io/restartedAt": datetime.now(timezone.utc).isoformat()
+                                "kubectl.kubernetes.io/restartedAt": datetime.now(UTC).isoformat()
                             }
                         }
                     }
@@ -417,7 +415,7 @@ class ActionExecutor:
                 },
             )
 
-        except asyncio.TimeoutError:
+        except TimeoutError:
             duration_ms = int((time.monotonic() - start_time) * 1000)
             error_msg = f"Operation timed out after {settings.K8S_OPERATION_TIMEOUT}s"
             logger.error(
@@ -480,9 +478,7 @@ class ActionExecutor:
                 target=target,
                 namespace=namespace,
                 message="[SHADOW MODE] Operation blocked - dry-run only",
-                would_execute="kubectl delete pod {name} -n {namespace}".format(
-                    name=name, namespace=namespace
-                ),
+                would_execute=f"kubectl delete pod {name} -n {namespace}",
             )
             return ExecutionResult(
                 success=True,
@@ -539,7 +535,7 @@ class ActionExecutor:
                 },
             )
 
-        except asyncio.TimeoutError:
+        except TimeoutError:
             duration_ms = int((time.monotonic() - start_time) * 1000)
             error_msg = f"Operation timed out after {settings.K8S_OPERATION_TIMEOUT}s"
             logger.error(
@@ -672,7 +668,7 @@ class ActionExecutor:
                     process.communicate(),
                     timeout=timeout_sec,
                 )
-            except asyncio.TimeoutError:
+            except TimeoutError:
                 process.kill()
                 await process.wait()
                 duration_ms = int((time.monotonic() - start_time) * 1000)
@@ -700,7 +696,7 @@ class ActionExecutor:
                 )
                 return ExecutionResult(
                     success=True,
-                    message=f"kubectl executed successfully",
+                    message="kubectl executed successfully",
                     operation_type=OperationType.RESTART_DEPLOYMENT,
                     target_resource=command[:50],
                     namespace="executed",
@@ -952,8 +948,8 @@ async def execute_approved_proposal(approval_id: str) -> ExecutionResult:
     Returns:
         ExecutionResult: 執行結果
     """
-    from src.services.multi_sig_redis import MultiSigRedisService
     from src.services.approval_db import get_approval_service
+    from src.services.multi_sig_redis import MultiSigRedisService
 
     logger.info(
         "execute_approved_proposal_start",
@@ -1064,7 +1060,7 @@ async def execute_approved_proposal(approval_id: str) -> ExecutionResult:
     # Step 5: 更新狀態
     new_status = "executed" if result.success else "failed"
     execution_log = {
-        "executed_at": datetime.now(timezone.utc).isoformat(),
+        "executed_at": datetime.now(UTC).isoformat(),
         "success": result.success,
         "stdout": result.k8s_response.get("stdout", "") if result.k8s_response else "",
         "stderr": result.error or "",

apps/api/src/services/graph_rag.py

@@ -440,7 +440,7 @@ class TopologyGraph:
 
 
 def create_mock_topology() -> TopologyGraph:
-    """
+    r"""
     建立 Mock 拓撲圖 (Phase 3 用)
 
     典型微服務架構:

apps/api/src/services/host_aggregator.py

@@ -20,7 +20,7 @@ Features:
 import asyncio
 import ssl
 from dataclasses import dataclass, field
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from enum import Enum
 from typing import Literal
 
@@ -94,7 +94,7 @@ class HostStatus:
     status: Literal["healthy", "degraded", "unhealthy", "unreachable"]
     services: list[ServiceStatus]
     metrics: HostMetrics | None = None
-    last_check: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
+    last_check: datetime = field(default_factory=lambda: datetime.now(UTC))
     error: str | None = None
 
 
@@ -222,7 +222,7 @@ async def _tcp_probe(ip: str, port: int, timeout: float = 3.0) -> tuple[bool, fl
         await writer.wait_closed()
         return True, round(latency, 2), None
 
-    except asyncio.TimeoutError:
+    except TimeoutError:
         return False, None, "timeout"
     except ConnectionRefusedError:
         return False, None, "connection refused"
@@ -481,7 +481,7 @@ class HostAggregator:
         )
 
         return AggregatedStatus(
-            timestamp=datetime.now(timezone.utc),
+            timestamp=datetime.now(UTC),
             environment=settings.ENVIRONMENT,
             mock_mode=False,  # Always real mode
             overall_status=overall,

apps/api/src/services/incident_engine.py

@@ -29,7 +29,7 @@ v1.1 重構內容 (2026-03-22 架構師審查後修正):
 """
 
 import json
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from typing import Any
 
 import structlog
@@ -40,7 +40,7 @@ from src.models.incident import (
     Severity,
     Signal,
 )
-from src.services.graph_rag import topology_graph, BlastRadiusResult
+from src.services.graph_rag import BlastRadiusResult, topology_graph
 from src.services.incident_memory import DualIncidentMemory, get_incident_memory
 
 logger = structlog.get_logger(__name__)
@@ -413,7 +413,7 @@ class IncidentEngine:
         # Signal 參數
         signal_json = signal.model_dump_json()
         severity_str = signal.severity.value
-        timestamp_str = datetime.now(timezone.utc).isoformat()
+        timestamp_str = datetime.now(UTC).isoformat()
 
         try:
             # 執行統一 Lua Script (原子操作)
@@ -604,7 +604,7 @@ class IncidentEngine:
             alert_name=signal_data.get("alert_name", "unknown"),
             severity=self._parse_severity(signal_data.get("severity", "warning")),
             source=self._parse_source(signal_data.get("source", "manual")),
-            fired_at=datetime.now(timezone.utc),
+            fired_at=datetime.now(UTC),
             labels=self._parse_dict(signal_data.get("labels", "{}")),
             annotations=self._parse_dict(signal_data.get("annotations", "{}")),
             fingerprint=signal_data.get("fingerprint"),

apps/api/src/services/incident_memory.py

@@ -17,7 +17,7 @@ NOTE: 此模組為 lewooogo-brain/adapters/incident_memory.py 的 apps/api 內
       待 Phase 6.4i 完成 monorepo Docker 解法後，將直接引用 lewooogo-brain 套件
 """
 
-from datetime import datetime, timezone, timedelta
+from datetime import UTC, datetime, timedelta
 from typing import Any, Protocol
 
 import structlog
@@ -334,7 +334,7 @@ class DualIncidentMemory:
                 return None
 
             # Step 3: 檢查聚合窗口
-            window_start = datetime.now(timezone.utc) - timedelta(minutes=window_minutes)
+            window_start = datetime.now(UTC) - timedelta(minutes=window_minutes)
             if incident.updated_at < window_start:
                 # 超出聚合窗口，不聚合
                 logger.debug(

apps/api/src/services/incident_service.py

@@ -17,7 +17,7 @@ Incident Service - Phase 6.2 雙層記憶寫入
 """
 
 import json
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from typing import Any, Literal
 
 import structlog
@@ -287,7 +287,7 @@ class IncidentService:
                 alert_name=signal_data.get("alert_name", "unknown"),
                 severity=self._parse_severity(signal_data.get("severity", "warning")),
                 source=self._parse_source(signal_data.get("source", "manual")),
-                fired_at=datetime.now(timezone.utc),
+                fired_at=datetime.now(UTC),
                 labels=self._parse_dict(signal_data.get("labels", "{}")),
                 annotations=self._parse_dict(signal_data.get("annotations", "{}")),
                 fingerprint=signal_data.get("fingerprint"),

apps/api/src/services/multi_sig_redis.py

@@ -16,12 +16,12 @@ Features:
 """
 
 import json
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from uuid import UUID
 
 import structlog
 
-from src.core.redis_client import get_redis, RedisLock
+from src.core.redis_client import RedisLock, get_redis
 
 logger = structlog.get_logger(__name__)
 
@@ -114,7 +114,7 @@ class MultiSigRedisService:
         """
         redis_client = get_redis()
         key = ApprovalStateRedis.get_key(approval_id)
-        now = datetime.now(timezone.utc).isoformat()
+        now = datetime.now(UTC).isoformat()
 
         state = {
             "id": str(approval_id),
@@ -236,7 +236,7 @@ class MultiSigRedisService:
                     raise RuntimeError(f"Already signed by: {signer_id}")
 
             # 新增簽名
-            now = datetime.now(timezone.utc).isoformat()
+            now = datetime.now(UTC).isoformat()
             new_signature = {
                 "signer_id": signer_id,
                 "signer_name": signer_name,
@@ -309,7 +309,7 @@ class MultiSigRedisService:
             if not state:
                 raise RuntimeError(f"Approval not found: {approval_id}")
 
-            now = datetime.now(timezone.utc).isoformat()
+            now = datetime.now(UTC).isoformat()
 
             updates = {
                 "status": status,
@@ -360,7 +360,7 @@ class MultiSigRedisService:
             if not state:
                 raise RuntimeError(f"Approval not found: {approval_id}")
 
-            now = datetime.now(timezone.utc).isoformat()
+            now = datetime.now(UTC).isoformat()
 
             await redis_client.hset(key, mapping={
                 "status": "rejected",

apps/api/src/services/notifications/__init__.py

@@ -9,7 +9,12 @@ NotificationProvider 介面 + 具體實作:
 - LineNotifyProvider (TODO)
 """
 
-from .base import NotificationProvider, NotificationMessage, NotificationResult, ExecutionStatus
+from .base import (
+    ExecutionStatus,
+    NotificationMessage,
+    NotificationProvider,
+    NotificationResult,
+)
 from .discord import DiscordWebhookProvider
 from .manager import NotificationManager, get_notification_manager
 

apps/api/src/services/notifications/base.py

@@ -11,7 +11,7 @@ Phase 6: leWOOOgo Output Plugins
 
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from enum import Enum
 from typing import Any
 
@@ -68,7 +68,7 @@ class NotificationMessage:
     confidence: float | None = None
 
     # 時間戳
-    timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
+    timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))
 
     @property
     def status_emoji(self) -> str:
@@ -117,7 +117,7 @@ class NotificationResult:
     message: str
     response_data: dict[str, Any] | None = None
     error: str | None = None
-    timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
+    timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))
 
 
 class NotificationProvider(ABC):

apps/api/src/services/notifications/discord.py

@@ -13,12 +13,13 @@ import httpx
 
 from src.core.config import settings
 from src.core.logging import get_logger
+
 from .base import (
-    NotificationProvider,
+    ExecutionStatus,
     NotificationMessage,
+    NotificationProvider,
     NotificationResult,
     NotificationStatus,
-    ExecutionStatus,
 )
 
 logger = get_logger("awoooi.notifications.discord")

apps/api/src/services/notifications/manager.py

@@ -7,9 +7,10 @@ Phase 6: leWOOOgo Output Plugins
 """
 
 from src.core.logging import get_logger
+
 from .base import (
-    NotificationProvider,
     NotificationMessage,
+    NotificationProvider,
     NotificationResult,
     NotificationStatus,
 )

apps/api/src/services/openclaw.py

@@ -20,10 +20,11 @@ Features:
 
 import hashlib
 import json
+import random
 import re
 import time
-import random
 from datetime import datetime
+
 import httpx
 import structlog
 
@@ -32,7 +33,7 @@ from src.core.redis_client import get_redis
 from src.models.ai import (
     OpenClawDecision,
 )
-from src.services.signoz_client import get_signoz_client, GoldMetrics
+from src.services.signoz_client import GoldMetrics, get_signoz_client
 
 logger = structlog.get_logger(__name__)
 

apps/api/src/services/proposal_service.py

@@ -18,7 +18,7 @@ Decision Proposal Service - Phase 6.4 決策輸出層
 - 所有決策必須可稽核
 """
 
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from uuid import UUID
 
 import structlog
@@ -32,6 +32,8 @@ from src.models.approval import (
     BlastRadius,
     DataImpact,
     DryRunCheck,
+)
+from src.models.approval import (
     RiskLevel as ApprovalRiskLevel,
 )
 from src.models.incident import (
@@ -40,8 +42,8 @@ from src.models.incident import (
     Severity,
 )
 from src.services.approval_db import get_approval_service
-from src.services.trust_engine import trust_engine, normalize_action_pattern
 from src.services.openclaw import get_openclaw
+from src.services.trust_engine import normalize_action_pattern, trust_engine
 
 logger = structlog.get_logger(__name__)
 
@@ -266,7 +268,7 @@ class ProposalService:
                     new_status="MITIGATING",
                 )
 
-            incident.updated_at = datetime.now(timezone.utc)
+            incident.updated_at = datetime.now(UTC)
 
             # 7. 更新 Redis + DB
             await self._persist_incident(incident)
@@ -551,7 +553,7 @@ class ProposalService:
                 incident = Incident.model_validate_json(data)
                 old_status = incident.status.value
                 incident.status = IncidentStatus.RESOLVED
-                incident.updated_at = datetime.now(timezone.utc)
+                incident.updated_at = datetime.now(UTC)
                 if incident.decision:
                     incident.decision.state = "completed"
                 await redis_client.set(key, incident.model_dump_json(), ex=604800)
@@ -579,7 +581,7 @@ class ProposalService:
                 record = result.scalar_one_or_none()
                 if record:
                     record.status = "resolved"
-                    record.updated_at = datetime.now(timezone.utc)
+                    record.updated_at = datetime.now(UTC)
                     await db.commit()
                     db_ok = True
                     logger.info(
@@ -607,7 +609,10 @@ class ProposalService:
         # 必須同步更新，否則下次 poll 會顯示 Y/n
         decision_ok = False
         try:
-            from src.services.decision_manager import get_decision_manager, DecisionState
+            from src.services.decision_manager import (
+                DecisionState,
+                get_decision_manager,
+            )
             decision_manager = get_decision_manager()
             existing_token = await decision_manager._find_existing_token(incident_id)
             if existing_token:

apps/api/src/services/signoz_client.py

@@ -14,10 +14,10 @@ Features:
 - ClickHouse HTTP API: 192.168.0.188:8123 (直查)
 """
 
-from dataclasses import dataclass, field
-from datetime import datetime, timezone, timedelta
 import json
 import time
+from dataclasses import dataclass, field
+from datetime import UTC, datetime, timedelta
 
 import structlog
 
@@ -237,7 +237,7 @@ class SignOzClient:
         Returns:
             GoldMetrics: 黃金指標數據
         """
-        now = datetime.now(timezone.utc)
+        now = datetime.now(UTC)
         start_time = now - timedelta(minutes=time_window_minutes)
         end_time = now
 
@@ -358,7 +358,7 @@ class SignOzClient:
             str: SignOz Trace URL with timestamps
         """
         if alert_timestamp is None:
-            alert_timestamp = datetime.now(timezone.utc)
+            alert_timestamp = datetime.now(UTC)
 
         link = SignOzTraceLink(
             base_url=self.signoz_url,
@@ -383,7 +383,7 @@ class SignOzClient:
 
         用於 High CPU / Disk Full 告警分析
         """
-        now = datetime.now(timezone.utc)
+        now = datetime.now(UTC)
         start_ms = int((now - timedelta(minutes=time_window_minutes)).timestamp() * 1000)
         end_ms = int(now.timestamp() * 1000)
 

apps/api/src/services/telegram_gateway.py

@@ -19,18 +19,18 @@ SOUL.md 鐵律 (4.1 Telegram 訊息壓縮原則):
 - 總長度: 800 字元 (v7.0 擴展以容納 SignOz 區塊)
 """
 
-from dataclasses import dataclass
-from datetime import datetime, timezone
 import asyncio
+from dataclasses import dataclass
+from datetime import UTC, datetime
 
 import httpx
 import structlog
 
 from src.core.config import settings
 from src.services.security_interceptor import (
-    get_security_interceptor,
-    UserNotWhitelistedError,
     NonceReplayError,
+    UserNotWhitelistedError,
+    get_security_interceptor,
 )
 
 logger = structlog.get_logger(__name__)
@@ -640,11 +640,11 @@ class TelegramGateway:
                     shadow_mode=True,
                 )
                 print(f"\n{'='*60}")
-                print(f"[SHADOW MODE] AI 生成的調優指令請求")
+                print("[SHADOW MODE] AI 生成的調優指令請求")
                 print(f"簽核單: {approval_id}")
                 print(f"執行者: @{username} (ID: {user_id})")
-                print(f"時間: {datetime.now(timezone.utc).isoformat()}")
-                print(f"狀態: 僅記錄，未實際執行")
+                print(f"時間: {datetime.now(UTC).isoformat()}")
+                print("狀態: 僅記錄，未實際執行")
                 print(f"{'='*60}\n")
 
                 return {
@@ -722,7 +722,7 @@ class TelegramGateway:
         elif action == "tune":
             stamp = f"⚡ 已由 @{username} 觸發自動調優 (Shadow Mode)"
             if extra_info:
-                stamp += f"\n📝 指令已記錄"
+                stamp += "\n📝 指令已記錄"
         else:
             stamp = f"✓ 已由 @{username} 處理"
 
@@ -1017,8 +1017,9 @@ class TelegramGateway:
             message_id: 訊息 ID
         """
         from uuid import UUID
-        from src.services.approval_db import get_approval_service
+
         from src.models.approval import Signature, SignatureSource
+        from src.services.approval_db import get_approval_service
 
         try:
             service = get_approval_service()
@@ -1043,11 +1044,11 @@ class TelegramGateway:
                         status=approval.status.value,
                     )
                     print(f"\n{'='*60}")
-                    print(f"✅ 統帥已授權執行！")
+                    print("✅ 統帥已授權執行！")
                     print(f"簽核單: {approval_id}")
                     print(f"簽核者: @{username} (ID: {user_id})")
                     print(f"狀態: {approval.status.value}")
-                    print(f"時間: {datetime.now(timezone.utc).isoformat()}")
+                    print(f"時間: {datetime.now(UTC).isoformat()}")
                     print(f"{'='*60}\n")
 
             elif action == "reject":
@@ -1065,7 +1066,7 @@ class TelegramGateway:
                         user_id=user_id,
                     )
                     print(f"\n{'='*60}")
-                    print(f"❌ 統帥已拒絕執行！")
+                    print("❌ 統帥已拒絕執行！")
                     print(f"簽核單: {approval_id}")
                     print(f"拒絕者: @{username}")
                     print(f"{'='*60}\n")
@@ -1101,7 +1102,7 @@ class TelegramGateway:
             if not self._initialized:
                 await self.initialize()
 
-            now = datetime.now(timezone.utc)
+            now = datetime.now(UTC)
 
             # 計算上次訊息時間
             last_msg_ago = "N/A"
@@ -1173,7 +1174,7 @@ class TelegramGateway:
             try:
                 # 1. 檢查沉默告警
                 if self._last_message_time:
-                    silence_duration = (datetime.now(timezone.utc) - self._last_message_time).total_seconds()
+                    silence_duration = (datetime.now(UTC) - self._last_message_time).total_seconds()
                     if silence_duration > silence_seconds:
                         # 發送沉默告警
                         hours = int(silence_duration // 3600)

apps/api/src/services/test_context_gatherer.py

@@ -8,6 +8,7 @@ Gate 2 Checkpoint: 驗證 ERROR Only 過濾邏輯
 """
 
 import pytest
+
 from src.services.context_gatherer import LogLevelFilter
 
 
@@ -192,7 +193,7 @@ Traceback (most recent call last):
         # 驗證: 過濾率應該很高 (約 60-70%)
         assert stats["removal_rate_percent"] > 50, f"Should filter >50%, got {stats['removal_rate_percent']}%"
 
-        print(f"\n📊 K8s Log Filter Stats:")
+        print("\n📊 K8s Log Filter Stats:")
         print(f"   Original: {stats['original_lines']} lines")
         print(f"   Filtered: {stats['filtered_lines']} lines")
         print(f"   Removed:  {stats['removed_lines']} lines ({stats['removal_rate_percent']}%)")

apps/api/src/workers/__init__.py

@@ -13,9 +13,9 @@ Phase 6.1: Event Bus Workers
 
 from src.workers.signal_worker import (
     SignalWorker,
+    close_signal_worker,
     get_signal_worker,
     init_signal_worker,
-    close_signal_worker,
 )
 
 __all__ = [

apps/api/src/workers/signal_worker.py

@@ -136,7 +136,7 @@ class SignalWorker:
             try:
                 # 給予 5 秒完成當前處理
                 await asyncio.wait_for(self._task, timeout=5.0)
-            except asyncio.TimeoutError:
+            except TimeoutError:
                 logger.warning("signal_worker_stop_timeout")
                 self._task.cancel()
             except asyncio.CancelledError:
@@ -325,8 +325,13 @@ async def _main() -> None:
 
     # Initialize settings first (loads env vars)
     from src.core.config import settings  # noqa: F401
-    from src.core.redis_client import init_redis_pool, close_redis_pool, init_worker_redis_pool, close_worker_redis_pool
-    from src.db.base import init_db, close_db
+    from src.core.redis_client import (
+        close_redis_pool,
+        close_worker_redis_pool,
+        init_redis_pool,
+        init_worker_redis_pool,
+    )
+    from src.db.base import close_db, init_db
 
     logger.info(
         "signal_worker_standalone_starting",

apps/api/tests/e2e_network_test.py

@@ -18,15 +18,13 @@ Phase 5 E2E 網路層測試 - HMAC 安全驗證 + Nonce 防重放
 import hashlib
 import hmac
 import json
-import pytest
 from unittest.mock import patch
 
-import httpx
+import pytest
 from httpx import ASGITransport, AsyncClient
 
-from src.main import app
 from src.core.config import settings
-
+from src.main import app
 
 # =============================================================================
 # Helper Functions
@@ -285,8 +283,8 @@ class TestTelegramSecurityInterceptor:
         同一個 Nonce 第二次使用應該被拒絕
         """
         from src.services.security_interceptor import (
-            TelegramSecurityInterceptor,
             NonceReplayError,
+            TelegramSecurityInterceptor,
         )
 
         interceptor = TelegramSecurityInterceptor()
@@ -432,7 +430,7 @@ class TestShadowMode:
 
         預期: 執行操作時僅記錄，不真正執行
         """
-        from src.services.executor import ActionExecutor, OperationType
+        from src.services.executor import ActionExecutor
 
         executor = ActionExecutor()
 

apps/api/tests/test_redis_multisig.py

@@ -16,9 +16,9 @@ Phase 6.1.1: 全自動單元自檢
 """
 
 import asyncio
-import sys
 import os
-from datetime import datetime, timezone
+import sys
+from datetime import UTC, datetime
 from uuid import uuid4
 
 # 添加專案路徑
@@ -43,7 +43,7 @@ async def test_redis_connection():
     logger.info("=" * 60)
     logger.info("TEST_1_REDIS_CONNECTION", status="starting")
 
-    from src.core.redis_client import init_redis_pool, get_redis, close_redis_pool
+    from src.core.redis_client import get_redis, init_redis_pool
 
     try:
         # 初始化連線池
@@ -253,8 +253,11 @@ async def test_ttl_verification():
     logger.info("=" * 60)
     logger.info("TEST_5_TTL_VERIFICATION", status="starting")
 
-    from src.services.multi_sig_redis import get_multi_sig_redis_service, APPROVAL_TTL_SECONDS
     from src.core.redis_client import get_redis
+    from src.services.multi_sig_redis import (
+        APPROVAL_TTL_SECONDS,
+        get_multi_sig_redis_service,
+    )
 
     service = get_multi_sig_redis_service()
     redis_client = get_redis()
@@ -409,7 +412,7 @@ async def main():
     """主測試入口"""
     logger.info("=" * 60)
     logger.info("PHASE_6_1_1_REDIS_MULTISIG_TEST", status="STARTING")
-    logger.info("timestamp", time=datetime.now(timezone.utc).isoformat())
+    logger.info("timestamp", time=datetime.now(UTC).isoformat())
     logger.info("=" * 60)
 
     results = {}

apps/api/tests/test_webhook_telegram_integration.py

@@ -14,17 +14,14 @@ Phase 5: 修復一級整合事故
     cd apps/api && pytest tests/test_webhook_telegram_integration.py -v
 """
 
-import json
-import pytest
-from unittest.mock import AsyncMock, patch, MagicMock
+from unittest.mock import AsyncMock, MagicMock, patch
 from uuid import UUID
 
-import httpx
+import pytest
 from httpx import ASGITransport, AsyncClient
 
-from src.main import app
 from src.core.config import settings
-
+from src.main import app
 
 # =============================================================================
 # Test Fixtures