feat(awooop): Phase 1-8 完整實作 — AwoooP Agent Platform 六平面架構

## Phase 1-3: Control Plane + Contract System
- awooop_phase1_control_plane_2026-05-04.sql: 12 張核心表 + RLS
- awooop_phase1_batch1_rls_2026-05-04.sql: 全部 FORCE RLS + GRANT
- packages/awooop-contracts/: 六合約 JSON Schema + golden fixtures
- src/models/awooop_contracts.py: Pydantic v2 contract models（extra=forbid）
- src/repositories/contract_repository.py: contract lifecycle（draft→published→active）
- src/services/contract_service.py: HMAC publish sig + Redis multi-sig activate
- src/services/schema_validator.py: LLM output validator（retry×3, E-SCHEMA-001）

## Phase 2: Tenant Isolation
- awooop_phase2_budget_ledger_2026-05-04.sql: budget_ledger + RLS
- src/services/budget_service.py: Token Budget Hard Kill 三層防線
- src/core/context.py: PROJECT_ID ContextVar（31 background loop 自動繼承）
- src/db/base.py + models.py: project_id 欄位 + RLS set_config 注入
- src/hermes/nl_gateway.py: project_id Redis key 前綴（Phase A 雙寫）
- src/services/anomaly_counter.py: per-project 改造（Phase A fallback）

## Phase 4: Platform Shell in Shadow Mode
- awooop_phase4_run_state_2026-05-04.sql: run_state + step_journal + idempotency
- src/services/run_state_machine.py: 8-state FSM + SKIP LOCKED + stale reaper
- src/services/platform_runtime.py: UUID v7 + W3C trace_id + shadow_execute
- src/services/audit_sink.py: PII/secret redaction 9 patterns
- src/api/v1/platform/runs.py: POST/GET /v1/platform/runs（Router→Service 架構）
- src/workers/platform_worker.py: SKIP LOCKED worker + heartbeat + reaper loop
- src/main.py: platform router + lifespan worker start/stop

## Phase 5: MCP Gateway 五閘門
- awooop_phase5_mcp_gateway_2026-05-04.sql: 4 表 + RLS
- src/plugins/mcp/gateway.py: McpGateway（Gate 1~5, E-MCP-GATE-001~009）
- src/plugins/mcp/redaction_middleware.py: 雙層 redaction + 16K 截斷
- src/plugins/mcp/registry.py: __provider name mangling（ADR-116）
- src/plugins/mcp/credential_resolver.py: k8s secret ref 解析
- tests/test_mcp_credential_isolation.py: 10 個迴歸測試（secret leak 防再現）

## Phase 6-8: EwoooC + Channel Hub + Approval Token
- awooop_phase6_ewoooc_onboarding_2026-05-04.sql: ewoooc tenant + 4 read-only MCP tools
- awooop_phase7_channel_hub_2026-05-04.sql: conversation_event + outbound_message
- src/services/provider_proxy.py: ProviderProxy + PlatformEnvelope（ADR-115）
- src/services/channel_hub.py: Telegram inbound mirror + Progressive Feedback（30s）
- src/services/awooop_approval_token.py: HS256 + jti NX replay 防護 + suggest mode

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

apps/api/src/services/anomaly_counter.py

@@ -115,8 +115,20 @@ class AnomalyCounter:
     # TTL 設定 (35 天，比清理週期長一點)
     TTL_SECONDS = 35 * 24 * 3600
 
-    def __init__(self, redis_client: redis.Redis) -> None:
+    def __init__(self, redis_client: redis.Redis, project_id: str = "awoooi") -> None:
         self.redis = redis_client
+        self.project_id = project_id
+
+    def _pkey(self, prefix: str, key: str) -> str:
+        """新格式 key: {project_id}:{prefix}{key}（Phase A 多租戶）"""
+        return f"{self.project_id}:{prefix}{key}"
+
+    async def _redis_get_with_fallback(self, prefix: str, key: str) -> bytes | None:
+        """Phase A: 讀新 key，fallback 到舊 key。"""
+        val = await self.redis.get(self._pkey(prefix, key))
+        if val is None:
+            val = await self.redis.get(f"{prefix}{key}")
+        return val
 
     @staticmethod
     def derive_key_from_incident(incident) -> str | None:
@@ -217,7 +229,7 @@ class AnomalyCounter:
     ) -> AnomalyFrequency:
         """實際的異常記錄邏輯（可能拋出 Redis 異常）"""
         timestamp = now.timestamp()
-        timeline_key = f"{self.PREFIX_TIMELINE}{anomaly_key}"
+        timeline_key = self._pkey(self.PREFIX_TIMELINE, anomaly_key)
 
         # 1. 添加到 Sorted Set (score = timestamp, member = timestamp string)
         await self.redis.zadd(timeline_key, {str(timestamp): timestamp})
@@ -270,27 +282,22 @@ class AnomalyCounter:
             else now
         )
 
-        # 6. 讀取修復統計
-        repair_count_str = await self.redis.get(
-            f"{self.PREFIX_REPAIR_COUNT}{anomaly_key}"
-        )
+        # 6. 讀取修復統計（Phase A: 讀新 key，fallback 到舊 key）
+        repair_count_str = await self._redis_get_with_fallback(self.PREFIX_REPAIR_COUNT, anomaly_key)
         auto_repair_count = int(repair_count_str) if repair_count_str else 0
 
-        permanent_fix_str = await self.redis.get(
-            f"{self.PREFIX_PERMANENT_FIX}{anomaly_key}"
-        )
-        permanent_fix = permanent_fix_str == "1"
+        permanent_fix_str = await self._redis_get_with_fallback(self.PREFIX_PERMANENT_FIX, anomaly_key)
+        permanent_fix = permanent_fix_str == b"1" or permanent_fix_str == "1"
 
         # 7. 儲存 metadata (首次記錄時)
-        metadata_key = f"{self.PREFIX_METADATA}{anomaly_key}"
-        if not await self.redis.exists(metadata_key):
-            await self.redis.hset(
-                metadata_key,
-                mapping={
-                    "signature": json.dumps(anomaly_signature),
-                    "first_seen": now.isoformat(),
-                },
-            )
+        metadata_key = self._pkey(self.PREFIX_METADATA, anomaly_key)
+        legacy_metadata_key = f"{self.PREFIX_METADATA}{anomaly_key}"
+        if not await self.redis.exists(metadata_key) and not await self.redis.exists(legacy_metadata_key):
+            metadata_payload = {
+                "signature": json.dumps(anomaly_signature),
+                "first_seen": now.isoformat(),
+            }
+            await self.redis.hset(metadata_key, mapping=metadata_payload)
             await self.redis.expire(metadata_key, self.TTL_SECONDS)
 
         # 8. 判斷升級等級
@@ -353,14 +360,14 @@ class AnomalyCounter:
             success: 是否成功
         """
         try:
-            repair_key = f"{self.PREFIX_REPAIR_COUNT}{anomaly_key}"
+            repair_key = self._pkey(self.PREFIX_REPAIR_COUNT, anomaly_key)
 
             # 遞增修復嘗試次數
             await self.redis.incr(repair_key)
             await self.redis.expire(repair_key, self.TTL_SECONDS)
 
             # 記錄修復歷史 (用於學習)
-            history_key = f"{self.PREFIX_REPAIR_HISTORY}{anomaly_key}"
+            history_key = self._pkey(self.PREFIX_REPAIR_HISTORY, anomaly_key)
             await self.redis.lpush(
                 history_key,
                 json.dumps(
@@ -411,7 +418,7 @@ class AnomalyCounter:
             return
 
         try:
-            key = f"{self.PREFIX_DISPOSITION}{anomaly_key}"
+            key = self._pkey(self.PREFIX_DISPOSITION, anomaly_key)
             await self.redis.hincrby(key, disposition_type, 1)
             await self.redis.hincrby(key, "total", 1)
             await self.redis.expire(key, self.TTL_SECONDS)
@@ -434,8 +441,11 @@ class AnomalyCounter:
                    "cold_start_trust": N, "total": N}
         """
         try:
-            key = f"{self.PREFIX_DISPOSITION}{anomaly_key}"
+            key = self._pkey(self.PREFIX_DISPOSITION, anomaly_key)
             raw = await self.redis.hgetall(key)
+            if not raw:
+                # Phase A: fallback 到舊 key
+                raw = await self.redis.hgetall(f"{self.PREFIX_DISPOSITION}{anomaly_key}")
             return {
                 "auto_repair": int(raw.get(b"auto_repair", raw.get("auto_repair", 0))),
                 "human_approved": int(raw.get(b"human_approved", raw.get("human_approved", 0))),
@@ -471,11 +481,25 @@ class AnomalyCounter:
 
         try:
             # S2 Fix: 使用 Pipeline 批次查詢，消除 N+1 問題
-            pattern = f"{self.PREFIX_DISPOSITION}*"
+            # Phase A: 先掃新前綴，若無資料 fallback 到舊前綴
+            new_pattern = f"{self.project_id}:{self.PREFIX_DISPOSITION}*"
+            new_strip = f"{self.project_id}:{self.PREFIX_DISPOSITION}"
+            legacy_pattern = f"{self.PREFIX_DISPOSITION}*"
+            legacy_strip = self.PREFIX_DISPOSITION
+
             keys: list = []
-            async for key in self.redis.scan_iter(match=pattern, count=100):
+            async for key in self.redis.scan_iter(match=new_pattern, count=100):
                 keys.append(key)
 
+            if keys:
+                strip_prefix = new_strip
+                meta_prefix = f"{self.project_id}:{self.PREFIX_METADATA}"
+            else:
+                async for key in self.redis.scan_iter(match=legacy_pattern, count=100):
+                    keys.append(key)
+                strip_prefix = legacy_strip
+                meta_prefix = self.PREFIX_METADATA
+
             if not keys:
                 return total_summary, by_anomaly
 
@@ -489,11 +513,11 @@ class AnomalyCounter:
             anomaly_keys_str = []
             for key in keys:
                 key_str = key.decode() if isinstance(key, bytes) else key
-                anomaly_keys_str.append(key_str.replace(self.PREFIX_DISPOSITION, ""))
+                anomaly_keys_str.append(key_str.replace(strip_prefix, ""))
 
             meta_pipe = self.redis.pipeline(transaction=False)
             for ak in anomaly_keys_str:
-                meta_pipe.hget(f"{self.PREFIX_METADATA}{ak}", "signature")
+                meta_pipe.hget(f"{meta_prefix}{ak}", "signature")
             meta_results = await meta_pipe.execute()
 
             for i, raw in enumerate(results):
@@ -547,13 +571,13 @@ class AnomalyCounter:
         """
         try:
             await self.redis.set(
-                f"{self.PREFIX_PERMANENT_FIX}{anomaly_key}",
+                self._pkey(self.PREFIX_PERMANENT_FIX, anomaly_key),
                 "1",
                 ex=90 * 24 * 3600,  # 90 天
             )
 
             # 記錄修復詳情
-            metadata_key = f"{self.PREFIX_METADATA}{anomaly_key}"
+            metadata_key = self._pkey(self.PREFIX_METADATA, anomaly_key)
             await self.redis.hset(
                 metadata_key,
                 mapping={
@@ -588,8 +612,11 @@ class AnomalyCounter:
             }
         """
         try:
-            history_key = f"{self.PREFIX_REPAIR_HISTORY}{anomaly_key}"
+            history_key = self._pkey(self.PREFIX_REPAIR_HISTORY, anomaly_key)
             history = await self.redis.lrange(history_key, 0, -1)
+            if not history:
+                # Phase A: fallback 到舊 key
+                history = await self.redis.lrange(f"{self.PREFIX_REPAIR_HISTORY}{anomaly_key}", 0, -1)
 
             total = 0
             success_count = 0
@@ -627,8 +654,11 @@ class AnomalyCounter:
             }
         """
         try:
-            history_key = f"{self.PREFIX_REPAIR_HISTORY}{anomaly_key}"
+            history_key = self._pkey(self.PREFIX_REPAIR_HISTORY, anomaly_key)
             history = await self.redis.lrange(history_key, 0, -1)
+            if not history:
+                # Phase A: fallback 到舊 key
+                history = await self.redis.lrange(f"{self.PREFIX_REPAIR_HISTORY}{anomaly_key}", 0, -1)
 
             stats: dict[str, dict] = {}
 
@@ -666,11 +696,14 @@ class AnomalyCounter:
             AnomalyFrequency 或 None (若無記錄 或 Redis 重連失敗)
         """
         try:
-            timeline_key = f"{self.PREFIX_TIMELINE}{anomaly_key}"
+            timeline_key = self._pkey(self.PREFIX_TIMELINE, anomaly_key)
+            legacy_timeline_key = f"{self.PREFIX_TIMELINE}{anomaly_key}"
 
-            # 檢查是否有記錄
+            # Phase A: 若新 key 無資料，改用舊 key
             if not await self.redis.exists(timeline_key):
-                return None
+                if not await self.redis.exists(legacy_timeline_key):
+                    return None
+                timeline_key = legacy_timeline_key
 
             now = datetime.now()
             cutoff_30d = (now - timedelta(days=30)).timestamp()
@@ -716,16 +749,12 @@ class AnomalyCounter:
                 else now
             )
 
-            # 讀取修復統計
-            repair_count_str = await self.redis.get(
-                f"{self.PREFIX_REPAIR_COUNT}{anomaly_key}"
-            )
+            # 讀取修復統計（Phase A: 讀新 key，fallback 到舊 key）
+            repair_count_str = await self._redis_get_with_fallback(self.PREFIX_REPAIR_COUNT, anomaly_key)
             auto_repair_count = int(repair_count_str) if repair_count_str else 0
 
-            permanent_fix_str = await self.redis.get(
-                f"{self.PREFIX_PERMANENT_FIX}{anomaly_key}"
-            )
-            permanent_fix = permanent_fix_str == "1"
+            permanent_fix_str = await self._redis_get_with_fallback(self.PREFIX_PERMANENT_FIX, anomaly_key)
+            permanent_fix = permanent_fix_str in (b"1", "1")
 
             escalation_level = self._get_escalation_level(count_24h)
 
@@ -797,7 +826,7 @@ def get_anomaly_counter() -> AnomalyCounter:
     if _anomaly_counter is None:
         from src.core.redis_client import get_redis
 
-        _anomaly_counter = AnomalyCounter(get_redis())
+        _anomaly_counter = AnomalyCounter(get_redis(), project_id="awoooi")
     return _anomaly_counter
 
 

