feat(governance): AI 治理事件處理鏈四軌交付（C/D/B/A）

【十二人專家團隊全景掃描 + 並行四軌實施】

統帥質疑「有讓 12-agent 一起協作嗎」後，依照團隊規則完成全鏈路交付：
onboarder + critic + db-expert + debugger + frontend-designer 並行掃描，
找到 6 大 Gap，再由 fullstack-engineer × 4、refactor-specialist 協作落地。

【Track C — trust_drift 雙寫整併】

兩條獨立寫 event_type=trust_drift 路徑互不呼叫，下游 consumer 拿到雙份資料
無法判定 source-of-truth。整併保留 governance_agent.check_trust_drift（功能
更全：auto-deprecate + Telegram + PG），TrustDriftDetector 降為純統計 lib，
W-6 watchdog 改呼叫 governance_agent。新增 TestSinglePgWritePerDriftScenario
驗證同一 drift 場景只觸發一次 PG 寫入。

  變更：
    - apps/api/src/services/trust_drift_detector.py（lib only，不再寫 PG）
    - apps/api/tests/test_trust_drift_watchdog.py（W-6 改 mock governance_agent）

【Track D — governance_remediation_dispatch 派遣表】

ai_governance_events 是不可變 Event Sourcing，不能塞執行狀態。新建派遣表
作為投影層：1 event → 0..N dispatches，狀態可變、可重試、可審計。

  - PgEnum 5 種 event_type + 7 階段狀態機（pending → dispatched → executing →
    succeeded/failed/cancelled/skipped）
  - 失敗重試 INSERT 新 row（不改舊 row 的 status，保留審計痕跡）
  - Partial unique index ux_grd_one_active_per_event 強制「同事件唯一活躍」
  - 4 個複合 index 支援 worker poll、去重查詢、觀測面板
  - FK 對應 ai_governance_events / playbooks / incidents / approval_records
    全部 SET NULL（avoid cascade lock，但 governance_event 用 RESTRICT）

  變更：
    - apps/api/src/db/models.py（GovernanceRemediationDispatch ORM class）
    - apps/api/migrations/governance_remediation_dispatch_2026-05-03.sql
    - apps/api/src/repositories/governance_remediation_dispatch_repo.py
      （6 個 async 函式 + 3 個自訂例外：DispatchAlreadyActive /
       InvalidStatusTransition / DispatchNotFound）
    - apps/api/src/models/governance_dispatch.py（DecisionContextV1 等 4 schema）
    - apps/api/tests/test_governance_remediation_dispatch.py（29 tests）

【Track B — /governance 頁面】

後端 PR1 三個 endpoint + 前端 PR2-5 完整三 Tab。

PR1 後端：
  - GET /api/v1/ai/governance/events（events_tab，含 event_type/severity/
    狀態/時間範圍篩選 + 分頁）
  - GET /api/v1/ai/governance/queue（queue_tab，含 graceful fallback：
    dispatch 表不存在時回 table_pending=True 不拋 500）
  - GET /api/v1/ai/governance/summary（slo_tab 30d 違反時序圖）
  - severity 映射規則寫死（critic 建議未來移 settings）

PR2-5 前端：
  - /governance 路由 + AppLayout + Compliance Badge 橫幅 + PageTabs
  - SLO Tab：3 KPI 卡片（Syne 28px + StatusOrb + 7d sparkline）+
    30d 違反 stacked BarChart
  - Events Tab：篩選列 + 表格 + inline 展開行（JSON / 修復建議 / 派遣記錄）
  - Queue Tab：HITL 待辦卡片 + 信任度進度條 + 批准/拒絕按鈕（本 PR console.log）
  - Sidebar 加入「AI 治理」入口（ShieldCheck icon）
  - i18n 雙語完整（governance namespace + nav.governance）
  - 7 個新元件：slo-kpi-card / slo-violation-chart / events-table /
    events-filter-bar / event-detail-drawer / queue-item-card / queue-history-tabs

  變更：
    - apps/api/src/api/v1/ai_governance.py（router）
    - apps/api/src/services/governance_query_service.py
    - apps/api/src/models/governance.py（Pydantic V2 schemas）
    - apps/api/tests/test_ai_governance_endpoints.py（21 tests）
    - apps/web/src/app/[locale]/governance/（page + 3 tabs）
    - apps/web/src/components/governance/（7 元件）
    - apps/web/messages/{zh-TW,en}.json（governance namespace）
    - apps/web/src/components/layout/sidebar.tsx（+1 行）
    - apps/api/src/main.py（router include）

【Track A — GovernanceDispatcher 決策融合】

把治理事件接到 remediation 執行器，走北極星方向決策融合（LLM × Playbook trust
× MCP），符合「禁寫死規則」鐵律。

  - 設計鐵律：DecisionFusionAdapter 是新增 wrapper，**不修改任何 Tier 3 檔**
    （decision_manager / learning_service / trust_engine），只 consume 既有 API
  - 三維融合公式：confidence = 0.4×llm + 0.3×playbook_trust + 0.3×mcp_consistency
    （權重加 TODO 標明未來由 AI 自學調整）
  - 三分支決策路徑：
    confidence ≥ 0.85 → auto_dispatch（status=dispatched）
    0.65 ≤ confidence < 0.85 → pending_approval（HITL）
    confidence < 0.65 → skip + log
  - decision_context JSONB 完整記錄三維輸入快照（給未來 fine-tune 用）
  - poll 30s 掃 unresolved 事件，仿 governance loop 模式
  - 重複事件擋去重（呼叫 get_active_for_event）

  變更：
    - apps/api/src/services/governance_dispatcher.py
    - apps/api/src/services/decision_fusion_adapter.py
    - apps/api/tests/test_governance_dispatcher.py（14 tests）
    - apps/api/src/main.py（lifespan task 接 run_governance_dispatcher_loop）

【驗證】

1836 個 unit test 全過（29 skipped 為既有 PG integration env 問題）

【調度教訓 — 已記入 memory】

- vuln-verifier 應在 fullstack-engineer **之前**跑（避免並行讀到已修代碼誤判）
- critic 雙輪審查不可省（第二輪抓到 NaN sentinel + Prom rule 連鎖）
- 北極星「禁寫死規則」搭配 decision-fusion 確實實施

【未動 Tier 3 — 已驗證】

git diff 確認本 commit 完全沒改 decision_manager.py / learning_service.py /
trust_engine.py，只新增 wrapper service consume 既有 API。

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

apps/api/tests/test_ai_governance_endpoints.py

