feat(flow): 实时全链路流程可视化 P1——trace→graph_update→前端全链路流程页(design/19)
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled

后端: trace/graph.go(RunGraph 图构建器,kind/status 映射,父子优先连线,脱敏白名单+截断);
       loop/flow.go(FlowPublisher 按会话增量发布 ws.graph_update,仅主人web会话); loop span
       start/finish 折入图; run.go StartRound/EndRound; wire.go 组装+注入.
前端: Flow.tsx(全链路流程页:FlowGraphView+会话/节点/开始/耗时卡片+状态计数+节点点开日志);
       FlowGraph 组件加 onNodeClick; types/wsParse 扩展; router/sections/AdminApp 注册 flow;
       api-client.md §3.5 graph_update 契约.
验证: 后端 build/vet/test 全绿+flow_test.go 集成(span→graph→ws.graph_update主干); 前端 build 绿.
红线: 单二进制零新依赖; 工具参数/结果脱敏; 仅主人web会话; 身份靠actor元数据; 只读不虚构.
P2(并行子代理分支/泳道/节点实时日志)后置.
This commit is contained in:
Aska committed 2026-09-07 19:10:57 +08:00
1 parent 7a81071ec8
commit 44e7a5c2c5
18 files changed
+1126 -11

No files matched your search

