diff --git a/Docs/api-client.md b/Docs/api-client.md index 23ae5ab..b5b9322 100644 --- a/Docs/api-client.md +++ b/Docs/api-client.md @@ -152,6 +152,16 @@ - **事件类型分流**（`BroadcastPersisted` 按 `sessionId` 分流）：`sessionId == "web"` 或其 `web:` 前缀 → **`message_persisted`**（Web 聊天实时同步）；**其它**（隔离 `sess-xxx`）→ **`channel_persisted`**（适配器日志专用）。 - **客户端处理**：Web 聊天**无** `channel_persisted` handler，天然忽略、不进会话；**IM 适配器日志页**订阅该事件，按 `adapterId === 当前适配器`（或总览 `adapterId != ''`）过滤后**仅追加**（按 `id` 去重、不整页重载），首次拉取仍由 `GET /api/v1/admin/channels/messages` + 轮询兜底。 +### 3.5 实时全链路流程事件（`graph_update`，design/19） +- **触发**：主会话 Web 每次会话轮次中，loop 的每个 trace span **开始/结束时**由 `loop.FlowPublisher` 经 bus `ws.graph_update` 广播增量 **FlowGraph**。**仅主人 Web 会话**（`sessionId == "web"` 或其 `web:` 前缀）发布——IM / 群聊 / 陌生人 / 子代理轮次**不**发布，防敏感信息泄向面板。管理员/主人 Web 端订阅后实时渲染「输入 → 思考(LLM) → 工具 → 输出」主干图。 +- **载荷**：`{type:"graph_update", data:{id, nodes:[{id,label,kind?,status?,startedAt?,endedAt?,props?,log?}], edges:[{from,to}], sessionId, startedAt, endedAt}}`。 + - `id` 为 `sessionId-round`（每轮一图）；`sessionId` 为会话 id；`startedAt/endedAt` 为 RFC3339。 + - `kind` ∈ `input|llm|tool|subagent|memory|guardrail|orchestration|output`。 + - `status` ∈ `idle|running|done|error`（`running` = span 在跑 / `done` = 成功 / `error` = 失败）。 + - `props` 仅含白名单摘要（如 `stop_reason` / `error_kind` / `result_summary`），**工具原始参数与密钥绝不前送**；长文本截断到 ~80 字。 + - `log` 为该节点累积的 span 事件 `[{time,type,props?}]`（设计 §2「节点实时日志」）。 +- **客户端处理**：`graph_update` → `setFlowGraph(parseGraph(data))`（`src/lib/wsParse.ts`），管理面板「全链路流程」页渲染 `FlowGraphView`，节点可点开看参数/结果/实时日志；无图时诚实显示「暂无流程」。 + ## 4. 客户端接入指引（Web / 移动 / 原生 / IM bridge 通用） ### 4.1 一次性接入流程（三步走） diff --git a/cmd/cyrene/wire.go b/cmd/cyrene/wire.go index be78576..be851c9 100644 --- a/cmd/cyrene/wire.go +++ b/cmd/cyrene/wire.go @@ -315,6 +315,17 @@ func (a *app) wireRuntime() error { store := a.memStore tracer := trace.New(trace.NewJSONRecorder(os.Stderr)) + // Realtime flow visualization (design/19 §1): the publisher accumulates the + // loop's trace spans into a per-session flow graph and publishes + // ws.graph_update on each span transition. It is gated to the owner's web + // sessions ("web" and web sub-sessions), so IM / stranger / sub-agent turns + // never leak into the admin panel. A nil flow keeps the loop's prior behavior. + flowPub := loop.NewFlowPublisher( + func(subject string, payload any) { _ = a.bus.Publish(context.Background(), subject, payload) }, + func(sessionID string) bool { + return sessionID == history.MainSessionID || strings.HasPrefix(sessionID, history.MainSessionID+":") + }, + ) checkpoint, perr := persist.New(context.Background(), a.db, a.log) if perr != nil { return fmt.Errorf("checkpoint store: %w", perr) @@ -348,6 +359,7 @@ func (a *app) wireRuntime() error { Extractor: memory.NewManualExtractor(personaID, "owner"), Checkpoint: checkpoint, Tracer: tracer, + Flow: flowPub, Guards: guards, PersonaID: personaID, Timezone: a.cfg.Timezone, diff --git a/frontend/web/src/admin/AdminApp.tsx b/frontend/web/src/admin/AdminApp.tsx index 32353d4..b3c2423 100644 --- a/frontend/web/src/admin/AdminApp.tsx +++ b/frontend/web/src/admin/AdminApp.tsx @@ -16,6 +16,7 @@ import { RagModule } from "./modules/Rag"; import { LlmModule } from "./modules/Llm"; import { AutomationsModule } from "./modules/Automations"; import { AgentsModule } from "./modules/Agents"; +import { FlowModule } from "./modules/Flow"; import { BackthinkModule } from "./modules/Backthink"; import { ObservabilityModule } from "./modules/Observability"; import { UsersModule } from "./modules/Users"; @@ -34,6 +35,7 @@ function renderSection(section: AdminSectionId): React.ReactNode { case "llm": return ; case "automations": return ; case "agents": return ; + case "flow": return ; case "backthink": return ; case "observability": return ; case "users": return ; diff --git a/frontend/web/src/admin/modules/Flow.tsx b/frontend/web/src/admin/modules/Flow.tsx new file mode 100644 index 0000000..e2af9ff --- /dev/null +++ b/frontend/web/src/admin/modules/Flow.tsx @@ -0,0 +1,189 @@ +import { useState } from "react"; +import { useAppStore } from "../../store/useAppStore"; +import type { AgentStatus, FlowGraph, FlowNode } from "../../lib/types"; +import { CardGrid, Notice, PageHeader, StatCard } from "../components"; +import { Surface, Tag, type TagTone } from "../../components/primitives"; +import { Icon } from "../../components/Icon"; +import { FlowGraphView } from "./FlowGraph"; + +const STATUS_LABEL: Record = { + idle: "空闲", + running: "运行中", + blocked: "阻塞", + done: "完成", + error: "错误", +}; +const STATUS_TONE: Record = { + idle: "neutral", + running: "accent", + blocked: "warn", + done: "ok", + error: "err", +}; +const STATUS_DOT: Record = { + idle: "var(--mu-surface-5)", + running: "var(--mu-accent)", + blocked: "var(--mu-warn)", + done: "var(--mu-ok)", + error: "var(--mu-err)", +}; + +/** 全链路流程页（design/19 §2）：实时渲染 ws.graph_update 驱动的单轮主干图， + * 输入 → 思考(LLM) → 工具 → 输出，节点可点开看实时日志/参数/结果。 */ +export function FlowModule(): React.ReactNode { + const graph = useAppStore((s) => s.flowGraph); + const [selectedId, setSelectedId] = useState(null); + const selected = graph?.nodes.find((n) => n.id === selectedId) ?? null; + + return ( +
+ + + {graph ? ( + <> + + } + label="会话" + value={{graph.sessionId ?? "web"}} + sub={graph.id} + tone="accent" + /> + } + label="节点" + value={graph.nodes.length} + sub={`${graph.edges.length} 条连线`} + tone="neutral" + /> + } + label="开始" + value={fmtStart(graph)} + sub="本轮开始时间" + tone="neutral" + /> + } + label="耗时" + value={fmtDuration(graph)} + sub="本轮时长" + tone="ok" + /> + + + +
+ {(["running", "done", "error", "idle"] as AgentStatus[]).map((s) => { + const n = countStatus(graph, s); + return ( + + + {STATUS_LABEL[s]} × {n} + + ); + })} +
+ setSelectedId(n.id)} /> + + + {selected ? ( + setSelectedId(null)} /> + ) : null} + + ) : ( + + 暂无流程 —— 接入 /ws/chat 并发起一次会话后，将实时出现「输入 → 思考 → 工具 → 输出」连通图；若某环节无对应 span，则诚实占位。 + + )} +
+ ); +} + +/** NodeDetail 展示单个节点的属性/结果与累积的 span 事件（实时日志）。 */ +function NodeDetail({ node, onClose }: { node: FlowNode; onClose: () => void }): React.ReactNode { + const status = node.status ?? "idle"; + const hasProps = !!node.props && Object.keys(node.props).length > 0; + const hasLog = !!node.log && node.log.length > 0; + + return ( + +
+ {node.label} + {node.kind ? {node.kind} : null} + {STATUS_LABEL[status]} + +
+
+ {node.startedAt ? 开始 {new Date(node.startedAt).toLocaleTimeString()} : null} + {node.endedAt ? 结束 {new Date(node.endedAt).toLocaleTimeString()} : null} + {node.id} +
+ + {hasProps ? ( +
+

参数 / 结果

+ 
+            {JSON.stringify(node.props, null, 2)}
+
+
+ ) : null} + + {hasLog ? ( +
+

实时日志（span 事件累积）

+ {node.log!.map((ev, i) => ( +
+ + {ev.time ? new Date(ev.time).toLocaleTimeString() : ""} + + {ev.type} + {ev.props ? ( + {JSON.stringify(ev.props)} + ) : null} +
+ ))} +
+ ) : null} + + {!hasProps && !hasLog ? 该节点暂无详细日志。 : null} + + ); +} + +/** countStatus returns how many nodes in graph are in the given status. */ +function countStatus(graph: FlowGraph, status: AgentStatus): number { + let n = 0; + for (const node of graph.nodes) { + if ((node.status ?? "idle") === status) n += 1; + } + return n; +} + +/** fmtStart renders the round start time as a locale time, or "—" when absent. */ +function fmtStart(graph: FlowGraph): string { + if (!graph.startedAt) return "—"; + const t = Date.parse(graph.startedAt); + return Number.isNaN(t) ? "—" : new Date(t).toLocaleTimeString(); +} + +/** fmtDuration renders the round duration from startedAt to endedAt (or now). */ +function fmtDuration(graph: FlowGraph): string { + if (!graph.startedAt) return "—"; + const start = Date.parse(graph.startedAt); + if (Number.isNaN(start)) return "—"; + const end = graph.endedAt ? Date.parse(graph.endedAt) : Date.now(); + const ms = Math.max(0, (Number.isNaN(end) ? Date.now() : end) - start); + if (ms >= 60000) return `${(ms / 60000).toFixed(1)} min`; + if (ms >= 1000) return `${(ms / 1000).toFixed(1)} s`; + return `${ms} ms`; +} diff --git a/frontend/web/src/admin/modules/FlowGraph.tsx b/frontend/web/src/admin/modules/FlowGraph.tsx index 4577c2f..f55d634 100644 --- a/frontend/web/src/admin/modules/FlowGraph.tsx +++ b/frontend/web/src/admin/modules/FlowGraph.tsx @@ -144,7 +144,13 @@ function clampLabel(label: string): string { return label.length > 20 ? `${label.slice(0, 19)}…` : label; } -export function FlowGraphView({ graph }: { graph: FlowGraph }): ReactNode { +export function FlowGraphView({ + graph, + onNodeClick, +}: { + graph: FlowGraph; + onNodeClick?: (node: FlowNode) => void; +}): ReactNode { const markerId = useId().replace(/[^a-zA-Z0-9]/g, ""); const { placed, edges, width, height } = layout(graph); @@ -186,7 +192,11 @@ export function FlowGraphView({ graph }: { graph: FlowGraph }): ReactNode { ); })} {placed.map((p) => ( - + onNodeClick(p.node) : undefined} + className={onNodeClick ? "cursor-pointer" : undefined} + > = new Set([ "llm", "automations", "agents", + "flow", "backthink", "observability", "system", diff --git a/frontend/web/src/admin/sections.tsx b/frontend/web/src/admin/sections.tsx index e22b7bb..703df50 100644 --- a/frontend/web/src/admin/sections.tsx +++ b/frontend/web/src/admin/sections.tsx @@ -10,7 +10,7 @@ export interface AdminSectionMeta { } /** - * The 15-module registry (design/10 §3). Order drives the sidebar; group drives + * The 16-module registry (design/10 §3). Order drives the sidebar; group drives * the section headers. All are clickable — several are shells until M10+ backend. */ export const ADMIN_SECTIONS: ReadonlyArray = [ @@ -25,6 +25,7 @@ export const ADMIN_SECTIONS: ReadonlyArray = [ { id: "llm", label: "LLM 模型", icon: "robot", group: "能力与连接" }, { id: "automations", label: "自动化与提醒", icon: "clock", group: "能力与连接" }, { id: "agents", label: "Agent 监控", icon: "agent", group: "运行时" }, + { id: "flow", label: "全链路流程", icon: "chart", group: "运行时" }, { id: "backthink", label: "后台思考", icon: "night", group: "运行时" }, { id: "observability", label: "可观测性", icon: "trend", group: "运行时" }, { id: "users", label: "用户与权限", icon: "users", group: "治理" }, diff --git a/frontend/web/src/lib/types.ts b/frontend/web/src/lib/types.ts index e264fed..808e361 100644 --- a/frontend/web/src/lib/types.ts +++ b/frontend/web/src/lib/types.ts @@ -156,6 +156,12 @@ export interface FlowNode { label: string; kind?: string; status?: AgentStatus; + /** Optional structured metadata (params/result summaries) for the node detail panel. */ + props?: Record; + /** Optional accumulated span events (log) for the node detail panel. */ + log?: TraceEvent[]; + startedAt?: string; + endedAt?: string; } export interface FlowEdge { from: string; @@ -165,6 +171,11 @@ export interface FlowGraph { id: string; nodes: FlowNode[]; edges: FlowEdge[]; + /** The graph's owning session id (backend `sessionId`). */ + sessionId?: string; + /** Round start/end times (RFC3339), when the backend provides them. */ + startedAt?: string; + endedAt?: string; } export interface SessionRecord { diff --git a/frontend/web/src/lib/wsParse.ts b/frontend/web/src/lib/wsParse.ts index 123f13e..27b9d65 100644 --- a/frontend/web/src/lib/wsParse.ts +++ b/frontend/web/src/lib/wsParse.ts @@ -109,8 +109,11 @@ export function parseAgent(payload: unknown): AgentInfo | null { export function parseGraph(payload: unknown): FlowGraph | null { const r = toRecord(payload); - if (!r || !Array.isArray(r.nodes)) return null; - const nodes = r.nodes + if (!r) return null; + // Support both a bare FlowGraph and a {sessionId, graph} envelope. + const graphRec = toRecord(r.graph) ?? r; + if (!Array.isArray(graphRec.nodes)) return null; + const nodes = graphRec.nodes .map((n) => toRecord(n)) .filter((n): n is Record => n !== null) .map((n) => ({ @@ -121,9 +124,22 @@ export function parseGraph(payload: unknown): FlowGraph | null { typeof n.status === "string" ? (n.status as FlowGraph["nodes"][number]["status"]) : undefined, + startedAt: typeof n.startedAt === "string" ? n.startedAt : undefined, + endedAt: typeof n.endedAt === "string" ? n.endedAt : undefined, + props: toRecord(n.props) ?? undefined, + log: Array.isArray(n.log) + ? n.log + .map((e) => toRecord(e)) + .filter((e): e is Record => e !== null) + .map((e) => ({ + time: typeof e.time === "string" ? e.time : "", + type: typeof e.type === "string" ? e.type : "", + props: toRecord(e.props) ?? undefined, + })) + : undefined, })); - const edges = Array.isArray(r.edges) - ? r.edges + const edges = Array.isArray(graphRec.edges) + ? graphRec.edges .map((e) => toRecord(e)) .filter((e): e is Record => e !== null) .map((e) => ({ @@ -131,7 +147,16 @@ export function parseGraph(payload: unknown): FlowGraph | null { to: typeof e.to === "string" ? e.to : "", })) : []; - return { id: typeof r.id === "string" ? r.id : "graph", nodes, edges }; + return { + id: typeof graphRec.id === "string" ? graphRec.id : "graph", + nodes, + edges, + sessionId: + (typeof r.sessionId === "string" ? r.sessionId : undefined) ?? + (typeof graphRec.sessionId === "string" ? graphRec.sessionId : undefined), + startedAt: typeof graphRec.startedAt === "string" ? graphRec.startedAt : undefined, + endedAt: typeof graphRec.endedAt === "string" ? graphRec.endedAt : undefined, + }; } /** Build a ChatMessage from a `message_persisted` payload (design/16 §3.3/§4.2): diff --git a/internal/runtime/loop/flow.go b/internal/runtime/loop/flow.go new file mode 100644 index 0000000..420e521 --- /dev/null +++ b/internal/runtime/loop/flow.go @@ -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 "" +} diff --git a/internal/runtime/loop/flow_test.go b/internal/runtime/loop/flow_test.go new file mode 100644 index 0000000..6c5b706 --- /dev/null +++ b/internal/runtime/loop/flow_test.go @@ -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 "" +} diff --git a/internal/runtime/loop/loop.go b/internal/runtime/loop/loop.go index d3266aa..1a760a3 100644 --- a/internal/runtime/loop/loop.go +++ b/internal/runtime/loop/loop.go @@ -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, diff --git a/internal/runtime/loop/options.go b/internal/runtime/loop/options.go index 2ee6998..b9e6a67 100644 --- a/internal/runtime/loop/options.go +++ b/internal/runtime/loop/options.go @@ -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. diff --git a/internal/runtime/loop/run.go b/internal/runtime/loop/run.go index 13000c5..6d5276f 100644 --- a/internal/runtime/loop/run.go +++ b/internal/runtime/loop/run.go @@ -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) diff --git a/internal/runtime/loop/trace.go b/internal/runtime/loop/trace.go index 2f20881..38e58b1 100644 --- a/internal/runtime/loop/trace.go +++ b/internal/runtime/loop/trace.go @@ -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 diff --git a/internal/runtime/trace/graph.go b/internal/runtime/trace/graph.go new file mode 100644 index 0000000..b789236 --- /dev/null +++ b/internal/runtime/trace/graph.go @@ -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]) + "…" +} diff --git a/internal/runtime/trace/graph_test.go b/internal/runtime/trace/graph_test.go new file mode 100644 index 0000000..7291dfc --- /dev/null +++ b/internal/runtime/trace/graph_test.go @@ -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) + } +} diff --git a/internal/runtime/trace/span.go b/internal/runtime/trace/span.go index abaa348..cbbc712 100644 --- a/internal/runtime/trace/span.go +++ b/internal/runtime/trace/span.go @@ -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 {