Cyrene_v2/internal/runtime/flowgraph/timeline.go

package flowgraph

// 本文件是**时间轴读侧**（design/19 §7；切片 W1，契约 v0.80 的
// `GET /api/v1/admin/flow/timeline`）：把该会话**保留期内**的多轮链路图按时间正序
// 取回来，供管理面板把「所有消息按时间顺序串起来」。
//
// 它补的是主人 2026-10-06 那句原话的缺口：「web 后台的全链路流程图我要的是所有消息按
// 时间顺序串起来的流程图，而不是只有最新一个来回」。数据早就在库里（迁移 0022，每轮/
// 每路一行、保留 168h），缺的只是**多轮聚合读侧** —— 本文件就是它。采集层（`loop` 的
// 发布器）与保留期清扫（`gc.go`）一字未改。
//
// 三条口径（每条都有测试钉住，不靠注释自觉）：
//
//  1. **按轮分组**：一轮 = 一条用户消息引出的全部工作；真并行时一轮有多路
//     （`graph.Round` 同值、`graph.TurnID` 各异，design/15 §2.5.5）⇒ 分组键是**轮次号**，
//     不是图 id。图上没有轮次号的老行（`round` 键缺席）**各自成组**，绝不猜它属于哪一轮。
//  2. **游标向更早翻页**：`before` 是**轮次开始时刻**（定宽 UTC 毫秒 RFC3339），只返回
//     **严格早于**它的轮次；调用方把响应里的 `nextBefore`（= 本页最早一轮的开始时刻）
//     原样回带即可连续翻页。
//  3. **查不到就是查不到**：已过保留期的行已被 `gc.go` 清扫、从未落盘的行本就不存在 ——
//     两者都不出现在返回值里，本包**不猜**、绝不用别的轮次顶替。保留期口径由调用方经
//     `Retention()` 如实转达给客户端（「更早的轮次已过保留期」这句话不能靠前端编）。
//  4. **清扫事实单独判**（切片 X1 新增，见 `EarlierRoundsSwept`）：`Timeline` 的第二个
//     返回值只回答「本次是不是全部」（`hasMore`），它**不**回答「更早的轮次是不是被
//     保留期清扫掉了」—— 那是另一件事、另一个判据。

import (
	"context"
	"fmt"
	"strconv"
	"strings"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
)

// TimelineRound 是时间轴里的**一轮**（design/19 §7.1）：该轮的开始时刻 + 该轮每一路的图。
//
// 「轮」（round）与「路」（leg）的区分见 design/19 §6.0：一轮是时间单元（一条用户消息
// 引出的全部工作），一路是同轮里同时推进的独立回合。**泳道**是图内的子代理带，与本类型
// 无关。
type TimelineRound struct {
	// Key 是**分组键**（`r:<轮次号>` / `t:<图 id>`），与 RoundKeyOf 逐字同源：键相同的
	// 两条路属于同一轮。调用方用它把**内存里的在飞路**并进同一轮（而不是另起一轮）。
	Key string
	// Round 是该轮的**会话轮次号**；0 = 未知（图上没有该键的旧行）。调用方据此让线上
	// JSON 出 `null`，**绝不用 0 冒充**一个真实轮次（design/19 §6.1 的缺席语义）。
	Round int
	// StartedAt 是该轮**最早一路**的开始时刻。图上的 `startedAt` 不可解析时回落该行的
	// 落盘时刻（与 SQL 侧的 COALESCE 同口径 —— 换的是同一条真实记录里的另一个时间字段，
	// 不是编造一个时刻）。
	StartedAt time.Time
	// Lanes 是该轮**每一路**的图（真并行时多于一条）。次序**不保证**：调用方按路的开始
	// 时刻自行排序展示（网关注入内存路后再统一排）。
	Lanes []*trace.FlowGraph
}

