test(dispatcher): RAG 管线集成测试 —— 检索→注入→生成→refs→忠实度评测整条链

补全核心链路最后一块。此前只测了 retriever 组件的解析，整条 RAG 管线没被集成覆盖。

新增 rag_integration_test.go 4 例：
- ConversationInjectsAndReturnsRefs：input→retriever→agent，断言 kb_search 被调、
  kb 按 owner 作用域(u42/travel)、检索片段注入 agent system prompt、refs 经 runGraph 回流。
- RefsDriveGroundedEval：有 refs 时 evaluate 走「含检索资料」的 grounded judge 路径，
  记录忠实度分（而非无来源路径）。
- ReportSectionInjectsRefs：报告 writeSection 检索命中注入撰写 prompt。
- DegradesWhenRetrieverDown：kb_search 失败 → 空 refs，agent 仍正常出答案、整图不失败。

go test -race ./internal/eino 干净，四模块全绿。至此核心编排链 + RAG 管线均有集成兜底。

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

project_analysis.md

@@ -135,7 +135,7 @@ Harness = 围绕 LLM 的可靠性 / 安全 / 质量治理层。4 个组件均为
 - **compose 引擎仍灰度**（默认走自研 graph.go）；compose 路径下 HITL 审批未实现。
 - **OCR 多模态是骨架**：`parse_document` 真支持 txt/md/csv/docx/xlsx/文本 PDF，但 MinerU/PaddleOCR 扫描件路径是 TODO 空实现（`mineru.py` 极小）。
 - **推理模型已部分适配**：reasoning 模型的 `reasoning_content` 已捕获并 surface 到 exec 轨迹「推理过程」事件（不污染答案）；尚未在最终答案区把思维链单独流式呈现给前端。
-- **测试纵深改善中**：除小函数单测，已补 `Handle()` 入口级集成测试（熔断/护栏/预算/状态机/工具→agent/异步评测 6 例，-race 干净）+ bus 真实 NATS e2e；剩 RAG 管线集成、前端 E2E。
+- **测试纵深改善中**：除小函数单测，已补 `Handle()` 入口级集成（熔断/护栏/预算/状态机/工具→agent/异步评测 6 例）+ **RAG 管线集成**（检索→注入 prompt→生成→refs 回流→grounded 忠实度评测 + owner 作用域 + 优雅降级 4 例）+ bus 真实 NATS e2e，均 -race 干净；剩前端 E2E。
 
 ---
 
@@ -208,7 +208,7 @@ Harness = 围绕 LLM 的可靠性 / 安全 / 质量治理层。4 个组件均为
 | ~~P1~~ 🟡 | 高可用（代码层就绪）：调度/工具本就队列组可多副本（实测 2 副本 8 任务 4/4 分摊）；**网关改为队列组订阅**（eval/usage/status/config），多副本不再重复落库/重复计费（单测+实测去重）。剩 NATS 集群 + 网关 LB + PG/Redis HA 属部署期 | 解单点 |
 | P1 | 备份/灾备演练（PG/Milvus/Neo4j） | 数据安全 |
 | ~~P1~~ ✅ | ~~优雅停机 drain~~：三 Go 服务全覆盖（gateway HTTP Shutdown / dispatcher 在途任务跑完 / mcp-go 在途工具回完），SIGTERM 后等在途至 SHUTDOWN_DRAIN_TIMEOUT(默认30s)再退；在途任务 ctx 脱离信号 ctx 不被掐断。**+ exec 轨迹 Redis 回放**（与 token 流同构，连晚/刷新重连不丢轨迹，实测事后连仍补齐全程）| 可靠性 |
-| ~~P1~~ 🟡 | ~~核心链路集成测试~~：Handle 入口级 6 例（熔断/护栏/预算/状态机/工具→agent/异步评测）已补。剩 RAG 管线集成 + 前端 E2E | 测试纵深 |
+| ~~P1~~ 🟡 | ~~核心链路集成测试~~：Handle 入口级 6 例 + RAG 管线 4 例（检索→注入→生成→refs→grounded 评测）已补。剩前端 E2E | 测试纵深 |
 | P2 | 计量计费闭环 + 按用户配额；OCR 完善；KbView 拆分；循环节点 | |
 | P2 | 安全审计 / 渗透（让"纵深防御"从自述变已验证） | 外部 |
 | P3 | 多租户/团队；插件体系；K8s Helm；代码签名 | |

sundynix-dispatcher/internal/eino/rag_integration_test.go

