awoooi/.agents/skills/02-lewooogo-backend-core.md

Skill 02: leWOOOgo Backend Core

後端核心引擎守護者

管轄範圍: apps/api/ (FastAPI, Python) 觸發條件: 任何 .py 檔案修改

---

文件資訊

欄位	值
版本	v1.5
建立日期	2026-03-20 (台北)
建立者	Claude Code
最後修改	2026-03-26 19:30 (台北)
修改者	Claude Code

變更紀錄

版本	日期	執行者	變更內容
v1.0	2026-03-20	Claude Code	初始建立
v1.1	2026-03-25	Claude Code	新增 MCP Tool 實作標準
v1.2	2026-03-25	Claude Code	加入文件資訊區塊
v1.3	2026-03-26	Claude Code	🔴🔴🔴 新增積木化強制執行章節 (32 項違規審計後)
v1.4	2026-03-26	Claude Code	📊 新增 Langfuse LLMOps 整合章節 (Phase 15.1)
v1.5	2026-03-26	Claude Code	🔴🔴 新增 UnitOfWork + Saga Pattern 章節 (ADR-027)

---

核心約束 (Six Iron Laws)

1. Async-First (非同步優先)

# ✅ 正確
async def get_health() -> HealthResponse:
    result = await check_services()
    return result

# ❌ 禁止
def get_health() -> HealthResponse:  # 同步函數
    result = check_services()
    return result

2. CORS 白名單鎖定 (無 Wildcard)

# ✅ 正確 (config.py)
CORS_ORIGINS: list[str] = [
    "https://awoooi.wooo.work",
    "http://localhost:3000",
]

# ❌ 絕對禁止
CORS_ORIGINS = ["*"]

3. Pydantic 強型別驗證

# ✅ 正確
class SignalPayload(BaseModel):
    source: str
    severity: Literal["P0", "P1", "P2", "P3"]
    message: str

# ❌ 禁止
def process_signal(data: dict):  # 無型別驗證
    pass

4. structlog 結構化日誌

# ✅ 正確
import structlog
logger = structlog.get_logger()
logger.info("signal_processed", signal_id=signal.id, severity=signal.severity)

# ❌ 禁止
print(f"Signal {signal.id} processed")
import logging  # 原生 logging

5. 台北時區鐵律 (禁止 UTC)

# ❌ 禁止 - UTC 時區
from datetime import UTC
datetime.now(UTC)
datetime.utcnow()

# ✅ 正確 - 台北時區工具
from src.utils.timezone import now_taipei, now_taipei_iso
now_taipei()           # datetime with +08:00
now_taipei_iso()       # '2026-03-25T02:08:04+08:00'

Memory 參考: feedback_timezone_taipei.md

6. ADR-013 Google Style Docstring

強制場景: 安全相關、複雜邏輯、外部依賴、危險操作

def restart_pod(namespace: str, pod_name: str) -> bool:
    """重啟指定 Pod。

    Args:
        namespace: K8s 命名空間
        pod_name: Pod 名稱 (不含 hash 後綴)

    Returns:
        True 表示重啟成功

    Raises:
        K8sPermissionError: 缺少 delete pods 權限

    Warning:
        此操作會導致短暫服務中斷
    """

---

OTEL 可觀測性 (P0 核心)

強制注入追蹤

# 所有 API 請求必須自動產生 traces
# 目標端點: http://192.168.0.188:24317 (SigNoz OTEL Collector)

# 環境變數 (必須設定)
OTEL_ENABLED=true
OTEL_EXPORTER_OTLP_ENDPOINT=http://192.168.0.188:24317
OTEL_SERVICE_NAME=awoooi-api

---

Langfuse LLMOps (Phase 15.1)

用途: LLM 呼叫追蹤、成本監控、Prompt 版本管理 端點: http://192.168.0.110:3100

環境變數

LANGFUSE_URL=http://192.168.0.110:3100
LANGFUSE_PUBLIC_KEY=pk-lf-xxx  # 從 K8s Secret
LANGFUSE_SECRET_KEY=sk-lf-xxx  # 從 K8s Secret

