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) 提交里（同一个工作时段内）。
376 lines
13 KiB
Go
376 lines
13 KiB
Go
package gateway
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"sort"
|
||
"strings"
|
||
"testing"
|
||
"time"
|
||
|
||
"github.com/gorilla/websocket"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
|
||
)
|
||
|
||
// 这些测试覆盖 design/19 流程链中此前唯一没有测试的一段：
|
||
// 生产者的 bus 事件 → hub 中继 → 下行 WS 帧。loop 一侧由
|
||
// internal/runtime/loop/flow_test.go 钉住；这里钉住网关中继以及
|
||
// 前端 parseGraph 所读的线上字段名，因此任一侧改名都会
|
||
// 在这里失败，而不是静默让全链路流程页面变空。
|
||
|
||
// TestFlowGraphUpdateRelayedToWS 在生产主题（ws.graph_update）上
|
||
// 发布真实的 trace.RunGraph 快照，并断言客户端收到一个
|
||
// 原样携带该快照的 `graph_update` 帧。
|
||
func TestFlowGraphUpdateRelayedToWS(t *testing.T) {
|
||
env := newEnv(t, nil, false)
|
||
tok := env.token(t, auth.RoleOwner)
|
||
wsURL := "ws" + strings.TrimPrefix(env.ts.URL, "http") + "/ws/chat?token=" + tok
|
||
|
||
conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil)
|
||
if err != nil {
|
||
t.Fatalf("dial: %v", err)
|
||
}
|
||
defer conn.Close()
|
||
_ = conn.SetReadDeadline(time.Now().Add(3 * time.Second))
|
||
|
||
// 先往返一次 mode_switch，使连接在发布前已注册
|
||
// （与 TestWSUpgradeAndBroadcast 相同的保护）。
|
||
if err := conn.WriteJSON(wsInbound{Type: "mode_switch", Data: json.RawMessage(`{"mode":"chat"}`)}); err != nil {
|
||
t.Fatalf("write mode_switch: %v", err)
|
||
}
|
||
if got := readWS(t, conn); got.Type != "mode_switch" {
|
||
t.Fatalf("first frame type = %q, want mode_switch", got.Type)
|
||
}
|
||
|
||
snap := sampleFlowSnapshot()
|
||
if err := env.bus.Publish(context.Background(), "ws.graph_update", snap); err != nil {
|
||
t.Fatalf("publish graph: %v", err)
|
||
}
|
||
raw := readRawWS(t, conn)
|
||
|
||
var frame struct {
|
||
Type string `json:"type"`
|
||
Data json.RawMessage `json:"data"`
|
||
}
|
||
if err := json.Unmarshal(raw, &frame); err != nil {
|
||
t.Fatalf("decode frame %s: %v", raw, err)
|
||
}
|
||
if frame.Type != "graph_update" {
|
||
t.Fatalf("frame type = %q, want graph_update (frame=%s)", frame.Type, raw)
|
||
}
|
||
|
||
// 中继的载荷必须是快照原文 —— 同样的 JSON，未被重塑。
|
||
want, err := json.Marshal(snap)
|
||
if err != nil {
|
||
t.Fatalf("marshal snapshot: %v", err)
|
||
}
|
||
if string(frame.Data) != string(want) {
|
||
t.Fatalf("relayed data differs from the published snapshot:\n got %s\nwant %s", frame.Data, want)
|
||
}
|
||
}
|
||
|
||
// TestFlowGraphUpdateWireFields 把快照每一层的 JSON 键名钉到
|
||
// 前端读取器（lib/wsParse.ts 中的 parseGraph）所需的字段上。
|
||
// 它刻意做精确键集比较：新增、改名或丢弃
|
||
// 前端所读的字段必须在这里失败。
|
||
func TestFlowGraphUpdateWireFields(t *testing.T) {
|
||
raw, err := json.Marshal(sampleFlowSnapshot())
|
||
if err != nil {
|
||
t.Fatalf("marshal snapshot: %v", err)
|
||
}
|
||
var graph map[string]json.RawMessage
|
||
if err := json.Unmarshal(raw, &graph); err != nil {
|
||
t.Fatalf("decode graph: %v", err)
|
||
}
|
||
|
||
assertGraphKeys(t, graph)
|
||
nodes := decodeObjects(t, graph["nodes"], "nodes")
|
||
assertNodeKeys(t, nodes)
|
||
assertEdgeKeys(t, decodeObjects(t, graph["edges"], "edges"))
|
||
assertLogKeys(t, nodes)
|
||
assertGraphMeta(t, raw)
|
||
}
|
||
|
||
// assertGraphKeys 钉住 parseGraph/Flow.tsx 所读的 FlowGraph 顶层键。
|
||
//
|
||
// v0.77 起多了两个键（纯增量）：`round`（会话轮次号，客户端据此只保留最新一轮的
|
||
// 泳道）与 `turnId`（这张图所属**路**的权威回合标识，真并行时两路各一个值）。
|
||
// 前者在每次建图时必有；后者只在「这条路有权威标识」时出现（omitempty），
|
||
// 因此本断言只钉它在场时可读，缺席语义由 TestFlowGraphDispatchLaneFields 钉住。
|
||
func assertGraphKeys(t *testing.T, graph map[string]json.RawMessage) {
|
||
t.Helper()
|
||
want := []string{"id", "nodes", "edges", "sessionId", "round", "startedAt", "endedAt"}
|
||
if got := sortedKeys(graph); !equalKeys(got, want) {
|
||
t.Fatalf("graph keys = %v, want %v", got, want)
|
||
}
|
||
}
|
||
|
||
// assertNodeKeys 检查每个发出的节点键都是前端会读的、每个
|
||
// 节点都携带始终读取的 id/label，以及持有 props+log 的节点具备
|
||
// 完整的键集（这也钉住了 omitempty 标签）。
|
||
func assertNodeKeys(t *testing.T, nodes []map[string]json.RawMessage) {
|
||
t.Helper()
|
||
want := []string{"id", "label", "kind", "status", "startedAt", "endedAt", "props", "log"}
|
||
fullSeen := false
|
||
for i, n := range nodes {
|
||
keys := sortedKeys(n)
|
||
for _, k := range keys {
|
||
if !containsKey(want, k) {
|
||
t.Fatalf("node[%d] has unexpected key %q; frontend reads %v", i, k, want)
|
||
}
|
||
}
|
||
if !containsKey(keys, "id") || !containsKey(keys, "label") {
|
||
t.Fatalf("node[%d] keys = %v, must contain id and label", i, keys)
|
||
}
|
||
if containsKey(keys, "props") && containsKey(keys, "log") {
|
||
fullSeen = true
|
||
if !equalKeys(keys, want) {
|
||
t.Fatalf("node[%d] keys = %v, want the full set %v", i, keys, want)
|
||
}
|
||
}
|
||
}
|
||
if !fullSeen {
|
||
t.Fatal("no node carried both props and log; the omitempty tags are unpinned")
|
||
}
|
||
}
|
||
|
||
// assertEdgeKeys 钉住 FlowEdge 形状（parseGraph 读取 from/to）。
|
||
func assertEdgeKeys(t *testing.T, edges []map[string]json.RawMessage) {
|
||
t.Helper()
|
||
if len(edges) == 0 {
|
||
t.Fatal("snapshot has no edges")
|
||
}
|
||
for i, e := range edges {
|
||
if got := sortedKeys(e); !equalKeys(got, []string{"from", "to"}) {
|
||
t.Fatalf("edge[%d] keys = %v, want [from to]", i, got)
|
||
}
|
||
}
|
||
}
|
||
|
||
// assertLogKeys 钉住节点的 log 条目形状（parseGraph 读取 time/type/props）。
|
||
func assertLogKeys(t *testing.T, nodes []map[string]json.RawMessage) {
|
||
t.Helper()
|
||
var logged []map[string]json.RawMessage
|
||
for _, n := range nodes {
|
||
rawLog, ok := n["log"]
|
||
if !ok {
|
||
continue
|
||
}
|
||
if err := json.Unmarshal(rawLog, &logged); err != nil {
|
||
t.Fatalf("decode log: %v", err)
|
||
}
|
||
}
|
||
if len(logged) == 0 {
|
||
t.Fatal("snapshot has no log entries to pin")
|
||
}
|
||
for i, ev := range logged {
|
||
if got := sortedKeys(ev); !equalKeys(got, []string{"time", "type", "props"}) {
|
||
t.Fatalf("log[%d] keys = %v, want [props time type]", i, got)
|
||
}
|
||
}
|
||
}
|
||
|
||
// assertGraphMeta 钉住前端渲染的信封值（graph id/session
|
||
// 对），以及时间戳以 RFC3339Nano 抵达（Date.parse 所期望的格式）。
|
||
func assertGraphMeta(t *testing.T, raw []byte) {
|
||
t.Helper()
|
||
var meta struct {
|
||
ID string `json:"id"`
|
||
SessionID string `json:"sessionId"`
|
||
StartedAt string `json:"startedAt"`
|
||
}
|
||
if err := json.Unmarshal(raw, &meta); err != nil {
|
||
t.Fatalf("decode meta: %v", err)
|
||
}
|
||
if meta.ID != "web-1" || meta.SessionID != "web" {
|
||
t.Fatalf("graph id/sessionId = %q/%q, want web-1/web", meta.ID, meta.SessionID)
|
||
}
|
||
if _, err := time.Parse(time.RFC3339Nano, meta.StartedAt); err != nil {
|
||
t.Fatalf("startedAt %q is not RFC3339Nano: %v", meta.StartedAt, err)
|
||
}
|
||
}
|
||
|
||
// TestFlowGraphUpdateRelaysPersonaProp 端到端钉住 v0.47 契约的后半段：
|
||
// 图构建器放行的 `props.persona`（子代理分支根节点上的目标角色 id）
|
||
// 经真实 bus 中继后，客户端收到的 graph_update 帧里仍然看得到它。
|
||
// 只在 trace 包断言白名单或快照不足以证明「客户端真的收得到」——
|
||
// 中继链路上任何一环重塑载荷都会让该键丢失，所以这条从生产者
|
||
// 快照一路查到下行 WS 帧。
|
||
func TestFlowGraphUpdateRelaysPersonaProp(t *testing.T) {
|
||
env := newEnv(t, nil, false)
|
||
tok := env.token(t, auth.RoleOwner)
|
||
wsURL := "ws" + strings.TrimPrefix(env.ts.URL, "http") + "/ws/chat?token=" + tok
|
||
|
||
conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil)
|
||
if err != nil {
|
||
t.Fatalf("dial: %v", err)
|
||
}
|
||
defer conn.Close()
|
||
_ = conn.SetReadDeadline(time.Now().Add(3 * time.Second))
|
||
|
||
// 与 TestFlowGraphUpdateRelayedToWS 相同：先往返一次 mode_switch，
|
||
// 保证连接在发布前已注册。
|
||
if err := conn.WriteJSON(wsInbound{Type: "mode_switch", Data: json.RawMessage(`{"mode":"chat"}`)}); err != nil {
|
||
t.Fatalf("write mode_switch: %v", err)
|
||
}
|
||
if got := readWS(t, conn); got.Type != "mode_switch" {
|
||
t.Fatalf("first frame type = %q, want mode_switch", got.Type)
|
||
}
|
||
|
||
if err := env.bus.Publish(context.Background(), "ws.graph_update", samplePersonaSnapshot()); err != nil {
|
||
t.Fatalf("publish graph: %v", err)
|
||
}
|
||
raw := readRawWS(t, conn)
|
||
|
||
var frame struct {
|
||
Type string `json:"type"`
|
||
Data struct {
|
||
Nodes []struct {
|
||
ID string `json:"id"`
|
||
Props map[string]any `json:"props"`
|
||
} `json:"nodes"`
|
||
} `json:"data"`
|
||
}
|
||
if err := json.Unmarshal(raw, &frame); err != nil {
|
||
t.Fatalf("decode frame %s: %v", raw, err)
|
||
}
|
||
if frame.Type != "graph_update" {
|
||
t.Fatalf("frame type = %q, want graph_update (frame=%s)", frame.Type, raw)
|
||
}
|
||
for _, n := range frame.Data.Nodes {
|
||
if n.ID != personaBranchNodeID {
|
||
continue
|
||
}
|
||
if got, _ := n.Props["persona"].(string); got != personaBranchRole {
|
||
t.Fatalf("线上帧里分支根 props.persona = %q, want %q (frame=%s)", got, personaBranchRole, raw)
|
||
}
|
||
return
|
||
}
|
||
t.Fatalf("线上帧里找不到分支根节点 %q: %s", personaBranchNodeID, raw)
|
||
}
|
||
|
||
// personaBranchNodeID / personaBranchRole 是下面 fixture 里分支根
|
||
// 的节点 id 与它所扮演的角色 id，供上面的断言引用。
|
||
const (
|
||
personaBranchNodeID = "span-sub"
|
||
personaBranchRole = "yuki"
|
||
)
|
||
|
||
// samplePersonaSnapshot 构建一张带子代理分支的真实生产者侧快照：
|
||
// 分支根节点经 AddSpan 折叠（因此走 safeSpanProps 白名单，与生产路径
|
||
// 同一道过滤），persona 值就是 orchestration/scheduler.go 会设进
|
||
// span prop 的目标角色 id。
|
||
func samplePersonaSnapshot() *trace.FlowGraph {
|
||
base := time.Date(2026, 10, 3, 21, 0, 0, 0, time.UTC)
|
||
g := trace.NewRunGraph("web")
|
||
g.Start(1, base)
|
||
g.SeedInput("帮我查一下天气")
|
||
g.AddSpan(trace.SpanRecord{
|
||
ID: personaBranchNodeID,
|
||
ParentID: "input",
|
||
Name: "subagent.sub-1",
|
||
Kind: trace.SpanKindSubagent,
|
||
Status: trace.SpanStatusOK,
|
||
StartedAt: base.Add(time.Millisecond),
|
||
EndedAt: base.Add(9 * time.Millisecond),
|
||
Props: map[string]any{"persona": personaBranchRole},
|
||
})
|
||
g.SeedOutput("今天晴，26 度。")
|
||
return g.Snapshot()
|
||
}
|
||
|
||
// sampleFlowSnapshot 构建真实的生产者侧快照（trace.RunGraph，与
|
||
// loop 的 FlowPublisher 发布的类型相同），其每个可选线上字段
|
||
// 都被填充，使上面的键集断言有意义。
|
||
func sampleFlowSnapshot() *trace.FlowGraph {
|
||
base := time.Date(2026, 10, 3, 21, 0, 0, 0, time.UTC)
|
||
g := trace.NewRunGraph("web")
|
||
g.Start(1, base)
|
||
g.SeedInput("帮我查一下天气")
|
||
g.AddSpanStart("span-llm", "llm.call", trace.SpanKindLLM, "", base.Add(time.Millisecond))
|
||
g.AddSpan(trace.SpanRecord{
|
||
ID: "span-llm",
|
||
Name: "llm.call",
|
||
Kind: trace.SpanKindLLM,
|
||
Status: trace.SpanStatusOK,
|
||
StartedAt: base.Add(time.Millisecond),
|
||
EndedAt: base.Add(20 * time.Millisecond),
|
||
Props: map[string]any{"stop_reason": "tool_calls", "step": 1},
|
||
Events: []trace.Event{{Time: base.Add(2 * time.Millisecond), Type: "llm.done", Props: map[string]any{"tools": 1}}},
|
||
})
|
||
g.AddSpanStart("span-tool", "tool.get_weather", trace.SpanKindTool, "span-llm", base.Add(21*time.Millisecond))
|
||
g.AddSpan(trace.SpanRecord{
|
||
ID: "span-tool",
|
||
ParentID: "span-llm",
|
||
Name: "tool.get_weather",
|
||
Kind: trace.SpanKindTool,
|
||
Status: trace.SpanStatusOK,
|
||
StartedAt: base.Add(21 * time.Millisecond),
|
||
EndedAt: base.Add(30 * time.Millisecond),
|
||
Props: map[string]any{"tool": "get_weather", "result": "晴 26℃"},
|
||
Events: []trace.Event{{Time: base.Add(25 * time.Millisecond), Type: "tool.done", Props: map[string]any{"ok": true}}},
|
||
})
|
||
g.SeedOutput("今天晴，26 度。")
|
||
return g.Snapshot()
|
||
}
|
||
|
||
// decodeObjects 解码原始 JSON 对象数组，形状畸形时失败。
|
||
func decodeObjects(t *testing.T, raw json.RawMessage, what string) []map[string]json.RawMessage {
|
||
t.Helper()
|
||
var out []map[string]json.RawMessage
|
||
if err := json.Unmarshal(raw, &out); err != nil {
|
||
t.Fatalf("decode %s: %v", what, err)
|
||
}
|
||
if len(out) == 0 {
|
||
t.Fatalf("snapshot has no %s", what)
|
||
}
|
||
return out
|
||
}
|
||
|
||
// readRawWS 读取一个文本帧并返回其原始字节。
|
||
func readRawWS(t *testing.T, conn *websocket.Conn) []byte {
|
||
t.Helper()
|
||
_, msg, err := conn.ReadMessage()
|
||
if err != nil {
|
||
t.Fatalf("read ws: %v", err)
|
||
}
|
||
return msg
|
||
}
|
||
|
||
// sortedKeys 按排序后的顺序返回该对象的 JSON 键。
|
||
func sortedKeys(obj map[string]json.RawMessage) []string {
|
||
out := make([]string, 0, len(obj))
|
||
for k := range obj {
|
||
out = append(out, k)
|
||
}
|
||
sort.Strings(out)
|
||
return out
|
||
}
|
||
|
||
// equalKeys 报告 got 是否恰好拥有 want 的全部成员（二者均已排序）。
|
||
func equalKeys(got []string, want []string) bool {
|
||
want = append([]string(nil), want...)
|
||
sort.Strings(want)
|
||
if len(got) != len(want) {
|
||
return false
|
||
}
|
||
for i := range got {
|
||
if got[i] != want[i] {
|
||
return false
|
||
}
|
||
}
|
||
return true
|
||
}
|
||
|
||
// containsKey 报告（已排序的）键列表中是否含有 k。
|
||
func containsKey(keys []string, k string) bool {
|
||
for _, key := range keys {
|
||
if key == k {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|