// RoundKeyOf 给出一个图的**轮次分组键**：有轮次号 ⇒ `r:<轮次号>`，否则 ⇒ `t:<图 id>`
// （旧图没有轮次号 ⇒ 每张图自成一组，绝不把它们猜成同一轮）。
//
// 它与 `timelineRoundKeySQL` **逐字同形**：一处是 Go、一处是 SQL，改一个必须改另一个。
// 两处的存在是必要的 —— SQL 侧要能在库里分组分页（不把整会话的行都读进内存），Go 侧要
// 能把**内存里的在飞路**归到同一个键上。
func RoundKeyOf(g *trace.FlowGraph) string {
	if g == nil {
		return ""
	}
	if g.Round > 0 {
		return "r:" + strconv.Itoa(g.Round)
	}
	return "t:" + g.ID
}

// CursorStamp 把一个时刻格式化成**时间轴游标口径**（定宽 UTC 毫秒 + 字面量 Z）。
//
// 它与 `timelineRoundStartSQL` 归一化出来的字符串**逐字同形**，因此可以直接当作
// `Timeline` 的 `before` 参数回带（游标精度到毫秒：这是 `strftime('%f')` 的精度，
// 也是本包全部时间比较的口径）。零值时刻返回空串 = 未知（调用方原样出线上字段，
// 不要用一个假时刻顶替）。
func CursorStamp(t time.Time) string {
	if t.IsZero() {
		return ""
	}
	return sqliteTimeStamp(t)
}

// timelineRoundKeySQL 是**分组键**的 SQL 表达式（与 `RoundKeyOf` 逐字同形）：
// 有轮次号 ⇒ `r:<轮次号>`，否则 ⇒ `t:<图 id>`。
//
// 外层的 `json_valid` 是**保护**：JSON 破损的行根本不去解析 —— 否则 `json_extract`
// 会让整条查询报错，把「一行坏数据」升级成「时间轴整体读不了」（且坏行若落在本页之外，
// 用户根本不该因此受罚）。用**嵌套 CASE**（而不是 `json_valid(...) AND ...`）是因为
// SQLite 不保证 AND 的操作数求值次序，而 CASE 只求值命中的那一支。
const timelineRoundKeySQL = `CASE WHEN json_valid(graph_json) THEN
        CASE WHEN CAST(json_extract(graph_json,'$.round') AS INTEGER) > 0
             THEN 'r:' || CAST(json_extract(graph_json,'$.round') AS INTEGER)
             ELSE 't:' || turn_id END
      ELSE 't:' || turn_id END`

// timelineRoundStartSQL 是**轮次开始时刻**的 SQL 表达式：把图上的 `$.startedAt` 归一成
// 定宽 UTC 毫秒字符串（与 `sqliteTimeStamp` 同口径 ⇒ 字典序等价于时间序，可以直接比较）；
// 取不到或不可解析时回落该行的 `created_at`（同为定宽 UTC 毫秒）。
//
// 归一化是必须的：图里的 `startedAt` 由 Go 的 `time.Time` 编码，可能是**本地时区偏移**
// 且**小数位数不定**，直接按字符串比较会得出错误的时间序。
//
// `json_type(...) = 'text'` 与迁移 0023 的回填条件同一条纪律：只认**字符串**形态的时间，
// 数值形态（畸形外部改动）不交给 `strftime` —— 它会把一个数字当儒略日解析出一个看似
// 合理的时刻，那是**编造**。这类行一律回落 `created_at`。两个条件嵌套（而非 AND）同
// 上一条的理由：SQLite 不保证操作数求值次序。
const timelineRoundStartSQL = `COALESCE(
        CASE WHEN json_valid(graph_json) THEN
             CASE WHEN json_type(graph_json,'$.startedAt') = 'text'
                  THEN strftime('%Y-%m-%dT%H:%M:%fZ', json_extract(graph_json,'$.startedAt')) END
        END,
        created_at)`