整合模式 (待實作)

from langfuse import Langfuse

langfuse = Langfuse(
    public_key=settings.LANGFUSE_PUBLIC_KEY,
    secret_key=settings.LANGFUSE_SECRET_KEY,
    host=settings.LANGFUSE_URL
)

# 包裝 LLM 呼叫
with langfuse.trace(name="openclaw_decision") as trace:
    generation = trace.generation(
        name="ollama_call",
        model="llama3.2:3b",
        input=prompt,
    )
    result = await _call_ollama(prompt)
    generation.end(output=result)

Memory 參考: reference_langfuse_credentials.md

---

機密管理 (嚴禁硬編碼)

# ✅ 正確 (從環境變數讀取)
settings.GEMINI_API_KEY
settings.CLAUDE_API_KEY
settings.WEBHOOK_HMAC_SECRET

# ❌ 絕對禁止
API_KEY = "sk-xxx..."  # 硬編碼機密

---

強制驗收程序 (Mandatory Validation)

修改任何 .py 後必須執行:

# Step 1: 語法檢查
cd apps/api && python -m py_compile src/core/config.py

# Step 2: 本地啟動測試 (MOCK_MODE)
docker run --rm -e MOCK_MODE=true -p 8888:8000 awoooi-api-test

# Step 3: Health Check
curl -sf http://localhost:8888/api/v1/health

驗收標準

檢查項目	通過條件
py_compile	無 SyntaxError
Docker Run	Application startup complete
Health API	HTTP 200, status: healthy/degraded

---

常見陷阱 (Known Pitfalls)

1. Pydantic v2 Field 命名: 禁止底線開頭 (_field_name)
2. JSON Array 環境變數: ConfigMap 中使用 '["a","b"]' 格式
3. httpx 超時: 預設 90 秒，長時間 AI 調用需調整

---

🚨 Redis Worker 專屬連線 (2026-03-23 教訓)

事故: Worker 使用 API 共用 Redis 連線池 (socket_timeout=5s)，與 XREADGROUP BLOCK 5000ms 衝突，導致每 5 秒 TimeoutError

鐵律: Worker 必須獨立連線池

# ✅ 正確: Worker 專屬長連線
async def init_worker_redis_pool() -> redis.Redis:
    return redis.from_url(
        settings.REDIS_URL,
        socket_timeout=None,  # 無限等待 (XREADGROUP 阻塞)
        socket_connect_timeout=10.0,
    )

# ❌ 禁止: Worker 共用 API 短超時連線
redis_client = get_redis()  # socket_timeout=5s
await redis_client.xreadgroup(block=5000)  # 必定超時

檢查清單

項目	正確值
API Redis socket_timeout	5.0 秒
Worker Redis socket_timeout	None (無限)
XREADGROUP block	5000ms

---

🚨 SQLite 禁止令 (2026-03-23 教訓)

事故: Worker 與 API Pod 各自使用本地 SQLite，導致「腦分裂」與 no such table: incidents 錯誤

鐵律: 絕對禁止 SQLite，強制 PostgreSQL

# ✅ 正確: 使用 DATABASE_URL (PostgreSQL)
def get_engine() -> AsyncEngine:
    database_url = settings.DATABASE_URL  # postgresql+asyncpg://...

    # 禁止 SQLite 守衛
    if "sqlite" in database_url.lower():
        logger.error("sqlite_forbidden")
        database_url = "postgresql+asyncpg://awoooi:xxx@192.168.0.188:5432/awoooi_prod"

    return create_async_engine(database_url, ...)

# ❌ 禁止: 使用 SQLITE_DATABASE_URL
settings.SQLITE_DATABASE_URL  # 已廢棄，禁止使用

生產環境連線

項目	值
主機	192.168.0.188
Port	5432
Database	awoooi_prod
連線字串前綴	postgresql+asyncpg://

---

🚨 函數重命名全域搜尋 (2026-03-23 教訓)

