package loop // 本文件钉住**每轮落一条**的落盘口径(切片 F2;design/19 §4 P3)。它是「写放大」 // 这条红线在代码里的落点:落盘只能发生在 EndRound 上(每轮一次),**绝不**在每个 // span 上发生;且只对发布会话白名单内的会话发生。用假 recorder 数调用次数, // 不依赖任何数据库。 import ( "errors" "sync" "testing" "time" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" ) // fakeRecorder 记录每次落盘调用(含图的深拷贝)并可选地报错。 type fakeRecorder struct { mu sync.Mutex graphs []*trace.FlowGraph err error } func (f *fakeRecorder) RecordRound(g *trace.FlowGraph) error { f.mu.Lock() defer f.mu.Unlock() f.graphs = append(f.graphs, g) return f.err } func (f *fakeRecorder) count() int { f.mu.Lock() defer f.mu.Unlock() return len(f.graphs) } func (f *fakeRecorder) last() *trace.FlowGraph { f.mu.Lock() defer f.mu.Unlock() if len(f.graphs) == 0 { return nil } return f.graphs[len(f.graphs)-1] } // runRoundWithSpans 跑一轮:起轮 → 三个 span(起止)→ 收轮。它不读任何全局状态, // 发布/落盘的计数由各用例自己的闭包与 recorder 负责。 func runRoundWithSpans(p *FlowPublisher, session string) { p.StartRound(session, "帮我查天气") for i, name := range []string{"llm.call", "tool_call", "llm.call"} { sp := &trace.Span{ID: "span-" + string(rune('a'+i)), Name: name, Kind: trace.SpanKindLLM, StartedAt: time.Now()} p.SpanStarted(session, sp) sp.Finish(trace.SpanStatusOK) p.SpanFinished(session, sp) } p.EndRound(session, "今天晴,26 度。") } // TestFlowPublisher_EndRoundPersistsOnePerRound_正常 是本片的核心断言:无论这一轮 // 有多少个 span,**落盘恰好一次**(每轮一行),且落的是**收尾后**的图(含 output 节点)。 func TestFlowPublisher_EndRoundPersistsOnePerRound_正常(t *testing.T) { rec := &fakeRecorder{} p := NewFlowPublisher(func(string, any) {}, func(string) bool { return true }, WithRoundRecorder(rec)) runRoundWithSpans(p, "web") if got := rec.count(); got != 1 { t.Fatalf("落盘次数 = %d, want 1(每轮一行;每 span 一行是写放大,本片红线)", got) } g := rec.last() if g == nil { t.Fatal("recorder 没收到图") } if g.ID != "web-1" { t.Fatalf("落盘 turn id = %q, want web-1", g.ID) } if g.SessionID != "web" { t.Fatalf("落盘 session = %q, want web", g.SessionID) } kinds := map[string]bool{} for _, n := range g.Nodes { kinds[n.Kind] = true } if !kinds["input"] || !kinds["output"] { t.Fatalf("落盘的图必须含主干端点 input/output: %+v", g.Nodes) } } // TestFlowPublisher_EndRoundSkipsWhitelistOutside_正常 钉住落盘的**白名单复用之**: // 发布会话范围外的会话(隔离渠道会话)连图都不建,EndRound 直接返回 ⇒ 落盘调用为零。 func TestFlowPublisher_EndRoundSkipsWhitelistOutside_正常(t *testing.T) { rec := &fakeRecorder{} allow := func(session string) bool { return session == "web" } p := NewFlowPublisher(func(string, any) {}, allow, WithRoundRecorder(rec)) runRoundWithSpans(p, "sess-im-stranger") if got := rec.count(); got != 0 { t.Fatalf("白名单外会话落盘 %d 次, want 0(白名单是同一处判据,绝不放行)", got) } runRoundWithSpans(p, "web") if got := rec.count(); got != 1 { t.Fatalf("白名单内会话落盘 %d 次, want 1(反向证明上面的 0 不是整体坏了)", got) } } // TestFlowPublisher_RecorderFailureDoesNotDropRound_异常 钉住「落盘失败不影响会话」: // recorder 报错时 EndRound 照常发布收尾帧(图仍在内存、面板仍实时看得到),且不 panic。 func TestFlowPublisher_RecorderFailureDoesNotDropRound_异常(t *testing.T) { frames := 0 rec := &fakeRecorder{err: errors.New("db down")} p := NewFlowPublisher(func(string, any) { frames++ }, func(string) bool { return true }, WithRoundRecorder(rec)) runRoundWithSpans(p, "web") if rec.count() != 1 { t.Fatalf("recorder 应被调用一次(失败也要尝试)") } if frames == 0 { t.Fatal("落盘失败不该吞掉实时帧 —— EndRound 仍须发布收尾帧") } // 实时图仍保留最近一轮(LastGraph 是收尾快照的出口)。 if snap := p.LastGraph("web"); snap == nil || len(snap.Nodes) == 0 { t.Fatal("落盘失败后最近一轮图仍在内存(LastGraph 仍可读)") } } // TestFlowPublisher_NoRecorderStillPublishes_边界 钉住 F2 是**纯增量**:没挂 recorder // 时 EndRound 行为与 F1 之前逐字相同(只发布、不落盘、不 panic)。 func TestFlowPublisher_NoRecorderStillPublishes_边界(t *testing.T) { frames := 0 p := NewFlowPublisher(func(string, any) { frames++ }, func(string) bool { return true }) runRoundWithSpans(p, "web") if frames == 0 { t.Fatal("没有 recorder 时 EndRound 仍须正常发布") } }