Cyrene_v2/internal/runtime/loop/flow_persist_test.go

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 仍须正常发布")
	}
}