Files
Cyrene_v2/internal/runtime/trace/graph_build.go
T
AskaEth 6e7ec56f9a
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled
feat(loop,gateway): 事件批 — tool_execution 三事件 + streamId + turn.start/end(裁决 01.4.1/01.4.2/01.4.3/01.4.5/03.4.4/1.12)
执行方案阶段 2 · P2-3(`DevDocs/audit/_plan.md`)。契约 `Docs/api-client.md` → **v0.50**。

- **01.4.1 `tool_execution` 拆三事件**:`model/stream.go` 删旧 `StreamKindToolExecution`,新增
  `tool_execution.start/update/end` + `ToolPhaseRunning/Done/Error`(与 design/19 status 同值);
  `StreamEvent` 增 `Tool`/`Phase`;`loop/tools.go` 的 `emitToolExecution` 重写为结构化字段、
  **Delta 恒空**(断言锁死),`ID` 复用 provider tool call id(可与 `tool_call.*` 关联)。
  ⚠️ **诚实边界**:start/end 有真实生产者;`update` 的 wire 映射已实现并单测锁定,但**工具分发器
  无中途进度上报源 ⇒ loop 暂不产出 update 帧**(已在 api-client §3.1 如实登记)。
- **03.4.4 phase 驱动图节点状态**:`invokeOne` 用**单一 phase 决策**同时决定 end 帧 phase、
  span 的 `phase` prop 与终态;`trace/graph.go` 新增 `FlowStatusForPhase`,`graph_build.go` 的
  `AddSpan` 对 tool 节点**优先消费 phase prop**(缺省回落 span 终态 ⇒ 两路不可能矛盾)。
- **01.4.2 `streamId`**:`StreamEvent` 增 `StreamID`;`loop/stream.go` 每次 ChatStream 生成
  `stream-<hex>`(crypto/rand,降级时间戳),该流所有帧共享、多步工具轮取新 id;`ws_bridge`
  把它并进**模型流帧** `data`(`tool_execution.*`/`turn.*` 不带)。语义区分:`turnId` = 哪轮回答,
  `streamId` = 这条流式消息。
- **01.4.3 `turn.start`/`turn.end`**:`loop/run.go` 入口发 start、defer 发 end,**覆盖全部出口**
  (最终答案/安全网终止/预算耗尽/步骤错误);帧**无 data 载荷**;静音的内部 run(backthink/子代理)
  不发。顺带把 91 行的 `Run` 拆成 `driveTurn` + `settleAndReturn`(清掉一处既有 >80 行违规)。
- **01.4.5**:仅 `Docs/api-client.md` §3.5 补 `graph_update` 三范围(`web`/`web:<uuid>`/`sess-backthink`),
  与 `wire_runtime.go` 的 allow 谓词一致,**零代码变更**。

验证:`go build ./...` 绿;本批四包(loop/model/trace/gateway)vet + 测试绿(含 `-count=1`);
linux/amd64 + darwin/arm64 交叉编译绿;本批 16 文件 ≤500 行、函数 ≤80 行。
**真 WS 帧断言 8 条**:帧序 `turn.start → message.delta → tool_execution.start →
tool_execution.end → message.delta → turn.end`(+`stream_end`)、turn 边界无载荷且整轮 turnId 一致、
两条模型流 `streamId` 互异、tool 帧 `{tool,phase,id}` 且无 `delta`/`streamId` 键、
`turn.end` 在步骤出错时也发出、内部静音 run 不发边界帧。
**变异验证**:把 `ev.StreamID = streamID` 改为赋空 → 两条断言 FAIL(`streamId` 为空/两流同空),
还原后复绿(直接删行会先被编译期 `declared and not used` 拦截)。

未验证:前端未消费这些帧(本轮未动 `frontend/**`,「前端能渲染」待前端批);无真实模型端到端。
2026-10-04 11:26:35 +08:00

