diff --git a/internal/runtime/flowgraph/flowgraph.go b/internal/runtime/flowgraph/flowgraph.go index 8f79270..3621fa8 100644 --- a/internal/runtime/flowgraph/flowgraph.go +++ b/internal/runtime/flowgraph/flowgraph.go @@ -164,6 +164,11 @@ func sourcedLog(l *slog.Logger) *slog.Logger { // 图 id(`trace.FlowGraph.ID`,也是本表的冲突键)。落盘时两个都写,使「按消息的那一路 // 查图」(`GraphsByTurnID`)是一次等值索引查询,而不是拿图 id 去反解。二次收尾同样 // 更新它:同一条路的图又被收一次时,标识必须跟着走到最新的那一份。 +// +// ⚠️ `lane_turn_id` 的**列**口径(迁移 0023 的注释)写的是「单路回合的图不带该键 ⇒ +// 落盘为空」,那是当时写侧的实情;写侧现已改为**每一路都取权威回合标识**(见 +// `loop.FlowPublisher.StartRoundWithTurnID`)。列的形状与语义一字未改,只是**新的 +// 单路行不再必然为空** —— 老行仍是空串(那时图里根本没存这个标识,任何回填都是编造)。 const upsertRoundSQL = `INSERT INTO flow_graphs (session_id, turn_id, lane_turn_id, graph_json, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?) @@ -179,8 +184,9 @@ ON CONFLICT(session_id, turn_id) DO UPDATE SET // 它同时盖上这一路的**回合标识**(`g.TurnID` → `lane_turn_id`,迁移 0023),使消息面上 // 「这条回复当时调了什么工具」能按**等值**查到这一轮的图(见 `GraphsByTurnID`)。 // 标识为空时如实写空串:那一行因此按路查不到,读取侧让 `toolCalls` 整体缺席 —— -// 这正是契约的语义,绝不用时间窗或正文去猜(`g.TurnID` 为空的两类情形见迁移 0023 -// 的注释:v0.77 之前的旧行与**当前**的单路回合)。 +// 这正是契约的语义,绝不用时间窗或正文去猜。`g.TurnID` 为空只剩一类情形: +// **没有 ctx 通道**的调用方(未接编排器的写入路径、旧后端落下的历史行)—— +// 编排器驱动的每一路(含普通单路回合与队首续跑路)都有权威回合标识。 // // 失败**上抛**给调用方(发布器记 WARN 后继续),绝不 panic、绝不改变图 —— 图是可视化 // 资产,落盘失败不该影响会话本身。 diff --git a/internal/runtime/loop/flow.go b/internal/runtime/loop/flow.go index dc9adc5..b5bb100 100644 --- a/internal/runtime/loop/flow.go +++ b/internal/runtime/loop/flow.go @@ -1,6 +1,7 @@ package loop import ( + "context" "log/slog" "sort" "sync" @@ -129,11 +130,39 @@ func NewFlowPublisher(publish func(subject string, payload any), allow func(sess // id(引入本切片之前的所有调用点原样可用),真并行路由经 flowOwner 传 // `{会话}#{路}`。两条路各起各的图 ⇒ 后起的那一路绝不重置前一路的图。 // +// 本入口**没有 ctx 通道**,因此这一路的身份只能取**归属桶里的路标识**:真并行路的桶 +// 正是网关按该路的 turnId 拼的,两者逐字同值;单路的桶里没有路标识 ⇒ 图上该键缺席。 +// 要覆盖**全部编排回合**(普通单路回合与队首续跑路也算)请用 StartRoundWithTurnID, +// 它是 `Run` 走的入口。 +func (p *FlowPublisher) StartRound(owner, inputText string) { + _, lane := splitFlowOwner(owner) + p.startRound(owner, lane, inputText) +} + +// StartRoundWithTurnID 与 StartRound 同一件事,但这一路的**权威回合标识**取自 ctx +// (`TurnIDFrom`,design/15 §2.5.9):编排器给**每一路**都盖它 —— 普通单路回合、队首 +// 续跑路、真并行路都有,因此它是「这张图属于哪一回合」的唯一正当来源,也是落盘 +// `flow_graphs.lane_turn_id` 与消息面 `toolCalls[]`(按 `chat_messages.turn_id` 等值 +// 回看)能对上的前提。 +// +// ctx 上没有权威标识时(后台思考、未接编排器的调用方)**保持缺席**:快照上的 `turnId` +// 键与落盘行的 `lane_turn_id` 都是空 ⇒ 读侧查不到这一轮的图。那是契约允许的诚实空态, +// **绝不**拿会话 id、时间戳之类去臆造一个身份。 +// +// 归属桶(owner)的语义一字未改:真并行两路仍各起各的图(桶是 `{会话}#{路}`),单路的 +// 桶仍是会话 id。本方法只决定**这张图属于哪一回合**,不决定它是不是另一张图。 +func (p *FlowPublisher) StartRoundWithTurnID(ctx context.Context, owner, inputText string) { + p.startRound(owner, TurnIDFrom(ctx), inputText) +} + +// startRound 是两条入口的共用实现:turnID 是这一路的权威回合标识(空 = 缺席,绝不 +// 臆造),owner 是归属桶。 +// // 轮次号只在**上一轮的路全部收尾**之后推进(inflight 为空),因此「同一条消息 // 引出的两路」共享同一个轮次号;新一轮开始时丢掉更早轮次的桶,内存因此不随路数 // 无界增长(图的历史在落盘里,见 flowgraph 包)。 -func (p *FlowPublisher) StartRound(owner, inputText string) { - session, lane := splitFlowOwner(owner) +func (p *FlowPublisher) startRound(owner, turnID, inputText string) { + session, _ := splitFlowOwner(owner) if session == "" || owner == "" || !p.allowed(session) { return } @@ -141,7 +170,7 @@ func (p *FlowPublisher) StartRound(owner, inputText string) { defer p.mu.Unlock() round := p.beginRoundLocked(session, owner) g := trace.NewRunGraph(session) - g.SetTurnID(lane) + g.SetTurnID(turnID) g.SetSeq(p.seq[session]) g.Start(round, time.Now()) g.SeedInput(inputText) diff --git a/internal/runtime/loop/flow_lane_test.go b/internal/runtime/loop/flow_lane_test.go index c2ab93d..cf9cb52 100644 --- a/internal/runtime/loop/flow_lane_test.go +++ b/internal/runtime/loop/flow_lane_test.go @@ -125,16 +125,20 @@ func TestFlowPublisher_RoundsShareRoundNumberButNeverID(t *testing.T) { } } -// TestFlowPublisher_SingleLegUnchanged 钉住**单路逐字不变**:归属桶就是会话 id, -// 图 id 仍是 `{会话}-{轮次}`(没有路后缀),帧上 `turnId` 键缺席。既有回归 -// (flow_snapshot_test.go 的 web-1/web-2)与本条是同一口径的两侧。 +// TestFlowPublisher_SingleLegUnchanged 钉住**老入口(无 ctx 通道)的单路逐字不变**: +// 归属桶就是会话 id,图 id 仍是 `{会话}-{轮次}`(没有后缀),帧上 `turnId` 键缺席。 +// 既有回归(flow_snapshot_test.go 的 web-1/web-2)与本条是同一口径的两侧。 +// +// 单路在**有 ctx 通道**时的形状(覆盖全部编排回合的那条路)见 flow_turnid_test.go: +// 那里 `turnId` 是**在场**的(权威回合标识),因为「单路没有权威标识」这件事本身 +// 已经随 CODE-TC-2 消失 —— 本入口的缺席只说明**它没有 ctx 可读**。 func TestFlowPublisher_SingleLegUnchanged(t *testing.T) { p := NewFlowPublisher(nil, func(string) bool { return true }) p.StartRound("web", "你好") p.EndRound("web", "你也好") g := p.LastGraph("web") if g == nil || g.ID != "web-1" || g.TurnID != "" { - t.Fatalf("单路图 = %+v,want id=web-1 且无 turnId(逐字不变)", g) + t.Fatalf("单路图 = %+v,want id=web-1 且无 turnId(老入口逐字不变)", g) } } diff --git a/internal/runtime/loop/flow_turnid_test.go b/internal/runtime/loop/flow_turnid_test.go new file mode 100644 index 0000000..3e2500f --- /dev/null +++ b/internal/runtime/loop/flow_turnid_test.go @@ -0,0 +1,279 @@ +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 +} diff --git a/internal/runtime/loop/run.go b/internal/runtime/loop/run.go index 394ef20..9f9a02e 100644 --- a/internal/runtime/loop/run.go +++ b/internal/runtime/loop/run.go @@ -87,9 +87,15 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa // 归属取**桶**(`{会话}[#{路}]`,见 FlowOwnerKey):真并行时本 Run 属于某一条路, // 两路因此各起各的图、各收各的图 —— 后起的那一路绝不重置前一路(design/15 // §2.5.10-1)。单路时桶逐字等于 sessionID,行为与引入本切片之前完全一致。 + // + // 这一路的**权威回合标识**(design/15 §2.5.9)取自 ctx:编排器给每一路都盖了它 + // (普通单路回合与队首续跑路算,见 `WithTurnID` / `beginLegLocked`),因此图上 + // `turnId` 与落盘行 `lane_turn_id` 与 `chat_messages.turn_id` 同值同源 —— + // 消息面「这条回复当时调了什么工具」据此按**等值**查回本轮的图(`toolCalls[]`)。 + // ctx 上没有它时(后台思考 / 未接编排器)如实缺席,绝不臆造。 if _, isBranch := flowBranch(ctx); l.flow != nil && !isBranch { owner := flowOwner(ctx, sessionID) - l.flow.StartRound(owner, lastUserText(msgs)) + l.flow.StartRoundWithTurnID(ctx, owner, lastUserText(msgs)) defer func() { l.flow.EndRound(owner, res.Text) }() } return l.driveTurn(ctx, sessionID, msgs, startStep, res) diff --git a/internal/runtime/trace/graph.go b/internal/runtime/trace/graph.go index 7564c76..de0a1d4 100644 --- a/internal/runtime/trace/graph.go +++ b/internal/runtime/trace/graph.go @@ -66,10 +66,12 @@ type FlowGraph struct { Edges []FlowEdge `json:"edges"` // SessionID 是拥有这张图的会话(帧壳与落盘行的归属坐标)。 SessionID string `json:"sessionId,omitempty"` - // TurnID 是这张图所属**路**的权威回合标识(design/15 §2.5.9):真并行时 - // 同一会话有多条路,每条路各有自己的图与本键。**缺席**语义:这条路没有 - // 权威标识(单路的历史帧、未接编排器的调用方、旧后端)——客户端据此把图 - // 归到会话那一张(兼容口径),绝不臆造一个 id。 + // TurnID 是这张图所属**路**的权威回合标识(design/15 §2.5.9):编排器给**每一路** + // 都发(普通单路回合、队首续跑路、真并行路都有),因此它就是这张图属于哪一回合的 + // 答案,也是落盘 `flow_graphs.lane_turn_id` 与 `chat_messages.turn_id` 的同值。 + // **缺席**语义:这条路没有权威标识(未接编排器的调用方、后台思考、旧后端)—— + // 客户端据此把图归到会话那一张(兼容口径),读侧据此让该轮的 `toolCalls` 整体 + // 缺席;两侧都**绝不**臆造一个 id。 TurnID string `json:"turnId,omitempty"` // Round 是这张图所属的**会话轮次号**(每会话单调递增;同一轮的各条路共享它)。 // 客户端据此只保留最新一轮的泳道,避免旧轮的路留在页面上。 @@ -105,7 +107,10 @@ type RunGraph struct { // 未经 SetSeq 建图的调用方(图构建器单测)保留 `{会话}-{轮次}` 的老形状。 seq int // turnID 是本图所属**路**的权威回合标识(design/15 §2.5.9)。空串表示这条路 - // 没有权威标识(单路 / 未接编排器 / 旧调用方),此时图上该键缺席。 + // 没有权威标识(未接编排器 / 旧调用方),此时图上该键缺席。它有一个可见后果: + // 图 id 的后缀取它(见 graphID),因此有权威标识的单路图也不再是 `{会话}-{序号}` + // 而是 `{会话}-{序号}#{回合}` —— 落盘唯一键 `(session_id, turn_id)` 因此跨进程 + // 不撞(序号只在进程内递增、重启归零,而回合标识每回合全新)。 turnID string nodes map[string]*FlowNode order []string @@ -151,7 +156,7 @@ func (g *RunGraph) Start(round int, now time.Time) { } // SetTurnID 盖上这张图所属**路**的权威回合标识(design/15 §2.5.9)。空串是 -// 合法入参(单路 / 未接编排器),此时快照里该键缺席(omitempty),客户端回落 +// 合法入参(未接编排器的调用方),此时快照里该键缺席(omitempty),客户端回落 // 到「按会话归一张图」的兼容口径。 func (g *RunGraph) SetTurnID(turnID string) { g.turnID = turnID } @@ -195,9 +200,11 @@ func (g *RunGraph) SeedOutput(text string) string { const laneIDSep = "#" // Snapshot returns a copy of the graph safe to encode and publish. 图 id 是 -// `{会话}-{轮次}`;这张图属于一条**路**时追加 `#{路标识}` —— 真并行下同一会话 -// 同一轮的两路必须各有各的 id,否则落盘(`flow_graphs` 的 `(session_id, turn_id)` -// 唯一键)会把其中一路整行顶掉、客户端也分不出两张图(design/15 §2.5.10-1)。 +// `{会话}-{轮次}`;这张图属于一条**有权威回合标识的路**时追加 `#{回合标识}` —— +// 真并行下同一会话同一轮的两路必须各有各的 id,否则落盘(`flow_graphs` 的 +// `(session_id, turn_id)` 唯一键)会把其中一路整行顶掉、客户端也分不出两张图 +// (design/15 §2.5.10-1);单路(编排器也给它盖了权威标识)同样带后缀,因此它的 +// 图 id 跨进程不撞(见 RunGraph.turnID)。 func (g *RunGraph) Snapshot() *FlowGraph { nodes := make([]FlowNode, 0, len(g.order)) for _, id := range g.order { @@ -215,9 +222,10 @@ func (g *RunGraph) Snapshot() *FlowGraph { } } -// graphID 拼出这张图的 id:`{会话}-{序号}[#{路标识}]`。序号取会话级建图序号 +// graphID 拼出这张图的 id:`{会话}-{序号}[#{权威回合标识}]`。序号取会话级建图序号 // (未设置时退回轮次号,见 RunGraph.seq),因此**同一轮里的每一条路也各有各的 id**; -// 路标识只在「这条路有权威标识」时追加。 +// 回合标识只在「这条路有权威标识」时追加(单路在有 ctx 通道时也有 —— 见 +// loop.FlowPublisher.StartRoundWithTurnID)。 func (g *RunGraph) graphID() string { n := g.seq if n == 0 {