feat(flow): 链路图本地落盘 + 多轮历史回放(design/19 §P3,契约 v0.73)
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled

主人要求「链路流程应该有本地数据,不然每次重载之后页面就没数据了」:
- 落盘:迁移 0022_flow_graphs(session_id/turn_id/graph_json/时间戳 + 唯一键 upsert)+ internal/runtime/flowgraph(RecordRound/Graph/Latest/Turns/Turn + 保留期 GC)
- 写入:loop.EndRound **每轮一条**(锁外深拷贝后写;失败 WARN 且不打印图内容);只落发布会话白名单(复用 flowSessionAllowed)
- 读回:端点加 turnId + source(live/persisted/none) + turns[](最新在前,空为 []);**内存优先、miss 才落盘**(最新一轮永远最新)
- 前端:flowGraphs 纳入 persistOptions.partialize(**刷新即刻有图**,version 3→4 + migrate)+ 历史轮选择器 + 诚实文案;顺带收 L1 转交的 flow/NodeDetail.tsx 节点日志倒序
- 新 env 键:FLOW_GRAPH_RETENTION=168h / FLOW_GRAPH_GC_INTERVAL=1h
- 变异实测:每 span 一行 / 保留期守卫失效 / partialize 去掉 flowGraphs / 去掉内存优先 / 白名单失效 ⇒ 全部打红
- 3 个禁动面文件的纯追加改动(Flow.tsx 挂选择器、FlowSnapshotHint 纠已成假话的文案、其渲染断言随之更新)由主会话裁定认可
- 登记存疑:图桶持久化无体积上限;useFlowTurns 的 effect 无自动化证据(项目无 jsdom)
This commit is contained in:
AskaEth committed 2026-10-05 16:19:32 +08:00
1 parent 84c124f5a9
commit a6cb44b8a8
38 files changed
+3047 -91

No files matched your search