// timelineRoundNoSQL 是**轮次号**的 SQL 表达式（没有轮次号时为 NULL，SQLite 的 DESC 会把
// NULL 排在最后）。它只用于 ORDER BY 的同毫秒**定序兜底**：两个轮次若落在同一毫秒，
// 就按轮次号降序（号大的在前 = 更新），绝不把次序交给数据库的自由发挥。
//
// 为什么不拿 `round_key` 直接当兜底：它是字符串，`'r:9'` 的字典序**大于** `'r:10'`，
// 单靠它定序会把第 9 轮排到第 10 轮之前。字符串键仍是最后一道兜底（保证完全确定），
// 但轮次号本身必须先比一遍。
const timelineRoundNoSQL = `CASE WHEN json_valid(graph_json)
        THEN CAST(json_extract(graph_json,'$.round') AS INTEGER) END`

// timelineKeysHeadSQL 取**最新一页**的轮次键（最新在前），多取一条用于判断还有没有更早的轮。
const timelineKeysHeadSQL = `SELECT ` + timelineRoundKeySQL + ` AS round_key,
        MIN(` + timelineRoundStartSQL + `) AS started_at
   FROM flow_graphs
  WHERE session_id = ?
  GROUP BY round_key
  ORDER BY started_at DESC, MAX(` + timelineRoundNoSQL + `) DESC, round_key DESC
  LIMIT ?`

// timelineKeysBeforeSQL 与上一条同形，但只取**开始时刻严格早于游标**的轮次（向更早翻页）。
const timelineKeysBeforeSQL = `SELECT ` + timelineRoundKeySQL + ` AS round_key,
        MIN(` + timelineRoundStartSQL + `) AS started_at
   FROM flow_graphs
  WHERE session_id = ?
  GROUP BY round_key
 HAVING MIN(` + timelineRoundStartSQL + `) < ?
  ORDER BY started_at DESC, MAX(` + timelineRoundNoSQL + `) DESC, round_key DESC
  LIMIT ?`

// Timeline 取该会话**保留期内**的多轮链路图，每轮带上它**每一路**的图。
//
// 返回值与语义：
//   - 轮次按开始时刻**升序**（最早在前）—— 时间轴要从上往下读成时间序；
//   - `limit` ≤0 取 `DefaultTurnsLimit`、超过 `MaxTurnsLimit` 即钳住（绝不回「全部」）；
//   - `before` 为零值 = 取**最新那一页**；给定时只取开始时刻**严格早于**它的轮次；
//   - 第二个返回值 `hasMore` 报告库里**还有更早的轮次**没被返回：调用方把本页最早一轮的
//     开始时刻回带成本次请求的 `before` 即可继续（`runs` 为空时 `hasMore` 恒为 false）；
//   - 该会话从未落盘（含发布会话范围外、已过保留期）时返回**空切片**与 `(false, nil)`：
//     这是一个**诚实的空态**，绝不用别的会话/别的轮次顶替。
//
// 一页最多跑两条 SQL：先取本页的轮次键（多取一条判 `hasMore`），再按键把这一页的**全部
// 路**取回来。**不按路分页** —— 那一轮的图会缺路（「两路只回一路」），正是 FV2 修掉的那类
// 静默丢数据的另一种形态。查询全部参数化；本方法只读，不写库、不碰保留期。
func (s *SQLStore) Timeline(ctx context.Context, sessionID string, limit int, before time.Time) ([]TimelineRound, bool, error) {
	if s == nil || s.db == nil {
		return nil, false, fmt.Errorf("flowgraph: query timeline: %w", ErrNilDB)
	}
	if strings.TrimSpace(sessionID) == "" {
		return []TimelineRound{}, false, nil
	}
	keys, more, err := s.timelineKeys(ctx, sessionID, normalizeLimit(limit), before)
	if err != nil {
		return nil, false, err
	}
	if len(keys) == 0 {
		return []TimelineRound{}, more, nil
	}
	rounds, err := s.timelineRounds(ctx, sessionID, keys)
	if err != nil {
		return nil, false, err
	}
	return rounds, more, nil
}

