Cyrene_v2/internal/runtime/flowgraph/flowgraph.go

// 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
}