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/臆造回退）。
348 lines
15 KiB
Go
348 lines
15 KiB
Go
// Package flowgraph 是**链路图本地落盘**的存储地基（design/19 §4 P3「多轮历史回放」；
|
||
// 切片 F2）。它补的是主人 2026-10-05 追加要求的那句「链路流程应该有本地数据，不然
|
||
// 每次重载之后页面就没数据了」—— 在此之前图只活在 `loop.FlowPublisher` 的**内存**里，
|
||
// 后端一重启、或该会话还没跑新轮次，就什么都查不到。
|
||
//
|
||
// 只做两件事：把**每一路的收尾图**（`trace.FlowGraph`）连会话与轮次标识落进
|
||
// `flow_graphs`（迁移 0022 / 0023），并支持按图 id 单轮读回、按**回合标识**批量读回
|
||
// （`GraphsByTurnID`，消息面回看工具调用的读侧入口）、按保留期清扫。
|
||
//
|
||
// 三条职责边界（别在这里越界）：
|
||
//
|
||
// - **不判身份、不判白名单**：哪些会话能产出图由发布器的发布会话范围
|
||
// （`cmd/cyrene` 的 `flowSessionAllowed`，design/19 §1 裁决 01.4.5）决定 ——
|
||
// 本包只被**已经过白名单的** `EndRound` 调用，因此不再存一份白名单。
|
||
// - **写入时机 = 每轮一条**（`EndRound`），**不是每 span 一条**：span 级写放大是纯浪费，
|
||
// 图在收尾时已完整（`output` 节点已并入）。真并行时一轮有多条路（design/15 §2.5.5），
|
||
// 因此实际是**每一路一条** —— 两路各有各的 `turn_id`，唯一键 `(session_id, turn_id)`
|
||
// 因此不会把其中一路顶上（FV2；design/15 §2.5.10-1）。
|
||
// - **不是图的真源**：实时图仍由发布器持有（`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 / 0023 显式列出、值全部参数化。
|
||
// 冲突键是 `(session_id, turn_id)`（迁移里的唯一索引承载）—— 同一会话同一轮的二次
|
||
// 收尾是同轮覆盖（更新图与 updated_at，**不动 created_at**），而不是插出两行
|
||
// 「同一轮」。
|
||
//
|
||
// `lane_turn_id`（迁移 0023）与 `turn_id` **不是同一个东西**：前者是这一**路**的权威
|
||
// 回合标识（`trace.FlowGraph.TurnID`，与 `chat_messages.turn_id` 同值同源），后者是
|
||
// 图 id（`trace.FlowGraph.ID`，也是本表的冲突键）。落盘时两个都写，使「按消息的那一路
|
||
// 查图」（`GraphsByTurnID`）是一次等值索引查询，而不是拿图 id 去反解。二次收尾同样
|
||
// 更新它：同一条路的图又被收一次时，标识必须跟着走到最新的那一份。
|
||
//
|
||
// ⚠️ `lane_turn_id` 的**列**口径（迁移 0023 的注释）写的是「单路回合的图不带该键 ⇒
|
||
// 落盘为空」，那是当时写侧的实情；写侧现已改为**每一路都取权威回合标识**（见
|
||
// `loop.FlowPublisher.StartRoundWithTurnID`）。列的形状与语义一字未改，只是**新的
|
||
// 单路行不再必然为空** —— 老行仍是空串（那时图里根本没存这个标识，任何回填都是编造）。
|
||
const upsertRoundSQL = `INSERT INTO flow_graphs
|
||
(session_id, turn_id, lane_turn_id, graph_json, created_at, updated_at)
|
||
VALUES (?, ?, ?, ?, ?, ?)
|
||
ON CONFLICT(session_id, turn_id) DO UPDATE SET
|
||
graph_json = excluded.graph_json,
|
||
lane_turn_id = excluded.lane_turn_id,
|
||
updated_at = excluded.updated_at`
|
||
|
||
// RecordRound 把一条**已收尾轮次**的图落盘（每轮一行）。它是发布器 `EndRound` 的落盘
|
||
// 出口，因此签名与 `loop.RoundRecorder` 一致：**不接收 ctx、自己用一个有界的后台
|
||
// context**（会话收尾路径不该被库拖住）。
|
||
//
|
||
// 它同时盖上这一路的**回合标识**（`g.TurnID` → `lane_turn_id`，迁移 0023），使消息面上
|
||
// 「这条回复当时调了什么工具」能按**等值**查到这一轮的图（见 `GraphsByTurnID`）。
|
||
// 标识为空时如实写空串：那一行因此按路查不到，读取侧让 `toolCalls` 整体缺席 ——
|
||
// 这正是契约的语义，绝不用时间窗或正文去猜。`g.TurnID` 为空只剩一类情形：
|
||
// **没有 ctx 通道**的调用方（未接编排器的写入路径、旧后端落下的历史行）——
|
||
// 编排器驱动的每一路（含普通单路回合与队首续跑路）都有权威回合标识。
|
||
//
|
||
// 失败**上抛**给调用方（发布器记 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, g.TurnID, 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
|
||
}
|