// timelineKeys 取本页的轮次键（最新 ⇒ 最旧）。它多取一条：拿到 `limit+1` 条就说明
// 库里还有更早的轮次（`hasMore`），多出来的那条**不返回**（它是下一页的第一条）。
func (s *SQLStore) timelineKeys(ctx context.Context, sessionID string, limit int, before time.Time) ([]string, bool, error) {
	query := timelineKeysHeadSQL
	args := []any{sessionID, limit + 1}
	if !before.IsZero() {
		query = timelineKeysBeforeSQL
		args = []any{sessionID, sqliteTimeStamp(before), limit + 1}
	}
	rows, err := s.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, false, fmt.Errorf("flowgraph: 查时间轴轮次(%s): %w", sessionID, err)
	}
	defer func() { _ = rows.Close() }()

	keys := make([]string, 0, limit)
	for rows.Next() {
		var (
			key       string
			startedAt string
		)
		if err := rows.Scan(&key, &startedAt); err != nil {
			return nil, false, fmt.Errorf("flowgraph: 读时间轴轮次行(%s): %w", sessionID, err)
		}
		keys = append(keys, key)
	}
	if err := rows.Err(); err != nil {
		return nil, false, fmt.Errorf("flowgraph: 遍历时间轴轮次行(%s): %w", sessionID, err)
	}
	if len(keys) > limit {
		return keys[:limit], true, nil
	}
	return keys, false, nil
}

// timelineRounds 按轮次键取回这一页**每一路**的图，并按时间**升序**装配成轮。
//
// 键的次序由 SQL 给出（最新 ⇒ 最旧，且已按时间定序），本方法取完整路后反转一次即为升序。
// 行内的 `graph_json` 解不开时返回错误（与 `Graph` / `Latest` 同一口径：读不懂就如实报错，
// 绝不把坏行当成「这一轮没有图」）；错误信息只带标识，**绝不打印图内容**。
func (s *SQLStore) timelineRounds(ctx context.Context, sessionID string, keys []string) ([]TimelineRound, error) {
	args := make([]any, 0, len(keys)+1)
	args = append(args, sessionID)
	placeholders := make([]string, 0, len(keys))
	for _, key := range keys {
		placeholders = append(placeholders, "?")
		args = append(args, key)
	}
	query := `SELECT turn_id, graph_json, created_at FROM flow_graphs
	   WHERE session_id = ? AND (` + timelineRoundKeySQL + `) IN (` + strings.Join(placeholders, ", ") + `)
	   ORDER BY created_at ASC, id ASC`
	rows, err := s.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("flowgraph: 查时间轴轮次图(%s, %d 轮): %w", sessionID, len(keys), err)
	}
	defer func() { _ = rows.Close() }()

	byKey := make(map[string]*TimelineRound, len(keys))
	for rows.Next() {
		var (
			turnID    string
			payload   string
			createdAt string
		)
		if err := rows.Scan(&turnID, &payload, &createdAt); err != nil {
			return nil, fmt.Errorf("flowgraph: 读时间轴轮次图行(%s): %w", sessionID, err)
		}
		g, err := decodeGraph(sessionID, turnID, payload)
		if err != nil {
			return nil, err
		}
		if err := mergeLane(byKey, g, createdAt); err != nil {
			return nil, fmt.Errorf("flowgraph: 归并时间轴轮次路(%s/%s): %w", sessionID, turnID, err)
		}
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("flowgraph: 遍历时间轴轮次图行(%s): %w", sessionID, err)
	}
	return roundsInKeyOrder(byKey, keys), nil
}

// mergeLane 把一路图并进它所属的那一轮（键相同即同一轮）。该轮尚未建组时新建一个。
//
// 轮次的开始时刻取**各路最早的那个**：真并行时两路的开始时刻相差毫秒级，轮的起点是
// 先起的那一路。
func mergeLane(byKey map[string]*TimelineRound, g *trace.FlowGraph, createdAt string) error {
	key := RoundKeyOf(g)
	round := byKey[key]
	if round == nil {
		round = &TimelineRound{Key: key, Round: g.Round}
		byKey[key] = round
	}
	start := g.StartedAt
	if start.IsZero() {
		// 图上的开始时刻不可用 ⇒ 回落该行的落盘时刻（与 SQL 的 COALESCE 同口径）。
		at, ok := parseSQLiteTime(createdAt)
		if !ok {
			return fmt.Errorf("轮次 %s 的 created_at 不可解析(%q): %w", g.ID, createdAt, ErrBadTimestamp)
		}
		start = at
	}
	if round.StartedAt.IsZero() || start.Before(round.StartedAt) {
		round.StartedAt = start
	}
	round.Lanes = append(round.Lanes, g)
	return nil
}

