diff --git a/sundynix-desktop/frontend/src/components/UpdateBanner.tsx b/sundynix-desktop/frontend/src/components/UpdateBanner.tsx
new file mode 100644
index 0000000..633c00b
--- /dev/null
+++ b/sundynix-desktop/frontend/src/components/UpdateBanner.tsx
@@ -0,0 +1,38 @@
+import { useEffect, useState } from "react";
+import { Download, X } from "lucide-react";
+import { checkUpdate, openExternal, type ReleaseInfo } from "../lib/version";
+
+// 启动时查 GitHub Releases，有新版则在顶部显示一条可关闭的横幅，点击去下载页。
+export function UpdateBanner() {
+ const [rel, setRel] = useState
(null);
+ const [dismissed, setDismissed] = useState(false);
+
+ useEffect(() => {
+ let alive = true;
+ void checkUpdate().then((r) => {
+ if (alive) setRel(r);
+ });
+ return () => {
+ alive = false;
+ };
+ }, []);
+
+ if (!rel || dismissed) return null;
+ return (
+
+
+
+ 新版本 v{rel.version} 可用
+
+
+
+
+ );
+}
diff --git a/sundynix-desktop/frontend/src/lib/version.ts b/sundynix-desktop/frontend/src/lib/version.ts
new file mode 100644
index 0000000..34d4348
--- /dev/null
+++ b/sundynix-desktop/frontend/src/lib/version.ts
@@ -0,0 +1,50 @@
+// 桌面端版本与"检查更新"——查 GitHub Releases 最新版，比当前版本新则提示下载。
+// 发版流程：bump 此处 APP_VERSION（= package.json version）→ 打 tag vX.Y.Z → release workflow 自动构建+发布。
+
+export const APP_VERSION = "0.1.0"; // 当前桌面端版本（与 git tag 去掉 v 前缀对齐）
+
+const REPO = "blizzardzhang/sundynix-agentix";
+
+export interface ReleaseInfo {
+ version: string; // 最新版本号（不含 v）
+ url: string; // release 页面（用户在此选 mac/win 安装包）
+ notes: string; // 更新日志
+}
+
+// checkUpdate 查 GitHub 最新 release；有比当前更新的版本则返回信息，否则 null（含出错/限流时静默）。
+export async function checkUpdate(): Promise {
+ try {
+ const res = await fetch(`https://api.github.com/repos/${REPO}/releases/latest`, {
+ headers: { Accept: "application/vnd.github+json" },
+ });
+ if (!res.ok) return null; // 404=还没发过 release / 403=限流 → 静默
+ const d = (await res.json()) as { tag_name?: string; html_url?: string; body?: string };
+ const latest = String(d.tag_name ?? "").replace(/^v/, "");
+ if (!latest || !isNewer(latest, APP_VERSION)) return null;
+ return { version: latest, url: d.html_url ?? `https://github.com/${REPO}/releases/latest`, notes: d.body ?? "" };
+ } catch {
+ return null; // 离线等 → 不打扰
+ }
+}
+
+// openExternal 在系统浏览器打开链接（Wails 用 runtime.BrowserOpenURL，浏览器模式回退 window.open）。
+export function openExternal(url: string): void {
+ const w = window as unknown as { runtime?: { BrowserOpenURL?: (u: string) => void } };
+ if (w.runtime?.BrowserOpenURL) {
+ w.runtime.BrowserOpenURL(url);
+ } else {
+ window.open(url, "_blank");
+ }
+}
+
+// isNewer 语义化版本比较：a 是否比 b 新（按 major.minor.patch）。
+function isNewer(a: string, b: string): boolean {
+ const pa = a.split(".").map((n) => parseInt(n, 10) || 0);
+ const pb = b.split(".").map((n) => parseInt(n, 10) || 0);
+ for (let i = 0; i < 3; i++) {
+ const x = pa[i] ?? 0;
+ const y = pb[i] ?? 0;
+ if (x !== y) return x > y;
+ }
+ return false;
+}
diff --git a/sundynix-desktop/frontend/src/studio/nodeCatalog.ts b/sundynix-desktop/frontend/src/studio/nodeCatalog.ts
index f28edac..800a988 100644
--- a/sundynix-desktop/frontend/src/studio/nodeCatalog.ts
+++ b/sundynix-desktop/frontend/src/studio/nodeCatalog.ts
@@ -58,8 +58,9 @@ export const NODE_KINDS: Record = {
{ key: "model", label: "模型", type: "select", options: ["占位 Pool", "ollama:qwen", "vllm:custom"], required: true },
{ key: "system", label: "系统提示词", type: "textarea", placeholder: "你是…" },
{ key: "temperature", label: "温度", type: "number" },
+ { key: "autonomous", label: "自主工具 (ReAct)", type: "checkbox" },
],
- defaults: { model: "占位 Pool", system: "", temperature: 0.7 },
+ defaults: { model: "占位 Pool", system: "", temperature: 0.7, autonomous: false },
},
tool: {
kind: "tool",
diff --git a/sundynix-dispatcher/cmd/dispatcher/main.go b/sundynix-dispatcher/cmd/dispatcher/main.go
index 65a0620..c0caa6f 100644
--- a/sundynix-dispatcher/cmd/dispatcher/main.go
+++ b/sundynix-dispatcher/cmd/dispatcher/main.go
@@ -41,8 +41,8 @@ func main() {
log.Printf("[dispatcher] subscribe model config: %v", err)
}
- // sub 同时作为 Token 回流出口（TokenSink）、MCP 工具调用出口（ToolCaller）与执行事件出口（ExecSink）。
- orch, err := eino.NewOrchestrator(pool, breaker, eval, sub, sub, sub)
+ // sub 同时作为 Token 回流(TokenSink)、MCP 工具调用(ToolCaller)、执行事件(ExecSink)与任务状态回写(StatusSink)出口。
+ orch, err := eino.NewOrchestrator(pool, breaker, eval, sub, sub, sub, sub)
if err != nil {
log.Fatalf("[dispatcher] build eino graph: %v", err)
}
diff --git a/sundynix-dispatcher/internal/eino/agent_collab_test.go b/sundynix-dispatcher/internal/eino/agent_collab_test.go
new file mode 100644
index 0000000..b479bd5
--- /dev/null
+++ b/sundynix-dispatcher/internal/eino/agent_collab_test.go
@@ -0,0 +1,46 @@
+package eino
+
+import (
+ "context"
+ "strings"
+ "testing"
+
+ "github.com/sundynix/sundynix-dispatcher/internal/harness"
+ "github.com/sundynix/sundynix-dispatcher/internal/llm"
+ "github.com/sundynix/sundynix-shared/contract"
+)
+
+// TestAgentCollaborationPassesOutput 验证编排式多智能体接力：
+// agent1 的产出注入 agent2 的上下文，agent2 基于它继续；最终成稿 = 下游 agent 的产出。
+func TestAgentCollaborationPassesOutput(t *testing.T) {
+ graph := `{"version":"1","nodes":[
+ {"id":"a1","kind":"agent","config":{"system":"研究"}},
+ {"id":"a2","kind":"agent","config":{"system":"撰写"}}
+ ],"edges":[{"source":"a1","target":"a2"}]}`
+
+ saw := false
+ ll := &fakeLLM{ready: true, stream: func(m []llm.ChatMessage) string {
+ var sys string
+ for _, x := range m {
+ if x.Role == "system" {
+ sys = x.Content
+ }
+ }
+ if strings.Contains(sys, "研究产出XYZ") { // agent1 的产出出现在 agent2 的 system → 接力成功
+ saw = true
+ return "最终报告：基于上游研究撰写"
+ }
+ return "研究产出XYZ"
+ }}
+ o := &Orchestrator{pool: ll, breaker: harness.NewCircuitBreaker(), sink: &fakeSink{}}
+ ans, err := o.runGraph(context.Background(), &contract.Task{ID: "tc", Graph: []byte(graph)}, &execTracer{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !saw {
+ t.Fatal("下游 agent 未看到上游 agent 的产出——协作没接力")
+ }
+ if !strings.Contains(ans, "最终报告") {
+ t.Fatalf("最终成稿应为下游 agent 产出，得 %q", ans)
+ }
+}
diff --git a/sundynix-dispatcher/internal/eino/compile.go b/sundynix-dispatcher/internal/eino/compile.go
index 56bd90c..82527d2 100644
--- a/sundynix-dispatcher/internal/eino/compile.go
+++ b/sundynix-dispatcher/internal/eino/compile.go
@@ -5,6 +5,7 @@ import (
"encoding/json"
"strings"
+ "github.com/cloudwego/eino/components/prompt"
"github.com/cloudwego/eino/schema"
)
@@ -18,10 +19,20 @@ type RunCtx struct {
Profile string // 召回的画像
History []*schema.Message // 短期历史
ToolOut []string // 工具/检索节点产出（含参考资料）
+ Upstream []string // 前序协作 agent 的产出（多 agent 接力时注入，让下游基于上游继续）
}
+// chatTemplate 是会话消息模板：系统提示词 + 历史占位 + 用户输入。
+// 用 Eino components/prompt.ChatTemplate（FString 仅解析模板串、值原样注入，故 JSON 花括号安全）。
+var chatTemplate = prompt.FromMessages(schema.FString,
+ schema.SystemMessage("{system}"),
+ schema.MessagesPlaceholder("history", true),
+ schema.UserMessage("{query}"),
+)
+
// buildMessages 把上下文组装为发给模型的消息序列（系统提示词 + 画像 + 工具产出 + 历史 + 用户输入）。
-func buildMessages(_ context.Context, rc *RunCtx) ([]*schema.Message, error) {
+// 系统串的动态拼装（按需注入画像/参考资料）留在 Go 侧；最终经 ChatTemplate 成型。
+func buildMessages(ctx context.Context, rc *RunCtx) ([]*schema.Message, error) {
var sys strings.Builder
sys.WriteString(rc.System)
if rc.Profile != "" {
@@ -33,11 +44,15 @@ func buildMessages(_ context.Context, rc *RunCtx) ([]*schema.Message, error) {
sys.WriteString("\n\n以下是工具/检索得到的参考资料：\n")
sys.WriteString(strings.Join(rc.ToolOut, "\n---\n"))
}
- msgs := make([]*schema.Message, 0, len(rc.History)+2)
- msgs = append(msgs, schema.SystemMessage(sys.String()))
- msgs = append(msgs, rc.History...)
- msgs = append(msgs, schema.UserMessage(rc.Query))
- return msgs, nil
+ if len(rc.Upstream) > 0 {
+ sys.WriteString("\n\n以下是前序协作 agent 的产出，请在此基础上继续完成你的部分（不要重头再来）：\n")
+ sys.WriteString(strings.Join(rc.Upstream, "\n---\n"))
+ }
+ return chatTemplate.Format(ctx, map[string]any{
+ "system": sys.String(),
+ "history": rc.History,
+ "query": rc.Query,
+ })
}
// previewArgs 把工具入参压成一行短预览。
diff --git a/sundynix-dispatcher/internal/eino/compose_callbacks.go b/sundynix-dispatcher/internal/eino/compose_callbacks.go
new file mode 100644
index 0000000..3466255
--- /dev/null
+++ b/sundynix-dispatcher/internal/eino/compose_callbacks.go
@@ -0,0 +1,68 @@
+package eino
+
+import (
+ "context"
+ "fmt"
+
+ "github.com/cloudwego/eino/callbacks"
+ "github.com/cloudwego/eino/components/model"
+ "github.com/cloudwego/eino/components/tool"
+ "github.com/cloudwego/eino/schema"
+ ucallbacks "github.com/cloudwego/eino/utils/callbacks"
+)
+
+// composeTracer 把 Eino compose 运行时的回调（ChatModel / Tool 的 start/end/error）
+// 翻译成我们现有的 ExecEvent 轨迹——可观测"归一"：用框架原生回调，而非各处手写 emit。
+// node 为该次运行在"运行·观测"里的归属节点 id（如 agent:xxx）。
+func composeTracer(tr *execTracer, node string) callbacks.Handler {
+ return ucallbacks.NewHandlerHelper().
+ ChatModel(&ucallbacks.ModelCallbackHandler{
+ OnStart: func(ctx context.Context, _ *callbacks.RunInfo, in *model.CallbackInput) context.Context {
+ tr.emit(node, "model", "start", "compose·ChatModel", inputMsgsPreview(in), 0)
+ return ctx
+ },
+ OnEnd: func(ctx context.Context, _ *callbacks.RunInfo, out *model.CallbackOutput) context.Context {
+ detail := ""
+ if out != nil && out.Message != nil {
+ detail = truncate(out.Message.Content, 120)
+ }
+ tr.emit(node, "model", "end", "compose·ChatModel", detail, 0)
+ return ctx
+ },
+ OnEndWithStreamOutput: func(ctx context.Context, _ *callbacks.RunInfo, out *schema.StreamReader[*model.CallbackOutput]) context.Context {
+ out.Close() // 仅观测：正文经主输出流消费，这里只标记结束
+ tr.emit(node, "model", "end", "compose·ChatModel", "流式完成", 0)
+ return ctx
+ },
+ OnError: func(ctx context.Context, _ *callbacks.RunInfo, err error) context.Context {
+ tr.emit(node, "model", "error", "compose·ChatModel", err.Error(), 0)
+ return ctx
+ },
+ }).
+ Tool(&ucallbacks.ToolCallbackHandler{
+ OnStart: func(ctx context.Context, _ *callbacks.RunInfo, in *tool.CallbackInput) context.Context {
+ if in != nil {
+ tr.emit(node, "tool", "start", "compose·工具", truncate(in.ArgumentsInJSON, 120), 0)
+ }
+ return ctx
+ },
+ OnEnd: func(ctx context.Context, _ *callbacks.RunInfo, out *tool.CallbackOutput) context.Context {
+ if out != nil {
+ tr.emit(node, "tool", "end", "compose·工具", truncate(out.Response, 160), 0)
+ }
+ return ctx
+ },
+ OnError: func(ctx context.Context, _ *callbacks.RunInfo, err error) context.Context {
+ tr.emit(node, "tool", "error", "compose·工具", err.Error(), 0)
+ return ctx
+ },
+ }).
+ Handler()
+}
+
+func inputMsgsPreview(in *model.CallbackInput) string {
+ if in == nil {
+ return ""
+ }
+ return fmt.Sprintf("%d 条消息", len(in.Messages))
+}
diff --git a/sundynix-dispatcher/internal/eino/compose_compiler.go b/sundynix-dispatcher/internal/eino/compose_compiler.go
new file mode 100644
index 0000000..d0dc0d7
--- /dev/null
+++ b/sundynix-dispatcher/internal/eino/compose_compiler.go
@@ -0,0 +1,217 @@
+package eino
+
+import (
+ "context"
+ "fmt"
+ "sync"
+
+ "github.com/cloudwego/eino/compose"
+
+ "github.com/sundynix/sundynix-dispatcher/internal/dsl"
+ "github.com/sundynix/sundynix-shared/contract"
+)
+
+// flowSignal 是 compose 编排图的边载荷（占位）：真实数据全走 compose 本地状态(*board)，
+// 边只传"该走了"的信号。注册 no-op 合并以支持 fan-in（多分支汇聚到一个节点）。
+type flowSignal struct{}
+
+var registerMergeOnce sync.Once
+
+func registerFlowMerge() {
+ registerMergeOnce.Do(func() {
+ compose.RegisterValuesMergeFunc(func([]flowSignal) (flowSignal, error) { return flowSignal{}, nil })
+ })
+}
+
+// executeGraph 按灰度开关选编排实现：compose.Graph（Phase C）或自研 graph.go（默认/权威）。
+func (o *Orchestrator) executeGraph(ctx context.Context, t *contract.Task, tr *execTracer) (string, error) {
+ if composeEnabled() {
+ return o.runComposeGraph(ctx, t, tr)
+ }
+ return o.runGraph(ctx, t, tr)
+}
+
+// runComposeGraph 把 DSL 图编译为 Eino compose.Graph 并执行（Phase C 编排归一）：
+// 节点体复用现有 execDSLNode；黑板进 compose 本地状态；branch 走 AddBranch；
+// DAG 触发模式让无依赖节点并行调度（效率）。编译失败即降级回自研 graph.go（安全网）。
+func (o *Orchestrator) runComposeGraph(ctx context.Context, t *contract.Task, tr *execTracer) (string, error) {
+ registerFlowMerge()
+ flow, ferr := dsl.Parse(t.Graph)
+ plan := dsl.Compile(t.Graph)
+ b := &board{
+ uid: meta(t, contract.MetaUserID),
+ sid: meta(t, contract.MetaSessionID),
+ query: plan.Query,
+ }
+
+ // 无图/空图：退化为 compose 单轮对话。
+ if ferr != nil || flow == nil || len(flow.Nodes) == 0 {
+ tr.info("task", "system", "无结构化图", "按单轮对话执行（compose）")
+ b.profile = o.fetchMemory(ctx, b.uid, b.query)
+ b.history = o.fetchHistory(ctx, b.sid)
+ o.runComposeConversation(ctx, t.ID, b, plan.System, tr, "agent")
+ return b.answer, nil
+ }
+
+ // 邻接 + 入度（只认两端都存在的边）。
+ nodeByID := make(map[string]dsl.Node, len(flow.Nodes))
+ outE := make(map[string][]dsl.Edge)
+ indeg := make(map[string]int, len(flow.Nodes))
+ for _, n := range flow.Nodes {
+ nodeByID[n.ID] = n
+ indeg[n.ID] = 0
+ }
+ for _, e := range flow.Edges {
+ if _, ok := nodeByID[e.Source]; !ok {
+ continue
+ }
+ if _, ok := nodeByID[e.Target]; !ok {
+ continue
+ }
+ outE[e.Source] = append(outE[e.Source], e)
+ indeg[e.Target]++
+ }
+
+ // 图里无 memory 节点 → 沿用默认：注入画像+历史（与 graph.go 对齐，避免回归）。
+ hasMemory := false
+ for _, n := range flow.Nodes {
+ if n.Kind == "memory" {
+ hasMemory = true
+ break
+ }
+ }
+ if !hasMemory {
+ b.profile = o.fetchMemory(ctx, b.uid, b.query)
+ b.history = o.fetchHistory(ctx, b.sid)
+ }
+
+ // 建 compose 图：黑板进本地状态（GenLocalState 闭包持有本任务的 b）。
+ g := compose.NewGraph[flowSignal, flowSignal](
+ compose.WithGenLocalState(func(context.Context) *board { return b }),
+ )
+ key := func(id string) string { return "n_" + id } // 节点 key 加前缀，避开 START/END 保留字
+
+ // 1) 加节点（branch 为 passthrough，路由交给 AddBranch）。
+ for _, n := range flow.Nodes {
+ node := n
+ if node.Kind == "branch" {
+ _ = g.AddLambdaNode(key(node.ID), compose.InvokableLambda(
+ func(context.Context, flowSignal) (flowSignal, error) { return flowSignal{}, nil }))
+ continue
+ }
+ _ = g.AddLambdaNode(key(node.ID), compose.InvokableLambda(
+ func(c context.Context, _ flowSignal) (flowSignal, error) {
+ perr := compose.ProcessState(c, func(sc context.Context, bd *board) error {
+ o.execDSLNode(sc, t, node, bd, plan, tr)
+ return nil
+ })
+ return flowSignal{}, perr
+ }))
+ }
+
+ // 2) 连边。branch 用 AddBranch（条件读 board 选下游）；其余直连；终端节点连 END。
+ for _, n := range flow.Nodes {
+ node := n
+ outs := outE[node.ID]
+ if node.Kind == "branch" {
+ endNodes := map[string]bool{compose.END: true}
+ for _, e := range outs {
+ if _, ok := nodeByID[e.Target]; ok {
+ endNodes[key(e.Target)] = true
+ }
+ }
+ brn := node
+ cond := func(c context.Context, _ flowSignal) (map[string]bool, error) {
+ chosen := map[string]bool{}
+ _ = compose.ProcessState(c, func(sc context.Context, bd *board) error {
+ for _, tgt := range o.branchNode(brn, bd, outE[brn.ID], nodeByID, tr) {
+ chosen[key(tgt)] = true
+ }
+ return nil
+ })
+ if len(chosen) == 0 {
+ chosen[compose.END] = true // 没选中任何下游 → 收口到 END，避免悬挂
+ }
+ return chosen, nil
+ }
+ _ = g.AddBranch(key(node.ID), compose.NewGraphMultiBranch(cond, endNodes))
+ continue
+ }
+ if len(outs) == 0 {
+ _ = g.AddEdge(key(node.ID), compose.END)
+ continue
+ }
+ for _, e := range outs {
+ if _, ok := nodeByID[e.Target]; ok {
+ _ = g.AddEdge(key(node.ID), key(e.Target))
+ }
+ }
+ }
+
+ // 3) 入口节点（入度 0）连 START。
+ for _, n := range flow.Nodes {
+ if indeg[n.ID] == 0 {
+ _ = g.AddEdge(compose.START, key(n.ID))
+ }
+ }
+
+ // 4) 编译（DAG 模式：无依赖节点并行调度）。编译失败 → 降级回自研 graph.go（安全网）。
+ r, cerr := g.Compile(ctx, compose.WithNodeTriggerMode(compose.AllPredecessor))
+ if cerr != nil {
+ tr.info("task", "system", "compose 编译失败", "退回自研 graph.go："+cerr.Error())
+ return o.runGraph(ctx, t, tr)
+ }
+ if _, ierr := r.Invoke(ctx, flowSignal{}); ierr != nil {
+ tr.info("task", "system", "compose 执行告警", ierr.Error()) // 副作用已落 board；下方按需补一段答复
+ }
+
+ // 图里无 agent 节点（纯工具/检索图）也要出一段答复。
+ if b.answer == "" {
+ o.runComposeConversation(ctx, t.ID, b, plan.System, tr, "agent")
+ }
+ return b.answer, nil
+}
+
+// execDSLNode 执行一个非 branch 的 DSL 节点（compose 编译器用；节点体与 graph.go 一致，
+// 区别仅 agent 直接走 runAgent/runReactAgent——compose 已是编排层，不再二次套 compose）。
+func (o *Orchestrator) execDSLNode(ctx context.Context, t *contract.Task, n dsl.Node, b *board, plan dsl.Plan, tr *execTracer) {
+ switch n.Kind {
+ case "input":
+ if txt := cstr(n.Config, "text"); txt != "" {
+ b.query = txt
+ }
+ tr.info("input:"+n.ID, "system", labelOf(n, "输入"), truncate(b.query, 80))
+ case "memory":
+ if cbool(n.Config, "profile") {
+ b.profile = o.fetchMemory(ctx, b.uid, b.query)
+ }
+ if cbool(n.Config, "history") {
+ b.history = o.fetchHistory(ctx, b.sid)
+ }
+ tr.info("memory:"+n.ID, "memory", labelOf(n, "记忆"),
+ fmt.Sprintf("画像 %d 字 · 历史 %d 条", len([]rune(b.profile)), len(b.history)))
+ case "retriever":
+ o.retrieverNode(ctx, n, b, tr)
+ case "tool":
+ o.execToolNode(ctx, t.ID, n, b, tr)
+ case "agent":
+ sys := firstNonEmpty(cstr(n.Config, "system"), plan.System)
+ if cbool(n.Config, "autonomous") {
+ o.runReactAgent(ctx, t.ID, b, sys, n, tr, "agent:"+n.ID)
+ } else {
+ o.runAgent(ctx, t.ID, b, sys, tr, "agent:"+n.ID)
+ }
+ case "aggregate":
+ merged := aggregate(cstr(n.Config, "strategy"), append(append([]string{}, b.refs...), b.toolOut...))
+ b.refs, b.toolOut = merged, nil
+ tr.info("aggregate:"+n.ID, "system", labelOf(n, "汇聚"), "策略："+firstNonEmpty(cstr(n.Config, "strategy"), "拼接"))
+ case "render":
+ o.renderNode(ctx, t.ID, n, b, tr)
+ case "map":
+ o.mapNode(ctx, t.ID, n, b, tr)
+ case "output":
+ tr.info("output:"+n.ID, "system", labelOf(n, "输出"), "目标："+firstNonEmpty(cstr(n.Config, "target"), "屏幕"))
+ default:
+ tr.info(n.Kind+":"+n.ID, "system", labelOf(n, n.Kind), "未识别节点，跳过")
+ }
+}
diff --git a/sundynix-dispatcher/internal/eino/compose_compiler_test.go b/sundynix-dispatcher/internal/eino/compose_compiler_test.go
new file mode 100644
index 0000000..1f39c7a
--- /dev/null
+++ b/sundynix-dispatcher/internal/eino/compose_compiler_test.go
@@ -0,0 +1,76 @@
+package eino
+
+import (
+ "context"
+ "testing"
+
+ "github.com/sundynix/sundynix-dispatcher/internal/harness"
+ "github.com/sundynix/sundynix-dispatcher/internal/llm"
+ "github.com/sundynix/sundynix-shared/contract"
+)
+
+// echoLLM 回显最后一条 user 消息内容（确定性），便于两条执行路径逐字对比。
+func echoLLM() *fakeLLM {
+ return &fakeLLM{
+ ready: true,
+ stream: func(m []llm.ChatMessage) string {
+ for i := len(m) - 1; i >= 0; i-- {
+ if m[i].Role == "user" {
+ return "ANS:" + m[i].Content
+ }
+ }
+ return "ANS:"
+ },
+ }
+}
+
+func runBoth(t *testing.T, graph string) (interp, comp string) {
+ t.Helper()
+ task := &contract.Task{ID: "t_eq", Graph: []byte(graph)}
+
+ o1 := &Orchestrator{pool: echoLLM(), breaker: harness.NewCircuitBreaker(), sink: &fakeSink{}}
+ a1, err := o1.runGraph(context.Background(), task, &execTracer{})
+ if err != nil {
+ t.Fatalf("runGraph: %v", err)
+ }
+ o2 := &Orchestrator{pool: echoLLM(), breaker: harness.NewCircuitBreaker(), sink: &fakeSink{}}
+ a2, err := o2.runComposeGraph(context.Background(), task, &execTracer{})
+ if err != nil {
+ t.Fatalf("runComposeGraph: %v", err)
+ }
+ return a1, a2
+}
+
+// TestComposeEquivalentLinear 多节点线性图：input→memory→agent→output，两路径成稿应逐字一致。
+func TestComposeEquivalentLinear(t *testing.T) {
+ graph := `{"version":"1","nodes":[
+ {"id":"in","kind":"input","config":{"text":"什么是图编排"}},
+ {"id":"m","kind":"memory","config":{}},
+ {"id":"a","kind":"agent","config":{"system":"你是助手"}},
+ {"id":"out","kind":"output","config":{}}
+ ],"edges":[
+ {"source":"in","target":"m"},{"source":"m","target":"a"},{"source":"a","target":"out"}
+ ]}`
+ interp, comp := runBoth(t, graph)
+ if interp == "" || interp != comp {
+ t.Fatalf("线性图不等价: interp=%q compose=%q", interp, comp)
+ }
+}
+
+// TestComposeEquivalentBranch 分支图：input→branch→(真)A/(假)B，条件恒真应都走 A，两路径一致。
+func TestComposeEquivalentBranch(t *testing.T) {
+ graph := `{"version":"1","nodes":[
+ {"id":"in","kind":"input","config":{"text":"hi"}},
+ {"id":"br","kind":"branch","config":{"condition":""}},
+ {"id":"a","kind":"agent","config":{"system":"A"}},
+ {"id":"b","kind":"agent","config":{"system":"B"}}
+ ],"edges":[
+ {"source":"in","target":"br"},
+ {"source":"br","target":"a","sourceHandle":"true"},
+ {"source":"br","target":"b","sourceHandle":"false"}
+ ]}`
+ interp, comp := runBoth(t, graph)
+ if interp == "" || interp != comp {
+ t.Fatalf("分支图不等价: interp=%q compose=%q", interp, comp)
+ }
+}
diff --git a/sundynix-dispatcher/internal/eino/compose_graph.go b/sundynix-dispatcher/internal/eino/compose_graph.go
new file mode 100644
index 0000000..e49d318
--- /dev/null
+++ b/sundynix-dispatcher/internal/eino/compose_graph.go
@@ -0,0 +1,97 @@
+package eino
+
+import (
+ "context"
+ "fmt"
+ "io"
+ "os"
+ "strings"
+ "time"
+
+ "github.com/cloudwego/eino/compose"
+ "github.com/cloudwego/eino/schema"
+
+ "github.com/sundynix/sundynix-dispatcher/internal/harness"
+)
+
+// composeEnabled 报告是否启用 compose.Graph 编排路径（Phase C 灰度开关，默认关 → 走自研 graph.go）。
+// 并存策略：EINO_COMPOSE=1 时对话主流程改走 Eino compose 运行时，行为对齐后再逐步退役 graph.go。
+func composeEnabled() bool { return os.Getenv("EINO_COMPOSE") == "1" }
+
+// runConversation 是对话/模型节点的统一入口：按灰度开关选 compose.Graph 或自研 runAgent。
+// 二者对外行为一致（据黑板拼消息 → 流式回流 token → 累计成稿），便于等价回归。
+func (o *Orchestrator) runConversation(ctx context.Context, taskID string, b *board, system string, tr *execTracer, node string) {
+ if composeEnabled() {
+ o.runComposeConversation(ctx, taskID, b, system, tr, node)
+ return
+ }
+ o.runAgent(ctx, taskID, b, system, tr, node)
+}
+
+// runComposeConversation 用 Eino compose.Graph 跑对话主流程：
+// START → ChatModel 节点 → END，编译为 Runnable 后流式执行；可观测经 callbacks 桥到 ExecEvent。
+// 模型未就绪 / 编译失败时降级回自研 runAgent，保证不回归。
+func (o *Orchestrator) runComposeConversation(ctx context.Context, taskID string, b *board, system string, tr *execTracer, node string) {
+ cm := o.pool.ChatModel()
+ if cm == nil {
+ o.runAgent(ctx, taskID, b, system, tr, node) // 无模型 → 走自研路径的降级桩
+ return
+ }
+
+ g := compose.NewGraph[[]*schema.Message, *schema.Message]()
+ if err := g.AddChatModelNode("model", cm); err != nil {
+ tr.info(node, "system", "compose 降级", "建图失败，退回自研路径："+err.Error())
+ o.runAgent(ctx, taskID, b, system, tr, node)
+ return
+ }
+ _ = g.AddEdge(compose.START, "model")
+ _ = g.AddEdge("model", compose.END)
+ r, err := g.Compile(ctx)
+ if err != nil {
+ tr.info(node, "system", "compose 降级", "编译失败，退回自研路径："+err.Error())
+ o.runAgent(ctx, taskID, b, system, tr, node)
+ return
+ }
+
+ rc := &RunCtx{
+ UserID: b.uid, SessionID: b.sid,
+ System: firstNonEmpty(system, defaultAgentSystem),
+ Query: b.query,
+ Profile: b.profile,
+ History: b.history,
+ ToolOut: append(append([]string{}, b.toolOut...), b.refs...),
+ Upstream: append([]string{}, b.agentOut...), // 前序协作 agent 产出 → 接力
+ }
+ msgs, _ := buildMessages(ctx, rc)
+
+ t0 := time.Now()
+ // ChatModel 的 start/end 由 composeTracer（callbacks）落轨迹，这里不再手写 emit（归一）。
+ sr, err := r.Stream(ctx, msgs, compose.WithCallbacks(composeTracer(tr, node)))
+ if err != nil {
+ tr.emit(node, "model", "error", "compose 图执行", err.Error(), time.Since(t0).Milliseconds())
+ return
+ }
+ defer sr.Close()
+
+ chunks := 0
+ var produced strings.Builder // 本节点产出（供下游 agent 接力）
+ for {
+ chunk, rerr := sr.Recv()
+ if rerr == io.EOF {
+ break
+ }
+ if rerr != nil {
+ tr.emit(node, "model", "error", "compose 图执行", rerr.Error(), time.Since(t0).Milliseconds())
+ return
+ }
+ if chunk.Content == "" {
+ continue
+ }
+ safe, _ := harness.RedactSecrets(chunk.Content) // 输出护栏：逐片脱敏
+ _ = o.sink.PublishToken(taskID, []byte(safe))
+ produced.WriteString(safe)
+ chunks++
+ }
+ o.recordAgentOutput(b, produced.String())
+ tr.info(node, "system", "compose 图", fmt.Sprintf("%d 段输出 / %d 字（Eino compose 运行时）", chunks, len([]rune(produced.String()))))
+}
diff --git a/sundynix-dispatcher/internal/eino/compose_graph_test.go b/sundynix-dispatcher/internal/eino/compose_graph_test.go
new file mode 100644
index 0000000..f5c5c6a
--- /dev/null
+++ b/sundynix-dispatcher/internal/eino/compose_graph_test.go
@@ -0,0 +1,89 @@
+package eino
+
+import (
+ "context"
+ "strings"
+ "testing"
+
+ "github.com/cloudwego/eino/components/model"
+ "github.com/cloudwego/eino/compose"
+ "github.com/cloudwego/eino/schema"
+
+ "github.com/sundynix/sundynix-dispatcher/internal/harness"
+ "github.com/sundynix/sundynix-dispatcher/internal/llm"
+)
+
+// stubModel 是实现 Eino model.BaseChatModel 的测试桩，固定回一段文本（确定性）。
+type stubModel struct{ reply string }
+
+func (s *stubModel) Generate(_ context.Context, _ []*schema.Message, _ ...model.Option) (*schema.Message, error) {
+ return schema.AssistantMessage(s.reply, nil), nil
+}
+
+func (s *stubModel) Stream(_ context.Context, _ []*schema.Message, _ ...model.Option) (*schema.StreamReader[*schema.Message], error) {
+ sr, sw := schema.Pipe[*schema.Message](1)
+ go func() {
+ sw.Send(schema.AssistantMessage(s.reply, nil), nil)
+ sw.Close()
+ }()
+ return sr, nil
+}
+
+// TestComposeGraphRuns 直接验证 compose.Graph（START→ChatModel→END）能编译并流式产出。
+func TestComposeGraphRuns(t *testing.T) {
+ g := compose.NewGraph[[]*schema.Message, *schema.Message]()
+ if err := g.AddChatModelNode("model", &stubModel{reply: "你好世界"}); err != nil {
+ t.Fatal(err)
+ }
+ _ = g.AddEdge(compose.START, "model")
+ _ = g.AddEdge("model", compose.END)
+ r, err := g.Compile(context.Background())
+ if err != nil {
+ t.Fatalf("compile: %v", err)
+ }
+ out, err := r.Invoke(context.Background(), []*schema.Message{schema.UserMessage("hi")})
+ if err != nil {
+ t.Fatalf("invoke: %v", err)
+ }
+ if out.Content != "你好世界" {
+ t.Fatalf("got %q", out.Content)
+ }
+}
+
+// TestComposeConversationEquivalent 验证 EINO_COMPOSE=1 时对话主流程走 compose 路径，
+// 把模型输出流式回流到 sink 并累计成稿——与自研 runAgent 行为等价（都逐片转发模型输出）。
+func TestComposeConversationEquivalent(t *testing.T) {
+ t.Setenv("EINO_COMPOSE", "1")
+ fs := &fakeSink{}
+ o := &Orchestrator{
+ pool: &fakeLLM{ready: true, cm: &stubModel{reply: "我是 compose 路径的回答"}},
+ breaker: harness.NewCircuitBreaker(),
+ sink: fs,
+ }
+ b := &board{query: "你好"}
+ tr := &execTracer{} // sink 为 nil → 轨迹发射空操作
+ o.runConversation(context.Background(), "task_compose", b, "", tr, "agent")
+
+ if !strings.Contains(b.answer, "compose 路径") {
+ t.Fatalf("成稿未含模型输出: %q", b.answer)
+ }
+ if !strings.Contains(fs.text(), "compose 路径") {
+ t.Fatalf("sink 未收到流式 token: %q", fs.text())
+ }
+}
+
+// TestComposeDisabledUsesRunAgent 验证开关关闭时仍走自研 runAgent（默认行为不变）。
+func TestComposeDisabledUsesRunAgent(t *testing.T) {
+ t.Setenv("EINO_COMPOSE", "0")
+ fs := &fakeSink{}
+ o := &Orchestrator{
+ pool: &fakeLLM{ready: true, stream: func([]llm.ChatMessage) string { return "自研路径回答" }},
+ breaker: harness.NewCircuitBreaker(),
+ sink: fs,
+ }
+ b := &board{query: "你好"}
+ o.runConversation(context.Background(), "task_legacy", b, "", &execTracer{}, "agent")
+ if !strings.Contains(b.answer, "自研路径") {
+ t.Fatalf("未走自研路径: %q", b.answer)
+ }
+}
diff --git a/sundynix-dispatcher/internal/eino/eino_components.go b/sundynix-dispatcher/internal/eino/eino_components.go
new file mode 100644
index 0000000..7e8b85c
--- /dev/null
+++ b/sundynix-dispatcher/internal/eino/eino_components.go
@@ -0,0 +1,62 @@
+package eino
+
+import (
+ "context"
+ "encoding/json"
+ "strings"
+
+ "github.com/cloudwego/eino/components/retriever"
+ "github.com/cloudwego/eino/schema"
+
+ "github.com/sundynix/sundynix-shared/contract"
+)
+
+const defaultRetrieveTopK = 4
+
+// ragRetriever 把 mcp-go 的混合检索（NATS kb_search）包成 Eino components/retriever.Retriever。
+// kb 在构造时绑定（owner 作用域库名）；查询经 NATS 调 mcp-go，命中转 *schema.Document。
+type ragRetriever struct {
+ caller ToolCaller
+ kb string
+ topK int
+}
+
+// newRetriever 构造一个绑定到指定库的 Eino Retriever 组件。
+func (o *Orchestrator) newRetriever(kb string) retriever.Retriever {
+ return &ragRetriever{caller: o.tools, kb: kb, topK: defaultRetrieveTopK}
+}
+
+// Retrieve 实现 retriever.Retriever：经 NATS kb_search 检索，返回命中文档（不可用时降级空）。
+func (r *ragRetriever) Retrieve(ctx context.Context, query string, _ ...retriever.Option) ([]*schema.Document, error) {
+ if r.caller == nil || r.kb == "" {
+ return nil, nil
+ }
+ topK := r.topK
+ if topK <= 0 {
+ topK = defaultRetrieveTopK
+ }
+ cctx, cancel := context.WithTimeout(ctx, toolCallTimeout)
+ defer cancel()
+ res, err := r.caller.CallTool(cctx, contract.ToolSubjectGo("kb_search"), &contract.ToolCall{
+ Tool: "kb_search", Args: map[string]any{"kb": r.kb, "q": query, "topK": topK},
+ })
+ if err != nil || res == nil || !res.OK || res.Content == "" || res.Content == "[]" {
+ return nil, nil
+ }
+ var hits []struct {
+ Text string `json:"text"`
+ Score float64 `json:"score"`
+ }
+ if json.Unmarshal([]byte(res.Content), &hits) != nil {
+ // 非结构化命中：整体作为单篇文档返回（与旧行为对齐）。
+ return []*schema.Document{{Content: res.Content}}, nil
+ }
+ docs := make([]*schema.Document, 0, len(hits))
+ for _, h := range hits {
+ docs = append(docs, &schema.Document{
+ Content: strings.TrimSpace(h.Text),
+ MetaData: map[string]any{"score": h.Score},
+ })
+ }
+ return docs, nil
+}
diff --git a/sundynix-dispatcher/internal/eino/eino_components_test.go b/sundynix-dispatcher/internal/eino/eino_components_test.go
new file mode 100644
index 0000000..8a05dcc
--- /dev/null
+++ b/sundynix-dispatcher/internal/eino/eino_components_test.go
@@ -0,0 +1,56 @@
+package eino
+
+import (
+ "context"
+ "strings"
+ "testing"
+
+ "github.com/cloudwego/eino/schema"
+
+ "github.com/sundynix/sundynix-shared/contract"
+)
+
+// TestBuildMessagesShape 锁定 ChatTemplate 产出的消息序列与旧手拼逻辑等价：
+// [system(含画像+参考资料)] + [历史...] + [user(query)]，且 JSON 花括号不被误解析。
+func TestBuildMessagesShape(t *testing.T) {
+ rc := &RunCtx{
+ System: "你是助手",
+ Query: "今天几号",
+ Profile: "称呼=Dexter",
+ History: []*schema.Message{schema.UserMessage("上一句"), schema.AssistantMessage("上一答", nil)},
+ ToolOut: []string{`{"k":"v"}`}, // 含花括号，验证 FString 不误解析
+ }
+ msgs, err := buildMessages(context.Background(), rc)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(msgs) != 4 {
+ t.Fatalf("应为 system+2历史+user=4 条，实际 %d", len(msgs))
+ }
+ if msgs[0].Role != schema.System || !strings.Contains(msgs[0].Content, "你是助手") ||
+ !strings.Contains(msgs[0].Content, "称呼=Dexter") || !strings.Contains(msgs[0].Content, `{"k":"v"}`) {
+ t.Fatalf("system 消息不对: %q", msgs[0].Content)
+ }
+ if msgs[3].Role != schema.User || msgs[3].Content != "今天几号" {
+ t.Fatalf("user 消息不对: %q", msgs[3].Content)
+ }
+}
+
+// fakeCaller 是 ToolCaller 测试替身，固定返回 kb_search 命中。
+type fakeCaller struct{ content string }
+
+func (f *fakeCaller) CallTool(_ context.Context, _ string, _ *contract.ToolCall) (*contract.ToolResult, error) {
+ return &contract.ToolResult{OK: true, Content: f.content}, nil
+}
+
+// TestRagRetrieverComponent 验证 Retriever 组件把 kb_search 命中转成 *schema.Document。
+func TestRagRetrieverComponent(t *testing.T) {
+ o := &Orchestrator{tools: &fakeCaller{content: `[{"text":"片段A","score":0.9},{"text":"片段B","score":0.8}]`}}
+ docs, err := o.newRetriever("u/kb").Retrieve(context.Background(), "q")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(docs) != 2 || docs[0].Content != "片段A" || docs[1].Content != "片段B" {
+ t.Fatalf("文档解析不对: %+v", docs)
+ }
+}
diff --git a/sundynix-dispatcher/internal/eino/graph.go b/sundynix-dispatcher/internal/eino/graph.go
index 5b95d86..5152bf0 100644
--- a/sundynix-dispatcher/internal/eino/graph.go
+++ b/sundynix-dispatcher/internal/eino/graph.go
@@ -27,7 +27,8 @@ type board struct {
refs []string // 检索 / 聚合得到的参考资料
toolOut []string // 工具节点产出
sections []reportSection // map 并行 fan-out 产出的分项成稿（供 render 多章渲染）
- answer string // 终端 agent / map 的成稿（流式累计）
+ answer string // 当前成稿（多 agent 协作时 = 最近一个 agent 的产出 = 成品）
+ agentOut []string // 各上游 agent 的产出（按序），注入下游 agent 上下文以实现接力协作
}
// runGraph 按 DSL 图的真实拓扑与连线执行（替代旧的线性拍平 compileFlow）。
@@ -50,7 +51,7 @@ func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execT
tr.info("task", "system", "无结构化图", "按单轮对话执行")
b.profile = o.fetchMemory(ctx, b.uid, b.query)
b.history = o.fetchHistory(ctx, b.sid)
- o.runAgent(ctx, t.ID, b, plan.System, tr, "agent")
+ o.runConversation(ctx, t.ID, b, plan.System, tr, "agent")
return b.answer, nil
}
@@ -123,7 +124,7 @@ func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execT
if cbool(n.Config, "autonomous") { // 开启自主工具 → ReAct（模型自己选工具）
o.runReactAgent(ctx, t.ID, b, sys, n, tr, "agent:"+n.ID)
} else {
- o.runAgent(ctx, t.ID, b, sys, tr, "agent:"+n.ID)
+ o.runConversation(ctx, t.ID, b, sys, tr, "agent:"+n.ID)
}
case "aggregate":
merged := aggregate(cstr(n.Config, "strategy"), append(append([]string{}, b.refs...), b.toolOut...))
@@ -147,7 +148,7 @@ func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execT
// 图里无 agent 节点（纯工具/检索图）也要出一段模型答复，否则没有输出。
if b.answer == "" {
- o.runAgent(ctx, t.ID, b, plan.System, tr, "agent")
+ o.runConversation(ctx, t.ID, b, plan.System, tr, "agent")
}
return b.answer, nil
}
@@ -230,16 +231,18 @@ func (o *Orchestrator) execToolNode(ctx context.Context, taskID string, n dsl.No
// runAgent 执行 agent/模型节点：据黑板拼消息 → 流式回流 token → 累计成稿。
func (o *Orchestrator) runAgent(ctx context.Context, taskID string, b *board, system string, tr *execTracer, node string) {
rc := &RunCtx{
- System: firstNonEmpty(system, defaultAgentSystem),
- Query: b.query,
- Profile: b.profile,
- History: b.history,
- ToolOut: append(append([]string{}, b.toolOut...), b.refs...),
+ System: firstNonEmpty(system, defaultAgentSystem),
+ Query: b.query,
+ Profile: b.profile,
+ History: b.history,
+ ToolOut: append(append([]string{}, b.toolOut...), b.refs...),
+ Upstream: append([]string{}, b.agentOut...), // 前序协作 agent 产出 → 接力
}
msgs, _ := buildMessages(ctx, rc)
tr.emit(node, "model", "start", "模型流式推理", "", 0)
t0 := time.Now()
n, redacted := 0, 0
+ var produced strings.Builder // 本节点自身产出（用于沿图向下游传递）
send := func(s string) {
if s == "" {
return
@@ -248,7 +251,7 @@ func (o *Orchestrator) runAgent(ctx context.Context, taskID string, b *board, sy
safe, hit := harness.RedactSecrets(s)
redacted += hit
_ = o.sink.PublishToken(taskID, []byte(safe))
- b.answer += safe
+ produced.WriteString(safe)
n++
}
var err error
@@ -264,8 +267,19 @@ func (o *Orchestrator) runAgent(ctx context.Context, taskID string, b *board, sy
if redacted > 0 {
tr.info(node, "system", "输出护栏", fmt.Sprintf("已脱敏 %d 处疑似密钥/令牌", redacted))
}
+ o.recordAgentOutput(b, produced.String()) // 产出入黑板：成当前成稿 + 供下游接力
tr.emit(node, "model", "end", "模型流式推理",
- fmt.Sprintf("%d tokens / %d 字", n, len([]rune(b.answer))), time.Since(t0).Milliseconds())
+ fmt.Sprintf("%d tokens / %d 字", n, len([]rune(produced.String()))), time.Since(t0).Milliseconds())
+}
+
+// recordAgentOutput 把一个 agent 节点的产出记入黑板：append 到 agentOut（供下游 agent 注入接力），
+// 并设为当前成稿 answer（多 agent 协作时，最后一个 agent 的产出即最终成品）。空产出忽略。
+func (o *Orchestrator) recordAgentOutput(b *board, out string) {
+ if strings.TrimSpace(out) == "" {
+ return
+ }
+ b.agentOut = append(b.agentOut, out)
+ b.answer = out
}
// renderNode 执行渲染节点：把当前成稿渲染成 Word（经 mcp-go report_render）。
diff --git a/sundynix-dispatcher/internal/eino/integration_test.go b/sundynix-dispatcher/internal/eino/integration_test.go
index 50da8fc..b3577c6 100644
--- a/sundynix-dispatcher/internal/eino/integration_test.go
+++ b/sundynix-dispatcher/internal/eino/integration_test.go
@@ -20,6 +20,7 @@ type fakeLLM struct {
ready bool
stream func(msgs []llm.ChatMessage) string // ChatStream 要回流的整段文本
chat func(msgs []llm.ChatMessage) (string, error) // Chat 返回
+ cm model.BaseChatModel // compose 路径用的 Eino 模型（可为 nil）
}
func (f *fakeLLM) Ready() bool { return f.ready }
@@ -43,6 +44,9 @@ func (f *fakeLLM) Chat(_ context.Context, msgs []llm.ChatMessage) (string, error
// ToolCallingModel：假模型不支持函数调用 → ReAct 路径会降级回普通对话。
func (f *fakeLLM) ToolCallingModel() model.ToolCallingChatModel { return nil }
+// ChatModel：compose 路径用；nil 时 runComposeConversation 降级回 runAgent。
+func (f *fakeLLM) ChatModel() model.BaseChatModel { return f.cm }
+
type fakeSink struct {
mu sync.Mutex
tokens []string
diff --git a/sundynix-dispatcher/internal/eino/orchestrator.go b/sundynix-dispatcher/internal/eino/orchestrator.go
index 5e6e197..448eb6c 100644
--- a/sundynix-dispatcher/internal/eino/orchestrator.go
+++ b/sundynix-dispatcher/internal/eino/orchestrator.go
@@ -4,6 +4,7 @@ package eino
import (
"context"
"encoding/json"
+ "errors"
"fmt"
"log"
"sync"
@@ -29,6 +30,11 @@ type ToolCaller interface {
CallTool(ctx context.Context, subject string, call *contract.ToolCall) (*contract.ToolResult, error)
}
+// StatusSink 回写任务生命周期状态（由 NATS bus 实现；可为 nil → 不回写）。
+type StatusSink interface {
+ PublishTaskStatus(taskID, status, detail string) error
+}
+
// LLM 是编排所需的语言模型能力（生产由 *llm.Pool 实现）。抽成接口便于测试注入假模型。
type LLM interface {
Ready() bool
@@ -37,11 +43,16 @@ type LLM interface {
Chat(ctx context.Context, msgs []llm.ChatMessage) (string, error)
// ToolCallingModel 返回支持函数调用的模型（ReAct agent 用）；不支持则返回 nil。
ToolCallingModel() model.ToolCallingChatModel
+ // ChatModel 返回 Eino ChatModel 组件（compose.Graph 编排用）；未就绪则 nil。
+ ChatModel() model.BaseChatModel
}
// 工具调用超时；超时即降级（不带工具上下文继续推理）。
const toolCallTimeout = 3 * time.Second
+// taskExecTimeout 是单个任务整体执行上限；超时即判 timeout（状态机），避免无限期"运行中"。
+const taskExecTimeout = 3 * time.Minute
+
// Orchestrator 把每个 DSL 任务动态编译为 Eino 图并执行（记忆召回 → 工具节点 → 注入 → 流式）。
type Orchestrator struct {
pool LLM
@@ -50,19 +61,48 @@ type Orchestrator struct {
sink TokenSink
tools ToolCaller
exec ExecSink
+ status StatusSink // 任务生命周期状态回写（可为 nil）
turnMu sync.Mutex // 保护 turns（攒批计数，多任务 goroutine 共享）
turns map[string]int // sessionID → 累计轮次，用于每 N 轮触发 consolidate
}
// NewOrchestrator 持有依赖；图按任务的 DSL 在 Handle 内动态编译。
-// exec 为执行可视化事件出口（可为 nil，则不发轨迹事件）；eval 为自动化评测（可为 nil）。
-func NewOrchestrator(pool LLM, breaker *harness.CircuitBreaker, eval *harness.Evaluator, sink TokenSink, tools ToolCaller, exec ExecSink) (*Orchestrator, error) {
- return &Orchestrator{pool: pool, breaker: breaker, eval: eval, sink: sink, tools: tools, exec: exec}, nil
+// exec 为执行可视化事件出口（可为 nil，则不发轨迹事件）；eval 为自动化评测（可为 nil）；
+// status 为任务生命周期状态回写出口（可为 nil）。
+func NewOrchestrator(pool LLM, breaker *harness.CircuitBreaker, eval *harness.Evaluator, sink TokenSink, tools ToolCaller, exec ExecSink, status StatusSink) (*Orchestrator, error) {
+ return &Orchestrator{pool: pool, breaker: breaker, eval: eval, sink: sink, tools: tools, exec: exec, status: status}, nil
+}
+
+// setStatus 回写一次任务状态流转（status 为 nil 时静默跳过）。
+func (o *Orchestrator) setStatus(taskID, status, detail string) {
+ if o.status == nil {
+ return
+ }
+ if err := o.status.PublishTaskStatus(taskID, status, detail); err != nil {
+ log.Printf("[eino] 回写任务状态 %s=%s 失败: %v", taskID, status, err)
+ }
+}
+
+// finishStatus 据收尾错误把任务置为 done / timeout / failed。
+func (o *Orchestrator) finishStatus(taskID string, err error) {
+ switch {
+ case err == nil:
+ o.setStatus(taskID, contract.TaskDone, "")
+ case errors.Is(err, context.DeadlineExceeded):
+ o.setStatus(taskID, contract.TaskTimeout, "执行超时")
+ default:
+ o.setStatus(taskID, contract.TaskFailed, truncate(err.Error(), 200))
+ }
}
// Handle 消费一个任务：按 DSL 编译 Eino 图并执行，把 Token 流回流到 sundynix.streams.。
func (o *Orchestrator) Handle(ctx context.Context, t *contract.Task) error {
+ // 护栏：丢弃空任务（无 id），避免误投/历史脏数据被当真任务处理并触发状态回写放大。
+ if t.ID == "" {
+ log.Printf("[eino] 跳过空任务（无 id）")
+ return nil
+ }
tr := o.tracer(t.ID)
defer tr.done()
@@ -72,22 +112,31 @@ func (o *Orchestrator) Handle(ctx context.Context, t *contract.Task) error {
tr.info("task", "system", "服务熔断", "后端连续失败，暂时拒绝新任务，请稍后重试")
_ = o.sink.PublishToken(t.ID, []byte("⚠️ 服务繁忙（已触发熔断保护），请稍后重试。"))
_ = o.sink.CompleteStream(t.ID)
+ o.setStatus(t.ID, contract.TaskFailed, "服务熔断")
return nil
}
+ // 任务状态机：进入执行 → running；整体加超时上限，超时判 timeout（杜绝无限期"运行中"）。
+ o.setStatus(t.ID, contract.TaskRunning, "")
+ tctx, cancel := context.WithTimeout(ctx, taskExecTimeout)
+ defer cancel()
+
// 报告生成走专用多步编排（规划→分章并行检索撰写→汇聚→渲染 Word），而非通用对话图。
if intent, _ := t.Meta[contract.MetaIntent].(string); intent == contract.IntentReport {
- return o.handleReport(ctx, t, tr)
+ err := o.handleReport(tctx, t, tr)
+ o.finishStatus(t.ID, err)
+ return err
}
log.Printf("[eino] task %s received (graph=%d bytes), 按图执行（拓扑+连线+分支）...", t.ID, len(t.Graph))
tr.info("task", "system", "任务受理", fmt.Sprintf("DSL %d 字节，按图执行", len(t.Graph)))
- // 按 DSL 图的真实拓扑/连线/分支执行（graph.go 解释器），agent 节点流式回流 token。
- answer, err := o.runGraph(ctx, t, tr)
+ // 按 DSL 图执行：compose.Graph（EINO_COMPOSE=1）或自研 graph.go（默认）；agent 节点流式回流 token。
+ answer, err := o.executeGraph(tctx, t, tr)
if err != nil {
log.Printf("[eino] task %s graph error: %v", t.ID, err)
_ = o.sink.CompleteStream(t.ID)
o.breaker.Report(false)
+ o.finishStatus(t.ID, err)
return err
}
@@ -96,6 +145,7 @@ func (o *Orchestrator) Handle(ctx context.Context, t *contract.Task) error {
}
log.Printf("[eino] task %s done (%d 字答复)", t.ID, len([]rune(answer)))
o.breaker.Report(true)
+ o.finishStatus(t.ID, nil)
// 写回阶段：离开热路径、异步落历史 + （TODO）抽取记忆。
go o.memorize(t, answer)
diff --git a/sundynix-dispatcher/internal/eino/react_agent.go b/sundynix-dispatcher/internal/eino/react_agent.go
index 7e8f922..52a87eb 100644
--- a/sundynix-dispatcher/internal/eino/react_agent.go
+++ b/sundynix-dispatcher/internal/eino/react_agent.go
@@ -6,6 +6,7 @@ import (
"fmt"
"io"
"log"
+ "strings"
"time"
"github.com/cloudwego/eino/components/tool"
@@ -85,42 +86,84 @@ func (m *mcpTool) InvokableRun(ctx context.Context, argsJSON string, _ ...tool.O
return res.Content, nil
}
-// agentTools 构建 ReAct 可用的工具集（按当前任务上下文绑定 uid/kb）。
-// 工具名是给"模型看"的语义名；mcpName 是实际 NATS 调用名。context 参数（user_id）服务端注入，不暴露给模型。
+// toolCatalogEntry 是 MCP list_tools 上报的一条工具元信息（与 mcp-go listTools 输出对齐）。
+type toolCatalogEntry struct {
+ Name string `json:"name"`
+ CN string `json:"cn"`
+ Desc string `json:"desc"`
+ Agent bool `json:"agent_exposed"`
+ AgentName string `json:"agent_name"`
+ Params []struct {
+ Name string `json:"name"`
+ Type string `json:"type"`
+ Desc string `json:"desc"`
+ Required bool `json:"required"`
+ } `json:"params"`
+ Inject []string `json:"inject"`
+}
+
+// agentTools 动态构建 ReAct 可用的工具集：分别向 mcp-go / mcp-py 探 list_tools 自描述目录，
+// 取 agent_exposed 的工具按上报参数 schema 建 InvokableTool；inject 参数（user_id/session_id/
+// kb/task_id）服务端运行时绑定、不暴露给模型。某台 MCP 离线即跳过（降级）。
+// 新增工具只需在对应 MCP 注册表标 agent，无需改这里（杜绝硬编码）。
func (o *Orchestrator) agentTools(b *board, taskID string, tr *execTracer) []tool.BaseTool {
if o.tools == nil {
return nil
}
- kbBind := map[string]any{}
- if b.kb != "" {
- kbBind["kb"] = b.kb
+ var out []tool.BaseTool
+ out = append(out, o.discoverTools(contract.ToolSubjectGo, b, taskID, tr)...)
+ out = append(out, o.discoverTools(contract.ToolSubjectPy, b, taskID, tr)...)
+ return out
+}
+
+// discoverTools 向某台 MCP（subject 前缀决定 go/py）探 list_tools，把 agent_exposed 的工具
+// 转成 Eino InvokableTool。该 MCP 不可用 / 无应答时返回空（不阻断）。
+func (o *Orchestrator) discoverTools(subject func(string) string, b *board, taskID string, tr *execTracer) []tool.BaseTool {
+ cctx, cancel := context.WithTimeout(context.Background(), toolCallTimeout)
+ defer cancel()
+ res, err := o.tools.CallTool(cctx, subject("list_tools"), &contract.ToolCall{Tool: "list_tools"})
+ if err != nil || res == nil || !res.OK {
+ return nil
}
- return []tool.BaseTool{
- &mcpTool{
- mcpName: "wiki_search",
- subject: contract.ToolSubjectGo,
- caller: o.tools, taskID: taskID, tr: tr,
- bind: kbBind,
- info: &schema.ToolInfo{
- Name: "wiki_search",
- Desc: "检索知识库，返回与查询最相关的资料片段。需要外部知识/事实依据时调用。",
- ParamsOneOf: schema.NewParamsOneOfByParams(map[string]*schema.ParameterInfo{
- "q": {Type: schema.String, Desc: "检索查询语句", Required: true},
- }),
- },
- },
- &mcpTool{
- mcpName: "memory_get",
- subject: contract.ToolSubjectGo,
- caller: o.tools, taskID: taskID, tr: tr,
- bind: map[string]any{"user_id": b.uid},
- info: &schema.ToolInfo{
- Name: "recall_user_memory",
- Desc: "召回当前用户的长期画像与偏好（称呼/职业/回答偏好等）。需要个性化、了解“我是谁”时调用。",
- ParamsOneOf: schema.NewParamsOneOfByParams(map[string]*schema.ParameterInfo{}),
- },
- },
+ var cat struct {
+ Tools []toolCatalogEntry `json:"tools"`
}
+ if json.Unmarshal([]byte(res.Content), &cat) != nil {
+ return nil
+ }
+ injectVal := map[string]any{"user_id": b.uid, "session_id": b.sid, "task_id": taskID, "kb": b.kb}
+
+ var out []tool.BaseTool
+ for _, e := range cat.Tools {
+ if !e.Agent {
+ continue
+ }
+ params := map[string]*schema.ParameterInfo{}
+ for _, p := range e.Params {
+ params[p.Name] = &schema.ParameterInfo{Type: schema.DataType(p.Type), Desc: p.Desc, Required: p.Required}
+ }
+ bind := map[string]any{}
+ for _, inj := range e.Inject {
+ if v, ok := injectVal[inj]; ok && v != "" {
+ bind[inj] = v
+ }
+ }
+ name := e.AgentName
+ if name == "" {
+ name = e.Name
+ }
+ out = append(out, &mcpTool{
+ mcpName: e.Name,
+ subject: subject,
+ caller: o.tools, taskID: taskID, tr: tr,
+ bind: bind,
+ info: &schema.ToolInfo{
+ Name: name, Desc: e.Desc,
+ ParamsOneOf: schema.NewParamsOneOfByParams(params),
+ },
+ })
+ }
+ return out
}
// runReactAgent 执行带"自主工具"的 agent 节点：模型在 ReAct 循环里自行决定调哪些 MCP 工具。
@@ -153,9 +196,10 @@ func (o *Orchestrator) runReactAgent(ctx context.Context, taskID string, b *boar
System: firstNonEmpty(system, defaultAgentSystem),
Query: b.query,
// 自主 agent 不预注入画像：让它经 recall_user_memory 工具按需自取（否则模型直接答、不调工具）。
- Profile: "",
- History: b.history,
- ToolOut: append(append([]string{}, b.toolOut...), b.refs...),
+ Profile: "",
+ History: b.history,
+ ToolOut: append(append([]string{}, b.toolOut...), b.refs...),
+ Upstream: append([]string{}, b.agentOut...), // 前序协作 agent 产出 → 接力
}
msgs, _ := buildMessages(ctx, rc)
@@ -169,6 +213,7 @@ func (o *Orchestrator) runReactAgent(ctx context.Context, taskID string, b *boar
defer sr.Close()
chunks := 0
+ var produced strings.Builder // 本节点产出（供下游 agent 接力）
for {
chunk, rerr := sr.Recv()
if rerr == io.EOF {
@@ -183,9 +228,10 @@ func (o *Orchestrator) runReactAgent(ctx context.Context, taskID string, b *boar
}
safe, _ := harness.RedactSecrets(chunk.Content)
_ = o.sink.PublishToken(taskID, []byte(safe))
- b.answer += safe
+ produced.WriteString(safe)
chunks++
}
+ o.recordAgentOutput(b, produced.String())
tr.emit(node, "model", "end", "ReAct 智能体",
- fmt.Sprintf("%d 段输出 / %d 字", chunks, len([]rune(b.answer))), time.Since(t0).Milliseconds())
+ fmt.Sprintf("%d 段输出 / %d 字", chunks, len([]rune(produced.String()))), time.Since(t0).Milliseconds())
}
diff --git a/sundynix-dispatcher/internal/eino/report.go b/sundynix-dispatcher/internal/eino/report.go
index ce15a98..679159f 100644
--- a/sundynix-dispatcher/internal/eino/report.go
+++ b/sundynix-dispatcher/internal/eino/report.go
@@ -214,29 +214,15 @@ func (o *Orchestrator) writeSection(ctx context.Context, topic, kb, heading stri
return strings.TrimSpace(txt)
}
-// retrieve 经 mcp-go kb_search 工具检索知识库，整理为可读参考资料。kb 为空或无召回则返回空。
+// retrieve 经 Eino Retriever 组件（包 mcp-go kb_search）检索知识库，整理为可读参考资料。
func (o *Orchestrator) retrieve(ctx context.Context, kb, query string) string {
- if o.tools == nil || kb == "" {
+ docs, err := o.newRetriever(kb).Retrieve(ctx, query)
+ if err != nil || len(docs) == 0 {
return ""
}
- cctx, cancel := context.WithTimeout(ctx, toolCallTimeout)
- defer cancel()
- res, err := o.tools.CallTool(cctx, contract.ToolSubjectGo("kb_search"), &contract.ToolCall{
- Tool: "kb_search", Args: map[string]any{"kb": kb, "q": query, "topK": 4},
- })
- if err != nil || res == nil || !res.OK || res.Content == "" || res.Content == "[]" {
- return ""
- }
- var hits []struct {
- Text string `json:"text"`
- Score float64 `json:"score"`
- }
- if json.Unmarshal([]byte(res.Content), &hits) != nil {
- return res.Content
- }
var b strings.Builder
- for i, h := range hits {
- fmt.Fprintf(&b, "%d. %s\n", i+1, strings.TrimSpace(h.Text))
+ for i, d := range docs {
+ fmt.Fprintf(&b, "%d. %s\n", i+1, strings.TrimSpace(d.Content))
}
return b.String()
}
diff --git a/sundynix-dispatcher/internal/llm/pool.go b/sundynix-dispatcher/internal/llm/pool.go
index 0b40538..8901603 100644
--- a/sundynix-dispatcher/internal/llm/pool.go
+++ b/sundynix-dispatcher/internal/llm/pool.go
@@ -76,6 +76,9 @@ func (p *Pool) model() model.BaseChatModel {
// Ready 报告是否已配置可用后端（且 ChatModel 构建成功）。
func (p *Pool) Ready() bool { return p.model() != nil }
+// ChatModel 返回当前 Eino ChatModel 组件（用于 compose.Graph 编排）；未就绪则 nil。
+func (p *Pool) ChatModel() model.BaseChatModel { return p.model() }
+
// ToolCallingModel 返回支持函数调用的模型（用于 ReAct agent）；未就绪 / 不支持则 nil。
func (p *Pool) ToolCallingModel() model.ToolCallingChatModel {
if tcm, ok := p.model().(model.ToolCallingChatModel); ok {
diff --git a/sundynix-dispatcher/internal/nats/subscriber.go b/sundynix-dispatcher/internal/nats/subscriber.go
index 99f3db9..9353e1a 100644
--- a/sundynix-dispatcher/internal/nats/subscriber.go
+++ b/sundynix-dispatcher/internal/nats/subscriber.go
@@ -4,6 +4,7 @@ package nats
import (
"context"
"log"
+ "time"
sharedbus "github.com/sundynix/sundynix-shared/bus"
"github.com/sundynix/sundynix-shared/contract"
@@ -73,6 +74,13 @@ func (s *Subscriber) ServeHealth(provide func() []byte) (func() error, error) {
return s.inner.ServeHealth(contract.SubjectHealthDispatcher, provide)
}
+// PublishTaskStatus 让 Subscriber 满足 eino.StatusSink，把任务状态流转回写给网关。
+func (s *Subscriber) PublishTaskStatus(taskID, status, detail string) error {
+ return s.inner.PublishTaskStatus(&contract.TaskStatusEvent{
+ TaskID: taskID, Status: status, Detail: detail, TS: time.Now().UnixMilli(),
+ })
+}
+
// RequestModelConfig 向控制面（Gateway）取当前激活的对话模型配置。
func (s *Subscriber) RequestModelConfig(ctx context.Context) (*contract.ModelConfig, error) {
return s.inner.RequestConfig(ctx, contract.ConfigKindChat)
diff --git a/sundynix-gateway/cmd/server/main.go b/sundynix-gateway/cmd/server/main.go
index 7652a0a..53c52bd 100644
--- a/sundynix-gateway/cmd/server/main.go
+++ b/sundynix-gateway/cmd/server/main.go
@@ -46,6 +46,15 @@ func main() {
}
}
+ // 任务生命周期：订阅 dispatcher 回写的状态流转（running/done/failed/timeout），落 PG 供 UI 查询。
+ if _, err := bus.SubscribeTaskStatus(func(ev *contract.TaskStatusEvent) {
+ if err := db.UpdateTaskStatus(context.Background(), ev.TaskID, ev.Status, ev.Detail); err != nil {
+ log.Printf("[gateway] 更新任务状态 %s=%s 失败: %v", ev.TaskID, ev.Status, err)
+ }
+ }); err != nil {
+ log.Printf("[gateway] subscribe task status: %v", err)
+ }
+
r := router.New(db, cache, bus, blobStore)
addr := envOr("GATEWAY_ADDR", ":8080")
log.Printf("[gateway] listening on %s", addr)
diff --git a/sundynix-gateway/go.mod b/sundynix-gateway/go.mod
index 60c531a..af49abe 100644
--- a/sundynix-gateway/go.mod
+++ b/sundynix-gateway/go.mod
@@ -4,6 +4,7 @@ go 1.25.0
require (
github.com/bwmarrin/snowflake v0.3.0
+ github.com/gin-contrib/sse v0.1.0
github.com/gin-gonic/gin v1.10.0
github.com/golang-jwt/jwt/v5 v5.3.1
github.com/minio/minio-go/v7 v7.2.0
@@ -26,7 +27,6 @@ require (
github.com/cloudwego/base64x v0.1.6 // indirect
github.com/dustin/go-humanize v1.0.1 // indirect
github.com/gabriel-vasile/mimetype v1.4.3 // indirect
- github.com/gin-contrib/sse v0.1.0 // indirect
github.com/go-playground/locales v0.14.1 // indirect
github.com/go-playground/universal-translator v0.18.1 // indirect
github.com/go-playground/validator/v10 v10.20.0 // indirect
diff --git a/sundynix-gateway/internal/handler/task_handler.go b/sundynix-gateway/internal/handler/task_handler.go
index 28eca24..b4c6be8 100644
--- a/sundynix-gateway/internal/handler/task_handler.go
+++ b/sundynix-gateway/internal/handler/task_handler.go
@@ -9,6 +9,7 @@ import (
"net/http"
"time"
+ "github.com/gin-contrib/sse"
"github.com/gin-gonic/gin"
"github.com/sundynix/sundynix-gateway/internal/blob"
@@ -53,23 +54,86 @@ func (h *Handler) SubmitTask(c *gin.Context) {
c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
return
}
+ // 从提交即开始把 token 流录进 Redis Stream（订阅早于 dispatcher 产 token）→
+ // SSE 可从中回放/断点续传，根治"连晚/重连丢 token"。
+ h.startTokenRecorder(task.ID)
c.JSON(http.StatusAccepted, gin.H{"task_id": task.ID})
}
-// StreamTask: 订阅 sundynix.streams.，以 SSE 把零拷贝 Token Stream 推给客户端。
+// startTokenRecorder 后台订阅 token 流并落 Redis Stream，与 SSE 客户端是否在线无关。
+// Redis 降级时为空操作（SSE 自动回退到 live NATS 路径）。
+func (h *Handler) startTokenRecorder(taskID string) {
+ if !h.cache.Enabled() {
+ return
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), 6*time.Minute) // 兜底防泄漏
+ unsub, err := h.bus.SubscribeTokens(taskID,
+ func(tok []byte) { _ = h.cache.StreamAppend(ctx, taskID, "token", string(tok)) },
+ func() { _ = h.cache.StreamAppend(ctx, taskID, "done", ""); cancel() },
+ )
+ if err != nil {
+ cancel()
+ return
+ }
+ go func() { <-ctx.Done(); _ = unsub() }()
+}
+
+// TaskStatus: GET /api/v1/tasks/:id —— 返回任务生命周期状态（供 UI 轮询，根治"卡运行中看不出来"）。
+func (h *Handler) TaskStatus(c *gin.Context) {
+ id := c.Param("id")
+ status, detail := h.db.GetTaskStatus(c.Request.Context(), id)
+ if status == "" {
+ c.JSON(http.StatusNotFound, gin.H{"error": "task not found"})
+ return
+ }
+ c.JSON(http.StatusOK, gin.H{"task_id": id, "status": status, "detail": detail})
+}
+
+// StreamTask: 以 SSE 把 Token Stream 推给客户端。
+// 优先从 Redis Stream 读（可回放 + 断点续传，根治连晚/重连丢 token）；Redis 降级时回退 live NATS。
func (h *Handler) StreamTask(c *gin.Context) {
taskID := c.Param("id")
c.Writer.Header().Set("Content-Type", "text/event-stream")
c.Writer.Header().Set("Cache-Control", "no-cache")
c.Writer.Header().Set("Connection", "keep-alive")
+ if !h.cache.Enabled() {
+ h.streamTaskLive(c, taskID) // Redis 不可用 → 旧的 live-NATS 路径兜底
+ return
+ }
+
+ // 断点续传：浏览器 EventSource 重连会带 Last-Event-ID；缺省 "0" 从头回放。
+ lastID := c.GetHeader("Last-Event-ID")
+ if lastID == "" {
+ lastID = "0"
+ }
+ ctx := c.Request.Context()
+ c.Stream(func(w io.Writer) bool {
+ entries, nl, err := h.cache.StreamRead(ctx, taskID, lastID, 20*time.Second)
+ if err != nil {
+ return false // ctx 取消（客户端断开）或后端故障
+ }
+ lastID = nl
+ for _, e := range entries {
+ if e.Kind == "done" {
+ _ = sse.Encode(w, sse.Event{Id: e.ID, Event: "done", Data: taskID})
+ return false
+ }
+ _ = sse.Encode(w, sse.Event{Id: e.ID, Event: "token", Data: e.Data})
+ }
+ return true // 继续阻塞读取后续 token
+ })
+}
+
+// streamTaskLive 是 Redis 降级时的兜底：直接订阅 NATS token 流转 SSE（无回放/续传能力）。
+func (h *Handler) streamTaskLive(c *gin.Context, taskID string) {
tokens := make(chan []byte, 256)
done := make(chan struct{})
unsub, err := h.bus.SubscribeTokens(taskID,
func(tok []byte) {
select {
case tokens <- tok:
- default: // 背压保护：客户端过慢则丢弃，避免阻塞 NATS 回调
+ default:
}
},
func() { close(done) },
@@ -79,8 +143,6 @@ func (h *Handler) StreamTask(c *gin.Context) {
return
}
defer func() { _ = unsub() }()
-
- // gin 的流式写：返回 false 即结束响应。
c.Stream(func(w io.Writer) bool {
select {
case tok := <-tokens:
diff --git a/sundynix-gateway/internal/nats/publisher.go b/sundynix-gateway/internal/nats/publisher.go
index cd328f8..181a8ea 100644
--- a/sundynix-gateway/internal/nats/publisher.go
+++ b/sundynix-gateway/internal/nats/publisher.go
@@ -59,6 +59,11 @@ func (b *Bus) Ping(ctx context.Context, subject string) ([]byte, error) {
return b.inner.Ping(ctx, subject)
}
+// SubscribeTaskStatus 订阅 dispatcher 回写的任务生命周期状态（落 PG）。
+func (b *Bus) SubscribeTaskStatus(onEvent func(*contract.TaskStatusEvent)) (func() error, error) {
+ return b.inner.SubscribeTaskStatus(onEvent)
+}
+
// ServeConfig 让网关作为配置控制面，响应某 kind 的配置请求。
func (b *Bus) ServeConfig(kind string, provide func() *contract.ModelConfig) (func() error, error) {
return b.inner.ServeConfig(kind, provide)
diff --git a/sundynix-gateway/internal/router/router.go b/sundynix-gateway/internal/router/router.go
index 0bb7275..1bff4f2 100644
--- a/sundynix-gateway/internal/router/router.go
+++ b/sundynix-gateway/internal/router/router.go
@@ -49,6 +49,7 @@ func New(db *store.Postgres, cache *store.Redis, bus *nats.Bus, blobStore *blob.
p := api.Group("", middleware.RequireAuth())
{
p.POST("/tasks", h.SubmitTask) // 解析 DSL 并 Publish 到 NATS（带已验证 uid）
+ p.GET("/tasks/:id", h.TaskStatus) // 任务生命周期状态（UI 轮询 submitted/running/done/failed/timeout）
p.PUT("/memory", h.SetMemory) // 偏好记忆登记（→ mcp-go memory_upsert）
p.GET("/memory", h.ListMemory) // 列出当前用户偏好（记忆面板）
p.DELETE("/memory", h.DeleteMemory) // 软删一条偏好（?key=）
diff --git a/sundynix-gateway/internal/store/models.go b/sundynix-gateway/internal/store/models.go
index fbe4ae1..81092f3 100644
--- a/sundynix-gateway/internal/store/models.go
+++ b/sundynix-gateway/internal/store/models.go
@@ -19,5 +19,6 @@ type Task struct {
BaseModel
TaskID string `gorm:"uniqueIndex;size:64"` // task_xxx
Graph string `gorm:"type:jsonb"` // React Flow 导出的 DSL 原文
- Status string `gorm:"size:32"` // submitted / done / failed
+ Status string `gorm:"size:32"` // submitted / running / done / failed / timeout
+ Detail string `gorm:"type:text"` // 失败/超时原因等（状态机回写）
}
diff --git a/sundynix-gateway/internal/store/pgsql.go b/sundynix-gateway/internal/store/pgsql.go
index 0e4c751..8b011f7 100644
--- a/sundynix-gateway/internal/store/pgsql.go
+++ b/sundynix-gateway/internal/store/pgsql.go
@@ -9,6 +9,8 @@ import (
"gorm.io/driver/postgres"
"gorm.io/gorm"
"gorm.io/gorm/schema"
+
+ "github.com/sundynix/sundynix-shared/contract"
)
// errStoreDisabled 表示 Postgres 处于降级（未连接）模式，写操作无法进行。
@@ -93,7 +95,29 @@ func (p *Postgres) SaveTask(ctx context.Context, id, graph string) error {
if p.db == nil {
return nil
}
- return p.db.WithContext(ctx).Create(&Task{TaskID: id, Graph: graph, Status: "submitted"}).Error
+ return p.db.WithContext(ctx).Create(&Task{TaskID: id, Graph: graph, Status: contract.TaskSubmitted}).Error
+}
+
+// UpdateTaskStatus 流转任务状态（running/done/failed/timeout），由 dispatcher 经 NATS 回写驱动。
+func (p *Postgres) UpdateTaskStatus(ctx context.Context, id, status, detail string) error {
+ if p.db == nil {
+ return nil
+ }
+ return p.db.WithContext(ctx).Model(&Task{}).
+ Where("task_id = ?", id).
+ Updates(map[string]any{"status": status, "detail": detail}).Error
+}
+
+// GetTaskStatus 取一条任务的当前状态（供 UI 轮询；不存在返回空串）。
+func (p *Postgres) GetTaskStatus(ctx context.Context, id string) (status, detail string) {
+ if p.db == nil {
+ return "", ""
+ }
+ var t Task
+ if err := p.db.WithContext(ctx).Select("status", "detail").Where("task_id = ?", id).First(&t).Error; err != nil {
+ return "", ""
+ }
+ return t.Status, t.Detail
}
// CountTasks 返回已提交任务数（降级模式返回 0）。
diff --git a/sundynix-gateway/internal/store/redis.go b/sundynix-gateway/internal/store/redis.go
index 798f427..03a06fd 100644
--- a/sundynix-gateway/internal/store/redis.go
+++ b/sundynix-gateway/internal/store/redis.go
@@ -48,6 +48,64 @@ func (r *Redis) Allow(ctx context.Context, key string, limit int64, window time.
return n <= limit, nil
}
+// ---- Token 流持久化（Redis Stream：可回放的追加日志，根治 SSE 连晚/重连丢 token）----
+
+const streamTTL = 10 * time.Minute
+
+func streamKey(taskID string) string { return "sundynix:stream:" + taskID }
+
+// StreamEntry 是 token 流里的一条记录（ID 用于 SSE 的 Last-Event-ID 断点续传）。
+type StreamEntry struct {
+ ID string
+ Kind string // token / done
+ Data string
+}
+
+// StreamAppend 把一条 token / 结束标记追加到任务的 Redis Stream（带 TTL 自动清理）。
+func (r *Redis) StreamAppend(ctx context.Context, taskID, kind, data string) error {
+ if r.rdb == nil {
+ return nil
+ }
+ k := streamKey(taskID)
+ if err := r.rdb.XAdd(ctx, &redis.XAddArgs{Stream: k, Values: map[string]any{"kind": kind, "data": data}}).Err(); err != nil {
+ return err
+ }
+ return r.rdb.Expire(ctx, k, streamTTL).Err()
+}
+
+// StreamRead 从 lastID 之后阻塞读取新条目（XREAD BLOCK）。lastID="0" 表示从头回放。
+// 阻塞超时无新数据时返回空切片 + 原 lastID（调用方据此继续轮询）。
+func (r *Redis) StreamRead(ctx context.Context, taskID, lastID string, block time.Duration) ([]StreamEntry, string, error) {
+ if r.rdb == nil {
+ return nil, lastID, nil
+ }
+ res, err := r.rdb.XRead(ctx, &redis.XReadArgs{
+ Streams: []string{streamKey(taskID), lastID}, Block: block, Count: 256,
+ }).Result()
+ if err == redis.Nil {
+ return nil, lastID, nil // 阻塞超时、无新条目
+ }
+ if err != nil {
+ return nil, lastID, err
+ }
+ var out []StreamEntry
+ nl := lastID
+ for _, st := range res {
+ for _, m := range st.Messages {
+ out = append(out, StreamEntry{ID: m.ID, Kind: asString(m.Values["kind"]), Data: asString(m.Values["data"])})
+ nl = m.ID
+ }
+ }
+ return out, nl, nil
+}
+
+func asString(v any) string {
+ if s, ok := v.(string); ok {
+ return s
+ }
+ return ""
+}
+
// Close 释放底层连接。
func (r *Redis) Close() {
if r.rdb != nil {
diff --git a/sundynix-mcp-go/internal/mcp/gateway.go b/sundynix-mcp-go/internal/mcp/gateway.go
index 730db3b..4e758a6 100644
--- a/sundynix-mcp-go/internal/mcp/gateway.go
+++ b/sundynix-mcp-go/internal/mcp/gateway.go
@@ -32,11 +32,25 @@ type Gateway struct {
tools map[string]toolDef // 工具注册表：唯一事实源，dispatch 与 list_tools 共用，杜绝漂移
}
-// toolDef 是一个注册工具的元信息（中文名 / 作用）+ 处理函数。
+// paramSpec 是一个工具参数的声明（供自主 agent 据此生成调用入参）。
+type paramSpec struct {
+ Name string `json:"name"`
+ Type string `json:"type"` // string / number / integer / boolean / object / array
+ Desc string `json:"desc"`
+ Required bool `json:"required"`
+}
+
+// toolDef 是一个注册工具的元信息 + 处理函数。新增 agent 暴露元信息，让自主 agent 能动态发现工具：
+// agent=是否给模型自主调用；agentName=模型可见名（空=注册键）；params=模型可填参数；
+// inject=服务端运行时注入、不暴露给模型的参数名（如 user_id / session_id / kb / task_id）。
type toolDef struct {
- cn string // 中文名
- desc string // 作用简述
- handler func(context.Context, *contract.ToolCall) *contract.ToolResult
+ cn string
+ desc string
+ agent bool
+ agentName string
+ params []paramSpec
+ inject []string
+ handler func(context.Context, *contract.ToolCall) *contract.ToolResult
}
func NewGateway(b *sharedbus.Bus, s *search.Hybrid, m *memory.Store, h *history.Store, r *rag.Engine) *Gateway {
@@ -60,29 +74,53 @@ func (g *Gateway) Serve(ctx context.Context) error {
// buildRegistry 注册 mcp-go 全部工具：名称 → (中文名, 作用, 处理函数)。
// 这是工具的唯一事实源——dispatch 据此路由、list_tools 据此上报，二者永不漂移。
+// 想让某工具能被自主 agent 调用：把 agent 设 true，写清 params（模型可填）与 inject（服务端注入）。
+// 加新工具只改这一处——dispatcher 经 list_tools 动态发现，无需改调度代码。
func (g *Gateway) buildRegistry() map[string]toolDef {
return map[string]toolDef{
- "wiki_search": {"知识检索", "向量检索知识库（Milvus），返回最相关片段", g.wikiSearch},
- "kb_ingest": {"知识入库", "文本切块 → 向量化 → 写入 Milvus / Bleve", g.kbIngest},
- "kb_search": {"检索台查询", "结构化返回命中内容与相似度分数", g.kbSearch},
- "kb_graph": {"知识图谱", "取某库的实体关系三元组（Neo4j）", g.kbGraph},
- "report_render": {"报告渲染", "把结构化报告渲染为 Word（.docx）", g.reportRender},
- "report_store": {"报告存源", "暂存报告源数据，供导出时按需渲染", g.reportStore},
- "report_export": {"报告导出", "按需把已存报告导出为 Word / Markdown", g.reportExport},
- "external_api": {"外部接口", "受控调用第三方 HTTP API（带 SSRF 校验）", g.externalAPI},
- "memory_get": {"记忆召回", "取用户长期画像（已按打分排序）", g.memoryGet},
- "memory_upsert": {"记忆写入", "新增 / 更新一条用户偏好（带重要度）", g.memoryUpsert},
- "memory_delete": {"记忆删除", "软删一条偏好（对账判定过时 / 矛盾时）", g.memoryDelete},
- "memory_list": {"记忆列表", "列出用户全部偏好（供管理面板查看）", g.memoryList},
- "history_get": {"历史召回", "取会话最近多轮对话", g.historyGet},
- "history_append": {"历史追加", "往会话写入一条消息", g.historyAppend},
- "health": {"健康检查", "上报 Milvus / Neo4j / embedding 就绪情况",
- func(_ context.Context, _ *contract.ToolCall) *contract.ToolResult {
+ // —— 暴露给自主 agent 的工具（带参数 schema / 注入声明）——
+ "wiki_search": {
+ cn: "知识检索", desc: "检索知识库，返回与查询最相关的资料片段。需要外部知识/事实依据时调用。",
+ agent: true,
+ params: []paramSpec{{Name: "q", Type: "string", Desc: "检索查询语句", Required: true}},
+ inject: []string{"kb"}, handler: g.wikiSearch,
+ },
+ "memory_get": {
+ cn: "记忆召回", desc: "召回当前用户的长期画像与偏好（称呼/职业/回答偏好等）。需要个性化、了解“我是谁”时调用。",
+ agent: true, agentName: "recall_user_memory", inject: []string{"user_id"}, handler: g.memoryGet,
+ },
+ "memory_upsert": {
+ cn: "记忆写入", desc: "把关于用户的一条事实/偏好长期记住（如称呼、职业、回答偏好）。",
+ agent: true, agentName: "remember_user_fact",
+ params: []paramSpec{
+ {Name: "key", Type: "string", Desc: "记忆条目的键，如 称呼/职业/回答偏好", Required: true},
+ {Name: "value", Type: "string", Desc: "记忆条目的值", Required: true},
+ },
+ inject: []string{"user_id"}, handler: g.memoryUpsert,
+ },
+ "history_get": {
+ cn: "历史召回", desc: "取当前会话最近多轮对话，用于理解上下文。",
+ agent: true, inject: []string{"session_id"}, handler: g.historyGet,
+ },
+
+ // —— 仅内部/流水线/管理用，不暴露给自主 agent ——
+ "kb_ingest": {cn: "知识入库", desc: "文本切块 → 向量化 → 写入 Milvus / Bleve", handler: g.kbIngest},
+ "kb_search": {cn: "检索台查询", desc: "结构化返回命中内容与相似度分数", handler: g.kbSearch},
+ "kb_graph": {cn: "知识图谱", desc: "取某库的实体关系三元组（Neo4j）", handler: g.kbGraph},
+ "report_render": {cn: "报告渲染", desc: "把结构化报告渲染为 Word（.docx）", handler: g.reportRender},
+ "report_store": {cn: "报告存源", desc: "暂存报告源数据，供导出时按需渲染", handler: g.reportStore},
+ "report_export": {cn: "报告导出", desc: "按需把已存报告导出为 Word / Markdown", handler: g.reportExport},
+ "external_api": {cn: "外部接口", desc: "受控调用第三方 HTTP API（带 SSRF 校验）", handler: g.externalAPI},
+ "memory_delete": {cn: "记忆删除", desc: "软删一条偏好（对账判定过时 / 矛盾时）", handler: g.memoryDelete},
+ "memory_list": {cn: "记忆列表", desc: "列出用户全部偏好（供管理面板查看）", handler: g.memoryList},
+ "history_append": {cn: "历史追加", desc: "往会话写入一条消息", handler: g.historyAppend},
+ "health": {cn: "健康检查", desc: "上报 Milvus / Neo4j / embedding 就绪情况",
+ handler: func(_ context.Context, _ *contract.ToolCall) *contract.ToolResult {
data, _ := json.Marshal(g.rag.Status())
return &contract.ToolResult{OK: true, Content: string(data)}
}},
- "echo": {"回显", "原样返回入参（调试用）",
- func(_ context.Context, call *contract.ToolCall) *contract.ToolResult {
+ "echo": {cn: "回显", desc: "原样返回入参（调试用）",
+ handler: func(_ context.Context, call *contract.ToolCall) *contract.ToolResult {
return &contract.ToolResult{OK: true, Content: fmt.Sprint(call.Args["text"])}
}},
}
@@ -102,16 +140,24 @@ func (g *Gateway) dispatch(ctx context.Context, call *contract.ToolCall) *contra
return td.handler(ctx, call)
}
-// listTools 自省：上报本服务注册的工具清单（名称 + 中文名 + 作用），供管理端展示。
+// listTools 自省：上报本服务注册的工具清单（名称 + 中文名 + 作用 + agent 暴露元信息），
+// 供管理端展示 & dispatcher 动态构建自主 agent 工具集（加工具只改注册表，无需改调度代码）。
func (g *Gateway) listTools() *contract.ToolResult {
type info struct {
- Name string `json:"name"`
- CN string `json:"cn"`
- Desc string `json:"desc"`
+ Name string `json:"name"`
+ CN string `json:"cn"`
+ Desc string `json:"desc"`
+ Agent bool `json:"agent_exposed"` // 是否给自主 agent
+ AgentName string `json:"agent_name,omitempty"`// 模型可见名（空=name）
+ Params []paramSpec `json:"params,omitempty"` // 模型可填参数
+ Inject []string `json:"inject,omitempty"` // 服务端注入参数（不暴露给模型）
}
out := make([]info, 0, len(g.tools))
for name, td := range g.tools {
- out = append(out, info{Name: name, CN: td.cn, Desc: td.desc})
+ out = append(out, info{
+ Name: name, CN: td.cn, Desc: td.desc,
+ Agent: td.agent, AgentName: td.agentName, Params: td.params, Inject: td.inject,
+ })
}
sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name }) // map 无序 → 稳定输出
data, _ := json.Marshal(map[string]any{"service": "mcp-go", "tools": out})
diff --git a/sundynix-mcp-go/internal/rag/chunk.go b/sundynix-mcp-go/internal/rag/chunk.go
new file mode 100644
index 0000000..3a9615a
--- /dev/null
+++ b/sundynix-mcp-go/internal/rag/chunk.go
@@ -0,0 +1,137 @@
+package rag
+
+import (
+ "strings"
+)
+
+// 切块参数（针对中文 + embedding token 上限调过；后续可配置化）。
+const (
+ chunkTargetRunes = 500 // 目标大小：贪心打包到接近此值就在句界收口
+ chunkOverlapRunes = 80 // 块间重叠：保跨块上下文连续
+ chunkMinRunes = 100 // 最小块：低于则并入相邻，避免碎块稀释向量
+ chunkMaxRunes = 1000 // 原子硬上限：超大无标点段落兜底窗口切
+)
+
+// chunk 把文本切成检索友好的语义块：按段落/句界切成原子 → 贪心打包到目标大小（句末收口）
+// → 块间加重叠。全程按 rune 操作，杜绝中文 UTF-8 被字节切碎。
+func chunk(text string) []string {
+ atoms := splitToAtoms(text)
+ packed := packAtoms(atoms, chunkTargetRunes, chunkMinRunes)
+ return addOverlap(packed, chunkOverlapRunes)
+}
+
+// splitToAtoms 把文本切成"原子"（句子/行）：在换行与句末标点处断开，去空白；
+// 超大无标点原子按 rune 窗口兜底切到 ≤ 目标大小。原子是打包的最小不可分单元。
+func splitToAtoms(text string) []string {
+ runes := []rune(strings.ReplaceAll(text, "\r\n", "\n"))
+ var atoms []string
+ start := 0
+ flush := func(end int) {
+ if s := strings.TrimSpace(string(runes[start:end])); s != "" {
+ atoms = append(atoms, s)
+ }
+ start = end
+ }
+ for i := 0; i < len(runes); i++ {
+ r := runes[i]
+ switch {
+ case r == '\n' || isCJKEnd(r):
+ flush(i + 1)
+ case r == '.' || r == '!' || r == '?' || r == ';':
+ // ASCII 句末：仅当其后为空白/行尾才断（避开缩写、小数点）。
+ if i+1 >= len(runes) || runes[i+1] == ' ' || runes[i+1] == '\n' {
+ flush(i + 1)
+ }
+ }
+ }
+ flush(len(runes))
+
+ out := make([]string, 0, len(atoms))
+ for _, a := range atoms {
+ if runeLen(a) <= chunkMaxRunes {
+ out = append(out, a)
+ } else {
+ out = append(out, splitByRuneWindow(a, chunkTargetRunes)...)
+ }
+ }
+ return out
+}
+
+// packAtoms 贪心打包：把原子拼进当前块，直到再加会超 target 就收口（块在句界结束）。
+// 末尾过小的块（< min）并入前一块，避免碎块。
+func packAtoms(atoms []string, target, min int) []string {
+ var chunks []string
+ var cur strings.Builder
+ curLen := 0
+ closeCur := func() {
+ if curLen > 0 {
+ chunks = append(chunks, cur.String())
+ cur.Reset()
+ curLen = 0
+ }
+ }
+ for _, a := range atoms {
+ al := runeLen(a)
+ if curLen > 0 && curLen+al > target {
+ closeCur()
+ }
+ if curLen > 0 {
+ cur.WriteByte('\n')
+ curLen++
+ }
+ cur.WriteString(a)
+ curLen += al
+ }
+ if curLen > 0 {
+ if n := len(chunks); n > 0 && curLen < min {
+ chunks[n-1] = chunks[n-1] + "\n" + cur.String() // 尾块太小 → 并入上一块
+ } else {
+ chunks = append(chunks, cur.String())
+ }
+ }
+ return chunks
+}
+
+// addOverlap 给每块前缀上一块尾部的 overlap 个 rune，保跨块上下文连续（首块不加）。
+func addOverlap(chunks []string, overlap int) []string {
+ if overlap <= 0 || len(chunks) <= 1 {
+ return chunks
+ }
+ out := make([]string, len(chunks))
+ out[0] = chunks[0]
+ for i := 1; i < len(chunks); i++ {
+ prev := []rune(chunks[i-1])
+ tail := prev
+ if len(prev) > overlap {
+ tail = prev[len(prev)-overlap:]
+ }
+ out[i] = strings.TrimSpace(string(tail)) + "\n" + chunks[i]
+ }
+ return out
+}
+
+func isCJKEnd(r rune) bool {
+ switch r {
+ case '。', '！', '？', '；', '…':
+ return true
+ }
+ return false
+}
+
+// splitByRuneWindow 按 rune 窗口硬切（兜底：超大无标点原子），保证不切碎多字节字符。
+func splitByRuneWindow(s string, size int) []string {
+ r := []rune(s)
+ var out []string
+ for len(r) > size {
+ if t := strings.TrimSpace(string(r[:size])); t != "" {
+ out = append(out, t)
+ }
+ r = r[size:]
+ }
+ if t := strings.TrimSpace(string(r)); t != "" {
+ out = append(out, t)
+ }
+ return out
+}
+
+func runeLen(s string) int { return len([]rune(s)) }
diff --git a/sundynix-mcp-go/internal/rag/chunk_test.go b/sundynix-mcp-go/internal/rag/chunk_test.go
new file mode 100644
index 0000000..be247f1
--- /dev/null
+++ b/sundynix-mcp-go/internal/rag/chunk_test.go
@@ -0,0 +1,122 @@
+package rag
+
+import (
+ "strings"
+ "testing"
+ "unicode/utf8"
+)
+
+func TestChunkEmpty(t *testing.T) {
+ if got := chunk(" \n\n "); len(got) != 0 {
+ t.Fatalf("空白文本应切出 0 块，得 %d", len(got))
+ }
+}
+
+func TestChunkShort(t *testing.T) {
+ got := chunk("这是一句很短的话。")
+ if len(got) != 1 || !strings.Contains(got[0], "很短") {
+ t.Fatalf("短文本应为单块，得 %v", got)
+ }
+}
+
+// TestChunkRuneSafe 是核心：中文必须按 rune 切，绝不能切碎出乱码（旧版按字节切的 bug）。
+func TestChunkRuneSafe(t *testing.T) {
+ // 1500 个汉字、无标点 → 触发窗口兜底切；每块必须是合法 UTF-8。
+ text := strings.Repeat("中", 1500)
+ for _, c := range chunk(text) {
+ if !utf8.ValidString(c) {
+ t.Fatalf("切出非法 UTF-8（中文被字节切碎）：%q", c)
+ }
+ for _, r := range c {
+ if r != '中' && !strings.ContainsRune("\n ", r) {
+ t.Fatalf("出现意外字符 %q，疑似切碎", r)
+ }
+ }
+ }
+}
+
+func TestChunkSizeBounds(t *testing.T) {
+ // 多段中文，每段以句号结尾。
+ var sb strings.Builder
+ for i := 0; i < 60; i++ {
+ sb.WriteString("这是用于测试切块大小上界的一个中文句子片段。")
+ }
+ for i, c := range chunk(sb.String()) {
+ if n := runeLen(c); n > chunkTargetRunes+chunkOverlapRunes+1 {
+ t.Fatalf("第 %d 块 %d 字，超过 target+overlap=%d", i, n, chunkTargetRunes+chunkOverlapRunes+1)
+ }
+ }
+}
+
+func TestChunkSentenceBoundary(t *testing.T) {
+ var sb strings.Builder
+ for i := 0; i < 80; i++ {
+ sb.WriteString("第一句话在这里。第二句话也在这里。")
+ }
+ chunks := chunk(sb.String())
+ if len(chunks) < 2 {
+ t.Fatalf("长文本应切出多块，得 %d", len(chunks))
+ }
+ // 多数块应以句号收口（允许重叠导致的少量例外）。
+ endsWell := 0
+ for _, c := range chunks {
+ if strings.HasSuffix(strings.TrimSpace(c), "。") {
+ endsWell++
+ }
+ }
+ if endsWell < len(chunks)/2 {
+ t.Fatalf("多数块应在句末收口，仅 %d/%d", endsWell, len(chunks))
+ }
+}
+
+func TestChunkOverlap(t *testing.T) {
+ var sb strings.Builder
+ for i := 0; i < 60; i++ {
+ sb.WriteString("用于验证相邻块之间存在上下文重叠的中文句子。")
+ }
+ chunks := chunk(sb.String())
+ if len(chunks) < 2 {
+ t.Skip("未切出多块，跳过重叠校验")
+ }
+ // 第 2 块开头应包含第 1 块尾部的一小段（重叠）。
+ prevTail := []rune(chunks[0])
+ tail := string(prevTail[max0(len(prevTail)-chunkOverlapRunes):])
+ // 取尾部一小片做包含判断（去掉可能的换行）。
+ probe := strings.TrimSpace(tail)
+ if len(probe) > 10 {
+ probe = probe[len(probe)-10:]
+ }
+ if probe != "" && !strings.Contains(chunks[1], strings.TrimSpace(string([]rune(probe)))) {
+ // 重叠是按 rune 尾部，probe 是字节尾部，宽松校验：第二块前缀应与首块尾部有交集
+ if !strings.HasPrefix(strings.TrimSpace(chunks[1]), strings.TrimSpace(tail)) {
+ t.Logf("重叠片段:%q\n块2前缀:%q", tail, []rune(chunks[1])[:min0(40, runeLen(chunks[1]))])
+ t.Fatalf("相邻块未见重叠")
+ }
+ }
+}
+
+func TestChunkOversizedNoPunct(t *testing.T) {
+ text := strings.Repeat("x", 3000) // 无标点超大块
+ chunks := chunk(text)
+ if len(chunks) < 3 {
+ t.Fatalf("3000 字无标点应窗口切成多块，得 %d", len(chunks))
+ }
+ for _, c := range chunks {
+ if runeLen(c) > chunkTargetRunes+chunkOverlapRunes+1 {
+ t.Fatalf("窗口切块超界：%d", runeLen(c))
+ }
+ }
+}
+
+func max0(n int) int {
+ if n < 0 {
+ return 0
+ }
+ return n
+}
+func min0(a, b int) int {
+ if a < b {
+ return a
+ }
+ return b
+}
diff --git a/sundynix-mcp-go/internal/rag/rag.go b/sundynix-mcp-go/internal/rag/rag.go
index e99f34a..f535fde 100644
--- a/sundynix-mcp-go/internal/rag/rag.go
+++ b/sundynix-mcp-go/internal/rag/rag.go
@@ -6,7 +6,6 @@ import (
"context"
"errors"
"log"
- "strings"
"sync"
"github.com/sundynix/sundynix-shared/contract"
@@ -252,19 +251,4 @@ func (e *Engine) Close() {
e.graph.close(context.Background())
}
-// chunk 朴素切块：按行切，去空白；过长再按长度切。真实系统应做版面/语义切块。
-func chunk(text string) []string {
- var out []string
- for _, line := range strings.Split(text, "\n") {
- s := strings.TrimSpace(line)
- if s == "" {
- continue
- }
- for len(s) > 2000 {
- out = append(out, s[:2000])
- s = s[2000:]
- }
- out = append(out, s)
- }
- return out
-}
+// chunk 的实现已移到 chunk.go（递归 + 句界 + 重叠 + rune 安全的语义切块）。
diff --git a/sundynix-mcp-py/src/sundynix_mcp_py/mcp_gateway.py b/sundynix-mcp-py/src/sundynix_mcp_py/mcp_gateway.py
index 4b60186..e4be237 100644
--- a/sundynix-mcp-py/src/sundynix_mcp_py/mcp_gateway.py
+++ b/sundynix-mcp-py/src/sundynix_mcp_py/mcp_gateway.py
@@ -25,12 +25,18 @@ log = logging.getLogger("mcp_py")
SUBJECT_PY_ALL = "sundynix.tools.py.>"
QUEUE_PY = "mcp-py-workers"
-# 工具元信息：名称 → (中文名, 作用简述)。list_tools 据此上报给管理端展示。
+# 工具元信息（与 mcp-go 注册表同形）：list_tools 上报给管理端展示 + dispatcher 动态构建自主 agent 工具集。
+# agent=是否暴露给自主 agent；params=模型可填参数；inject=服务端注入参数（不暴露给模型）。
TOOL_META = {
- "echo": ("回显", "原样返回入参（调试用）"),
- "run_code": ("代码执行", "静态守卫 + Docker 隔离沙箱运行代码（标准档 256m/10s）"),
- "parse_document": ("文档解析", "文件 → 纯文本（MinerU / PaddleOCR）"),
- "secure_sandbox": ("安全沙箱", "更严资源档（128m/5s）的隔离执行，用于高风险代码"),
+ "echo": {"cn": "回显", "desc": "原样返回入参（调试用）"},
+ "run_code": {
+ "cn": "代码执行",
+ "desc": "在隔离沙箱里执行 Python 代码（静态守卫 + Docker，256m/10s）。用于计算、数据处理、逻辑验证。",
+ "agent": True,
+ "params": [{"name": "code", "type": "string", "desc": "要执行的 Python 代码", "required": True}],
+ },
+ "parse_document": {"cn": "文档解析", "desc": "文件 → 纯文本（MinerU / PaddleOCR）"},
+ "secure_sandbox": {"cn": "安全沙箱", "desc": "更严资源档（128m/5s）的隔离执行，用于高风险代码"},
}
@@ -102,10 +108,18 @@ class McpGateway:
return str(args.get("text", ""))
async def _list_tools(self, args: dict) -> str:
- """自省：上报业务工具清单（名称 + 中文名 + 作用），供管理端探活 + 展示。"""
+ """自省：上报业务工具清单（名称/中文名/作用 + agent 暴露元信息），供管理端展示 + dispatcher 动态发现。"""
tools = [
- {"name": n, "cn": cn, "desc": d}
- for n, (cn, d) in TOOL_META.items()
+ {
+ "name": n,
+ "cn": m["cn"],
+ "desc": m["desc"],
+ "agent_exposed": m.get("agent", False),
+ "agent_name": m.get("agent_name", ""),
+ "params": m.get("params", []),
+ "inject": m.get("inject", []),
+ }
+ for n, m in TOOL_META.items()
if n in self._tools # 仅上报真正注册的业务工具（list_tools 自身不计入）
]
return json.dumps({"service": "mcp-py", "tools": tools})
diff --git a/sundynix-shared/bus/bus.go b/sundynix-shared/bus/bus.go
index 63631cf..dca7377 100644
--- a/sundynix-shared/bus/bus.go
+++ b/sundynix-shared/bus/bus.go
@@ -237,6 +237,31 @@ func (b *Bus) Ping(ctx context.Context, subject string) ([]byte, error) {
return msg.Data, nil
}
+// ---- 任务生命周期状态回写（core NATS pub-sub）----
+
+// PublishTaskStatus 广播一次任务状态流转（dispatcher 调用）。
+func (b *Bus) PublishTaskStatus(ev *contract.TaskStatusEvent) error {
+ data, err := json.Marshal(ev)
+ if err != nil {
+ return err
+ }
+ return b.nc.Publish(contract.SubjectTaskStatus, data)
+}
+
+// SubscribeTaskStatus 订阅任务状态流转（网关调用，落 PG + 推 UI）。
+func (b *Bus) SubscribeTaskStatus(onEvent func(*contract.TaskStatusEvent)) (unsub func() error, err error) {
+ sub, err := b.nc.Subscribe(contract.SubjectTaskStatus, func(m *nats.Msg) {
+ var ev contract.TaskStatusEvent
+ if json.Unmarshal(m.Data, &ev) == nil {
+ onEvent(&ev)
+ }
+ })
+ if err != nil {
+ return nil, fmt.Errorf("subscribe task status: %w", err)
+ }
+ return sub.Unsubscribe, nil
+}
+
// ---- 配置控制面（core NATS request-reply + broadcast）----
// RequestConfig 向控制面（Gateway）请求某 kind 当前激活配置（chat/embedding）。
diff --git a/sundynix-shared/contract/task.go b/sundynix-shared/contract/task.go
index 73c7614..fa20099 100644
--- a/sundynix-shared/contract/task.go
+++ b/sundynix-shared/contract/task.go
@@ -29,6 +29,31 @@ const (
// request-reply 心跳主题让控制面（管理端「服务状态」）能判定它在不在线。
SubjectHealthDispatcher = "sundynix.health.dispatcher"
+ // 任务生命周期状态回写：dispatcher 开跑/跑完/出错经此主题广播，网关订阅落 PG 并推 UI。
+ // core NATS pub-sub（状态是幂等覆盖，丢一条由下一条纠正，无需持久化）。
+ // 注意：必须在 sundynix.tasks.> 之外，否则会被任务流捕获成"幽灵任务"自我放大。
+ SubjectTaskStatus = "sundynix.status.task"
+)
+
+// 任务生命周期状态机：submitted（网关建任务）→ running（dispatcher 开跑）
+// → done / failed / timeout（dispatcher 收尾）。
+const (
+ TaskSubmitted = "submitted"
+ TaskRunning = "running"
+ TaskDone = "done"
+ TaskFailed = "failed"
+ TaskTimeout = "timeout"
+)
+
+// TaskStatusEvent 是一次任务状态流转事件（经 SubjectTaskStatus 回流给网关）。
+type TaskStatusEvent struct {
+ TaskID string `json:"task_id"`
+ Status string `json:"status"` // running / done / failed / timeout
+ Detail string `json:"detail,omitempty"` // 失败原因等
+ TS int64 `json:"ts"` // unix 毫秒
+}
+
+const (
// MetaUserID 是 Task.Meta 中承载已登录用户标识的键（用于偏好记忆召回）。
MetaUserID = "user_id"
// MetaSessionID 是 Task.Meta 中承载会话标识的键（用于短期多轮历史）。