+324
View File
@@ -0,0 +1,324 @@
// Package flowgraph 是**链路图本地落盘**的存储地基(design/19 §4 P3「多轮历史回放」;
// 切片 F2)。它补的是主人 2026-10-05 追加要求的那句「链路流程应该有本地数据,不然
// 每次重载之后页面就没数据了」—— 在此之前图只活在 `loop.FlowPublisher` 的**内存**里,
// 后端一重启、或该会话还没跑新轮次,就什么都查不到。
//
// 只做一件事:把**每一轮的收尾图**(`trace.FlowGraph`)连会话与轮次标识落进
// `flow_graphs`(迁移 0022),并支持按轮次读回、按保留期清扫。
//
// 三条职责边界(别在这里越界):
//
// - **不判身份、不判白名单**:哪些会话能产出图由发布器的发布会话范围
// (`cmd/cyrene` 的 `flowSessionAllowed`,design/19 §1 裁决 01.4.5)决定 ——
// 本包只被**已经过白名单的** `EndRound` 调用,因此不再存一份白名单。
// - **写入时机 = 每轮一条**(`EndRound`),**不是每 span 一条**:span 级写放大是纯浪费,
// 图在收尾时已完整(`output` 节点已并入)。
// - **不是图的真源**:实时图仍由发布器持有(`graph_update` 帧的唯一来源);本包只是
// 一份**收尾存档**,供「打开页面后翻历史轮」。读写失败都**不许**影响会话收尾。
//
// 脱敏:`graph_json` 里是**出图前已清洗**的图(节点 props 走 design/19 §1.1 的放行表 +
// 凭据形状清洗,label 也经 `trace.RedactText`)。本包**只用结构化字段**(session_id /
// turn_id)打日志,**绝不打印图内容** —— 图里可能含用户输入文本。
package flowgraph
import (
"context"
"database/sql"
"encoding/json"
"errors"
"fmt"
"io"
"log/slog"
"strings"
"time"
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
)
// 存储侧的哨兵错误。调用方一律用 `errors.Is` 分类,**绝不按错误文本匹配**。
var (
// ErrNilDB 报告构造 Store 时没有可用的数据库连接。宁可明确失败,也不给出
// 一个写不进去的替身(与 `events.NewStore` / `history.NewStore` 同一口径)。
ErrNilDB = errors.New("flowgraph: 缺少数据库连接")
// ErrNoGraphID 报告一条待落盘的图缺少会话 id 或轮次标识。两者是「这一行属于哪一轮」
// 的唯一坐标,缺一个就既查不到也归不了档 —— 这是调用方的编程错误,必须显式失败。
ErrNoGraphID = errors.New("flowgraph: 轮次图缺少会话或轮次标识")
// ErrBadTimestamp 报告读回的轮次行携带一个无法解析的时间戳。宁可如实报错,也不
// 静默给一个零值时间(那会让 GC 与列表排序基于假时刻做判断)。
ErrBadTimestamp = errors.New("flowgraph: 轮次行时间戳不可解析")
)
const (
// sqliteTimeLayout 与迁移 0022 的默认值 `strftime('%Y-%m-%dT%H:%M:%fZ','now')`
// 逐字同形(UTC、毫秒、字面量 Z)。写入与解析共用它,使「我们写进去的」与
// 「SQLite 默认生成的」可以直接比较(与 `events` / `documents` 同一手法)。
sqliteTimeLayout = "2006-01-02T15:04:05.000"
// recordTimeout 是**单次落盘**的时长上界。写入发生在 `EndRound`(会话收尾)上,
// 必须有一个界:一次库卡死绝不允许把收尾挂住。
recordTimeout = 5 * time.Second
// flowgraphLogSource 是全部日志的 source 标签(AGENTS §10:禁无来源裸日志)。
flowgraphLogSource = "runtime/flowgraph"
// DefaultTurnsLimit / MaxTurnsLimit 是轮次列表的缺省页大小与硬上界
// (超过即钳住,绝不静默返回全表)。多轮历史回放看的是「最近几轮」,
// 因此缺省比事件列表小。
DefaultTurnsLimit = 50
MaxTurnsLimit = 200
)
// Config 是链路图存储的接线参数。时间与开关口径由调用方(组合根)给出:本包**不读
// 环境变量**、不猜数据目录布局(与 `events.Config` / `documents.GCConfig` 同一理由)。
type Config struct {
// DB 是活连接(组合根注入既有 SQLite 连接池;测试注入临时库)。为 nil 时
// NewStore 返回 ErrNilDB。
DB *sql.DB
// Log 是基础 logger;nil 时替换为标记来源 runtime/flowgraph 的丢弃 logger。
Log *slog.Logger
// Now 是可注入时钟(测试用);nil 时用 time.Now。
Now func() time.Time
// GCInterval 是保留期清扫的间隔(`FLOW_GRAPH_GC_INTERVAL`)。**≤0 = 关闭 GC**
// (含启动期清扫),默认 DefaultGCInterval。
GCInterval time.Duration
// Retention 是 `flow_graphs` 的保留期(`FLOW_GRAPH_RETENTION`)。**≤0 = 关闭 GC**,
// 默认 DefaultRetention。0 刻意**不**解释成「立即清扫」:那会让一次笔误把刚落盘的
// 历史轮清空。
Retention time.Duration
}
// SQLStore 是基于 SQLite 的链路图存储实现。它可安全并发使用:连接池本身串行化写
// (`db.Open` 的 SetMaxOpenConns(1))。
type SQLStore struct {
db *sql.DB
log *slog.Logger
now func() time.Time
gcInterval time.Duration
retention time.Duration
gcEnabled bool
}
// Turn 是 `flow_graphs` 里一轮的**摘要**(供面板选历史轮),不含图本身。
type Turn struct {
// TurnID 是轮次标识(`trace.FlowGraph.ID`,形如 "web-7")。
TurnID string
// CreatedAt 是该轮落盘时刻(UTC)。
CreatedAt time.Time
}
// NewStore 构造链路图存储。唯一的失败是 DB 为 nil(ErrNilDB);GC 的关闭**不是**
// 构造失败,它是可见的正常状态(Enabled() 为 false + 一条记明原因的 WARN)。
func NewStore(cfg Config) (*SQLStore, error) {
if cfg.DB == nil {
return nil, fmt.Errorf("flowgraph: new store: %w", ErrNilDB)
}
log := sourcedLog(cfg.Log)
now := cfg.Now
if now == nil {
now = time.Now
}
s := &SQLStore{
db: cfg.DB,
log: log,
now: now,
gcInterval: cfg.GCInterval,
retention: cfg.Retention,
}
switch {
case cfg.GCInterval <= 0:
log.Warn("flow graph gc disabled: interval <= 0", "interval", cfg.GCInterval.String())
case cfg.Retention <= 0:
log.Warn("flow graph gc disabled: retention <= 0", "retention", cfg.Retention.String())
default:
s.gcEnabled = true
}
log.Info("flow graph store ready",
"gc_enabled", s.gcEnabled,
"gc_interval", s.gcInterval.String(),
"retention", s.retention.String())
return s, nil
}
// sourcedLog 把基础 logger 标上 source=runtime/flowgraph;nil 时给出丢弃 logger,
// 使调用方永远可以安全记日志(与 `events` / `documents` 同一手法)。
func sourcedLog(l *slog.Logger) *slog.Logger {
if l == nil {
l = slog.New(slog.NewTextHandler(io.Discard, nil))
}
return logger.WithSource(l, flowgraphLogSource)
}
// upsertRoundSQL 是落盘的唯一语句:列按迁移 0022 显式列出、值全部参数化。冲突键是
// `(session_id, turn_id)`(迁移里的唯一索引承载)—— 同一会话同一轮的二次收尾是同轮
// 覆盖(更新图与 updated_at,**不动 created_at**),而不是插出两行「同一轮」。
const upsertRoundSQL = `INSERT INTO flow_graphs
(session_id, turn_id, graph_json, created_at, updated_at)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(session_id, turn_id) DO UPDATE SET
graph_json = excluded.graph_json,
updated_at = excluded.updated_at`
// RecordRound 把一条**已收尾轮次**的图落盘(每轮一行)。它是发布器 `EndRound` 的落盘
// 出口,因此签名与 `loop.RoundRecorder` 一致:**不接收 ctx、自己用一个有界的后台
// context**(会话收尾路径不该被库拖住)。
//
// 失败**上抛**给调用方(发布器记 WARN 后继续),绝不 panic、绝不改变图 —— 图是可视化
// 资产,落盘失败不该影响会话本身。
func (s *SQLStore) RecordRound(g *trace.FlowGraph) error {
if s == nil || s.db == nil {
return fmt.Errorf("flowgraph: record round: %w", ErrNilDB)
}
if g == nil || strings.TrimSpace(g.SessionID) == "" || strings.TrimSpace(g.ID) == "" {
return fmt.Errorf("flowgraph: record round(session=%q turn=%q): %w", sessionOf(g), idOf(g), ErrNoGraphID)
}
payload, err := json.Marshal(g)
if err != nil {
return fmt.Errorf("flowgraph: 编码轮次图(%s/%s): %w", g.SessionID, g.ID, err)
}
ctx, cancel := context.WithTimeout(context.Background(), recordTimeout)
defer cancel()
stamp := sqliteTimeStamp(s.now())
if _, err := s.db.ExecContext(ctx, upsertRoundSQL,
g.SessionID, g.ID, string(payload), stamp, stamp); err != nil {
return fmt.Errorf("flowgraph: 落轮次图(%s/%s): %w", g.SessionID, g.ID, err)
}
return nil
}
// Graph 按 `(sessionID, turnID)` 取回一轮的图。该轮不存在时返回 `(nil, nil)` —— 这是
// 一个**诚实的空态**(不是错误):客户端必须据此渲染空态,绝不允许补画任何节点/假图。
func (s *SQLStore) Graph(ctx context.Context, sessionID, turnID string) (*trace.FlowGraph, error) {
if s == nil || s.db == nil {
return nil, fmt.Errorf("flowgraph: query round: %w", ErrNilDB)
}
const query = `SELECT graph_json FROM flow_graphs WHERE session_id = ? AND turn_id = ?`
var payload string
err := s.db.QueryRowContext(ctx, query, sessionID, turnID).Scan(&payload)
if errors.Is(err, sql.ErrNoRows) {
return nil, nil
}
if err != nil {
return nil, fmt.Errorf("flowgraph: 查轮次图(%s/%s): %w", sessionID, turnID, err)
}
return decodeGraph(sessionID, turnID, payload)
}
// Latest 取该会话**最近一轮**已落盘的图(按 created_at 倒序,同毫秒用 id 定序)。
// 没有已落盘的轮次时返回 `(nil, nil)`(诚实空态)。
func (s *SQLStore) Latest(ctx context.Context, sessionID string) (*trace.FlowGraph, error) {
if s == nil || s.db == nil {
return nil, fmt.Errorf("flowgraph: query latest round: %w", ErrNilDB)
}
const query = `SELECT turn_id, graph_json FROM flow_graphs
WHERE session_id = ? ORDER BY created_at DESC, id DESC LIMIT 1`
var (
turnID string
payload string
)
err := s.db.QueryRowContext(ctx, query, sessionID).Scan(&turnID, &payload)
if errors.Is(err, sql.ErrNoRows) {
return nil, nil
}
if err != nil {
return nil, fmt.Errorf("flowgraph: 查最近轮次图(%s): %w", sessionID, err)
}
return decodeGraph(sessionID, turnID, payload)
}
// Turns 返回该会话**已落盘轮次的摘要**,最新在前(供面板选历史轮)。limit ≤0 取
// DefaultTurnsLimit,超过 MaxTurnsLimit 即钳住。
func (s *SQLStore) Turns(ctx context.Context, sessionID string, limit int) ([]Turn, error) {
if s == nil || s.db == nil {
return nil, fmt.Errorf("flowgraph: query turns: %w", ErrNilDB)
}
n := normalizeLimit(limit)
const query = `SELECT turn_id, created_at FROM flow_graphs
WHERE session_id = ? ORDER BY created_at DESC, id DESC LIMIT ?`
rows, err := s.db.QueryContext(ctx, query, sessionID, n)
if err != nil {
return nil, fmt.Errorf("flowgraph: 查轮次列表(%s): %w", sessionID, err)
}
defer func() { _ = rows.Close() }()
out := make([]Turn, 0, n)
for rows.Next() {
var (
turnID string
createdAt string
)
if err := rows.Scan(&turnID, &createdAt); err != nil {
return nil, fmt.Errorf("flowgraph: 读轮次行(%s): %w", sessionID, err)
}
at, ok := parseSQLiteTime(createdAt)
if !ok {
return nil, fmt.Errorf("flowgraph: 轮次 %s 的 created_at 不可解析(%q): %w", turnID, createdAt, ErrBadTimestamp)
}
out = append(out, Turn{TurnID: turnID, CreatedAt: at})
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("flowgraph: 遍历轮次行(%s): %w", sessionID, err)
}
return out, nil
}
// decodeGraph 把一行 `graph_json` 解析回 `trace.FlowGraph`(与 `graph_update` 帧同形)。
func decodeGraph(sessionID, turnID, payload string) (*trace.FlowGraph, error) {
var g trace.FlowGraph
if err := json.Unmarshal([]byte(payload), &g); err != nil {
return nil, fmt.Errorf("flowgraph: 解析轮次图 JSON(%s/%s): %w", sessionID, turnID, err)
}
return &g, nil
}
// normalizeLimit 把轮次列表页大小归一:≤0 取缺省、超过硬上界即钳住。
func normalizeLimit(limit int) int {
switch {
case limit <= 0:
return DefaultTurnsLimit
case limit > MaxTurnsLimit:
return MaxTurnsLimit
default:
return limit
}
}
// sqliteTimeStamp 把时刻格式化成库里的时间戳口径(UTC 毫秒 + 字面量 Z)。
func sqliteTimeStamp(t time.Time) string {
return t.UTC().Format(sqliteTimeLayout) + "Z"
}
// parseSQLiteTime 解析库里的时间戳列。它先试本包写出的口径,再退回 RFC3339
// (覆盖带偏移或非毫秒精度的历史值);两者都不成立时返回 false,调用方据此保守
// 处理(绝不把读不懂的时间当成一个真实时刻)。
func parseSQLiteTime(raw string) (time.Time, bool) {
trimmed := strings.TrimSpace(raw)
if trimmed == "" {
return time.Time{}, false
}
if t, err := time.Parse(sqliteTimeLayout+"Z", trimmed); err == nil {
return t.UTC(), true
}
if t, err := time.Parse(time.RFC3339Nano, trimmed); err == nil {
return t.UTC(), true
}
return time.Time{}, false
}
// sessionOf / idOf 是为**错误信息**安全取标识的助手:g 为 nil 时也不 panic
// (RecordRound 的校验顺序里要先能报出「缺了哪个坐标」)。
func sessionOf(g *trace.FlowGraph) string {
if g == nil {
return ""
}
return g.SessionID
}
func idOf(g *trace.FlowGraph) string {
if g == nil {
return ""
}
return g.ID
}
@@ -0,0 +1,317 @@
package flowgraph
// 本文件是**链路图落盘存储**的验收(design/19 §4 P3;切片 F2)。断言的是**库里的事实**
// (行在不在、读回来的图是不是原图、每轮一行),不按日志文本断言 —— 与
// `internal/runtime/events` 的测试同一口径。
//
// 核心一条(主人原话的内核):「后端重启后仍能查回历史轮次的图」——
// `TestStore_SurvivesRestart_正常` 用**同一个文件库、两条连接、两个 store** 模拟
// 「进程重启」,第二个 store 内存里没有任何图,只能靠落盘读回。
import (
"context"
"database/sql"
"errors"
"path/filepath"
"testing"
"time"
"git.yeij.top/AskaEth/Cyrene_v2/internal/db"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
)
// testNow 是夹具的固定时钟(可复现:所有时间断言都基于它)。
func testNow() time.Time {
return time.Date(2026, 10, 5, 12, 0, 0, 0, time.UTC)
}
// newTestStore 建一套隔离夹具:一个跑过迁移的内存库 + 一个把时钟钉在 testNow 的
// 存储。默认 **关闭 GC**(存储用例与保留期无关,GC 由 gc_test.go 专门覆盖),
// tune 可覆盖任何配置项。第二个返回值是底层连接,供用例做结构级断言。
func newTestStore(t *testing.T, tune func(*Config)) (*SQLStore, *sql.DB) {
t.Helper()
ctx := context.Background()
conn, err := db.Open(ctx, ":memory:", nil)
if err != nil {
t.Fatalf("打开内存库: %v", err)
}
if err := db.MigrateUp(ctx, conn, nil); err != nil {
_ = conn.Close()
t.Fatalf("内存库跑迁移: %v", err)
}
t.Cleanup(func() { _ = conn.Close() })
cfg := Config{DB: conn, Now: testNow, GCInterval: 0, Retention: 0}
if tune != nil {
tune(&cfg)
}
st, err := NewStore(cfg)
if err != nil {
t.Fatalf("NewStore: %v", err)
}
return st, conn
}
// sampleGraph 造一轮最小图(input→llm→output),轮次标识 = "<session>-<round>",
// 与 `trace.RunGraph.Snapshot` 的命名逐字同形。
func sampleGraph(session string, round int) *trace.FlowGraph {
return &trace.FlowGraph{
ID: session + "-" + itoa(round),
SessionID: session,
StartedAt: testNow().Add(-time.Minute),
EndedAt: testNow(),
Nodes: []trace.FlowNode{
{ID: "input", Label: "你好", Kind: "input", Status: trace.FlowStatusDone},
{ID: "llm-1", Label: "思考", Kind: "llm", Status: trace.FlowStatusDone},
{ID: "output", Label: "回复", Kind: "output", Status: trace.FlowStatusDone},
},
Edges: []trace.FlowEdge{{From: "input", To: "llm-1"}, {From: "llm-1", To: "output"}},
}
}
// itoa 是个极小的十进制转换(避免为一个测试引入 fmt 的字符串依赖)。
func itoa(n int) string {
if n == 0 {
return "0"
}
digits := make([]byte, 0, 4)
for n > 0 {
digits = append([]byte{byte('0' + n%10)}, digits...)
n /= 10
}
return string(digits)
}
// TestNewStore_NilDB_异常:没有数据库就没有存储 —— 宁可明确失败,也不给一个写不进去
// 的替身(与 events / documents 同一口径)。
func TestNewStore_NilDB_异常(t *testing.T) {
t.Parallel()
if _, err := NewStore(Config{}); !errors.Is(err, ErrNilDB) {
t.Fatalf("NewStore(nil DB) err = %v, want ErrNilDB", err)
}
}
// TestRecordRound_ReadBack_正常 钉住「写进去什么、按轮次读回来就是什么」:Latest 回
// 最新一轮、Graph 按 turnId 精确取,且主干端点与边都完整往返。
func TestRecordRound_ReadBack_正常(t *testing.T) {
t.Parallel()
ctx := context.Background()
st, _ := newTestStore(t, nil)
for round := 1; round <= 3; round++ {
if err := st.RecordRound(sampleGraph("web", round)); err != nil {
t.Fatalf("RecordRound(round=%d): %v", round, err)
}
}
latest, err := st.Latest(ctx, "web")
if err != nil {
t.Fatalf("Latest: %v", err)
}
if latest == nil || latest.ID != "web-3" {
t.Fatalf("Latest = %+v, want web-3", latest)
}
got, err := st.Graph(ctx, "web", "web-2")
if err != nil {
t.Fatalf("Graph(web-2): %v", err)
}
if got == nil || got.ID != "web-2" {
t.Fatalf("Graph(web-2) = %+v, want web-2", got)
}
if len(got.Nodes) != 3 || len(got.Edges) != 2 {
t.Fatalf("round-trip lost structure: nodes=%d edges=%d", len(got.Nodes), len(got.Edges))
}
if got.SessionID != "web" {
t.Fatalf("SessionID = %q, want web(图自带会话,前端按它键入桶)", got.SessionID)
}
}
// TestGraph_MissingIsHonestNil_边界 钉住「该轮不存在」是**诚实的空**(nil, nil),
// 不是错误 —— 客户端据此渲染空态,绝不补画假图。
func TestGraph_MissingIsHonestNil_边界(t *testing.T) {
t.Parallel()
ctx := context.Background()
st, _ := newTestStore(t, nil)
if err := st.RecordRound(sampleGraph("web", 1)); err != nil {
t.Fatalf("RecordRound: %v", err)
}
for _, turn := range []string{"web-999", "sess-backthink-1"} {
got, err := st.Graph(ctx, "web", turn)
if err != nil {
t.Fatalf("Graph(%s) err = %v, want nil(缺席不是错误)", turn, err)
}
if got != nil {
t.Fatalf("Graph(%s) = %+v, want nil", turn, got)
}
}
latest, err := st.Latest(ctx, "sess-backthink")
if err != nil {
t.Fatalf("Latest(unknown session) err = %v, want nil", err)
}
if latest != nil {
t.Fatalf("Latest(unknown session) = %+v, want nil", latest)
}
}
// TestRecordRound_SameTurnMerges_正常 钉住唯一键:同一会话同一轮二次收尾是**同轮覆盖**
// (一行、取后者),绝不插出两行「同一轮」。
func TestRecordRound_SameTurnMerges_正常(t *testing.T) {
t.Parallel()
ctx := context.Background()
st, conn := newTestStore(t, nil)
first := sampleGraph("web", 1)
if err := st.RecordRound(first); err != nil {
t.Fatalf("record first: %v", err)
}
second := sampleGraph("web", 1)
second.Nodes = second.Nodes[:1] // 第二次少两个节点:读回来必须看到「后写的那份」
if err := st.RecordRound(second); err != nil {
t.Fatalf("record second: %v", err)
}
var rows int
if err := conn.QueryRowContext(ctx,
`SELECT count(*) FROM flow_graphs WHERE session_id = ? AND turn_id = ?`, "web", "web-1").Scan(&rows); err != nil {
t.Fatalf("count rows: %v", err)
}
if rows != 1 {
t.Fatalf("rows = %d, want 1 (unique key must merge, not duplicate)", rows)
}
got, err := st.Graph(ctx, "web", "web-1")
if err != nil {
t.Fatalf("Graph: %v", err)
}
if got == nil || len(got.Nodes) != 1 {
t.Fatalf("merged round = %+v, want the second (1 node)", got)
}
}
// TestRecordRound_MissingID_异常 钉住缺坐标是显式的编程错误(ErrNoGraphID),
// 绝不写一行「不知道是哪一轮」的记录进库。
func TestRecordRound_MissingID_异常(t *testing.T) {
t.Parallel()
st, _ := newTestStore(t, nil)
cases := map[string]*trace.FlowGraph{
"nil graph": nil,
"no session": {ID: "web-1"},
"no turn id": {SessionID: "web"},
"blank turn id": {SessionID: "web", ID: " "},
"blank session": {SessionID: " ", ID: "web-1"},
}
for name, g := range cases {
if err := st.RecordRound(g); !errors.Is(err, ErrNoGraphID) {
t.Fatalf("%s: err = %v, want ErrNoGraphID", name, err)
}
}
}
// TestTurns_NewestFirstAndClamped_正常 钉住轮次列表的排序口径(最新在前)与分页钳制:
// 面板据此选历史轮,顺序错了就等于把「最近几轮」讲反。
func TestTurns_NewestFirstAndClamped_正常(t *testing.T) {
t.Parallel()
ctx := context.Background()
// 时钟递增,保证 created_at 严格不同(排序可复现)。
base := testNow()
clock := base
st, _ := newTestStore(t, func(c *Config) { c.Now = func() time.Time { return clock } })
for round := 1; round <= 4; round++ {
clock = base.Add(time.Duration(round) * time.Minute)
if err := st.RecordRound(sampleGraph("web", round)); err != nil {
t.Fatalf("RecordRound(round=%d): %v", round, err)
}
}
turns, err := st.Turns(ctx, "web", 0)
if err != nil {
t.Fatalf("Turns: %v", err)
}
if len(turns) != 4 {
t.Fatalf("turns = %d, want 4", len(turns))
}
want := []string{"web-4", "web-3", "web-2", "web-1"}
for i, w := range want {
if turns[i].TurnID != w {
t.Fatalf("turns[%d] = %q, want %q(最新在前)", i, turns[i].TurnID, w)
}
}
// 负/零取缺省;超上限被钳住(这里给 2 取 2)。
limited, err := st.Turns(ctx, "web", 2)
if err != nil {
t.Fatalf("Turns(limit=2): %v", err)
}
if len(limited) != 2 || limited[0].TurnID != "web-4" {
t.Fatalf("Turns(limit=2) = %+v, want newest 2", limited)
}
if MaxTurnsLimit < 3 || normalizeLimit(MaxTurnsLimit+1000) != MaxTurnsLimit {
t.Fatalf("normalizeLimit must clamp to MaxTurnsLimit=%d", MaxTurnsLimit)
}
if normalizeLimit(0) != DefaultTurnsLimit || normalizeLimit(-5) != DefaultTurnsLimit {
t.Fatalf("normalizeLimit(≤0) must fall back to DefaultTurnsLimit=%d", DefaultTurnsLimit)
}
}
// TestStore_SurvivesRestart_正常 是**本切片的内核**(主人要求):后端重启后仍能查回
// 历史轮次的图。用同一个**文件库**、先后两条连接 + 两个 store 模拟进程重启 —— 第二个
// store 内存里没有任何图,唯一能拿到 web-3 的方式就是从库里读回。
func TestStore_SurvivesRestart_正常(t *testing.T) {
t.Parallel()
ctx := context.Background()
path := filepath.Join(t.TempDir(), "cyrene.db")
// 第一个「进程」:建库、迁移、落一轮。
conn1, err := db.Open(ctx, path, nil)
if err != nil {
t.Fatalf("open db (first): %v", err)
}
if err := db.MigrateUp(ctx, conn1, nil); err != nil {
_ = conn1.Close()
t.Fatalf("migrate (first): %v", err)
}
st1, err := NewStore(Config{DB: conn1, Now: testNow})
if err != nil {
_ = conn1.Close()
t.Fatalf("NewStore (first): %v", err)
}
if err := st1.RecordRound(sampleGraph("web", 3)); err != nil {
_ = conn1.Close()
t.Fatalf("RecordRound: %v", err)
}
_ = conn1.Close() // 模拟进程退出
// 第二个「进程」:新连接、新 store(内存里没有任何图),只能靠落盘。
conn2, err := db.Open(ctx, path, nil)
if err != nil {
t.Fatalf("open db (second): %v", err)
}
t.Cleanup(func() { _ = conn2.Close() })
if err := db.MigrateUp(ctx, conn2, nil); err != nil {
t.Fatalf("migrate (second): %v", err)
}
st2, err := NewStore(Config{DB: conn2, Now: testNow})
if err != nil {
t.Fatalf("NewStore (second): %v", err)
}
got, err := st2.Graph(ctx, "web", "web-3")
if err != nil {
t.Fatalf("Graph after restart: %v", err)
}
if got == nil || got.ID != "web-3" {
t.Fatalf("重启后查回 = %+v, want web-3(这正是主人要的「本地数据」)", got)
}
latest, err := st2.Latest(ctx, "web")
if err != nil {
t.Fatalf("Latest after restart: %v", err)
}
if latest == nil || latest.ID != "web-3" {
t.Fatalf("重启后 Latest = %+v, want web-3", latest)
}
turns, err := st2.Turns(ctx, "web", 0)
if err != nil {
t.Fatalf("Turns after restart: %v", err)
}
if len(turns) != 1 || turns[0].TurnID != "web-3" {
t.Fatalf("重启后 turns = %+v, want [web-3]", turns)
}
}
+119
View File
@@ -0,0 +1,119 @@
package flowgraph
// 本文件归属**保留期清扫(Sweep)**(切片 F2)。它逐字沿用 `internal/runtime/events/gc.go`
// 的既有 GC 范式(design/19 §4 P3 的落盘件照 `events` / `documents` 口径,不另造):
//
// 1. **启动期扫一次、此后按间隔定时扫**(间隔默认 DefaultGCInterval)。启动期那次
// 与定时循环都**不**阻塞启动、不绑在磁盘上 —— 常驻循环住在组合根
// (`cmd/cyrene/wire_flow_graphs.go` 逐字对齐 `wire_channel_events.go`),本包只提供
// `Sweep` 与它需要的三个口径(Enabled / Interval / Retention)。
// 2. **失败只记 WARN、绝不让进程退出**:`Sweep` 把查询/删除失败**上抛**给调用方
// (组合根记 WARN 后继续等下一轮);已经删掉的行不会因为后续批次失败而「回滚」,
// 未删的行也不会被清空 —— 一轮失败就只是「这一轮没扫干净」。
// 3. **≤0 = 关闭**(显式),关闭原因在 NewStore 里记进日志 —— 「配了却不生效」不存在
// (与文档/事件 GC 同口径)。
//
// 清理条件只有一条:`created_at < now − 保留期` 的 `flow_graphs` 行逐批删除。落盘件本身
// **没有**「不参与 GC」的例外:图是可视化资产,过期即可回收(没有它,库会无限增长)。
import (
"context"
"fmt"
"time"
)
const (
// DefaultGCInterval 是 `FLOW_GRAPH_GC_INTERVAL` 的默认值:每小时清扫一次。
DefaultGCInterval = time.Hour
// DefaultRetention 是 `FLOW_GRAPH_RETENTION` 的默认值:7 天(168h,与文档保留期同值)。
// 图的用途是「回看最近若干轮」,不是事后定案证据(后者是 channel_events 的 720h 口径)——
// 一周足以覆盖「翻前几轮」,也把每轮一份的图 JSON 挡在可行的体积内。
DefaultRetention = 168 * time.Hour
// sweepBatchSize 是单批删除的行数上限;maxSweepBatches 是一轮清扫的批数上限。
// 两者一起把「一轮清扫」的时长与单次内存占用**有界**(AGENTS §10 的压测纪律:
// 有界,不靠堆进程):积压时不是一次 DELETE 全表,而是每轮最多
// maxSweepBatches × sweepBatchSize 行,剩下的留给下一轮。
sweepBatchSize = 500
maxSweepBatches = 100
)
// SweepStats 是一轮清扫的汇总(成功件只记总数,绝不记任何行内容 / 图内容)。
type SweepStats struct {
// Batches 是实际执行的删除批数。
Batches int
// Deleted 是删除的过期行数。
Deleted int64
}
// Enabled 报告清扫是否真的会删行(保留期与间隔都 > 0)。
func (s *SQLStore) Enabled() bool { return s != nil && s.gcEnabled }
// Interval 返回定时清扫间隔(关闭时为 0)。
func (s *SQLStore) Interval() time.Duration {
if s == nil {
return 0
}
return s.gcInterval
}
// Retention 返回轮次图保留期(关闭时为 0)。
func (s *SQLStore) Retention() time.Duration {
if s == nil {
return 0
}
return s.retention
}
// sweepSQL 按时间戳删一批过期行。子查询带 `ORDER BY created_at LIMIT ?`,因此它
// 吃得上 `idx_flow_graphs_created_at`,且每批的扫描面有界。时间戳用定宽 UTC 字符串
// 比较:字典序等价于时间序(见 flowgraph.go 的口径注释)。
const sweepSQL = `DELETE FROM flow_graphs WHERE id IN (
SELECT id FROM flow_graphs WHERE created_at < ? ORDER BY created_at LIMIT ?)`
// Sweep 按保留期清扫 `flow_graphs` 的过期行,返回一轮汇总。
//
// 关闭时(Enabled() 为 false)是**空操作**:返回零值与 nil 错误、不碰任何一行。这与
// 「配了却不生效」是两回事 —— 关闭原因在 NewStore 里已记 WARN。
//
// 失败语义:查询/删除失败返回错误并**带上已完成的统计**(调用方据此知道这一轮走到了
// 哪一步,记 WARN 后等下一轮)。ctx 取消时立刻停下:**已删的保持删除、未删的绝不因此
// 被清空**。
func (s *SQLStore) Sweep(ctx context.Context) (SweepStats, error) {
if !s.Enabled() {
return SweepStats{}, nil
}
cutoff := sqliteTimeStamp(s.now().Add(-s.retention))
var stats SweepStats
for i := 0; i < maxSweepBatches; i++ {
if err := ctx.Err(); err != nil {
return stats, fmt.Errorf("flowgraph: sweep: %w", err)
}
deleted, err := s.sweepBatch(ctx, cutoff)
if err != nil {
return stats, err
}
stats.Batches++
stats.Deleted += deleted
if deleted < int64(sweepBatchSize) {
break
}
}
s.log.Info("flow graph sweep",
"deleted", stats.Deleted, "batches", stats.Batches, "retention", s.retention.String())
return stats, nil
}
// sweepBatch 删一批过期行并返回本批删掉的行数。
func (s *SQLStore) sweepBatch(ctx context.Context, cutoff string) (int64, error) {
res, err := s.db.ExecContext(ctx, sweepSQL, cutoff, sweepBatchSize)
if err != nil {
return 0, fmt.Errorf("flowgraph: sweep 删过期行: %w", err)
}
n, err := res.RowsAffected()
if err != nil {
return 0, fmt.Errorf("flowgraph: sweep 读取删除行数: %w", err)
}
return n, nil
}
+144
View File
@@ -0,0 +1,144 @@
package flowgraph
// 本文件是**保留期清扫**的验收(切片 F2)。断言的是库里的事实(哪些行还在),
// 与 `internal/runtime/events/gc_test.go` 同一口径。
import (
"context"
"testing"
"time"
)
// TestSweep_DisabledIsNoOp_边界 钉住「配了 ≤0 = 关闭 GC」:`Sweep` 是**空操作**,
// 一行都不删。0 刻意不解释成「立即清空」—— 那会让一次笔误抹掉刚落盘的历史轮。
func TestSweep_DisabledIsNoOp_边界(t *testing.T) {
t.Parallel()
ctx := context.Background()
for _, tc := range []struct {
name string
gcInterval, retain time.Duration
}{
{"interval=0", 0, time.Hour},
{"interval<0", -time.Hour, time.Hour},
{"retention=0", time.Hour, 0},
{"retention<0", time.Hour, -time.Hour},
} {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
st, _ := newTestStore(t, func(c *Config) {
c.GCInterval = tc.gcInterval
c.Retention = tc.retain
})
if st.Enabled() {
t.Fatalf("Enabled() = true, want false (%s)", tc.name)
}
if err := st.RecordRound(sampleGraph("web", 1)); err != nil {
t.Fatalf("RecordRound: %v", err)
}
stats, err := st.Sweep(ctx)
if err != nil {
t.Fatalf("Sweep: %v", err)
}
if stats.Deleted != 0 || stats.Batches != 0 {
t.Fatalf("disabled Sweep = %+v, want zero", stats)
}
turns, err := st.Turns(ctx, "web", 0)
if err != nil {
t.Fatalf("Turns: %v", err)
}
if len(turns) != 1 {
t.Fatalf("行被删了(turns=%d),但 GC 关闭时必须是空操作", len(turns))
}
})
}
}
// TestSweep_DeletesOnlyExpired_正常 钉住「正数只删超期」:保留期内的行**必须**留下,
// 只有 `created_at < now − 保留期` 的行被回收。时钟注入让这条判定可复现。
func TestSweep_DeletesOnlyExpired_正常(t *testing.T) {
t.Parallel()
ctx := context.Background()
base := testNow()
clock := base
st, _ := newTestStore(t, func(c *Config) {
c.GCInterval = time.Hour
c.Retention = 24 * time.Hour
c.Now = func() time.Time { return clock }
})
if !st.Enabled() {
t.Fatal("Enabled() = false, want true(间隔与保留期都 > 0)")
}
// 旧轮:T0 落盘。
if err := st.RecordRound(sampleGraph("web", 1)); err != nil {
t.Fatalf("record old round: %v", err)
}
// 时钟拨到 T0+48h,落一个新轮。
clock = base.Add(48 * time.Hour)
if err := st.RecordRound(sampleGraph("web", 2)); err != nil {
t.Fatalf("record new round: %v", err)
}
// cutoff = (T0+48h) − 24h = T0+24h ⇒ web-1(T0)超期、web-2(T0+48h)留下。
stats, err := st.Sweep(ctx)
if err != nil {
t.Fatalf("Sweep: %v", err)
}
if stats.Deleted != 1 {
t.Fatalf("Deleted = %d, want 1(只删超期的那一轮)", stats.Deleted)
}
turns, err := st.Turns(ctx, "web", 0)
if err != nil {
t.Fatalf("Turns: %v", err)
}
if len(turns) != 1 || turns[0].TurnID != "web-2" {
t.Fatalf("turns after sweep = %+v, want [web-2](保留期内的那轮必须留下)", turns)
}
if _, err := st.Graph(ctx, "web", "web-1"); err != nil {
t.Fatalf("Graph(被删轮) err = %v, want nil(缺席不是错误)", err)
}
old, err := st.Graph(ctx, "web", "web-1")
if err != nil {
t.Fatalf("Graph(web-1): %v", err)
}
if old != nil {
t.Fatalf("web-1 应已被回收,却仍读得到: %+v", old)
}
}
// TestSweep_ScopedPerSession_正常 钉住清扫是**全表按时间**的(不是按会话):过期的
// 会话行照样回收,不会因为另一个会话还有新轮就整表留着。
func TestSweep_ScopedPerSession_正常(t *testing.T) {
t.Parallel()
ctx := context.Background()
base := testNow()
clock := base
st, _ := newTestStore(t, func(c *Config) {
c.GCInterval = time.Hour
c.Retention = time.Hour
c.Now = func() time.Time { return clock }
})
if err := st.RecordRound(sampleGraph("web", 1)); err != nil {
t.Fatalf("record web: %v", err)
}
clock = base.Add(3 * time.Hour)
if err := st.RecordRound(sampleGraph("sess-backthink", 1)); err != nil {
t.Fatalf("record backthink: %v", err)
}
if _, err := st.Sweep(ctx); err != nil {
t.Fatalf("Sweep: %v", err)
}
webTurns, err := st.Turns(ctx, "web", 0)
if err != nil {
t.Fatalf("Turns(web): %v", err)
}
if len(webTurns) != 0 {
t.Fatalf("web turns = %+v, want empty(3h 前的轮已超 1h 保留期)", webTurns)
}
btTurns, err := st.Turns(ctx, "sess-backthink", 0)
if err != nil {
t.Fatalf("Turns(backthink): %v", err)
}
if len(btTurns) != 1 {
t.Fatalf("backthink turns = %+v, want [sess-backthink-1]", btTurns)
}
}
+74 -3
View File
@@ -1,9 +1,11 @@
package loop
import (
"log/slog"
"sync"
"time"
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
)
@@ -13,11 +15,41 @@ import (
// 作为 WS 事件类型 graph_update 转发(design/19 §1)。
const graphUpdateSubject = "ws.graph_update"
// flowLogSource 是发布器日志的 source 标签(落盘失败记 WARN;AGENTS §10 禁裸日志)。
const flowLogSource = "runtime/loop/flow"
// allowedSessions 是用于控制哪些会话可发布
// flow 图的谓词签名(主人的 web 会话),使 IM/陌生人/子代理轮次绝不
// 泄露进管理面板。谓词为 nil 时允许所有会话。
type allowedSessions func(sessionID string) bool
// RoundRecorder 把一条**已收尾轮次**的图落盘(design/19 §4 的 P3「多轮历史回放」;
// 切片 F2)。生产实现是 `internal/runtime/flowgraph` 的 `*SQLStore`。
//
// 语义约定(实现方必须遵守):**每轮一条**(只在 EndRound 里调用,绝不在每个 span
// 上调用),失败**上抛**给发布器(发布器记 WARN 后继续)—— 图是可视化资产,落盘失败
// 绝不允许影响会话收尾,也绝不改变实时帧。
//
// 白名单不在实现方:EndRound 对发布会话范围外的会话**直接返回**,因此落盘出口天然
// 只被白名单内的会话触发(复用 flowSessionAllowed,不存在第二份白名单)。
type RoundRecorder interface {
RecordRound(g *trace.FlowGraph) error
}
// FlowPublisherOption 配置 FlowPublisher 的可选行为(落盘出口 + 日志)。
type FlowPublisherOption func(*FlowPublisher)
// WithRoundRecorder 挂一个「每轮收尾时落一条」的落盘出口。传 nil 等价于不落盘。
func WithRoundRecorder(rec RoundRecorder) FlowPublisherOption {
return func(p *FlowPublisher) { p.record = rec }
}
// WithFlowLogger 给发布器一个 logger:落盘失败要**可见**(记 WARN),但绝不把图内容
// 写进日志(图里可能含用户输入文本,只记 session/turn 标识)。
func WithFlowLogger(log *slog.Logger) FlowPublisherOption {
return func(p *FlowPublisher) { p.log = log }
}
// FlowPublisher 从循环的 trace span 累积每会话 flow 图,
// 并在每个 span 开始/结束时发布 ws.graph_update,给前端
// 一张实时的 input→think→tool→output 图(design/19 §1)。它容忍 nil
@@ -34,19 +66,32 @@ type FlowPublisher struct {
allow allowedSessions
graphs map[string]*trace.RunGraph
rounds map[string]int
// record 是每轮收尾的落盘出口(design/19 §4 P3;切片 F2)。nil = 不落盘
// (保持 F1 之前的纯内存行为,且测试可只测发布)。
record RoundRecorder
// log 用于把落盘失败记成 WARN(带 source)。nil 时静默(测试/未接线)。
log *slog.Logger
}
// NewFlowPublisher 构建一个 FlowPublisher。publish 非 nil 时,会在
// 每次 span 状态迁移时以 graphUpdateSubject 与 *trace.FlowGraph
// 快照调用。allow 非 nil 时,控制哪些 session id 可发布(返回
// false 则整个会话被丢弃)。两者都可以为 nil(nil publish 为空操作)。
func NewFlowPublisher(publish func(subject string, payload any), allow func(sessionID string) bool) *FlowPublisher {
return &FlowPublisher{
//
// 可变项经 Functional Options 传入(不改变既有两参调用):落盘出口(F2)与日志。
func NewFlowPublisher(publish func(subject string, payload any), allow func(sessionID string) bool, opts ...FlowPublisherOption) *FlowPublisher {
p := &FlowPublisher{
publish: publish,
allow: allow,
graphs: make(map[string]*trace.RunGraph),
rounds: make(map[string]int),
}
for _, opt := range opts {
if opt != nil {
opt(p)
}
}
return p
}
// StartRound 为 sessionID 开始一张新图,用用户文本
@@ -72,18 +117,34 @@ func (p *FlowPublisher) StartRound(sessionID, inputText string) {
// **重放的出口是 `LastGraph`**(网关只读查询,契约 v0.71),而不是又发一帧:
// 实时帧向来只推给当时在线的连接,刷新/晚开的页面只能靠只读拉取补齐
// (`graph_update` 本身不重放)。
//
// 收尾也是**落盘**的唯一时机(切片 F2;design/19 §4 的 P3「多轮历史回放」):每轮落
// **一条**(绝不是每 span 一条),且只在该会话处于发布会话白名单内时发生(本方法对
// 白名单外的会话直接返回 ⇒ 落盘出口天然只被白名单会话触发)。深拷贝与写库在**放锁
// 之后**做:一次库写绝不把其他会话的 span 发布挡住;失败只记 WARN,绝不影响本轮收尾。
func (p *FlowPublisher) EndRound(sessionID, outputText string) {
if sessionID == "" || !p.allowed(sessionID) {
return
}
p.mu.Lock()
defer p.mu.Unlock()
g := p.graphs[sessionID]
if g == nil {
p.mu.Unlock()
return
}
g.SeedOutput(outputText)
p.publishLocked(g)
var snap *trace.FlowGraph
if p.record != nil {
snap = g.Snapshot().Clone()
}
p.mu.Unlock()
if snap != nil {
if err := p.record.RecordRound(snap); err != nil {
p.logPersistError(sessionID, snap.ID, err)
}
}
}
// MarkInputSteered 给本轮的 input 节点打上 `steered:true` 标记并发布一帧
@@ -188,6 +249,16 @@ func (p *FlowPublisher) allowed(sessionID string) bool {
return p.allow(sessionID)
}
// logPersistError 把一次落盘失败记成带来源的 WARN。它只记 session/turn 标识,
// **绝不打印图内容**(图里可能含用户输入文本);logger 未接线时静默。
func (p *FlowPublisher) logPersistError(sessionID, turnID string, err error) {
if p.log == nil {
return
}
logger.WithSource(p.log, flowLogSource).Warn("flow round persist failed",
"session", sessionID, "turn", turnID, "err", err)
}
// lastUserText 返回 messages 中最新用户消息的文本,用于
// 标注该轮的 input 节点。没有用户文本时返回 ""。
func lastUserText(messages []model.Message) string {
+138
View File
@@ -0,0 +1,138 @@
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 仍须正常发布")
}
}