Cyrene_v2/internal/runtime/loop/flow_parallel_test.go

package loop

import (
	"context"
	"strings"
	"sync"
	"testing"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
)

// 本文件是**真并行两路互不覆盖**的回归（design/15 §2.5.10-1 登记的那条真 bug）：
// 同一会话的两条并行路（`chat.parallel`）各自开一轮，`graphs[sessionID]` 只有一张 ⇒
// 后起的那一路把前一路的图整张顶掉、两路的节点互相长在对方的图上、落盘两路撞同一个
// `flow_graphs.turn_id`。
//
// 用例刻意只用**引入本切片之前就存在**的 API 驱动真实路径（真 `Loop.Run` + 网关实际
// 挂的那两条 ctx 通道 `WithTurnID` / `WithParallelLeg`），因此它在修复前能编译、能跑、
// 会红 —— 「改前红」的证据就是它（见 FV2-report §4）。判据：
//
//   - 两条路各落一条（`RecordRound` 两次）且**图 id 不同**（同一 turn_id 会被
//     `(session_id, turn_id)` 唯一键 upsert 成一行）；
//   - 以 A 的输入播种的那张图里只有 A 的工具节点，反之亦然；
//   - 任何一帧 `graph_update` 都不同时含两条路的工具节点。

// barrierTool 是执行期停在闸门上的工具：它报告「已到执行点」后阻塞，使测试能确定地
// 构造「A 的工具在飞时 B 起步」的交错时序，而不是靠 sleep 赌调度。
type barrierTool struct {
	name    string
	arrived chan struct{}
	gate    chan struct{}
	once    sync.Once
}

func newBarrierTool(name string) *barrierTool {
	return &barrierTool{name: name, arrived: make(chan struct{}, 1), gate: make(chan struct{})}
}

func (b *barrierTool) Name() string        { return b.name }
func (b *barrierTool) Description() string { return "barrier tool " + b.name }

func (b *barrierTool) Parameters() (map[string]any, error) {
	return map[string]any{"type": "object", "properties": map[string]any{}}, nil
}

func (b *barrierTool) Execute(ctx context.Context, _ any) (any, error) {
	b.once.Do(func() { b.arrived <- struct{}{} })
	select {
	case <-b.gate:
		return b.name + " 完成", nil
	case <-ctx.Done():
		return nil, ctx.Err()
	}
}

// laneAdapter 按「这一路的窗口里是哪个标记文本」分流，先请求该路的工具，工具结果回灌
// 之后给出该路的最终答复。两条路共用一个实例 —— 与生产同形（同一个 runner 跑两路）。
type laneAdapter struct {
	tool   map[string]string // 标记文本 → 该路请求的工具名
	answer map[string]string // 标记文本 → 该路的最终答复
}

func (a *laneAdapter) Name() string { return "lane" }

func (a *laneAdapter) Chat(_ context.Context, req model.Request, _ model.Options) (model.Response, error) {
	mark := a.markOf(req)
	if mark == "" {
		return model.Response{Content: "无法归属的请求"}, nil
	}
	if lastMessageRole(req.Messages) == model.RoleTool {
		return model.Response{Content: a.answer[mark]}, nil
	}
	return model.Response{Content: `{"tool_calls":[{"name":"` + a.tool[mark] + `","args":{}}]}`}, nil
}

// markOf 找出请求里出现的那一路的标记文本；两路的窗口互不共享（I4），因此最多一命中。
func (a *laneAdapter) markOf(req model.Request) string {
	for _, m := range req.Messages {
		if m.Role != model.RoleUser {
			continue
		}
		for mark := range a.tool {
			if strings.Contains(messageText(m), mark) {
				return mark
			}
		}
	}
	return ""
}

// ChatStream 是流式那一半：本用例的循环走非流式路径（未挂 OnStream），因此它只把
// 请求委托给 Chat 并原样交出聚合结果 —— 刻意**不**再造第二套脚本，也刻意不发任何
// 增量事件，避免「哪个入口真的跑过」变成模糊事实。
func (a *laneAdapter) ChatStream(ctx context.Context, req model.Request, _ func(model.StreamEvent)) (model.Response, error) {
	return a.Chat(ctx, req, model.Options{})
}

