From 2945108f666c6d602d4cafeabf2bb9da301c2a4b Mon Sep 17 00:00:00 2001
From: Andrew <zhhongyu01@gmail.com>
Date: Wed, 9 Sep 2026 20:00:06 +0800
Subject: [PATCH] feat(kb): Enhance knowledge base with new collections and
 search functionality

- Added new configuration for knowledge base root directory in `.env.example` and `settings.py`.
- Implemented `find_products` method in `CoreReadOnlyRepository` for fuzzy product search based on user queries.
- Introduced `search_cs_knowledge` function in `rag_service.py` to facilitate semantic search across new `fin_*` collections.
- Updated document parsing to support Markdown and YAML front-matter for knowledge base entries.
- Created multiple new FAQ and policy documents in the `data/kb_collections` directory to enrich the knowledge base.

This update significantly improves the knowledge retrieval capabilities for customer service interactions, ensuring more relevant and accurate responses.
---
 .env.example                                  |   3 +
 AGENTS.md                                     |   2 +-
 app/config/settings.py                        |   1 +
 app/repository/core_ro.py                     |  18 +++
 app/service/rag_service.py                    |  73 ++++++------
 app/service/visitor_service.py                |  21 +++-
 app/tool/document_parser.py                   |  66 ++++++++++-
 app/tool/embedding_tool.py                    |  23 +++-
 app/tool/milvus_tool.py                       | 104 +++++++++++++++++-
 .../fin_faq_collection/faq-open-account.md    |   8 ++
 .../faq-redeem-settlement.md                  |   8 ++
 .../fin_faq_collection/faq-reset-password.md  |   8 ++
 .../fin_policy_collection/policy-aml-kyc.md   |   7 ++
 .../fin_policy_collection/policy-complaint.md |   7 ++
 .../policy-suitability-match.md               |   7 ++
 .../prod-balanced-r3.md                       |   8 ++
 .../prod-money-market-r1.md                   |   8 ++
 .../prod-tech-growth-r4.md                    |   8 ++
 docs/memory/MEMORY.md                         |  10 +-
 docs/memory/TODO.md                           |  14 ++-
 docs/项目框架设计/客服Agent-合并说明.md       |  17 ++-
 scripts/kb/test_search.py                     |  51 +++++----
 tests/conftest.py                             |  74 +++++++++++--
 tests/test_wave2_core_ro_tool.py              |  11 +-
 tests/test_wave4_e2e.py                       |  87 ++++++++-------
 tests/test_wave5_notes.py                     |  12 +-
 26 files changed, 514 insertions(+), 142 deletions(-)
 create mode 100644 data/kb_collections/fin_faq_collection/faq-open-account.md
 create mode 100644 data/kb_collections/fin_faq_collection/faq-redeem-settlement.md
 create mode 100644 data/kb_collections/fin_faq_collection/faq-reset-password.md
 create mode 100644 data/kb_collections/fin_policy_collection/policy-aml-kyc.md
 create mode 100644 data/kb_collections/fin_policy_collection/policy-complaint.md
 create mode 100644 data/kb_collections/fin_policy_collection/policy-suitability-match.md
 create mode 100644 data/kb_collections/fin_product_collection/prod-balanced-r3.md
 create mode 100644 data/kb_collections/fin_product_collection/prod-money-market-r1.md
 create mode 100644 data/kb_collections/fin_product_collection/prod-tech-growth-r4.md

diff --git a/.env.example b/.env.example
index 31c080b..c297032 100644
--- a/.env.example
+++ b/.env.example
@@ -21,6 +21,9 @@ NEO4J_PASSWORD=
 # 注意：路径含中文会触发 Milvus Lite(faiss) 文件 IO 失败，Windows 中文用户目录请改纯英文绝对路径
 MILVUS_URI=./data/milvus.db
 
+# 客服 KB 第二套（fin_faq / fin_product / fin_policy 原料目录）
+KB_ROOT_DIR=./data/kb_collections
+
 # Ollama Embedding
 OLLAMA_BASE_URL=http://127.0.0.1:11434
 EMBED_MODEL=bge-m3
