feat(p11): RAG 自主學習 + Promotion Gate 4 階段護欄（feature flag OFF）

Operation Ollama-First v5.0 / Phase 11 / RAG 自主學習迴圈

services/rag_service.py (532 行)
- RAGService.query() — bge-m3 embed + cosine 0.85 threshold + top_k=5
- get_embedding_signature() — v5.0 護欄 #3 一致性檢查 (SHA1[:12])
- fire-and-forget rag_query_log INSERT (不阻塞主流程)
- feedback() — Telegram 👍/👎 寫回 feedback_score
- RAG_ENABLED 預設 OFF（戰前行為不變）

services/learning_pipeline.py (750 行)
- Distiller — 純 Hermes 規則引擎，零 LLM 成本
  Quality 規則：MCP >200 字 0.8 / LLM JSON ok 0.9 / TextRank 0.6 / 👍 1.0 / 👎 0.0
- PromotionGate — Owen v5.0 護欄 #1 鐵律
  Stage 1: quality_score >= 0.7
  Stage 2: 無幻覺檢測（規則引擎，零 LLM）
  Stage 3: 與既有 insight 相似度 < 0.95（Stage 3 在 episode embed 後啟用）
  Stage 4: weight >= 0.8 必經 Telegram 👍/👎
- expire_stale_reviews() — 24h 無回應自動降級 weight=0.5
- hash_human_approver — Telegram username SHA1[:8] PII 保護

services/hermes_analyst_service.py — 新增 analyze() RAG-first
- RAG hit → return synthesize（不燒 LLM）
- RAG miss → 既有 LLM 路徑 + enqueue learning_episodes

services/openclaw_strategist_service.py — Q&A 入口接 RAG-first
- 不動週/月/年報（敘事報告 RAG hit 機率低）

services/telegram_templates.py
- rag_feedback_keyboard() — 👍/👎 inline keyboard
- promotion_review_keyboard() — Stage 4 人工驗收按鈕

routes/openclaw_bot_routes.py — 3 組 callback handler
- rag_fb:{id}:{score} → rag_service.feedback()
- pg_ok:{episode_id} → PromotionGate.promote()
- pg_no:{episode_id} → PromotionGate.reject()

70 unit tests 全綠 + 全戰役 196 tests zero regression（4:17 跑完）

剩餘 limitations（Phase 12+ 補）：
1. learning_episodes.embedding 寫入路徑（Stage 3 dedup 暫 skip）
2. PromotionGate worker cron 未掛
3. Telegram awaiting_review 推播未接（callback handler 已就位）

灰度開啟條件（建議 1 週後）：
- ANTHROPIC_API_KEY 設定 + RAG_ENABLED=true + threshold=0.90 保守
- feedback_score >= 4 比率 > 70% → threshold 降至 0.85

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

tests/test_learning_pipeline.py