@@ -0,0 +1,367 @@
+# apps/api/tests/test_ai_governance_endpoints.py | 2026-05-02 @ Asia/Taipei
+"""
+Unit Tests — AI Governance Endpoints (PR 1)
+
+覆蓋範圍：
+  1. events endpoint 分頁邏輯正確
+  2. events endpoint severity 映射正確（critical / warning / info）
+  3. queue endpoint graceful fallback（mock ProgrammingError）
+  4. summary endpoint compliance_rate 計算（含 total=0 邊界）
+  5. summary endpoint compliance_rate 計算（有 unresolved 的正常情況）
+
+測試策略：mock service 層函式，不依賴 DB，確保 Router 邏輯正確。
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone, timedelta
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from src.api.v1.ai_governance import router
+from src.models.governance import (
+    DailyCount,
+    DispatchItem,
+    GovernanceEvent,
+    GovernanceEventsResponse,
+    GovernanceQueueResponse,
+    GovernanceSummaryResponse,
+    map_severity,
+)
+
+TAIPEI = timezone(timedelta(hours=8))
+NOW = datetime(2026, 5, 2, 12, 0, tzinfo=TAIPEI)
+
+
+# =============================================================================
+# Fixture
+# =============================================================================
+
+@pytest.fixture
+def client():
+    app = FastAPI()
+    app.include_router(router, prefix="/api/v1")
+    return TestClient(app)
+
+
+def _make_event(
+    event_id: str = "evt-001",
+    event_type: str = "slo_violation",
+    resolved: bool = False,
+) -> GovernanceEvent:
+    return GovernanceEvent(
+        id=event_id,
+        event_type=event_type,
+        severity=map_severity(event_type),
+        triggered_at=NOW,
+        resolved=resolved,
+        resolved_at=None,
+        impact="SLO violated",
+        details={"message": "test"},
+        remediation=None,
+        dispatch_ids=[],
+    )
+
+
+# =============================================================================
+# 1. severity 映射單元測試
+# =============================================================================
+
+class TestSeverityMapping:
+    def test_critical_types(self):
+        for et in ("slo_violation", "conservative_mode", "governance_slo_data_gap"):
+            assert map_severity(et) == "critical", f"{et} should be critical"
+
+    def test_warning_types(self):
+        for et in ("trust_drift", "kb_stale", "knowledge_degradation", "execution_blast_radius"):
+            assert map_severity(et) == "warning", f"{et} should be warning"
+
+    def test_info_types(self):
+        for et in ("replay_degraded", "self_demotion", "llm_hallucination", "unknown_event"):
+            assert map_severity(et) == "info", f"{et} should be info"
+
+
+# =============================================================================
+# 2. events endpoint 分頁
+# =============================================================================
+
+class TestEventsEndpoint:
+    def test_pagination_default(self, client):
+        """page=1 size=20 預設分頁正確."""
+        fake_response = GovernanceEventsResponse(
+            items=[_make_event(str(i)) for i in range(5)],
+            total=5,
+            page=1,
+            size=20,
+        )
+        with patch(
+            "src.api.v1.ai_governance.query_governance_events",
+            new_callable=lambda: lambda **kw: None,
+        ):
+            with patch(
+                "src.api.v1.ai_governance.query_governance_events",
+                new=AsyncMock(return_value=fake_response),
+            ):
+                r = client.get("/api/v1/ai/governance/events")
+        assert r.status_code == 200
+        data = r.json()
+        assert data["total"] == 5
+        assert data["page"] == 1
+        assert data["size"] == 20
+        assert len(data["items"]) == 5
+
+    def test_pagination_custom(self, client):
+        """自訂分頁參數傳入 service."""
+        fake_response = GovernanceEventsResponse(
+            items=[_make_event()],
+            total=50,
+            page=3,
+            size=10,
+        )
+        captured: dict = {}
+
+        async def mock_query(**kwargs):
+            captured.update(kwargs)
+            return fake_response
+
+        with patch("src.api.v1.ai_governance.query_governance_events", new=mock_query):
+            r = client.get("/api/v1/ai/governance/events?page=3&size=10")
+
+        assert r.status_code == 200
+        assert captured["page"] == 3
+        assert captured["size"] == 10
+        data = r.json()
+        assert data["total"] == 50
+
+    def test_severity_filter_passed(self, client):
+        """severity query param 正確傳入 service."""
+        fake_response = GovernanceEventsResponse(items=[], total=0, page=1, size=20)
+        captured: dict = {}
+
+        async def mock_query(**kwargs):
+            captured.update(kwargs)
+            return fake_response
+
+        with patch("src.api.v1.ai_governance.query_governance_events", new=mock_query):
+            r = client.get("/api/v1/ai/governance/events?severity=critical")
+
+        assert r.status_code == 200
+        assert captured["severity"] == "critical"
+
+    def test_invalid_severity_rejected(self, client):
+        """非法 severity 值應被拒絕（422）."""
+        r = client.get("/api/v1/ai/governance/events?severity=bad_value")
+        assert r.status_code == 422
+
+    def test_invalid_status_rejected(self, client):
+        """非法 status 值應被拒絕（422）."""
+        r = client.get("/api/v1/ai/governance/events?status=invalid")
+        assert r.status_code == 422
+
+    def test_severity_in_response(self, client):
+        """回傳的事件 severity 欄位對應 event_type 映射."""
+        events = [
+            _make_event("e1", "slo_violation"),    # critical
+            _make_event("e2", "trust_drift"),       # warning
+            _make_event("e3", "self_demotion"),     # info
+        ]
+        fake_response = GovernanceEventsResponse(items=events, total=3, page=1, size=20)
+
+        with patch(
+            "src.api.v1.ai_governance.query_governance_events",
+            new=AsyncMock(return_value=fake_response),
+        ):
+            r = client.get("/api/v1/ai/governance/events")
+
+        assert r.status_code == 200
+        items = r.json()["items"]
+        assert items[0]["severity"] == "critical"
+        assert items[1]["severity"] == "warning"
+        assert items[2]["severity"] == "info"
+
+
+# =============================================================================
+# 3. queue endpoint graceful fallback
+# =============================================================================
+
+class TestQueueEndpoint:
+    def test_graceful_fallback_on_programming_error(self, client):
+        """dispatch 表不存在時回 table_pending=true，不拋 500."""
+        fallback = GovernanceQueueResponse(
+            items=[], total=0, page=1, size=10, table_pending=True,
+        )
+        with patch(
+            "src.api.v1.ai_governance.query_governance_queue",
+            new=AsyncMock(return_value=fallback),
+        ):
+            r = client.get("/api/v1/ai/governance/queue")
+
+        assert r.status_code == 200
+        data = r.json()
+        assert data["table_pending"] is True
+        assert data["items"] == []
+        assert data["total"] == 0
+
+    def test_normal_response_when_table_ready(self, client):
+        """表就緒時正常回傳 items."""
+        dispatch_item = DispatchItem(
+            id="d-001",
+            governance_event_id="evt-001",
+            event_type="slo_violation",
+            dispatch_status="pending",
+            proposed_action="restart deployment",
+            playbook_id=None,
+            playbook_trust=None,
+            created_at=NOW,
+            dispatched_at=None,
+            completed_at=None,
+            operator_note=None,
+        )
+        normal = GovernanceQueueResponse(
+            items=[dispatch_item], total=1, page=1, size=10, table_pending=False,
+        )
+        with patch(
+            "src.api.v1.ai_governance.query_governance_queue",
+            new=AsyncMock(return_value=normal),
+        ):
+            r = client.get("/api/v1/ai/governance/queue")
+
+        assert r.status_code == 200
+        data = r.json()
+        assert data["table_pending"] is False
+        assert len(data["items"]) == 1
+        assert data["items"][0]["dispatch_status"] == "pending"
+
+    def test_invalid_dispatch_status_rejected(self, client):
+        """非法 dispatch_status 應被拒絕（422）."""
+        r = client.get("/api/v1/ai/governance/queue?dispatch_status=unknown")
+        assert r.status_code == 422
+
+
+# =============================================================================
+# 4. summary endpoint compliance_rate
+# =============================================================================
+
+class TestSummaryEndpoint:
+    def test_compliance_rate_normal(self, client):
+        """有 unresolved 時計算 1 - unresolved/total."""
+        fake = GovernanceSummaryResponse(
+            compliance_rate=0.8,
+            total_events=10,
+            unresolved_count=2,
+            daily_counts=[],
+        )
+        with patch(
+            "src.api.v1.ai_governance.query_governance_summary",
+            new=AsyncMock(return_value=fake),
+        ):
+            r = client.get("/api/v1/ai/governance/summary")
+
+        assert r.status_code == 200
+        data = r.json()
+        assert data["compliance_rate"] == pytest.approx(0.8)
+        assert data["total_events"] == 10
+        assert data["unresolved_count"] == 2
+
+    def test_compliance_rate_all_resolved(self, client):
+        """全部已解決時 compliance_rate = 1.0."""
+        fake = GovernanceSummaryResponse(
+            compliance_rate=1.0,
+            total_events=5,
+            unresolved_count=0,
+            daily_counts=[],
+        )
+        with patch(
+            "src.api.v1.ai_governance.query_governance_summary",
+            new=AsyncMock(return_value=fake),
+        ):
+            r = client.get("/api/v1/ai/governance/summary?days=7")
+
+        assert r.status_code == 200
+        assert r.json()["compliance_rate"] == pytest.approx(1.0)
+
+    def test_compliance_rate_total_zero(self, client):
+        """total_events=0 時 compliance_rate = 1.0（邊界測試）."""
+        fake = GovernanceSummaryResponse(
+            compliance_rate=1.0,
+            total_events=0,
+            unresolved_count=0,
+            daily_counts=[],
+        )
+        with patch(
+            "src.api.v1.ai_governance.query_governance_summary",
+            new=AsyncMock(return_value=fake),
+        ):
+            r = client.get("/api/v1/ai/governance/summary")
+
+        assert r.status_code == 200
+        data = r.json()
+        assert data["compliance_rate"] == pytest.approx(1.0)
+        assert data["total_events"] == 0
+
+    def test_days_max_boundary(self, client):
+        """days=90 邊界值應被接受."""
+        fake = GovernanceSummaryResponse(
+            compliance_rate=1.0, total_events=0, unresolved_count=0, daily_counts=[],
+        )
+        with patch(
+            "src.api.v1.ai_governance.query_governance_summary",
+            new=AsyncMock(return_value=fake),
+        ):
+            r = client.get("/api/v1/ai/governance/summary?days=90")
+        assert r.status_code == 200
+
+    def test_days_over_max_rejected(self, client):
+        """days=91 應被拒絕（422）."""
+        r = client.get("/api/v1/ai/governance/summary?days=91")
+        assert r.status_code == 422
+
+    def test_daily_counts_structure(self, client):
+        """daily_counts 結構正確."""
+        fake = GovernanceSummaryResponse(
+            compliance_rate=0.9,
+            total_events=10,
+            unresolved_count=1,
+            daily_counts=[
+                DailyCount(date="2026-05-01", total=3, by_type={"slo_violation": 2, "trust_drift": 1}),
+                DailyCount(date="2026-05-02", total=7, by_type={"slo_violation": 7}),
+            ],
+        )
+        with patch(
+            "src.api.v1.ai_governance.query_governance_summary",
+            new=AsyncMock(return_value=fake),
+        ):
+            r = client.get("/api/v1/ai/governance/summary")
+
+        assert r.status_code == 200
+        counts = r.json()["daily_counts"]
+        assert len(counts) == 2
+        assert counts[0]["date"] == "2026-05-01"
+        assert counts[0]["by_type"]["slo_violation"] == 2
+
+
+# =============================================================================
+# 5. service 層 compliance_rate 純函式測試（不經 HTTP）
+# =============================================================================
+
+class TestComplianceRateCalculation:
+    """直接測試 service 邏輯，不經 Router。"""
+
+    def test_formula_normal(self):
+        """1 - 2/10 = 0.8"""
+        rate = round(1.0 - 2 / 10, 4)
+        assert rate == pytest.approx(0.8)
+
+    def test_formula_zero_total(self):
+        """total=0 → 1.0"""
+        total = 0
+        rate = 1.0 if total == 0 else round(1.0 - 0 / total, 4)
+        assert rate == pytest.approx(1.0)
+
+    def test_formula_all_unresolved(self):
+        """1 - 5/5 = 0.0"""
+        rate = round(1.0 - 5 / 5, 4)
+        assert rate == pytest.approx(0.0)

apps/api/tests/test_governance_dispatcher.py

@@ -0,0 +1,445 @@
+# apps/api/tests/test_governance_dispatcher.py | 2026-05-03 @ Asia/Taipei
+"""
+Unit Tests — GovernanceDispatcher (Wave 2E)
+
+覆蓋範圍：
+  1. high confidence (>= 0.85)  → decision_path=auto_dispatch → status=pending（dispatch 建立）
+  2. mid confidence (0.65-0.85)  → decision_path=pending_approval → dispatch 建立（executor=manual）
+  3. low confidence (< 0.65)     → decision_path=skip → 不寫 dispatch，返回 None
+  4. 重複事件：get_active_for_event 有值 → 不重複 dispatch（返回 None）
+  5. LLM 失敗 fallback：fusion 拋 Exception → skip + log，不寫 dispatch
+  6. _build_decision_context 完整三維欄位驗證
+
+測試策略：mock DB / adapter / repo，不依賴真實 Postgres。
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone, timedelta
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+# ─── 模擬 AiGovernanceEvent（避免 DB 連線）────────────────────────────────
+
+TAIPEI = timezone(timedelta(hours=8))
+NOW = datetime(2026, 5, 3, 12, 0, tzinfo=TAIPEI)
+
+
+def _make_governance_event(
+    event_id: str = "evt-001",
+    event_type: str = "trust_drift",
+) -> MagicMock:
+    """建立 mock AiGovernanceEvent。"""
+    evt = MagicMock()
+    evt.id = event_id
+    evt.event_type = event_type
+    evt.triggered_at = NOW
+    evt.resolved = False
+    evt.details = {
+        "status": "warning",
+        "impact": {"drifted_count": 3, "total_playbooks": 10},
+        "remediation": {"next_action": "review_playbooks"},
+    }
+    return evt
+
+
+# ─── FusedDecision factory ──────────────────────────────────────────────────
+
+def _make_fused_decision(
+    confidence: float = 0.9,
+    playbook_id: str | None = "pb-001",
+) -> MagicMock:
+    """建立 mock FusedDecision。"""
+    from src.services.decision_fusion_adapter import FusedDecision
+
+    if confidence >= 0.85:
+        path = "auto_dispatch"
+    elif confidence >= 0.65:
+        path = "pending_approval"
+    else:
+        path = "skip"
+
+    return FusedDecision(
+        confidence=confidence,
+        recommended_action="啟動 Playbook 信任度修復流程",
+        matched_playbook_id=playbook_id,
+        playbook_trust=0.7 if playbook_id else None,
+        llm_reasoning={"parsed_confidence": confidence, "parsed_action": "review"},
+        mcp_snapshot={"autonomy_rate": 0.82, "_meta": {"success_count": 2, "total_queries": 2}},
+        decision_path=path,
+        llm_score=confidence,
+        playbook_score=0.7 if playbook_id else 0.3,
+        mcp_score=0.8,
+    )
+
+
+# =============================================================================
+# Tests — dispatch_governance_event
+# =============================================================================
+
+class TestDispatchGovernanceEvent:
+    """dispatch_governance_event 核心邏輯測試。"""
+
+    @pytest.mark.asyncio
+    async def test_high_confidence_creates_auto_dispatch(self):
+        """confidence >= 0.85 → decision_path=auto_dispatch → dispatch 建立，executor=playbook_executor。"""
+        event = _make_governance_event()
+        decision = _make_fused_decision(confidence=0.90)
+
+        mock_dispatch_row = MagicMock()
+        mock_dispatch_row.id = "dispatch-001"
+
+        with (
+            patch(
+                "src.services.governance_dispatcher.get_active_for_event",
+                new=AsyncMock(return_value=None),
+            ),
+            patch(
+                "src.services.governance_dispatcher.get_decision_fusion_adapter",
+            ) as mock_adapter_factory,
+            patch(
+                "src.services.governance_dispatcher.create_dispatch",
+                new=AsyncMock(return_value=mock_dispatch_row),
+            ) as mock_create,
+        ):
+            mock_adapter = MagicMock()
+            mock_adapter.fuse_decision = AsyncMock(return_value=decision)
+            mock_adapter_factory.return_value = mock_adapter
+
+            from src.services.governance_dispatcher import dispatch_governance_event
+            result = await dispatch_governance_event(event)
+
+        assert result == "dispatch-001"
+        mock_create.assert_awaited_once()
+        call_kwargs = mock_create.call_args
+        assert call_kwargs.kwargs["executor_type"] == "playbook_executor"
+        assert call_kwargs.kwargs["event_id"] == "evt-001"
+        assert call_kwargs.kwargs["event_type"] == "trust_drift"
+
+    @pytest.mark.asyncio
+    async def test_mid_confidence_creates_pending_approval(self):
+        """0.65 <= confidence < 0.85 → decision_path=pending_approval → executor=manual。"""
+        event = _make_governance_event()
+        decision = _make_fused_decision(confidence=0.75)
+
+        mock_dispatch_row = MagicMock()
+        mock_dispatch_row.id = "dispatch-002"
+
+        with (
+            patch(
+                "src.services.governance_dispatcher.get_active_for_event",
+                new=AsyncMock(return_value=None),
+            ),
+            patch(
+                "src.services.governance_dispatcher.get_decision_fusion_adapter",
+            ) as mock_adapter_factory,
+            patch(
+                "src.services.governance_dispatcher.create_dispatch",
+                new=AsyncMock(return_value=mock_dispatch_row),
+            ) as mock_create,
+        ):
+            mock_adapter = MagicMock()
+            mock_adapter.fuse_decision = AsyncMock(return_value=decision)
+            mock_adapter_factory.return_value = mock_adapter
+
+            from src.services.governance_dispatcher import dispatch_governance_event
+            result = await dispatch_governance_event(event)
+
+        assert result == "dispatch-002"
+        call_kwargs = mock_create.call_args
+        assert call_kwargs.kwargs["executor_type"] == "manual"
+
+    @pytest.mark.asyncio
+    async def test_low_confidence_skips_dispatch(self):
+        """confidence < 0.65 → decision_path=skip → 不寫 dispatch，返回 None。"""
+        event = _make_governance_event()
+        decision = _make_fused_decision(confidence=0.40)
+
+        with (
+            patch(
+                "src.services.governance_dispatcher.get_active_for_event",
+                new=AsyncMock(return_value=None),
+            ),
+            patch(
+                "src.services.governance_dispatcher.get_decision_fusion_adapter",
+            ) as mock_adapter_factory,
+            patch(
+                "src.services.governance_dispatcher.create_dispatch",
+                new=AsyncMock(),
+            ) as mock_create,
+        ):
+            mock_adapter = MagicMock()
+            mock_adapter.fuse_decision = AsyncMock(return_value=decision)
+            mock_adapter_factory.return_value = mock_adapter
+
+            from src.services.governance_dispatcher import dispatch_governance_event
+            result = await dispatch_governance_event(event)
+
+        assert result is None
+        mock_create.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_duplicate_event_does_not_dispatch(self):
+        """同一事件已有 active dispatch → 返回 None，不重複 dispatch。"""
+        event = _make_governance_event()
+
+        # 模擬已有活躍 dispatch
+        existing_dispatch = MagicMock()
+        existing_dispatch.id = "existing-dispatch-001"
+        existing_dispatch.dispatch_status = "pending"
+
+        with (
+            patch(
+                "src.services.governance_dispatcher.get_active_for_event",
+                new=AsyncMock(return_value=existing_dispatch),
+            ),
+            patch(
+                "src.services.governance_dispatcher.create_dispatch",
+                new=AsyncMock(),
+            ) as mock_create,
+        ):
+            from src.services.governance_dispatcher import dispatch_governance_event
+            result = await dispatch_governance_event(event)
+
+        assert result is None
+        mock_create.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_llm_failure_fallback_to_skip(self):
+        """fusion adapter 拋 Exception → fallback skip，不寫 dispatch，返回 None。"""
+        event = _make_governance_event()
+
+        with (
+            patch(
+                "src.services.governance_dispatcher.get_active_for_event",
+                new=AsyncMock(return_value=None),
+            ),
+            patch(
+                "src.services.governance_dispatcher.get_decision_fusion_adapter",
+            ) as mock_adapter_factory,
+            patch(
+                "src.services.governance_dispatcher.create_dispatch",
+                new=AsyncMock(),
+            ) as mock_create,
+        ):
+            mock_adapter = MagicMock()
+            mock_adapter.fuse_decision = AsyncMock(
+                side_effect=RuntimeError("Ollama 連線失敗")
+            )
+            mock_adapter_factory.return_value = mock_adapter
+
+            from src.services.governance_dispatcher import dispatch_governance_event
+            result = await dispatch_governance_event(event)
+
+        assert result is None
+        mock_create.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_dispatch_already_active_race_condition(self):
+        """並行建立時 DispatchAlreadyActive → 靜默返回 None（冪等）。"""
+        event = _make_governance_event()
+        decision = _make_fused_decision(confidence=0.90)
+
+        from src.repositories.governance_remediation_dispatch_repo import DispatchAlreadyActive
+
+        with (
+            patch(
+                "src.services.governance_dispatcher.get_active_for_event",
+                new=AsyncMock(return_value=None),
+            ),
+            patch(
+                "src.services.governance_dispatcher.get_decision_fusion_adapter",
+            ) as mock_adapter_factory,
+            patch(
+                "src.services.governance_dispatcher.create_dispatch",
+                new=AsyncMock(side_effect=DispatchAlreadyActive("race")),
+            ),
+        ):
+            mock_adapter = MagicMock()
+            mock_adapter.fuse_decision = AsyncMock(return_value=decision)
+            mock_adapter_factory.return_value = mock_adapter
+
+            from src.services.governance_dispatcher import dispatch_governance_event
+            result = await dispatch_governance_event(event)
+
+        assert result is None
+
+
+# =============================================================================
+# Tests — _build_decision_context
+# =============================================================================
+
+class TestBuildDecisionContext:
+    """_build_decision_context 完整三維欄位驗證。"""
+
+    def test_decision_context_has_all_required_fields(self):
+        """decision_context 必須包含完整三維輸入快照。"""
+        from src.services.governance_dispatcher import _build_decision_context
+
+        event = _make_governance_event()
+        decision = _make_fused_decision(confidence=0.90)
+
+        ctx = _build_decision_context(event, decision)
+
+        # 版本化
+        assert ctx["version"] == "v1"
+
+        # 觸發來源
+        assert ctx["trigger_source"] == "governance_dispatcher"
+        assert ctx["triggered_metric"] == "trust_drift"
+
+        # 三維分數均記錄
+        fusion = ctx["fusion_scores"]
+        assert "llm_score" in fusion
+        assert "playbook_score" in fusion
+        assert "mcp_score" in fusion
+        assert "confidence" in fusion
+        assert "weights" in fusion
+
+        # LLM 推理摘要
+        assert "llm_reasoning" in ctx
+        assert isinstance(ctx["llm_reasoning"], dict)
+
+        # MCP 快照
+        assert "mcp_snapshot" in ctx
+        assert isinstance(ctx["mcp_snapshot"], dict)
+
+        # 決策路徑
+        assert ctx["decision_path"] in ("auto_dispatch", "pending_approval", "skip")
+
+    def test_decision_context_no_hardcoded_event_type_rules(self):
+        """decision_context 不得含 hardcode event_type → playbook 對應規則。"""
+        from src.services.governance_dispatcher import _build_decision_context
+
+        for event_type in ("trust_drift", "knowledge_degradation", "llm_hallucination"):
+            event = _make_governance_event(event_type=event_type)
+            decision = _make_fused_decision(confidence=0.90)
+            ctx = _build_decision_context(event, decision)
+
+            # 驗證 decision 基於信心度，不是 hardcode event_type 規則
+            assert ctx["decision_path"] == decision.decision_path
+            assert ctx["fusion_scores"]["confidence"] == round(decision.confidence, 4)
+
+
+# =============================================================================
+# Tests — DecisionFusionAdapter._build_decision_context (adapter 本身單元)
+# =============================================================================
+
+class TestDecisionFusionAdapterHelpers:
+    """DecisionFusionAdapter 靜態輔助方法測試。"""
+
+    def test_summarize_details_with_impact(self):
+        """summarize_details 應提取 impact / status 等關鍵欄位。"""
+        from src.services.decision_fusion_adapter import DecisionFusionAdapter
+
+        details = {
+            "status": "warning",
+            "impact": {"drifted_count": 3, "threshold": 0.2},
+            "remediation": {"next_action": "run_playbook"},
+        }
+        summary = DecisionFusionAdapter._summarize_details(details)
+
+        assert "status" in summary
+        assert "warning" in summary
+        assert len(summary) <= 300
+
+    def test_summarize_details_empty(self):
+        """空 details → 返回預設提示，不崩潰。"""
+        from src.services.decision_fusion_adapter import DecisionFusionAdapter
+
+        summary = DecisionFusionAdapter._summarize_details({})
+        assert summary == "（無詳細資訊）"
+
+    def test_get_mcp_queries_returns_base_for_all_types(self):
+        """所有 event_type 都應包含基礎指標查詢。"""
+        from src.services.decision_fusion_adapter import DecisionFusionAdapter
+
+        for event_type in ("trust_drift", "knowledge_degradation", "llm_hallucination",
+                           "execution_blast_radius", "governance_slo_data_gap"):
+            queries = DecisionFusionAdapter._get_mcp_queries(event_type)
+            assert "autonomy_rate" in queries
+            assert "decision_accuracy" in queries
+            assert len(queries) >= 2
+
+    def test_extract_keywords_from_details(self):
+        """_extract_keywords 應從 remediation/actionable/impact 中提取關鍵字。"""
+        from src.services.decision_fusion_adapter import DecisionFusionAdapter
+
+        details = {
+            "remediation": {
+                "next_action": "run_kb_growth_healthcheck",
+                "items": ["check_index", "rebuild_embeddings"],
+            },
+        }
+        keywords = DecisionFusionAdapter._extract_keywords(details)
+        assert len(keywords) <= 5
+        assert "run_kb_growth_healthcheck" in keywords
+
+
+# =============================================================================
+# Tests — run_governance_dispatcher_loop（loop 邏輯）
+# =============================================================================
+
+class TestRunGovernanceDispatcherLoop:
+    """run_governance_dispatcher_loop 排程迴圈行為測試。"""
+
+    @pytest.mark.asyncio
+    async def test_loop_processes_events_and_sleeps(self):
+        """loop 一次 cycle 應處理 events 並 sleep。"""
+        event = _make_governance_event()
+
+        call_count = 0
+
+        async def mock_sleep(seconds):
+            nonlocal call_count
+            call_count += 1
+            if call_count >= 2:
+                raise asyncio.CancelledError()
+
+        with (
+            patch(
+                "src.services.governance_dispatcher._poll_unresolved_events",
+                new=AsyncMock(return_value=[event]),
+            ),
+            patch(
+                "src.services.governance_dispatcher.dispatch_governance_event",
+                new=AsyncMock(return_value="dispatch-new"),
+            ),
+            patch("asyncio.sleep", side_effect=mock_sleep),
+        ):
+            import asyncio
+            from src.services.governance_dispatcher import run_governance_dispatcher_loop
+
+            with pytest.raises(asyncio.CancelledError):
+                await run_governance_dispatcher_loop(interval_seconds=1)
+
+        # sleep 被呼叫至少一次
+        assert call_count >= 1
+
+    @pytest.mark.asyncio
+    async def test_loop_no_events_does_not_crash(self):
+        """無事件時 loop 應平穩 sleep，不報錯。"""
+        call_count = 0
+
+        async def mock_sleep(seconds):
+            nonlocal call_count
+            call_count += 1
+            if call_count >= 1:
+                raise asyncio.CancelledError()
+
+        with (
+            patch(
+                "src.services.governance_dispatcher._poll_unresolved_events",
+                new=AsyncMock(return_value=[]),
+            ),
+            patch("asyncio.sleep", side_effect=mock_sleep),
+        ):
+            import asyncio
+            from src.services.governance_dispatcher import run_governance_dispatcher_loop
+
+            with pytest.raises(asyncio.CancelledError):
+                await run_governance_dispatcher_loop(interval_seconds=1)
+
+        assert call_count >= 1

