fix(flow): graph_update 按路分桶——修真并行两路的图互相覆盖(FV2)
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled

【取证(症状成立,且比登记描述更糟)】两路真并行时两路最终落盘**同一个 turn_id**,
其中一路的图彻底消失,且 A 路的工具被画在 B 路的图上。逐层定位:发布器 graphs 只按
sessionID 建键(后起的路整张顶掉前者)· 帧载荷基线的键集被既有测试精确断言为
{id,nodes,edges,sessionId,startedAt,endedAt}(无任何区分路的键)· 前端每会话单桶 ·
落盘 (session_id,turn_id) 唯一键两路相撞 · 冷启动 LastGraph 只有一张。
该 bug 早已公开登记(design/15 §2.5.10-1「本轮未解决」、契约 §6-19「仍**未修**」)。

【修法】
- 归属桶 {会话}[#{路}](loop.FlowOwnerKey,与 checkpoint 键同字符口径);路只来自
  ParallelLegFrom(ctx) ⇒ 单路逐字不变(桶 = 会话 id),既有调用点语义零改动
- 图 id 的第二分量改用**建图序号**(每次 +1、永不复用):同一轮里可以有并行路 +
  队首续跑的串行路,用轮次号会让两条串行路撞同一个 web-1
- 轮次号只在上一轮的路全部收尾后推进,并丢掉更早轮次的桶(否则内存随路数无界增长)
- 新增只读出口 LatestLanes ⇒ GET /admin/flow/graph 的 lanes[](最近一轮每一路),
  graph 仍是「最新一路」= 向后兼容
- 同源洞一并修:子代理归属原取 AgentSpec.ParentSession(会话、不含路)⇒ 并行路里派的
  子代理 span 被静默丢弃;新增 AgentSpec.FlowOwner 承载「会话#路」,ParentSession 继续
  只承载会话(总线载荷 subagent_result.sessionId 不受影响)
- 前端:store 增 flowLanes(按 turnId 分格、按轮淘汰、硬上限 4)+ LaneTracks 泳道卡;
  单路/未给 lanes 时渲染路径与改前逐字相同

【无迁移】turn_id 是自由文本、每路 id 唯一 ⇒ 0022 的唯一索引天然分两行(0023 仍空着)

【我的验收】① 隔离 worktree(30 文件)build/vet/linux/darwin 全过、69 包全绿 0 FAIL;
② 我亲手打红变异:让 FlowOwnerKey 忽略路 ⇒ TestFlowPublisher_ParallelLegsDoNotOverwriteEachOther
FAIL;③ 前端独立复跑 npx tsc --noEmit + npm run build 通过(4.12s)

⚠️ 契约 v0.77 的登记在紧随其后的 docs(contract) 提交里(同一个工作时段内)。
This commit is contained in:
AskaEth committed 2026-10-06 08:16:03 +08:00
1 parent c9144c1096
commit 89434eadef
30 files changed
+2039 -148

No files matched your search

+210 -51
View File
@@ -2,6 +2,7 @@ package loop
import (
"log/slog"
"sort"
"sync"
"time"
@@ -21,6 +22,9 @@ const flowLogSource = "runtime/loop/flow"
// allowedSessions 是用于控制哪些会话可发布
// flow 图的谓词签名(主人的 web 会话),使 IM/陌生人/子代理轮次绝不
// 泄露进管理面板。谓词为 nil 时允许所有会话。
//
// 谓词收的是**会话 id**(不是归属桶):调用方先经 splitFlowOwner 拆掉路标识,
// 因此白名单永远只按会话判 —— 路的存在绝不放宽或收紧任何一档。
type allowedSessions func(sessionID string) bool
// RoundRecorder 把一条**已收尾轮次**的图落盘(design/19 §4 的 P3「多轮历史回放」;
@@ -30,6 +34,10 @@ type allowedSessions func(sessionID string) bool
// 上调用),失败**上抛**给发布器(发布器记 WARN 后继续)—— 图是可视化资产,落盘失败
// 绝不允许影响会话收尾,也绝不改变实时帧。
//
// 「每轮一条」在真并行下的含义是**每一路一条**:两路各有自己的图 id(见
// trace.RunGraph 的 id 口径),因此 `flow_graphs` 的 `(session_id, turn_id)` 唯一键
// 不会被两路撞上(design/15 §2.5.10-1)。
//
// 白名单不在实现方:EndRound 对发布会话范围外的会话**直接返回**,因此落盘出口天然
// 只被白名单内的会话触发(复用 flowSessionAllowed,不存在第二份白名单)。
type RoundRecorder interface {
@@ -50,22 +58,39 @@ func WithFlowLogger(log *slog.Logger) FlowPublisherOption {
return func(p *FlowPublisher) { p.log = log }
}
// FlowPublisher 从循环的 trace span 累积每会话 flow 图,
// FlowPublisher 从循环的 trace span 累积**按路**的 flow 图,
// 并在每个 span 开始/结束时发布 ws.graph_update,给前端
// 一张实时的 input→think→tool→output 图(design/19 §1)。它容忍 nil
// (nil 发布者为空操作)且并发安全;会话图
// (nil 发布者为空操作)且并发安全;同一会话的图
// 在同一把互斥锁下串行化,因为并行的工具 span 会并发完成。
//
// 归属:一个会话拥有一个实时轮次。声明自己是另一会话分支的
// Run(loop.WithFlowBranch)绝不调用 StartRound/EndRound,
// 而是把其 span 发布到所属会话下,因此子代理渲染为
// 归属:**一个 Run 拥有一个路 bucket**。单路时 bucket 就是会话 id(行为与引入
// 「路」之前逐字相同);真并行时每一路各有自己的 bucket(`{会话}#{路}`,见
// FlowOwnerKey),因此两路的节点绝不互相长在对方的图上、也绝不互相重置
// (design/15 §2.5.10-1 登记的那条「graph_update 互相顶掉」)。声明自己是另一
// 会话分支的 Run(loop.WithFlowBranch)绝不调用 StartRound/EndRound,
// 而是把其 span 发布到所属路的 bucket 下,因此子代理渲染为
// 父图的一条泳道,而不是第二张图(design/19 §2)。
type FlowPublisher struct {
mu sync.Mutex
publish func(subject string, payload any)
allow allowedSessions
graphs map[string]*trace.RunGraph
rounds map[string]int
// graphs 是「归属桶 → 该桶当前这张图」。键见 FlowOwnerKey。
graphs map[string]*trace.RunGraph
// seq 是**每会话**的建图序号:每次 StartRound 加一,并写进图 id
// (`{会话}-{序号}[#{路}]`)。用序号而不是轮次号做 id 的第二个分量,是为了让
// 「同一轮里的第二条路」也拿到一个全新的 id —— 否则两路会撞同一个
// `flow_graphs.turn_id`(唯一键),后收尾的那一路把前一路整行顶掉。
seq map[string]int
// rounds 是每会话当前的**轮次号**(= 该轮第一条路的建图序号)。同一轮的各条路
// 共享它,客户端据此只保留最新一轮的泳道。
rounds map[string]int
// roundOf 是「归属桶 → 它属于哪一轮」。
roundOf map[string]int
// inflight 是每会话**在飞的桶**集合。新一轮的判据是「上一轮的路全部收尾」
// (集合为空),因此它是轮次号推进的唯一依据,也是内存不随路数无界增长的
// 依据(新一轮起时丢掉更早轮次的桶)。
inflight map[string]map[string]bool
// record 是每轮收尾的落盘出口(design/19 §4 P3;切片 F2)。nil = 不落盘
// (保持 F1 之前的纯内存行为,且测试可只测发布)。
record RoundRecorder
@@ -81,10 +106,13 @@ type FlowPublisher struct {
// 可变项经 Functional Options 传入(不改变既有两参调用):落盘出口(F2)与日志。
func NewFlowPublisher(publish func(subject string, payload any), allow func(sessionID string) bool, opts ...FlowPublisherOption) *FlowPublisher {
p := &FlowPublisher{
publish: publish,
allow: allow,
graphs: make(map[string]*trace.RunGraph),
rounds: make(map[string]int),
publish: publish,
allow: allow,
graphs: make(map[string]*trace.RunGraph),
seq: make(map[string]int),
rounds: make(map[string]int),
roundOf: make(map[string]int),
inflight: make(map[string]map[string]bool),
}
for _, opt := range opts {
if opt != nil {
@@ -94,40 +122,83 @@ func NewFlowPublisher(publish func(subject string, payload any), allow func(sess
return p
}
// StartRound 为 sessionID 开始一张新图,用用户文本
// 播种一个 input 节点并发布初始帧。它重置该会话的上一
// 轮次,因此每个用户 turn 都是自己的一张图。
func (p *FlowPublisher) StartRound(sessionID, inputText string) {
if sessionID == "" || !p.allowed(sessionID) {
// StartRound 为 owner 开始一张新图,用用户文本
// 播种一个 input 节点并发布初始帧。
//
// owner 是**归属桶**(`{会话}[#{路}]`,见 FlowOwnerKey):单路调用方传的就是会话
// id(引入本切片之前的所有调用点原样可用),真并行路由经 flowOwner 传
// `{会话}#{路}`。两条路各起各的图 ⇒ 后起的那一路绝不重置前一路的图。
//
// 轮次号只在**上一轮的路全部收尾**之后推进(inflight 为空),因此「同一条消息
// 引出的两路」共享同一个轮次号;新一轮开始时丢掉更早轮次的桶,内存因此不随路数
// 无界增长(图的历史在落盘里,见 flowgraph 包)。
func (p *FlowPublisher) StartRound(owner, inputText string) {
session, lane := splitFlowOwner(owner)
if session == "" || owner == "" || !p.allowed(session) {
return
}
p.mu.Lock()
defer p.mu.Unlock()
round := p.rounds[sessionID] + 1
g := trace.NewRunGraph(sessionID)
round := p.beginRoundLocked(session, owner)
g := trace.NewRunGraph(session)
g.SetTurnID(lane)
g.SetSeq(p.seq[session])
g.Start(round, time.Now())
g.SeedInput(inputText)
p.graphs[sessionID] = g
p.rounds[sessionID] = round
p.graphs[owner] = g
p.roundOf[owner] = round
if p.inflight[session] == nil {
p.inflight[session] = make(map[string]bool, 2)
}
p.inflight[session][owner] = true
p.publishLocked(g)
}
// beginRoundLocked 分配(或沿用)本轮次号并清掉更早轮次的桶。调用方必须持有 p.mu。
func (p *FlowPublisher) beginRoundLocked(session, owner string) int {
p.seq[session]++
if len(p.inflight[session]) == 0 {
p.rounds[session] = p.seq[session]
p.dropEarlierRoundsLocked(session)
}
return p.rounds[session]
}
// dropEarlierRoundsLocked 丢掉该会话**更早轮次**留下的桶与图。图的历史由落盘承载
// (`flow_graphs`),实时桶只需覆盖当前这一轮 —— 否则每一条并行路(turnId 每回合
// 全新、bucket 键因此永不复用)都会在内存里永久留一张图。调用方必须持有 p.mu。
func (p *FlowPublisher) dropEarlierRoundsLocked(session string) {
current := p.rounds[session]
for owner, round := range p.roundOf {
if round >= current {
continue
}
if s, _ := splitFlowOwner(owner); s != session {
continue
}
delete(p.roundOf, owner)
delete(p.graphs, owner)
}
}
// EndRound 用回复播种一个 output 节点并发布最终帧,
// 以此收尾该轮次。它保留该图,使错过最终帧的客户端能重放最后一份快照 ——
// **重放的出口是 `LastGraph`**(网关只读查询,契约 v0.71),而不是又发一帧:
// 实时帧向来只推给当时在线的连接,刷新/晚开的页面只能靠只读拉取补齐
// (`graph_update` 本身不重放)。
// 以此收尾该路的图。它保留该图,使错过最终帧的客户端能重放最后一份快照 ——
// **重放的出口是 `LastGraph` / `LatestLanes`**(网关只读查询,契约 v0.71/v0.77),
// 而不是又发一帧:实时帧向来只推给当时在线的连接,刷新/晚开的页面只能靠只读拉取
// 补齐(`graph_update` 本身不重放)。
//
// 收尾也是**落盘**的唯一时机(切片 F2;design/19 §4 的 P3「多轮历史回放」):每轮落
// **一条**(绝不是每 span 一条),且只在该会话处于发布会话白名单内时发生(本方法对
// 白名单外的会话直接返回 ⇒ 落盘出口天然只被白名单会话触发)。深拷贝与写库在**放锁
// 之后**做:一次库写绝不把其他会话的 span 发布挡住;失败只记 WARN,绝不影响本轮收尾。
func (p *FlowPublisher) EndRound(sessionID, outputText string) {
if sessionID == "" || !p.allowed(sessionID) {
// 收尾也是**落盘**的唯一时机(切片 F2;design/19 §4 的 P3「多轮历史回放」):每轮
// 落**一条**(绝不是每 span 一条),真并行下**每一路各一条**;且只在该会话处于发布
// 会话白名单内时发生(本方法对白名单外的会话直接返回 ⇒ 落盘出口天然只被白名单会话
// 触发)。深拷贝与写库在**放锁之后**做:一次库写绝不把其他会话的 span 发布挡住;
// 失败只记 WARN,绝不影响本轮收尾。
func (p *FlowPublisher) EndRound(owner, outputText string) {
session, _ := splitFlowOwner(owner)
if session == "" || owner == "" || !p.allowed(session) {
return
}
p.mu.Lock()
g := p.graphs[sessionID]
g := p.graphs[owner]
if g == nil {
p.mu.Unlock()
return
@@ -138,29 +209,33 @@ func (p *FlowPublisher) EndRound(sessionID, outputText string) {
if p.record != nil {
snap = g.Snapshot().Clone()
}
// 本路退出在飞集合(**只删自己的桶**):集合空了才说明整轮收尾,下一轮才会
// 拿到新的轮次号 —— 因此轮次号在同轮并行的两路之间保持同一个值。
delete(p.inflight[session], owner)
p.mu.Unlock()
if snap != nil {
if err := p.record.RecordRound(snap); err != nil {
p.logPersistError(sessionID, snap.ID, err)
p.logPersistError(session, snap.ID, err)
}
}
}
// MarkInputSteered 给本轮的 input 节点打上 `steered:true` 标记并发布一帧
// (design/15 §5 / 裁决 13.4.5「注入时给 input 节点打 steered:true 标记」)。
// 它让链路图如实表达「这一轮的输入在回合中途被追加过」——图仍然只有一条主干
// MarkInputSteered 给本路的 input 节点打上 `steered:true` 标记并发布一帧
// (design/15 §5 / 裁决 13.4.5「注入时给 input 节点贴 steered:true 标记」)。
// 它让链路图如实表达「这一路的输入在回合中途被追加过」——图仍然只有一条主干
// (I1:同会话恒只有一个在飞回合),而不是被画成第二轮。
//
// nil 接收者、空会话、发布白名单外的会话、或该会话还没有图时都是空操作;标记
// nil 接收者、空归属、发布白名单外的会话、或该路还没有图时都是空操作;标记
// 只并入既有 input 节点(Upsert 的合并语义),绝不新建节点、绝不改动它的标签。
func (p *FlowPublisher) MarkInputSteered(sessionID string) {
if p == nil || sessionID == "" || !p.allowed(sessionID) {
func (p *FlowPublisher) MarkInputSteered(owner string) {
session, _ := splitFlowOwner(owner)
if p == nil || owner == "" || session == "" || !p.allowed(session) {
return
}
p.mu.Lock()
defer p.mu.Unlock()
g := p.graphs[sessionID]
g := p.graphs[owner]
if g == nil {
return
}
@@ -174,13 +249,13 @@ func (p *FlowPublisher) MarkInputSteered(sessionID string) {
// SpanStarted 为一个刚开始的 span 记录一个运行中的占位
// 节点。state_transition span 会被图构建器省略。
func (p *FlowPublisher) SpanStarted(sessionID string, sp *trace.Span) {
if sp == nil || sessionID == "" || !p.allowed(sessionID) {
func (p *FlowPublisher) SpanStarted(owner string, sp *trace.Span) {
if sp == nil {
return
}
p.mu.Lock()
defer p.mu.Unlock()
g := p.graphs[sessionID]
g := p.bucketLocked(owner)
if g == nil {
return
}
@@ -190,14 +265,14 @@ func (p *FlowPublisher) SpanStarted(sessionID string, sp *trace.Span) {
}
// SpanFinished 把一个刚结束的 span 作为最终节点(状态 done/error)
// 并入会话图并发布。必须在 span 的 Finish 之后调用。
func (p *FlowPublisher) SpanFinished(sessionID string, sp *trace.Span) {
if sp == nil || sessionID == "" || !p.allowed(sessionID) {
// 并入该路的图并发布。必须在 span 的 Finish 之后调用。
func (p *FlowPublisher) SpanFinished(owner string, sp *trace.Span) {
if sp == nil {
return
}
p.mu.Lock()
defer p.mu.Unlock()
g := p.graphs[sessionID]
g := p.bucketLocked(owner)
if g == nil {
return
}
@@ -205,16 +280,30 @@ func (p *FlowPublisher) SpanFinished(sessionID string, sp *trace.Span) {
p.publishLocked(g)
}
// LastGraph 返回 sessionID **最近一轮**链路图的一份只读深拷贝快照,供
// bucketLocked 取归属桶当前那张图;归属非法或该路还没有图时返回 nil。调用方必须
// 持有 p.mu。白名单判据与建图处同一处(bucket 拆出的会话),因此**不可能**让
// 白名单外的会话经 span 折叠进图。
func (p *FlowPublisher) bucketLocked(owner string) *trace.RunGraph {
session, _ := splitFlowOwner(owner)
if owner == "" || session == "" || !p.allowed(session) {
return nil
}
return p.graphs[owner]
}
// LastGraph 返回 sessionID **最近一路**链路图的一份只读深拷贝快照,供
// 管理面板在「轮次已经结束、实时帧已经错过」时补拉(契约 v0.71 的
// `GET /api/v1/admin/flow/graph`;design/19 §4 的 P3「多轮历史回放」的
// 最小一步)。
//
// 「最近」= 该会话所有在内存的路里**开始最晚**的那一路(同刻用图 id 定序)。
// 真并行时它只是最近的一路,**全貌**是 `LatestLanes`(契约 v0.77 的 `lanes[]`)。
//
// 它是**只读**的:不发布、不改图、不动轮次计数 —— 因此补拉绝不会在
// 管理面板里凭空造出一帧 `graph_update`,也绝不改任何会话的实时图。
//
// 该会话还没有图时返回 nil:从未跑过带 span 的轮次、或该会话本就**不在
// 发布会话范围内**(`StartRound` 对白名单外的会话直接返回 ⇒ 从不建图,
// 发布会话范围**(`StartRound` 对白名单外的会话直接返回 ⇒ 从不建图,
// 隔离渠道会话因此天然查不到)。调用方**必须**把 nil 读作「确实没有图」,
// 绝不允许据此补画任何节点或假图。
//
@@ -226,11 +315,81 @@ func (p *FlowPublisher) LastGraph(sessionID string) *trace.FlowGraph {
}
p.mu.Lock()
defer p.mu.Unlock()
g := p.graphs[sessionID]
if g == nil {
latest := p.latestLocked(sessionID)
if latest == nil {
return nil
}
return g.Snapshot().Clone()
return latest.Clone()
}
// LatestLanes 返回 sessionID **最近一轮每一路**的图快照(每条路一份深拷贝),
// 按路的开始时间升序(最早的在前)。契约 v0.77 的 `lanes[]` 就是它:真并行时
// 同一会话同一轮有多条路,`LastGraph` 只给出其中最新的一条,管理面板要靠本方法
// 才看得全(design/19 §2「并行分支独立显示」)。
//
// 没有图时返回空切片(不是 nil):调用方据此渲染诚实空态,绝不补画。
// 与 LastGraph 一样是**只读**的,且返回深拷贝(读者在锁外编码)。
func (p *FlowPublisher) LatestLanes(sessionID string) []*trace.FlowGraph {
if p == nil || sessionID == "" {
return nil
}
p.mu.Lock()
defer p.mu.Unlock()
round := p.rounds[sessionID]
owners := make([]string, 0, 2)
for owner, r := range p.roundOf {
if r != round {
continue
}
if s, _ := splitFlowOwner(owner); s == sessionID {
owners = append(owners, owner)
}
}
out := make([]*trace.FlowGraph, 0, len(owners))
for _, owner := range owners {
g := p.graphs[owner]
if g == nil {
continue
}
out = append(out, g.Snapshot().Clone())
}
sortLanes(out)
return out
}
// latestLocked 取该会话开始最晚的那一路的活图(没有时 nil)。调用方必须持有 p.mu。
func (p *FlowPublisher) latestLocked(sessionID string) *trace.FlowGraph {
var best *trace.FlowGraph
for owner, g := range p.graphs {
if s, _ := splitFlowOwner(owner); s != sessionID {
continue
}
snap := g.Snapshot()
if best == nil || laneNewer(snap, best) {
best = snap
}
}
return best
}
// laneNewer 报告 a 是否比 b 更「新」:先比开始时刻,同刻用图 id 定序(图 id 里含
// 会话级单调序号,因此定序不依赖时间分辨率)。
func laneNewer(a, b *trace.FlowGraph) bool {
if a.StartedAt.Equal(b.StartedAt) {
return a.ID > b.ID
}
return a.StartedAt.After(b.StartedAt)
}
// sortLanes 按「开始时刻升序、同刻按图 id」把一路路图排序,使一个轮次里的多条路
// 有一个确定的展示次序(绝不让展示次序随 map 遍历序抖动)。
func sortLanes(lanes []*trace.FlowGraph) {
sort.SliceStable(lanes, func(i, j int) bool {
if lanes[i].StartedAt.Equal(lanes[j].StartedAt) {
return lanes[i].ID < lanes[j].ID
}
return lanes[i].StartedAt.Before(lanes[j].StartedAt)
})
}
// publishLocked 发布 g 的一份快照。调用方必须持有 p.mu。
+85 -19
View File
@@ -1,6 +1,40 @@
package loop
import "context"
import (
"context"
"strings"
)
// flowOwnerSep 是 flow 图**归属桶**(`{会话}#{路}`)的分隔符。它与
// design/15 §2.5.9 的 checkpoint 归属键(`{sessionID}#{turnID}`)以及
// trace 的图 id 拼接口径(`{会话}-{轮次}#{路}`)取同一个字符:**一条路**
// 在流程图、checkpoint 与落盘行三处因此是同一个标识。
//
// 之所以能在同一个字符串里编码「会话 + 路」:会话 id 的取值族是
// `web` / `web:<uuid>` / `sess-backthink`(`cmd/cyrene` 的 flowSessionAllowed),
// 权威 turnId 是 `turn-<hex>`(`gateway.newTurnID`),两者都不含 `#`。
const flowOwnerSep = "#"
// FlowOwnerKey 拼出「会话[#路]」的 flow 归属桶;lane 为空时就是会话本身
// (单路 / 未声明并行的运行,逐字等于引入本特性之前的会话 id)。
//
// 导出它是因为**归属桶的串形状只有一个权威**:发布器的建图出口、子代理 span 的
// 归属、以及跨包的验收测试都按它拼串,任何一处自己拼 `#` 都会在换分隔符时静默走偏。
func FlowOwnerKey(sessionID, lane string) string {
if lane == "" {
return sessionID
}
return sessionID + flowOwnerSep + lane
}
// splitFlowOwner 把归属桶拆回(会话, 路)。没有 `#` 时整串就是会话、路为空 ——
// 这正是单路运行与**旧调用方**传给发布器的形态,因此拆分是向后兼容的。
func splitFlowOwner(owner string) (string, string) {
if i := strings.Index(owner, flowOwnerSep); i >= 0 {
return owner[:i], owner[i+1:]
}
return owner, ""
}
// flowBranchKey 是标记某个 Run 为另一会话 flow 图分支的 context key。
// 它由 orchestration 层(经 WithFlowBranch)在驱动
@@ -8,24 +42,28 @@ import "context"
type flowBranchKey struct{}
// WithFlowBranch 返回 ctx,把当前 Run 标记为分支:它的 span 并入
// parentSession 的实时 flow 图,且该 Run 自身既不开始也不结束
// parent 的实时 flow 图,且该 Run 自身既不开始也不结束
// 轮次 —— 由所属会话的轮次负责。orchestration 层正是靠它把
// 子代理的工作归属到派生它的会话,使并行的
// 子代理渲染为同一张图的泳道,而不是用各自的图
// 替换前端唯一的 flowGraph(design/19 §2)。
//
// 空的 parentSession 仍把该 Run 标记为分支:它不拥有轮次、
// parent 是**归属桶**(`{会话}#{路}`,见 FlowOwnerKey):并行路里派发的子代理必须
// 并进**那一路**的图,而不是会话主干那张 —— 否则两路的子代理节点会混进同一张图,
// 正是本切片要治的「两路互相覆盖」(design/15 §2.5.10-1)。单路时它逐字等于会话 id。
//
// 空的 parent 仍把该 Run 标记为分支:它不拥有轮次、
// 不发布任何东西。这正是无人归属到会话的后台子代理的
// 状态,它不得悄悄变成属于自己的一张图。nil ctx
// 原样返回。
func WithFlowBranch(ctx context.Context, parentSession string) context.Context {
func WithFlowBranch(ctx context.Context, parent string) context.Context {
if ctx == nil {
return ctx
}
return context.WithValue(ctx, flowBranchKey{}, parentSession)
return context.WithValue(ctx, flowBranchKey{}, parent)
}
// flowBranch 报告 WithFlowBranch 记录的父会话以及
// flowBranch 报告 WithFlowBranch 记录的父归属桶以及
// 该 Run 是否本就是分支。非分支的 Run 拥有自己的轮次并在
// 自己的 session id 下发布。
func flowBranch(ctx context.Context) (string, bool) {
@@ -40,27 +78,55 @@ func flowBranch(ctx context.Context) (string, bool) {
return s, true
}
// flowOwner 返回应当接收该 Run span 的 flow 图所属会话:
// flowOwner 返回应当接收该 Run span 的 flow 归属桶:
// 当该 Run 是子代理分支时取分支父级(分支未归属时为空,
// 此时抑制发布),否则取该 Run 自身的 sessionID。
// 此时抑制发布),否则取该 Run 自身的 session 加上它所属的**路**
// (真并行路的标识由网关经 WithParallelLeg 盖上;单路时为会话本身)。
func flowOwner(ctx context.Context, sessionID string) string {
if parent, isBranch := flowBranch(ctx); isBranch {
return parent
}
return sessionID
return FlowOwnerKey(sessionID, ParallelLegFrom(ctx))
}
// FlowOwnerFromContext 返回拥有从 ctx 启动的工作的 flow 图所属会话:
// 当 ctx 运行于子代理分支内时取分支父级,否则
// 取外层 Run 被驱动的 session id(两者都未设置时为空)。
// FlowOwnerFromContext 返回拥有从 ctx 启动的工作的 flow 图**归属桶**
// (`{会话}#{路}`,见 FlowOwnerKey):当 ctx 运行于子代理分支内时取分支父级,
// 否则取外层 Run 被驱动的 session id 加上它所属的路(两者都未设置时为空)。
//
// 派发后台工作的工具 —— 子代理派生工具 —— 把它
// 作为父会话传入,使在另一个子代理内部派生的子代理
// 保持在同一个顶层图的归属之下,而不是归属到一个不拥有
// 任何图的 id。
// 派发后台工作的工具 —— 子代理派生工具 —— 把它作为父归属桶传入,使在另一个
// 子代理内部派生的子代理保持在同一个顶层图的归属之下,而不是归属到一个不拥有
// 任何图的 id;真并行路里派发的子代理也因此落进**那一路**的图。
//
// ⚠️ 它返回的是**归属桶**而不是会话 id。要把子代理的归属会话交给总线载荷
// (`subagent_result` 的 sessionId 必须是真会话 id)时必须用
// FlowSessionFromContext 取会话那一半。
func FlowOwnerFromContext(ctx context.Context) string {
if parent, isBranch := flowBranch(ctx); isBranch {
return parent
if owner, isBranch := flowBranch(ctx); isBranch {
return owner
}
return SessionFromContext(ctx)
session := SessionFromContext(ctx)
if session == "" {
// 没有会话就没有归属:绝不回一个 `#路` 形态的串(那既查不到图,也会让
// 调用方误以为有归属)。分支与子代理因此保持「不发布」的既有语义。
return ""
}
return FlowOwnerKey(session, FlowLaneFromContext(ctx))
}
// FlowSessionFromContext 返回从 ctx 启动的工作的**归属会话 id**(不含路):
// 真并行时同一会话的每一路都是同一个会话 id,总线载荷与日志要的就是它。
func FlowSessionFromContext(ctx context.Context) string {
session, _ := splitFlowOwner(FlowOwnerFromContext(ctx))
return session
}
// FlowLaneFromContext 返回从 ctx 启动的工作所属的**路**标识(design/15 §2.5.9
// 的 turnId);单路 / 未声明并行时为空。分支运行取分支归属桶里的那一半(子代理
// 因此继承派生它的那一路),非分支运行取本 Run 的并发路。
func FlowLaneFromContext(ctx context.Context) string {
if owner, isBranch := flowBranch(ctx); isBranch {
_, lane := splitFlowOwner(owner)
return lane
}
return ParallelLegFrom(ctx)
}
+232
View File
@@ -0,0 +1,232 @@
package loop
import (
"context"
"strings"
"testing"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
)
// 本文件钉住**按路分桶**这条改造的其余口径(核心行为已在 flow_parallel_test.go 用真
// 循环跑过):桶 / 图 id / 帧上 `turnId` 键三者同源,最近一轮的每一路都拿得回来,
// 且新一轮开始后不再把上一轮的路留在内存里。这些用例走发布器的公开面(真调用点
// 传给它的就是这些字符串),不依赖任何数据库。
// laneIDs 返回每一路图的 id 与 turnId 键,供断言用。
func laneIDs(lanes []*trace.FlowGraph) []string {
out := make([]string, 0, len(lanes))
for _, g := range lanes {
out = append(out, g.ID+"@"+g.TurnID)
}
return out
}
// TestFlowPublisher_ParallelLanesGetOwnGraphs 钉住真并行两路各有各的桶:
// 图 id 不同(落盘不撞 key)、都归属同一个会话、都带上自己的路标识(帧上 `turnId`),
// 且最近一轮的两路都能经 LatestLanes 拿回来。
func TestFlowPublisher_ParallelLanesGetOwnGraphs(t *testing.T) {
p := NewFlowPublisher(nil, func(string) bool { return true })
p.StartRound(FlowOwnerKey("web", "turn-a"), "标记A")
p.SpanStarted(FlowOwnerKey("web", "turn-a"), spanNamed("span-a", "tool.probe_a"))
p.SpanFinished(FlowOwnerKey("web", "turn-a"), finishedSpan("span-a", "tool.probe_a"))
p.StartRound(FlowOwnerKey("web", "turn-b"), "标记B")
p.SpanStarted(FlowOwnerKey("web", "turn-b"), spanNamed("span-b", "tool.probe_b"))
p.SpanFinished(FlowOwnerKey("web", "turn-b"), finishedSpan("span-b", "tool.probe_b"))
ga := p.LastGraph("web")
if ga == nil {
t.Fatal("LastGraph = nil,但该会话刚起了两路")
}
// 最近一路 = 后起的那一路;它的图里只有自己的工具。
if ga.TurnID != "turn-b" || !hasNode(ga, "probe_b") || hasNode(ga, "probe_a") {
t.Fatalf("LastGraph 不是 B 那一路的图:id=%s turnId=%q nodes=%s", ga.ID, ga.TurnID, describeNodes(ga))
}
lanes := p.LatestLanes("web")
if len(lanes) != 2 {
t.Fatalf("最近一轮的路数 = %d,want 2(两路各自一张图):%v", len(lanes), laneIDs(lanes))
}
if got := laneIDs(lanes); got[0] != "web-1#turn-a@turn-a" || got[1] != "web-2#turn-b@turn-b" {
t.Fatalf("两路的 id@turnId = %v,want [web-1#turn-a@turn-a web-2#turn-b@turn-b]", got)
}
if lanes[0].ID == lanes[1].ID {
t.Fatalf("两路撞同一个图 id %q(落盘会互相顶掉)", lanes[0].ID)
}
for _, g := range lanes {
if g.SessionID != "web" {
t.Fatalf("图 %s 的 sessionId = %q,want web(路标识绝不改变会话归属)", g.ID, g.SessionID)
}
if g.Round != lanes[0].Round {
t.Fatalf("同一轮的两路轮次号不同:%d vs %d", g.Round, lanes[0].Round)
}
}
// 收尾后仍是最近一轮的每一路(页面在这之后才打开也看得到两条)。
p.EndRound(FlowOwnerKey("web", "turn-a"), "答复A")
p.EndRound(FlowOwnerKey("web", "turn-b"), "答复B")
if got := p.LatestLanes("web"); len(got) != 2 {
t.Fatalf("收轮后最近一轮的路数 = %d,want 2", len(got))
}
}
// TestFlowPublisher_NewRoundDropsEarlierLanes 钉住轮次推进与内存边界:新一轮只在
// 上一轮的路全部收尾之后开始(该轮的路因此共享轮次号),并丢掉更早轮次的桶 ——
// 否则每一条真并行的路(turnId 每回合全新、桶键永不复用)都会在内存里永久留一张图。
func TestFlowPublisher_NewRoundDropsEarlierLanes(t *testing.T) {
p := NewFlowPublisher(nil, func(string) bool { return true })
p.StartRound(FlowOwnerKey("web", "turn-a"), "标记A")
p.StartRound(FlowOwnerKey("web", "turn-b"), "标记B")
round := p.LatestLanes("web")[0].Round
p.EndRound(FlowOwnerKey("web", "turn-a"), "答复A")
p.EndRound(FlowOwnerKey("web", "turn-b"), "答复B")
// 上一轮尚未被新轮次顶掉时,两条路都还在。
if got := p.LatestLanes("web"); len(got) != 2 {
t.Fatalf("上一轮的路数 = %d,want 2(收轮后仍可补拉)", len(got))
}
p.StartRound("web", "单路的下一轮")
next := p.LatestLanes("web")
if len(next) != 1 || next[0].TurnID != "" {
t.Fatalf("新一轮的路 = %v,want 恰好一条无路标识的图(上一轮的桶必须被丢掉)", laneIDs(next))
}
if next[0].Round <= round {
t.Fatalf("新一轮轮次号 = %d,want > %d", next[0].Round, round)
}
if g := p.LastGraph("web"); g == nil || g.TurnID != "" {
t.Fatalf("新一轮的 LastGraph = %+v,want 单路那张(上一轮的路已不在内存)", g)
}
}
// TestFlowPublisher_RoundsShareRoundNumberButNeverID 钉住「轮次号」与「图 id 的序号」
// 是两个量:同一轮里可以有**不止一条路**(真并行的那一路 + 队首续跑的串行路),它们
// 共享轮次号(客户端据此把同一轮的路并排画),但**各有各的图 id** —— 否则落盘的
// `(session_id, turn_id)` 唯一键会把同一会话的两轮 upsert 成一行,前一轮整张图消失。
func TestFlowPublisher_RoundsShareRoundNumberButNeverID(t *testing.T) {
rec := &fakeRecorder{}
p := NewFlowPublisher(nil, func(string) bool { return true }, WithRoundRecorder(rec))
// 一条并行路先起(占用本轮),随后两条串行路在它仍飞着的时候先后跑完。
p.StartRound(FlowOwnerKey("web", "turn-a"), "并行的一路")
p.StartRound("web", "串行第一轮")
p.EndRound("web", "答复1")
p.StartRound("web", "串行第二轮")
p.EndRound("web", "答复2")
lines := recordedRounds(rec)
if len(lines) != 2 {
t.Fatalf("落盘 %d 条,want 2(两条串行路各一条)", len(lines))
}
if lines[0].ID == lines[1].ID {
t.Fatalf("同一轮的两条串行路撞同一个 id %q ⇒ 落盘会把前一轮整行顶掉", lines[0].ID)
}
if lines[0].Round != lines[1].Round {
t.Fatalf("同一轮的两条路轮次号不同:%d vs %d", lines[0].Round, lines[1].Round)
}
}
// TestFlowPublisher_SingleLegUnchanged 钉住**单路逐字不变**:归属桶就是会话 id,
// 图 id 仍是 `{会话}-{轮次}`(没有路后缀),帧上 `turnId` 键缺席。既有回归
// (flow_snapshot_test.go 的 web-1/web-2)与本条是同一口径的两侧。
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)
}
}
// TestFlowOwnershipHelpers 钉住 loop 与 orchestration 之间的归属口径:
// 真并行路由 `{会话}#{路}` 归属、真会话 id 与路标识各自可单独取出(总线载荷要的是
// 会话、图归属要的是桶),子代理分支继承派生它的那一路。
func TestFlowOwnershipHelpers(t *testing.T) {
// 与 Run 同形:会话先经 withLoopSession 播种,网关再加路标识(WithParallelLeg)。
ctx := withLoopSession(WithParallelLeg(WithTurnID(context.Background(), "turn-a"), "turn-a"), "web")
if got := flowOwner(ctx, "web"); got != "web#turn-a" {
t.Fatalf("并行路的归属桶 = %q,want web#turn-a", got)
}
if got := FlowOwnerFromContext(ctx); got != "web#turn-a" {
t.Fatalf("FlowOwnerFromContext = %q,want web#turn-a", got)
}
if got := FlowSessionFromContext(ctx); got != "web" {
t.Fatalf("FlowSessionFromContext = %q,want web(总线载荷要真会话 id)", got)
}
if got := FlowLaneFromContext(ctx); got != "turn-a" {
t.Fatalf("FlowLaneFromContext = %q,want turn-a", got)
}
// 子代理(分支 Run):归属继承分支桶,会话与路各自拆得回来。
branch := WithFlowBranch(context.Background(), FlowOwnerFromContext(ctx))
if got := flowOwner(branch, "sub-1"); got != "web#turn-a" {
t.Fatalf("子代理 span 的归属桶 = %q,want web#turn-a(必须落进派生它的那一路)", got)
}
if got := FlowSessionFromContext(branch); got != "web" {
t.Fatalf("子代理的 FlowSessionFromContext = %q,want web", got)
}
if got := FlowLaneFromContext(branch); got != "turn-a" {
t.Fatalf("子代理的 FlowLaneFromContext = %q,want turn-a", got)
}
// 单路:桶、会话、路三者不出意外的形态(路为空)。
single := withLoopSession(context.Background(), "web")
if got := flowOwner(single, "web"); got != "web" {
t.Fatalf("单路归属桶 = %q,want web", got)
}
if got := FlowOwnerFromContext(single); got != "web" {
t.Fatalf("单路 FlowOwnerFromContext = %q,want web", got)
}
if got := FlowLaneFromContext(single); got != "" {
t.Fatalf("单路 FlowLaneFromContext = %q,want 空", got)
}
// 没有会话就没有归属:绝不回 `#路` 这种既查不到图、又骗调用方的串。
if got := FlowOwnerFromContext(WithParallelLeg(context.Background(), "turn-a")); got != "" {
t.Fatalf("无会话时 FlowOwnerFromContext = %q,want 空", got)
}
if got := FlowSessionFromContext(WithParallelLeg(context.Background(), "turn-a")); got != "" {
t.Fatalf("无会话时 FlowSessionFromContext = %q,want 空", got)
}
if _, lane := splitFlowOwner("web"); lane != "" {
t.Fatalf("无 # 的归属串应拆出空路,得到 %q", lane)
}
}
// spanNamed / finishedSpan 造一个刚起 / 刚结束的 span,让图里有真实的节点。
func spanNamed(id, name string) *trace.Span {
return &trace.Span{ID: id, Name: name, Kind: trace.SpanKindTool}
}
func finishedSpan(id, name string) *trace.Span {
sp := spanNamed(id, name)
sp.Finish(trace.SpanStatusOK)
return sp
}
// TestFlowPublisher_LaneFramesCarryLaneKey 钉住**帧载荷**上的 `turnId` 键:
// 客户端据此把两路的帧分成两条泳道(旧客户端忽略未知键仍可用)。
func TestFlowPublisher_LaneFramesCarryLaneKey(t *testing.T) {
var frames []*trace.FlowGraph
p := NewFlowPublisher(func(_ string, payload any) {
if g, ok := payload.(*trace.FlowGraph); ok {
frames = append(frames, g)
}
}, func(string) bool { return true })
p.StartRound(FlowOwnerKey("web", "turn-a"), "标记A")
p.StartRound(FlowOwnerKey("web", "turn-b"), "标记B")
if len(frames) < 2 {
t.Fatalf("发布帧数 = %d,want ≥2", len(frames))
}
for _, g := range frames {
if g.TurnID == "" {
t.Fatalf("并行路的帧缺 turnId 键(客户端无法分成两条泳道):id=%s", g.ID)
}
if !strings.Contains(g.ID, g.TurnID) {
t.Fatalf("图 id %q 里找不到路标识 %q(两者必须同源)", g.ID, g.TurnID)
}
if g.Round == 0 {
t.Fatalf("帧缺轮次号:id=%s", g.ID)
}
}
}
+292
View File
@@ -0,0 +1,292 @@
package loop
import (
"context"
"strings"
"sync"
"testing"
"time"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
)
// 本文件是**真并行两路互不覆盖**的回归(design/15 §2.5.10-1 登记的那条真 bug):
// 同一会话的两条并行路(`chat.parallel`)各自开一轮,`graphs[sessionID]` 只有一张 ⇒
// 后起的那一路把前一路的图整张顶掉、两路的节点互相长在对方的图上、落盘两路撞同一个
// `flow_graphs.turn_id`。
//
// 用例刻意只用**引入本切片之前就存在**的 API 驱动真实路径(真 `Loop.Run` + 网关实际
// 挂的那两条 ctx 通道 `WithTurnID` / `WithParallelLeg`),因此它在修复前能编译、能跑、
// 会红 —— 「改前红」的证据就是它(见 FV2-report §4)。判据:
//
// - 两条路各落一条(`RecordRound` 两次)且**图 id 不同**(同一 turn_id 会被
// `(session_id, turn_id)` 唯一键 upsert 成一行);
// - 以 A 的输入播种的那张图里只有 A 的工具节点,反之亦然;
// - 任何一帧 `graph_update` 都不同时含两条路的工具节点。
// barrierTool 是执行期停在闸门上的工具:它报告「已到执行点」后阻塞,使测试能确定地
// 构造「A 的工具在飞时 B 起步」的交错时序,而不是靠 sleep 赌调度。
type barrierTool struct {
name string
arrived chan struct{}
gate chan struct{}
once sync.Once
}
func newBarrierTool(name string) *barrierTool {
return &barrierTool{name: name, arrived: make(chan struct{}, 1), gate: make(chan struct{})}
}
func (b *barrierTool) Name() string { return b.name }
func (b *barrierTool) Description() string { return "barrier tool " + b.name }
func (b *barrierTool) Parameters() (map[string]any, error) {
return map[string]any{"type": "object", "properties": map[string]any{}}, nil
}
func (b *barrierTool) Execute(ctx context.Context, _ any) (any, error) {
b.once.Do(func() { b.arrived <- struct{}{} })
select {
case <-b.gate:
return b.name + " 完成", nil
case <-ctx.Done():
return nil, ctx.Err()
}
}
// laneAdapter 按「这一路的窗口里是哪个标记文本」分流,先请求该路的工具,工具结果回灌
// 之后给出该路的最终答复。两条路共用一个实例 —— 与生产同形(同一个 runner 跑两路)。
type laneAdapter struct {
tool map[string]string // 标记文本 → 该路请求的工具名
answer map[string]string // 标记文本 → 该路的最终答复
}
func (a *laneAdapter) Name() string { return "lane" }
func (a *laneAdapter) Chat(_ context.Context, req model.Request, _ model.Options) (model.Response, error) {
mark := a.markOf(req)
if mark == "" {
return model.Response{Content: "无法归属的请求"}, nil
}
if lastMessageRole(req.Messages) == model.RoleTool {
return model.Response{Content: a.answer[mark]}, nil
}
return model.Response{Content: `{"tool_calls":[{"name":"` + a.tool[mark] + `","args":{}}]}`}, nil
}
// markOf 找出请求里出现的那一路的标记文本;两路的窗口互不共享(I4),因此最多一命中。
func (a *laneAdapter) markOf(req model.Request) string {
for _, m := range req.Messages {
if m.Role != model.RoleUser {
continue
}
for mark := range a.tool {
if strings.Contains(messageText(m), mark) {
return mark
}
}
}
return ""
}
// ChatStream 是流式那一半:本用例的循环走非流式路径(未挂 OnStream),因此它只把
// 请求委托给 Chat 并原样交出聚合结果 —— 刻意**不**再造第二套脚本,也刻意不发任何
// 增量事件,避免「哪个入口真的跑过」变成模糊事实。
func (a *laneAdapter) ChatStream(ctx context.Context, req model.Request, _ func(model.StreamEvent)) (model.Response, error) {
return a.Chat(ctx, req, model.Options{})
}
// lastMessageRole 返回末条消息的角色(空窗口时为空串)。
func lastMessageRole(msgs []model.Message) model.Role {
if len(msgs) == 0 {
return ""
}
return msgs[len(msgs)-1].Role
}
// parallelLaneFixture 是两路交错跑完之后的全部观测面:落盘的每一路、发布的每一帧、
// 以及两路各自的最终答复(用来确认哪张图属于哪一路)。
type parallelLaneFixture struct {
pub *FlowPublisher
recorder *fakeRecorder
mu sync.Mutex
frames []*trace.FlowGraph
}
// run 让两条平行路交错跑完一轮:A 起轮 → A 的工具在飞 → B 起轮 → B 的工具在飞 →
// 依次放行。返回时两路的 Run 都已返回。
func (f *parallelLaneFixture) run(t *testing.T) {
t.Helper()
toolA, toolB := newBarrierTool("probe_a"), newBarrierTool("probe_b")
ad := &laneAdapter{
tool: map[string]string{"标记A": "probe_a", "标记B": "probe_b"},
answer: map[string]string{"标记A": "答复A", "标记B": "答复B"},
}
l := New(Options{
Agent: newTestAgent(t, ad, dispatcherWith(t, toolA, toolB)),
MaxRounds: 8,
Flow: f.pub,
Tracer: trace.New(nil),
})
ctxA := WithParallelLeg(WithTurnID(context.Background(), "turn-a"), "turn-a")
ctxB := WithParallelLeg(WithTurnID(context.Background(), "turn-b"), "turn-b")
var wg sync.WaitGroup
var resA, resB *Result
var errA, errB error
wg.Add(2)
go func() {
defer wg.Done()
resA, errA = l.Run(ctxA, "web", []model.Message{model.TextMessage(model.RoleUser, "标记A:请跑一次工具")})
}()
// A 的工具确实在飞之后才起 B —— 这才叫「同一会话两路并行」,也正是缺陷现场。
awaitBarrier(t, toolA.arrived, "A 的工具执行点")
go func() {
defer wg.Done()
resB, errB = l.Run(ctxB, "web", []model.Message{model.TextMessage(model.RoleUser, "标记B:请跑一次工具")})
}()
awaitBarrier(t, toolB.arrived, "B 的工具执行点")
close(toolA.gate)
close(toolB.gate)
wg.Wait()
if errA != nil || errB != nil {
t.Fatalf("两路 Run 报错:A=%v B=%v", errA, errB)
}
if resA.Text != "答复A" || resB.Text != "答复B" {
t.Fatalf("两路答复 = %q / %q,want 答复A / 答复B(窗口串了)", resA.Text, resB.Text)
}
}
// awaitBarrier 等一个闸门信号,超时即让测试失败(避免缺陷把它挂死成超时)。
func awaitBarrier(t *testing.T, ch <-chan struct{}, what string) {
t.Helper()
select {
case <-ch:
case <-time.After(20 * time.Second):
t.Fatalf("等待 %s 超时", what)
}
}
// newParallelLaneFixture 装一台接了假落盘出口的发布器(真 bus 不在本用例范围内)。
func newParallelLaneFixture(t *testing.T) *parallelLaneFixture {
t.Helper()
f := &parallelLaneFixture{recorder: &fakeRecorder{}}
f.pub = NewFlowPublisher(func(subject string, payload any) {
if subject != graphUpdateSubject {
return
}
g, ok := payload.(*trace.FlowGraph)
if !ok {
return
}
f.mu.Lock()
f.frames = append(f.frames, g)
f.mu.Unlock()
}, func(string) bool { return true }, WithRoundRecorder(f.recorder))
return f
}
// recorded 返回落盘出口收到的每一路图(副本,调用方无需持锁)。
func (f *parallelLaneFixture) recorded() []*trace.FlowGraph {
return recordedRounds(f.recorder)
}
// recordedRounds 返回假落盘出口收到的每一路图(副本,调用方无需持锁)。
func recordedRounds(rec *fakeRecorder) []*trace.FlowGraph {
rec.mu.Lock()
defer rec.mu.Unlock()
return append([]*trace.FlowGraph(nil), rec.graphs...)
}
// publishedFrames 返回发布过的每一帧(副本)。
func (f *parallelLaneFixture) publishedFrames() []*trace.FlowGraph {
f.mu.Lock()
defer f.mu.Unlock()
return append([]*trace.FlowGraph(nil), f.frames...)
}
// TestFlowPublisher_ParallelLegsDoNotOverwriteEachOther 是本切片的核心用例:
// 两路并行 ⇒ 两桶互不覆盖、各自落一条、任何一帧都不含对方节点。
func TestFlowPublisher_ParallelLegsDoNotOverwriteEachOther(t *testing.T) {
f := newParallelLaneFixture(t)
f.run(t)
recorded := f.recorded()
if len(recorded) != 2 {
t.Fatalf("落盘 %d 条,want 2(每路一条):%s", len(recorded), describeGraphs(recorded))
}
if recorded[0].ID == recorded[1].ID {
t.Fatalf("两路撞同一个 turn_id %q ⇒ flow_graphs 的 (session_id, turn_id) 会把其中一路整行顶掉:%s",
recorded[0].ID, describeGraphs(recorded))
}
graphA := graphWithInput(recorded, "标记A")
graphB := graphWithInput(recorded, "标记B")
if graphA == nil || graphB == nil {
t.Fatalf("两路没有各自的图(A=%v B=%v):%s", graphA != nil, graphB != nil, describeGraphs(recorded))
}
assertLaneIsolated(t, "A", graphA, "probe_a", "probe_b")
assertLaneIsolated(t, "B", graphB, "probe_b", "probe_a")
for i, frame := range f.publishedFrames() {
if hasNode(frame, "probe_a") && hasNode(frame, "probe_b") {
t.Fatalf("第 %d 帧同时含两条路的工具节点(B 的节点长在 A 的图上):%s", i, describeNodes(frame))
}
}
}
// assertLaneIsolated 断言某一路的图里有自己的工具节点、且**没有**对方的工具节点。
func assertLaneIsolated(t *testing.T, lane string, g *trace.FlowGraph, own, other string) {
t.Helper()
if !hasNode(g, own) {
t.Fatalf("%s 路的图里没有自己的工具节点 %q:%s", lane, own, describeNodes(g))
}
if hasNode(g, other) {
t.Fatalf("%s 路的图里出现了另一路的工具节点 %q:%s", lane, other, describeNodes(g))
}
if g.SessionID != "web" {
t.Fatalf("%s 路图的 sessionId = %q,want web(路不能改会话归属)", lane, g.SessionID)
}
}
// graphWithInput 找出输入节点标签含 mark 的那张图(两路各自被用户文本播种)。
func graphWithInput(graphs []*trace.FlowGraph, mark string) *trace.FlowGraph {
for _, g := range graphs {
for _, n := range g.Nodes {
if n.Kind == "input" && strings.Contains(n.Label, mark) {
return g
}
}
}
return nil
}
// hasNode 报告图里有没有标签等于(或前缀为)name 的工具节点。
func hasNode(g *trace.FlowGraph, name string) bool {
for _, n := range g.Nodes {
if n.Kind == "tool" && strings.Contains(n.Label, name) {
return true
}
}
return false
}
// describeGraphs / describeNodes 只打印**结构与标签**,供失败信息定位(测试里没有
// 敏感内容,但仍只打标识)。
func describeGraphs(graphs []*trace.FlowGraph) string {
var b strings.Builder
for _, g := range graphs {
b.WriteString("\n id=" + g.ID + " session=" + g.SessionID)
b.WriteString(" nodes=" + describeNodes(g))
}
return b.String()
}
func describeNodes(g *trace.FlowGraph) string {
parts := make([]string, 0, len(g.Nodes))
for _, n := range g.Nodes {
parts = append(parts, n.Kind+":"+n.Label)
}
return "[" + strings.Join(parts, " ") + "]"
}
+7 -2
View File
@@ -83,9 +83,14 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa
// 分支 Run(子代理)把其 span 并入父会话的实时
// 图;只有所属的 Run 开始与结束轮次。没有它,子代理
// 会重置前端唯一的 flowGraph,并让父级自身的节点出现断档。
//
// 归属取**桶**(`{会话}[#{路}]`,见 FlowOwnerKey):真并行时本 Run 属于某一条路,
// 两路因此各起各的图、各收各的图 —— 后起的那一路绝不重置前一路(design/15
// §2.5.10-1)。单路时桶逐字等于 sessionID,行为与引入本切片之前完全一致。
if _, isBranch := flowBranch(ctx); l.flow != nil && !isBranch {
l.flow.StartRound(sessionID, lastUserText(msgs))
defer func() { l.flow.EndRound(sessionID, res.Text) }()
owner := flowOwner(ctx, sessionID)
l.flow.StartRound(owner, lastUserText(msgs))
defer func() { l.flow.EndRound(owner, res.Text) }()
}
return l.driveTurn(ctx, sessionID, msgs, startStep, res)
}
+1 -1
View File
@@ -98,7 +98,7 @@ func (l *Loop) steerAtBoundary(ctx context.Context, sessionID string, step int,
// 链路图如实标记「本轮输入被中途追加过」(design/15 §5)。子代理 Run 不拥有
// 自己的图(它的 span 并入派生它的会话),因此分支 Run 不打这个标记。
if _, isBranch := flowBranch(ctx); !isBranch {
l.flow.MarkInputSteered(sessionID)
l.flow.MarkInputSteered(flowOwner(ctx, sessionID))
}
l.log.Info("steer injected", "session", sessionID, "step", step, "messages", len(msgs))
return append(steered, msgs...)