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 }