apps/api/src/services/audit_sink.py

@@ -0,0 +1,227 @@
+"""
+Audit Sink with PII/Secret Redaction
+======================================
+AwoooP Phase 4.4: Audit log 寫入前的 sanitization pipeline（ADR-116）
+2026-05-04 ogt + Claude Sonnet 4.6
+
+設計原則：
+- audit log 不記錄 raw LLM input/output，只記 hash + schema validation result
+- PII / secret pattern 硬攔（不可被 caller 繞過）
+- 攔截清單：GCP IP、PostgreSQL password、Telegram token、SSH key、Bearer token 等
+- redaction 後原值不可還原（替換為 [REDACTED:<type>]）
+- 所有 audit 寫入透過此 sink（禁止其他 service 直接 INSERT audit_logs）
+
+使用：
+    from src.services.audit_sink import write_audit
+
+    await write_audit(
+        project_id="awoooi",
+        action="run.completed",
+        resource_type="run",
+        resource_id=str(run_id),
+        details={"trace_id": trace_id, "cost_usd": 0.012},
+    )
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from typing import Any
+
+import structlog
+
+logger = structlog.get_logger(__name__)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Redaction patterns（ADR-116 P1-08）
+# ─────────────────────────────────────────────────────────────────────────────
+
+# 每個 pattern: (compiled_re, replacement_tag)
+_REDACTION_PATTERNS: list[tuple[re.Pattern[str], str]] = [
+    # Telegram bot token（數字:英數字母混合 32~64 字元）
+    (re.compile(r"\d{8,12}:[A-Za-z0-9_-]{32,64}"), "TELEGRAM_TOKEN"),
+
+    # PostgreSQL connection string
+    (re.compile(r"postgresql(?:\+asyncpg)?://[^:]+:[^@]+@[^/\s]+"), "PG_DSN"),
+
+    # Generic password in URL / config
+    (re.compile(r"(?i)(?:password|passwd|pwd)\s*[:=]\s*\S+"), "PASSWORD"),
+
+    # Bearer / Authorization header value
+    (re.compile(r"(?i)(?:bearer|token)\s+[A-Za-z0-9\-._~+/]+=*"), "BEARER_TOKEN"),
+
+    # AWS / GCP / NVIDIA API key patterns
+    (re.compile(r"(?i)(?:api[_-]?key|apikey)\s*[:=]\s*[A-Za-z0-9\-._]{20,}"), "API_KEY"),
+
+    # Private GCP internal IPs（ADR-116 禁止 GCP 內網 IP 進 log）
+    (re.compile(r"\b10\.\d{1,3}\.\d{1,3}\.\d{1,3}\b"), "INTERNAL_IP"),
+    (re.compile(r"\b172\.(?:1[6-9]|2\d|3[0-1])\.\d{1,3}\.\d{1,3}\b"), "INTERNAL_IP"),
+    (re.compile(r"\b192\.168\.\d{1,3}\.\d{1,3}\b"), "INTERNAL_IP"),
+
+    # SSH private key
+    (re.compile(r"-----BEGIN (?:RSA|EC|OPENSSH) PRIVATE KEY-----[\s\S]*?-----END [A-Z ]+ PRIVATE KEY-----"), "SSH_PRIVATE_KEY"),
+
+    # JWT（三段 base64 以 . 分隔）
+    (re.compile(r"eyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+"), "JWT_TOKEN"),
+
+    # Hex secret >= 32 位（可能是 HMAC key / session token）
+    (re.compile(r"\b[0-9a-f]{64}\b"), "HEX_SECRET_64"),
+]
+
+# 欄位名稱黑名單：這些 key 的 value 直接替換（不做 pattern 掃描）
+_BLOCKED_FIELD_NAMES = frozenset({
+    "password", "passwd", "pwd", "secret", "token", "api_key", "apikey",
+    "private_key", "private_key_pem", "bot_token", "telegram_token",
+    "hmac_key", "jwt", "authorization", "cookie", "session",
+})
+
+# LLM raw input/output 欄位名稱（只記 hash）
+_LLM_RAW_FIELDS = frozenset({
+    "raw_input", "raw_output", "llm_input", "llm_output",
+    "prompt", "completion", "system_prompt",
+})
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Sanitization pipeline
+# ─────────────────────────────────────────────────────────────────────────────
+
+def _redact_string(value: str) -> str:
+    """對字串套用所有 redaction patterns"""
+    for pattern, tag in _REDACTION_PATTERNS:
+        value = pattern.sub(f"[REDACTED:{tag}]", value)
+    return value
+
+
+def sanitize(details: dict[str, Any]) -> dict[str, Any]:
+    """
+    遞迴處理 details dict，套用所有 redaction 規則。
+
+    規則優先序：
+    1. key 在 _BLOCKED_FIELD_NAMES → value 替換為 [REDACTED:BLOCKED_FIELD]
+    2. key 在 _LLM_RAW_FIELDS → value 替換為 sha256(str(value))（只記 hash）
+    3. string value → pattern redaction
+    4. nested dict/list → 遞迴處理
+    """
+    return _sanitize_value(details, depth=0)
+
+
+def _sanitize_value(value: Any, depth: int = 0) -> Any:
+    if depth > 10:
+        return "[REDACTED:MAX_DEPTH]"
+
+    if isinstance(value, dict):
+        return {k: _sanitize_dict_entry(k, v, depth) for k, v in value.items()}
+    if isinstance(value, list):
+        return [_sanitize_value(item, depth + 1) for item in value]
+    if isinstance(value, str):
+        return _redact_string(value)
+    return value
+
+
+def _sanitize_dict_entry(key: str, value: Any, depth: int) -> Any:
+    key_lower = key.lower()
+
+    if key_lower in _BLOCKED_FIELD_NAMES:
+        return "[REDACTED:BLOCKED_FIELD]"
+
+    if key_lower in _LLM_RAW_FIELDS:
+        # 只記 sha256 hash，不記原始內容
+        raw_str = json.dumps(value, ensure_ascii=False) if not isinstance(value, str) else value
+        return f"[LLM_RAW_HASH:{hashlib.sha256(raw_str.encode()).hexdigest()[:16]}]"
+
+    return _sanitize_value(value, depth + 1)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Audit write
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def write_audit(
+    *,
+    project_id: str,
+    action: str,
+    resource_type: str,
+    resource_id: str,
+    details: dict[str, Any] | None = None,
+    run_id: str | None = None,
+    trace_id: str | None = None,
+) -> None:
+    """
+    統一 audit log 寫入入口（Phase 4+ 所有 service 必須透過此方法）。
+
+    1. sanitize details（PII / secret redaction）
+    2. 附加 run_id / trace_id（可觀測性）
+    3. INSERT audit_logs（非阻擋 background task）
+    """
+    import asyncio
+
+    asyncio.create_task(
+        _write_audit_impl(
+            project_id=project_id,
+            action=action,
+            resource_type=resource_type,
+            resource_id=resource_id,
+            details=details,
+            run_id=run_id,
+            trace_id=trace_id,
+        ),
+        name="audit_sink_write",
+    )
+
+
+async def _write_audit_impl(
+    *,
+    project_id: str,
+    action: str,
+    resource_type: str,
+    resource_id: str,
+    details: dict[str, Any] | None,
+    run_id: str | None,
+    trace_id: str | None,
+) -> None:
+    try:
+        from sqlalchemy import text as sa_text
+        from src.db.base import get_db_context
+
+        clean_details: dict[str, Any] = sanitize(details or {})
+        if run_id:
+            clean_details["_run_id"] = run_id
+        if trace_id:
+            clean_details["_trace_id"] = trace_id
+
+        async with get_db_context(project_id) as db:
+            await db.execute(
+                sa_text("""
+                    INSERT INTO audit_logs
+                        (project_id, action, resource_type, resource_id, details)
+                    VALUES
+                        (:project_id, :action, :resource_type, :resource_id, :details::jsonb)
+                """),
+                {
+                    "project_id": project_id,
+                    "action": action,
+                    "resource_type": resource_type,
+                    "resource_id": resource_id,
+                    "details": json.dumps(clean_details),
+                },
+            )
+    except Exception as exc:
+        logger.warning(
+            "audit_sink_write_failed",
+            action=action,
+            resource_id=resource_id,
+            error=str(exc),
+        )
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Convenience：可在測試中驗證 sanitization 結果
+# ─────────────────────────────────────────────────────────────────────────────
+
+def sanitize_for_test(details: dict[str, Any]) -> dict[str, Any]:
+    """同步 sanitize，供測試使用"""
+    return sanitize(details)

apps/api/src/services/awooop_approval_token.py

@@ -0,0 +1,357 @@
+"""
+AwoooP Approval Token — HS256 簽核令牌 + Multi-sig + Suggest Mode
+==================================================================
+AwoooP Phase 8: ADR-116 Gate 5 approval flow
+2026-05-04 ogt + Claude Sonnet 4.6
+
+功能：
+  1. HS256 Approval Token（自製，不依賴 PyJWT）：
+     - issue_approval_token() → signed token（3 段 base64url）
+     - verify_approval_token() → payload（含 jti/exp/sub/approver）
+     - jti 存 Redis NX（TTL = exp - now）防 token replay
+     - TTL = 15 分鐘（APPROVAL_TOKEN_TTL = 900s）
+
+  2. Multi-sig quorum：
+     - record_approval() → 驗 token + NX jti + SADD approver_id → 目前簽核數
+     - check_approval_quorum(required=1) → bool | raise QuorumNotMetError
+     - Redis Set TTL = 1h
+
+  3. Suggest Mode（AWOOOP_SUGGEST_MODE feature flag）：
+     - is_suggest_mode_enabled() → bool
+     - build_suggest_action(action_type, target) → SuggestedAction（dry-run）
+     - 支援 3 個 SRE flow：rollback / scale / restart
+
+Redis key 前綴（與 legacy multi_sig_redis.py 不衝突）：
+  awooop_appr:jti:{jti}       — NX token replay 防護
+  awooop_appr:sigs:{project_id}:{run_id}:{tool_name}  — 簽核人 Set
+
+錯誤碼：
+  E-APPR-001  token 無效或已過期
+  E-APPR-002  jti 已使用（replay attack）
+  E-APPR-003  quorum 未達
+  E-APPR-004  approver 重複簽核
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac as _hmac_module
+import json
+import os
+import time
+import uuid
+from dataclasses import dataclass, field
+from typing import Any
+
+import structlog
+
+logger = structlog.get_logger(__name__)
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 常數
+# ─────────────────────────────────────────────────────────────────────────────
+
+APPROVAL_TOKEN_TTL = 900           # 15 分鐘
+_JTI_KEY_PREFIX = "awooop_appr:jti:"
+_SIG_SET_PREFIX = "awooop_appr:sigs:"
+_SIG_TTL_SECONDS = 3600            # 簽核 Set 1h TTL
+_SUGGEST_MODE_ENV = "AWOOOP_SUGGEST_MODE"
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 錯誤定義
+# ─────────────────────────────────────────────────────────────────────────────
+
+class InvalidApprovalTokenError(Exception):
+    error_code = "E-APPR-001"
+
+class TokenReplayError(Exception):
+    error_code = "E-APPR-002"
+
+class QuorumNotMetError(Exception):
+    error_code = "E-APPR-003"
+
+class DuplicateApproverError(Exception):
+    error_code = "E-APPR-004"
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# HS256 Token 實作
+# ─────────────────────────────────────────────────────────────────────────────
+
+def _b64url_encode(data: bytes) -> str:
+    return base64.urlsafe_b64encode(data).rstrip(b"=").decode()
+
+
+def _b64url_decode(s: str) -> bytes:
+    padding = 4 - len(s) % 4
+    if padding != 4:
+        s += "=" * padding
+    return base64.urlsafe_b64decode(s)
+
+
+def _get_hmac_key() -> bytes:
+    try:
+        from src.core.config import settings
+        key = getattr(settings, "APPROVAL_HMAC_KEY", None) or ""
+    except Exception:
+        key = ""
+    key = key or os.environ.get("APPROVAL_HMAC_KEY", "")
+    if not key:
+        logger.warning("approval_hmac_key_not_set_using_dev_fallback")
+        key = "dev-awooop-approval-hmac-fallback"
+    return key.encode()
+
+
+def issue_approval_token(
+    *,
+    project_id: str,
+    run_id: str,
+    tool_name: str,
+    approver_id: str,
+    ttl_seconds: int = APPROVAL_TOKEN_TTL,
+) -> str:
+    """
+    產生 HS256 Approval Token。
+
+    payload：
+      jti = uuid4().hex（唯一 token ID，用於 Redis NX 防 replay）
+      iss = "awooop-approval"
+      sub = "{project_id}:{run_id}:{tool_name}"
+      approver = approver_id
+      iat / exp
+    """
+    now = int(time.time())
+    jti = uuid.uuid4().hex
+
+    header = {"alg": "HS256", "typ": "JWT"}
+    payload = {
+        "jti": jti,
+        "iss": "awooop-approval",
+        "sub": f"{project_id}:{run_id}:{tool_name}",
+        "approver": approver_id,
+        "iat": now,
+        "exp": now + ttl_seconds,
+    }
+
+    h_b64 = _b64url_encode(json.dumps(header, separators=(",", ":")).encode())
+    p_b64 = _b64url_encode(json.dumps(payload, separators=(",", ":")).encode())
+    signing_input = f"{h_b64}.{p_b64}"
+
+    sig = _hmac_module.new(
+        _get_hmac_key(),
+        signing_input.encode(),
+        hashlib.sha256,
+    ).digest()
+    return f"{signing_input}.{_b64url_encode(sig)}"
+
+
+def verify_approval_token(token: str) -> dict[str, Any]:
+    """
+    驗證 HS256 token，回傳 payload。
+
+    Raises:
+        InvalidApprovalTokenError: 簽名無效/過期/格式錯誤
+    """
+    try:
+        parts = token.split(".")
+        if len(parts) != 3:
+            raise InvalidApprovalTokenError("token 非 3 段格式")
+
+        h_b64, p_b64, sig_b64 = parts
+        signing_input = f"{h_b64}.{p_b64}"
+
+        expected_sig = _hmac_module.new(
+            _get_hmac_key(),
+            signing_input.encode(),
+            hashlib.sha256,
+        ).digest()
+
+        if not _hmac_module.compare_digest(sig_b64, _b64url_encode(expected_sig)):
+            raise InvalidApprovalTokenError("token 簽名無效")
+
+        payload = json.loads(_b64url_decode(p_b64))
+
+        if int(time.time()) > payload.get("exp", 0):
+            raise InvalidApprovalTokenError("token 已過期")
+
+        return payload
+
+    except InvalidApprovalTokenError:
+        raise
+    except Exception as exc:
+        raise InvalidApprovalTokenError(f"token 解析失敗: {exc}") from exc
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Multi-sig Redis approval
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def record_approval(
+    *,
+    project_id: str,
+    run_id: str,
+    tool_name: str,
+    approver_id: str,
+    token: str,
+) -> int:
+    """
+    記錄一筆簽核。步驟：
+    1. verify_approval_token（HS256 + exp）
+    2. sub 匹配驗證
+    3. Redis NX jti（防 replay）
+    4. Redis SADD approver_id（防重複）
+    5. 回傳目前簽核數
+
+    Raises:
+        InvalidApprovalTokenError, TokenReplayError, DuplicateApproverError
+    """
+    payload = verify_approval_token(token)
+
+    expected_sub = f"{project_id}:{run_id}:{tool_name}"
+    if payload.get("sub") != expected_sub:
+        raise InvalidApprovalTokenError(
+            f"token sub 不符（期望 '{expected_sub}'，實際 '{payload.get('sub')}'）"
+        )
+
+    jti = payload["jti"]
+    exp = payload["exp"]
+
+    try:
+        import aioredis
+        from src.core.config import settings
+
+        redis = aioredis.from_url(settings.REDIS_URL)
+
+        # jti NX
+        jti_key = f"{_JTI_KEY_PREFIX}{jti}"
+        ttl_remaining = max(exp - int(time.time()), 1)
+        ok = await redis.set(jti_key, "1", nx=True, ex=ttl_remaining)
+        if not ok:
+            await redis.aclose()
+            raise TokenReplayError(f"jti={jti!r} 已使用")
+
+        # SADD approver
+        sig_key = f"{_SIG_SET_PREFIX}{project_id}:{run_id}:{tool_name}"
+        added = await redis.sadd(sig_key, approver_id)
+        if added == 0:
+            await redis.aclose()
+            raise DuplicateApproverError(f"approver '{approver_id}' 已簽核")
+
+        await redis.expire(sig_key, _SIG_TTL_SECONDS)
+        count = int(await redis.scard(sig_key))
+        await redis.aclose()
+
+        logger.info(
+            "awooop_approval_recorded",
+            project_id=project_id,
+            run_id=run_id,
+            tool_name=tool_name,
+            approver_id=approver_id,
+            count=count,
+        )
+        return count
+
+    except (InvalidApprovalTokenError, TokenReplayError, DuplicateApproverError):
+        raise
+    except Exception as exc:
+        logger.exception("awooop_approval_redis_error", error=str(exc))
+        raise InvalidApprovalTokenError(f"Redis 錯誤: {exc}") from exc
+
+
+async def check_approval_quorum(
+    *,
+    project_id: str,
+    run_id: str,
+    tool_name: str,
+    required_count: int = 1,
+) -> bool:
+    """
+    檢查 quorum。Raises QuorumNotMetError if 不足。
+    """
+    try:
+        import aioredis
+        from src.core.config import settings
+
+        redis = aioredis.from_url(settings.REDIS_URL)
+        sig_key = f"{_SIG_SET_PREFIX}{project_id}:{run_id}:{tool_name}"
+        count = int(await redis.scard(sig_key))
+        await redis.aclose()
+
+        if count < required_count:
+            raise QuorumNotMetError(f"簽核數不足（{count}/{required_count}）")
+        return True
+
+    except QuorumNotMetError:
+        raise
+    except Exception as exc:
+        raise QuorumNotMetError(f"Redis 查詢失敗: {exc}") from exc
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Suggest Mode
+# ─────────────────────────────────────────────────────────────────────────────
+
+@dataclass
+class SuggestedAction:
+    """Suggest mode dry-run 結果（不真正執行）"""
+    action_type: str           # 'rollback' | 'scale' | 'restart'
+    target: str
+    suggested_command: str
+    rollback_evidence: dict[str, Any] = field(default_factory=dict)
+    dry_run: bool = True
+    approval_required: bool = True
+
+
+def is_suggest_mode_enabled() -> bool:
+    return os.environ.get(_SUGGEST_MODE_ENV, "").lower() in ("true", "1", "yes")
+
+
+async def build_suggest_action(
+    action_type: str,
+    *,
+    target: str,
+    run_id: str,
+    project_id: str,
+) -> SuggestedAction:
+    """
+    Suggest mode：返回 dry-run 建議，不執行真實操作。
+    支援 rollback / scale / restart 三個 SRE flow。
+    """
+    if action_type not in ("rollback", "scale", "restart"):
+        raise ValueError(f"不支援的 action_type: {action_type!r}")
+
+    if action_type == "rollback":
+        command = f"kubectl rollout undo deployment/{target}"
+        evidence: dict[str, Any] = {
+            "note": f"需確認 deployment/{target} 當前 image 與 rollout history",
+            "suggested_verification": f"kubectl rollout history deployment/{target}",
+        }
+    elif action_type == "scale":
+        command = f"kubectl scale deployment/{target} --replicas=<N>"
+        evidence = {
+            "note": f"需確認 deployment/{target} 當前 replicas 數量",
+            "suggested_verification": f"kubectl get deployment/{target} -o json | jq .spec.replicas",
+        }
+    else:  # restart
+        command = f"kubectl rollout restart deployment/{target}"
+        evidence = {
+            "note": f"需確認 deployment/{target} 當前 pod 狀態",
+            "suggested_verification": f"kubectl get pods -l app={target}",
+        }
+
+    logger.info(
+        "suggest_action_built",
+        project_id=project_id,
+        run_id=run_id,
+        action_type=action_type,
+        target=target,
+    )
+
+    return SuggestedAction(
+        action_type=action_type,
+        target=target,
+        suggested_command=command,
+        rollback_evidence=evidence,
+    )