262 lines
7.7 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package trace
import (
"strings"
"time"
)
// Node kinds the builder emits. They are named constants because the builder
// itself branches on the subagent kind when it resolves a node's lane
// (design/19 §2: parallel sub-agent branches render as separate swimlanes).
const (
nodeKindLLM = "llm"
nodeKindTool = "tool"
nodeKindSubagent = "subagent"
nodeKindMemory = "memory"
nodeKindGuardrail = "guardrail"
nodeKindOrchestrate = "orchestration"
)
// maxLaneWalk bounds how far the builder walks a span's parent chain when it
// resolves the lane (nearest enclosing subagent node) of a new node. Span
// parents form a tree, so the bound only guards against malformed input — it
// keeps the walk finite without a magic number in the loop (project rule §10).
const maxLaneWalk = 64
// trunkLane is the lane key of the main trunk: the lane of every span that has no
// sub-agent ancestor. It is a named constant so the map lookups read as intent.
const trunkLane = ""
// 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 span
// belongs to a lane (subagent branch) or the main trunk and is chained after the
// last node of that same lane — so a lane stays a coherent step chain even while
// several lanes are in flight, and the first node of a lane hangs off its
// subagent root (design/19 §2: 泳道内展示该子代理的步骤链).
func (g *RunGraph) Upsert(spec NodeSpec) {
if spec.ID == "" {
return
}
g.rememberParent(spec.ID, spec.ParentID)
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)
lane := g.laneOf(spec.ParentID)
from := g.laneLast[lane]
if spec.ParentID != "" {
if _, ok := g.nodes[spec.ParentID]; ok {
from = spec.ParentID
}
}
if from == "" {
from = lane
}
if from != "" && from != spec.ID {
g.addEdge(from, spec.ID)
}
// A sub-agent root is drawn on the trunk it hangs off, but it does not become
// the trunk's continuation point: the main chain resumes where the branch left
// it (the spawn node). Without this, every node after a spawn would look like a
// step of the last branch (the frontend derives lanes from the edges). The
// first trunk node still advances the cursor, so a graph built without an input
// seed cannot orphan its later trunk nodes.
if lane != "" || n.Kind != nodeKindSubagent || g.laneLast[trunkLane] == "" {
g.laneLast[lane] = spec.ID
}
}
// rememberParent records a span's parent link even when the span is not rendered
// as a node, so laneOf can walk through skipped containers.
func (g *RunGraph) rememberParent(id, parent string) {
if id == "" {
return
}
g.parents[id] = parent
}
// laneOf returns the id of the nearest enclosing subagent node for a span's
// parent chain, or "" when the chain belongs to the main trunk. Only nodes
// already in the graph can be a lane root, so a lane resolves as soon as its
// first step is folded in.
func (g *RunGraph) laneOf(parentID string) string {
for id, hops := parentID, 0; id != "" && hops < maxLaneWalk; hops++ {
if n, ok := g.nodes[id]; ok && n.Kind == nodeKindSubagent {
return id
}
id = g.parents[id]
}
return ""
}
// 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})
}
// 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 — but
// their parent link is remembered so lane resolution can walk through them.
// tool 节点的终态优先由其携带的细粒度 phase 驱动(design/04 §1.1 /
// design/19 §1:running/done/error),缺失时回落到 span 终态推导,
// 因此两套来源对同一节点绝不可能得出不同状态。
func (g *RunGraph) AddSpan(rec SpanRecord) {
kind := kindForSpan(rec.Kind)
if kind == "" {
g.rememberParent(rec.ID, rec.ParentID)
return
}
status := statusForFinished(rec.Status)
if kind == nodeKindTool {
if ph := FlowStatusForPhase(spanPhase(rec)); ph != "" {
status = ph
}
}
g.Upsert(NodeSpec{
ID: rec.ID,
Label: labelForSpan(rec),
Kind: kind,
Status: status,
StartedAt: rec.StartedAt,
EndedAt: rec.EndedAt,
Props: safeSpanProps(rec),
Log: redactEvents(rec.Events),
ParentID: rec.ParentID,
})
}
// spanPhase 读取 tool span 上记录的执行 phase;缺失或类型不符时返回
// 空串(调用方据此回落),不区分「没有」与「读不到」。
func spanPhase(rec SpanRecord) string {
ph, _ := rec.Props["phase"].(string)
return ph
}
// 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 (their parent link is still remembered for lane resolution), 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 == "" {
g.rememberParent(id, parent)
return false
}
g.Upsert(NodeSpec{
ID: id,
Label: labelForStart(name, kind),
Kind: nodeKind,
Status: FlowStatusRunning,
StartedAt: at,
ParentID: parent,
})
return true
}
// 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 nodeKindLLM
case SpanKindTool:
return nodeKindTool
case SpanKindSubagent:
return nodeKindSubagent
case SpanKindMemory:
return nodeKindMemory
case SpanKindGuardrail:
return nodeKindGuardrail
case SpanKindOrchestration:
return nodeKindOrchestrate
default:
return ""
}
}
// labelForSpan renders a node label from a finished span.
func labelForSpan(rec SpanRecord) string {
if kind := kindForSpan(rec.Kind); kind == nodeKindTool {
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)
}
// 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
}
}