// lastMessageRole 返回末条消息的角色（空窗口时为空串）。
func lastMessageRole(msgs []model.Message) model.Role {
	if len(msgs) == 0 {
		return ""
	}
	return msgs[len(msgs)-1].Role
}

// parallelLaneFixture 是两路交错跑完之后的全部观测面：落盘的每一路、发布的每一帧、
// 以及两路各自的最终答复（用来确认哪张图属于哪一路）。
type parallelLaneFixture struct {
	pub      *FlowPublisher
	recorder *fakeRecorder
	mu       sync.Mutex
	frames   []*trace.FlowGraph
}

// run 让两条平行路交错跑完一轮：A 起轮 → A 的工具在飞 → B 起轮 → B 的工具在飞 →
// 依次放行。返回时两路的 Run 都已返回。
func (f *parallelLaneFixture) run(t *testing.T) {
	t.Helper()
	toolA, toolB := newBarrierTool("probe_a"), newBarrierTool("probe_b")
	ad := &laneAdapter{
		tool:   map[string]string{"标记A": "probe_a", "标记B": "probe_b"},
		answer: map[string]string{"标记A": "答复A", "标记B": "答复B"},
	}
	l := New(Options{
		Agent:     newTestAgent(t, ad, dispatcherWith(t, toolA, toolB)),
		MaxRounds: 8,
		Flow:      f.pub,
		Tracer:    trace.New(nil),
	})

	ctxA := WithParallelLeg(WithTurnID(context.Background(), "turn-a"), "turn-a")
	ctxB := WithParallelLeg(WithTurnID(context.Background(), "turn-b"), "turn-b")
	var wg sync.WaitGroup
	var resA, resB *Result
	var errA, errB error
	wg.Add(2)
	go func() {
		defer wg.Done()
		resA, errA = l.Run(ctxA, "web", []model.Message{model.TextMessage(model.RoleUser, "标记A：请跑一次工具")})
	}()
	// A 的工具确实在飞之后才起 B —— 这才叫「同一会话两路并行」，也正是缺陷现场。
	awaitBarrier(t, toolA.arrived, "A 的工具执行点")
	go func() {
		defer wg.Done()
		resB, errB = l.Run(ctxB, "web", []model.Message{model.TextMessage(model.RoleUser, "标记B：请跑一次工具")})
	}()
	awaitBarrier(t, toolB.arrived, "B 的工具执行点")
	close(toolA.gate)
	close(toolB.gate)
	wg.Wait()

	if errA != nil || errB != nil {
		t.Fatalf("两路 Run 报错：A=%v B=%v", errA, errB)
	}
	if resA.Text != "答复A" || resB.Text != "答复B" {
		t.Fatalf("两路答复 = %q / %q，want 答复A / 答复B（窗口串了）", resA.Text, resB.Text)
	}
}

// awaitBarrier 等一个闸门信号，超时即让测试失败（避免缺陷把它挂死成超时）。
func awaitBarrier(t *testing.T, ch <-chan struct{}, what string) {
	t.Helper()
	select {
	case <-ch:
	case <-time.After(20 * time.Second):
		t.Fatalf("等待 %s 超时", what)
	}
}

// newParallelLaneFixture 装一台接了假落盘出口的发布器（真 bus 不在本用例范围内）。
func newParallelLaneFixture(t *testing.T) *parallelLaneFixture {
	t.Helper()
	f := &parallelLaneFixture{recorder: &fakeRecorder{}}
	f.pub = NewFlowPublisher(func(subject string, payload any) {
		if subject != graphUpdateSubject {
			return
		}
		g, ok := payload.(*trace.FlowGraph)
		if !ok {
			return
		}
		f.mu.Lock()
		f.frames = append(f.frames, g)
		f.mu.Unlock()
	}, func(string) bool { return true }, WithRoundRecorder(f.recorder))
	return f
}

// recorded 返回落盘出口收到的每一路图（副本，调用方无需持锁）。
func (f *parallelLaneFixture) recorded() []*trace.FlowGraph {
	return recordedRounds(f.recorder)
}

