Cyrene_v2/internal/runtime/loop/flow_turnid_test.go

package loop

// 本文件钉住**权威回合标识覆盖到全部编排回合**（切片 CODE-TC-2）：`toolCalls[]` 的
// 覆盖率此前只到**真并行路** —— 图的 `TurnID` 取自归属桶里的路标识，而桶里的路标识
// 只有真并行路才有 ⇒ 普通单路回合（Web 单路聊天、队首续跑路）的图没有权威标识，
// 落盘后 `flow_graphs.lane_turn_id` 为空，消息面拿 `chat_messages.turn_id` 查不到图
// ⇒ `toolCalls` 整体缺席（契约允许，但不是需求本意）。
//
// 三条判据都用**真 `Loop.Run`**（或网关实际挂的那几条 ctx 通道）驱动，不靠手搓图：
//
//  1. 单路回合（只挂 `WithTurnID`）⇒ 帧 / 图 / 落盘行都带权威回合标识，且能按它查回
//     （`GraphsByTurnID` 正是 gateway 的消息读取面取图的那个入口）；
//  2. 没挂权威标识（后台思考 / 未接编排器）⇒ 如实缺席，**绝不**拿会话 id 之类的去臆造；
//  3. 真并行两路 ⇒ 归属桶一字未改：两路各一张图、各落一行、各按自己的回合标识查得回。

import (
	"context"
	"database/sql"
	"strings"
	"sync"
	"testing"

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

// frameCapture 收集发布闭包收到的每一帧图快照（发布器可能被多路并发驱动，故加锁）。
type frameCapture struct {
	mu     sync.Mutex
	frames []*trace.FlowGraph
}

func (c *frameCapture) add(g *trace.FlowGraph) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.frames = append(c.frames, g)
}

func (c *frameCapture) snapshot() []*trace.FlowGraph {
	c.mu.Lock()
	defer c.mu.Unlock()
	return append([]*trace.FlowGraph(nil), c.frames...)
}

// newFlowGraphStore 开一个临时库（内存）并把迁移跑满，返回**真落盘出口**与连接。
// 断言走真 store（`GraphsByTurnID` / `Turns`）与真 SQL（列里的字面值），与
// `internal/runtime/flowgraph` 的 newTestStore 同一手法 —— 本文件要验的是「库里真的
// 落下了那个值」，替身做不到。
func newFlowGraphStore(t *testing.T) (*flowgraph.SQLStore, *sql.DB) {
	t.Helper()
	ctx := context.Background()
	conn, err := db.Open(ctx, ":memory:", nil)
	if err != nil {
		t.Fatalf("打开内存库: %v", err)
	}
	if err := db.MigrateUp(ctx, conn, nil); err != nil {
		_ = conn.Close()
		t.Fatalf("内存库跑迁移: %v", err)
	}
	t.Cleanup(func() { _ = conn.Close() })
	st, err := flowgraph.NewStore(flowgraph.Config{DB: conn, Log: testLogger()})
	if err != nil {
		t.Fatalf("flowgraph.NewStore: %v", err)
	}
	return st, conn
}

// captureFrames 造一台接了**真落盘出口**的发布器，并把每一帧图快照收集下来。
// 发布闭包只认本片的主题（`ws.graph_update`），与生产同一个 subject。
func captureFrames(t *testing.T, store *flowgraph.SQLStore) (*FlowPublisher, *frameCapture) {
	t.Helper()
	cap := &frameCapture{}
	pub := NewFlowPublisher(func(subject string, payload any) {
		if subject != graphUpdateSubject {
			return
		}
		if g, ok := payload.(*trace.FlowGraph); ok {
			cap.add(g)
		}
	}, func(string) bool { return true }, WithRoundRecorder(store))
	return pub, cap
}

// toolCallAdapter 是「先请求一次工具、再给最终答复」的最小脚本适配器：它的存在只为
// 让图里真的长出一个工具节点（`toolCalls[]` 的数据源就是它）。
func toolCallAdapter(tool string) *mockAdapter {
	return &mockAdapter{chat: []mgResp{
		{resp: model.Response{Content: `{"tool_calls":[{"name":"` + tool + `","args":{}}]}`}},
		{resp: model.Response{Content: "答复"}},
	}}
}

// toolCallLoop 装一台「一次工具调用 + 最终答复」的循环，接上给定发布器。
func toolCallLoop(t *testing.T, pub *FlowPublisher) *Loop {
	t.Helper()
	return New(Options{
		Agent:     newTestAgent(t, toolCallAdapter("probe"), dispatcherWith(t, &mockTool{name: "probe", result: "工具结果"})),
		MaxRounds: 8,
		Flow:      pub,
		Tracer:    trace.New(nil),
	})
}

