AskaEth
4a937bd269
- 新增 StartRoundWithTurnID(ctx, owner, text)：从 ctx 取权威回合标识（与 chat_messages.turn_id
同值）；Run 的唯一生产建图入口切到它；原 StartRound 语义逐字不变（1 行包装）
- 归属桶（并行按路分图）保持不变；落盘查询键 lane_turn_id 对单路也有值 ⇒ 消息面按等值查得到图
- 无权威标识时**保持缺席**（不拿桶里的路标识顶上：不许臆造）
- 不需要迁移 0024（0023 的列+索引已在位；老单路行标识根本没存过，回填=编造）
形状变化（已登记契约）：graph.id 对单路也追加 #{回合标识}（web-7 → web-7#turn-…），
顺带修掉「重启后单路轮次 upsert 掉旧行」的隐患。3 个变异自检（退回老入口/无视 ctx/臆造回退）。
237 lines
11 KiB
Go
237 lines
11 KiB
Go
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)
|
||
}
|
||
}
|
||
}
|