feat(flow): toolCalls 覆盖率追到全部编排回合(L3)—— 图带权威回合标识
- 新增 StartRoundWithTurnID(ctx, owner, text):从 ctx 取权威回合标识(与 chat_messages.turn_id
同值);Run 的唯一生产建图入口切到它;原 StartRound 语义逐字不变(1 行包装)
- 归属桶(并行按路分图)保持不变;落盘查询键 lane_turn_id 对单路也有值 ⇒ 消息面按等值查得到图
- 无权威标识时**保持缺席**(不拿桶里的路标识顶上:不许臆造)
- 不需要迁移 0024(0023 的列+索引已在位;老单路行标识根本没存过,回填=编造)
形状变化(已登记契约):graph.id 对单路也追加 #{回合标识}(web-7 → web-7#turn-…),
顺带修掉「重启后单路轮次 upsert 掉旧行」的隐患。3 个变异自检(退回老入口/无视 ctx/臆造回退)。
This commit is contained in:
1 parent
6475eb875d
commit
4a937bd269
6 files changed
+353
-21
No files matched your search
@@ -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、绝不改变图 —— 图是可视化
|
||||
// 资产,落盘失败不该影响会话本身。
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in new issue
Block a user