// TestRun_SingleLegRoundCarriesAuthoritativeTurnID_正常 是本片的核心：一次**普通单路
// 回合**（只挂网关那条 `WithTurnID`，不挂 `WithParallelLeg`）跑完之后，权威回合标识
// 一路到达「帧 → 图 → 落盘行」，且能被 `GraphsByTurnID` 按**消息那一个 id** 查回。
func TestRun_SingleLegRoundCarriesAuthoritativeTurnID_正常(t *testing.T) {
	t.Parallel()
	const turnID = "turn-single-1"
	ctx := context.Background()
	store, conn := newFlowGraphStore(t)
	pub, frames := captureFrames(t, store)

	// 网关在驱动每一路之前就这么挂（ws_hub_turn.go / beginLegLocked）：单路也有。
	res, err := toolCallLoop(t, pub).Run(WithTurnID(ctx, turnID), "web",
		[]model.Message{model.TextMessage(model.RoleUser, "帮我跑一次工具")})
	if err != nil {
		t.Fatalf("Run: %v", err)
	}
	if res.Text != "答复" {
		t.Fatalf("答复 = %q, want 答复", res.Text)
	}

	// ① 帧：单路的每一帧都带权威 `turnId`（此前缺席），会话归属与轮次号照旧。
	all := frames.snapshot()
	if len(all) == 0 {
		t.Fatal("一帧都没发布")
	}
	for i, g := range all {
		if g.TurnID != turnID {
			t.Fatalf("第 %d 帧 turnId = %q, want %q（单路也必须带权威回合标识）", i, g.TurnID, turnID)
		}
		if g.SessionID != "web" {
			t.Fatalf("第 %d 帧 sessionId = %q, want web（路不能改会话归属）", i, g.SessionID)
		}
		if g.Round == 0 {
			t.Fatalf("第 %d 帧缺轮次号", i)
		}
	}

	// ② 落盘行：`lane_turn_id` 就是那个权威回合标识（与 `chat_messages.turn_id` 同值，
	// 因此消息面按等值就能查到本轮的图 —— gateway 的 `toolCalls[]` 正是这么取的）。
	wantGraphID := "web-1#" + turnID
	var lane string
	if err := conn.QueryRowContext(ctx,
		`SELECT lane_turn_id FROM flow_graphs WHERE session_id = ?`, "web").Scan(&lane); err != nil {
		t.Fatalf("读落盘行的 lane_turn_id: %v", err)
	}
	if lane != turnID {
		t.Fatalf("落盘 lane_turn_id = %q, want %q（单路行不再为空）", lane, turnID)
	}
	graphs, err := store.GraphsByTurnID(ctx, "web", []string{turnID})
	if err != nil {
		t.Fatalf("GraphsByTurnID: %v", err)
	}
	got := graphs[turnID]
	if got == nil {
		t.Fatalf("按 %q 查不到本轮图（命中 %d 条）—— 消息面因此给不出 toolCalls", turnID, len(graphs))
	}
	if got.ID != wantGraphID {
		t.Fatalf("图 id = %q, want %q（会话级序号 + 权威回合标识）", got.ID, wantGraphID)
	}
	if !hasNode(got, "probe") {
		t.Fatalf("查回来的图里没有工具节点（toolCalls 的数据源）：%s", describeNodes(got))
	}
	// 每轮一行（不是每 span 一行），且 `Turns` 的键就是这张图的 id（面板翻历史轮用它）。
	turns, err := store.Turns(ctx, "web", 0)
	if err != nil {
		t.Fatalf("Turns: %v", err)
	}
	if len(turns) != 1 || turns[0].TurnID != wantGraphID {
		t.Fatalf("Turns = %+v, want 恰好一轮 %s", turns, wantGraphID)
	}
}