// EarlierRoundsSwept 报告「该会话**确有更早的轮次已被保留期清扫删掉**」这一事实
// （design/19 §7.1 的清扫信号；切片 X1，2026-10-06）。
//
// **它补的是真机挖出的第二处真缺陷**：此前契约与设计把 `truncated` 读作「更早的轮次
// 已过保留期」，而后端的 `truncated = hasMore || !storeWired` 说的是「本次不是全部」
// —— 一页还没拉完时，面板就会**对着仍然可翻的历史说假话**。两件事必须分开：
// `truncated` 回答「本次是不是全部」，本函数回答「更早的是不是被清扫掉了」。
//
// **判据（一条链，逐环可复核）**：
//  1. 轮次号**自 1 起**：一个会话的第一次发号必然是 1（`loop` 的计数器从 0 起，其轮
//     次号 = 该轮首路的建图序号），且此后**每会话单调**（跨重启也单调 —— 发布器在
//     本进程第一次给该会话发号之前，用 `MaxRound` 把计数器播到已落盘的最大值之上）；
//  2. 因此「现存**最早**一轮的轮次号 > 1」只可能由**删除**造成：号更小的那些轮次确实
//     发过号（= 确实建过图、收尾时确实尝试落盘）；
//  3. `flow_graphs` 的**唯一**删除者就是保留期清扫（`gc.go`：按 `created_at` 删过期行；
//     读取侧只读、发布器只 upsert）。
//     ⇒ 最早一轮的轮次号 > 1 时，「更早的轮次已被保留期清扫」成立。
//
// **调用方必须保证 rounds 是该会话**完整**的轮次列表**（即 `Timeline` 返回 `hasMore`
// 为 false 的那一页，可再并入内存里的在飞路）：只看一页就说「已过保留期」，正是本切片
// 要治的那句假话。本函数是**纯判断**，不查库、不猜时间。
//
// **诚实的盲区（不美化）**：① 一个会话若**整段历史都被清扫**（一条可用行都不剩），
// 发号会从 1 重新开始 ⇒ 本函数无法断言，返回 false（**少说**，绝不多说）；② 同一会话
// 里有轮次号缺席的老行时同样返回 false（号不可信）；③ 唯一可能**错报**的路径是该会话
// 最早的轮次**落盘失败**（`RecordRound` 报错，发布器记 WARN）—— 那一轮从未进库、号却
// 已用掉，本函数会把它算成「已清扫」。概率低、只在库写失败时出现，如实登记在
// design/19 §7 与契约里，不掩盖。
func EarlierRoundsSwept(rounds []TimelineRound) bool {
	if len(rounds) == 0 {
		return false
	}
	if rounds[0].Round <= 0 {
		return false
	}
	floor := rounds[0].Round
	for i := 1; i < len(rounds); i++ {
		if rounds[i].Round <= 0 {
			// 有轮次号未知的轮次 ⇒ 「号更小的都删过了」这条推论不成立，保守为否。
			return false
		}
		if rounds[i].Round < floor {
			floor = rounds[i].Round
		}
	}
	return floor > 1
}

// roundsInKeyOrder 按 SQL 给出的键次序（最新 ⇒ 最旧）装配轮次并**反转为升序**
// （时间轴从上往下读成时间序）。键没有对应行时跳过 —— 那意味着并发清扫刚删掉了它。
func roundsInKeyOrder(byKey map[string]*TimelineRound, keys []string) []TimelineRound {
	out := make([]TimelineRound, 0, len(keys))
	for i := len(keys) - 1; i >= 0; i-- {
		if round := byKey[keys[i]]; round != nil {
			out = append(out, *round)
		}
	}
	return out
}