事故: init_redis() 重命名為 init_redis_pool()，但 Worker standalone 模式遺漏更新，導致 ImportError

鐵律: 重命名前必須全域搜尋

# 修改函數名稱前，必須執行:
grep -rn "old_function_name" apps/api/src/

# 確認所有呼叫點都已更新後，才能提交

高風險區域

• workers/signal_worker.py standalone _main() 函數
• main.py lifespan 初始化
• 測試檔案 tests/

---

🧱 leWOOOgo Memory Providers (Phase 6.4d - 2026-03-23)

新架構: 雙層記憶體 (Working + Episodic)

記憶層級

層級	Provider	儲存	TTL
Working Memory	RedisMemoryProvider	Redis	7 天
Episodic Memory	PgMemoryProvider	PostgreSQL	永久
雙層整合	DualMemoryProvider	兩者同步	-

使用方式

from lewooogo_data.providers import (
    RedisMemoryProvider,
    PgMemoryProvider,
    DualMemoryProvider,
    init_redis_pool,
    init_pg_engine,
)

# 初始化連線池 (啟動時執行)
await init_redis_pool()
await init_pg_engine()

# 建立 Provider
from your_models import Incident

# 單層使用
redis_memory = RedisMemoryProvider(Incident, key_prefix="incidents")
pg_memory = PgMemoryProvider(Incident)

# 雙層使用 (推薦)
dual_memory = DualMemoryProvider(Incident, key_prefix="incidents")

# CRUD 操作
await dual_memory.save("inc-001", incident)
data = await dual_memory.load("inc-001")  # Working 優先，Episodic 備援

鐵律

規則	說明
TTL 必須設定	Redis 所有 key 必須有 TTL，禁止無限累積
雙層同步	寫入時 Working + Episodic 同步
優雅降級	Redis 斷線不影響主流程
禁止直接存取	所有記憶體操作必須透過 Provider

檔案位置

packages/lewooogo-data/src/lewooogo_data/
├── interfaces/
│   └── memory_provider.py    # IMemoryProvider, IDualMemoryProvider
└── providers/
    ├── redis_memory.py       # RedisMemoryProvider
    ├── pg_memory.py          # PgMemoryProvider
    └── dual_memory.py        # DualMemoryProvider

---

🔴🔴 UnitOfWork + Saga Pattern (ADR-027)

用途: 確保 Incident-Approval 雙層寫入原子性 批准日期: 2026-03-26 Memory: feedback_incident_approval_sync.md, project_incident_approval_sync.md

問題背景

Incident 和 Approval 的建立涉及兩層儲存 (PostgreSQL + Redis)，必須確保：

1. 兩者同時成功或同時失敗
2. 狀態變更時雙向同步

核心模式

# ✅ 正確: 使用 IncidentApprovalService
from src.services.incident_approval_service import IncidentApprovalService

async def handle_alert(data: AlertData):
    service = IncidentApprovalService(session_factory, redis_client)
    incident, approval = await service.create_with_approval(
        incident_data, approval_data
    )

# ❌ 禁止: 分別建立 Incident 和 Approval
incident = await incident_repo.create(...)  # PostgreSQL
approval = await approval_repo.create(...)  # Redis 可能失敗，導致孤兒 Incident

UnitOfWork 模式

from src.core.unit_of_work import UnitOfWork

async with UnitOfWork(session_factory) as uow:
    # 所有操作在同一事務中
    incident = await self.incident_repo.create(uow.session, data)
    approval = await self.approval_repo.create(uow.session, data)

    # Redis 寫入 (事務外，需 Saga 補償)
    try:
        await self._write_to_redis(incident, approval)
    except RedisError:
        await uow.rollback()  # Saga 補償
        raise

鐵律

規則	說明
禁止單獨建立	Incident/Approval 必須透過 IncidentApprovalService
狀態同步	Approval 變更時必須同步 Incident
TTL 統一	使用 src/core/constants.py 定義的 TTL
Redis 失敗回滾	使用 Saga Pattern 補償 PostgreSQL 事務

檔案位置