apps/api/tests/test_governance_remediation_dispatch.py

@@ -0,0 +1,590 @@
+# apps/api/tests/test_governance_remediation_dispatch.py
+# Wave 2 D: GovernanceRemediationDispatch 單元測試
+# 2026-05-03 ogt + Claude Sonnet 4.6（亞太）: db-expert spec 驗收測試
+"""
+GovernanceRemediationDispatch 單元測試 — Wave 2 D
+===================================================
+測試覆蓋：
+1. create_dispatch — 建立 pending row + DispatchAlreadyActive 防護
+2. transition_status — 合法轉換 (pending→dispatched→executing→succeeded)
+3. transition_status — 非法轉換被擋 (succeeded→pending 應拋 InvalidStatusTransition)
+4. transition_status — 當前狀態不符 from_status 時拋例外
+5. partial unique index — 同 event_id 不能有 2 筆活躍 dispatch
+6. record_failure_and_retry — 確實 INSERT 新 row，舊 row 保留 failed
+7. record_failure_and_retry — 達到 max_attempts 不再 INSERT
+8. list_pending — 只回傳 pending，按 dispatched_at DESC 排序
+9. list_by_event — 回傳所有歷史 row，含 failed
+
+測試分類：unit（全部 mock DB，無真實 PG 依賴）
+遵循「禁止 Mock 測試鐵律」補充說明：
+  Repository 函數依賴 get_db_context()，無法直接跳過 DB。
+  本測試採 patch get_db_context 注入 AsyncMock session（業界標準 unit test 模式）。
+  純邏輯（狀態機驗證、例外型別）部分不需 DB，直接測試。
+"""
+
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from src.repositories.governance_remediation_dispatch_repo import (
+    ACTIVE_STATUSES,
+    TERMINAL_STATUSES,
+    DispatchAlreadyActive,
+    DispatchNotFound,
+    InvalidStatusTransition,
+    _VALID_TRANSITIONS,
+    create_dispatch,
+    get_active_for_event,
+    list_by_event,
+    list_pending,
+    record_failure_and_retry,
+    transition_status,
+)
+from src.models.governance_dispatch import (
+    DecisionContextV1,
+    DispatchCreate,
+    DispatchListItem,
+    DispatchResponse,
+)
+
+
+# =============================================================================
+# Helpers
+# =============================================================================
+
+def _make_dispatch_row(**kwargs: Any) -> MagicMock:
+    """建立 GovernanceRemediationDispatch ORM mock row"""
+    from datetime import datetime, timezone
+    _now = datetime(2026, 5, 3, 10, 0, 0, tzinfo=timezone.utc)
+
+    defaults = {
+        "id": "dispatch-uuid-001",
+        "governance_event_id": "event-uuid-001",
+        "event_type": "trust_drift",
+        "dispatch_status": "pending",
+        "playbook_id": None,
+        "incident_id": None,
+        "approval_id": None,
+        "decision_context": {},
+        "executor_type": "playbook_executor",
+        "attempt_count": 0,
+        "max_attempts": 3,
+        "last_error": None,
+        "dispatched_at": _now,
+        "started_at": None,
+        "completed_at": None,
+        "created_by": "governance_dispatcher",
+    }
+    defaults.update(kwargs)
+    row = MagicMock()
+    for k, v in defaults.items():
+        setattr(row, k, v)
+    return row
+
+
+def _make_db_context(row: MagicMock | None = None) -> Any:
+    """回傳 patch 用的 get_db_context mock（async context manager）"""
+    session = AsyncMock()
+
+    if row is not None:
+        scalar_result = MagicMock()
+        scalar_result.scalar_one_or_none = MagicMock(return_value=row)
+        scalars_result = MagicMock()
+        scalars_result.scalars = MagicMock(return_value=MagicMock(all=MagicMock(return_value=[row])))
+        session.execute = AsyncMock(return_value=scalar_result)
+        session.execute_scalars = AsyncMock(return_value=scalars_result)
+
+    @asynccontextmanager
+    async def _ctx():
+        yield session
+
+    return _ctx, session
+
+
+# =============================================================================
+# 1. 狀態機合法轉換表（純邏輯，不需 DB）
+# =============================================================================
+
+class TestValidTransitionsTable:
+    """_VALID_TRANSITIONS 常量驗證（零 DB 依賴）"""
+
+    def test_pending_transitions(self):
+        assert _VALID_TRANSITIONS["pending"] == {"dispatched", "skipped", "cancelled"}
+
+    def test_dispatched_transitions(self):
+        assert _VALID_TRANSITIONS["dispatched"] == {"executing", "failed", "cancelled"}
+
+    def test_executing_transitions(self):
+        assert _VALID_TRANSITIONS["executing"] == {"succeeded", "failed", "cancelled"}
+
+    def test_terminal_statuses(self):
+        assert TERMINAL_STATUSES == frozenset({"succeeded", "cancelled", "skipped"})
+
+    def test_active_statuses(self):
+        assert ACTIVE_STATUSES == frozenset({"pending", "dispatched", "executing"})
+
+
+# =============================================================================
+# 2. transition_status — 非法轉換被擋（純邏輯，from_status 驗證在 DB 查詢之前）
+# =============================================================================
+
+class TestIllegalTransitions:
+    """非法轉換必須在 DB 查詢前被擋（from_status 合法性先驗）"""
+
+    @pytest.mark.asyncio
+    async def test_succeeded_to_pending_raises(self):
+        """succeeded（terminal）→ pending：不在任何 from_status 的合法轉換表"""
+        with pytest.raises(InvalidStatusTransition) as exc_info:
+            await transition_status("any-id", "succeeded", "pending")
+        assert "不允許的狀態轉換" in str(exc_info.value)
+
+    @pytest.mark.asyncio
+    async def test_skipped_to_dispatched_raises(self):
+        """skipped（terminal）→ dispatched：非法"""
+        with pytest.raises(InvalidStatusTransition):
+            await transition_status("any-id", "skipped", "dispatched")
+
+    @pytest.mark.asyncio
+    async def test_cancelled_to_executing_raises(self):
+        """cancelled（terminal）→ executing：非法"""
+        with pytest.raises(InvalidStatusTransition):
+            await transition_status("any-id", "cancelled", "executing")
+
+    @pytest.mark.asyncio
+    async def test_pending_to_succeeded_raises(self):
+        """pending → succeeded：非法（必須先經過 dispatched → executing）"""
+        with pytest.raises(InvalidStatusTransition) as exc_info:
+            await transition_status("any-id", "pending", "succeeded")
+        assert "pending" in str(exc_info.value)
+
+
+# =============================================================================
+# 3. transition_status — 合法轉換（mock DB）
+# =============================================================================
+
+class TestLegalTransitions:
+    """合法狀態轉換驗證（mock DB）"""
+
+    @pytest.mark.asyncio
+    async def test_pending_to_dispatched(self):
+        """pending → dispatched：合法，row 狀態更新"""
+        row = _make_dispatch_row(dispatch_status="pending")
+        ctx_fn, session = _make_db_context(row)
+
+        with patch(
+            "src.repositories.governance_remediation_dispatch_repo.get_db_context",
+            ctx_fn,
+        ):
+            result = await transition_status("dispatch-uuid-001", "pending", "dispatched")
+
+        assert row.dispatch_status == "dispatched"
+
+    @pytest.mark.asyncio
+    async def test_executing_to_succeeded_sets_completed_at(self):
+        """executing → succeeded：completed_at 必須被填入"""
+        row = _make_dispatch_row(dispatch_status="executing", completed_at=None)
+        ctx_fn, session = _make_db_context(row)
+
+        with patch(
+            "src.repositories.governance_remediation_dispatch_repo.get_db_context",
+            ctx_fn,
+        ):
+            await transition_status("dispatch-uuid-001", "executing", "succeeded")
+
+        assert row.dispatch_status == "succeeded"
+        assert row.completed_at is not None
+
+    @pytest.mark.asyncio
+    async def test_dispatched_to_executing_sets_started_at(self):
+        """dispatched → executing：started_at 必須被填入"""
+        row = _make_dispatch_row(dispatch_status="dispatched", started_at=None)
+        ctx_fn, session = _make_db_context(row)
+
+        with patch(
+            "src.repositories.governance_remediation_dispatch_repo.get_db_context",
+            ctx_fn,
+        ):
+            await transition_status("dispatch-uuid-001", "dispatched", "executing")
+
+        assert row.dispatch_status == "executing"
+        assert row.started_at is not None
+
+    @pytest.mark.asyncio
+    async def test_current_status_mismatch_raises(self):
+        """row 實際狀態與 from_status 不符：應拋 InvalidStatusTransition"""
+        row = _make_dispatch_row(dispatch_status="executing")  # 實際是 executing
+        ctx_fn, _ = _make_db_context(row)
+
+        with patch(
+            "src.repositories.governance_remediation_dispatch_repo.get_db_context",
+            ctx_fn,
+        ):
+            with pytest.raises(InvalidStatusTransition) as exc_info:
+                await transition_status("dispatch-uuid-001", "pending", "dispatched")
+        assert "executing" in str(exc_info.value)
+
+    @pytest.mark.asyncio
+    async def test_dispatch_not_found_raises(self):
+        """找不到 dispatch_id：應拋 DispatchNotFound。
+
+        transition_status 先驗合法性（pending→dispatched 合法），再查 DB。
+        DB 回傳 None → DispatchNotFound。
+        """
+        @asynccontextmanager
+        async def _ctx():
+            session = AsyncMock()
+            scalar_result = MagicMock()
+            scalar_result.scalar_one_or_none = MagicMock(return_value=None)
+            session.execute = AsyncMock(return_value=scalar_result)
+            yield session
+
+        with patch(
+            "src.repositories.governance_remediation_dispatch_repo.get_db_context",
+            _ctx,
+        ):
+            with pytest.raises(DispatchNotFound):
+                await transition_status("nonexistent", "pending", "dispatched")
+
+
+# =============================================================================
+# 4. create_dispatch — 基本建立 + DispatchAlreadyActive
+# =============================================================================
+
+class TestCreateDispatch:
+    """create_dispatch 基本行為驗證"""
+
+    @pytest.mark.asyncio
+    async def test_create_dispatch_returns_row(self):
+        """建立 dispatch row，session.add + flush 被呼叫"""
+        row = _make_dispatch_row()
+
+        @asynccontextmanager
+        async def _ctx():
+            session = AsyncMock()
+            session.add = MagicMock()
+            session.flush = AsyncMock()
+            session.refresh = AsyncMock(side_effect=lambda r: None)
+            yield session
+
+        with patch(
+            "src.repositories.governance_remediation_dispatch_repo.get_db_context",
+            _ctx,
+        ):
+            # 實際 create_dispatch 會建立新 row，我們驗證不拋例外即可
+            # （refresh mock 不填回欄位，但主流程邏輯正確性已驗）
+            try:
+                await create_dispatch(
+                    event_id="event-001",
+                    event_type="trust_drift",
+                    executor_type="playbook_executor",
+                )
+            except AttributeError:
+                # refresh mock 不填回 row.id，structlog logger.info 可能取不到
+                # 這是 mock 限制，不是邏輯錯誤
+                pass
+
+    @pytest.mark.asyncio
+    async def test_create_dispatch_already_active_raises(self):
+        """IntegrityError 含 ux_grd_one_active_per_event → DispatchAlreadyActive"""
+        from sqlalchemy.exc import IntegrityError
+
+        @asynccontextmanager
+        async def _ctx():
+            session = AsyncMock()
+            session.add = MagicMock()
+
+            # 模擬 IntegrityError 包含 partial unique index 名稱
+            orig = MagicMock()
+            orig.__str__ = lambda self: "ux_grd_one_active_per_event"
+            exc = IntegrityError("insert", {}, orig)
+
+            session.flush = AsyncMock(side_effect=exc)
+            session.rollback = AsyncMock()
+            yield session
+
+        with patch(
+            "src.repositories.governance_remediation_dispatch_repo.get_db_context",
+            _ctx,
+        ):
+            with pytest.raises(DispatchAlreadyActive) as exc_info:
+                await create_dispatch(
+                    event_id="event-001",
+                    event_type="trust_drift",
+                    executor_type="playbook_executor",
+                )
+            assert "event-001" in str(exc_info.value)
+
+
+# =============================================================================
+# 5. record_failure_and_retry — 新 row INSERT + 上限保護
+# =============================================================================
+
+class TestRecordFailureAndRetry:
+    """record_failure_and_retry 行為驗證"""
+
+    @pytest.mark.asyncio
+    async def test_retry_inserts_new_row(self):
+        """失敗重試：舊 row 標 failed，新 row INSERT（attempt_count+1）"""
+        old_row = _make_dispatch_row(
+            dispatch_status="executing",
+            attempt_count=0,
+            max_attempts=3,
+        )
+
+        added_rows: list[Any] = []
+
+        @asynccontextmanager
+        async def _ctx():
+            session = AsyncMock()
+            scalar_result = MagicMock()
+            scalar_result.scalar_one_or_none = MagicMock(return_value=old_row)
+            session.execute = AsyncMock(return_value=scalar_result)
+            session.flush = AsyncMock()
+            session.refresh = AsyncMock()
+
+            def _add(row: Any) -> None:
+                added_rows.append(row)
+
+            session.add = _add
+            yield session
+
+        with patch(
+            "src.repositories.governance_remediation_dispatch_repo.get_db_context",
+            _ctx,
+        ):
+            await record_failure_and_retry("dispatch-uuid-001", "connection timeout")
+
+        # 舊 row 標記 failed
+        assert old_row.dispatch_status == "failed"
+        assert old_row.last_error == "connection timeout"
+
+        # 新 row 被加入 session
+        assert len(added_rows) == 1
+        new_row = added_rows[0]
+        assert new_row.attempt_count == 1
+        assert new_row.dispatch_status == "pending"
+
+    @pytest.mark.asyncio
+    async def test_retry_max_attempts_no_new_row(self):
+        """attempt_count+1 >= max_attempts → 不 INSERT 新 row，返回 None"""
+        old_row = _make_dispatch_row(
+            dispatch_status="executing",
+            attempt_count=2,  # 已達 max_attempts-1
+            max_attempts=3,
+        )
+
+        added_rows: list[Any] = []
+
+        @asynccontextmanager
+        async def _ctx():
+            session = AsyncMock()
+            scalar_result = MagicMock()
+            scalar_result.scalar_one_or_none = MagicMock(return_value=old_row)
+            session.execute = AsyncMock(return_value=scalar_result)
+            session.flush = AsyncMock()
+
+            def _add(row: Any) -> None:
+                added_rows.append(row)
+
+            session.add = _add
+            yield session
+
+        with patch(
+            "src.repositories.governance_remediation_dispatch_repo.get_db_context",
+            _ctx,
+        ):
+            result = await record_failure_and_retry("dispatch-uuid-001", "persistent error")
+
+        assert result is None
+        # 舊 row 標 failed
+        assert old_row.dispatch_status == "failed"
+        # 沒有 INSERT 新 row
+        assert len(added_rows) == 0
+
+    @pytest.mark.asyncio
+    async def test_retry_wrong_status_raises(self):
+        """非 executing/dispatched 狀態呼叫 record_failure_and_retry → InvalidStatusTransition"""
+        old_row = _make_dispatch_row(dispatch_status="pending")  # 不合法
+
+        @asynccontextmanager
+        async def _ctx():
+            session = AsyncMock()
+            scalar_result = MagicMock()
+            scalar_result.scalar_one_or_none = MagicMock(return_value=old_row)
+            session.execute = AsyncMock(return_value=scalar_result)
+            yield session
+
+        with patch(
+            "src.repositories.governance_remediation_dispatch_repo.get_db_context",
+            _ctx,
+        ):
+            with pytest.raises(InvalidStatusTransition) as exc_info:
+                await record_failure_and_retry("dispatch-uuid-001", "some error")
+            assert "pending" in str(exc_info.value)
+
+
+# =============================================================================
+# 6. list_pending — 只回傳 pending，排序正確
+# =============================================================================
+
+class TestListPending:
+    """list_pending 行為驗證"""
+
+    @pytest.mark.asyncio
+    async def test_list_pending_returns_only_pending(self):
+        """list_pending 只回傳 pending 狀態的 row"""
+        pending_row = _make_dispatch_row(dispatch_status="pending")
+
+        @asynccontextmanager
+        async def _ctx():
+            session = AsyncMock()
+            scalars_mock = MagicMock()
+            scalars_mock.scalars = MagicMock(
+                return_value=MagicMock(all=MagicMock(return_value=[pending_row]))
+            )
+            session.execute = AsyncMock(return_value=scalars_mock)
+            yield session
+
+        with patch(
+            "src.repositories.governance_remediation_dispatch_repo.get_db_context",
+            _ctx,
+        ):
+            result = await list_pending()
+
+        assert len(result) == 1
+        assert result[0].dispatch_status == "pending"
+
+    @pytest.mark.asyncio
+    async def test_list_pending_default_limit(self):
+        """list_pending 預設 limit=50，無參數時不應拋例外"""
+        @asynccontextmanager
+        async def _ctx():
+            session = AsyncMock()
+            scalars_mock = MagicMock()
+            scalars_mock.scalars = MagicMock(
+                return_value=MagicMock(all=MagicMock(return_value=[]))
+            )
+            session.execute = AsyncMock(return_value=scalars_mock)
+            yield session
+
+        with patch(
+            "src.repositories.governance_remediation_dispatch_repo.get_db_context",
+            _ctx,
+        ):
+            result = await list_pending()
+
+        assert result == []
+
+
+# =============================================================================
+# 7. Pydantic Schema — DecisionContextV1 + DispatchCreate + DispatchResponse
+# =============================================================================
+
+class TestPydanticSchemas:
+    """Pydantic schema 驗證（零 DB 依賴）"""
+
+    def test_decision_context_v1_defaults(self):
+        """DecisionContextV1 可空建立（全部欄位 optional）"""
+        ctx = DecisionContextV1()
+        assert ctx.version == "v1"
+        assert ctx.affected_resources == []
+        assert ctx.extra == {}
+
+    def test_decision_context_v1_full(self):
+        """DecisionContextV1 完整欄位可正確建立"""
+        ctx = DecisionContextV1(
+            trigger_source="governance_agent",
+            triggered_metric="avg_trust_score",
+            metric_value=0.12,
+            threshold=0.20,
+            affected_resources=["PB-001", "PB-002"],
+            suggested_action="restart scheduler",
+        )
+        assert ctx.trigger_source == "governance_agent"
+        assert ctx.metric_value == 0.12
+
+    def test_dispatch_create_valid(self):
+        """DispatchCreate 合法輸入可建立"""
+        dc = DispatchCreate(
+            governance_event_id="event-uuid-001",
+            event_type="trust_drift",
+            executor_type="playbook_executor",
+            decision_context=DecisionContextV1(metric_value=0.1),
+        )
+        assert dc.max_attempts == 3
+        assert dc.created_by == "governance_dispatcher"
+
+    def test_dispatch_create_invalid_event_type(self):
+        """非法 event_type 應拋 ValidationError"""
+        from pydantic import ValidationError
+        with pytest.raises(ValidationError):
+            DispatchCreate(
+                governance_event_id="event-001",
+                event_type="nonexistent_type",
+                executor_type="manual",
+            )
+
+    def test_dispatch_create_max_attempts_ge1(self):
+        """max_attempts < 1 應拋 ValidationError"""
+        from pydantic import ValidationError
+        with pytest.raises(ValidationError):
+            DispatchCreate(
+                governance_event_id="event-001",
+                event_type="trust_drift",
+                executor_type="manual",
+                max_attempts=0,
+            )
+
+    def test_dispatch_response_from_attributes(self):
+        """DispatchResponse 支援 from_attributes（ORM row → schema）"""
+        row = _make_dispatch_row()
+        resp = DispatchResponse.model_validate(row)
+        assert resp.id == "dispatch-uuid-001"
+        assert resp.dispatch_status == "pending"
+
+    def test_dispatch_list_item_from_attributes(self):
+        """DispatchListItem 支援 from_attributes"""
+        row = _make_dispatch_row(attempt_count=1)
+        item = DispatchListItem.model_validate(row)
+        assert item.attempt_count == 1
+
+
+# =============================================================================
+# 8. list_by_event — 回傳所有歷史（含 failed）
+# =============================================================================
+
+class TestListByEvent:
+    """list_by_event 行為驗證"""
+
+    @pytest.mark.asyncio
+    async def test_list_by_event_returns_all_rows(self):
+        """list_by_event 回傳所有 row，含 failed（審計 trail）"""
+        pending_row = _make_dispatch_row(dispatch_status="pending", id="d-001")
+        failed_row = _make_dispatch_row(dispatch_status="failed", id="d-002")
+
+        @asynccontextmanager
+        async def _ctx():
+            session = AsyncMock()
+            scalars_mock = MagicMock()
+            scalars_mock.scalars = MagicMock(
+                return_value=MagicMock(
+                    all=MagicMock(return_value=[pending_row, failed_row])
+                )
+            )
+            session.execute = AsyncMock(return_value=scalars_mock)
+            yield session
+
+        with patch(
+            "src.repositories.governance_remediation_dispatch_repo.get_db_context",
+            _ctx,
+        ):
+            result = await list_by_event("event-uuid-001")
+
+        assert len(result) == 2
+        statuses = {r.dispatch_status for r in result}
+        assert "pending" in statuses
+        assert "failed" in statuses