// TestRun_NoAuthorityStaysAbsent_边界 钉住**没有权威标识就不许臆造**：未挂 `WithTurnID`
// 的运行（后台思考 / 未接编排器的调用方）图上 `turnId` 缺席、落盘 `lane_turn_id` 为空串
// （既不是 NULL，也不是会话 id、时间戳之类的替身）⇒ 读侧按任何值都查不到这一轮。
func TestRun_NoAuthorityStaysAbsent_边界(t *testing.T) {
	t.Parallel()
	const session = "sess-backthink"
	ctx := context.Background()
	store, conn := newFlowGraphStore(t)
	pub, frames := captureFrames(t, store)

	if _, err := toolCallLoop(t, pub).Run(ctx, session,
		[]model.Message{model.TextMessage(model.RoleUser, "跑一次工具")}); err != nil {
		t.Fatalf("Run: %v", err)
	}

	for i, g := range frames.snapshot() {
		if g.TurnID != "" {
			t.Fatalf("第 %d 帧凭空有了 turnId=%q（没有权威标识时必须缺席）", i, g.TurnID)
		}
	}
	var (
		lane  string
		isNul int
	)
	if err := conn.QueryRowContext(ctx,
		`SELECT lane_turn_id, lane_turn_id IS NULL FROM flow_graphs WHERE session_id = ?`,
		session).Scan(&lane, &isNul); err != nil {
		t.Fatalf("读落盘行的 lane_turn_id: %v", err)
	}
	if lane != "" || isNul != 0 {
		t.Fatalf("lane_turn_id = %q (is null = %d), want 空串且非 NULL", lane, isNul)
	}
	// 读侧的两个候选值都查不到：会话 id 与任意回合标识都不是「这一轮的图」，诚实地缺席。
	found, err := store.GraphsByTurnID(ctx, session, []string{session, "turn-any"})
	if err != nil {
		t.Fatalf("GraphsByTurnID: %v", err)
	}
	if len(found) != 0 {
		t.Fatalf("没有权威标识的轮次竟被查到 %d 条：%v", len(found), found)
	}
}

// TestFlowPublisher_ParallelLanesKeepSeparateRows_正常 钉住**归属桶一字未改**：
// 同轮两路（网关挂的那两条 ctx 通道：`WithTurnID` + `WithParallelLeg`）仍各起各的图、
// 各落一行，各按自己的回合标识查得回 —— 换成权威标识当落盘/查询键，绝不能让两路
// 互相顶掉，也绝不能让两路混进同一张图。
func TestFlowPublisher_ParallelLanesKeepSeparateRows_正常(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store, conn := newFlowGraphStore(t)
	pub, _ := captureFrames(t, store)

	// 两路都在飞（同一轮）：A、B 先后起轮，再各自收轮。
	for _, lane := range []string{"turn-a", "turn-b"} {
		legCtx := WithParallelLeg(WithTurnID(ctx, lane), lane)
		pub.StartRoundWithTurnID(legCtx, FlowOwnerKey("web", lane), "标记"+lane)
	}
	for _, lane := range []string{"turn-a", "turn-b"} {
		pub.EndRound(FlowOwnerKey("web", lane), "答复"+lane)
	}
	// 建图序号是**会话级**递增的（A 先起 ⇒ 1），因此两路的 id 各带自己的后缀。
	wantID := map[string]string{"turn-a": "web-1#turn-a", "turn-b": "web-2#turn-b"}

	var rows int
	if err := conn.QueryRowContext(ctx,
		`SELECT COUNT(*) FROM flow_graphs WHERE session_id = ?`, "web").Scan(&rows); err != nil {
		t.Fatalf("COUNT 行数: %v", err)
	}
	if rows != 2 {
		t.Fatalf("flow_graphs 行数 = %d, want 2（两路各一行，谁都不许顶掉谁）", rows)
	}
	found, err := store.GraphsByTurnID(ctx, "web", []string{"turn-a", "turn-b"})
	if err != nil {
		t.Fatalf("GraphsByTurnID: %v", err)
	}
	for _, lane := range []string{"turn-a", "turn-b"} {
		g := found[lane]
		if g == nil {
			t.Fatalf("%s 那一路按自己的回合标识查不到（被另一路顶掉了？命中 %d 条）", lane, len(found))
		}
		if g.ID != wantID[lane] {
			t.Fatalf("%s 的图 id = %q, want %q（各带自己那一路的后缀）", lane, g.ID, wantID[lane])
		}
		if !hasInputMark(g, "标记"+lane) {
			t.Fatalf("%s 的图里没有自己那一路的输入（两路混进同一张图了）：%s", lane, describeNodes(g))
		}
	}
	// 最近一轮的两路都在内存里，页面按会话取图仍是两条泳道。
	if lanes := pub.LatestLanes("web"); len(lanes) != 2 {
		t.Fatalf("最近一轮的路数 = %d, want 2", len(lanes))
	}
}

// hasInputMark 报告图里有没有标签含 mark 的 input 节点（两路各自被自己的用户文本播种）。
func hasInputMark(g *trace.FlowGraph, mark string) bool {
	for _, n := range g.Nodes {
		if n.Kind == "input" && strings.Contains(n.Label, mark) {
			return true
		}
	}
	return false
}