Cyrene_v2/internal/runtime/flowgraph/flowgraph_lane_test.go

package flowgraph

// 本文件钉住**真并行两路的落盘不撞 key**（design/15 §2.5.10-1 + design/19 §4 P3；
// 切片 FV2）。症状：两路在同一会话各收一轮，若两路的 `turn_id` 相同，
// `(session_id, turn_id)` 唯一键会让两路 upsert 成**一行** —— 后收尾的那一路把前一路
// 整张图顶掉，翻历史轮时那一路的图彻底消失。
//
// 断言的是**库里的事实**（几行、每行 id 是什么、读回来的图属于哪一路），不按日志断言，
// 与 flowgraph_test.go 同一口径。夹具用 `trace.RunGraph` 走真建图路径（两路各自
// SetTurnID + SetSeq），因此库里的 turn_id 与线上图 id 逐字同源。

import (
	"context"
	"database/sql"
	"testing"
	"time"

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

// laneGraph 用真建图器造一条路的收尾图：seq 是会话级建图序号（两路必不同），
// lane 是这一路的权威 turnId。
func laneGraph(session string, seq int, lane, input, output string) *trace.FlowGraph {
	g := trace.NewRunGraph(session)
	g.SetTurnID(lane)
	g.SetSeq(seq)
	g.Start(seq, testNow().Add(-time.Minute))
	g.SeedInput(input)
	g.Upsert(trace.NodeSpec{ID: "span-" + lane, Label: lane, Kind: "tool", Status: trace.FlowStatusDone, ParentID: "input"})
	g.SeedOutput(output)
	return g.Snapshot()
}

// TestStore_ParallelLanesPersistSeparateRows_正常 是本文件的核心断言：同一会话同一轮
// 的两路各占**一行**，按各自的路标识查得回来，`Turns` 列出两条，任一路都没有被
// 另一路顶掉。
func TestStore_ParallelLanesPersistSeparateRows_正常(t *testing.T) {
	t.Parallel()
	st, conn := newTestStore(t, nil)
	ctx := context.Background()

	laneA := laneGraph("web", 1, "turn-a", "标记A", "答复A")
	laneB := laneGraph("web", 2, "turn-b", "标记B", "答复B")
	if laneA.ID == laneB.ID {
		t.Fatalf("夹具自身就撞了 id：%q", laneA.ID)
	}
	for _, g := range []*trace.FlowGraph{laneA, laneB} {
		if err := st.RecordRound(g); err != nil {
			t.Fatalf("落盘 %s: %v", g.ID, err)
		}
	}

	if got := countRows(t, conn, "web"); got != 2 {
		t.Fatalf("flow_graphs 里该会话的行数 = %d，want 2（两路各一行）", got)
	}
	for _, want := range []struct {
		graph  *trace.FlowGraph
		input  string
		output string
	}{{laneA, "标记A", "答复A"}, {laneB, "标记B", "答复B"}} {
		got, err := st.Graph(ctx, "web", want.graph.ID)
		if err != nil {
			t.Fatalf("查 %s: %v", want.graph.ID, err)
		}
		if got == nil {
			t.Fatalf("查不到 %s —— 这一路被另一路顶掉了", want.graph.ID)
		}
		if got.TurnID != want.graph.TurnID || got.Round != want.graph.Round {
			t.Fatalf("读回 %s 的路/轮次 = %q/%d，want %q/%d",
				want.graph.ID, got.TurnID, got.Round, want.graph.TurnID, want.graph.Round)
		}
		if !hasNodeLabel(got, want.input) || !hasNodeLabel(got, want.output) {
			t.Fatalf("读回 %s 的图不是这一路的（输入/回复节点对不上）：%+v", want.graph.ID, got.Nodes)
		}
	}

	turns, err := st.Turns(ctx, "web", 0)
	if err != nil {
		t.Fatalf("Turns: %v", err)
	}
	if len(turns) != 2 {
		t.Fatalf("Turns 条数 = %d，want 2（两路都能翻到）", len(turns))
	}
	if latest, err := st.Latest(ctx, "web"); err != nil || latest == nil {
		t.Fatalf("Latest = %+v / %v，want 一条真图", latest, err)
	}
}

// TestStore_SameLaneTwiceIsOneRow_边界 钉住唯一键的另一半语义：**同一路**二次收尾
// 是同轮覆盖（一行），这正是分桶改造**不该改动**的既有行为 —— 两路分行靠的是两路
// 各有各的 id，而不是靠放弃 upsert。
func TestStore_SameLaneTwiceIsOneRow_边界(t *testing.T) {
	t.Parallel()
	st, conn := newTestStore(t, nil)
	first := laneGraph("web", 7, "turn-x", "标记A", "答复A")
	if err := st.RecordRound(first); err != nil {
		t.Fatalf("首次落盘: %v", err)
	}
	again := laneGraph("web", 7, "turn-x", "标记A", "答复A（重放）")
	if err := st.RecordRound(again); err != nil {
		t.Fatalf("二次落盘: %v", err)
	}
	if got := countRows(t, conn, "web"); got != 1 {
		t.Fatalf("同一路二次收尾后行数 = %d，want 1（同轮覆盖）", got)
	}
	back, err := st.Graph(context.Background(), "web", first.ID)
	if err != nil || back == nil {
		t.Fatalf("Graph = %+v / %v", back, err)
	}
	if !hasNodeLabel(back, "答复A（重放）") {
		t.Fatalf("覆盖后的图不是最新那份：%+v", back.Nodes)
	}
}

// countRows 数该会话在 flow_graphs 里的行数（结构级断言，不经过 store 的读方法）。
func countRows(t *testing.T, conn *sql.DB, session string) int {
	t.Helper()
	var n int
	if err := conn.QueryRowContext(context.Background(),
		`SELECT COUNT(*) FROM flow_graphs WHERE session_id = ?`, session).Scan(&n); err != nil {
		t.Fatalf("COUNT 行数: %v", err)
	}
	return n
}

// hasNodeLabel 报告图里有没有标签等于 label 的节点。
func hasNodeLabel(g *trace.FlowGraph, label string) bool {
	for _, n := range g.Nodes {
		if n.Label == label {
			return true
		}
	}
	return false
}