EINO_ADOPTION.md

@@ -9,7 +9,7 @@
 
 - [x] **Phase A · 地基**：`llm.Pool` 换 Eino ChatModel 组件（commit d84b1ec，验收通过）
 - [x] **Phase B · 质变**：MCP 工具→`InvokableTool` + ReAct agent（模型自主调工具，验收 7/7 命中）
-- [ ] **Phase C · 编排归一**：`Flow→compose.Graph` + callbacks 桥接
+- [~] **Phase C · 编排归一**：✅ 对话主流程跑 `compose.Graph`（EINO_COMPOSE 灰度开关，默认关）+ callbacks→ExecEvent 归一 / ⬜ branch·map·render 等节点逐步迁移（并存+等价回归）
 - [~] **Phase D · 状态化执行**：✅ 任务生命周期 FSM（已完成）/ ⬜ HITL 中断恢复 / ⬜ 多智能体（按场景）
 
 ---
@@ -117,7 +117,17 @@ github.com/cloudwego/eino-ext/...             # ⚠️ 官方组件实现（open
 
 ---
 
-## Phase C · 编排归一：迁到 compose.Graph 〔P2〕
+## Phase C · 编排归一：迁到 compose.Graph 〔P2〕🟡 进行中（核心已落地，并存灰度）
+
+> 落地（并存+等价回归策略）：对话主流程已可跑在 `compose.Graph` 上——
+> - `compose_graph.go`：`runConversation` 按 `EINO_COMPOSE` 开关分流；`runComposeConversation` 建图 `START→ChatModel→END`、`Compile`→`Stream`，token 回流；模型未就绪/编译失败降级回 `runAgent`。**默认关，graph.go 仍是默认且权威。**
+> - `compose_callbacks.go`：`composeTracer` 用 `utils/callbacks` 把 ChatModel/Tool 的 start/end/error 桥到 ExecEvent（可观测归一）。
+> - 测试：compose 图编译+运行、compose 对话流式回流、开关关→走 runAgent，三个单测；live 实测 compose 路径出 54 字答复 + eval 1.00，默认路径 eval 1.00。
+> - **顺带修真 bug**：`SubjectTaskStatus` 原为 `sundynix.tasks.status`，落在任务流捕获通配 `sundynix.tasks.>` 内 → 状态事件被当成"幽灵任务"自我放大（实测污染 2300+ 条）。已挪到 `sundynix.status.task` + dispatcher 加空任务护栏。
+>
+> 剩余：branch / map / render / retriever / prompt 等节点逐步迁 compose（同并存+等价回归），对齐后 graph.go 退役。
+
+
 
 **目标**：自研解释器退役，DSL 图编译为 `compose.Graph`，吃到流式 reduce/branch、类型化边、自动并发、callbacks。
 

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))
+}

sundynix-dispatcher/internal/eino/compose_graph.go

@@ -0,0 +1,93 @@
+package eino
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"os"
+	"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...),
+	}
+	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
+	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))
+		b.answer += safe
+		chunks++
+	}
+	tr.info(node, "system", "compose 图", fmt.Sprintf("%d 段输出 / %d 字（Eino compose 运行时）", chunks, len([]rune(b.answer))))
+}

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)
+	}
+}

sundynix-dispatcher/internal/eino/graph.go

@@ -50,7 +50,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 +123,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 +147,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
 }

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

sundynix-dispatcher/internal/eino/orchestrator.go

@@ -43,6 +43,8 @@ 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
 }
 
 // 工具调用超时；超时即降级（不带工具上下文继续推理）。
@@ -96,6 +98,11 @@ func (o *Orchestrator) finishStatus(taskID string, err error) {
 
 // Handle 消费一个任务：按 DSL 编译 Eino 图并执行，把 Token 流回流到 sundynix.streams.<id>。
 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()
 

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 {

sundynix-shared/contract/task.go

@@ -31,7 +31,8 @@ const (
 
 	// 任务生命周期状态回写：dispatcher 开跑/跑完/出错经此主题广播，网关订阅落 PG 并推 UI。
 	// core NATS pub-sub（状态是幂等覆盖，丢一条由下一条纠正，无需持久化）。
-	SubjectTaskStatus = "sundynix.tasks.status"
+	// 注意：必须在 sundynix.tasks.> 之外，否则会被任务流捕获成"幽灵任务"自我放大。
+	SubjectTaskStatus = "sundynix.status.task"
 )
 
 // 任务生命周期状态机：submitted（网关建任务）→ running（dispatcher 开跑）