AskaEth
89434eadef
【取证（症状成立，且比登记描述更糟）】两路真并行时两路最终落盘**同一个 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) 提交里（同一个工作时段内）。
282 lines
11 KiB
Go
282 lines
11 KiB
Go
package trace
|
||
|
||
import (
|
||
"fmt"
|
||
"time"
|
||
)
|
||
|
||
// Flow graph wire types (DevDocs/design/19 §1). The JSON field names are
|
||
// camelCase so the frontend's FlowGraph type and parseGraph read them directly.
|
||
// Node status strings mirror the frontend AgentStatus keys so parseGraph and
|
||
// FlowGraphView pass them through unchanged.
|
||
|
||
// Node status labels kept as named constants so no bare string appears in the
|
||
// mapping code (project rule §10).
|
||
const (
|
||
// FlowStatusIdle is a node that has not started.
|
||
FlowStatusIdle = "idle"
|
||
// FlowStatusRunning is a node whose span is in flight.
|
||
FlowStatusRunning = "running"
|
||
// FlowStatusDone is a successfully completed node.
|
||
FlowStatusDone = "done"
|
||
// FlowStatusError is a node whose span finished with an error.
|
||
FlowStatusError = "error"
|
||
)
|
||
|
||
// FlowStatusForPhase 把工具执行的细粒度 phase（design/04 §1.1，
|
||
// loop 随 tool_execution.end 帧一同记在 tool span 上）映射为流程图
|
||
// 节点状态：phase 取值与本包状态枚举（design/19 §1 统一枚举）
|
||
// 同构（running/done/error）。未知 phase 返回空串，调用方回落到
|
||
// span 终态推导，因此缺 phase 的旧 span 不受影响。
|
||
func FlowStatusForPhase(phase string) string {
|
||
switch phase {
|
||
case FlowStatusRunning:
|
||
return FlowStatusRunning
|
||
case FlowStatusDone:
|
||
return FlowStatusDone
|
||
case FlowStatusError:
|
||
return FlowStatusError
|
||
default:
|
||
return ""
|
||
}
|
||
}
|
||
|
||
// FlowNode is one node in a flow graph. Status/SpanKind mirror the frontier.
|
||
type FlowNode struct {
|
||
ID string `json:"id"`
|
||
Label string `json:"label"`
|
||
Kind string `json:"kind,omitempty"`
|
||
Status string `json:"status,omitempty"`
|
||
StartedAt time.Time `json:"startedAt,omitempty"`
|
||
EndedAt time.Time `json:"endedAt,omitempty"`
|
||
Props map[string]any `json:"props,omitempty"`
|
||
Log []Event `json:"log,omitempty"`
|
||
}
|
||
|
||
// FlowEdge is a directed edge between two nodes.
|
||
type FlowEdge struct {
|
||
From string `json:"from"`
|
||
To string `json:"to"`
|
||
}
|
||
|
||
// FlowGraph is a live, single-round flow graph published as ws.graph_update.
|
||
type FlowGraph struct {
|
||
ID string `json:"id"`
|
||
Nodes []FlowNode `json:"nodes"`
|
||
Edges []FlowEdge `json:"edges"`
|
||
// SessionID 是拥有这张图的会话（帧壳与落盘行的归属坐标）。
|
||
SessionID string `json:"sessionId,omitempty"`
|
||
// TurnID 是这张图所属**路**的权威回合标识（design/15 §2.5.9）：真并行时
|
||
// 同一会话有多条路，每条路各有自己的图与本键。**缺席**语义：这条路没有
|
||
// 权威标识（单路的历史帧、未接编排器的调用方、旧后端）——客户端据此把图
|
||
// 归到会话那一张（兼容口径），绝不臆造一个 id。
|
||
TurnID string `json:"turnId,omitempty"`
|
||
// Round 是这张图所属的**会话轮次号**（每会话单调递增；同一轮的各条路共享它）。
|
||
// 客户端据此只保留最新一轮的泳道，避免旧轮的路留在页面上。
|
||
// **缺席**（0）语义：旧后端/未设置 —— 客户端按「每次帧都算同一轮」的保守口径处理。
|
||
Round int `json:"round,omitempty"`
|
||
StartedAt time.Time `json:"startedAt,omitempty"`
|
||
EndedAt time.Time `json:"endedAt,omitempty"`
|
||
}
|
||
|
||
// NodeSpec is the input to RunGraph.Upsert: the full description of a node plus
|
||
// the parent id used to wire its incoming edge.
|
||
type NodeSpec struct {
|
||
ID string
|
||
Label string
|
||
Kind string
|
||
Status string
|
||
StartedAt time.Time
|
||
EndedAt time.Time
|
||
Props map[string]any
|
||
Log []Event
|
||
ParentID string
|
||
}
|
||
|
||
// RunGraph accumulates the spans of one round into a FlowGraph. It is a pure
|
||
// structure: it knows nothing about the bus, sessions or the loop, so it is
|
||
// unit-testable in isolation. A caller serializes access (see loop.FlowPublisher).
|
||
type RunGraph struct {
|
||
session string
|
||
round int
|
||
// seq 是本图的**会话级建图序号**（调用方每次建图递增、永不复用）。图 id 的第二
|
||
// 分量取它而不是轮次号：真并行时同一轮里的第二条路必须也拿到全新的 id，否则两路
|
||
// 撞同一个 `flow_graphs.turn_id`（唯一键），后收尾的那一路把前一路整行顶掉。
|
||
// 未经 SetSeq 建图的调用方（图构建器单测）保留 `{会话}-{轮次}` 的老形状。
|
||
seq int
|
||
// turnID 是本图所属**路**的权威回合标识（design/15 §2.5.9）。空串表示这条路
|
||
// 没有权威标识（单路 / 未接编排器 / 旧调用方），此时图上该键缺席。
|
||
turnID string
|
||
nodes map[string]*FlowNode
|
||
order []string
|
||
edges map[string]bool
|
||
edgeOrder []FlowEdge
|
||
// parents remembers every span's parent link, including spans that are not
|
||
// rendered as nodes (a state_transition step container), so a node's lane can
|
||
// be resolved by walking through the skipped containers.
|
||
parents map[string]string
|
||
// laneLast is the per-lane cursor used to chain sequential steps: keyed by the
|
||
// enclosing subagent node id, and by "" for the main trunk. One cursor per lane
|
||
// (instead of one per graph) is what keeps two parallel sub-agent lanes from
|
||
// cross-linking when their spans interleave.
|
||
laneLast map[string]string
|
||
started time.Time
|
||
ended time.Time
|
||
}
|
||
|
||
// NewRunGraph allocates an empty graph for one session round.
|
||
func NewRunGraph(session string) *RunGraph {
|
||
return &RunGraph{
|
||
session: session,
|
||
nodes: make(map[string]*FlowNode),
|
||
edges: make(map[string]bool),
|
||
parents: make(map[string]string),
|
||
laneLast: make(map[string]string),
|
||
}
|
||
}
|
||
|
||
// Start marks the round began at now and resets the node/edge sets so a reused
|
||
// graph starts fresh for a new round. 路口标识不在重置之列：它是**本次建图**的
|
||
// 身份，由 SetTurnID 在 Start 之后立刻盖上（见 loop.FlowPublisher）。
|
||
func (g *RunGraph) Start(round int, now time.Time) {
|
||
g.round = round
|
||
g.started = now
|
||
g.ended = time.Time{}
|
||
g.nodes = make(map[string]*FlowNode)
|
||
g.order = nil
|
||
g.edges = make(map[string]bool)
|
||
g.edgeOrder = nil
|
||
g.parents = make(map[string]string)
|
||
g.laneLast = make(map[string]string)
|
||
}
|
||
|
||
// SetTurnID 盖上这张图所属**路**的权威回合标识（design/15 §2.5.9）。空串是
|
||
// 合法入参（单路 / 未接编排器），此时快照里该键缺席（omitempty），客户端回落
|
||
// 到「按会话归一张图」的兼容口径。
|
||
func (g *RunGraph) SetTurnID(turnID string) { g.turnID = turnID }
|
||
|
||
// SetSeq 盖上这张图的**会话级建图序号**（见 RunGraph.seq）。未设置（0）时图 id
|
||
// 退回 `{会话}-{轮次}` 的老形状，因此只调 Start 的既有调用方/单测逐字不变。
|
||
func (g *RunGraph) SetSeq(seq int) { g.seq = seq }
|
||
|
||
// SeedInput creates the "input" node for the round. It returns the node id so a
|
||
// caller can reference it. The label is redacted (design/19 §3), so a message that
|
||
// happens to carry a credential cannot publish one.
|
||
func (g *RunGraph) SeedInput(text string) string {
|
||
g.Upsert(NodeSpec{
|
||
ID: "input",
|
||
Label: RedactText(text),
|
||
Kind: "input",
|
||
Status: FlowStatusDone,
|
||
StartedAt: g.started,
|
||
})
|
||
return "input"
|
||
}
|
||
|
||
// SeedOutput creates the "output" node for the round, wired to the last node so
|
||
// the trunk ends at the answer. The label is redacted like SeedInput's.
|
||
func (g *RunGraph) SeedOutput(text string) string {
|
||
g.ended = time.Now()
|
||
g.Upsert(NodeSpec{
|
||
ID: "output",
|
||
Label: RedactText(text),
|
||
Kind: "output",
|
||
Status: FlowStatusDone,
|
||
StartedAt: g.ended,
|
||
EndedAt: g.ended,
|
||
})
|
||
return "output"
|
||
}
|
||
|
||
// laneIDSep 是图 id 里「会话-轮次」与「路标识」之间的分隔符。它与 `loop` 的
|
||
// flow 归属串（`{会话}#{路}`）以及 design/15 §2.5.9 的 checkpoint 归属键
|
||
// （`{sessionID}#{turnID}`）取同一个字符：**一条路**在流程图、checkpoint 与
|
||
// 落盘行三处因此是同一个拼接口径（loop 侧的同名常量见 loop/flow_branch.go）。
|
||
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）。
|
||
func (g *RunGraph) Snapshot() *FlowGraph {
|
||
nodes := make([]FlowNode, 0, len(g.order))
|
||
for _, id := range g.order {
|
||
nodes = append(nodes, *g.nodes[id])
|
||
}
|
||
return &FlowGraph{
|
||
ID: g.graphID(),
|
||
Nodes: nodes,
|
||
Edges: append([]FlowEdge(nil), g.edgeOrder...),
|
||
SessionID: g.session,
|
||
TurnID: g.turnID,
|
||
Round: g.round,
|
||
StartedAt: g.started,
|
||
EndedAt: g.ended,
|
||
}
|
||
}
|
||
|
||
// graphID 拼出这张图的 id：`{会话}-{序号}[#{路标识}]`。序号取会话级建图序号
|
||
// （未设置时退回轮次号，见 RunGraph.seq），因此**同一轮里的每一条路也各有各的 id**；
|
||
// 路标识只在「这条路有权威标识」时追加。
|
||
func (g *RunGraph) graphID() string {
|
||
n := g.seq
|
||
if n == 0 {
|
||
n = g.round
|
||
}
|
||
base := fmt.Sprintf("%s-%d", g.session, n)
|
||
if g.turnID == "" {
|
||
return base
|
||
}
|
||
return base + laneIDSep + g.turnID
|
||
}
|
||
|
||
// Clone 深拷贝一份图，供**锁外**的读者使用（契约 v0.71 的
|
||
// `GET /api/v1/admin/flow/graph` 就近取材于 `loop.FlowPublisher.LastGraph`）。
|
||
//
|
||
// 为什么必须深拷贝：`Snapshot` 只复制节点结构体，`Props` 映射与 `Log` 切片
|
||
// 仍与活图**共享底层存储**；而 `RunGraph.Upsert` 对同一个节点是**就地**合并
|
||
// （`merge` 往同一个 map 写键、往同一个切片追加）。发布路径在持有发布器
|
||
// 互斥锁时就把图编码掉了，因此共享无碍；但只读查询把图交出去之后还要在
|
||
// 锁外编码，与下一帧 span 的合并并发 ⇒ 会读到撕裂的 map（Go 里并发读写
|
||
// map 直接是致命错误）。深拷贝把这份共享切断，且只发生在人为的只读查询上，
|
||
// 不在实时发布路径上，因此不影响每帧开销。
|
||
func (g *FlowGraph) Clone() *FlowGraph {
|
||
if g == nil {
|
||
return nil
|
||
}
|
||
out := &FlowGraph{
|
||
ID: g.ID,
|
||
Nodes: make([]FlowNode, 0, len(g.Nodes)),
|
||
Edges: append([]FlowEdge(nil), g.Edges...),
|
||
SessionID: g.SessionID,
|
||
TurnID: g.TurnID,
|
||
Round: g.Round,
|
||
StartedAt: g.StartedAt,
|
||
EndedAt: g.EndedAt,
|
||
}
|
||
for _, n := range g.Nodes {
|
||
cp := n
|
||
if n.Props != nil {
|
||
cp.Props = make(map[string]any, len(n.Props))
|
||
for k, v := range n.Props {
|
||
cp.Props[k] = v
|
||
}
|
||
}
|
||
if n.Log != nil {
|
||
cp.Log = make([]Event, len(n.Log))
|
||
for i, e := range n.Log {
|
||
ev := e
|
||
if e.Props != nil {
|
||
ev.Props = make(map[string]any, len(e.Props))
|
||
for k, v := range e.Props {
|
||
ev.Props[k] = v
|
||
}
|
||
}
|
||
cp.Log[i] = ev
|
||
}
|
||
}
|
||
out.Nodes = append(out.Nodes, cp)
|
||
}
|
||
return out
|
||
}
|