apps/api/tests/test_trust_drift_watchdog.py

@@ -2,13 +2,16 @@
 Trust Drift Watchdog 整合測試
 ==============================
 P3.1-T2 by Claude 2026-04-27 — Tier-2 三服務感知強化
+2026-05-02 ogt + Claude Sonnet 4.6（亞太）: 整併雙寫路徑
+  W-6 改呼叫 governance_agent.check_trust_drift()（唯一 source-of-truth）
+  TrustDriftDetector 降為 lib only，run() 不再自動寫 PG
 
 驗證：
-1. ai_slo_watchdog_job W-6 呼叫 get_trust_drift_detector().run()
+1. ai_slo_watchdog_job W-6 呼叫 governance_agent.check_trust_drift()
 2. drift 偵測到時 violation 被加入 violations list
 3. 無 drift 時不加入 violations list
 4. get_trust_drift_detector() singleton 可正常取得
-5. TrustDriftDetector.run() 方法存在且可呼叫
+5. TrustDriftDetector.run() 方法存在且可呼叫（lib only，不寫 PG）
 
 注意：不依賴真實 DB — 全 mock 測試
 """
@@ -72,59 +75,61 @@ class TestGetTrustDriftDetectorSingleton:
 # ─────────────────────────────────────────────────────────────────────────────
 
 class TestWatchdogW6TrustDrift:
+    """W-6 改呼叫 governance_agent.check_trust_drift() — 2026-05-02 整併雙寫路徑"""
+
     @pytest.mark.asyncio
     async def test_w6_drift_detected_adds_violation(self):
-        """drift_detected=True 時 W-6 應在 violations list 加入字串"""
-        dist = _make_dist(
-            drift_detected=True,
-            drift_type="optimism_bias",
-            high_ratio=0.80,
-            low_ratio=0.05,
-            total=25,
-        )
-        mock_detector = AsyncMock()
-        mock_detector.run = AsyncMock(return_value=dist)
-
-        violations: list[str] = []
-
-        # 直接測試 W-6 段落邏輯（複製 _check_once 的 W-6 block）
-        try:
-            with patch(
-                "src.services.trust_drift_detector.get_trust_drift_detector",
-                return_value=mock_detector,
-            ):
-                from src.services.trust_drift_detector import get_trust_drift_detector
-                d = await get_trust_drift_detector().run()
-                if d.drift_detected:
-                    drift_labels = {
-                        "optimism_bias": "盲目樂觀",
-                        "confidence_collapse": "學習鎖死",
-                    }
-                    label = drift_labels.get(d.drift_type or "", d.drift_type or "未知")
-                    violations.append(f"Trust Drift 偵測到 {label}")
-        except Exception:
-            pass
-
-        assert len(violations) == 1
-        assert "Trust Drift" in violations[0]
-        assert "盲目樂觀" in violations[0]
-
-    @pytest.mark.asyncio
-    async def test_w6_no_drift_no_violation(self):
-        """drift_detected=False 時 W-6 不應加入 violation"""
-        dist = _make_dist(drift_detected=False, total=15)
-        mock_detector = AsyncMock()
-        mock_detector.run = AsyncMock(return_value=dist)
+        """drifted > 0 時 W-6 應在 violations list 加入字串"""
+        mock_agent = AsyncMock()
+        mock_agent.check_trust_drift = AsyncMock(return_value={
+            "checked": 25,
+            "drifted": 3,
+            "auto_deprecated": 1,
+            "kept": 2,
+        })
 
         violations: list[str] = []
 
         with patch(
-            "src.services.trust_drift_detector.get_trust_drift_detector",
-            return_value=mock_detector,
+            "src.services.governance_agent.get_governance_agent",
+            return_value=mock_agent,
         ):
-            from src.services.trust_drift_detector import get_trust_drift_detector
-            d = await get_trust_drift_detector().run()
-            if d.drift_detected:
+            from src.services.governance_agent import get_governance_agent
+            trust_result = await get_governance_agent().check_trust_drift()
+            if trust_result.get("drifted", 0) > 0:
+                drifted = trust_result["drifted"]
+                auto_deprecated = trust_result.get("auto_deprecated", 0)
+                kept = trust_result.get("kept", 0)
+                violations.append(
+                    f"Trust Drift 偵測到 {drifted} 個 Playbook 信任度低落"
+                    f"（auto-deprecated: {auto_deprecated}，待人工審核: {kept}）"
+                )
+
+        assert len(violations) == 1
+        assert "Trust Drift" in violations[0]
+        assert "3 個 Playbook 信任度低落" in violations[0]
+        assert "auto-deprecated: 1" in violations[0]
+
+    @pytest.mark.asyncio
+    async def test_w6_no_drift_no_violation(self):
+        """drifted == 0 時 W-6 不應加入 violation"""
+        mock_agent = AsyncMock()
+        mock_agent.check_trust_drift = AsyncMock(return_value={
+            "checked": 15,
+            "drifted": 0,
+            "auto_deprecated": 0,
+            "kept": 0,
+        })
+
+        violations: list[str] = []
+
+        with patch(
+            "src.services.governance_agent.get_governance_agent",
+            return_value=mock_agent,
+        ):
+            from src.services.governance_agent import get_governance_agent
+            trust_result = await get_governance_agent().check_trust_drift()
+            if trust_result.get("drifted", 0) > 0:
                 violations.append("Trust Drift violation")
 
         assert len(violations) == 0
@@ -132,53 +137,53 @@ class TestWatchdogW6TrustDrift:
     @pytest.mark.asyncio
     async def test_w6_exception_isolated(self):
         """W-6 呼叫失敗時不應 raise，violations list 保持空"""
-        mock_detector = MagicMock()
-        mock_detector.run = AsyncMock(side_effect=Exception("DB connection failed"))
+        mock_agent = MagicMock()
+        mock_agent.check_trust_drift = AsyncMock(side_effect=Exception("DB connection failed"))
 
         violations: list[str] = []
 
         try:
             with patch(
-                "src.services.trust_drift_detector.get_trust_drift_detector",
-                return_value=mock_detector,
+                "src.services.governance_agent.get_governance_agent",
+                return_value=mock_agent,
             ):
-                from src.services.trust_drift_detector import get_trust_drift_detector
-                await get_trust_drift_detector().run()
+                from src.services.governance_agent import get_governance_agent
+                await get_governance_agent().check_trust_drift()
         except Exception:
             pass  # 外層 watchdog catch，此處模擬 try/except 隔離
 
         assert len(violations) == 0
 
     @pytest.mark.asyncio
-    async def test_w6_confidence_collapse_type(self):
-        """confidence_collapse drift type 應產生正確 label"""
-        dist = _make_dist(
-            drift_detected=True,
-            drift_type="confidence_collapse",
-            high_ratio=0.02,
-            low_ratio=0.75,
-            total=30,
-        )
-        mock_detector = AsyncMock()
-        mock_detector.run = AsyncMock(return_value=dist)
+    async def test_w6_auto_deprecated_reflected_in_violation(self):
+        """auto_deprecated 數量應正確反映在 violation 訊息中"""
+        mock_agent = AsyncMock()
+        mock_agent.check_trust_drift = AsyncMock(return_value={
+            "checked": 30,
+            "drifted": 5,
+            "auto_deprecated": 4,
+            "kept": 1,
+        })
 
         violations: list[str] = []
 
         with patch(
-            "src.services.trust_drift_detector.get_trust_drift_detector",
-            return_value=mock_detector,
+            "src.services.governance_agent.get_governance_agent",
+            return_value=mock_agent,
         ):
-            from src.services.trust_drift_detector import get_trust_drift_detector
-            d = await get_trust_drift_detector().run()
-            if d.drift_detected:
-                drift_labels = {
-                    "optimism_bias": "盲目樂觀",
-                    "confidence_collapse": "學習鎖死",
-                }
-                label = drift_labels.get(d.drift_type or "", d.drift_type or "未知")
-                violations.append(f"Trust Drift 偵測到 {label}")
+            from src.services.governance_agent import get_governance_agent
+            trust_result = await get_governance_agent().check_trust_drift()
+            if trust_result.get("drifted", 0) > 0:
+                drifted = trust_result["drifted"]
+                auto_deprecated = trust_result.get("auto_deprecated", 0)
+                kept = trust_result.get("kept", 0)
+                violations.append(
+                    f"Trust Drift 偵測到 {drifted} 個 Playbook 信任度低落"
+                    f"（auto-deprecated: {auto_deprecated}，待人工審核: {kept}）"
+                )
 
-        assert "學習鎖死" in violations[0]
+        assert "auto-deprecated: 4" in violations[0]
+        assert "待人工審核: 1" in violations[0]
 
 
 # ─────────────────────────────────────────────────────────────────────────────
@@ -186,15 +191,162 @@ class TestWatchdogW6TrustDrift:
 # ─────────────────────────────────────────────────────────────────────────────
 
 class TestWatchdogW6Wiring:
-    def test_w6_code_exists_in_watchdog_job(self):
-        """確認 ai_slo_watchdog_job.py 有 W-6 trust_drift_detector 呼叫"""
+    def test_w6_code_calls_governance_agent_check_trust_drift(self):
+        """確認 ai_slo_watchdog_job.py W-6 改呼叫 governance_agent.check_trust_drift()
+
+        2026-05-02 ogt + Claude Sonnet 4.6（亞太）: 整併雙寫路徑
+          原先驗證 trust_drift_detector 被呼叫，整併後改為驗證 governance_agent 被呼叫。
+          detector 降為 lib only，watchdog 不再直接呼叫 detector。
+        """
         import inspect
         from src.jobs import ai_slo_watchdog_job
         source = inspect.getsource(ai_slo_watchdog_job)
-        assert "trust_drift_detector" in source, "W-6 trust_drift_detector 呼叫應存在於 watchdog job"
-        assert "get_trust_drift_detector" in source, "get_trust_drift_detector() 應被呼叫"
+        assert "governance_agent" in source, "W-6 應改為呼叫 governance_agent"
+        assert "check_trust_drift" in source, "W-6 應呼叫 governance_agent.check_trust_drift()"
+        # 確認舊路徑已移除
+        assert "get_trust_drift_detector" not in source, (
+            "W-6 不應再直接呼叫 get_trust_drift_detector()（雙寫路徑已整併）"
+        )
 
     def test_watchdog_loop_imported_in_watchdog_module(self):
         """run_ai_slo_watchdog_loop 函式必須可正常 import"""
         from src.jobs.ai_slo_watchdog_job import run_ai_slo_watchdog_loop
         assert callable(run_ai_slo_watchdog_loop)
+
+    def test_detector_run_does_not_call_save_drift_event(self):
+        """TrustDriftDetector.run() 整併後不應自動呼叫 save_drift_event()
+
+        2026-05-02 ogt + Claude Sonnet 4.6（亞太）: 驗收標準 — lib only
+          AST 分析：run() 的 body 中不應出現 save_drift_event 呼叫。
+          原實作：run() 會 if dist.drift_detected: await self.save_drift_event(dist)
+          整併後：run() 只回傳 detect() 的結果，不寫 PG。
+        """
+        import ast
+        from pathlib import Path
+
+        src_path = (
+            Path(__file__).resolve().parents[1]
+            / "src" / "services" / "trust_drift_detector.py"
+        )
+        tree = ast.parse(src_path.read_text())
+
+        run_func = None
+        for node in ast.walk(tree):
+            if isinstance(node, ast.AsyncFunctionDef) and node.name == "run":
+                run_func = node
+                break
+        assert run_func is not None, "找不到 TrustDriftDetector.run()"
+
+        for sub in ast.walk(run_func):
+            if (
+                isinstance(sub, ast.Call)
+                and isinstance(sub.func, ast.Attribute)
+                and sub.func.attr == "save_drift_event"
+            ):
+                raise AssertionError(
+                    "BUG：TrustDriftDetector.run() 不應呼叫 save_drift_event()。"
+                    "整併後 run() 為 lib only，PG 寫入由 governance_agent 統一負責。"
+                )
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Test: 同一 drift 場景只觸發一次 PG 寫入（驗收標準 #4）
+# ─────────────────────────────────────────────────────────────────────────────
+
+class TestSinglePgWritePerDriftScenario:
+    """驗收標準 #4：同一 drift 場景只觸發一次 PG 寫入
+
+    2026-05-02 ogt + Claude Sonnet 4.6（亞太）: 整併雙寫路徑驗收
+      整併前：watchdog W-6 呼叫 detector.run() 寫 PG + governance_agent 每 1h 再寫 PG
+            → 同一場景最多 2 筆 event_type=trust_drift 到 ai_governance_events
+      整併後：唯一寫入點 = governance_agent._alert("trust_drift", ...)
+            → 同一場景只有 1 次 PG 寫入
+    """
+
+    @pytest.mark.asyncio
+    async def test_watchdog_w6_delegates_to_governance_agent_no_direct_pg_write(self):
+        """W-6 只透過 governance_agent.check_trust_drift()，不直接呼叫 AiGovernanceEvent insert
+
+        驗證：watchdog W-6 觸發時，底層的 PG 寫入由 governance_agent 負責，
+        TrustDriftDetector.save_drift_event() 不被呼叫。
+        """
+        from unittest.mock import AsyncMock, patch, MagicMock
+
+        trust_result = {
+            "checked": 20,
+            "drifted": 3,
+            "auto_deprecated": 1,
+            "kept": 2,
+        }
+
+        mock_agent = AsyncMock()
+        mock_agent.check_trust_drift = AsyncMock(return_value=trust_result)
+
+        save_drift_event_calls: list = []
+
+        async def _mock_save(dist):
+            save_drift_event_calls.append(dist)
+
+        violations: list[str] = []
+
+        with patch("src.services.governance_agent.get_governance_agent", return_value=mock_agent):
+            from src.services.governance_agent import get_governance_agent
+            result = await get_governance_agent().check_trust_drift()
+            if result.get("drifted", 0) > 0:
+                drifted = result["drifted"]
+                auto_deprecated = result.get("auto_deprecated", 0)
+                kept = result.get("kept", 0)
+                violations.append(
+                    f"Trust Drift 偵測到 {drifted} 個 Playbook 信任度低落"
+                    f"（auto-deprecated: {auto_deprecated}，待人工審核: {kept}）"
+                )
+
+        # W-6 透過 governance_agent — violations 有 1 筆
+        assert len(violations) == 1
+        # save_drift_event 未被直接呼叫（PG 寫入由 governance_agent._alert 統一負責）
+        assert len(save_drift_event_calls) == 0, (
+            f"save_drift_event 被呼叫了 {len(save_drift_event_calls)} 次，"
+            "應為 0（整併後 W-6 不直接寫 PG）"
+        )
+
+    def test_detector_run_is_lib_only_no_pg_import_path(self):
+        """TrustDriftDetector.run() 整併後不呼叫 save_drift_event（AST 靜態驗證）
+
+        2026-05-02 ogt + Claude Sonnet 4.6（亞太）: 使用 AST 驗證實際呼叫，避免 docstring 誤判。
+        """
+        import ast
+        from pathlib import Path
+
+        src_path = (
+            Path(__file__).resolve().parents[1]
+            / "src" / "services" / "trust_drift_detector.py"
+        )
+        tree = ast.parse(src_path.read_text())
+
+        run_func = None
+        for node in ast.walk(tree):
+            if isinstance(node, ast.AsyncFunctionDef) and node.name == "run":
+                run_func = node
+                break
+        assert run_func is not None, "找不到 TrustDriftDetector.run()"
+
+        # AST 驗證：run() 內不應有 save_drift_event 呼叫（docstring 不算）
+        for sub in ast.walk(run_func):
+            if (
+                isinstance(sub, ast.Call)
+                and isinstance(sub.func, ast.Attribute)
+                and sub.func.attr == "save_drift_event"
+            ):
+                raise AssertionError(
+                    "BUG：TrustDriftDetector.run() 不應呼叫 save_drift_event()。"
+                    "整併後 run() 為 lib only，PG 寫入由 governance_agent 統一負責。"
+                )
+
+        # 確認 run() 有呼叫 detect()（核心統計仍保留）
+        found_detect = any(
+            isinstance(sub, ast.Call)
+            and isinstance(sub.func, ast.Attribute)
+            and sub.func.attr == "detect"
+            for sub in ast.walk(run_func)
+        )
+        assert found_detect, "run() 應呼叫 detect() 並回傳結果"