diff --git a/AGENTS.md b/AGENTS.md
index 0199cc1..a8cf79a 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -38,6 +38,6 @@ app/repository/core_ro.py # Core 只读 + check_suitability（R-02）
 scripts/core/reset.ps1    # 本地灌 Core 模拟库
 ```
 
-**当前分支：** `merger` · **测试基线：** `python -m pytest` → 554 passed, 0 skipped
+**当前分支：** `merger` · **测试基线：** `python -m pytest` → 730 passed, 0 skipped
 
 技术选型硬阀门见 MEMORY 第 3、7 节。Cursor 以 `.cursor/rules/project-memory.mdc` 为准。
diff --git a/app/config/settings.py b/app/config/settings.py
index a3051bf..ebcf0ff 100644
--- a/app/config/settings.py
+++ b/app/config/settings.py
@@ -24,6 +24,7 @@ class Settings(BaseSettings):
     neo4j_password: str = ""
 
     milvus_uri: str = "./data/milvus.db"
+    kb_root_dir: str = "./data/kb_collections"
 
     ollama_base_url: str = "http://127.0.0.1:11434"
     embed_model: str = "bge-m3"
diff --git a/app/repository/core_ro.py b/app/repository/core_ro.py
index 0a451c7..023ab90 100644
--- a/app/repository/core_ro.py
+++ b/app/repository/core_ro.py
@@ -466,6 +466,24 @@ class CoreReadOnlyRepository:
             row = conn.execute(sql, {"pid": product_id}).mappings().first()
             return dict(row) if row else None
 
+    def find_products(self, keyword: str, limit: int = 3) -> list[dict[str, Any]]:
+        """按产品名称模糊匹配在售产品（客服适当性 Tool 用）。"""
+        kw = (keyword or "").strip()
+        if not kw:
+            return []
+        sql = text(
+            """
+            SELECT * FROM core_product
+            WHERE is_open = 1
+              AND (product_name LIKE :pat OR product_id LIKE :pat)
+            ORDER BY product_id
+            LIMIT :lim
+            """
+        )
+        pat = f"%{kw}%"
+        with self._engine.connect() as conn:
+            return [dict(r) for r in conn.execute(sql, {"pat": pat, "lim": limit}).mappings()]
+
     def get_latest_nav(self, product_id: str) -> dict[str, Any] | None:
         sql = text(
             """
diff --git a/app/service/rag_service.py b/app/service/rag_service.py
index 59290b5..5916410 100644
--- a/app/service/rag_service.py
+++ b/app/service/rag_service.py
@@ -18,6 +18,7 @@ from __future__ import annotations
 from typing import Any
 
 from app.service import embedding, milvus_service
+from app.tool.milvus_tool import get_milvus_client
 
 # 对外默认 TopK（对话 Tool 引用；脚本/服务可显式覆盖）
 DEFAULT_TOP_K = 3
@@ -78,44 +79,52 @@ def search_knowledge(
 
 
 # ---------------------------------------------------------------------------
-# 客服线 · 游客/客户 RAG 接缝（映射 T-21 search_knowledge，不另建 collection）
+# 客服线 · fin_* 三库 RAG（build_collections.py · 与 kb_product_rules 并存）
 # ---------------------------------------------------------------------------
 
-_VISITOR_INTENT_DOC_TYPE: dict[str, str | None] = {
-    "product_consult": None,
-    "policy_interpret": "policy",
-    "faq": "faq",
+_CS_INTENT_COLLECTION: dict[str, str] = {
+    "product_consult": "fin_product",
+    "policy_interpret": "fin_policy",
+    "faq": "fin_faq",
 }
 
 
+def search_cs_knowledge(intent: str, query: str, top_k: int = 5) -> tuple[str, list[dict]]:
+    """客服/游客 RAG：按意图选 fin_* collection 语义检索。"""
+    collection = _CS_INTENT_COLLECTION.get(intent)
+    if not collection or not query or not query.strip():
+        return "", []
+    try:
+        vector = _embed(query.strip())
+        client = get_milvus_client()
+        try:
+            hits = client.search(collection, vector, top_k=top_k)
+        finally:
+            client.close()
+    except Exception:
+        return "", []
+    if not hits:
+        return "", []
+    context_parts: list[str] = []
+    sources: list[dict] = []
+    for i, hit in enumerate(hits, 1):
+        chunk_text = hit.get("chunk_text") or ""
+        source_doc = hit.get("source_doc") or hit.get("id") or ""
+        context_parts.append(f"[来源: {source_doc} | 片段 {i}]\n{chunk_text}")
+        sources.append(
+            {
+                "source_doc": source_doc,
+                "chunk_no": hit.get("chunk_no", i),
+                "score": round(float(hit.get("score", 0.0)), 4),
+            }
+        )
+    return "\n\n".join(context_parts), sources
+
+
 class VisitorRagService:
-    """游客/客户 RAG：intent → search_knowledge → 上下文 + 溯源。"""
+    """游客/客户 RAG：intent → fin_* Milvus 检索 + 溯源。"""
 
     def retrieve(self, intent: str, query: str, top_k: int = 5) -> tuple[str, list[dict]]:
-        if intent not in _VISITOR_INTENT_DOC_TYPE:
+        if intent not in _CS_INTENT_COLLECTION:
             return "", []
-        try:
-            payload = search_knowledge(
-                query,
-                doc_type=_VISITOR_INTENT_DOC_TYPE[intent],
-                top_k=top_k,
-            )
-        except Exception:
-            return "", []
-        hits = payload.get("results") or []
-        if not hits:
-            return "", []
-        context_parts: list[str] = []
-        sources: list[dict] = []
-        for i, hit in enumerate(hits, 1):
-            chunk_text = hit.get("chunk_text") or hit.get("text") or ""
-            source_doc = hit.get("source_doc_id") or hit.get("source_doc") or ""
-            context_parts.append(f"[来源: {source_doc} | 片段 {i}]\n{chunk_text}")
-            sources.append(
-                {
-                    "source_doc": source_doc,
-                    "chunk_no": hit.get("chunk_no", i),
-                    "score": round(float(hit.get("score", 0.0)), 4),
-                }
-            )
-        return "\n\n".join(context_parts), sources
+        return search_cs_knowledge(intent, query, top_k=top_k)
diff --git a/app/service/visitor_service.py b/app/service/visitor_service.py
index e562800..65fdda2 100644
--- a/app/service/visitor_service.py
+++ b/app/service/visitor_service.py
@@ -103,12 +103,18 @@ _REJECT_REALTIME_KW = (
 
 
 def recall_memory(state: VisitorState) -> VisitorState:
-    """节点 1：从 Redis 读取两类短期记忆。"""
+    """节点 1：从 Redis 读取两类短期记忆（Redis 不可用则降级为空）。"""
     mem = VisitorMemoryService()
     sid = state["session_id"]
+    try:
+        chitchat_memory = mem.recall(sid, "chitchat")
+        consult_memory = mem.recall(sid, "consult")
+    except Exception:
+        chitchat_memory = []
+        consult_memory = []
     return {
-        "chitchat_memory": mem.recall(sid, "chitchat"),
-        "consult_memory": mem.recall(sid, "consult"),
+        "chitchat_memory": chitchat_memory,
+        "consult_memory": consult_memory,
     }
 
 
@@ -257,7 +263,7 @@ def fallback(state: VisitorState) -> VisitorState:
 
 
 def save_memory(state: VisitorState) -> VisitorState:
-    """节点 8：保存短期记忆（先写 user，再写 assistant）。"""
+    """节点 8：保存短期记忆（先写 user，再写 assistant；Redis 失败不阻断）。"""
     mem = VisitorMemoryService()
     sid = state["session_id"]
     intent = state.get("intent", "fallback")
@@ -268,8 +274,11 @@ def save_memory(state: VisitorState) -> VisitorState:
     else:
         kind = "chitchat"
 
-    mem.append(sid, kind, "user", state["message"])
-    mem.append(sid, kind, "assistant", state["reply"])
+    try:
+        mem.append(sid, kind, "user", state["message"])
+        mem.append(sid, kind, "assistant", state["reply"])
+    except Exception:
+        pass
 
     return {}
 
diff --git a/app/tool/document_parser.py b/app/tool/document_parser.py
index 9d93bcf..11afd0f 100644
--- a/app/tool/document_parser.py
+++ b/app/tool/document_parser.py
@@ -1 +1,65 @@
-"""文档解析：PDF/Word → 文本块（本地 data/kb/，不用 MinIO）。"""
+"""文档解析：Markdown + YAML front-matter → Chunk（客服 fin_* 知识库目录）。"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+FRONT_MATTER_RE = re.compile(r"^---\s*\n(?P<body>.*?)\n---\s*\n", re.DOTALL)
+
+
+@dataclass
+class Chunk:
+    chunk_id: str
+    text: str
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+def _parse_front_matter(text: str) -> tuple[dict[str, str], str]:
+    match = FRONT_MATTER_RE.match(text)
+    if not match:
+        return {}, text.strip()
+    meta: dict[str, str] = {}
+    for line in match.group("body").splitlines():
+        line = line.strip()
+        if not line or line.startswith("#") or ":" not in line:
+            continue
+        key, _, value = line.partition(":")
+        meta[key.strip()] = value.strip()
+    return meta, text[match.end() :].strip()
+
+
+def _slug(value: str, fallback: str) -> str:
+    cleaned = re.sub(r"[^A-Za-z0-9._-]+", "-", (value or fallback).strip())[:48]
+    return cleaned or fallback
+
+
+def _parse_markdown_file(path: Path, chunk_no: int) -> Chunk:
+    raw = path.read_text(encoding="utf-8")
+    meta, body = _parse_front_matter(raw)
+    source_doc = meta.get("source_doc") or path.stem
+    chunk_text = body or meta.get("answer") or meta.get("question") or path.stem
+    text = chunk_text
+    if meta.get("question") and meta.get("answer"):
+        text = f"{meta['question']}\n{meta['answer']}"
+    elif meta.get("question"):
+        text = f"{meta['question']}\n{chunk_text}"
+
+    record_meta = dict(meta)
+    record_meta.setdefault("source_doc", source_doc)
+    record_meta["chunk_no"] = str(chunk_no)
+    record_meta["chunk_text"] = chunk_text
+
+    chunk_id = f"{_slug(source_doc, path.stem)}_{chunk_no}"
+    return Chunk(chunk_id=chunk_id, text=text, metadata=record_meta)
+
+
+def parse_collection_dir(dir_path: Path) -> tuple[list[Chunk], dict[str, Any]]:
+    """解析目录下全部 ``*.md``，每文件 1 块（chunk_no 从 1 递增）。"""
+    if not dir_path.is_dir():
+        return [], {"file_count": 0}
+    files = sorted(dir_path.glob("*.md"))
+    chunks = [_parse_markdown_file(path, i + 1) for i, path in enumerate(files)]
+    return chunks, {"file_count": len(files)}
diff --git a/app/tool/embedding_tool.py b/app/tool/embedding_tool.py
index 6787d69..55a34a7 100644
--- a/app/tool/embedding_tool.py
+++ b/app/tool/embedding_tool.py
@@ -1 +1,22 @@
-"""向量化：Ollama bge-m3，1024 维。"""
+"""向量化：Ollama bge-m3，1024 维（客服 KB 灌库 / fin_* 检索用）。
+
+薄封装 `app.service.embedding`，与 T-21 共用同一 Ollama 配置。
+"""
+
+from __future__ import annotations
+
+from app.service import embedding
+
+
+class Embedder:
+    """批量/单条 embedding（build_collections / test_search 调用面）。"""
+
+    def embed(self, text: str) -> list[float]:
+        return embedding.embed_text(text)
+
+    def embed_batch(self, texts: list[str]) -> list[list[float]]:
+        return embedding.embed_texts(texts)
+
+
+def get_embedder() -> Embedder:
+    return Embedder()
diff --git a/app/tool/milvus_tool.py b/app/tool/milvus_tool.py
index d3f63c8..84470cf 100644
--- a/app/tool/milvus_tool.py
+++ b/app/tool/milvus_tool.py
@@ -1 +1,103 @@
-"""Milvus 封装：Milvus Lite 连接、Collection CRUD、语义检索。"""
+"""Milvus 封装：Milvus Lite 连接、fin_* Collection CRUD、语义检索（客服 KB 第二套）。"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from app.config.settings import settings
+from app.service import milvus_service
+from pymilvus import DataType  # noqa: E402  milvus_service 已做 MILVUS_URI 环境防御
+
+# build_collections.py 与检索共用
+COLLECTION_FIELDS: dict[str, list[str]] = {
+    "fin_faq": ["question", "answer", "category", "source_doc", "chunk_no", "chunk_text"],
+    "fin_product": ["product_name", "risk_level", "doc_type", "source_doc", "chunk_no", "chunk_text"],
+    "fin_policy": ["policy_name", "chapter", "source_doc", "chunk_no", "chunk_text"],
+}
+
+_VARCHAR_MAX = {
+    "question": 512,
+    "answer": 2048,
+    "category": 64,
+    "product_name": 256,
+    "risk_level": 8,
+    "doc_type": 32,
+    "policy_name": 256,
+    "chapter": 128,
+    "source_doc": 128,
+    "chunk_text": 8192,
+}
+
+
+def _schema_for(collection: str, dim: int):
+    fields = COLLECTION_FIELDS[collection]
+    schema = milvus_service.MilvusClient.create_schema(auto_id=False, enable_dynamic_field=False)
+    schema.add_field("id", DataType.VARCHAR, is_primary=True, max_length=64)
+    schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=dim)
+    for name in fields:
+        if name == "chunk_no":
+            schema.add_field(name, DataType.INT64)
+        else:
+            schema.add_field(name, DataType.VARCHAR, max_length=_VARCHAR_MAX.get(name, 256))
+    return schema
+
+
+class MilvusKbClient:
+    """fin_faq / fin_product / fin_policy 三库客户端。"""
+
+    def __init__(self, client: milvus_service.MilvusClient | None = None) -> None:
+        self._client = client or milvus_service.milvus_client()
+
+    def ensure_collection(self, name: str) -> None:
+        if name not in COLLECTION_FIELDS:
+            raise ValueError(f"unknown collection: {name}")
+        dim = settings.embed_dim
+        if self._client.has_collection(name):
+            self._client.load_collection(name)
+            return
+        index = self._client.prepare_index_params()
+        index.add_index(field_name="embedding", index_type="AUTOINDEX", metric_type="COSINE")
+        self._client.create_collection(name, schema=_schema_for(name, dim), index_params=index)
+        self._client.load_collection(name)
+
+    def drop_collection(self, name: str) -> None:
+        if self._client.has_collection(name):
+            self._client.drop_collection(name)
+
+    def insert(self, name: str, records: list[dict[str, Any]]) -> int:
+        if not records:
+            return 0
+        self.ensure_collection(name)
+        self._client.upsert(name, data=records)
+        return len(records)
+
+    def search(self, name: str, vector: list[float], top_k: int = 3) -> list[dict[str, Any]]:
+        if not self._client.has_collection(name):
+            return []
+        self._client.load_collection(name)
+        output_fields = list(COLLECTION_FIELDS[name])
+        results = self._client.search(
+            name,
+            data=[vector],
+            limit=top_k,
+            output_fields=output_fields,
+        )
+        hits = results[0] if results else []
+        out: list[dict[str, Any]] = []
+        for hit in hits:
+            entity = hit.get("entity", {})
+            out.append(
+                {
+                    "id": hit.get("id"),
+                    "score": float(hit.get("distance", 0.0)),
+                    **entity,
+                }
+            )
+        return out
+
+    def close(self) -> None:
+        self._client.close()
+
+
+def get_milvus_client() -> MilvusKbClient:
+    return MilvusKbClient()
diff --git a/data/kb_collections/fin_faq_collection/faq-open-account.md b/data/kb_collections/fin_faq_collection/faq-open-account.md
new file mode 100644
index 0000000..9110e3e
--- /dev/null
+++ b/data/kb_collections/fin_faq_collection/faq-open-account.md
@@ -0,0 +1,8 @@
+---
+question: 开户需要准备哪些材料？
+answer: 需要有效身份证件、本人银行卡，并完成实名认证与风险测评问卷。线上开户还需手机号验证与人脸识别。
+category: 开户
+source_doc: faq-open-account
+---
+
+开户需要有效身份证件、本人银行卡，并完成实名认证与风险测评问卷。
\ No newline at end of file
diff --git a/data/kb_collections/fin_faq_collection/faq-redeem-settlement.md b/data/kb_collections/fin_faq_collection/faq-redeem-settlement.md
new file mode 100644
index 0000000..b412884
--- /dev/null
+++ b/data/kb_collections/fin_faq_collection/faq-redeem-settlement.md
@@ -0,0 +1,8 @@
+---
+question: 赎回资金多久到账？
+answer: 货币基金一般 T+1 个工作日到账；股票型、混合型产品通常为 T+3 至 T+5 个工作日，具体以产品说明书为准。
+category: 交易
+source_doc: faq-redeem-settlement
+---
+
+赎回资金到账时间因产品类型而异，请以产品说明书与交易确认信息为准。
\ No newline at end of file
diff --git a/data/kb_collections/fin_faq_collection/faq-reset-password.md b/data/kb_collections/fin_faq_collection/faq-reset-password.md
new file mode 100644
index 0000000..00277bf
--- /dev/null
+++ b/data/kb_collections/fin_faq_collection/faq-reset-password.md
@@ -0,0 +1,8 @@
+---
+question: App 忘记登录密码怎么办？
+answer: 可在登录页选择「忘记密码」，通过绑定的手机号或邮箱接收验证码后重置。若手机号已变更，请联系客服或到网点办理。
+category: 账户操作
+source_doc: faq-reset-password
+---
+
+可在登录页选择「忘记密码」，通过绑定的手机号或邮箱接收验证码后重置密码。
\ No newline at end of file
diff --git a/data/kb_collections/fin_policy_collection/policy-aml-kyc.md b/data/kb_collections/fin_policy_collection/policy-aml-kyc.md
new file mode 100644
index 0000000..0b0e8cf
--- /dev/null
+++ b/data/kb_collections/fin_policy_collection/policy-aml-kyc.md
@@ -0,0 +1,7 @@
+---
+policy_name: 反洗钱客户身份识别
+chapter: KYC
+source_doc: policy-aml-kyc
+---
+
+反洗钱要求：开立账户或办理一定金额以上交易时，须核实客户身份、受益所有人，并持续监测可疑交易。命中名单或异常模式须按规定上报。
\ No newline at end of file
diff --git a/data/kb_collections/fin_policy_collection/policy-complaint.md b/data/kb_collections/fin_policy_collection/policy-complaint.md
new file mode 100644
index 0000000..6de63b5
--- /dev/null
+++ b/data/kb_collections/fin_policy_collection/policy-complaint.md
@@ -0,0 +1,7 @@
+---
+policy_name: 销售合规与投诉处理
+chapter: 投诉处理
+source_doc: policy-complaint
+---
+
+客户投诉应在受理后按规定时限调查并反馈。涉及销售误导、适当性争议的，须留存录音录像与适当性匹配记录备查。
\ No newline at end of file
diff --git a/data/kb_collections/fin_policy_collection/policy-suitability-match.md b/data/kb_collections/fin_policy_collection/policy-suitability-match.md
new file mode 100644
index 0000000..6d15bf7
--- /dev/null
+++ b/data/kb_collections/fin_policy_collection/policy-suitability-match.md
@@ -0,0 +1,7 @@
+---
+policy_name: 投资者适当性管理指引
+chapter: 适当性匹配
+source_doc: policy-suitability-match
+---
+
+投资者适当性匹配规则：客户风险承受能力等级 C1~C5 须与产品风险等级 R1~R5 相匹配，客户只能购买风险等级不超过自身承受能力的产品。销售机构须做好双录与风险提示。
\ No newline at end of file
diff --git a/data/kb_collections/fin_product_collection/prod-balanced-r3.md b/data/kb_collections/fin_product_collection/prod-balanced-r3.md
new file mode 100644
index 0000000..49a5d11
--- /dev/null
+++ b/data/kb_collections/fin_product_collection/prod-balanced-r3.md
@@ -0,0 +1,8 @@
+---
+product_name: 均衡配置混合
+risk_level: R3
+doc_type: rule
+source_doc: prod-balanced-r3
+---
+
+均衡配置混合（R3）股债均衡，适合稳健型（C3）及以上客户。申购 T 日 15:00 前按当日净值确认，赎回一般 T+3 至 T+4 个工作日到账。
\ No newline at end of file
diff --git a/data/kb_collections/fin_product_collection/prod-money-market-r1.md b/data/kb_collections/fin_product_collection/prod-money-market-r1.md
new file mode 100644
index 0000000..fa9f41f
--- /dev/null
+++ b/data/kb_collections/fin_product_collection/prod-money-market-r1.md
@@ -0,0 +1,8 @@
+---
+product_name: 稳健货币A
+risk_level: R1
+doc_type: prospectus
+source_doc: prod-money-market-r1
+---
+
+稳健货币A（R1）为货币市场型基金，风险等级 R1，流动性较好，适合保守型（C1）及以上客户作为现金管理工具。不保本不保收益。
\ No newline at end of file
diff --git a/data/kb_collections/fin_product_collection/prod-tech-growth-r4.md b/data/kb_collections/fin_product_collection/prod-tech-growth-r4.md
new file mode 100644
index 0000000..4409fd5
--- /dev/null
+++ b/data/kb_collections/fin_product_collection/prod-tech-growth-r4.md
@@ -0,0 +1,8 @@
+---
+product_name: 科技成长主题
+risk_level: R4
+doc_type: prospectus
+source_doc: prod-tech-growth-r4
+---
+
+科技成长主题（R4）为股票型主题基金，聚焦半导体、人工智能、云计算等方向。适合积极型（C4）及以上客户，短期波动较大，建议以 3 年以上资金参与。
\ No newline at end of file
diff --git a/docs/memory/MEMORY.md b/docs/memory/MEMORY.md
index f8153b8..a857acc 100644
--- a/docs/memory/MEMORY.md
+++ b/docs/memory/MEMORY.md
@@ -9,7 +9,7 @@
 
 **项目是什么：** 金融四 Agent（客户财富 / 代理人 / 数据分析 / 风控）共用数据层与合规底座；**不**互调 LLM，跨 Agent 走 L1/L2/L3 画像与预警表。
 
-**当前进度：** 需求与表设计已定 · **风控模块 B1~B9b + C4~C6 + chat B/C + AL-09 合并** · **代销平台 API v0.1 已落地** · **客服 Agent S2 接缝已接线**（visitor 试聊 + customer 并行分流） · **554 passed 0 skipped** · **`web/` 静奢智能 UI 已落地** · **`npm run build/test/lint` 绿（14 例 Vitest）**。**下一步：ChatPanel/SSE · 迁移 SQL CS-C-11 · Wave 1/2/4/5 解禁。**
+**当前进度：** 需求与表设计已定 · **风控模块 B1~B9b + C4~C6 + chat B/C + AL-09 合并** · **代销平台 API v0.1 已落地** · **客服 Agent S2 接缝收尾**（visitor 试聊 + customer 并行分流 · Wave 1~5 测试绿 · CS-C-11 迁移已执行） · **730 passed 0 skipped** · **`web/` 静奢智能 UI 已落地** · **`npm run build/test/lint` 绿（14 例 Vitest）**。**下一步：ChatPanel/SSE · 前端 customer Chat。**
 
 **工作分支：** 团队开发在 **`merger`**（已 merge 风控模块）；历史开发分支 `risk-control-agent` 交付冻结。旧文档中「待 AL-09 合并」口径已过时。
 
@@ -41,7 +41,7 @@
 | `scripts/core/*.sql` + `reset.ps1` | **已实现** | Core 模拟库 DDL + 种子 |
 | `scripts/agent/` `scripts/demo/` `scripts/dev/` | **已实现** | AML 名单种子 + 风控演示数据 + `fix_utf8_seed.py`/`run_sql_file.py`（Windows UTF-8 灌库）+ issue_dev_token |
 | `scripts/sync/*.py` | **已实现** | 归属同步 + Neo4j 全图 |
-| `tests/` | **已实现** | 37+ 测试模块 **554 用例 0 skipped**（含 `test_wave3_customer_service`；sqlite + 真 MySQL 集成；`test_module_boundary` 宿主 D 类排除）。改路由必同步 `tests/test_main.py::test_all_routers_mounted` |
+| `tests/` | **已实现** | 37+ 测试模块 **730 用例 0 skipped**（含客服 Wave 1~5；sqlite + 真 MySQL 集成；`test_module_boundary` 宿主 D 类排除）。改路由必同步 `tests/test_main.py::test_all_routers_mounted` |
 | `docs/需求拆解/` | 已定 | 场景 P0、矩阵、合规原文 |
 | `docs/PRD/PRD-风控监测Agent.md` | **已冻结（v1.1）** | 风控 PRD v1.0 + v1.1 追加 FR-8/9/10（§4A）+ 规则表附录 |
 | `docs/项目框架设计/实现方案-风控追加需求v1.1-C4C6.md` | **已定稿** | C4~C6 编码依据（经独立 AI 评审修订闭环）；分支/进度速览另见项目根 `交接文档.md` |
@@ -65,7 +65,7 @@
 5. mysql … < scripts/agent/seed-aml-list.sql       # AML 名单（Windows 乱码：`python scripts/dev/fix_utf8_seed.py`）
    （风控演示：scripts/demo/prepare_risk_demo.sql，reset 后重跑）
 6. python scripts/sync/sync_advisor_rel.py && python scripts/sync/sync_neo4j.py
-7. uvicorn app.main:app --reload → GET /health；`docker compose up -d redis`（限流/会话窗口）；python -m pytest（**554 绿**，系统 Python 3.13.14）
+7. uvicorn app.main:app --reload → GET /health；`docker compose up -d redis`（限流/会话窗口）；python -m pytest（**730 绿**，系统 Python 3.13.14）
 ```
 
 **AL-09 合并后架构（一句话）：** 宿主 `gateway/` + 模块 `deps.py` **双栈并存**；对外登录/token **统一**；chat/risk 均走模块鉴权；接缝 S2 用 `auth_adapter`。
@@ -157,7 +157,7 @@ Core 模拟：scripts/core/reset.ps1 · 文档 docs/项目框架设计/Core模
 种子：scripts/agent/seed-aml-list.sql（AML 名单）· scripts/demo/prepare_risk_demo.sql（reset 后重跑）
 依赖：requirements.txt（LangGraph + langchain-core/openai + FastAPI + SQLAlchemy）
 启动：uvicorn app.main:app --reload → GET /health
-测试：python -m pytest（**554 绿**；集成测试需本机 MySQL + `risk_aml_list` + `prepare_risk_demo.sql`）
+测试：python -m pytest（**730 绿**；集成测试需本机 MySQL + `risk_aml_list` + `prepare_risk_demo.sql`）
 运维/演示脚本：scripts/demo/subscribe_alerts.py（订阅推送演示）· rebuild_alerts.py TRD-xxx（引擎异常补偿重放）
 JWT 联调：python scripts/dev/issue_dev_token.py --sub STAFF-30001 --roles risk_officer（+ Authorization: Bearer + X-Agent-Type）
 配置：.env（见 .env.example）
@@ -205,6 +205,6 @@ RBAC 联调账号：scripts/dev/rbac-seed-reference.md
 2. 改动属于 api / service / tool / repository 哪一层？
 3. 是否需 customer_id 归属与 JWT RBAC？
 4. Core 是模拟库只读还是 agent 库读写？
-5. 如何验证？（`python -m pytest` 全量（当前 **554 绿**）· `docker compose up -d redis` · uvicorn + /health · 平台 `/api/customers/*` · `tests/test_module_boundary.py` 全绿）
+5. 如何验证？（`python -m pytest` 全量（当前 **730 绿**）· `docker compose up -d redis` · uvicorn + /health · 平台 `/api/customers/*` · `tests/test_module_boundary.py` 全绿）
 
 大任务：FRAMEWORK/FLOW 与实现状态不符时先更新 memory 再编码（用户确认跳过除外）。
diff --git a/docs/memory/TODO.md b/docs/memory/TODO.md
index f911d40..c19b52a 100644
--- a/docs/memory/TODO.md
+++ b/docs/memory/TODO.md
@@ -37,12 +37,20 @@
 
 > 清单：`docs/项目框架设计/客服Agent-合并说明.md` §3–§4 · 原则只增不盖、不覆盖 `main.py`/`chat.py`/`agent_service.py` 整体。
 
-- [ ] **跑迁移 SQL**：`scripts/agent/migrate-customer-agent-cs-c11.sql`（`conversation_archive` · `customer_notes`）
+- [x] **跑迁移 SQL**：`scripts/agent/migrate-customer-agent-cs-c11.sql`（`conversation_archive` · `customer_notes`）（2026-09-09 本机已执行）
 - [x] **挂游客路由**：`POST /api/chat/visitor` → `main.py`
 - [x] **编排接缝（并行）**：`X-Agent-Type: customer` → `customer_service.run_customer_chat`；其余 Agent 仍 `agent_service.chat`
 - [x] **Wave 3 测试解禁**：`test_wave3_customer_service.py` 19 例绿 · 基线 **554 passed**
-- [ ] **Wave 1/2/4/5 解禁**：`tests/conftest.py` 移出剩余 `collect_ignore` · 修接缝后全 wave 绿
-- [ ] **可选接线**：`profile_service` / `note_service` / 客服 KB 脚本与 Milvus 集合（`scripts/kb/*`）— 按需求择项
+- [x] **Wave 1/2 解禁**：`test_wave1_*` · `test_wave2_*` 已移出 `collect_ignore` 并绿（132+ 例）
+- [x] **Wave 3 profile / 4 / 5 解禁**：`test_wave3_profile_service` · `test_wave4_e2e` · `test_wave5_notes` · 基线 **730 passed**
+- [x] **客服 KB 第二套（fin_*）· Ollama 灌库**（2026-09-09 本机）：
+  1. `ollama pull bge-m3` ✓
+  2. Ollama 服务 `http://127.0.0.1:11434` ✓
+  3. **Windows**：`MILVUS_URI` 须英文绝对路径（如 `C:/Users/Windows/.jinrong/milvus/milvus.db`），中文项目路径 faiss 会炸
+  4. `python scripts/kb/build_collections.py --rebuild`（原料 `data/kb_collections/`）✓ 9 块
+  5. `python scripts/kb/test_search.py` ✓
+- [x] **客服 RAG 接 fin_* 三库**：`VisitorRagService` → `search_cs_knowledge`（`fin_product` / `fin_policy` / `fin_faq`）；宿主 Tool 仍走 `kb_product_rules`（T-21）
+- [x] **tool 层实现 + 种子文档**：`document_parser` / `embedding_tool` / `milvus_tool` + `data/kb_collections/*`（2026-09-09）
 
 ### 风控 Agent · 后端已就绪 · 前端/运维未接（盘点 2026-09-09）
 
diff --git a/docs/项目框架设计/客服Agent-合并说明.md b/docs/项目框架设计/客服Agent-合并说明.md
index 75c92e1..f026d6a 100644
--- a/docs/项目框架设计/客服Agent-合并说明.md
+++ b/docs/项目框架设计/客服Agent-合并说明.md
@@ -66,17 +66,23 @@ scripts/agent/migrate-customer-agent-cs-c11.sql
 
 **仍待手工：**
 
-- 执行 `scripts/agent/migrate-customer-agent-cs-c11.sql`（`conversation_archive` · `customer_notes`）
-- Wave 1/2/4/5 测试仍 `collect_ignore`，逐步解禁
+- 本机真跑服务：`docker compose up -d redis`（游客/客服短期记忆与限流；无 Redis 时客服/游客对话可降级，但无跨轮记忆）
+
+**RAG 双轨（2026-09-09）：**
+
+| 链路 | Milvus | 灌库 |
+| --- | --- | --- |
+| 客服/游客 RAG | `fin_faq` / `fin_product` / `fin_policy` | `scripts/kb/build_collections.py` · 原料 `data/kb_collections/` |
+| 宿主 Tool | `kb_product_rules` | `scripts/kb/build_kb.py` · 原料 `data/kb/` |
 
 ---
 
 ## 4. 测试预期
 
-- 基线 `python -m pytest` → **554 passed**（含 `test_wave3_customer_service.py` 19 例）
-- Wave 1/2/4/5 仍在 `tests/conftest.py` `collect_ignore`，接线验证后逐步移出
+- 基线 `python -m pytest` → **730 passed**（Wave 1~5 客服测试已全部解禁）
+- `tests/conftest.py` `collect_ignore` 已清空
 - 新增 `pytest.ini`：`testpaths = tests`（避免 `scripts/kb/test_*.py` 被误收集）
-- 客服线全量（剩余 wave 解禁后）：`pytest tests/test_wave1_data_masker.py … test_wave5_notes.py`
+- 客服线全量：`pytest tests/test_wave1_data_masker.py … test_wave5_notes.py`
 
 ---
 
@@ -84,4 +90,5 @@ scripts/agent/migrate-customer-agent-cs-c11.sql
 
 | 日期 | 说明 |
 | --- | --- |
+| 2026-09-09 | Wave 1~5 解禁 · 迁移 SQL 已执行 · pytest **730 passed** · Scope B 冒烟（customer 200 · visitor Redis 降级） |
 | 2026-09-09 | S2 接缝接线：visitor 路由 + customer 分流 + wave3 测试绿 · pytest 554 passed |
diff --git a/scripts/kb/test_search.py b/scripts/kb/test_search.py
index 9b8444a..913642c 100644
--- a/scripts/kb/test_search.py
+++ b/scripts/kb/test_search.py
@@ -1,6 +1,11 @@
-"""快速测试 Milvus 检索是否正常工作。"""
+"""快速测试 fin_* Milvus 检索（需先 build_collections 灌库）。"""
+from __future__ import annotations
+
 import sys
-sys.path.insert(0, "d:/金融系统")
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT))
 
 from app.tool.embedding_tool import get_embedder
 from app.tool.milvus_tool import get_milvus_client
@@ -8,29 +13,23 @@ from app.tool.milvus_tool import get_milvus_client
 e = get_embedder()
 m = get_milvus_client()
 
-# 测试产品检索
-v = e.embed("R3中风险产品有哪些")
-hits = m.search("fin_product", v, top_k=3)
-print("=== 产品检索 ===")
-for h in hits:
-    score = h.get("score", 0)
-    text = h.get("chunk_text", "")[:100]
-    print(f"score={score:.4f} | {text}")
+try:
+    v = e.embed("R3中风险产品有哪些")
+    hits = m.search("fin_product", v, top_k=3)
+    print("=== 产品检索 ===")
+    for h in hits:
+        print(f"score={h.get('score', 0):.4f} | {(h.get('chunk_text') or '')[:100]}")
 
-# 测试 FAQ 检索
-v2 = e.embed("开户需要什么材料")
-hits2 = m.search("fin_faq", v2, top_k=3)
-print("\n=== FAQ 检索 ===")
-for h in hits2:
-    score = h.get("score", 0)
-    text = h.get("chunk_text", "")[:100]
-    print(f"score={score:.4f} | {text}")
+    v2 = e.embed("开户需要什么材料")
+    hits2 = m.search("fin_faq", v2, top_k=3)
+    print("\n=== FAQ 检索 ===")
+    for h in hits2:
+        print(f"score={h.get('score', 0):.4f} | {(h.get('chunk_text') or '')[:100]}")
 
-# 测试政策检索
-v3 = e.embed("投资者适当性匹配规则")
-hits3 = m.search("fin_policy", v3, top_k=3)
-print("\n=== 政策检索 ===")
-for h in hits3:
-    score = h.get("score", 0)
-    text = h.get("chunk_text", "")[:100]
-    print(f"score={score:.4f} | {text}")
+    v3 = e.embed("投资者适当性匹配规则")
+    hits3 = m.search("fin_policy", v3, top_k=3)
+    print("\n=== 政策检索 ===")
+    for h in hits3:
+        print(f"score={h.get('score', 0):.4f} | {(h.get('chunk_text') or '')[:100]}")
+finally:
+    m.close()
diff --git a/tests/conftest.py b/tests/conftest.py
index a31f1f8..1985246 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -74,22 +74,76 @@ class FakeRedis:
     def hgetall(self, key):
         return dict(self.hashes.get(key, {}))
 
+    def scan_iter(self, match=None):
+        import fnmatch
+
+        pat = match or "*"
+        keys = set(self.strings) | set(self.lists) | set(self.hashes)
+        for key in keys:
+            if fnmatch.fnmatch(key, pat):
+                yield key
+
+
+@pytest.fixture(autouse=True)
+def _wave_customer_fake_redis(request, monkeypatch):
+    """Wave 4/5：无本机 Redis 时用 FakeRedis（与 test_wave3 口径一致）。"""
+    nodeid = request.node.nodeid
+    if "test_wave4_e2e" not in nodeid and "test_wave5_notes" not in nodeid:
+        yield
+        return
+    r = FakeRedis()
+    monkeypatch.setattr("app.config.database.get_redis_client", lambda: r)
+    from app.service.risk import redis_gateway
+
+    monkeypatch.setattr(redis_gateway, "_gateway", r)
+    from app.service import customer_service as cs
+
+    monkeypatch.setattr(cs, "_spawn", lambda fn, *a: fn(*a))
+    yield r
+
+
+@pytest.fixture(autouse=True)
+def _wave_customer_test_session(request):
+    """Wave 4/5：在 agent_session 预建测试会话（merger chat 续聊须 SessionGuard 命中）。"""
+    nodeid = request.node.nodeid
+    if "test_wave4_e2e" not in nodeid and "test_wave5_notes" not in nodeid:
+        yield
+        return
+
+    from app.config.database import get_agent_engine
+    from app.repository.session_repository import SessionRepository
+
+    session_id = "sess-e2e-001" if "test_wave4_e2e" in nodeid else "sess-note-001"
+    cust = "CUST-9527"
+    engine = get_agent_engine()
+    with engine.begin() as conn:
+        conn.execute(
+            text("DELETE FROM agent_session WHERE session_id = :sid"),
+            {"sid": session_id},
+        )
+    SessionRepository(engine=engine).create_session(
+        session_id=session_id,
+        trace_id="trc-wave-customer-test",
+        agent_type="customer",
+        actor_id=cust,
+        actor_role="customer",
+        customer_id=cust,
+    )
+    yield
+    with engine.begin() as conn:
+        conn.execute(
+            text("DELETE FROM agent_session WHERE session_id = :sid"),
+            {"sid": session_id},
+        )
+
 
 @pytest.fixture
 def fake_redis():
     return FakeRedis()
 
 
-# 客服 Agent 线测试：接缝接线后逐步移出 ignore（见 客服Agent-合并说明.md）
-collect_ignore = [
-    "test_wave1_data_masker.py",
-    "test_wave2_core_ro_tool.py",
-    "test_wave2_profile_slots.py",
-    "test_wave2_prompts.py",
-    "test_wave3_profile_service.py",
-    "test_wave4_e2e.py",
-    "test_wave5_notes.py",
-]
+# 客服 Agent Wave 测试已全部解禁（见 客服Agent-合并说明.md）
+collect_ignore: list[str] = []
 
 
 @pytest.fixture(autouse=True)
diff --git a/tests/test_wave2_core_ro_tool.py b/tests/test_wave2_core_ro_tool.py
index f934c62..b92456a 100644
--- a/tests/test_wave2_core_ro_tool.py
+++ b/tests/test_wave2_core_ro_tool.py
@@ -9,6 +9,8 @@
 
 from __future__ import annotations
 
+from datetime import date, timedelta
+
 import pytest
 
 from app.repository.core_ro import CoreReadOnlyRepository
@@ -75,13 +77,18 @@ def test_holdings_empty(ro):
 # ---------------------------------------------------------------------------
 
 def test_trades_default_3_months(ro):
+    since = date.today() - timedelta(days=90)
+    rows = ro.list_trades("CUST-9527", since=since, limit=50)
+    confirmed = [r for r in rows if r.get("trade_status") == "confirmed"]
+
     r = query_trades("CUST-9527", repo=ro)
     assert r["ok"] is True
     assert r["tool"] == "transaction_query"
     assert "近 3 个月" in r["fact_text"]
-    assert "2 笔" in r["fact_text"]
+    assert f"{len(confirmed)} 笔" in r["fact_text"]
     assert "申购" in r["fact_text"]
-    assert len(r["facts"]) == 2
+    assert len(r["facts"]) == len(rows)
+    assert len(confirmed) >= 2  # Wave 0 种子至少 2 笔；联调库可能更多
 
 
 def test_trades_custom_months(ro):
diff --git a/tests/test_wave4_e2e.py b/tests/test_wave4_e2e.py
index 0d6ca3e..276df13 100644
--- a/tests/test_wave4_e2e.py
+++ b/tests/test_wave4_e2e.py
@@ -16,7 +16,8 @@ from unittest.mock import patch
 import pytest
 from fastapi.testclient import TestClient
 
-from app.config.database import get_agent_engine, get_core_engine, get_redis_client
+from app.config import database as db
+from app.config.database import get_agent_engine, get_core_engine
 from app.gateway.jwt_service import issue_token
 from app.main import app
 from app.service import customer_service
@@ -75,7 +76,7 @@ def _clean_redis():
     conftest mock 的 session_repo.ensure_session 返回 'sess-test-001'，
     所以实际 Redis key 用的是 'sess-test-001' 而非 body.session_id。
     """
-    r = get_redis_client()
+    r = db.get_redis_client()
     patterns = [
         "customer:sess-e2e-001:*",
         "customer:sess-test-001:*",
@@ -129,7 +130,9 @@ def _chat(client: TestClient, message: str, *, token: str | None = None,
             "X-Agent-Type": "customer",
         },
     )
-    return resp.json()
+    assert resp.status_code == 200, resp.text
+    raw = resp.json()
+    return raw.get("data", raw)
 
 
 def _core_scalar(sql: str, params: dict | None = None):
@@ -148,10 +151,8 @@ def test_01_holding_query(client):
     fake = _make_fake_invoke({"意图分类器": "holding_query",
                               "事实数据": "您的持仓包括以下产品。"})
     with patch.object(customer_service, "_invoke", side_effect=fake):
-        body = _chat(client, "我的持仓有哪些", token=_customer_token())
+        data = _chat(client, "我的持仓有哪些", token=_customer_token())
 
-    assert body["code"] == 0
-    data = body["data"]
     assert data["intent"] == "holding_query"
     assert data["reply"]
     assert data["agent_type"] == "customer"
@@ -172,11 +173,10 @@ def test_02_transaction_query(client):
         "事实数据": "近3个月交易记录如下。",
     })
     with patch.object(customer_service, "_invoke", side_effect=fake):
-        body = _chat(client, "最近3个月交易记录", token=_customer_token())
+        data = _chat(client, "最近3个月交易记录", token=_customer_token())
 
-    assert body["code"] == 0
-    assert body["data"]["intent"] == "transaction_query"
-    assert body["data"]["reply"]
+    assert data["intent"] == "transaction_query"
+    assert data["reply"]
     # 交叉验证：近 3 月 confirmed 交易数
     db_count = _core_scalar(
         "SELECT COUNT(*) FROM core_trade "
@@ -199,10 +199,9 @@ def test_03_risk_assessment_query(client):
         "事实数据": "您的风险评级为C3，有效期至...",
     })
     with patch.object(customer_service, "_invoke", side_effect=fake):
-        body = _chat(client, "我的风险等级是什么", token=_customer_token())
+        data = _chat(client, "我的风险等级是什么", token=_customer_token())
 
-    assert body["code"] == 0
-    assert body["data"]["intent"] == "risk_assessment_query"
+    assert data["intent"] == "risk_assessment_query"
     # 交叉验证：DB 中的风评等级
     risk_code = _core_scalar(
         "SELECT risk_code FROM core_customer_risk "
@@ -211,7 +210,7 @@ def test_03_risk_assessment_query(client):
         {"cid": CUST},
     )
     assert risk_code is not None
-    assert f"C{risk_code[-1]}" in body["data"]["reply"] or risk_code is not None
+    assert f"C{risk_code[-1]}" in data["reply"] or risk_code is not None
 
 
 # ---------------------------------------------------------------------------
@@ -233,10 +232,9 @@ def test_04_suitability_check_readonly(client):
         "事实数据": "根据适当性匹配，您的C3评级可购买R3及以下产品。",
     })
     with patch.object(customer_service, "_invoke", side_effect=fake):
-        body = _chat(client, "我能买R3产品吗", token=_customer_token())
+        data = _chat(client, "我能买R3产品吗", token=_customer_token())
 
-    assert body["code"] == 0
-    assert body["data"]["intent"] == "suitability_check"
+    assert data["intent"] == "suitability_check"
     # 验证 risk_suitability_log 无新记录
     with agent_engine.connect() as conn:
         after = conn.execute(text(
@@ -258,12 +256,11 @@ def test_05_rag_product_consult(client):
     })
     with patch.object(customer_service, "_invoke", side_effect=fake), \
          patch.object(VisitorRagService, "retrieve", return_value=fake_rag):
-        body = _chat(client, "货币基金A是什么产品", token=_customer_token())
+        data = _chat(client, "货币基金A是什么产品", token=_customer_token())
 
-    assert body["code"] == 0
-    assert body["data"]["intent"] == "product_consult"
-    assert body["data"]["reply"]
-    assert body["data"]["has_disclaimer"] is True
+    assert data["intent"] == "product_consult"
+    assert data["reply"]
+    assert data.get("has_disclaimer") is True
 
 
 # ---------------------------------------------------------------------------
@@ -273,12 +270,11 @@ def test_05_rag_product_consult(client):
 def test_06_reject_investment_advice(client):
     """说「推荐稳赚基金」→ reject 拒绝话术，不转人工。"""
     # 不 mock LLM（keyword_route 直接命中）
-    body = _chat(client, "推荐稳赚的基金", token=_customer_token())
+    data = _chat(client, "推荐稳赚的基金", token=_customer_token())
 
-    assert body["code"] == 0
-    assert body["data"]["intent"] == "reject"
-    assert body["data"]["transfer_to_human"] is False
-    assert body["data"]["reply"]  # 非空拒绝话术
+    assert data["intent"] == "reject"
+    assert data["transfer_to_human"] is False
+    assert data["reply"]
 
 
 # ---------------------------------------------------------------------------
@@ -287,11 +283,10 @@ def test_06_reject_investment_advice(client):
 
 def test_07_transfer_human_complaint(client):
     """说「我要投诉」→ transfer_human。"""
-    body = _chat(client, "我要投诉你们的服务", token=_customer_token())
+    data = _chat(client, "我要投诉你们的服务", token=_customer_token())
 
-    assert body["code"] == 0
-    assert body["data"]["intent"] == "transfer_human"
-    assert body["data"]["transfer_to_human"] is True
+    assert data["intent"] == "transfer_human"
+    assert data["transfer_to_human"] is True
 
 
 # ---------------------------------------------------------------------------
@@ -369,7 +364,7 @@ def test_09_profile_injection_next_day(client):
         })})
 
     # 刷新 Redis 热缓存（手动写入空 dict 触发 DB 回源）
-    r = get_redis_client()
+    r = db.get_redis_client()
     r.delete(f"profile:l1:{CUST}")
 
     captured = {}
@@ -386,9 +381,8 @@ def test_09_profile_injection_next_day(client):
         return "好的"
 
     with patch.object(customer_service, "_invoke", side_effect=capturing_invoke):
-        body = _chat(client, "今天天气怎么样", token=_customer_token())
+        data = _chat(client, "今天天气怎么样", token=_customer_token())
 
-    assert body["code"] == 0
     # 验证画像上下文注入了 LLM prompt
     user_prompt = captured.get("last_user", "")
     assert "指数基金" in user_prompt or "上海" in user_prompt, \
@@ -451,13 +445,19 @@ def test_11_cross_customer_forbidden(client):
         assert_customer_access(ctx, CUST_B)
     assert "403" in str(exc_info.value.error_code) or "NOT_OWNER" in exc_info.value.error_code
 
-    # 端到端：body 传别人 ID，但 resolve 忽略，数据仍是自己的
-    fake = _make_fake_invoke({"意图分类器": "holding_query",
-                              "事实数据": "您的持仓如下。"})
-    with patch.object(customer_service, "_invoke", side_effect=fake):
-        body = _chat(client, "我的持仓", token=_customer_token(), customer_id=CUST_B)
-
-    assert body["code"] == 0  # 不报错（resolve 忽略 body.customer_id）
+    # 端到端：customer token 传他人 customer_id → merger chat 层 403
+    tok = _customer_token()
+    resp = client.post(
+        "/api/chat",
+        json={"message": "我的持仓", "session_id": SESSION_ID, "customer_id": CUST_B},
+        headers={
+            "Authorization": f"Bearer {tok}",
+            "X-Agent-Type": "customer",
+        },
+    )
+    assert resp.status_code == 403
+    err = resp.json()
+    assert err.get("error_code") == "AUTH_403_NOT_OWNER"
 
 
 # ---------------------------------------------------------------------------
@@ -484,9 +484,8 @@ def test_12_redis_down_degrade(client):
          patch("app.service.customer_service.ProfileHotCache") as mock_hc:
         mock_hc_instance = mock_hc.return_value
         mock_hc_instance.get_style_tags.return_value = {}
-        body = _chat(client, "你好", token=_customer_token())
+        data = _chat(client, "你好", token=_customer_token())
 
-    assert body["code"] == 0
-    assert body["data"]["reply"]
+    assert data["reply"]
 
 
diff --git a/tests/test_wave5_notes.py b/tests/test_wave5_notes.py
index 38fccda..d0f9632 100644
--- a/tests/test_wave5_notes.py
+++ b/tests/test_wave5_notes.py
@@ -16,7 +16,8 @@ import pytest
 from fastapi.testclient import TestClient
 from sqlalchemy import text
 
-from app.config.database import get_agent_engine, get_redis_client
+from app.config import database as db
+from app.config.database import get_agent_engine
 from app.gateway.jwt_service import issue_token
 from app.repository.note_repository import CustomerNoteRepository
 from app.service import customer_service, note_service
@@ -57,7 +58,7 @@ class FakeLLM:
 
 @pytest.fixture(autouse=True)
 def _clean_env():
-    r = get_redis_client()
+    r = db.get_redis_client()
     for pattern in (
         f"customer:{SESSION_ID}:*",
         f"customer:sess-test-001:*",
@@ -93,15 +94,16 @@ def _customer_token() -> str:
 
 
 def _chat(client: TestClient, message: str) -> dict:
-    raw = client.post(
+    resp = client.post(
         "/api/chat",
         json={"message": message, "session_id": SESSION_ID},
         headers={
             "Authorization": f"Bearer {_customer_token()}",
             "X-Agent-Type": "customer",
         },
-    ).json()
-    # 响应包了一层 {code, message, data: {...}, trace_id}，测试用 data 字段
+    )
+    assert resp.status_code == 200, resp.text
+    raw = resp.json()
     return raw.get("data", raw)