@@ -0,0 +1,132 @@
+package eino
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+	"testing"
+
+	"github.com/sundynix/sundynix-dispatcher/internal/harness"
+	"github.com/sundynix/sundynix-dispatcher/internal/llm"
+	"github.com/sundynix/sundynix-shared/contract"
+)
+
+// ---- RAG 管线集成测试：检索 → 注入 prompt → 生成 → refs 回流（喂忠实度评测）整条链。----
+
+const ragSnippet = "西湖是杭州著名景点，三面环山。"
+
+// kbTool 返回一个把 kb_search 应答成固定命中、并捕获入参的 fakeTools。
+func kbTool(capture *contract.ToolCall) *fakeTools {
+	return &fakeTools{fn: func(c *contract.ToolCall) *contract.ToolResult {
+		if c.Tool == "kb_search" {
+			if capture != nil {
+				*capture = *c
+			}
+			return &contract.ToolResult{OK: true, Content: `[{"text":"` + ragSnippet + `","score":0.92}]`}
+		}
+		return &contract.ToolResult{OK: true, Content: ""}
+	}}
+}
+
+// 对话 RAG：input→retriever→agent。检索命中应注入 agent 的 system prompt，且 refs 经 runGraph 回流。
+func TestRAG_ConversationInjectsAndReturnsRefs(t *testing.T) {
+	g := `{"nodes":[
+		{"id":"i","kind":"input","config":{"text":"介绍杭州西湖"}},
+		{"id":"r","kind":"retriever","config":{"kb":"travel"}},
+		{"id":"a","kind":"agent","config":{"system":"你是导游"}}
+	],"edges":[{"source":"i","target":"r"},{"source":"r","target":"a"}]}`
+	// fakeLLM 回显 system 消息 → 借此断言检索片段已注入 prompt。
+	ll := &fakeLLM{ready: true, stream: func(m []llm.ChatMessage) string { return m[0].Content }}
+	var captured contract.ToolCall
+	o := newOrch(ll, kbTool(&captured), &fakeSink{}, &fakeExec{})
+	tk := &contract.Task{ID: "t1", Graph: json.RawMessage(g), Meta: map[string]any{contract.MetaUserID: "u42"}}
+
+	ans, refs, err := o.runGraph(context.Background(), tk, o.tracer("t1"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if captured.Tool != "kb_search" {
+		t.Fatal("应调用 kb_search 检索")
+	}
+	// owner 作用域：未带 "/" 的库名应被前缀成 uid/kb。
+	if kb, _ := captured.Args["kb"].(string); kb != "u42/travel" {
+		t.Errorf("kb 应按 owner 作用域为 u42/travel，got %q", kb)
+	}
+	if !strings.Contains(ans, ragSnippet) {
+		t.Errorf("检索片段应注入 agent prompt，prompt=%q", ans)
+	}
+	// refs 回流（喂忠实度评测的 sources）。
+	if len(refs) == 0 || !strings.Contains(strings.Join(refs, ""), ragSnippet) {
+		t.Errorf("runGraph 应回流含检索片段的 refs，got %v", refs)
+	}
+}
+
+// refs 回流后应触发「有来源」的忠实度评测路径（grounded judge），而非无来源路径。
+func TestRAG_RefsDriveGroundedEval(t *testing.T) {
+	ll := &fakeLLM{ready: true}
+	o := newOrch(ll, &fakeTools{}, &fakeSink{}, &fakeExec{})
+	var sawSources bool
+	o.eval = harness.NewEvaluator(func() bool { return true },
+		func(_ context.Context, _, user string) (string, error) {
+			// grounded judge 的 user 含「检索资料」段；无来源路径则没有。
+			if strings.Contains(user, "检索资料") {
+				sawSources = true
+				return `{"quality":5,"faithfulness":5,"unsupported":[],"reason":"忠实"}`, nil
+			}
+			return `{"score":3,"reason":"无来源"}`, nil
+		})
+	sink := &fakeEvalSink{}
+	o.evalSink = sink
+
+	o.evaluate(&contract.Task{ID: "t1"}, "介绍西湖", "西湖三面环山。", []string{ragSnippet})
+
+	if !sawSources {
+		t.Fatal("有 refs 时应走 grounded（含检索资料）评测路径")
+	}
+	if ev := sink.get(); ev == nil || ev.Faithful == 0 {
+		t.Errorf("应记录忠实度分，got %+v", ev)
+	}
+}
+
+// 报告 RAG：writeSection 检索命中应注入撰写 prompt。
+func TestRAG_ReportSectionInjectsRefs(t *testing.T) {
+	// writeSection 走 Chat（非流式）→ fakeLLM.chat 回显 user 消息以验证注入。
+	ll := &fakeLLM{ready: true, chat: func(m []llm.ChatMessage) (string, error) {
+		return m[len(m)-1].Content, nil // user 消息（含参考资料）
+	}}
+	o := newOrch(ll, kbTool(nil), &fakeSink{}, &fakeExec{})
+
+	body := o.writeSection(context.Background(), "杭州旅游", "u1/travel", "西湖", o.tracer("t1"), "section:0")
+	if !strings.Contains(body, ragSnippet) {
+		t.Errorf("报告章节撰写 prompt 应含检索资料，got %q", body)
+	}
+}
+
+// 优雅降级：检索工具不可用时返回空 refs，agent 仍正常出答案、不报错。
+func TestRAG_DegradesWhenRetrieverDown(t *testing.T) {
+	g := `{"nodes":[
+		{"id":"i","kind":"input","config":{"text":"介绍杭州"}},
+		{"id":"r","kind":"retriever","config":{"kb":"travel"}},
+		{"id":"a","kind":"agent","config":{"system":"导游答复"}}
+	],"edges":[{"source":"i","target":"r"},{"source":"r","target":"a"}]}`
+	ll := &fakeLLM{ready: true, stream: func(m []llm.ChatMessage) string { return m[0].Content }}
+	// kb_search 返回失败 → retrieve 降级空。
+	ft := &fakeTools{fn: func(c *contract.ToolCall) *contract.ToolResult {
+		if c.Tool == "kb_search" {
+			return &contract.ToolResult{OK: false, Error: "milvus down"}
+		}
+		return &contract.ToolResult{OK: true}
+	}}
+	o := newOrch(ll, ft, &fakeSink{}, &fakeExec{})
+
+	ans, refs, err := o.runGraph(context.Background(), &contract.Task{ID: "t1", Graph: json.RawMessage(g), Meta: map[string]any{}}, o.tracer("t1"))
+	if err != nil {
+		t.Fatalf("检索失败不应让整图失败: %v", err)
+	}
+	if len(refs) != 0 {
+		t.Errorf("检索失败应空 refs，got %v", refs)
+	}
+	if !strings.Contains(ans, "导游答复") {
+		t.Errorf("无检索仍应正常出答案，got %q", ans)
+	}
+}