apps/api/src/services/budget_service.py

@@ -0,0 +1,378 @@
+"""AwoooP Token Budget Hard Kill Service
+=======================================
+ADR-120: 三層 Hard Kill 防護架構
+2026-05-04 ogt + Claude Sonnet 4.6（Phase 2.6）
+
+防線：
+1. Pre-call check（呼叫前）— Layer 1 Tenant + Layer 2 Platform + Layer 3 Emergency Kill
+2. Post-call accounting（呼叫後）— 寫 budget_ledger + 更新 Redis cache
+3. 告警閾值通知（80% / 95% Telegram 告警）
+
+注意：Layer 0 Run budget 需要 awooop_run_state（Phase 3 SAGA 實作後補加）
+"""
+from __future__ import annotations
+
+import time
+from decimal import Decimal
+
+import structlog
+
+from src.core.config import settings
+
+logger = structlog.get_logger(__name__)
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 告警閾值（ADR-120 D4）
+# ─────────────────────────────────────────────────────────────────────────────
+BUDGET_ALERT_THRESHOLDS = {
+    "warn": Decimal("0.80"),
+    "critical": Decimal("0.95"),
+    "hard_kill": Decimal("1.00"),
+}
+
+# Redis key 前綴
+_EMERGENCY_KILL_KEY = "platform:budget:emergency_kill"
+_TENANT_BUDGET_KEY_PREFIX = "budget:tenant:"   # {project_id}:daily_used_usd
+_PLATFORM_BUDGET_KEY = "budget:platform:daily_used_usd"
+_BUDGET_CACHE_TTL = 300  # 5 分鐘，每次寫入後 refresh
+
+
+class BudgetExhaustedError(Exception):
+    """LLM call 被 hard kill 攔截"""
+
+    def __init__(self, error_code: str, message: str) -> None:
+        self.error_code = error_code
+        super().__init__(f"[{error_code}] {message}")
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 費用計算（按模型定價估算）
+# ─────────────────────────────────────────────────────────────────────────────
+
+# USD per 1M tokens（in + out）
+_COST_PER_MILLION_TOKENS: dict[str, tuple[float, float]] = {
+    # (prompt_per_M, completion_per_M)
+    "claude-opus-4-7":        (15.0, 75.0),
+    "claude-sonnet-4-6":      (3.0, 15.0),
+    "claude-haiku-4-5":       (0.8, 4.0),
+    "gpt-4o":                 (5.0, 15.0),
+    "gpt-4o-mini":            (0.15, 0.6),
+    "gemini-2.0-flash":       (0.075, 0.3),
+    "deepseek-r1:14b":        (0.0, 0.0),   # local Ollama — 無費用
+    "qwen3:8b":               (0.0, 0.0),   # local Ollama — 無費用
+}
+_DEFAULT_COST_PER_M = (3.0, 15.0)  # fallback → claude-sonnet
+
+
+def estimate_cost(
+    prompt_tokens: int,
+    completion_tokens: int,
+    model: str,
+) -> Decimal:
+    """估算一次 LLM call 的費用（USD）"""
+    prompt_rate, completion_rate = _COST_PER_MILLION_TOKENS.get(
+        model, _DEFAULT_COST_PER_M
+    )
+    cost = (prompt_tokens / 1_000_000 * prompt_rate +
+            completion_tokens / 1_000_000 * completion_rate)
+    return Decimal(str(round(cost, 6)))
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Pre-call Budget Check（ADR-120 D2 防線 1）
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def check_budget_before_llm_call(
+    project_id: str,
+    model: str,
+    estimated_prompt_tokens: int = 4000,
+    *,
+    agent_id: str | None = None,
+) -> None:
+    """
+    LLM call 前的三層 budget check。
+
+    超出任一層預算 → 拋出 BudgetExhaustedError，阻止 API call。
+    Redis 不可用時 fail-open（不阻擋呼叫，但記 warning）。
+
+    Args:
+        project_id: 租戶 ID
+        model: 模型名稱（用於費用估算）
+        estimated_prompt_tokens: 預估 prompt token 數（保守估計 × 1.5 已含在外）
+    """
+    # Layer 3：Emergency Kill Switch（最優先）
+    await check_emergency_kill()
+
+    # Local Ollama 模型無費用，跳過 Layer 1/2
+    if model in {"deepseek-r1:14b", "qwen3:8b"} or model.startswith("ollama/"):
+        return
+
+    estimated_cost = estimate_cost(estimated_prompt_tokens, 0, model)
+
+    # Layer 2：Tenant Budget
+    await _check_tenant_budget(project_id, estimated_cost)
+
+    # Layer 1：Platform Budget
+    await _check_platform_budget(estimated_cost)
+
+
+async def check_emergency_kill() -> None:
+    """Layer 3: Emergency Kill Switch — Redis key platform:budget:emergency_kill"""
+    try:
+        from src.core.redis_client import get_redis
+        redis = get_redis()
+        if await redis.exists(_EMERGENCY_KILL_KEY):
+            raise BudgetExhaustedError(
+                "E-BUDGET-004",
+                "Emergency kill switch activated — contact platform admin",
+            )
+    except BudgetExhaustedError:
+        raise
+    except Exception as exc:
+        logger.warning("budget_emergency_kill_check_failed", error=str(exc))
+
+
+async def _check_tenant_budget(project_id: str, estimated_cost: Decimal) -> None:
+    """Layer 2: Tenant Budget（Redis 快取 + awooop_projects.budget_limit_usd）"""
+    try:
+        from src.core.redis_client import get_redis
+        redis = get_redis()
+
+        # 讀取 Tenant 每日已用金額
+        cache_key = f"{_TENANT_BUDGET_KEY_PREFIX}{project_id}"
+        used_raw = await redis.get(cache_key)
+        used_usd = Decimal(used_raw.decode() if isinstance(used_raw, bytes) else used_raw or "0")
+
+        # 讀取 Tenant 預算上限（從 awooop_projects 表）
+        limit_usd = await _get_tenant_budget_limit(project_id)
+        if limit_usd is None:
+            return  # 無上限 → 放行
+
+        if used_usd + estimated_cost > limit_usd:
+            raise BudgetExhaustedError(
+                "E-BUDGET-002",
+                f"Tenant {project_id} budget exhausted: "
+                f"used ${used_usd:.4f} / ${limit_usd:.4f}",
+            )
+
+        # 告警閾值
+        usage_pct = (used_usd + estimated_cost) / limit_usd
+        if usage_pct >= BUDGET_ALERT_THRESHOLDS["critical"]:
+            logger.warning(
+                "budget_tenant_critical",
+                project_id=project_id,
+                usage_pct=float(usage_pct),
+                used_usd=float(used_usd),
+                limit_usd=float(limit_usd),
+            )
+        elif usage_pct >= BUDGET_ALERT_THRESHOLDS["warn"]:
+            logger.warning(
+                "budget_tenant_warn",
+                project_id=project_id,
+                usage_pct=float(usage_pct),
+                used_usd=float(used_usd),
+                limit_usd=float(limit_usd),
+            )
+
+    except BudgetExhaustedError:
+        raise
+    except Exception as exc:
+        logger.warning("budget_tenant_check_failed", project_id=project_id, error=str(exc))
+
+
+async def _check_platform_budget(estimated_cost: Decimal) -> None:
+    """Layer 1: Platform Budget（config 靜態上限 + Redis 累計）"""
+    platform_limit = getattr(settings, "PLATFORM_DAILY_BUDGET_USD", None)
+    if not platform_limit:
+        return  # 未設定 → 放行
+
+    try:
+        from src.core.redis_client import get_redis
+        redis = get_redis()
+        used_raw = await redis.get(_PLATFORM_BUDGET_KEY)
+        used_usd = Decimal(used_raw.decode() if isinstance(used_raw, bytes) else used_raw or "0")
+        limit_usd = Decimal(str(platform_limit))
+
+        if used_usd + estimated_cost > limit_usd:
+            raise BudgetExhaustedError(
+                "E-BUDGET-003",
+                f"Platform budget exhausted: used ${used_usd:.4f} / ${limit_usd:.4f} — "
+                "all LLM calls suspended",
+            )
+    except BudgetExhaustedError:
+        raise
+    except Exception as exc:
+        logger.warning("budget_platform_check_failed", error=str(exc))
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Post-call Accounting（ADR-120 D2 防線 2）
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def record_token_usage(
+    *,
+    project_id: str,
+    model: str,
+    provider: str,
+    prompt_tokens: int,
+    completion_tokens: int,
+    agent_id: str | None = None,
+    run_id: str | None = None,
+) -> Decimal:
+    """
+    LLM call 完成後記帳。
+
+    1. 計算實際費用
+    2. INSERT budget_ledger
+    3. 更新 Redis budget cache（async，不阻擋回傳）
+    4. 觸發告警閾值通知
+
+    Returns:
+        actual_cost_usd
+    """
+    import asyncio
+    from uuid import UUID
+
+    actual_cost = estimate_cost(prompt_tokens, completion_tokens, model)
+
+    # 寫入 budget_ledger（非阻擋）
+    asyncio.create_task(
+        _write_budget_ledger(
+            project_id=project_id,
+            agent_id=agent_id,
+            run_id=UUID(run_id) if run_id else None,
+            model=model,
+            provider=provider,
+            prompt_tokens=prompt_tokens,
+            completion_tokens=completion_tokens,
+            cost_usd=actual_cost,
+        ),
+        name="budget_ledger_write",
+    )
+
+    # 更新 Redis cache（非阻擋）
+    asyncio.create_task(
+        _update_budget_cache(project_id, actual_cost),
+        name="budget_cache_update",
+    )
+
+    logger.info(
+        "token_usage_recorded",
+        project_id=project_id,
+        model=model,
+        prompt_tokens=prompt_tokens,
+        completion_tokens=completion_tokens,
+        cost_usd=float(actual_cost),
+    )
+    return actual_cost
+
+
+async def _write_budget_ledger(
+    *,
+    project_id: str,
+    agent_id: str | None,
+    run_id,  # UUID | None
+    model: str,
+    provider: str,
+    prompt_tokens: int,
+    completion_tokens: int,
+    cost_usd: Decimal,
+) -> None:
+    """INSERT budget_ledger（leWOOOgo: DB 寫入在 Service 層，非 Router）"""
+    try:
+        from sqlalchemy import text
+        from src.db.base import get_db_context
+        async with get_db_context(project_id) as db:
+            await db.execute(
+                text("""
+                    INSERT INTO budget_ledger
+                        (project_id, agent_id, run_id, model, provider,
+                         prompt_tokens, completion_tokens, cost_usd)
+                    VALUES
+                        (:project_id, :agent_id, :run_id, :model, :provider,
+                         :prompt_tokens, :completion_tokens, :cost_usd)
+                """),
+                {
+                    "project_id": project_id,
+                    "agent_id": agent_id,
+                    "run_id": run_id,
+                    "model": model,
+                    "provider": provider,
+                    "prompt_tokens": prompt_tokens,
+                    "completion_tokens": completion_tokens,
+                    "cost_usd": cost_usd,
+                },
+            )
+    except Exception as exc:
+        logger.warning("budget_ledger_write_failed", project_id=project_id, error=str(exc))
+
+
+async def _update_budget_cache(project_id: str, cost: Decimal) -> None:
+    """用 Redis INCRBYFLOAT 更新 Tenant + Platform daily budget cache"""
+    try:
+        from src.core.redis_client import get_redis
+        redis = get_redis()
+        cost_f = float(cost)
+
+        # Tenant daily budget
+        tenant_key = f"{_TENANT_BUDGET_KEY_PREFIX}{project_id}"
+        await redis.incrbyfloat(tenant_key, cost_f)
+        await redis.expire(tenant_key, 86400)  # 24h TTL（每日重置）
+
+        # Platform daily budget
+        await redis.incrbyfloat(_PLATFORM_BUDGET_KEY, cost_f)
+        await redis.expire(_PLATFORM_BUDGET_KEY, 86400)
+
+    except Exception as exc:
+        logger.warning("budget_cache_update_failed", project_id=project_id, error=str(exc))
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Helper：從 DB 讀取 Tenant budget limit
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def _get_tenant_budget_limit(project_id: str) -> Decimal | None:
+    """從 awooop_projects.budget_limit_usd 讀取 Tenant 每日上限（允許 NULL = 無上限）"""
+    try:
+        from sqlalchemy import text
+        from src.db.base import get_db_context
+        async with get_db_context() as db:
+            row = await db.execute(
+                text("SELECT budget_limit_usd FROM awooop_projects WHERE project_id = :pid"),
+                {"pid": project_id},
+            )
+            result = row.scalar_one_or_none()
+            return Decimal(str(result)) if result is not None else None
+    except Exception as exc:
+        logger.warning("get_tenant_budget_limit_failed", project_id=project_id, error=str(exc))
+        return None
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Emergency Kill Switch 管理（Admin 工具）
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def activate_emergency_kill(reason: str = "") -> None:
+    """啟動緊急停機 — SET platform:budget:emergency_kill"""
+    from src.core.redis_client import get_redis
+    redis = get_redis()
+    await redis.set(_EMERGENCY_KILL_KEY, reason or "activated", ex=86400 * 7)
+    logger.warning("budget_emergency_kill_activated", reason=reason)
+
+
+async def deactivate_emergency_kill() -> None:
+    """解除緊急停機"""
+    from src.core.redis_client import get_redis
+    redis = get_redis()
+    await redis.delete(_EMERGENCY_KILL_KEY)
+    logger.info("budget_emergency_kill_deactivated")
+
+
+async def is_emergency_kill_active() -> bool:
+    """查詢緊急停機狀態"""
+    try:
+        from src.core.redis_client import get_redis
+        redis = get_redis()
+        return bool(await redis.exists(_EMERGENCY_KILL_KEY))
+    except Exception:
+        return False

