AskaEth
4a937bd269
- 新增 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/臆造回退）。
290 lines
11 KiB
Go
290 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）：编排器给**每一路**
|
||
// 都发（普通单路回合、队首续跑路、真并行路都有），因此它就是这张图属于哪一回合的
|
||
// 答案，也是落盘 `flow_graphs.lane_turn_id` 与 `chat_messages.turn_id` 的同值。
|
||
// **缺席**语义：这条路没有权威标识（未接编排器的调用方、后台思考、旧后端）——
|
||
// 客户端据此把图归到会话那一张（兼容口径），读侧据此让该轮的 `toolCalls` 整体
|
||
// 缺席；两侧都**绝不**臆造一个 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）。空串表示这条路
|
||
// 没有权威标识（未接编排器 / 旧调用方），此时图上该键缺席。它有一个可见后果：
|
||
// 图 id 的后缀取它（见 graphID），因此有权威标识的单路图也不再是 `{会话}-{序号}`
|
||
// 而是 `{会话}-{序号}#{回合}` —— 落盘唯一键 `(session_id, turn_id)` 因此跨进程
|
||
// 不撞（序号只在进程内递增、重启归零，而回合标识每回合全新）。
|
||
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）；单路（编排器也给它盖了权威标识）同样带后缀，因此它的
|
||
// 图 id 跨进程不撞（见 RunGraph.turnID）。
|
||
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**；
|
||
// 回合标识只在「这条路有权威标识」时追加（单路在有 ctx 通道时也有 —— 见
|
||
// loop.FlowPublisher.StartRoundWithTurnID）。
|
||
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
|
||
}
|