@@ -0,0 +1,274 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+tests/test_learning_pipeline.py
+Operation Ollama-First v5.0 / Phase 11 — Distiller + LearningPipeline 單元測試
+
+涵蓋：
+  - Distiller 各 quality_score 規則（mcp / llm_response / user_feedback / manual_curated）
+  - LearningPipeline.enqueue() DB 寫入路徑
+  - expire_stale_reviews() 24h 自動降級
+  - hash_human_approver() PII 保護
+"""
+
+from __future__ import annotations
+
+import json
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Distiller 各規則
+# ─────────────────────────────────────────────────────────────────────────────
+class TestDistillerMcpResult:
+    def test_long_with_keywords_high_quality(self):
+        from services.learning_pipeline import Distiller
+        d = Distiller()
+        text = "本週業績分析顯示，建議聚焦保濕品類。" + "詳細說明 " * 80  # > 200 字
+        result = d.distill(episode_type='mcp_result', raw_content=text)
+        assert result is not None
+        assert result.quality_score == 0.8
+        assert result.episode_type == 'mcp_result'
+
+    def test_long_no_keywords_medium_quality(self):
+        from services.learning_pipeline import Distiller
+        d = Distiller()
+        text = "啦啦啦" * 100  # > 200 字但無關鍵字
+        result = d.distill(episode_type='mcp_result', raw_content=text)
+        assert result.quality_score == 0.65
+
+    def test_short_low_quality(self):
+        from services.learning_pipeline import Distiller
+        d = Distiller()
+        text = "短內容"
+        result = d.distill(episode_type='mcp_result', raw_content=text)
+        assert result.quality_score == 0.5
+
+    def test_empty_returns_none(self):
+        from services.learning_pipeline import Distiller
+        d = Distiller()
+        assert d.distill(episode_type='mcp_result', raw_content='') is None
+        assert d.distill(episode_type='mcp_result', raw_content='   ') is None
+
+
+class TestDistillerLlmResponse:
+    def test_json_structured_high_quality(self):
+        from services.learning_pipeline import Distiller
+        d = Distiller()
+        text = json.dumps({"status": "ok", "summary": "本週重點"})
+        result = d.distill(episode_type='llm_response', raw_content=text)
+        assert result.quality_score == 0.9
+
+    def test_json_array_non_empty_high(self):
+        from services.learning_pipeline import Distiller
+        d = Distiller()
+        text = json.dumps([{"sku": "A001", "risk": "HIGH"}])
+        result = d.distill(episode_type='llm_response', raw_content=text)
+        assert result.quality_score == 0.9
+
+    def test_json_dict_no_status_lower(self):
+        from services.learning_pipeline import Distiller
+        d = Distiller()
+        text = json.dumps({"some_field": "value"})
+        result = d.distill(episode_type='llm_response', raw_content=text)
+        # dict 非空 → 0.9 (status_ok 條件含 "len(obj)>0")
+        assert result.quality_score == 0.9
+
+    def test_free_text_long_with_numbers(self):
+        from services.learning_pipeline import Distiller
+        d = Distiller()
+        text = "本週業績漲了 15.3%。" + "詳細說明 " * 100  # > 500 字 + 數字
+        result = d.distill(episode_type='llm_response', raw_content=text)
+        assert result.quality_score == 0.65
+
+    def test_free_text_long_no_numbers(self):
+        from services.learning_pipeline import Distiller
+        d = Distiller()
+        text = "本週業績趨勢上升。" + "詳細說明 " * 100  # > 500 字無數字
+        result = d.distill(episode_type='llm_response', raw_content=text)
+        assert result.quality_score == 0.55
+
+    def test_free_text_short_below_quality_gate(self):
+        from services.learning_pipeline import Distiller
+        d = Distiller()
+        text = "本週業績有變化"  # 短文本
+        result = d.distill(episode_type='llm_response', raw_content=text)
+        # 0.4 → Stage 1 會 reject
+        assert result.quality_score == 0.4
+
+
+class TestDistillerUserFeedback:
+    def test_score_5_high_quality(self):
+        from services.learning_pipeline import Distiller
+        d = Distiller()
+        result = d.distill(
+            episode_type='user_feedback',
+            raw_content='這個建議幫我增加了 8% 銷量',
+            user_feedback_score=5,
+        )
+        assert result.quality_score == 1.0
+        assert result.weight == 0.9  # 高權重 → Stage 4 人工驗收
+
+    def test_score_1_negative_sample(self):
+        from services.learning_pipeline import Distiller
+        d = Distiller()
+        result = d.distill(
+            episode_type='user_feedback',
+            raw_content='完全沒幫助',
+            user_feedback_score=1,
+        )
+        assert result.quality_score == 0.0  # Stage 1 reject
+
+    def test_default_score_3_mid(self):
+        from services.learning_pipeline import Distiller
+        d = Distiller()
+        result = d.distill(
+            episode_type='user_feedback',
+            raw_content='普通',
+            user_feedback_score=None,
+        )
+        # 預設 3 → (3-1)/4 = 0.5
+        assert result.quality_score == 0.5
+
+
+class TestDistillerManualCurated:
+    def test_max_quality_and_weight(self):
+        from services.learning_pipeline import Distiller
+        d = Distiller()
+        result = d.distill(episode_type='manual_curated', raw_content='手動入庫')
+        assert result.quality_score == 1.0
+        assert result.weight == 1.0
+
+
+class TestDistillerInvalidType:
+    def test_unknown_type_returns_none(self):
+        from services.learning_pipeline import Distiller
+        d = Distiller()
+        result = d.distill(episode_type='garbage', raw_content='whatever')
+        assert result is None
+
+
+class TestDistillerLengthGuard:
+    def test_distilled_text_truncated_to_16kb(self):
+        from services.learning_pipeline import Distiller, DISTILLED_TEXT_MAX_BYTES
+        d = Distiller()
+        text = '建議分析 ' * 5000  # 遠超 16KB
+        result = d.distill(episode_type='mcp_result', raw_content=text)
+        encoded = result.distilled_text.encode('utf-8')
+        assert len(encoded) <= DISTILLED_TEXT_MAX_BYTES
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# LearningPipeline.enqueue
+# ─────────────────────────────────────────────────────────────────────────────
+class TestLearningPipelineEnqueue:
+    def test_enqueue_returns_id_on_success(self, monkeypatch):
+        from services.learning_pipeline import learning_pipeline
+
+        fake_session = MagicMock()
+        fake_row = MagicMock()
+        fake_row.__getitem__.return_value = 42
+        fake_session.execute.return_value.fetchone.return_value = fake_row
+        monkeypatch.setattr('database.manager.get_session', lambda: fake_session)
+
+        new_id = learning_pipeline.enqueue(
+            episode_type='manual_curated',
+            raw_content='手動入庫測試內容',
+        )
+        assert new_id == 42
+        fake_session.commit.assert_called_once()
+
+    def test_enqueue_returns_none_when_distill_fails(self):
+        from services.learning_pipeline import learning_pipeline
+        # 空內容 → distill 回 None → enqueue 回 None
+        result = learning_pipeline.enqueue(
+            episode_type='mcp_result',
+            raw_content='',
+        )
+        assert result is None
+
+    def test_enqueue_db_failure_returns_none(self, monkeypatch):
+        from services.learning_pipeline import learning_pipeline
+
+        fake_session = MagicMock()
+        fake_session.execute.side_effect = RuntimeError("db down")
+        monkeypatch.setattr('database.manager.get_session', lambda: fake_session)
+
+        result = learning_pipeline.enqueue(
+            episode_type='manual_curated',
+            raw_content='測試內容',
+        )
+        assert result is None
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# expire_stale_reviews
+# ─────────────────────────────────────────────────────────────────────────────
+class TestExpireStaleReviews:
+    def test_expire_uses_correct_sql(self, monkeypatch):
+        from services.learning_pipeline import expire_stale_reviews
+
+        fake_session = MagicMock()
+        fake_result = MagicMock()
+        fake_result.rowcount = 3
+        fake_session.execute.return_value = fake_result
+        monkeypatch.setattr('database.manager.get_session', lambda: fake_session)
+
+        count = expire_stale_reviews(hours=24)
+        assert count == 3
+        # 確認 commit 跑了
+        fake_session.commit.assert_called_once()
+
+    def test_expire_db_failure_returns_zero(self, monkeypatch):
+        from services.learning_pipeline import expire_stale_reviews
+
+        fake_session = MagicMock()
+        fake_session.execute.side_effect = RuntimeError("db down")
+        monkeypatch.setattr('database.manager.get_session', lambda: fake_session)
+
+        count = expire_stale_reviews(hours=24)
+        assert count == 0
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# hash_human_approver
+# ─────────────────────────────────────────────────────────────────────────────
+class TestHashHumanApprover:
+    def test_returns_8_char_hex(self):
+        from services.learning_pipeline import hash_human_approver
+        h = hash_human_approver('owen.tsai')
+        assert len(h) == 8
+        assert all(c in '0123456789abcdef' for c in h)
+
+    def test_empty_returns_empty(self):
+        from services.learning_pipeline import hash_human_approver
+        assert hash_human_approver('') == ''
+        assert hash_human_approver(None) == ''  # type: ignore
+
+    def test_deterministic(self):
+        from services.learning_pipeline import hash_human_approver
+        a = hash_human_approver('alice')
+        b = hash_human_approver('alice')
+        c = hash_human_approver('bob')
+        assert a == b
+        assert a != c
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 工具函式：_detect_simple_contradiction
+# ─────────────────────────────────────────────────────────────────────────────
+class TestContradictionDetector:
+    def test_no_contradiction_returns_none(self):
+        from services.learning_pipeline import _detect_simple_contradiction
+        text = "業績是上升。市場是競爭。"
+        # subject=業績→上升, subject=市場→競爭，沒矛盾
+        assert _detect_simple_contradiction(text) is None
+
+    def test_contradiction_detected(self):
+        from services.learning_pipeline import _detect_simple_contradiction
+        text = "A是黑色。A是白色。"
+        result = _detect_simple_contradiction(text)
+        assert result is not None
+        assert 'A' in result

tests/test_promotion_gate.py

@@ -0,0 +1,390 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+tests/test_promotion_gate.py
+Operation Ollama-First v5.0 / Phase 11 — PromotionGate 4 階段晉升閘單元測試
+
+涵蓋：
+  Stage 1: quality_score < 0.7 → rejected_quality
+  Stage 2: 規則引擎幻覺檢測（hedge words / 矛盾）
+  Stage 3: cosine similarity >= 0.95 → rejected_duplicate
+  Stage 4: weight >= 0.8 強制 awaiting_review（不能跳）
+  promote() / reject() / mark_awaiting_review() DB 操作
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pytest
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 共用工具
+# ─────────────────────────────────────────────────────────────────────────────
+def _fake_episode(
+    id_=1, episode_type='llm_response',
+    distilled_text='內容', quality_score=0.8, weight=0.5,
+    embedding=None, status='pending',
+):
+    return {
+        'id': id_,
+        'episode_type': episode_type,
+        'distilled_text': distilled_text,
+        'quality_score': quality_score,
+        'weight': weight,
+        'embedding': embedding,
+        'promotion_status': status,
+    }
+
+
+def _patch_load_episode(monkeypatch, episode):
+    """讓 PromotionGate._load_episode 直接回 episode（不走 DB）。"""
+    from services.learning_pipeline import PromotionGate
+    monkeypatch.setattr(
+        PromotionGate, '_load_episode',
+        staticmethod(lambda episode_id: episode if episode else None),
+    )
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Stage 1: quality_score
+# ─────────────────────────────────────────────────────────────────────────────
+class TestStage1Quality:
+    def test_low_quality_rejected(self, monkeypatch):
+        from services.learning_pipeline import PromotionGate
+        ep = _fake_episode(quality_score=0.5)
+        _patch_load_episode(monkeypatch, ep)
+
+        gate = PromotionGate()
+        decision = gate.can_promote(1)
+        assert decision.can_promote is False
+        assert decision.reason == 'rejected_quality'
+        assert '0.500' in (decision.detail or '')
+
+    def test_quality_at_threshold_passes(self, monkeypatch):
+        from services.learning_pipeline import PromotionGate, STAGE_1_AUTO_QUALITY
+        ep = _fake_episode(quality_score=STAGE_1_AUTO_QUALITY, weight=0.5)
+        _patch_load_episode(monkeypatch, ep)
+
+        gate = PromotionGate()
+        decision = gate.can_promote(1)
+        # 過 Stage 1 + 2 + 3 + 4 自動晉升
+        assert decision.can_promote is True
+        assert decision.reason == 'approved'
+
+    def test_episode_not_found(self, monkeypatch):
+        from services.learning_pipeline import PromotionGate
+        _patch_load_episode(monkeypatch, None)
+
+        gate = PromotionGate()
+        decision = gate.can_promote(99999)
+        assert decision.can_promote is False
+        assert decision.reason == 'rejected_quality'
+        assert 'not found' in (decision.detail or '')
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Stage 2: 幻覺檢測
+# ─────────────────────────────────────────────────────────────────────────────
+class TestStage2Hallucination:
+    def test_hedge_words_without_numbers_rejected(self, monkeypatch):
+        from services.learning_pipeline import PromotionGate
+        ep = _fake_episode(
+            quality_score=0.8,
+            distilled_text='我猜本週業績可能會有點成長吧，也許不錯。',
+        )
+        _patch_load_episode(monkeypatch, ep)
+
+        gate = PromotionGate()
+        decision = gate.can_promote(1)
+        assert decision.can_promote is False
+        assert decision.reason == 'rejected_hallucination'
+
+    def test_hedge_words_with_numbers_passes(self, monkeypatch):
+        from services.learning_pipeline import PromotionGate
+        ep = _fake_episode(
+            quality_score=0.8, weight=0.5,
+            distilled_text='我猜本週業績會漲 5.2%，根據過去 30 天平均。',
+        )
+        _patch_load_episode(monkeypatch, ep)
+
+        gate = PromotionGate()
+        decision = gate.can_promote(1)
+        # 通過 Stage 2（有具體數字）→ 進到 Stage 4 → approved
+        assert decision.can_promote is True
+
+    def test_contradiction_rejected(self, monkeypatch):
+        from services.learning_pipeline import PromotionGate
+        ep = _fake_episode(
+            quality_score=0.8,
+            distilled_text='A是黑色。A是白色。',
+        )
+        _patch_load_episode(monkeypatch, ep)
+
+        gate = PromotionGate()
+        decision = gate.can_promote(1)
+        assert decision.can_promote is False
+        assert decision.reason == 'rejected_hallucination'
+        assert '自相矛盾' in (decision.detail or '')
+
+    def test_clean_text_passes(self, monkeypatch):
+        from services.learning_pipeline import PromotionGate
+        ep = _fake_episode(
+            quality_score=0.8, weight=0.5,
+            distilled_text='本週業績漲 5.2%，建議聚焦保濕品類。',
+        )
+        _patch_load_episode(monkeypatch, ep)
+
+        gate = PromotionGate()
+        decision = gate.can_promote(1)
+        assert decision.can_promote is True
+        assert decision.reason == 'approved'
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Stage 3: 去重
+# ─────────────────────────────────────────────────────────────────────────────
+class TestStage3Dedup:
+    def test_high_similarity_rejected(self, monkeypatch):
+        from services.learning_pipeline import PromotionGate, STAGE_3_DEDUP_THRESHOLD
+        ep = _fake_episode(
+            quality_score=0.8, weight=0.5,
+            distilled_text='本週業績漲 5%。',
+            embedding=[0.1] * 1024,  # 模擬非空 embedding
+        )
+        _patch_load_episode(monkeypatch, ep)
+
+        # 模擬 DB 回 similarity=0.96
+        fake_row = MagicMock()
+        fake_row.id = 999
+        fake_row.similarity = 0.96
+        fake_session = MagicMock()
+        fake_session.execute.return_value.fetchone.return_value = fake_row
+        monkeypatch.setattr('database.manager.get_session', lambda: fake_session)
+
+        gate = PromotionGate()
+        decision = gate.can_promote(1)
+        assert decision.can_promote is False
+        assert decision.reason == 'rejected_duplicate'
+        assert decision.similar_insight_id == 999
+
+    def test_low_similarity_passes(self, monkeypatch):
+        from services.learning_pipeline import PromotionGate
+        ep = _fake_episode(
+            quality_score=0.8, weight=0.5,
+            distilled_text='全新內容', embedding=[0.1] * 1024,
+        )
+        _patch_load_episode(monkeypatch, ep)
+
+        fake_row = MagicMock()
+        fake_row.id = 999
+        fake_row.similarity = 0.5
+        fake_session = MagicMock()
+        fake_session.execute.return_value.fetchone.return_value = fake_row
+        monkeypatch.setattr('database.manager.get_session', lambda: fake_session)
+
+        gate = PromotionGate()
+        decision = gate.can_promote(1)
+        assert decision.can_promote is True
+
+    def test_null_embedding_skips_dedup(self, monkeypatch):
+        """蒸餾時尚未 embed → 略過 Stage 3，不阻擋晉升。"""
+        from services.learning_pipeline import PromotionGate
+        ep = _fake_episode(quality_score=0.8, weight=0.5, embedding=None)
+        _patch_load_episode(monkeypatch, ep)
+
+        # DB 不應被呼叫
+        called = {'count': 0}
+
+        def _spy_session():
+            called['count'] += 1
+            return MagicMock()
+
+        monkeypatch.setattr('database.manager.get_session', _spy_session)
+
+        gate = PromotionGate()
+        decision = gate.can_promote(1)
+        assert decision.can_promote is True
+        assert called['count'] == 0
+
+    def test_dedup_query_failure_passes(self, monkeypatch):
+        """DB 查詢失敗 → 視為通過（避免 DB 故障阻塞晉升）。"""
+        from services.learning_pipeline import PromotionGate
+        ep = _fake_episode(
+            quality_score=0.8, weight=0.5,
+            embedding=[0.1] * 1024,
+        )
+        _patch_load_episode(monkeypatch, ep)
+
+        fake_session = MagicMock()
+        fake_session.execute.side_effect = RuntimeError("db down")
+        monkeypatch.setattr('database.manager.get_session', lambda: fake_session)
+
+        gate = PromotionGate()
+        decision = gate.can_promote(1)
+        assert decision.can_promote is True
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Stage 4: 強制人工驗收（v5.0 護欄 #1 核心）
+# ─────────────────────────────────────────────────────────────────────────────
+class TestStage4HumanReview:
+    def test_high_weight_forces_awaiting_review(self, monkeypatch):
+        """weight=0.85 (>=0.8) 必經人工驗收，不能跳 Stage 4。"""
+        from services.learning_pipeline import PromotionGate, STAGE_4_HUMAN_REVIEW_WEIGHT
+        ep = _fake_episode(quality_score=0.9, weight=0.85)
+        _patch_load_episode(monkeypatch, ep)
+
+        gate = PromotionGate()
+        decision = gate.can_promote(1)
+        assert decision.can_promote is False
+        assert decision.reason == 'awaiting_review'
+        assert str(STAGE_4_HUMAN_REVIEW_WEIGHT) in (decision.detail or '')
+
+    def test_high_weight_at_threshold_forces_review(self, monkeypatch):
+        """weight 剛好 0.8 也要進人工驗收（>= not >）。"""
+        from services.learning_pipeline import PromotionGate
+        ep = _fake_episode(quality_score=0.9, weight=0.8)
+        _patch_load_episode(monkeypatch, ep)
+
+        gate = PromotionGate()
+        decision = gate.can_promote(1)
+        assert decision.reason == 'awaiting_review'
+
+    def test_low_weight_auto_promoted(self, monkeypatch):
+        from services.learning_pipeline import PromotionGate
+        ep = _fake_episode(quality_score=0.9, weight=0.79)
+        _patch_load_episode(monkeypatch, ep)
+
+        gate = PromotionGate()
+        decision = gate.can_promote(1)
+        assert decision.can_promote is True
+        assert decision.reason == 'approved'
+
+    def test_high_weight_user_feedback_forces_review(self, monkeypatch):
+        """user_feedback episode_type 預設 weight=0.9 → 必經人工。"""
+        from services.learning_pipeline import PromotionGate
+        ep = _fake_episode(
+            episode_type='user_feedback',
+            quality_score=1.0, weight=0.9,
+        )
+        _patch_load_episode(monkeypatch, ep)
+
+        gate = PromotionGate()
+        decision = gate.can_promote(1)
+        assert decision.reason == 'awaiting_review'
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# promote() DB 操作
+# ─────────────────────────────────────────────────────────────────────────────
+class TestPromote:
+    def test_promote_inserts_ai_insights_and_updates_episode(self, monkeypatch):
+        from services.learning_pipeline import PromotionGate
+        ep = _fake_episode(
+            id_=1, episode_type='llm_response',
+            quality_score=0.9, weight=0.5,
+            distilled_text='本週業績漲 5%',
+        )
+        _patch_load_episode(monkeypatch, ep)
+
+        # 模擬 INSERT RETURNING id = 555
+        fake_row = MagicMock()
+        fake_row.__getitem__.return_value = 555
+        fake_session = MagicMock()
+        fake_session.execute.return_value.fetchone.return_value = fake_row
+        monkeypatch.setattr('database.manager.get_session', lambda: fake_session)
+
+        gate = PromotionGate()
+        insight_id = gate.promote(1)
+        assert insight_id == 555
+        # 檢查 INSERT + UPDATE 各跑一次（execute 至少 2 次）
+        assert fake_session.execute.call_count >= 2
+        fake_session.commit.assert_called_once()
+
+    def test_promote_episode_not_found_returns_none(self, monkeypatch):
+        from services.learning_pipeline import PromotionGate
+        _patch_load_episode(monkeypatch, None)
+        gate = PromotionGate()
+        assert gate.promote(99999) is None
+
+    def test_promote_db_failure_returns_none(self, monkeypatch):
+        from services.learning_pipeline import PromotionGate
+        ep = _fake_episode(quality_score=0.9, weight=0.5)
+        _patch_load_episode(monkeypatch, ep)
+
+        fake_session = MagicMock()
+        fake_session.execute.side_effect = RuntimeError("db down")
+        monkeypatch.setattr('database.manager.get_session', lambda: fake_session)
+
+        gate = PromotionGate()
+        assert gate.promote(1) is None
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# reject() / mark_awaiting_review()
+# ─────────────────────────────────────────────────────────────────────────────
+class TestRejectAndMark:
+    def test_reject_valid_reason(self, monkeypatch):
+        from services.learning_pipeline import PromotionGate
+
+        fake_session = MagicMock()
+        monkeypatch.setattr('database.manager.get_session', lambda: fake_session)
+
+        gate = PromotionGate()
+        ok = gate.reject(1, 'rejected_quality', detail='quality 0.3 < 0.7')
+        assert ok is True
+        fake_session.commit.assert_called_once()
+
+    def test_reject_invalid_reason_returns_false(self):
+        from services.learning_pipeline import PromotionGate
+        gate = PromotionGate()
+        assert gate.reject(1, 'invalid_reason') is False
+
+    def test_reject_db_failure_returns_false(self, monkeypatch):
+        from services.learning_pipeline import PromotionGate
+
+        fake_session = MagicMock()
+        fake_session.execute.side_effect = RuntimeError("db down")
+        monkeypatch.setattr('database.manager.get_session', lambda: fake_session)
+
+        gate = PromotionGate()
+        assert gate.reject(1, 'rejected_quality') is False
+
+    def test_mark_awaiting_review_runs_update(self, monkeypatch):
+        from services.learning_pipeline import PromotionGate
+
+        fake_session = MagicMock()
+        monkeypatch.setattr('database.manager.get_session', lambda: fake_session)
+
+        gate = PromotionGate()
+        ok = gate.mark_awaiting_review(1)
+        assert ok is True
+        fake_session.execute.assert_called_once()
+        fake_session.commit.assert_called_once()
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 完整 4 階段流程串接（高 weight 必經人工）
+# ─────────────────────────────────────────────────────────────────────────────
+class TestEndToEndFlow:
+    def test_user_feedback_high_quality_must_await_review(self, monkeypatch):
+        """v5.0 護欄 #1 核心案例：user_feedback weight=0.9 強制 awaiting_review，
+        即使 quality=1.0 + 無幻覺 + 無重複也不能直接晉升。
+        """
+        from services.learning_pipeline import PromotionGate
+        ep = _fake_episode(
+            episode_type='user_feedback',
+            quality_score=1.0,        # Stage 1 過
+            distilled_text='2026-04-29 業績漲 12%，廣告 ROI 4.2 倍。',  # 有數字 → Stage 2 過
+            embedding=None,           # Stage 3 略過（無 embedding）
+            weight=0.9,               # >= 0.8 → 強制 Stage 4
+        )
+        _patch_load_episode(monkeypatch, ep)
+
+        gate = PromotionGate()
+        decision = gate.can_promote(1)
+        # 鐵律：高權重必經人工
+        assert decision.can_promote is False
+        assert decision.reason == 'awaiting_review'

tests/test_rag_service.py

@@ -0,0 +1,435 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+tests/test_rag_service.py
+Operation Ollama-First v5.0 / Phase 11 — RAGService 單元測試
+
+涵蓋：
+  - feature flag RAG_ENABLED 預設 OFF → skip 不查 DB / 不寫 log
+  - RAG_ENABLED=true 時的 hit / miss 分支
+  - embedding signature 不一致 → log warning + 不採該筆
+  - fire-and-forget rag_query_log 失敗不影響主流程
+  - feedback() / invalidate_by_caller() / get_embedding_signature()
+
+Mock 策略：
+  - ollama_service.generate_embedding 用 monkeypatch
+  - database.manager.get_session 用 MagicMock 回 session（不真的查 DB）
+"""
+
+from __future__ import annotations
+
+import os
+import time
+from typing import Any, Dict, List
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 共用 fixture
+# ─────────────────────────────────────────────────────────────────────────────
+@pytest.fixture(autouse=True)
+def _reset_rag_kill_switch():
+    """每個測試重置 kill-switch，避免互相污染。"""
+    from services.rag_service import _reset_kill_switch
+    _reset_kill_switch()
+    yield
+    _reset_kill_switch()
+
+
+@pytest.fixture
+def rag_disabled(monkeypatch):
+    monkeypatch.setenv('RAG_ENABLED', 'false')
+
+
+@pytest.fixture
+def rag_enabled(monkeypatch):
+    monkeypatch.setenv('RAG_ENABLED', 'true')
+
+
+def _fake_embedding(dim: int = 1024) -> List[float]:
+    """產生穩定的假 embedding（避免每次 random）。"""
+    return [0.01 * (i % 100) for i in range(dim)]
+
+
+def _make_row_obj(**fields):
+    """模擬 SQLAlchemy row（支援屬性與 row.id 風格）。"""
+    obj = MagicMock()
+    for k, v in fields.items():
+        setattr(obj, k, v)
+    return obj
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Test 1: feature flag 預設 OFF → 短路
+# ─────────────────────────────────────────────────────────────────────────────
+class TestFeatureFlagOff:
+    def test_query_skips_when_disabled(self, rag_disabled):
+        """RAG_ENABLED=false → 不查 DB / 不 embed / 回空 hits。"""
+        from services.rag_service import rag_service
+
+        with patch('services.ollama_service.ollama_service.generate_embedding') as mock_embed:
+            result = rag_service.query("本週業績", caller='test_caller')
+
+        assert result.hits == []
+        assert result.has_high_confidence is False
+        mock_embed.assert_not_called()
+
+    def test_synthesize_returns_empty_when_no_hits(self, rag_disabled):
+        from services.rag_service import rag_service
+
+        result = rag_service.query("本週業績", caller='test_caller')
+        assert result.synthesize() == ""
+
+    def test_default_flag_is_off(self, monkeypatch):
+        """RAG_ENABLED 未設 → 預設 OFF。"""
+        monkeypatch.delenv('RAG_ENABLED', raising=False)
+        from services.rag_service import is_rag_enabled
+        assert is_rag_enabled() is False
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Test 2: RAG_ENABLED=true 時的 hit / miss
+# ─────────────────────────────────────────────────────────────────────────────
+class TestRagEnabledHits:
+    def test_high_confidence_hit_returns_synthesized(self, rag_enabled, monkeypatch):
+        """top-1 distance=0.05 → similarity=0.95 >= threshold 0.85 → has_high_confidence。"""
+        from services import rag_service as rs
+
+        monkeypatch.setattr(
+            'services.ollama_service.ollama_service.generate_embedding',
+            lambda text, model="bge-m3:latest": _fake_embedding(),
+        )
+
+        sig = rs.get_embedding_signature()
+        fake_rows = [
+            _make_row_obj(
+                id=101, insight_type='weekly_strategy', period='2026-W17',
+                content='本週業績漲 5%，建議聚焦保濕品類。',
+                embedding_signature=sig, distance=0.05,
+            ),
+            _make_row_obj(
+                id=102, insight_type='weekly_strategy', period='2026-W17',
+                content='競品 PChome 同類降價 8%。',
+                embedding_signature=sig, distance=0.10,
+            ),
+        ]
+        fake_session = MagicMock()
+        fake_session.execute.return_value.fetchall.return_value = fake_rows
+        monkeypatch.setattr('database.manager.get_session', lambda: fake_session)
+
+        result = rs.rag_service.query("本週業績趨勢", caller='openclaw_qa')
+
+        assert len(result.hits) == 2
+        assert result.hits[0]['score'] == pytest.approx(0.95, abs=1e-3)
+        assert result.has_high_confidence is True
+        assert '本週業績漲' in result.synthesize()
+
+    def test_low_confidence_no_hit(self, rag_enabled, monkeypatch):
+        """所有結果 distance>0.15 → similarity<0.85 → SQL 已過濾，回空 → has_high_confidence False。"""
+        from services import rag_service as rs
+
+        monkeypatch.setattr(
+            'services.ollama_service.ollama_service.generate_embedding',
+            lambda text, model="bge-m3:latest": _fake_embedding(),
+        )
+        fake_session = MagicMock()
+        fake_session.execute.return_value.fetchall.return_value = []
+        monkeypatch.setattr('database.manager.get_session', lambda: fake_session)
+
+        result = rs.rag_service.query("非常冷僻的問題", caller='openclaw_qa')
+
+        assert result.hits == []
+        assert result.has_high_confidence is False
+        assert result.synthesize() == ""
+
+    def test_empty_query_short_circuits(self, rag_enabled, monkeypatch):
+        from services import rag_service as rs
+
+        # generate_embedding 不應被呼叫
+        embed_called = {'count': 0}
+
+        def _spy(*a, **kw):
+            embed_called['count'] += 1
+            return _fake_embedding()
+
+        monkeypatch.setattr(
+            'services.ollama_service.ollama_service.generate_embedding', _spy,
+        )
+
+        result = rs.rag_service.query("", caller='openclaw_qa')
+        assert result.hits == []
+        assert embed_called['count'] == 0
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Test 3: embedding signature 不一致 → 不採用
+# ─────────────────────────────────────────────────────────────────────────────
+class TestEmbeddingSignature:
+    def test_signature_format(self):
+        from services.rag_service import get_embedding_signature
+        sig = get_embedding_signature()
+        # 12 碼 hex
+        assert len(sig) == 12
+        assert all(c in '0123456789abcdef' for c in sig)
+
+    def test_signature_changes_with_model(self):
+        from services.rag_service import get_embedding_signature
+        a = get_embedding_signature(model='bge-m3:latest', dim=1024, normalize=True)
+        b = get_embedding_signature(model='bge-m3:v2', dim=1024, normalize=True)
+        c = get_embedding_signature(model='bge-m3:latest', dim=512, normalize=True)
+        d = get_embedding_signature(model='bge-m3:latest', dim=1024, normalize=False)
+        assert a != b and a != c and a != d
+
+    def test_mismatched_signature_rows_skipped(self, rag_enabled, monkeypatch, caplog):
+        """row.embedding_signature != expected → 不採用 + log warning。"""
+        from services import rag_service as rs
+        import logging
+
+        monkeypatch.setattr(
+            'services.ollama_service.ollama_service.generate_embedding',
+            lambda text, model="bge-m3:latest": _fake_embedding(),
+        )
+        sig = rs.get_embedding_signature()
+        fake_rows = [
+            _make_row_obj(
+                id=201, insight_type='x', period=None, content='舊簽名資料',
+                embedding_signature='deadbeef0001',  # 不同
+                distance=0.05,
+            ),
+            _make_row_obj(
+                id=202, insight_type='x', period=None, content='新簽名資料',
+                embedding_signature=sig,
+                distance=0.06,
+            ),
+        ]
+        fake_session = MagicMock()
+        fake_session.execute.return_value.fetchall.return_value = fake_rows
+        monkeypatch.setattr('database.manager.get_session', lambda: fake_session)
+
+        with caplog.at_level(logging.WARNING, logger='services.rag_service'):
+            result = rs.rag_service.query("query", caller='openclaw_qa')
+
+        # 只剩 1 筆（簽名一致的）
+        assert len(result.hits) == 1
+        assert result.hits[0]['id'] == 202
+        # warning 有寫
+        assert any('signature mismatch' in r.message for r in caplog.records)
+
+    def test_null_signature_passes_through(self, rag_enabled, monkeypatch):
+        """既有未回填的 row（signature=NULL）暫時放行避免戰前資料完全失效。"""
+        from services import rag_service as rs
+
+        monkeypatch.setattr(
+            'services.ollama_service.ollama_service.generate_embedding',
+            lambda text, model="bge-m3:latest": _fake_embedding(),
+        )
+        fake_rows = [
+            _make_row_obj(
+                id=301, insight_type='x', period=None, content='legacy data',
+                embedding_signature=None,
+                distance=0.1,
+            ),
+        ]
+        fake_session = MagicMock()
+        fake_session.execute.return_value.fetchall.return_value = fake_rows
+        monkeypatch.setattr('database.manager.get_session', lambda: fake_session)
+
+        result = rs.rag_service.query("query", caller='openclaw_qa')
+        assert len(result.hits) == 1
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Test 4: fire-and-forget log 失敗不影響主流程
+# ─────────────────────────────────────────────────────────────────────────────
+class TestFireAndForgetLog:
+    def test_log_write_failure_does_not_crash(self, rag_enabled, monkeypatch):
+        """rag_query_log INSERT 失敗 → main 仍回 RAGResult。"""
+        from services import rag_service as rs
+
+        monkeypatch.setattr(
+            'services.ollama_service.ollama_service.generate_embedding',
+            lambda text, model="bge-m3:latest": _fake_embedding(),
+        )
+        # SELECT session 正常；INSERT session 故意 raise
+        select_session = MagicMock()
+        select_session.execute.return_value.fetchall.return_value = []
+        insert_session = MagicMock()
+        insert_session.execute.side_effect = RuntimeError("DB unavailable")
+
+        sessions_iter = iter([select_session, insert_session])
+        monkeypatch.setattr(
+            'database.manager.get_session', lambda: next(sessions_iter),
+        )
+
+        # 應不 raise（fire-and-forget thread 內部包 try/except）
+        result = rs.rag_service.query("query", caller='openclaw_qa')
+        assert result.hits == []
+        # 等 daemon thread 跑完
+        time.sleep(0.2)
+
+    def test_embedding_failure_falls_back_to_empty(self, rag_enabled, monkeypatch):
+        """embedding 回 [] → 不查 DB → 回空 hits 給 caller fallback LLM。"""
+        from services import rag_service as rs
+
+        monkeypatch.setattr(
+            'services.ollama_service.ollama_service.generate_embedding',
+            lambda text, model="bge-m3:latest": [],
+        )
+        # DB 不應被呼叫
+        called = {'count': 0}
+
+        def _spy_session():
+            called['count'] += 1
+            return MagicMock()
+
+        monkeypatch.setattr('database.manager.get_session', _spy_session)
+
+        result = rs.rag_service.query("query", caller='openclaw_qa')
+        assert result.hits == []
+        # SELECT 不應發生（log 寫入仍會用 session）
+        # rag_query_log 寫入也許會跑（embedding=None 仍 log），所以 called 可能是 0 或 1
+        time.sleep(0.2)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Test 5: feedback() / invalidate_by_caller()
+# ─────────────────────────────────────────────────────────────────────────────
+class TestFeedback:
+    def test_feedback_writes_score(self, monkeypatch):
+        from services.rag_service import rag_service
+
+        fake_session = MagicMock()
+        monkeypatch.setattr('database.manager.get_session', lambda: fake_session)
+
+        ok = rag_service.feedback(123, 5)
+        assert ok is True
+        fake_session.execute.assert_called_once()
+        fake_session.commit.assert_called_once()
+
+    def test_feedback_clamps_score(self, monkeypatch):
+        """score=99 應被夾擠到 5；score=0 應被夾擠到 1。"""
+        from services.rag_service import rag_service
+
+        captured = {}
+        fake_session = MagicMock()
+
+        def _exec(stmt, params):
+            captured.update(params)
+            return MagicMock()
+
+        fake_session.execute.side_effect = _exec
+        monkeypatch.setattr('database.manager.get_session', lambda: fake_session)
+
+        rag_service.feedback(1, 99)
+        assert captured['score'] == 5
+
+        captured.clear()
+        rag_service.feedback(1, 0)
+        assert captured['score'] == 1
+
+    def test_feedback_invalid_id_returns_false(self):
+        from services.rag_service import rag_service
+        assert rag_service.feedback(0, 5) is False
+        assert rag_service.feedback(None, 5) is False  # type: ignore
+
+    def test_feedback_db_failure_returns_false(self, monkeypatch):
+        from services.rag_service import rag_service
+
+        fake_session = MagicMock()
+        fake_session.execute.side_effect = RuntimeError("db fail")
+        monkeypatch.setattr('database.manager.get_session', lambda: fake_session)
+
+        ok = rag_service.feedback(123, 5)
+        assert ok is False
+
+    def test_invalidate_by_caller_is_noop(self):
+        """v5.0 暫無 cache 層 → no-op，不 raise。"""
+        from services.rag_service import rag_service
+        rag_service.invalidate_by_caller('openclaw_qa')  # 不應 raise
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Test 6: RAGResult 屬性與 synthesize
+# ─────────────────────────────────────────────────────────────────────────────
+class TestRAGResultStructure:
+    def test_synthesize_takes_top_3(self):
+        from services.rag_service import RAGResult
+        result = RAGResult(
+            query='q', embedding_signature='abc',
+            hits=[{'content': f'第 {i} 筆內容'} for i in range(5)],
+        )
+        out = result.synthesize()
+        assert '第 0 筆' in out
+        assert '第 1 筆' in out
+        assert '第 2 筆' in out
+        assert '第 3 筆' not in out
+        assert '第 4 筆' not in out
+
+    def test_has_high_confidence_threshold(self):
+        from services.rag_service import RAGResult
+        # top-1 score=0.86 >= 0.85 → True
+        r1 = RAGResult(
+            query='q', embedding_signature='abc',
+            hits=[{'score': 0.86}], threshold=0.85,
+        )
+        assert r1.has_high_confidence is True
+
+        # top-1 score=0.84 < 0.85 → False
+        r2 = RAGResult(
+            query='q', embedding_signature='abc',
+            hits=[{'score': 0.84}], threshold=0.85,
+        )
+        assert r2.has_high_confidence is False
+
+        # 無 hits → False
+        r3 = RAGResult(query='q', embedding_signature='abc', hits=[])
+        assert r3.has_high_confidence is False
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Test 7: top_k / threshold 護欄夾擠
+# ─────────────────────────────────────────────────────────────────────────────
+class TestParamGuards:
+    def test_top_k_clamped_to_50(self, rag_enabled, monkeypatch):
+        from services import rag_service as rs
+
+        monkeypatch.setattr(
+            'services.ollama_service.ollama_service.generate_embedding',
+            lambda text, model="bge-m3:latest": _fake_embedding(),
+        )
+        fake_session = MagicMock()
+        fake_session.execute.return_value.fetchall.return_value = []
+        captured = {}
+
+        def _exec(stmt, params):
+            captured.update(params)
+            return MagicMock(fetchall=lambda: [])
+
+        fake_session.execute.side_effect = _exec
+        monkeypatch.setattr('database.manager.get_session', lambda: fake_session)
+
+        rs.rag_service.query("query", caller='openclaw_qa', top_k=999)
+        assert captured.get('lim', 0) <= 100  # top_k=50, fetch_limit = 100
+
+    def test_threshold_clamped_to_unit_interval(self, rag_enabled, monkeypatch):
+        from services import rag_service as rs
+
+        monkeypatch.setattr(
+            'services.ollama_service.ollama_service.generate_embedding',
+            lambda text, model="bge-m3:latest": _fake_embedding(),
+        )
+        fake_session = MagicMock()
+        captured = {}
+
+        def _exec(stmt, params):
+            captured.update(params)
+            return MagicMock(fetchall=lambda: [])
+
+        fake_session.execute.side_effect = _exec
+        monkeypatch.setattr('database.manager.get_session', lambda: fake_session)
+
+        rs.rag_service.query("query", caller='openclaw_qa', threshold=2.0)
+        # 2.0 → clamp 1.0；max_distance = 1.0 - 1.0 = 0.0
+        assert captured.get('max_distance', 1.0) == pytest.approx(0.0)