apps/api/src/services/channel_hub.py

@@ -0,0 +1,418 @@
+"""
+Channel Hub — AwoooP 入站事件統一路由 + Progressive Feedback Policy
+====================================================================
+AwoooP Phase 7: ADR-106（channel_event family）
+2026-05-04 ogt + Claude Sonnet 4.6
+
+功能：
+  1. Telegram 入站事件鏡像（記錄到 awooop_conversation_event）
+  2. 建立 platform run（呼叫 platform_runtime.create_run）
+  3. Progressive Feedback Policy：
+     - run 進入 WAITING_TOOL 狀態 → 30 秒後若未 complete → 發 interim Telegram 訊息
+     - 訊息記錄到 awooop_outbound_message
+  4. Shadow Mode：不發任何 Telegram 訊息（只記錄到 outbound_message, status='shadow'）
+
+Progressive Feedback Policy 設計（ADR-106 P2-03）：
+  - 用 asyncio.create_task 啟動 30s 計時器
+  - 30s 後查詢 run state：若仍在 WAITING_TOOL → 發 interim 訊息
+  - interim 訊息：「AI 正在分析中，請稍候...」（不洩漏 run 細節）
+  - Final reply 由 shadow_execute() 完成後觸發（Phase 8 實作）
+
+與 legacy telegram_gateway.py 的關係：
+  - 完全獨立，不修改 legacy gateway
+  - legacy 繼續處理 legacy flow（signal_worker 觸發的 approval/notification）
+  - AwoooP run 只走本模組
+"""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import json
+from datetime import datetime, timezone
+from typing import Any
+from uuid import UUID
+
+import structlog
+from sqlalchemy import select, text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.db.awooop_models import AwoooPRunState
+from src.services.audit_sink import _redact_string
+from src.services.platform_runtime import create_run
+
+logger = structlog.get_logger(__name__)
+
+# Progressive Feedback Policy：等待超過此秒數才發 interim 訊息
+_INTERIM_WAIT_SECONDS = 30
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 入站事件記錄
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def mirror_inbound_event(
+    db: AsyncSession,
+    *,
+    project_id: str,
+    channel_type: str,
+    provider_event_id: str,
+    platform_subject_id: str | None = None,
+    channel_user_id: str | None = None,
+    channel_chat_id: str | None = None,
+    content_type: str = "text",
+    raw_content: str | None = None,
+    attachment_sha256: str | None = None,
+    provider_ts: datetime | None = None,
+    run_id: UUID | None = None,
+    is_duplicate: bool = False,
+) -> UUID:
+    """
+    記錄入站 channel event 到 awooop_conversation_event。
+
+    raw_content 只用於計算 hash 和 preview，不入庫明文。
+    回傳 event_id。
+    """
+    content_hash: str | None = None
+    content_preview: str | None = None
+
+    if raw_content is not None:
+        content_hash = hashlib.sha256(raw_content.encode()).hexdigest()
+        # preview：redact 後截取前 256 字元
+        redacted = _redact_string(raw_content)
+        content_preview = redacted[:256] if len(redacted) > 256 else redacted
+
+    result = await db.execute(
+        text("""
+            INSERT INTO awooop_conversation_event (
+                project_id, channel_type, provider_event_id,
+                platform_subject_id, channel_user_id, channel_chat_id,
+                run_id, content_type, content_hash, content_preview,
+                attachment_sha256, is_duplicate, provider_ts, received_at
+            ) VALUES (
+                :project_id, :channel_type, :provider_event_id,
+                :platform_subject_id, :channel_user_id, :channel_chat_id,
+                :run_id, :content_type, :content_hash, :content_preview,
+                :attachment_sha256, :is_duplicate, :provider_ts, NOW()
+            )
+            ON CONFLICT (project_id, channel_type, provider_event_id) DO UPDATE SET
+                is_duplicate = TRUE,
+                run_id = COALESCE(EXCLUDED.run_id, awooop_conversation_event.run_id)
+            RETURNING event_id
+        """),
+        {
+            "project_id": project_id,
+            "channel_type": channel_type,
+            "provider_event_id": provider_event_id,
+            "platform_subject_id": platform_subject_id,
+            "channel_user_id": channel_user_id,
+            "channel_chat_id": channel_chat_id,
+            "run_id": run_id,
+            "content_type": content_type,
+            "content_hash": content_hash,
+            "content_preview": content_preview,
+            "attachment_sha256": attachment_sha256,
+            "is_duplicate": is_duplicate,
+            "provider_ts": provider_ts,
+        },
+    )
+    row = result.fetchone()
+    event_id: UUID = row[0]
+    logger.info(
+        "channel_event_mirrored",
+        project_id=project_id,
+        channel_type=channel_type,
+        event_id=str(event_id),
+        is_duplicate=is_duplicate,
+    )
+    return event_id
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 出站訊息記錄
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def record_outbound_message(
+    db: AsyncSession,
+    *,
+    project_id: str,
+    run_id: UUID,
+    channel_type: str,
+    channel_chat_id: str,
+    message_type: str,           # 'interim' | 'final' | 'error' | 'approval_request'
+    content: str | None = None,
+    provider_message_id: str | None = None,
+    send_status: str = "pending",
+    conversation_event_id: UUID | None = None,
+    triggered_by_state: str | None = None,
+    waiting_since: datetime | None = None,
+    is_shadow: bool = True,
+) -> UUID:
+    """
+    記錄出站訊息到 awooop_outbound_message。
+
+    is_shadow=True：status='shadow'（不實際發送，只記錄）
+    """
+    content_hash: str | None = None
+    content_preview: str | None = None
+    if content is not None:
+        content_hash = hashlib.sha256(content.encode()).hexdigest()
+        redacted = _redact_string(content)
+        content_preview = redacted[:256]
+
+    actual_status = "shadow" if is_shadow else send_status
+
+    result = await db.execute(
+        text("""
+            INSERT INTO awooop_outbound_message (
+                project_id, run_id, conversation_event_id,
+                channel_type, channel_chat_id, message_type,
+                content_hash, content_preview, provider_message_id,
+                send_status, queued_at,
+                triggered_by_state, waiting_since
+            ) VALUES (
+                :project_id, :run_id, :conversation_event_id,
+                :channel_type, :channel_chat_id, :message_type,
+                :content_hash, :content_preview, :provider_message_id,
+                :send_status, NOW(),
+                :triggered_by_state, :waiting_since
+            )
+            RETURNING message_id
+        """),
+        {
+            "project_id": project_id,
+            "run_id": run_id,
+            "conversation_event_id": conversation_event_id,
+            "channel_type": channel_type,
+            "channel_chat_id": channel_chat_id,
+            "message_type": message_type,
+            "content_hash": content_hash,
+            "content_preview": content_preview,
+            "provider_message_id": provider_message_id,
+            "send_status": actual_status,
+            "triggered_by_state": triggered_by_state,
+            "waiting_since": waiting_since,
+        },
+    )
+    row = result.fetchone()
+    message_id: UUID = row[0]
+    logger.info(
+        "outbound_message_recorded",
+        project_id=project_id,
+        run_id=str(run_id),
+        message_type=message_type,
+        send_status=actual_status,
+        message_id=str(message_id),
+    )
+    return message_id
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Progressive Feedback Policy
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def schedule_interim_feedback(
+    *,
+    project_id: str,
+    run_id: UUID,
+    channel_type: str,
+    channel_chat_id: str,
+    conversation_event_id: UUID | None = None,
+    is_shadow: bool = True,
+    wait_seconds: int = _INTERIM_WAIT_SECONDS,
+) -> None:
+    """
+    Progressive Feedback Policy：
+    等待 wait_seconds 秒後，若 run 仍在 WAITING_TOOL → 發 interim 訊息。
+
+    Shadow Mode：記錄到 outbound_message(status='shadow')，不實際發 Telegram 訊息。
+    """
+    asyncio.create_task(
+        _interim_feedback_task(
+            project_id=project_id,
+            run_id=run_id,
+            channel_type=channel_type,
+            channel_chat_id=channel_chat_id,
+            conversation_event_id=conversation_event_id,
+            is_shadow=is_shadow,
+            wait_seconds=wait_seconds,
+        ),
+        name=f"interim_feedback_{str(run_id)[:8]}",
+    )
+
+
+async def _interim_feedback_task(
+    *,
+    project_id: str,
+    run_id: UUID,
+    channel_type: str,
+    channel_chat_id: str,
+    conversation_event_id: UUID | None,
+    is_shadow: bool,
+    wait_seconds: int,
+) -> None:
+    """等待後查 run state，仍 waiting_tool 才發 interim"""
+    await asyncio.sleep(wait_seconds)
+
+    try:
+        from src.db.base import get_db_context
+
+        async with get_db_context(project_id) as db:
+            result = await db.execute(
+                select(AwoooPRunState.state, AwoooPRunState.is_shadow).where(
+                    AwoooPRunState.run_id == run_id,
+                    AwoooPRunState.project_id == project_id,
+                )
+            )
+            row = result.first()
+
+            if row is None:
+                logger.warning(
+                    "interim_feedback_run_not_found",
+                    run_id=str(run_id),
+                )
+                return
+
+            state, run_is_shadow = row
+            if state != "waiting_tool":
+                # run 已推進（complete/failed 等），不需要 interim
+                return
+
+            waiting_since = datetime.now(timezone.utc)
+            interim_content = "AI 正在分析中，請稍候... ⏳"
+
+            await record_outbound_message(
+                db,
+                project_id=project_id,
+                run_id=run_id,
+                channel_type=channel_type,
+                channel_chat_id=channel_chat_id,
+                message_type="interim",
+                content=interim_content,
+                send_status="pending",
+                conversation_event_id=conversation_event_id,
+                triggered_by_state="waiting_tool",
+                waiting_since=waiting_since,
+                is_shadow=is_shadow or run_is_shadow,
+            )
+
+            if not (is_shadow or run_is_shadow):
+                # Non-shadow：實際發 Telegram 訊息
+                await _send_telegram_interim(
+                    channel_chat_id=channel_chat_id,
+                    content=interim_content,
+                    run_id=run_id,
+                )
+
+            logger.info(
+                "interim_feedback_sent",
+                project_id=project_id,
+                run_id=str(run_id),
+                is_shadow=is_shadow or run_is_shadow,
+            )
+
+    except Exception as exc:
+        logger.exception(
+            "interim_feedback_task_error",
+            run_id=str(run_id),
+            error=str(exc),
+        )
+
+
+async def _send_telegram_interim(
+    *,
+    channel_chat_id: str,
+    content: str,
+    run_id: UUID,
+) -> None:
+    """實際發送 Telegram interim 訊息（non-shadow 專用）"""
+    try:
+        import os
+
+        import httpx
+
+        bot_token = os.environ.get("TELEGRAM_BOT_TOKEN")
+        if not bot_token:
+            logger.warning("interim_telegram_no_token", run_id=str(run_id))
+            return
+
+        async with httpx.AsyncClient(timeout=10) as client:
+            await client.post(
+                f"https://api.telegram.org/bot{bot_token}/sendMessage",
+                json={
+                    "chat_id": channel_chat_id,
+                    "text": content,
+                    "parse_mode": "HTML",
+                },
+            )
+    except Exception as exc:
+        logger.warning(
+            "interim_telegram_send_failed",
+            run_id=str(run_id),
+            error=str(exc),
+        )
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Channel Hub 主入口（Telegram inbound）
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def handle_telegram_inbound(
+    db: AsyncSession,
+    *,
+    project_id: str,
+    agent_id: str,
+    message_id: str,
+    user_id: str,
+    chat_id: str,
+    text: str | None = None,
+    is_shadow: bool = True,
+) -> dict[str, Any]:
+    """
+    Telegram 入站訊息的統一處理入口：
+    1. mirror_inbound_event（記錄）
+    2. create_run（建立 platform run）
+    3. schedule_interim_feedback（Progressive Feedback）
+    4. 回傳 {event_id, run_id, is_duplicate}
+    """
+    # Step 1: 嘗試建立 run（有冪等保護）
+    run_id, is_duplicate = await create_run(
+        project_id=project_id,
+        agent_id=agent_id,
+        trigger_type="channel_event",
+        trigger_ref=f"telegram:{message_id}",
+        input_payload={"chat_id": chat_id, "user_id": user_id},
+        channel_type="telegram",
+        provider_event_id=message_id,
+    )
+
+    # Step 2: Mirror event（含 run_id）
+    event_id = await mirror_inbound_event(
+        db,
+        project_id=project_id,
+        channel_type="telegram",
+        provider_event_id=message_id,
+        channel_user_id=user_id,
+        channel_chat_id=chat_id,
+        content_type="text" if text else "callback_query",
+        raw_content=text,
+        run_id=run_id,
+        is_duplicate=is_duplicate,
+    )
+
+    # Step 3: Progressive Feedback（30s 計時器）
+    if not is_duplicate:
+        await schedule_interim_feedback(
+            project_id=project_id,
+            run_id=run_id,
+            channel_type="telegram",
+            channel_chat_id=chat_id,
+            conversation_event_id=event_id,
+            is_shadow=is_shadow,
+        )
+
+    return {
+        "event_id": str(event_id),
+        "run_id": str(run_id),
+        "is_duplicate": is_duplicate,
+    }