apps/api/src/
├── core/
│   ├── unit_of_work.py        # 事務管理
│   └── constants.py           # TTL 常量
└── services/
    └── incident_approval_service.py  # 原子同步服務

---

🧰 MCP Tool 實作規範 (Phase 13.2)

目標: 將 Mock MCP Tool 升級為真實系統連接 優先級: P0 最優先

MCP Bridge 位置

apps/api/src/plugins/mcp/
├── __init__.py
├── mcp_bridge.py      # 核心橋樑
└── tools/             # 新增目錄
    ├── signoz_tool.py     # #79 SignOz MCP
    ├── kubernetes_tool.py # #80 Kubernetes MCP
    └── postgres_tool.py   # #81 PostgreSQL MCP

Tool 實作模板

from src.plugins.mcp.mcp_bridge import MCPTool, MCPToolResult

class SignOzTool:
    """SignOz 監控查詢 Tool"""

    async def query_traces(
        self,
        service_name: str,
        start_time: datetime,
        end_time: datetime,
    ) -> MCPToolResult:
        """查詢服務 Traces

        Args:
            service_name: 服務名稱 (awoooi-api, awoooi-web)
            start_time: 查詢起始時間 (台北時區)
            end_time: 查詢結束時間

        Returns:
            MCPToolResult with trace data

        Warning:
            SignOz API 有速率限制，避免高頻查詢
        """
        # 實作邏輯...

Tool 實作鐵律

規則	說明
Privacy Shield	所有 Tool 調用必須經過 RehydrationEngine
錯誤處理	外部系統失敗不影響主流程
日誌脫敏	禁止 log 敏感參數 (IP/Token)
超時設定	每個 Tool 必須設定合理 timeout
台北時區	所有時間參數必須使用 +8 時區

Tool 狀態

Tool	狀態	說明
Kubernetes	🟡 Mock	#80 待實作真實 kubectl
Database	🟡 Mock	#81 待連接 PostgreSQL
SignOz	❌ 缺失	#79 P0 查詢 Trace/Logs
維運手冊 RAG	❌ 缺失	#84 Markdown 向量化

---

🔴🔴🔴 積木化強制執行 (2026-03-26 審計後新增)

審計結果: 32 項違規，3 週開發工時浪費 Memory: feedback_lewooogo_modular_enforcement.md

修改前必問 5 題

1. 這個邏輯是否已存在於 packages/？ → 用 packages，禁止重寫
2. Router 是否只做 HTTP 轉發？ → 禁止 Router 直接存取 Redis/DB
3. Service 是否依賴 Interface？ → 必須用 Protocol/ABC
4. 是否可被其他模組重用？ → 可重用邏輯放 packages/
5. 是否遵循依賴注入？ → 禁止 global singleton

禁止清單 (Router 層)

# ❌ 禁止
from src.core.redis_client import get_redis  # 應透過 Service
from src.db.base import get_session          # 應透過 Repository
LUA_SCRIPT = """..."""                        # 應放在 Repository

正確架構層次

api/v1/*.py (Router) → services/*.py (Service) → packages/lewooogo-*/ (積木)

違規時

1. 停止修改，通報統帥
2. 必須修改違規檔案 → 同時修復違規
3. 新增功能 → 放在正確層次

---

參考文檔

• apps/api/src/core/config.py: 設定中心
• apps/api/src/main.py: FastAPI 應用入口
• apps/api/src/plugins/mcp/mcp_bridge.py: MCP Bridge 核心
• packages/lewooogo-data/: 記憶體 Provider 積木
• packages/lewooogo-brain/: AI 引擎積木
• memory/feedback_lewooogo_modular_enforcement.md: 積木化強制執行鐵律
• memory/feedback_incident_approval_sync.md: Incident-Approval 同步鐵律
• ADR-001: MCP Protocol 採用
• ADR-005: BFF 閘道架構
• ADR-006: AI 備援策略
• ADR-008: Python 模組化獨立積木架構
• ADR-027: Incident-Approval 同步架構 (UnitOfWork + Saga)