package loop import ( "context" "strings" "testing" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" ) // 本文件钉住**按路分桶**这条改造的其余口径(核心行为已在 flow_parallel_test.go 用真 // 循环跑过):桶 / 图 id / 帧上 `turnId` 键三者同源,最近一轮的每一路都拿得回来, // 且新一轮开始后不再把上一轮的路留在内存里。这些用例走发布器的公开面(真调用点 // 传给它的就是这些字符串),不依赖任何数据库。 // laneIDs 返回每一路图的 id 与 turnId 键,供断言用。 func laneIDs(lanes []*trace.FlowGraph) []string { out := make([]string, 0, len(lanes)) for _, g := range lanes { out = append(out, g.ID+"@"+g.TurnID) } return out } // TestFlowPublisher_ParallelLanesGetOwnGraphs 钉住真并行两路各有各的桶: // 图 id 不同(落盘不撞 key)、都归属同一个会话、都带上自己的路标识(帧上 `turnId`), // 且最近一轮的两路都能经 LatestLanes 拿回来。 func TestFlowPublisher_ParallelLanesGetOwnGraphs(t *testing.T) { p := NewFlowPublisher(nil, func(string) bool { return true }) p.StartRound(FlowOwnerKey("web", "turn-a"), "标记A") p.SpanStarted(FlowOwnerKey("web", "turn-a"), spanNamed("span-a", "tool.probe_a")) p.SpanFinished(FlowOwnerKey("web", "turn-a"), finishedSpan("span-a", "tool.probe_a")) p.StartRound(FlowOwnerKey("web", "turn-b"), "标记B") p.SpanStarted(FlowOwnerKey("web", "turn-b"), spanNamed("span-b", "tool.probe_b")) p.SpanFinished(FlowOwnerKey("web", "turn-b"), finishedSpan("span-b", "tool.probe_b")) ga := p.LastGraph("web") if ga == nil { t.Fatal("LastGraph = nil,但该会话刚起了两路") } // 最近一路 = 后起的那一路;它的图里只有自己的工具。 if ga.TurnID != "turn-b" || !hasNode(ga, "probe_b") || hasNode(ga, "probe_a") { t.Fatalf("LastGraph 不是 B 那一路的图:id=%s turnId=%q nodes=%s", ga.ID, ga.TurnID, describeNodes(ga)) } lanes := p.LatestLanes("web") if len(lanes) != 2 { t.Fatalf("最近一轮的路数 = %d,want 2(两路各自一张图):%v", len(lanes), laneIDs(lanes)) } if got := laneIDs(lanes); got[0] != "web-1#turn-a@turn-a" || got[1] != "web-2#turn-b@turn-b" { t.Fatalf("两路的 id@turnId = %v,want [web-1#turn-a@turn-a web-2#turn-b@turn-b]", got) } if lanes[0].ID == lanes[1].ID { t.Fatalf("两路撞同一个图 id %q(落盘会互相顶掉)", lanes[0].ID) } for _, g := range lanes { if g.SessionID != "web" { t.Fatalf("图 %s 的 sessionId = %q,want web(路标识绝不改变会话归属)", g.ID, g.SessionID) } if g.Round != lanes[0].Round { t.Fatalf("同一轮的两路轮次号不同:%d vs %d", g.Round, lanes[0].Round) } } // 收尾后仍是最近一轮的每一路(页面在这之后才打开也看得到两条)。 p.EndRound(FlowOwnerKey("web", "turn-a"), "答复A") p.EndRound(FlowOwnerKey("web", "turn-b"), "答复B") if got := p.LatestLanes("web"); len(got) != 2 { t.Fatalf("收轮后最近一轮的路数 = %d,want 2", len(got)) } } // TestFlowPublisher_NewRoundDropsEarlierLanes 钉住轮次推进与内存边界:新一轮只在 // 上一轮的路全部收尾之后开始(该轮的路因此共享轮次号),并丢掉更早轮次的桶 —— // 否则每一条真并行的路(turnId 每回合全新、桶键永不复用)都会在内存里永久留一张图。 func TestFlowPublisher_NewRoundDropsEarlierLanes(t *testing.T) { p := NewFlowPublisher(nil, func(string) bool { return true }) p.StartRound(FlowOwnerKey("web", "turn-a"), "标记A") p.StartRound(FlowOwnerKey("web", "turn-b"), "标记B") round := p.LatestLanes("web")[0].Round p.EndRound(FlowOwnerKey("web", "turn-a"), "答复A") p.EndRound(FlowOwnerKey("web", "turn-b"), "答复B") // 上一轮尚未被新轮次顶掉时,两条路都还在。 if got := p.LatestLanes("web"); len(got) != 2 { t.Fatalf("上一轮的路数 = %d,want 2(收轮后仍可补拉)", len(got)) } p.StartRound("web", "单路的下一轮") next := p.LatestLanes("web") if len(next) != 1 || next[0].TurnID != "" { t.Fatalf("新一轮的路 = %v,want 恰好一条无路标识的图(上一轮的桶必须被丢掉)", laneIDs(next)) } if next[0].Round <= round { t.Fatalf("新一轮轮次号 = %d,want > %d", next[0].Round, round) } if g := p.LastGraph("web"); g == nil || g.TurnID != "" { t.Fatalf("新一轮的 LastGraph = %+v,want 单路那张(上一轮的路已不在内存)", g) } } // TestFlowPublisher_RoundsShareRoundNumberButNeverID 钉住「轮次号」与「图 id 的序号」 // 是两个量:同一轮里可以有**不止一条路**(真并行的那一路 + 队首续跑的串行路),它们 // 共享轮次号(客户端据此把同一轮的路并排画),但**各有各的图 id** —— 否则落盘的 // `(session_id, turn_id)` 唯一键会把同一会话的两轮 upsert 成一行,前一轮整张图消失。 func TestFlowPublisher_RoundsShareRoundNumberButNeverID(t *testing.T) { rec := &fakeRecorder{} p := NewFlowPublisher(nil, func(string) bool { return true }, WithRoundRecorder(rec)) // 一条并行路先起(占用本轮),随后两条串行路在它仍飞着的时候先后跑完。 p.StartRound(FlowOwnerKey("web", "turn-a"), "并行的一路") p.StartRound("web", "串行第一轮") p.EndRound("web", "答复1") p.StartRound("web", "串行第二轮") p.EndRound("web", "答复2") lines := recordedRounds(rec) if len(lines) != 2 { t.Fatalf("落盘 %d 条,want 2(两条串行路各一条)", len(lines)) } if lines[0].ID == lines[1].ID { t.Fatalf("同一轮的两条串行路撞同一个 id %q ⇒ 落盘会把前一轮整行顶掉", lines[0].ID) } if lines[0].Round != lines[1].Round { t.Fatalf("同一轮的两条路轮次号不同:%d vs %d", lines[0].Round, lines[1].Round) } } // TestFlowPublisher_SingleLegUnchanged 钉住**老入口(无 ctx 通道)的单路逐字不变**: // 归属桶就是会话 id,图 id 仍是 `{会话}-{轮次}`(没有后缀),帧上 `turnId` 键缺席。 // 既有回归(flow_snapshot_test.go 的 web-1/web-2)与本条是同一口径的两侧。 // // 单路在**有 ctx 通道**时的形状(覆盖全部编排回合的那条路)见 flow_turnid_test.go: // 那里 `turnId` 是**在场**的(权威回合标识),因为「单路没有权威标识」这件事本身 // 已经随 CODE-TC-2 消失 —— 本入口的缺席只说明**它没有 ctx 可读**。 func TestFlowPublisher_SingleLegUnchanged(t *testing.T) { p := NewFlowPublisher(nil, func(string) bool { return true }) p.StartRound("web", "你好") p.EndRound("web", "你也好") g := p.LastGraph("web") if g == nil || g.ID != "web-1" || g.TurnID != "" { t.Fatalf("单路图 = %+v,want id=web-1 且无 turnId(老入口逐字不变)", g) } } // TestFlowOwnershipHelpers 钉住 loop 与 orchestration 之间的归属口径: // 真并行路由 `{会话}#{路}` 归属、真会话 id 与路标识各自可单独取出(总线载荷要的是 // 会话、图归属要的是桶),子代理分支继承派生它的那一路。 func TestFlowOwnershipHelpers(t *testing.T) { // 与 Run 同形:会话先经 withLoopSession 播种,网关再加路标识(WithParallelLeg)。 ctx := withLoopSession(WithParallelLeg(WithTurnID(context.Background(), "turn-a"), "turn-a"), "web") if got := flowOwner(ctx, "web"); got != "web#turn-a" { t.Fatalf("并行路的归属桶 = %q,want web#turn-a", got) } if got := FlowOwnerFromContext(ctx); got != "web#turn-a" { t.Fatalf("FlowOwnerFromContext = %q,want web#turn-a", got) } if got := FlowSessionFromContext(ctx); got != "web" { t.Fatalf("FlowSessionFromContext = %q,want web(总线载荷要真会话 id)", got) } if got := FlowLaneFromContext(ctx); got != "turn-a" { t.Fatalf("FlowLaneFromContext = %q,want turn-a", got) } // 子代理(分支 Run):归属继承分支桶,会话与路各自拆得回来。 branch := WithFlowBranch(context.Background(), FlowOwnerFromContext(ctx)) if got := flowOwner(branch, "sub-1"); got != "web#turn-a" { t.Fatalf("子代理 span 的归属桶 = %q,want web#turn-a(必须落进派生它的那一路)", got) } if got := FlowSessionFromContext(branch); got != "web" { t.Fatalf("子代理的 FlowSessionFromContext = %q,want web", got) } if got := FlowLaneFromContext(branch); got != "turn-a" { t.Fatalf("子代理的 FlowLaneFromContext = %q,want turn-a", got) } // 单路:桶、会话、路三者不出意外的形态(路为空)。 single := withLoopSession(context.Background(), "web") if got := flowOwner(single, "web"); got != "web" { t.Fatalf("单路归属桶 = %q,want web", got) } if got := FlowOwnerFromContext(single); got != "web" { t.Fatalf("单路 FlowOwnerFromContext = %q,want web", got) } if got := FlowLaneFromContext(single); got != "" { t.Fatalf("单路 FlowLaneFromContext = %q,want 空", got) } // 没有会话就没有归属:绝不回 `#路` 这种既查不到图、又骗调用方的串。 if got := FlowOwnerFromContext(WithParallelLeg(context.Background(), "turn-a")); got != "" { t.Fatalf("无会话时 FlowOwnerFromContext = %q,want 空", got) } if got := FlowSessionFromContext(WithParallelLeg(context.Background(), "turn-a")); got != "" { t.Fatalf("无会话时 FlowSessionFromContext = %q,want 空", got) } if _, lane := splitFlowOwner("web"); lane != "" { t.Fatalf("无 # 的归属串应拆出空路,得到 %q", lane) } } // spanNamed / finishedSpan 造一个刚起 / 刚结束的 span,让图里有真实的节点。 func spanNamed(id, name string) *trace.Span { return &trace.Span{ID: id, Name: name, Kind: trace.SpanKindTool} } func finishedSpan(id, name string) *trace.Span { sp := spanNamed(id, name) sp.Finish(trace.SpanStatusOK) return sp } // TestFlowPublisher_LaneFramesCarryLaneKey 钉住**帧载荷**上的 `turnId` 键: // 客户端据此把两路的帧分成两条泳道(旧客户端忽略未知键仍可用)。 func TestFlowPublisher_LaneFramesCarryLaneKey(t *testing.T) { var frames []*trace.FlowGraph p := NewFlowPublisher(func(_ string, payload any) { if g, ok := payload.(*trace.FlowGraph); ok { frames = append(frames, g) } }, func(string) bool { return true }) p.StartRound(FlowOwnerKey("web", "turn-a"), "标记A") p.StartRound(FlowOwnerKey("web", "turn-b"), "标记B") if len(frames) < 2 { t.Fatalf("发布帧数 = %d,want ≥2", len(frames)) } for _, g := range frames { if g.TurnID == "" { t.Fatalf("并行路的帧缺 turnId 键(客户端无法分成两条泳道):id=%s", g.ID) } if !strings.Contains(g.ID, g.TurnID) { t.Fatalf("图 id %q 里找不到路标识 %q(两者必须同源)", g.ID, g.TurnID) } if g.Round == 0 { t.Fatalf("帧缺轮次号:id=%s", g.ID) } } }