sundynix-agentix/sundynix-dispatcher/internal/eino/compose_graph_test.go

package eino

import (
	"context"
	"encoding/json"
	"strings"
	"sync"
	"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"
	"github.com/sundynix/sundynix-shared/contract"
)

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

// TestComposeConversation 验证对话主流程走 compose 路径：模型输出流式回流到 sink 并累计成稿。
func TestComposeConversation(t *testing.T) {
	fs := &fakeSink{}
	o := &Orchestrator{
		pool:    &fakeLLM{ready: true, cm: &stubModel{reply: "我是 compose 路径的回答"}},
		breaker: harness.NewCircuitBreaker(),
		sink:    fs,
	}
	b := &board{query: "你好"}
	o.runComposeConversation(context.Background(), "task_compose", b, "", &execTracer{}, "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())
	}
}

// TestComposeConversationDegradesToRunAgent 验证无 Eino ChatModel 时降级回 runAgent（流式回流不丢）。
func TestComposeConversationDegradesToRunAgent(t *testing.T) {
	fs := &fakeSink{}
	o := &Orchestrator{
		// cm 为 nil → ChatModel() 返回 nil → runComposeConversation 降级 runAgent（走 stream 桩）。
		pool:    &fakeLLM{ready: true, stream: func([]llm.ChatMessage) string { return "降级路径回答" }},
		breaker: harness.NewCircuitBreaker(),
		sink:    fs,
	}
	b := &board{query: "你好"}
	o.runComposeConversation(context.Background(), "task_degrade", b, "", &execTracer{}, "agent", "模型流式推理")
	if !strings.Contains(b.answer, "降级路径") {
		t.Fatalf("无 ChatModel 应降级 runAgent 出稿: %q", b.answer)
	}
}

// 图里可以并排好几个 agent（研究/撰写/审查）。之前 runAgent 把轨迹标签写死成
// "模型流式推理"，用户在编排里起的节点名整个丢了 —— 轨迹里三行同名，团队视图里
// 工位只能退回节点 ID（r/w/rev），谁是谁全靠猜。这里钉住：标签必须来自节点 label。
type execCapture struct {
	mu  sync.Mutex
	buf [][]byte
}

func (c *execCapture) PublishExec(_ string, data []byte) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.buf = append(c.buf, append([]byte(nil), data...))
	return nil
}
func (c *execCapture) CompleteExec(string) error { return nil }
func (c *execCapture) labels() []string {
	c.mu.Lock()
	defer c.mu.Unlock()
	var out []string
	for _, b := range c.buf {
		var e contract.ExecEvent
		if json.Unmarshal(b, &e) == nil {
			out = append(out, e.Label)
		}
	}
	return out
}

func TestRunAgentUsesNodeLabel(t *testing.T) {
	cap := &execCapture{}
	o := &Orchestrator{
		pool:    &fakeLLM{ready: true, cm: &stubModel{reply: "要点若干"}},
		breaker: harness.NewCircuitBreaker(),
		sink:    &fakeSink{},
		exec:    cap,
	}
	o.runAgent(context.Background(), "task_label", &board{query: "选型"}, "", o.tracer("task_label"), "agent:r", "研究 Agent")

	got := strings.Join(cap.labels(), "|")
	if !strings.Contains(got, "研究 Agent") {
		t.Fatalf("轨迹标签应为节点名「研究 Agent」，实际: %q", got)
	}
	if strings.Contains(got, "模型流式推理") {
		t.Fatalf("不该再出现写死的兜底标签: %q", got)
	}
}

func TestRunAgentLabelFallback(t *testing.T) {
	cap := &execCapture{}
	o := &Orchestrator{
		pool:    &fakeLLM{ready: true, cm: &stubModel{reply: "x"}},
		breaker: harness.NewCircuitBreaker(),
		sink:    &fakeSink{},
		exec:    cap,
	}
	o.runAgent(context.Background(), "task_fb", &board{query: "q"}, "", o.tracer("task_fb"), "agent:x", "")

	if got := strings.Join(cap.labels(), "|"); !strings.Contains(got, "模型流式推理") {
		t.Fatalf("节点没起名时应回兜底标签: %q", got)
	}
}