// recordedRounds 返回假落盘出口收到的每一路图（副本，调用方无需持锁）。
func recordedRounds(rec *fakeRecorder) []*trace.FlowGraph {
	rec.mu.Lock()
	defer rec.mu.Unlock()
	return append([]*trace.FlowGraph(nil), rec.graphs...)
}

// publishedFrames 返回发布过的每一帧（副本）。
func (f *parallelLaneFixture) publishedFrames() []*trace.FlowGraph {
	f.mu.Lock()
	defer f.mu.Unlock()
	return append([]*trace.FlowGraph(nil), f.frames...)
}

// TestFlowPublisher_ParallelLegsDoNotOverwriteEachOther 是本切片的核心用例：
// 两路并行 ⇒ 两桶互不覆盖、各自落一条、任何一帧都不含对方节点。
func TestFlowPublisher_ParallelLegsDoNotOverwriteEachOther(t *testing.T) {
	f := newParallelLaneFixture(t)
	f.run(t)

	recorded := f.recorded()
	if len(recorded) != 2 {
		t.Fatalf("落盘 %d 条，want 2（每路一条）：%s", len(recorded), describeGraphs(recorded))
	}
	if recorded[0].ID == recorded[1].ID {
		t.Fatalf("两路撞同一个 turn_id %q ⇒ flow_graphs 的 (session_id, turn_id) 会把其中一路整行顶掉：%s",
			recorded[0].ID, describeGraphs(recorded))
	}
	graphA := graphWithInput(recorded, "标记A")
	graphB := graphWithInput(recorded, "标记B")
	if graphA == nil || graphB == nil {
		t.Fatalf("两路没有各自的图（A=%v B=%v）：%s", graphA != nil, graphB != nil, describeGraphs(recorded))
	}
	assertLaneIsolated(t, "A", graphA, "probe_a", "probe_b")
	assertLaneIsolated(t, "B", graphB, "probe_b", "probe_a")

	for i, frame := range f.publishedFrames() {
		if hasNode(frame, "probe_a") && hasNode(frame, "probe_b") {
			t.Fatalf("第 %d 帧同时含两条路的工具节点（B 的节点长在 A 的图上）：%s", i, describeNodes(frame))
		}
	}
}

// assertLaneIsolated 断言某一路的图里有自己的工具节点、且**没有**对方的工具节点。
func assertLaneIsolated(t *testing.T, lane string, g *trace.FlowGraph, own, other string) {
	t.Helper()
	if !hasNode(g, own) {
		t.Fatalf("%s 路的图里没有自己的工具节点 %q：%s", lane, own, describeNodes(g))
	}
	if hasNode(g, other) {
		t.Fatalf("%s 路的图里出现了另一路的工具节点 %q：%s", lane, other, describeNodes(g))
	}
	if g.SessionID != "web" {
		t.Fatalf("%s 路图的 sessionId = %q，want web（路不能改会话归属）", lane, g.SessionID)
	}
}

// graphWithInput 找出输入节点标签含 mark 的那张图（两路各自被用户文本播种）。
func graphWithInput(graphs []*trace.FlowGraph, mark string) *trace.FlowGraph {
	for _, g := range graphs {
		for _, n := range g.Nodes {
			if n.Kind == "input" && strings.Contains(n.Label, mark) {
				return g
			}
		}
	}
	return nil
}

// hasNode 报告图里有没有标签等于（或前缀为）name 的工具节点。
func hasNode(g *trace.FlowGraph, name string) bool {
	for _, n := range g.Nodes {
		if n.Kind == "tool" && strings.Contains(n.Label, name) {
			return true
		}
	}
	return false
}

// describeGraphs / describeNodes 只打印**结构与标签**，供失败信息定位（测试里没有
// 敏感内容，但仍只打标识）。
func describeGraphs(graphs []*trace.FlowGraph) string {
	var b strings.Builder
	for _, g := range graphs {
		b.WriteString("\n  id=" + g.ID + " session=" + g.SessionID)
		b.WriteString(" nodes=" + describeNodes(g))
	}
	return b.String()
}

func describeNodes(g *trace.FlowGraph) string {
	parts := make([]string, 0, len(g.Nodes))
	for _, n := range g.Nodes {
		parts = append(parts, n.Kind+":"+n.Label)
	}
	return "[" + strings.Join(parts, " ") + "]"
}