apps/api/src/services/contract_service.py

@@ -0,0 +1,449 @@
+"""
+Contract Lifecycle Service
+===========================
+AwoooP Phase 3: 合約生命週期管理（ADR-107/ADR-112）
+2026-05-04 ogt + Claude Sonnet 4.6
+
+生命週期狀態機：
+  draft → published → active → revoked
+  ↑                           ↓（新 active 把舊的設為 revoked）
+
+操作：
+  draft()      — 建立 draft revision（schema 驗證 + body_hash）
+  publish()    — HMAC 簽章驗證後 draft → published
+  activate()   — approval 確認後 published → active + outbox
+  get_active() — runtime 唯一讀取路徑（只返回 active revision）
+
+安全機制：
+  - body_hash = sha256(canonical JSON)（ADR-112）
+  - publish() 需 HMAC 簽章（settings.CONTRACT_HMAC_KEY）
+  - activate() 需 Redis multi_sig 確認（ADR-112 approval workflow）
+  - 所有操作寫入 audit_log
+"""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+from datetime import datetime, timezone
+from typing import Any
+from uuid import UUID
+
+import structlog
+from pydantic import ValidationError
+
+from src.core.config import settings
+from src.db.awooop_models import AwoooPContractRevision
+from src.models.awooop_contracts import validate_contract_body
+from src.repositories import contract_repository
+
+logger = structlog.get_logger(__name__)
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 錯誤定義
+# ─────────────────────────────────────────────────────────────────────────────
+
+class ContractError(Exception):
+    """合約操作基礎錯誤"""
+    def __init__(self, error_code: str, message: str) -> None:
+        self.error_code = error_code
+        super().__init__(f"[{error_code}] {message}")
+
+
+class ContractSchemaError(ContractError):
+    """body_json 不符合 schema"""
+    def __init__(self, family: str, details: str) -> None:
+        super().__init__("E-CONTRACT-001", f"Contract family={family} schema 驗證失敗: {details}")
+
+
+class ContractSignatureError(ContractError):
+    """HMAC 簽章驗證失敗"""
+    def __init__(self) -> None:
+        super().__init__("E-CONTRACT-002", "Contract publish 簽章驗證失敗")
+
+
+class ContractStateError(ContractError):
+    """非法狀態轉換"""
+    def __init__(self, from_state: str, to_state: str) -> None:
+        super().__init__(
+            "E-CONTRACT-003",
+            f"非法狀態轉換 {from_state!r} → {to_state!r}",
+        )
+
+
+class ContractApprovalError(ContractError):
+    """缺少必要的 activation approval"""
+    def __init__(self, revision_id: str) -> None:
+        super().__init__(
+            "E-CONTRACT-004",
+            f"revision {revision_id} 尚未取得足夠的 approval 簽核",
+        )
+
+
+class ContractNotFoundError(ContractError):
+    """Revision 不存在"""
+    def __init__(self, revision_id: str) -> None:
+        super().__init__(
+            "E-CONTRACT-005",
+            f"Revision {revision_id!r} 不存在或無權限存取",
+        )
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Body hash（ADR-112 artifact integrity）
+# ─────────────────────────────────────────────────────────────────────────────
+
+def _compute_body_hash(body_json: dict[str, Any]) -> str:
+    """
+    計算 body_json 的 SHA-256 hex digest。
+    使用 canonical JSON（sorted keys, no spaces）確保確定性。
+    """
+    canonical = json.dumps(body_json, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
+
+
+def _verify_publish_signature(
+    revision_id: str,
+    body_hash: str,
+    publisher_id: str,
+    signature: str,
+) -> bool:
+    """
+    驗證 publish HMAC 簽章。
+    message = f"{revision_id}:{body_hash}:{publisher_id}"
+    secret  = settings.CONTRACT_HMAC_KEY（base64 or hex）
+    """
+    secret = getattr(settings, "CONTRACT_HMAC_KEY", "")
+    if not secret:
+        # 未設定 HMAC key → 開發環境放行（但記錄 warning）
+        logger.warning(
+            "contract_hmac_key_not_set",
+            warning="CONTRACT_HMAC_KEY 未設定，publish 簽章驗證跳過（非 production 行為）",
+        )
+        return True
+
+    message = f"{revision_id}:{body_hash}:{publisher_id}".encode("utf-8")
+    expected = hmac.new(
+        secret.encode("utf-8"), message, hashlib.sha256
+    ).hexdigest()
+    return hmac.compare_digest(expected, signature)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Multi-sig approval（ADR-112 activation approval）
+# ─────────────────────────────────────────────────────────────────────────────
+
+_APPROVAL_KEY_PREFIX = "contract:approval:"
+_APPROVAL_REQUIRED = 1  # Phase 3：1 人核准即可；Phase 5+ 升為 2
+
+
+async def _check_activation_approval(revision_id: str, project_id: str) -> bool:
+    """
+    檢查 Redis 中是否有足夠的 activation approval。
+    key = contract:approval:{project_id}:{revision_id}
+    value = JSON list of approver IDs
+    """
+    try:
+        from src.core.redis_client import get_redis
+        redis = get_redis()
+        key = f"{_APPROVAL_KEY_PREFIX}{project_id}:{revision_id}"
+        raw = await redis.get(key)
+        if not raw:
+            return False
+        approvers = json.loads(raw.decode() if isinstance(raw, bytes) else raw)
+        return len(approvers) >= _APPROVAL_REQUIRED
+    except Exception as exc:
+        logger.warning("contract_approval_check_failed", revision_id=revision_id, error=str(exc))
+        return False
+
+
+async def record_activation_approval(
+    revision_id: str,
+    project_id: str,
+    approver_id: str,
+) -> int:
+    """
+    記錄一個 approver 的核准簽名。
+    Returns: 目前收到的 approval 數。
+    """
+    from src.core.redis_client import get_redis
+    redis = get_redis()
+    key = f"{_APPROVAL_KEY_PREFIX}{project_id}:{revision_id}"
+    raw = await redis.get(key)
+    approvers: list[str] = json.loads(raw.decode() if isinstance(raw, bytes) else raw or "[]")
+    if approver_id not in approvers:
+        approvers.append(approver_id)
+    await redis.set(key, json.dumps(approvers), ex=86400)  # 24h TTL
+    logger.info(
+        "contract_approval_recorded",
+        revision_id=revision_id,
+        approver_id=approver_id,
+        total_approvals=len(approvers),
+    )
+    return len(approvers)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Core lifecycle operations
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def draft(
+    *,
+    project_id: str,
+    contract_family: str,
+    contract_id: str,
+    version_major: int,
+    version_minor: int,
+    body_json: dict[str, Any],
+    body_schema_version: str = "v1.0",
+) -> AwoooPContractRevision:
+    """
+    Step 1: 建立 draft revision。
+
+    - 驗證 body_json 符合 contract_family 的 Pydantic schema
+    - 計算 body_hash（sha256 canonical JSON）
+    - 寫入 DB（lifecycle_status='draft'）
+    - 寫入 audit log
+
+    draft revision 不可被 runtime 讀取（get_active() 只返回 active）。
+    """
+    # Schema 驗證
+    try:
+        validate_contract_body(contract_family, body_json)
+    except ValidationError as exc:
+        raise ContractSchemaError(contract_family, exc.json(indent=0)) from exc
+    except ValueError as exc:
+        raise ContractSchemaError(contract_family, str(exc)) from exc
+
+    body_hash = _compute_body_hash(body_json)
+
+    revision = await contract_repository.create_draft(
+        project_id=project_id,
+        contract_family=contract_family,
+        contract_id=contract_id,
+        version_major=version_major,
+        version_minor=version_minor,
+        body_json=body_json,
+        body_hash=body_hash,
+        body_schema_version=body_schema_version,
+    )
+
+    await _write_audit(
+        project_id=project_id,
+        action="contract.drafted",
+        resource_type="contract_revision",
+        resource_id=str(revision.revision_id),
+        details={
+            "contract_family": contract_family,
+            "contract_id": contract_id,
+            "version": f"{version_major}.{version_minor}",
+            "body_hash": body_hash,
+        },
+    )
+    return revision
+
+
+async def publish(
+    *,
+    revision_id: UUID,
+    project_id: str,
+    publisher_id: str,
+    signature: str,
+) -> AwoooPContractRevision:
+    """
+    Step 2: draft → published。
+
+    - 讀取 revision（驗證 lifecycle_status='draft'）
+    - HMAC 簽章驗證（publisher_id + body_hash + revision_id）
+    - 更新 lifecycle_status='published'
+    - 寫入 audit log
+    """
+    revision = await contract_repository.get_revision(revision_id, project_id)
+    if revision is None:
+        raise ContractNotFoundError(str(revision_id))
+    if revision.lifecycle_status != "draft":
+        raise ContractStateError(revision.lifecycle_status, "published")
+
+    if not _verify_publish_signature(
+        str(revision_id), revision.body_hash, publisher_id, signature
+    ):
+        raise ContractSignatureError()
+
+    published_at = datetime.now(timezone.utc)
+    revision = await contract_repository.mark_published(
+        revision_id=revision_id,
+        project_id=project_id,
+        publisher_id=publisher_id,
+        publish_signature=signature,
+        published_at=published_at,
+    )
+
+    await _write_audit(
+        project_id=project_id,
+        action="contract.published",
+        resource_type="contract_revision",
+        resource_id=str(revision_id),
+        details={
+            "publisher_id": publisher_id,
+            "published_at": published_at.isoformat(),
+            "body_hash": revision.body_hash,
+        },
+    )
+    return revision
+
+
+async def activate(
+    *,
+    revision_id: UUID,
+    project_id: str,
+    activator_id: str,
+    bypass_approval: bool = False,
+) -> AwoooPContractRevision:
+    """
+    Step 3: published → active。
+
+    - 讀取 revision（驗證 lifecycle_status='published'）
+    - 確認 Redis approval（除非 bypass_approval=True）
+    - 更新 active pointer（UPSERT awooop_active_revisions）
+    - 舊 active revision → revoked
+    - 寫入 outbox event（ADR-113）
+    - 寫入 audit log
+    """
+    revision = await contract_repository.get_revision(revision_id, project_id)
+    if revision is None:
+        raise ContractNotFoundError(str(revision_id))
+    if revision.lifecycle_status != "published":
+        raise ContractStateError(revision.lifecycle_status, "active")
+
+    if not bypass_approval:
+        approved = await _check_activation_approval(str(revision_id), project_id)
+        if not approved:
+            raise ContractApprovalError(str(revision_id))
+
+    # 找舊 active revision（如果有）
+    old_revision = await contract_repository.get_active_revision(
+        project_id=project_id,
+        contract_family=revision.contract_family,
+        contract_id=revision.contract_id,
+    )
+    old_revision_id = old_revision.revision_id if old_revision else None
+
+    revision = await contract_repository.mark_active(
+        revision_id=revision_id,
+        project_id=project_id,
+        contract_family=revision.contract_family,
+        contract_id=revision.contract_id,
+        old_revision_id=old_revision_id,
+    )
+
+    await _write_audit(
+        project_id=project_id,
+        action="contract.activated",
+        resource_type="contract_revision",
+        resource_id=str(revision_id),
+        details={
+            "activator_id": activator_id,
+            "old_revision_id": str(old_revision_id) if old_revision_id else None,
+            "contract_family": revision.contract_family,
+            "contract_id": revision.contract_id,
+        },
+    )
+    return revision
+
+
+async def get_active(
+    *,
+    project_id: str,
+    contract_family: str,
+    contract_id: str,
+    verify_hash: bool = True,
+) -> AwoooPContractRevision | None:
+    """
+    Runtime 讀取路徑：只返回 active revision。
+
+    verify_hash=True（預設）：從 DB 讀取後驗證 body_hash，
+    確保 body_json 未被竄改（ADR-112 artifact integrity）。
+    """
+    revision = await contract_repository.get_active_revision(
+        project_id=project_id,
+        contract_family=contract_family,
+        contract_id=contract_id,
+    )
+    if revision is None:
+        return None
+
+    if verify_hash:
+        computed = _compute_body_hash(revision.body_json)
+        if computed != revision.body_hash:
+            logger.error(
+                "contract_hash_mismatch",
+                revision_id=str(revision.revision_id),
+                expected=revision.body_hash,
+                computed=computed,
+            )
+            raise ContractError(
+                "E-CONTRACT-006",
+                f"revision {revision.revision_id} body_hash 不符（資料可能被竄改）",
+            )
+
+    return revision
+
+
+async def get_active_body(
+    *,
+    project_id: str,
+    contract_family: str,
+    contract_id: str,
+) -> dict[str, Any] | None:
+    """
+    便利方法：直接返回 body_json（含 hash 驗證）。
+    None = 沒有 active revision。
+    """
+    revision = await get_active(
+        project_id=project_id,
+        contract_family=contract_family,
+        contract_id=contract_id,
+    )
+    return revision.body_json if revision else None
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Audit log helper
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def _write_audit(
+    *,
+    project_id: str,
+    action: str,
+    resource_type: str,
+    resource_id: str,
+    details: dict[str, Any],
+) -> None:
+    """寫入 audit_log（非阻擋，失敗只 warning）"""
+    try:
+        from sqlalchemy import text as sa_text
+        from src.db.base import get_db_context
+        async with get_db_context(project_id) as db:
+            await db.execute(
+                sa_text("""
+                    INSERT INTO audit_logs
+                        (project_id, action, resource_type, resource_id, details)
+                    VALUES
+                        (:project_id, :action, :resource_type, :resource_id, :details::jsonb)
+                """),
+                {
+                    "project_id": project_id,
+                    "action": action,
+                    "resource_type": resource_type,
+                    "resource_id": resource_id,
+                    "details": json.dumps(details),
+                },
+            )
+    except Exception as exc:
+        logger.warning(
+            "contract_audit_write_failed",
+            action=action,
+            resource_id=resource_id,
+            error=str(exc),
+        )

apps/api/src/services/platform_runtime.py

@@ -0,0 +1,375 @@
+"""
+Platform Runtime（Shadow Mode Shell）
+======================================
+AwoooP Phase 4: 第一個 runtime shell，只跑 shadow，不改 legacy 行為（ADR-106）
+2026-05-04 ogt + Claude Sonnet 4.6
+
+Shadow Mode 保證：
+  1. 0 user-visible response（不發送 Telegram/Slack 任何訊息）
+  2. 0 destructive tool call（is_destructive=true 的工具全部攔截）
+  3. 所有執行記錄寫入 awooop_run_state + step_journal（可觀測）
+  4. budget_service hard kill 同樣作用（防止 shadow 跑出超額費用）
+
+Idempotency（ADR-114）：
+  (project_id, channel_type, provider_event_id) 複合唯一
+  Redis NX 先攔（快），PG constraint 最後防（準確）
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import uuid
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+import structlog
+from sqlalchemy import select
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+
+from src.db.awooop_models import AwoooPRunIdempotency, AwoooPRunState, AwoooPRunStepJournal
+from src.db.base import get_db_context
+from src.services.run_state_machine import LEASE_TTL_SECONDS, transition
+
+logger = structlog.get_logger(__name__)
+
+# Shadow mode 設定
+_SHADOW_ENABLED = True  # Phase 4 固定 True；Phase 6+ 由 migration_mode 控制
+_DESTRUCTIVE_TOOL_KEYWORDS = frozenset({
+    "delete", "drop", "remove", "kill", "terminate", "destroy",
+    "rollback", "revert", "patch", "apply", "exec", "execute",
+    "restart", "scale", "cordon", "drain",
+})
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# UUID v7（時間有序）
+# ─────────────────────────────────────────────────────────────────────────────
+
+def _uuid7() -> uuid.UUID:
+    """
+    生成 UUID v7（時間有序，適合資料庫 PK）。
+    格式：48-bit Unix timestamp ms + version(7) + 74-bit random
+    """
+    now_ms = int(datetime.now(timezone.utc).timestamp() * 1000)
+    rand_bits = int.from_bytes(uuid.uuid4().bytes[6:], "big") & 0x3FFFFFFFFFFFFFFF
+    val = (now_ms << 80) | (0x7 << 76) | (0x8 << 72) | rand_bits
+    return uuid.UUID(int=val)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# W3C traceparent 生成
+# ─────────────────────────────────────────────────────────────────────────────
+
+def _new_trace_id() -> str:
+    """生成 W3C traceparent-compatible trace_id"""
+    trace_id_hex = uuid.uuid4().hex + uuid.uuid4().hex[:16]  # 128-bit
+    span_id_hex = uuid.uuid4().hex[:16]                       # 64-bit
+    return f"00-{trace_id_hex}-{span_id_hex}-01"
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Idempotency
+# ─────────────────────────────────────────────────────────────────────────────
+
+_IDEMPOTENCY_REDIS_PREFIX = "awooop:run:idem:"
+_IDEMPOTENCY_REDIS_TTL = 86400  # 24h
+
+
+async def check_idempotency(
+    project_id: str,
+    channel_type: str,
+    provider_event_id: str,
+) -> uuid.UUID | None:
+    """
+    檢查 (project_id, channel_type, provider_event_id) 是否已有對應 run_id。
+
+    Layer 1：Redis NX（快速攔截，TTL 24h）
+    Layer 2：PostgreSQL awooop_run_idempotency（準確）
+
+    Returns: 既有 run_id，或 None（尚未處理）
+    """
+    idem_key = f"{_IDEMPOTENCY_REDIS_PREFIX}{project_id}:{channel_type}:{provider_event_id}"
+
+    # Layer 1: Redis
+    try:
+        from src.core.redis_client import get_redis
+        redis = get_redis()
+        cached = await redis.get(idem_key)
+        if cached:
+            run_id_str = cached.decode() if isinstance(cached, bytes) else cached
+            logger.info(
+                "idempotency_hit_redis",
+                project_id=project_id,
+                provider_event_id=provider_event_id,
+                run_id=run_id_str,
+            )
+            return uuid.UUID(run_id_str)
+    except Exception as exc:
+        logger.warning("idempotency_redis_check_failed", error=str(exc))
+
+    # Layer 2: PostgreSQL
+    try:
+        async with get_db_context(project_id) as db:
+            result = await db.execute(
+                select(AwoooPRunIdempotency.run_id).where(
+                    AwoooPRunIdempotency.project_id == project_id,
+                    AwoooPRunIdempotency.channel_type == channel_type,
+                    AwoooPRunIdempotency.provider_event_id == provider_event_id,
+                )
+            )
+            row = result.fetchone()
+            if row:
+                return uuid.UUID(str(row[0]))
+    except Exception as exc:
+        logger.warning("idempotency_pg_check_failed", error=str(exc))
+
+    return None
+
+
+async def _register_idempotency(
+    project_id: str,
+    channel_type: str,
+    provider_event_id: str,
+    run_id: uuid.UUID,
+) -> None:
+    """寫入 idempotency 記錄（Redis + PostgreSQL）"""
+    idem_key = f"{_IDEMPOTENCY_REDIS_PREFIX}{project_id}:{channel_type}:{provider_event_id}"
+    run_id_str = str(run_id)
+
+    # Redis NX（若已有其他 worker 寫入，NX 失敗，無害）
+    try:
+        from src.core.redis_client import get_redis
+        redis = get_redis()
+        await redis.set(idem_key, run_id_str, ex=_IDEMPOTENCY_REDIS_TTL, nx=True)
+    except Exception as exc:
+        logger.warning("idempotency_redis_write_failed", error=str(exc))
+
+    # PostgreSQL（INSERT OR IGNORE）
+    try:
+        async with get_db_context(project_id) as db:
+            stmt = pg_insert(AwoooPRunIdempotency).values(
+                project_id=project_id,
+                channel_type=channel_type,
+                provider_event_id=provider_event_id,
+                run_id=run_id,
+            ).on_conflict_do_nothing(constraint="uix_run_idempotency_key")
+            await db.execute(stmt)
+    except Exception as exc:
+        logger.warning("idempotency_pg_write_failed", error=str(exc))
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Shadow destructive tool check
+# ─────────────────────────────────────────────────────────────────────────────
+
+def is_destructive_tool(tool_name: str, is_destructive_flag: bool = False) -> bool:
+    """
+    判斷 tool call 是否為破壞性操作。
+    Shadow mode 下一律攔截。
+
+    判斷邏輯：
+    1. MCP Gateway contract 的 is_destructive=True flag
+    2. tool_name 包含破壞性關鍵字（fallback，無 contract 時使用）
+    """
+    if is_destructive_flag:
+        return True
+    tool_lower = tool_name.lower()
+    return any(kw in tool_lower for kw in _DESTRUCTIVE_TOOL_KEYWORDS)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Run 建立
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def create_run(
+    *,
+    project_id: str,
+    agent_id: str,
+    trigger_type: str,
+    trigger_ref: str | None = None,
+    input_payload: dict[str, Any] | None = None,
+    channel_type: str | None = None,
+    provider_event_id: str | None = None,
+    timeout_seconds: int = 600,
+) -> tuple[uuid.UUID, bool]:
+    """
+    建立新 run（或返回既有 run，若重複事件）。
+
+    Returns:
+        (run_id, is_duplicate) — is_duplicate=True 表示冪等命中
+
+    Shadow mode：is_shadow=True，不產生 user response。
+    """
+    # Idempotency 檢查
+    if channel_type and provider_event_id:
+        existing_run_id = await check_idempotency(project_id, channel_type, provider_event_id)
+        if existing_run_id:
+            logger.info(
+                "run_creation_idempotent",
+                project_id=project_id,
+                channel_type=channel_type,
+                provider_event_id=provider_event_id,
+                existing_run_id=str(existing_run_id),
+            )
+            return existing_run_id, True
+
+    run_id = _uuid7()
+    trace_id = _new_trace_id()
+    timeout_at = datetime.now(timezone.utc) + timedelta(seconds=timeout_seconds)
+
+    # 計算 input_sha256
+    input_sha256 = None
+    if input_payload:
+        canonical = json.dumps(input_payload, sort_keys=True, separators=(",", ":"))
+        input_sha256 = hashlib.sha256(canonical.encode()).hexdigest()
+
+    async with get_db_context(project_id) as db:
+        run = AwoooPRunState(
+            run_id=run_id,
+            project_id=project_id,
+            agent_id=agent_id,
+            state="pending",
+            trace_id=trace_id,
+            trigger_type=trigger_type,
+            trigger_ref=trigger_ref,
+            is_shadow=_SHADOW_ENABLED,
+            input_sha256=input_sha256,
+            timeout_at=timeout_at,
+        )
+        db.add(run)
+
+    # 寫入 idempotency 記錄
+    if channel_type and provider_event_id:
+        await _register_idempotency(project_id, channel_type, provider_event_id, run_id)
+
+    logger.info(
+        "run_created",
+        run_id=str(run_id),
+        project_id=project_id,
+        agent_id=agent_id,
+        is_shadow=_SHADOW_ENABLED,
+        trace_id=trace_id,
+        trigger_type=trigger_type,
+    )
+    return run_id, False
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Shadow Execution（Phase 4 主邏輯）
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def shadow_execute(run: AwoooPRunState) -> None:
+    """
+    Shadow mode 執行一個 run。
+
+    Phase 4 行為：
+    - 解析 agent contract（get_active()）
+    - 執行 tool calls（全部攔截，不實際執行）
+    - 記錄 step_journal
+    - 完成後 COMPLETED（無 user response）
+
+    Phase 6+ 才接真實 LLM + channel adapter。
+    """
+    run_id = run.run_id
+    project_id = run.project_id
+    agent_id = run.agent_id
+
+    logger.info(
+        "shadow_execute_start",
+        run_id=str(run_id),
+        project_id=project_id,
+        agent_id=agent_id,
+    )
+
+    try:
+        # 解析 agent contract（取得工具清單）
+        from src.services.contract_service import get_active_body
+        agent_contract = await get_active_body(
+            project_id=project_id,
+            contract_family="agent",
+            contract_id=agent_id,
+        )
+
+        tools = agent_contract.get("tools", []) if agent_contract else []
+
+        # Shadow step journal：記錄每個工具會被攔截
+        step_seq = 0
+        async with get_db_context(project_id) as db:
+            for tool in tools:
+                tool_name = tool.get("tool_name", "unknown")
+                blocked = is_destructive_tool(tool_name)
+                step = AwoooPRunStepJournal(
+                    run_id=run_id,
+                    project_id=project_id,
+                    step_seq=step_seq,
+                    tool_name=tool_name,
+                    mcp_gateway_id=tool.get("mcp_gateway_id"),
+                    result_status="success" if not blocked else "pending",
+                    was_blocked=blocked,
+                    block_reason="shadow_mode_destructive_blocked" if blocked else None,
+                )
+                db.add(step)
+                step_seq += 1
+
+        # 完成 run（shadow mode：無 user response）
+        await transition(
+            run_id=run_id,
+            project_id=project_id,
+            to_state="completed",
+            step_count_delta=step_seq,
+        )
+
+        logger.info(
+            "shadow_execute_completed",
+            run_id=str(run_id),
+            steps=step_seq,
+        )
+
+    except Exception as exc:
+        logger.exception(
+            "shadow_execute_failed",
+            run_id=str(run_id),
+            error=str(exc),
+        )
+        await transition(
+            run_id=run_id,
+            project_id=project_id,
+            to_state="failed",
+            error_code="E-RUN-001",
+            error_detail=str(exc)[:500],
+        )
+
+
+async def get_run_status(run_id: uuid.UUID, project_id: str) -> dict[str, Any] | None:
+    """
+    查詢單一 run 的 FSM 狀態。回傳 None 表示不存在。
+    Router 層透過此 service 函數存取，不直接操作 DB。
+    """
+    async with get_db_context(project_id) as db:
+        result = await db.execute(
+            select(AwoooPRunState).where(
+                AwoooPRunState.run_id == run_id,
+                AwoooPRunState.project_id == project_id,
+            )
+        )
+        run = result.scalar_one_or_none()
+
+    if run is None:
+        return None
+
+    return {
+        "run_id": str(run.run_id),
+        "project_id": run.project_id,
+        "agent_id": run.agent_id,
+        "state": run.state,
+        "is_shadow": run.is_shadow,
+        "trace_id": run.trace_id,
+        "attempt_count": run.attempt_count,
+        "cost_usd": float(run.cost_usd),
+        "step_count": run.step_count,
+        "error_code": run.error_code,
+        "created_at": run.created_at.isoformat() if run.created_at else None,
+        "started_at": run.started_at.isoformat() if run.started_at else None,
+        "completed_at": run.completed_at.isoformat() if run.completed_at else None,
+    }

apps/api/src/services/provider_proxy.py

@@ -0,0 +1,240 @@
+"""
+Provider Proxy Adapter — EwoooC AwoooP Envelope 注入
+=====================================================
+AwoooP Phase 6: ADR-115 D3
+2026-05-04 ogt + Claude Sonnet 4.6
+
+功能：
+  EwoooC（或任何外部 tenant）的請求在進入 AwoooP 前，
+  必須注入完整的 platform envelope，確保：
+  - project_id 正確（budget/audit/RLS 有效）
+  - agent_id 存在（Gate 2 通過）
+  - trace_id / run_id 有 W3C traceparent format
+  - platform_subject_id 已建立（channel user 身份映射）
+
+使用方式：
+    from src.services.provider_proxy import ProviderProxy
+
+    proxy = ProviderProxy(project_id="ewoooc", db=db)
+    envelope = await proxy.build_envelope(
+        agent_id="openclaw-biz",
+        channel_type="telegram",
+        channel_user_id="123456789",
+        channel_chat_id="123456789",
+    )
+    # envelope 可直接作為 GatewayContext 的初始化參數
+
+設計原則（ADR-115 D3）：
+  - Proxy 只做 envelope 注入（<1ms），不做額外複雜 IO
+  - platform_subject upsert 是唯一 DB write（auto-provisioning）
+  - run_id 由 platform_runtime.create_run() 分配，Proxy 不自行生成
+  - 每個 tenant 有獨立的 budget partition 和 RLS 隔離
+"""
+
+from __future__ import annotations
+
+import hashlib
+import os
+import re
+import struct
+import time
+import uuid
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Any
+from uuid import UUID
+
+import structlog
+from sqlalchemy import select, text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.db.awooop_models import AwoooPPlatformSubject, AwoooPProject
+
+logger = structlog.get_logger(__name__)
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Platform Envelope
+# ─────────────────────────────────────────────────────────────────────────────
+
+@dataclass
+class PlatformEnvelope:
+    """
+    AwoooP Platform Envelope — 每個 EwoooC 請求注入的 metadata。
+
+    下游（Gateway / Budget / Audit）都依賴這個 envelope。
+    """
+    project_id: str
+    agent_id: str
+    trace_id: str              # W3C traceparent
+    platform_subject_id: str   # "{project_id}:{channel_type}:{channel_user_id}"
+    channel_type: str
+    channel_user_id: str
+    channel_chat_id: str | None = None
+    run_id: UUID | None = None             # 由 create_run() 填入
+    policy_revision_id: str | None = None  # active policy contract revision
+    tags: dict[str, Any] = field(default_factory=dict)
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "project_id": self.project_id,
+            "agent_id": self.agent_id,
+            "trace_id": self.trace_id,
+            "platform_subject_id": self.platform_subject_id,
+            "channel_type": self.channel_type,
+            "channel_user_id": self.channel_user_id,
+            "channel_chat_id": self.channel_chat_id,
+            "run_id": str(self.run_id) if self.run_id else None,
+            "policy_revision_id": self.policy_revision_id,
+        }
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# W3C traceparent 生成
+# ─────────────────────────────────────────────────────────────────────────────
+
+def _new_trace_id() -> str:
+    """生成 W3C traceparent 格式 trace_id。格式：00-{32hex}-{16hex}-01"""
+    trace_id = uuid.uuid4().hex  # 32 hex chars = 128 bits
+    span_id = uuid.uuid4().hex[:16]  # 16 hex chars = 64 bits
+    return f"00-{trace_id}-{span_id}-01"
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# platform_subject_id 格式
+# ─────────────────────────────────────────────────────────────────────────────
+
+def build_platform_subject_id(project_id: str, channel_type: str, channel_user_id: str) -> str:
+    """
+    格式：{project_id}:{channel_type}:{channel_user_id}
+    例：ewoooc:telegram:123456789
+    """
+    return f"{project_id}:{channel_type}:{channel_user_id}"
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# ProviderProxy
+# ─────────────────────────────────────────────────────────────────────────────
+
+class ProviderProxy:
+    """
+    AwoooP Provider Proxy Adapter（ADR-115 D3）。
+
+    職責：
+    1. 驗證 project 存在且不是 legacy mode
+    2. upsert platform_subject（auto-provisioning）
+    3. 生成 trace_id（W3C traceparent）
+    4. 返回 PlatformEnvelope 供下游使用
+    """
+
+    def __init__(self, project_id: str, db: AsyncSession) -> None:
+        self.project_id = project_id
+        self._db = db
+
+    async def build_envelope(
+        self,
+        *,
+        agent_id: str,
+        channel_type: str,
+        channel_user_id: str,
+        channel_chat_id: str | None = None,
+        display_name: str | None = None,
+        extra_tags: dict[str, Any] | None = None,
+    ) -> PlatformEnvelope:
+        """
+        建立 PlatformEnvelope：
+        1. 驗證 project_id（不是 legacy mode）
+        2. upsert platform_subject（auto-provisioning）
+        3. 生成 trace_id
+        4. 返回 envelope
+        """
+        await self._validate_project()
+        await self._upsert_platform_subject(
+            channel_type=channel_type,
+            channel_user_id=channel_user_id,
+            channel_chat_id=channel_chat_id,
+            display_name=display_name,
+        )
+
+        platform_subject_id = build_platform_subject_id(
+            self.project_id, channel_type, channel_user_id
+        )
+        trace_id = _new_trace_id()
+
+        logger.info(
+            "provider_proxy_envelope_built",
+            project_id=self.project_id,
+            agent_id=agent_id,
+            channel_type=channel_type,
+            platform_subject_id=platform_subject_id,
+            trace_id=trace_id[:32] + "...",  # 只 log 前 32 字元
+        )
+
+        return PlatformEnvelope(
+            project_id=self.project_id,
+            agent_id=agent_id,
+            trace_id=trace_id,
+            platform_subject_id=platform_subject_id,
+            channel_type=channel_type,
+            channel_user_id=channel_user_id,
+            channel_chat_id=channel_chat_id,
+            tags=extra_tags or {},
+        )
+
+    async def _validate_project(self) -> None:
+        """project 必須存在且不是 legacy_awoooi_default mode"""
+        result = await self._db.execute(
+            select(AwoooPProject).where(
+                AwoooPProject.project_id == self.project_id,
+                AwoooPProject.migration_mode != "legacy_awoooi_default",
+            )
+        )
+        project = result.scalar_one_or_none()
+        if project is None:
+            raise ValueError(
+                f"project '{self.project_id}' 不存在或 migration_mode=legacy_awoooi_default"
+                "（EwoooC 接入需要至少 migration_mode='shadow'）"
+            )
+
+    async def _upsert_platform_subject(
+        self,
+        *,
+        channel_type: str,
+        channel_user_id: str,
+        channel_chat_id: str | None,
+        display_name: str | None,
+    ) -> None:
+        """
+        Auto-provisioning：第一次看到這個 channel user 就建立 platform_subject。
+        後續請求更新 last_seen_at。
+        """
+        platform_subject_id = build_platform_subject_id(
+            self.project_id, channel_type, channel_user_id
+        )
+        now = datetime.now(timezone.utc)
+
+        await self._db.execute(
+            text("""
+                INSERT INTO awooop_platform_subjects (
+                    project_id, channel_type, channel_user_id, channel_chat_id,
+                    platform_subject_id, display_name, roles,
+                    first_seen_at, last_seen_at
+                ) VALUES (
+                    :project_id, :channel_type, :channel_user_id, :channel_chat_id,
+                    :platform_subject_id, :display_name, '["viewer"]'::jsonb,
+                    :now, :now
+                )
+                ON CONFLICT (project_id, channel_type, channel_user_id) DO UPDATE SET
+                    last_seen_at = :now,
+                    channel_chat_id = COALESCE(EXCLUDED.channel_chat_id, awooop_platform_subjects.channel_chat_id),
+                    display_name = COALESCE(EXCLUDED.display_name, awooop_platform_subjects.display_name)
+            """),
+            {
+                "project_id": self.project_id,
+                "channel_type": channel_type,
+                "channel_user_id": channel_user_id,
+                "channel_chat_id": channel_chat_id,
+                "platform_subject_id": platform_subject_id,
+                "display_name": display_name,
+                "now": now,
+            },
+        )

apps/api/src/services/run_state_machine.py

@@ -0,0 +1,304 @@
+"""
+Run State Machine
+==================
+AwoooP Phase 4: Run FSM 轉換規則 + Worker Lease（ADR-114/ADR-119）
+2026-05-04 ogt + Claude Sonnet 4.6
+
+狀態機：
+  PENDING → RUNNING（worker 取得 lease）
+  RUNNING → WAITING_TOOL（等待 tool call 完成）
+  RUNNING → WAITING_APPROVAL（等待人工審核）
+  RUNNING → COMPLETED / FAILED / CANCELLED
+  WAITING_TOOL → RUNNING（tool call 完成）
+  WAITING_TOOL → FAILED（tool call 失敗 + 超過 max_attempts）
+  WAITING_APPROVAL → RUNNING（核准）
+  WAITING_APPROVAL → CANCELLED（拒絕/超時）
+  * → TIMEOUT（lease_until 過期且超過 max_attempts）
+
+SKIP LOCKED：
+  Worker 以 SELECT ... FOR UPDATE SKIP LOCKED 取單，防 double-pickup。
+  Lease TTL = 60 秒；Heartbeat 每 15 秒更新。
+
+Stale run reaper：
+  每分鐘掃描 lease_until < NOW() 的 running run：
+    attempt_count < max_attempts → 重設 PENDING
+    attempt_count >= max_attempts → 標記 FAILED(E-RUN-002)
+"""
+
+from __future__ import annotations
+
+import socket
+import uuid
+from datetime import datetime, timedelta, timezone
+from typing import TYPE_CHECKING
+
+import structlog
+from sqlalchemy import select, text, update
+
+from src.db.awooop_models import AwoooPRunState
+from src.db.base import get_db_context
+
+if TYPE_CHECKING:
+    from uuid import UUID
+
+logger = structlog.get_logger(__name__)
+
+# Worker lease TTL（秒）
+LEASE_TTL_SECONDS = 60
+HEARTBEAT_INTERVAL_SECONDS = 15
+STALE_REAPER_INTERVAL_SECONDS = 60
+
+# 有效的 FSM 轉換表
+# key: from_state, value: set of valid to_states
+_VALID_TRANSITIONS: dict[str, frozenset[str]] = {
+    "pending":          frozenset({"running", "cancelled"}),
+    "running":          frozenset({"waiting_tool", "waiting_approval", "completed", "failed", "cancelled", "timeout"}),
+    "waiting_tool":     frozenset({"running", "failed", "cancelled"}),
+    "waiting_approval": frozenset({"running", "cancelled", "timeout"}),
+    "completed":        frozenset(),  # terminal
+    "failed":           frozenset(),  # terminal
+    "cancelled":        frozenset(),  # terminal
+    "timeout":          frozenset(),  # terminal
+}
+
+TERMINAL_STATES = frozenset({"completed", "failed", "cancelled", "timeout"})
+
+_WORKER_ID = f"{socket.gethostname()}:{uuid.uuid4().hex[:8]}"
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# FSM 驗證
+# ─────────────────────────────────────────────────────────────────────────────
+
+class InvalidStateTransitionError(Exception):
+    def __init__(self, from_state: str, to_state: str) -> None:
+        self.from_state = from_state
+        self.to_state = to_state
+        super().__init__(f"非法 FSM 轉換: {from_state!r} → {to_state!r}")
+
+
+def validate_transition(from_state: str, to_state: str) -> None:
+    """驗證 FSM 轉換是否合法，非法則拋出 InvalidStateTransitionError"""
+    valid_targets = _VALID_TRANSITIONS.get(from_state, frozenset())
+    if to_state not in valid_targets:
+        raise InvalidStateTransitionError(from_state, to_state)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Worker Lease（SKIP LOCKED）
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def acquire_pending_run(
+    project_id: str,
+    worker_id: str = _WORKER_ID,
+) -> AwoooPRunState | None:
+    """
+    以 SKIP LOCKED 取得一筆 PENDING run，並設定 lease。
+
+    同時只有一個 worker 可取得同一筆 run（PostgreSQL SKIP LOCKED 保證）。
+    Returns None 表示目前沒有待處理的 run。
+    """
+    lease_until = datetime.now(timezone.utc) + timedelta(seconds=LEASE_TTL_SECONDS)
+    now = datetime.now(timezone.utc)
+
+    async with get_db_context(project_id) as db:
+        # SKIP LOCKED：其他 worker 已鎖定的 row 直接跳過
+        result = await db.execute(
+            text("""
+                SELECT run_id FROM awooop_run_state
+                WHERE project_id = :project_id
+                  AND state = 'pending'
+                  AND (lease_until IS NULL OR lease_until < NOW())
+                ORDER BY created_at ASC
+                LIMIT 1
+                FOR UPDATE SKIP LOCKED
+            """),
+            {"project_id": project_id},
+        )
+        row = result.fetchone()
+        if row is None:
+            return None
+
+        run_id = row[0]
+
+        # 更新 lease + 轉為 RUNNING
+        await db.execute(
+            update(AwoooPRunState)
+            .where(
+                AwoooPRunState.run_id == run_id,
+                AwoooPRunState.project_id == project_id,
+            )
+            .values(
+                state="running",
+                lease_until=lease_until,
+                heartbeat_at=now,
+                worker_id=worker_id,
+                started_at=now,
+                attempt_count=AwoooPRunState.attempt_count + 1,
+            )
+        )
+
+        # 重新讀取完整 record
+        result2 = await db.execute(
+            select(AwoooPRunState).where(AwoooPRunState.run_id == run_id)
+        )
+        run = result2.scalar_one()
+
+    logger.info(
+        "run_lease_acquired",
+        run_id=str(run_id),
+        project_id=project_id,
+        worker_id=worker_id,
+        attempt_count=run.attempt_count,
+    )
+    return run
+
+
+async def heartbeat(run_id: "UUID", project_id: str) -> None:
+    """更新 run 的 heartbeat + 延長 lease TTL"""
+    new_lease = datetime.now(timezone.utc) + timedelta(seconds=LEASE_TTL_SECONDS)
+    async with get_db_context(project_id) as db:
+        await db.execute(
+            update(AwoooPRunState)
+            .where(
+                AwoooPRunState.run_id == run_id,
+                AwoooPRunState.state == "running",
+            )
+            .values(
+                heartbeat_at=datetime.now(timezone.utc),
+                lease_until=new_lease,
+            )
+        )
+
+
+async def transition(
+    run_id: "UUID",
+    project_id: str,
+    to_state: str,
+    *,
+    error_code: str | None = None,
+    error_detail: str | None = None,
+    output_sha256: str | None = None,
+    cost_usd_delta: float = 0.0,
+    step_count_delta: int = 0,
+) -> None:
+    """
+    執行 FSM 狀態轉換（含驗證）。
+
+    先從 DB 讀取 current state，驗證轉換合法性，再 UPDATE。
+    terminal state 同時寫入 completed_at。
+    """
+    async with get_db_context(project_id) as db:
+        result = await db.execute(
+            select(AwoooPRunState.state).where(
+                AwoooPRunState.run_id == run_id,
+                AwoooPRunState.project_id == project_id,
+            )
+        )
+        row = result.fetchone()
+        if row is None:
+            raise ValueError(f"run {run_id} 不存在或無 RLS 權限")
+
+        from_state = row[0]
+        validate_transition(from_state, to_state)
+
+        values: dict = {"state": to_state}
+        if error_code:
+            values["error_code"] = error_code
+        if error_detail:
+            values["error_detail"] = error_detail
+        if output_sha256:
+            values["output_sha256"] = output_sha256
+        if cost_usd_delta:
+            values["cost_usd"] = AwoooPRunState.cost_usd + cost_usd_delta
+        if step_count_delta:
+            values["step_count"] = AwoooPRunState.step_count + step_count_delta
+        if to_state in TERMINAL_STATES:
+            values["completed_at"] = datetime.now(timezone.utc)
+            values["lease_until"] = None
+            values["worker_id"] = None
+
+        await db.execute(
+            update(AwoooPRunState)
+            .where(AwoooPRunState.run_id == run_id)
+            .values(**values)
+        )
+
+    logger.info(
+        "run_state_transition",
+        run_id=str(run_id),
+        from_state=from_state,
+        to_state=to_state,
+        error_code=error_code,
+    )
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Stale Run Reaper
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def reap_stale_runs(project_id: str) -> int:
+    """
+    掃描 lease_until < NOW() 的 RUNNING run。
+    - attempt_count < max_attempts → 重設 PENDING（retry）
+    - attempt_count >= max_attempts → FAILED(E-RUN-002)
+
+    Returns: 處理的 stale run 數
+    """
+    now = datetime.now(timezone.utc)
+    reaped = 0
+
+    async with get_db_context(project_id) as db:
+        # 找所有 stale RUNNING runs
+        result = await db.execute(
+            select(AwoooPRunState).where(
+                AwoooPRunState.project_id == project_id,
+                AwoooPRunState.state == "running",
+                AwoooPRunState.lease_until < now,
+            )
+        )
+        stale_runs = list(result.scalars().all())
+
+        for run in stale_runs:
+            if run.attempt_count < run.max_attempts:
+                # Retry：重設為 PENDING
+                await db.execute(
+                    update(AwoooPRunState)
+                    .where(AwoooPRunState.run_id == run.run_id)
+                    .values(
+                        state="pending",
+                        lease_until=None,
+                        worker_id=None,
+                        heartbeat_at=None,
+                    )
+                )
+                logger.warning(
+                    "stale_run_requeued",
+                    run_id=str(run.run_id),
+                    attempt_count=run.attempt_count,
+                    max_attempts=run.max_attempts,
+                )
+            else:
+                # 超過最大重試次數 → FAILED
+                await db.execute(
+                    update(AwoooPRunState)
+                    .where(AwoooPRunState.run_id == run.run_id)
+                    .values(
+                        state="failed",
+                        error_code="E-RUN-002",
+                        error_detail=f"max_attempts={run.max_attempts} 超過，stale run 已廢棄",
+                        completed_at=now,
+                        lease_until=None,
+                        worker_id=None,
+                    )
+                )
+                logger.error(
+                    "stale_run_failed",
+                    run_id=str(run.run_id),
+                    attempt_count=run.attempt_count,
+                )
+            reaped += 1
+
+    if reaped:
+        logger.info("stale_run_reaper_done", project_id=project_id, reaped=reaped)
+    return reaped

apps/api/src/services/schema_validator.py

@@ -0,0 +1,262 @@
+"""
+LLM Output Schema Validator
+=============================
+AwoooP Phase 3.3: LLM 輸出 → schema 驗證 → retry 機制（ADR-112）
+2026-05-04 ogt + Claude Sonnet 4.6
+
+設計原則：
+- LLM 輸出必須通過 Pydantic schema 驗證才能到達 channel adapter
+- 驗證失敗 → 自動 retry（最多 3 次，含 retry prompt）
+- 3 次全部失敗 → 拋出 SchemaValidationError（E-SCHEMA-001）
+- 支援六合約家族 + 自訂 Pydantic model
+
+位置：介於 LLM response 和 channel adapter 之間
+呼叫方：任何需要結構化 LLM 輸出的 service（playbook_generator, decision_manager 等）
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from typing import Any, TypeVar
+
+import structlog
+from pydantic import BaseModel, ValidationError
+
+logger = structlog.get_logger(__name__)
+
+T = TypeVar("T", bound=BaseModel)
+
+_MAX_RETRIES = 3
+_JSON_EXTRACT_RE = re.compile(r"```(?:json)?\s*(\{[\s\S]*?\})\s*```|(\{[\s\S]*\})", re.DOTALL)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 錯誤定義
+# ─────────────────────────────────────────────────────────────────────────────
+
+class SchemaValidationError(Exception):
+    """LLM 輸出連續 3 次 schema 驗證失敗"""
+
+    error_code: str = "E-SCHEMA-001"
+
+    def __init__(self, model_name: str, attempts: int, last_error: str) -> None:
+        self.model_name = model_name
+        self.attempts = attempts
+        self.last_error = last_error
+        super().__init__(
+            f"[E-SCHEMA-001] LLM 輸出 {attempts} 次驗證失敗 "
+            f"(model={model_name}): {last_error}"
+        )
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# JSON 萃取（容錯解析）
+# ─────────────────────────────────────────────────────────────────────────────
+
+def extract_json_from_llm_output(raw: str) -> dict[str, Any] | None:
+    """
+    從 LLM 原始輸出中萃取 JSON。
+    策略：
+    1. 直接 json.loads（最常見：LLM 直接回傳 JSON）
+    2. 從 ```json ... ``` 程式碼區塊萃取
+    3. 找第一個 { ... } 區塊嘗試解析
+    """
+    raw = raw.strip()
+
+    # 策略 1：直接解析
+    try:
+        obj = json.loads(raw)
+        if isinstance(obj, dict):
+            return obj
+    except json.JSONDecodeError:
+        pass
+
+    # 策略 2 + 3：正則萃取
+    for match in _JSON_EXTRACT_RE.finditer(raw):
+        candidate = match.group(1) or match.group(2)
+        if candidate:
+            try:
+                obj = json.loads(candidate)
+                if isinstance(obj, dict):
+                    return obj
+            except json.JSONDecodeError:
+                continue
+
+    return None
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Retry prompt builder
+# ─────────────────────────────────────────────────────────────────────────────
+
+def build_retry_prompt(
+    original_prompt: str,
+    failed_output: str,
+    validation_error: str,
+    model_name: str,
+    attempt: int,
+) -> str:
+    """
+    建立包含錯誤回饋的 retry prompt。
+    讓 LLM 知道上次輸出哪裡出錯，引導修正。
+    """
+    return (
+        f"{original_prompt}\n\n"
+        f"---\n"
+        f"[SCHEMA VALIDATION RETRY {attempt}/{_MAX_RETRIES}]\n"
+        f"上次回應未通過結構驗證（{model_name}），請修正以下問題後重新回應：\n\n"
+        f"驗證錯誤：\n{validation_error}\n\n"
+        f"上次回應（供參考）：\n{failed_output[:500]}...\n"
+        f"---\n\n"
+        f"請只回傳符合格式的 JSON 物件，不要包含任何額外說明。"
+    )
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Core validator
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def validate_llm_output(
+    *,
+    raw_output: str,
+    model_cls: type[T],
+    llm_caller: Any,  # Callable[[str], Awaitable[str]] — 供 retry 使用
+    original_prompt: str,
+    context: dict[str, Any] | None = None,
+) -> T:
+    """
+    驗證 LLM 輸出是否符合 Pydantic model。
+
+    Args:
+        raw_output:      LLM 第一次回傳的原始字串
+        model_cls:       目標 Pydantic model class
+        llm_caller:      async callable(prompt: str) -> str，用於 retry
+        original_prompt: 原始 prompt（retry 時附加錯誤回饋）
+        context:         額外 logging context
+
+    Returns:
+        驗證成功的 model instance
+
+    Raises:
+        SchemaValidationError: 連續 3 次失敗後拋出
+    """
+    model_name = model_cls.__name__
+    ctx = context or {}
+    current_output = raw_output
+    last_error = ""
+
+    for attempt in range(1, _MAX_RETRIES + 1):
+        # 1. 萃取 JSON
+        parsed = extract_json_from_llm_output(current_output)
+        if parsed is None:
+            last_error = "無法從 LLM 輸出中萃取 JSON 物件"
+            logger.warning(
+                "schema_validator_no_json",
+                model_name=model_name,
+                attempt=attempt,
+                output_preview=current_output[:200],
+                **ctx,
+            )
+        else:
+            # 2. Pydantic 驗證
+            try:
+                instance = model_cls.model_validate(parsed)
+                logger.info(
+                    "schema_validator_passed",
+                    model_name=model_name,
+                    attempt=attempt,
+                    **ctx,
+                )
+                return instance
+            except ValidationError as exc:
+                last_error = exc.json(indent=None)
+                logger.warning(
+                    "schema_validator_failed",
+                    model_name=model_name,
+                    attempt=attempt,
+                    error=last_error[:500],
+                    **ctx,
+                )
+
+        # 3. Retry（如果不是最後一次）
+        if attempt < _MAX_RETRIES:
+            retry_prompt = build_retry_prompt(
+                original_prompt=original_prompt,
+                failed_output=current_output,
+                validation_error=last_error,
+                model_name=model_name,
+                attempt=attempt,
+            )
+            try:
+                current_output = await llm_caller(retry_prompt)
+            except Exception as exc:
+                logger.warning(
+                    "schema_validator_llm_retry_failed",
+                    model_name=model_name,
+                    attempt=attempt,
+                    error=str(exc),
+                    **ctx,
+                )
+                # LLM 呼叫本身失敗，保留上次 output，繼續嘗試（或直接結束）
+                break
+
+    # 3 次全失敗
+    logger.error(
+        "schema_validator_exhausted",
+        model_name=model_name,
+        total_attempts=_MAX_RETRIES,
+        last_error=last_error[:500],
+        **ctx,
+    )
+    raise SchemaValidationError(model_name, _MAX_RETRIES, last_error)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 便利方法：從 contract family 名稱驗證（不需知道具體 model class）
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def validate_llm_output_by_family(
+    *,
+    raw_output: str,
+    contract_family: str,
+    llm_caller: Any,
+    original_prompt: str,
+    context: dict[str, Any] | None = None,
+) -> BaseModel:
+    """
+    依 contract_family 自動選擇 model class 並驗證。
+    適合 generic pipeline 呼叫（不知道具體 model）。
+    """
+    from src.models.awooop_contracts import CONTRACT_FAMILY_MODELS, VALID_CONTRACT_FAMILIES
+
+    model_cls = CONTRACT_FAMILY_MODELS.get(contract_family)
+    if model_cls is None:
+        raise ValueError(
+            f"未知 contract_family: {contract_family!r}，"
+            f"合法值：{sorted(VALID_CONTRACT_FAMILIES)}"
+        )
+    return await validate_llm_output(
+        raw_output=raw_output,
+        model_cls=model_cls,
+        llm_caller=llm_caller,
+        original_prompt=original_prompt,
+        context=context,
+    )
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 同步版本（非 LLM retry，只做一次驗證）— 供測試和非 LLM 路徑使用
+# ─────────────────────────────────────────────────────────────────────────────
+
+def validate_once(raw: str | dict[str, Any], model_cls: type[T]) -> T:
+    """
+    單次驗證，不做 retry。
+    適合：已知格式正確的內部資料、測試 fixture 驗證。
+    """
+    if isinstance(raw, str):
+        parsed = extract_json_from_llm_output(raw)
+        if parsed is None:
+            raise SchemaValidationError(model_cls.__name__, 1, "無法萃取 JSON")
+        return model_cls.model_validate(parsed)
+    return model_cls.model_validate(raw)