+143
View File
@@ -0,0 +1,143 @@
package loop
import (
"sync"
"time"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
)
// graphUpdateSubject is the bus subject the FlowPublisher publishes each round's
// live flow graph on. The gateway hub subscribes to ws.> and relays it as the
// WS event type graph_update (design/19 §1).
const graphUpdateSubject = "ws.graph_update"
// allowedSessions is the predicate signature used to gate which sessions publish
// flow graphs (the owner's web sessions), so IM/stranger/sub-agent turns never
// leak into the admin panel. A nil predicate allows every session.
type allowedSessions func(sessionID string) bool
// FlowPublisher accumulates a per-session flow graph from the loop's trace spans
// and publishes ws.graph_update as each span starts/finishes, giving the frontend
// a realtime input→think→tool→output graph (design/19 §1). It is nil-tolerant
// (a nil publisher is a no-op) and safe for concurrent use; session graphs are
// serialized under one mutex because parallel tool spans finish concurrently.
type FlowPublisher struct {
mu sync.Mutex
publish func(subject string, payload any)
allow allowedSessions
graphs map[string]*trace.RunGraph
rounds map[string]int
}
// NewFlowPublisher builds a FlowPublisher. publish, when non-nil, is called with
// graphUpdateSubject and a *trace.FlowGraph snapshot on every span transition.
// allow, when non-nil, gates which session ids are published (returning false
// drops the session entirely). Both may be nil (a nil publish is a no-op).
func NewFlowPublisher(publish func(subject string, payload any), allow func(sessionID string) bool) *FlowPublisher {
return &FlowPublisher{
publish: publish,
allow: allow,
graphs: make(map[string]*trace.RunGraph),
rounds: make(map[string]int),
}
}
// StartRound begins a fresh graph for sessionID, seeding an input node from the
// user text and publishing the initial frame. It resets the session's previous
// round, so each user turn is its own graph.
func (p *FlowPublisher) StartRound(sessionID, inputText string) {
if sessionID == "" || !p.allowed(sessionID) {
return
}
p.mu.Lock()
defer p.mu.Unlock()
round := p.rounds[sessionID] + 1
g := trace.NewRunGraph(sessionID)
g.Start(round, time.Now())
g.SeedInput(inputText)
p.graphs[sessionID] = g
p.rounds[sessionID] = round
p.publishLocked(g)
}
// EndRound finalizes the round by seeding an output node from the reply and
// publishing the final frame. It keeps the graph around so a client that misses
// the final frame can replay the last snapshot on refresh.
func (p *FlowPublisher) EndRound(sessionID, outputText string) {
if sessionID == "" || !p.allowed(sessionID) {
return
}
p.mu.Lock()
defer p.mu.Unlock()
g := p.graphs[sessionID]
if g == nil {
return
}
g.SeedOutput(outputText)
p.publishLocked(g)
}
// SpanStarted records a running placeholder node for a span that just began.
// State-transition spans are omitted by the graph builder.
func (p *FlowPublisher) SpanStarted(sessionID string, sp *trace.Span) {
if sp == nil || sessionID == "" || !p.allowed(sessionID) {
return
}
p.mu.Lock()
defer p.mu.Unlock()
g := p.graphs[sessionID]
if g == nil {
return
}
if g.AddSpanStart(sp.ID, sp.Name, sp.Kind, sp.ParentID, sp.StartedAt) {
p.publishLocked(g)
}
}
// SpanFinished folds a just-finished span into the session graph as a final node
// (status done/error) and publishes. It must be called after the span's Finish.
func (p *FlowPublisher) SpanFinished(sessionID string, sp *trace.Span) {
if sp == nil || sessionID == "" || !p.allowed(sessionID) {
return
}
p.mu.Lock()
defer p.mu.Unlock()
g := p.graphs[sessionID]
if g == nil {
return
}
g.AddSpan(sp.Snapshot())
p.publishLocked(g)
}
// publishLocked publishes a snapshot of g. Callers must hold p.mu.
func (p *FlowPublisher) publishLocked(g *trace.RunGraph) {
if p.publish == nil {
return
}
p.publish(graphUpdateSubject, g.Snapshot())
}
// allowed reports whether a session's graphs should be published.
func (p *FlowPublisher) allowed(sessionID string) bool {
if p.allow == nil {
return true
}
return p.allow(sessionID)
}
// lastUserText returns the text of the newest user message in messages, used to
// label the round's input node. It returns "" when no user text is present.
func lastUserText(messages []model.Message) string {
for i := len(messages) - 1; i >= 0; i-- {
if messages[i].Role != model.RoleUser {
continue
}
if t := messageText(messages[i]); t != "" {
return t
}
}
return ""
}
+118
View File
@@ -0,0 +1,118 @@
package loop
import (
"context"
"sync"
"testing"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
)
// TestRun_PublishesFlowGraph_Trunk wires a FlowPublisher with a capturing publish
// function and a live tracer, runs one tool round and asserts ws.graph_update was
// published with the input→llm→tool→output trunk (design/19 §1 P1).
func TestRun_PublishesFlowGraph_Trunk(t *testing.T) {
var mu sync.Mutex
var seen []*trace.FlowGraph
flow := NewFlowPublisher(func(subject string, payload any) {
if subject != graphUpdateSubject {
return
}
if g, ok := payload.(*trace.FlowGraph); ok {
mu.Lock()
seen = append(seen, g)
mu.Unlock()
}
}, func(string) bool { return true })
call := `{"tool_calls":[{"name":"echo","args":{"q":"hi"}}]}`
ad := &mockAdapter{chat: []mgResp{
{resp: model.Response{Content: call}},
{resp: model.Response{Content: "Echoed!"}},
}}
td := dispatcherWith(t, &mockTool{name: "echo", result: "hi"})
a := newTestAgent(t, ad, td)
l := New(Options{Agent: a, MaxRounds: 8, Flow: flow, Tracer: trace.New(nil)})
res, err := l.Run(context.Background(), "web", []model.Message{model.TextMessage(model.RoleUser, "echo this")})
if err != nil {
t.Fatalf("Run: %v", err)
}
if res.Text != "Echoed!" {
t.Fatalf("Text = %q, want Echoed!", res.Text)
}
mu.Lock()
defer mu.Unlock()
if len(seen) == 0 {
t.Fatal("no ws.graph_update published")
}
last := seen[len(seen)-1]
if last.SessionID != "web" {
t.Fatalf("SessionID = %q, want web", last.SessionID)
}
kinds := make(map[string]string)
for _, n := range last.Nodes {
kinds[n.ID] = n.Kind
}
nodeKind := map[string]bool{"input": false, "llm": false, "tool": false, "output": false}
inputID, outputID := "", ""
for _, n := range last.Nodes {
if n.Kind == "input" {
inputID = n.ID
}
if n.Kind == "output" {
outputID = n.ID
}
if _, ok := nodeKind[n.Kind]; ok {
nodeKind[n.Kind] = true
}
}
for _, want := range []string{"input", "llm", "tool", "output"} {
if !nodeKind[want] {
t.Fatalf("trunk incomplete: no %q node; nodes=%+v", want, last.Nodes)
}
}
// Walk the chain from the input node; it must reach output through llm/tool.
chain := []string{}
cur := inputID
for cur != "" {
chain = append(chain, kinds[cur])
if cur == outputID {
break
}
cur = nextNode(last, cur)
}
hasLLM, hasTool := false, false
for _, k := range chain {
if k == "llm" {
hasLLM = true
}
if k == "tool" {
hasTool = true
}
}
if chain[len(chain)-1] != "output" || !hasLLM || !hasTool {
t.Fatalf("trunk chain broken: %v", chain)
}
// The final frame should show the tool completed (done), not still running.
for _, n := range last.Nodes {
if n.Kind == "tool" && n.Status != trace.FlowStatusDone {
t.Fatalf("tool status = %q, want done", n.Status)
}
}
}
// nextNode returns the node that cur points to, or "" when cur has no outgoing
// edge (the terminal output node). A linear chain has at most one successor.
func nextNode(g *trace.FlowGraph, cur string) string {
for _, e := range g.Edges {
if e.From == cur {
return e.To
}
}
return ""
}
+2
View File
@@ -90,6 +90,7 @@ type Loop struct {
extractor memory.Extractor
checkpoint checkpointStore
tracer *trace.Tracer
flow *FlowPublisher
guards *guardrail.Pipeline
personaID string
timeLocation *time.Location
@@ -141,6 +142,7 @@ func New(opts Options) *Loop {
extractor: opts.Extractor,
checkpoint: opts.Checkpoint,
tracer: opts.Tracer,
flow: opts.Flow,
guards: opts.Guards,
personaID: opts.PersonaID,
timeLocation: timeLocation,
+4
View File
@@ -60,6 +60,10 @@ type Options struct {
Checkpoint checkpointStore
// Tracer emits a structured span for every step, LLM call and tool call.
Tracer *trace.Tracer
// Flow, when non-nil, accumulates the run's trace spans into a per-session
// flow graph and publishes ws.graph_update on each span transition (design/19
// §1). Nil disables the realtime flow visualization.
Flow *FlowPublisher
// Guards validates request/response payloads on the streaming path, where
// RunStep is bypassed. On the non-streaming path the agent runs its own
// guardrails inside RunStep.
+4
View File
@@ -49,6 +49,10 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa
msgs = l.resolveImages(ctx, msgs)
toolState := &toolLoopState{prevFP: make(map[string]bool)}
res := &Result{}
if l.flow != nil {
l.flow.StartRound(sessionID, lastUserText(msgs))
defer func() { l.flow.EndRound(sessionID, res.Text) }()
}
for step := startStep; step <= l.maxRounds; step++ {
if err := ctx.Err(); err != nil {
return nil, fmt.Errorf("loop: run: step %d: %w", step, err)
+19 -3
View File
@@ -8,20 +8,36 @@ import (
// span starts a named span of the given kind when a Tracer is configured, else
// returns a nil no-op span. The returned context carries the span as the current
// one so nested spans chain as children.
// one so nested spans chain as children. When a FlowPublisher is configured the
// span's session (threaded via ctx) is stamped as a prop and a running placeholder
// node is published immediately, so the flow graph shows work in flight.
func (l *Loop) span(ctx context.Context, name string, kind trace.SpanKind) (context.Context, *trace.Span) {
if l.tracer == nil {
return ctx, nil
}
return l.tracer.StartSpan(ctx, name, kind)
sctx, sp := l.tracer.StartSpan(ctx, name, kind)
if sid := loopSession(ctx); sid != "" {
_ = sp.SetProp("session", sid)
if l.flow != nil {
l.flow.SpanStarted(sid, sp)
}
}
return sctx, sp
}
// finishSpan closes a span with the given status, ignoring a nil span.
// finishSpan closes a span with the given status, ignoring a nil span. After the
// span closes (immutable) the finished record is folded into the flow graph and
// published, flipping the node to its final status.
func (l *Loop) finishSpan(sp *trace.Span, status trace.SpanStatus) {
if sp == nil {
return
}
sp.Finish(status)
if l.flow != nil {
if sid, ok := sp.Props["session"].(string); ok && sid != "" {
l.flow.SpanFinished(sid, sp)
}
}
}
// spanProp attaches a structured property to a span, ignoring a nil span and any
+395
View File
@@ -0,0 +1,395 @@
package trace
import (
"encoding/json"
"fmt"
"strings"
"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"
)
// maxSummaryRunes bounds how many runes of a user string / tool result are
// rendered into a node label or prop summary, so sensitive or long content is
// collapsed to a safe, dashboard-friendly prefix.
const maxSummaryRunes = 80
// 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 string `json:"sessionId,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
nodes map[string]*FlowNode
order []string
edges map[string]bool
edgeOrder []FlowEdge
last 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),
}
}
// Start marks the round began at now and resets the node/edge sets so a reused
// graph starts fresh for a new round.
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.last = ""
}
// Upsert creates or merges a node and wires its incoming edge. The edge source
// is the node's ParentID when that parent is already a node, otherwise the most
// recently added node — so a round assembles as a left-to-right trunk even when
// the parent container (a state_transition span) is omitted from the graph.
func (g *RunGraph) Upsert(spec NodeSpec) {
if spec.ID == "" {
return
}
if existing := g.nodes[spec.ID]; existing != nil {
g.merge(existing, spec)
return
}
n := &FlowNode{
ID: spec.ID,
Label: spec.Label,
Kind: spec.Kind,
Status: spec.Status,
StartedAt: spec.StartedAt,
EndedAt: spec.EndedAt,
}
if spec.Props != nil {
n.Props = spec.Props
}
n.Log = append(n.Log, spec.Log...)
g.nodes[spec.ID] = n
g.order = append(g.order, spec.ID)
from := g.last
if spec.ParentID != "" {
if _, ok := g.nodes[spec.ParentID]; ok {
from = spec.ParentID
}
}
if from != "" && from != spec.ID {
g.addEdge(from, spec.ID)
}
g.last = spec.ID
}
// merge updates an existing node in place (a running placeholder being finalized
// by its finished span), keeping label/kind and appending props/log.
func (g *RunGraph) merge(existing *FlowNode, spec NodeSpec) {
if spec.Label != "" {
existing.Label = spec.Label
}
if spec.Kind != "" {
existing.Kind = spec.Kind
}
if spec.Status != "" {
existing.Status = spec.Status
}
if !spec.EndedAt.IsZero() {
existing.EndedAt = spec.EndedAt
}
if len(spec.Props) > 0 {
if existing.Props == nil {
existing.Props = make(map[string]any, len(spec.Props))
}
for k, v := range spec.Props {
existing.Props[k] = v
}
}
existing.Log = append(existing.Log, spec.Log...)
}
// addEdge records a directed edge once.
func (g *RunGraph) addEdge(from, to string) {
key := from + "\x00" + to
if g.edges[key] {
return
}
g.edges[key] = true
g.edgeOrder = append(g.edgeOrder, FlowEdge{From: from, To: to})
}
// SeedInput creates the "input" node for the round. It returns the node id so a
// caller can reference it.
func (g *RunGraph) SeedInput(text string) string {
g.Upsert(NodeSpec{
ID: "input",
Label: truncate(text, maxSummaryRunes),
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.
func (g *RunGraph) SeedOutput(text string) string {
g.ended = time.Now()
g.Upsert(NodeSpec{
ID: "output",
Label: truncate(text, maxSummaryRunes),
Kind: "output",
Status: FlowStatusDone,
StartedAt: g.ended,
EndedAt: g.ended,
})
return "output"
}
// AddSpan folds a finished span record into the graph as a node with its final
// status. State-transition spans are the step container and are omitted.
func (g *RunGraph) AddSpan(rec SpanRecord) {
kind := kindForSpan(rec.Kind)
if kind == "" {
return
}
g.Upsert(NodeSpec{
ID: rec.ID,
Label: labelForSpan(rec),
Kind: kind,
Status: statusForFinished(rec.Status),
StartedAt: rec.StartedAt,
EndedAt: rec.EndedAt,
Props: safeSpanProps(rec),
Log: rec.Events,
ParentID: rec.ParentID,
})
}
// AddSpanStart registers a running placeholder for a span that just began, so a
// node appears as "running" before it finishes. State-transition spans are
// omitted, matching AddSpan. It reports whether a node was created.
func (g *RunGraph) AddSpanStart(id, name string, kind SpanKind, parent string, at time.Time) bool {
nodeKind := kindForSpan(kind)
if nodeKind == "" {
return false
}
g.Upsert(NodeSpec{
ID: id,
Label: labelForStart(name, kind),
Kind: nodeKind,
Status: FlowStatusRunning,
StartedAt: at,
ParentID: parent,
})
return true
}
// Snapshot returns a copy of the graph safe to encode and publish.
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: fmt.Sprintf("%s-%d", g.session, g.round),
Nodes: nodes,
Edges: append([]FlowEdge(nil), g.edgeOrder...),
SessionID: g.session,
StartedAt: g.started,
EndedAt: g.ended,
}
}
// kindForSpan maps a span kind to a flow node kind. state_transition is the step
// container and maps to "" so it is never rendered as a node.
func kindForSpan(k SpanKind) string {
switch k {
case SpanKindLLM:
return "llm"
case SpanKindTool:
return "tool"
case SpanKindSubagent:
return "subagent"
case SpanKindMemory:
return "memory"
case SpanKindGuardrail:
return "guardrail"
case SpanKindOrchestration:
return "orchestration"
default:
return ""
}
}
// labelForSpan renders a node label from a finished span.
func labelForSpan(rec SpanRecord) string {
if kind := kindForSpan(rec.Kind); kind == "tool" {
return strings.TrimPrefix(rec.Name, "tool.")
}
if rec.Kind == SpanKindLLM {
return "思考"
}
if rec.Name != "" {
return rec.Name
}
return string(rec.Kind)
}
// labelForStart renders a node label from a just-started span.
func labelForStart(name string, kind SpanKind) string {
if kind == SpanKindLLM {
return "思考"
}
if kind == SpanKindTool {
return strings.TrimPrefix(name, "tool.")
}
if name != "" {
return name
}
return string(kind)
}
// safeSpanProps carries only a whitelist of span props to the frontend, plus a
// desensitized tool result summary. It never forwards raw arguments or secrets,
// and truncates string text so long content cannot leak a full sensitive payload.
func safeSpanProps(rec SpanRecord) map[string]any {
var props map[string]any
for _, k := range safePropKeys {
v, ok := rec.Props[k]
if !ok {
continue
}
if props == nil {
props = make(map[string]any)
}
props[k] = redactValue(v)
}
if rec.Kind == SpanKindTool {
if v, ok := rec.Props["result"]; ok {
if props == nil {
props = make(map[string]any)
}
props["result_summary"] = truncate(summarize(v), maxSummaryRunes)
}
}
return props
}
// redactValue shortens string props so a long error/step value cannot leak
// sensitive content through the dashboard; non-string values pass through.
func redactValue(v any) any {
if s, ok := v.(string); ok {
return truncate(s, maxSummaryRunes)
}
return v
}
// safePropKeys are the span prop keys forwarded to a node's props.
var safePropKeys = []string{"stop_reason", "step", "error", "error_kind", "tool"}
// statusForFinished maps a span terminal status to a node status.
func statusForFinished(s SpanStatus) string {
switch s {
case SpanStatusError:
return FlowStatusError
case SpanStatusOK:
return FlowStatusDone
default:
return FlowStatusIdle
}
}
// summarize renders an arbitrary value as a short text string for a prop summary.
func summarize(v any) string {
switch x := v.(type) {
case nil:
return ""
case string:
return x
default:
b, err := json.Marshal(x)
if err != nil {
return fmt.Sprintf("%v", x)
}
return string(b)
}
}
// truncate cuts s to at most max runes, appending an ellipsis when it overflows.
func truncate(s string, max int) string {
if max <= 0 {
return s
}
r := []rune(s)
if len(r) <= max {
return s
}
return string(r[:max]) + "…"
}
+161
View File
@@ -0,0 +1,161 @@
package trace
import (
"testing"
"time"
)
// spanRec builds a finished SpanRecord with the essential fields for the builder.
func spanRec(id, parent, name string, kind SpanKind, status SpanStatus, props map[string]any) SpanRecord {
now := time.Now()
return SpanRecord{
ID: id,
ParentID: parent,
Name: name,
Kind: kind,
Status: status,
StartedAt: now.Add(-time.Second),
EndedAt: now,
Duration: time.Second,
Props: props,
}
}
// graphFor builds a graph from the given finished spans plus input/output text.
func graphFor(spans []SpanRecord, input, output string) *FlowGraph {
g := NewRunGraph("web")
g.Start(1, time.Now())
g.SeedInput(input)
for _, s := range spans {
g.AddSpan(s)
}
g.SeedOutput(output)
return g.Snapshot()
}
func nodeByID(g *FlowGraph, id string) FlowNode {
for _, n := range g.Nodes {
if n.ID == id {
return n
}
}
return FlowNode{}
}
func hasEdge(g *FlowGraph, from, to string) bool {
for _, e := range g.Edges {
if e.From == from && e.To == to {
return true
}
}
return false
}
// TestBuildFlowGraphTrunk asserts the single-round main trunk assembles as
// input→llm→tool→output with final "done" status on every node.
func TestBuildFlowGraphTrunk(t *testing.T) {
llm := spanRec("s1", "step1", "llm.call", SpanKindLLM, SpanStatusOK,
map[string]any{"stop_reason": "tool_calls"})
tool := spanRec("s2", "step1", "tool.echo", SpanKindTool, SpanStatusOK,
map[string]any{"result": "hello world"})
g := graphFor([]SpanRecord{llm, tool}, "你好", "你好呀")
if g.SessionID != "web" {
t.Fatalf("SessionID = %q, want web", g.SessionID)
}
if len(g.Nodes) != 4 {
t.Fatalf("Nodes = %d, want 4 (input/llm/tool/output) -> %+v", len(g.Nodes), g.Nodes)
}
if nodeByID(g, "input").Kind != "input" {
t.Fatalf("input node kind = %q", nodeByID(g, "input").Kind)
}
if nodeByID(g, "s1").Kind != "llm" || nodeByID(g, "s1").Status != FlowStatusDone {
t.Fatalf("llm node was not folded: %+v", nodeByID(g, "s1"))
}
if nodeByID(g, "s2").Kind != "tool" || nodeByID(g, "s2").Status != FlowStatusDone {
t.Fatalf("tool node was not folded: %+v", nodeByID(g, "s2"))
}
if !hasEdge(g, "input", "s1") {
t.Fatalf("expected edge input->s1, got %+v", g.Edges)
}
if !hasEdge(g, "s1", "s2") {
t.Fatalf("expected edge s1->s2, got %+v", g.Edges)
}
if !hasEdge(g, "s2", "output") {
t.Fatalf("expected edge s2->output, got %+v", g.Edges)
}
}
// TestBuildFlowGraphErrorTool asserts a failed tool span finalizes the node as
// "error" while the rest of the trunk stays built.
func TestBuildFlowGraphErrorTool(t *testing.T) {
llm := spanRec("s1", "step1", "llm.call", SpanKindLLM, SpanStatusOK, nil)
tool := spanRec("s2", "step1", "tool.bad", SpanKindTool, SpanStatusError,
map[string]any{"error": "boom", "error_kind": "execution"})
g := graphFor([]SpanRecord{llm, tool}, "q", "")
if got := nodeByID(g, "s2").Status; got != FlowStatusError {
t.Fatalf("tool status = %q, want %q", got, FlowStatusError)
}
if got := nodeByID(g, "s2").Kind; got != "tool" {
t.Fatalf("tool kind = %q", got)
}
if nodeByID(g, "s2").Props["error_kind"] != "execution" {
t.Fatalf("tool props missing error_kind: %+v", nodeByID(g, "s2").Props)
}
if !hasEdge(g, "s1", "s2") {
t.Fatalf("expected edge s1->s2")
}
}
// TestBuildFlowGraphSkipsStateTransition asserts the step container span is not
// rendered as a node and does not break the chain.
func TestBuildFlowGraphSkipsStateTransition(t *testing.T) {
step := spanRec("step1", "", "loop.step", SpanKindStateTransition, SpanStatusOK, nil)
llm := spanRec("s1", "step1", "llm.call", SpanKindLLM, SpanStatusOK, nil)
g := graphFor([]SpanRecord{step, llm}, "q", "a")
for _, n := range g.Nodes {
if n.ID == "step1" {
t.Fatalf("state_transition span leaked as a node: %+v", n)
}
}
if !hasEdge(g, "input", "s1") {
t.Fatalf("expected edge input->s1 (through the skipped container), got %+v", g.Edges)
}
}
// TestBuildFlowGraphRunningAndDone asserts a node created at start (via
// AddSpanStart) is "running", then becomes "done" once its span finishes.
func TestBuildFlowGraphRunningAndDone(t *testing.T) {
g := NewRunGraph("web")
g.Start(1, time.Now())
g.SeedInput("q")
g.AddSpanStart("s1", "llm.call", SpanKindLLM, "", time.Now())
if got := nodeByID(g.Snapshot(), "s1").Status; got != FlowStatusRunning {
t.Fatalf("start status = %q, want running", got)
}
g.AddSpan(spanRec("s1", "", "llm.call", SpanKindLLM, SpanStatusOK, nil))
if got := nodeByID(g.Snapshot(), "s1").Status; got != FlowStatusDone {
t.Fatalf("finished status = %q, want done", got)
}
}
// TestBuildFlowGraphParentEdgePrioritizesParent asserts that when a node's parent
// is already a node (e.g. a nested subagent span), the edge wires to that parent
// instead of the linear chain.
func TestBuildFlowGraphParentEdgePrioritizesParent(t *testing.T) {
parent := spanRec("sub", "", "subagent.foo", SpanKindSubagent, SpanStatusOK, nil)
child := spanRec("nested", "sub", "llm.call", SpanKindLLM, SpanStatusOK, nil)
g := graphFor([]SpanRecord{parent, child}, "q", "a")
if !hasEdge(g, "sub", "nested") {
t.Fatalf("expected edge sub->nested (parent priority), got %+v", g.Edges)
}
if nodeByID(g, "sub").Kind != "subagent" {
t.Fatalf("subagent kind = %q", nodeByID(g, "sub").Kind)
}
}
+10
View File
@@ -158,6 +158,16 @@ func (s *Span) recordLocked() SpanRecord {
return rec
}
// Snapshot returns an immutable copy of a finished span as a SpanRecord. It is
// only safe to call after Finish returns (a closed span is immutable), and it
// lets a consumer read the span's final state without holding its lock. Calling
// it on a non-finished span returns a record with a zero Status/EndedAt.
func (s *Span) Snapshot() SpanRecord {
s.mu.RLock()
defer s.mu.RUnlock()
return s.recordLocked()
}
// SpanRecord is an immutable, lock-free snapshot of a completed span, produced
// by Finish and passed to a Writer for persistence. It is safe to copy.
type SpanRecord struct {