From 1efe3a6a7998446aec329aa927c6dc327a31409a Mon Sep 17 00:00:00 2001 From: AskaEth Date: Tue, 6 Oct 2026 15:59:09 +0800 Subject: [PATCH] =?UTF-8?q?feat(flow):=20=E9=93=BE=E8=B7=AF=E5=9B=BE?= =?UTF-8?q?=E6=97=B6=E9=97=B4=E8=BD=B4=E5=87=BA=E5=8F=A3=EF=BC=88W1?= =?UTF-8?q?=EF=BC=89=E2=80=94=E2=80=94=20=E6=89=80=E6=9C=89=E8=BD=AE?= =?UTF-8?q?=E6=AC=A1=E6=8C=89=E6=97=B6=E9=97=B4=E6=AD=A3=E5=BA=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 GET /api/v1/admin/flow/timeline?sessionId=&limit=&before=：时间正序多轮， 每轮 {turnId,round,startedAt,endedAt,lanes[]}；limit 缺省 50/上限 200 + before 游标 - 数据源=落盘 flow_graphs + 内存最近一轮（同图 id 去重）；保留期 168h 外的轮次不返回， 但响应给 retentionHours/retentionEnabled/truncated ⇒ 前端如实标注，不悄悄少一块 - 老出口 GET /admin/flow/graph 零 diff；角色门 writeAuth（owner|admin） - 4 个变异自检（游标 <= 重复、分组降级、去重缺失、truncated 静默） --- internal/runtime/flowgraph/timeline.go | 297 ++++++++++++ internal/runtime/flowgraph/timeline_test.go | 263 ++++++++++ .../services/gateway/admin_flow_timeline.go | 382 +++++++++++++++ .../gateway/admin_flow_timeline_test.go | 454 ++++++++++++++++++ internal/services/gateway/routes.go | 5 + 5 files changed, 1401 insertions(+) create mode 100644 internal/runtime/flowgraph/timeline.go create mode 100644 internal/runtime/flowgraph/timeline_test.go create mode 100644 internal/services/gateway/admin_flow_timeline.go create mode 100644 internal/services/gateway/admin_flow_timeline_test.go diff --git a/internal/runtime/flowgraph/timeline.go b/internal/runtime/flowgraph/timeline.go new file mode 100644 index 0000000..5a6cad4 --- /dev/null +++ b/internal/runtime/flowgraph/timeline.go @@ -0,0 +1,297 @@ +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()` 如实转达给客户端（「更早的轮次已过保留期」这句话不能靠前端编）。 + +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 +} + +// 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 +} diff --git a/internal/runtime/flowgraph/timeline_test.go b/internal/runtime/flowgraph/timeline_test.go new file mode 100644 index 0000000..acd0523 --- /dev/null +++ b/internal/runtime/flowgraph/timeline_test.go @@ -0,0 +1,263 @@ +package flowgraph + +// 本文件钉住**时间轴读侧**（design/19 §7；切片 W1）：多轮按时间正序取回、一轮含其 +// **每一路**、`before` 游标向更早翻页不重不漏、没有轮次号的老行各自成组。 +// +// 断言的是**库里读回来的事实**（几轮、每轮几路、次序、`hasMore`），不按日志断言 —— +// 与 flowgraph_test.go / flowgraph_lane_test.go 同一口径。 + +import ( + "context" + "fmt" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// timelineGraph 用真建图器造一条路的收尾图：round 是会话轮次号（同轮的多路共享）， +// seq 是会话级建图序号（每条路必不同，决定图 id），lane 是这一路的权威回合标识。 +func timelineGraph(session string, round, seq int, lane string, started time.Time) *trace.FlowGraph { + g := trace.NewRunGraph(session) + g.SetTurnID(lane) + g.SetSeq(seq) + g.Start(round, started) + g.SeedInput("输入-" + lane) + g.SeedOutput("回复-" + lane) + return g.Snapshot() +} + +// recordTimelineRound 落一条时间轴夹具（失败即 Fatal）。 +func recordTimelineRound(t *testing.T, st *SQLStore, g *trace.FlowGraph) { + t.Helper() + if err := st.RecordRound(g); err != nil { + t.Fatalf("落盘 %s: %v", g.ID, err) + } +} + +// TestTimeline_GroupsLanesByRound_正常 是本文件的核心断言：三轮按**时间正序**返回， +// 中间那一轮的**两路**并进同一轮（`lanes` 两条），而不是被拆成两轮、也不是只剩一路。 +func TestTimeline_GroupsLanesByRound_正常(t *testing.T) { + t.Parallel() + st, _ := newTestStore(t, nil) + ctx := context.Background() + + base := testNow().Add(-time.Hour) + recordTimelineRound(t, st, timelineGraph("web", 1, 1, "turn-a", base.Add(30*time.Minute))) + recordTimelineRound(t, st, timelineGraph("web", 2, 2, "turn-b", base.Add(40*time.Minute))) + recordTimelineRound(t, st, timelineGraph("web", 2, 3, "turn-c", base.Add(40*time.Minute).Add(time.Second))) + recordTimelineRound(t, st, timelineGraph("web", 3, 4, "turn-d", base.Add(50*time.Minute))) + + rounds, more, err := st.Timeline(ctx, "web", 0, time.Time{}) + if err != nil { + t.Fatalf("Timeline: %v", err) + } + if more { + t.Fatal("hasMore = true，但库里只有这三轮") + } + if len(rounds) != 3 { + t.Fatalf("轮数 = %d, want 3（同一轮的两路属**同一轮**）: %+v", len(rounds), roundKeys(rounds)) + } + for i, want := range []string{"r:1", "r:2", "r:3"} { + if rounds[i].Key != want { + t.Fatalf("rounds[%d].Key = %q, want %q（时间正序）", i, rounds[i].Key, want) + } + } + for i, want := range []int{1, 2, 3} { + if rounds[i].Round != want { + t.Fatalf("rounds[%d].Round = %d, want %d", i, rounds[i].Round, want) + } + } + if got := len(rounds[1].Lanes); got != 2 { + t.Fatalf("第 2 轮的 lanes = %d, want 2（两路都在）: %+v", got, rounds[1].Lanes) + } + if !rounds[1].StartedAt.Equal(base.Add(40 * time.Minute)) { + t.Fatalf("第 2 轮开始时刻 = %v, want %v（取最早那一路）", rounds[1].StartedAt, base.Add(40*time.Minute)) + } + if rounds[0].Lanes[0].ID != "web-1#turn-a" { + t.Fatalf("第 1 轮的路 id = %q, want web-1#turn-a", rounds[0].Lanes[0].ID) + } +} + +// TestTimeline_BeforeCursorPagesWithoutGapOrDup_边界 钉住分页：`limit=2` 逐页向更早翻， +// 各页**不重不漏**、`hasMore` 如实、翻到底后是诚实空态。 +func TestTimeline_BeforeCursorPagesWithoutGapOrDup_边界(t *testing.T) { + t.Parallel() + st, _ := newTestStore(t, nil) + ctx := context.Background() + + base := testNow().Add(-10 * time.Hour) + for round := 1; round <= 5; round++ { + started := base.Add(time.Duration(round) * time.Hour) + recordTimelineRound(t, st, timelineGraph("web", round, round, fmt.Sprintf("turn-%d", round), started)) + } + + seen := map[string]int{} + before := time.Time{} + wantPages := [][]string{ + {"r:4", "r:5"}, + {"r:2", "r:3"}, + {"r:1"}, + {}, + } + wantMorePages := []bool{true, true, false, false} + for page, want := range wantPages { + rounds, more, err := st.Timeline(ctx, "web", 2, before) + if err != nil { + t.Fatalf("第 %d 页 Timeline: %v", page+1, err) + } + if got := roundKeys(rounds); !sameKeys(got, want) { + t.Fatalf("第 %d 页轮次 = %v, want %v", page+1, got, want) + } + if more != wantMorePages[page] { + t.Fatalf("第 %d 页 hasMore = %v, want %v", page+1, more, wantMorePages[page]) + } + if len(rounds) == 0 { + continue + } + for _, r := range rounds { + seen[r.Key]++ + } + before = rounds[0].StartedAt + } + for round := 1; round <= 5; round++ { + if key := fmt.Sprintf("r:%d", round); seen[key] != 1 { + t.Fatalf("轮 %s 出现 %d 次, want 1（翻页不重不漏；全部=%v）", key, seen[key], seen) + } + } +} + +// TestTimeline_LegacyRowWithoutRoundIsOwnGroup_边界 钉住老行的诚实口径：图上没有轮次号 +// （`round` 键缺席）时**每张图自成一组**，绝不把它们猜成同一轮，`Round` 如实为 0 +// （调用方据此让线上 JSON 出 `null`）。 +func TestTimeline_LegacyRowWithoutRoundIsOwnGroup_边界(t *testing.T) { + t.Parallel() + st, _ := newTestStore(t, nil) + ctx := context.Background() + + // 手工造两张「老图」：Round 为 0（= 键缺席）、图 id 不同。 + for i, id := range []string{"web-old-1", "web-old-2"} { + g := &trace.FlowGraph{ + ID: id, + SessionID: "web", + StartedAt: testNow().Add(-time.Duration(10-i) * time.Minute), + EndedAt: testNow(), + Nodes: []trace.FlowNode{{ID: "input", Label: "你好", Kind: "input"}}, + } + recordTimelineRound(t, st, g) + } + + rounds, _, err := st.Timeline(ctx, "web", 0, time.Time{}) + if err != nil { + t.Fatalf("Timeline: %v", err) + } + if len(rounds) != 2 { + t.Fatalf("轮数 = %d, want 2（无轮次号的行各自成组，绝不猜）", len(rounds)) + } + if rounds[0].Key != "t:web-old-1" || rounds[1].Key != "t:web-old-2" { + t.Fatalf("分组键 = %q/%q, want t:web-old-1 / t:web-old-2", rounds[0].Key, rounds[1].Key) + } + for i, r := range rounds { + if r.Round != 0 { + t.Fatalf("rounds[%d].Round = %d, want 0（未知，不得冒充真实轮次）", i, r.Round) + } + } +} + +// TestTimeline_StartFallsBackToCreatedAt_边界 钉住时间回落：图上的 `startedAt` 不可用时， +// 轮次开始时刻取该行的**落盘时刻**（同一条记录里的另一个真实时间字段），而不是零值。 +func TestTimeline_StartFallsBackToCreatedAt_边界(t *testing.T) { + t.Parallel() + st, _ := newTestStore(t, nil) + ctx := context.Background() + + recordTimelineRound(t, st, &trace.FlowGraph{ + ID: "web-9", + SessionID: "web", + Round: 9, + Nodes: []trace.FlowNode{{ID: "input", Label: "你好", Kind: "input"}}, + }) + + rounds, _, err := st.Timeline(ctx, "web", 0, time.Time{}) + if err != nil { + t.Fatalf("Timeline: %v", err) + } + if len(rounds) != 1 { + t.Fatalf("轮数 = %d, want 1", len(rounds)) + } + if !rounds[0].StartedAt.Equal(testNow()) { + t.Fatalf("开始时刻 = %v, want %v（图上没有 startedAt ⇒ 回落落盘时刻）", rounds[0].StartedAt, testNow()) + } +} + +// TestTimeline_LimitClamped_边界 钉住页大小：缺省 50、超过硬上限 200 即钳住且 `hasMore` +// 如实为 true（绝不静默返回全表）。 +func TestTimeline_LimitClamped_边界(t *testing.T) { + t.Parallel() + st, _ := newTestStore(t, nil) + ctx := context.Background() + + base := testNow().Add(-time.Hour) + total := MaxTurnsLimit + 5 + for round := 1; round <= total; round++ { + started := base.Add(time.Duration(round) * time.Second) + recordTimelineRound(t, st, timelineGraph("web", round, round, "", started)) + } + + rounds, more, err := st.Timeline(ctx, "web", MaxTurnsLimit+999, time.Time{}) + if err != nil { + t.Fatalf("Timeline(超上限): %v", err) + } + if len(rounds) != MaxTurnsLimit || !more { + t.Fatalf("超上限时 = %d 轮 / hasMore=%v, want %d / true", len(rounds), more, MaxTurnsLimit) + } + rounds, more, err = st.Timeline(ctx, "web", 0, time.Time{}) + if err != nil { + t.Fatalf("Timeline(缺省): %v", err) + } + if len(rounds) != DefaultTurnsLimit || !more { + t.Fatalf("缺省时 = %d 轮 / hasMore=%v, want %d / true", len(rounds), more, DefaultTurnsLimit) + } +} + +// TestTimeline_EmptyAndNilDB_边界 钉住两个空态：没有任何落盘行时返回**空切片**（不是 +// nil、更不是错误），未接数据库时如实报 ErrNilDB。 +func TestTimeline_EmptyAndNilDB_边界(t *testing.T) { + t.Parallel() + st, _ := newTestStore(t, nil) + + rounds, more, err := st.Timeline(context.Background(), "web", 0, time.Time{}) + if err != nil { + t.Fatalf("Timeline(空会话): %v", err) + } + if rounds == nil || len(rounds) != 0 || more { + t.Fatalf("空会话 = %+v/%v, want 空切片/false", rounds, more) + } + + var nilStore *SQLStore + if _, _, err := nilStore.Timeline(context.Background(), "web", 0, time.Time{}); err == nil { + t.Fatal("nil store 必须如实报错，不得假装「没有轮次」") + } +} + +// roundKeys 取出轮次键序列（断言用）。 +func roundKeys(rounds []TimelineRound) []string { + out := make([]string, 0, len(rounds)) + for _, r := range rounds { + out = append(out, r.Key) + } + return out +} + +// sameKeys 报告两个键序列是否逐字相等。 +func sameKeys(got, want []string) bool { + if len(got) != len(want) { + return false + } + for i := range got { + if got[i] != want[i] { + return false + } + } + return true +} diff --git a/internal/services/gateway/admin_flow_timeline.go b/internal/services/gateway/admin_flow_timeline.go new file mode 100644 index 0000000..d7f02dd --- /dev/null +++ b/internal/services/gateway/admin_flow_timeline.go @@ -0,0 +1,382 @@ +package gateway + +// 本文件是**多轮时间轴**出口（design/19 §7；切片 W1，契约 v0.80）： +// `GET /api/v1/admin/flow/timeline?sessionId=&limit=&before=` +// +// 主人 2026-10-06 的原话：「web 后台的全链路流程图我要的是**所有消息按时间顺序串起来**的 +// 流程图，而不是只有最新一个来回。」老出口 `GET /admin/flow/graph` 只给**最近一轮** +// 的图（v0.71 的冷启动快照）与 `turns[]` 摘要（v0.73 的轮次键列表），没有任何一处把 +// 多轮按时间串起来 —— 本文件补的就是它，且**纯增量**：老出口逐字未改（老客户端零影响）。 +// +// 四条口径： +// +// 1. **只读**：经读侧出口拿快照，网关不发布、不改图、不动任何会话、不落盘、不审计 +// （与同族的 `/admin/flow/graph` 同门同档：writeAuth = owner|admin，design/19 §3）。 +// 2. **两个数据源合用**：主体是**落盘**的 `flow_graphs`（迁移 0022，保留期见 +// `Retention()`），另把**内存里的最近一轮**（在飞/刚落盘失败的那一轮）并进来 —— +// 面板打开时最新一轮还在跑，不该在时间轴上缺席。 +// 3. **时间正序 + 游标向更早翻页**：轮次按开始时刻升序返回；`before` 是「取开始时刻 +// 严格早于它的轮次」，响应带 `nextBefore` 供客户端原样回带（分页不重不漏的口径由 +// `flowgraph.SQLStore.Timeline` 保证并测试）。 +// 4. **保留期如实**：过期轮次已被清扫，不在返回值里；响应给 `retentionHours` / +// `retentionEnabled` / `truncated`，由前端据此标注「更早的轮次已过保留期」—— +// 绝不静默少一块（design/19 §7.1 的原话）。 +// +// 未接线（内存与落盘**都没接**）时回 `500 flow_unavailable`，与老出口同一码、同一口径： +// 绝不回一份空的 `rounds` 把「图源没接线」伪装成「这台会话没有轮次」。 + +import ( + "context" + "fmt" + "net/http" + "sort" + "strconv" + "strings" + "time" + + apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/flowgraph" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// flowTimelineSource 是**落盘多轮时间轴**的读侧出口（契约 v0.80）。它**故意**与 +// `FlowGraphStore` 分开声明：那条接口是为「最小权限读面」定的四个方法（`contracts.go` +// 的注释逐条写明不许扩），而时间轴还要两件它没有的东西 —— 分组分页读（`Timeline`）与 +// 保留期值（`Retention`）。分开声明使既有查询面与它的桩（`history_toolcalls_test.go` +// 的 `stubFlowGraphStore`）一字不动。 +// +// 生产实现恒为 `*flowgraph.SQLStore`（由下面的编译期断言钉住）；组合根若把 +// `Deps.FlowGraphs` 换成别的实现而它没有这两个方法，端点会**显式**回 +// `500 flow_unavailable`（见 handler），绝不静默退化成「只有内存」假装一切正常。 +type flowTimelineSource interface { + // Timeline 按时间正序取该会话保留期内的多轮（每轮含该轮每一路），并报告库里 + // 是否还有更早的轮次（`hasMore`）。 + Timeline(ctx context.Context, sessionID string, limit int, before time.Time) ([]flowgraph.TimelineRound, bool, error) + // Retention 是该存储的保留期；≤0 = 未启用保留期清扫（`flowgraph.Config` 的口径）。 + Retention() time.Duration +} + +// 编译期断言：链路图落盘存储必须能满足时间轴读侧（否则生产组合根一换实现就会 500）。 +var _ flowTimelineSource = (*flowgraph.SQLStore)(nil) + +// flowTimelineLaneDTO 是时间轴里一轮中的**某一路**：`turnId` 是那一路的路标识 +// （`graph.turnId`，design/15 §2.5.9；**可能为空** = 这一路没有权威标识），`graph` 与 +// `graph_update` 帧的 data 逐字同形（与老出口的 `lanes[]` 同一形状）。 +type flowTimelineLaneDTO struct { + TurnID string `json:"turnId"` + Graph any `json:"graph"` +} + +// flowTimelineRoundDTO 是时间轴里的**一轮**（design/19 §7.1 给定了形状）。 +// +// 「轮」= 一条用户消息引出的全部工作；真并行时一轮有多路，各路的图都在 `lanes[]` 里。 +type flowTimelineRoundDTO struct { + // TurnID 是这一轮**首路**的图 id（一个真 id：`{会话}-{序号}[#{路}]`）。多路时它只代表 + // 首路，全貌在 `lanes[]`（每路的图 id 见 `lanes[].graph.id`）—— 绝不臆造一个 + // 「轮次 id」出来（design/19 §6.1 的诚实口径）。 + TurnID string `json:"turnId"` + // Round 是轮次号；`null` = **未知**（那批行早于轮次号落库）—— 不用 0 冒充。 + Round *int `json:"round"` + // StartedAt / EndedAt 是这一轮的开始/结束时刻（UTC RFC3339 **毫秒**，与 `before` + // 游标同一口径）。EndedAt 为空串 = 这一轮**还没有收尾**（内存里的在飞轮）， + // 客户端据此显示「进行中」，而不是画一个假的结束时刻。 + StartedAt string `json:"startedAt"` + EndedAt string `json:"endedAt"` + // Lanes 是该轮每一路，按路的开始时刻升序。恒非 nil。 + Lanes []flowTimelineLaneDTO `json:"lanes"` +} + +// adminFlowTimelineResponse 是 `GET /api/v1/admin/flow/timeline` 的响应体（契约 v0.80）。 +type adminFlowTimelineResponse struct { + SessionID string `json:"sessionId"` + // Source ∈ persisted / live / none（沿用老出口的取值域）： + // persisted = 本页**至少有一轮**来自落盘（可能同时含内存里的在飞路）； + // live = 本页**全部**来自内存（落盘未接线，或该会话还没有落盘轮）； + // none = 没有任何轮次（`rounds:[]`）。 + Source string `json:"source"` + // Rounds 恒非 nil（没有轮次时是 `[]`），按时间**正序**（最早在前）—— 时间轴就是 + // 从上往下读成时间序。 + Rounds []flowTimelineRoundDTO `json:"rounds"` + // NextBefore 是**本页最早一轮**的开始时刻：`hasMore` 为 true 时原样回带成下一次 + // 请求的 `before` 即可继续向更早翻页。没有轮次时为空串。 + NextBefore string `json:"nextBefore"` + // HasMore 报告库里**还有更早的轮次**没被本次返回（用 NextBefore 继续取）。 + HasMore bool `json:"hasMore"` + // Truncated 报告本次**不是全部**：`hasMore`（分页未到底）或落盘源未接线（只剩内存 + // 最近的这一轮）。前端据此如实标注「更早的轮次未显示」，**绝不静默少一块**。 + Truncated bool `json:"truncated"` + // RetentionHours 是落盘保留期（小时，下取整）；`RetentionEnabled` 为 false 时保留期 + // 清扫**未启用** ⇒ "更早的轮次" 不是被清扫掉的，别把它说成过期。 + // `FLOW_GRAPH_RETENTION` 是既有旋钮（默认 168h），本片**不新增任何配置键**。 + RetentionHours int `json:"retentionHours"` + RetentionEnabled bool `json:"retentionEnabled"` +} + +// adminFlowTimeline 是 `GET /api/v1/admin/flow/timeline` 的处理器（契约 v0.80； +// design/19 §7「所有消息按时间顺序串起来」）。 +// +// 参数： +// - `sessionId` 省略时落到主会话 `web`（与老出口同一默认，见 mainSessionID）； +// - `limit` 缺省 50、硬上限 200（超过即钳住，绝不返回全表）； +// - `before` 取「开始时刻严格早于它」的轮次（省略 = 最新那一页），取值是响应里的 +// `nextBefore`，也接受 Unix 毫秒（与 `llm/usage` / 渠道事件同一解析口径）。 +// +// 角色门：`writeAuth`（owner|admin），与老出口、与同一张图的 WS 下行帧同档。 +func (s *Server) adminFlowTimeline(w http.ResponseWriter, r *http.Request) { + sessionID := strings.TrimSpace(r.URL.Query().Get("sessionId")) + if sessionID == "" { + sessionID = mainSessionID + } + if s.flow == nil && s.flowGraphs == nil { + s.log.Warn("flow timeline source not wired", "session", sessionID) + writeError(w, apperr.Wrap(apperr.ErrInternal, flowUnavailableCode, "flow graph source is not wired")) + return + } + limit, err := timelineLimitParam(r.URL.Query().Get("limit")) + if err != nil { + writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "%s", err.Error())) + return + } + before, err := parseInstantParam(r.URL.Query().Get("before")) + if err != nil { + writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "before 非法：%s", err.Error())) + return + } + store, storeWired := s.flowGraphs.(flowTimelineSource) + if s.flowGraphs != nil && !storeWired { + s.log.Warn("flow timeline store lacks timeline reads", "session", sessionID) + writeError(w, apperr.Wrap(apperr.ErrInternal, flowUnavailableCode, "链路图落盘源不支持时间轴读取")) + return + } + rounds := []flowgraph.TimelineRound{} + hasMore := false + retention := time.Duration(0) + if storeWired { + rounds, hasMore, err = store.Timeline(r.Context(), sessionID, limit, before) + if err != nil { + // 原始错误（可能带 SQL/路径）只进日志；响应只给稳定 code + 中文说明。 + s.log.Warn("flow timeline read failed", "session", sessionID, "err", err) + writeError(w, apperr.Wrap(apperr.ErrInternal, flowUnavailableCode, "读取链路图时间轴失败")) + return + } + retention = store.Retention() + } + persisted := len(rounds) > 0 + // 并入内存里的最近一轮：同一轮并进已有那一轮，库里还没有的那一轮当最新一轮。 + merged, droppedOldest := mergeLiveRounds(rounds, s.liveFlowLanes(sessionID), before, limit) + if droppedOldest { + // 被挤掉的那一轮仍在库里 ⇒ 必须如实说「还有更早的」，否则客户端以为已翻到底。 + hasMore = true + } + writeJSON(w, http.StatusOK, adminFlowTimelineResponse{ + SessionID: sessionID, + Source: timelineSource(persisted, len(merged) > 0), + Rounds: flowTimelineRoundsOf(merged), + NextBefore: timelineNextBefore(merged), + HasMore: hasMore, + Truncated: hasMore || !storeWired, + RetentionHours: int(retention / time.Hour), + RetentionEnabled: retention > 0, + }) +} + +// timelineLimitParam 解析 `?limit=`：空串 = 未给（0 ⇒ 存储层用缺省 `DefaultTurnsLimit`）； +// 非法整数报错（400，绝不静默当缺省）；负值归一成 0；超过硬上限 `MaxTurnsLimit` 即钳住。 +func timelineLimitParam(raw string) (int, error) { + trimmed := strings.TrimSpace(raw) + if trimmed == "" { + return 0, nil + } + n, err := strconv.Atoi(trimmed) + if err != nil { + return 0, fmt.Errorf("limit 必须是整数：%q", raw) + } + switch { + case n < 0: + return 0, nil + case n > flowgraph.MaxTurnsLimit: + return flowgraph.MaxTurnsLimit, nil + default: + return n, nil + } +} + +// liveFlowLanes 取该会话**内存最近一轮每一路**的图（图源未接线时为空切片）。 +func (s *Server) liveFlowLanes(sessionID string) []*trace.FlowGraph { + if s.flow == nil { + return nil + } + return s.flow.LatestLanes(sessionID) +} + +// mergeLiveRounds 把内存里的实时路并进落盘读回的时间轴，返回合并后的轮次（时间升序）与 +// 「因超出 limit 丢掉了本页最早一轮」的标志。 +// +// 三条规则： +// - **同一分组键 = 同一轮**：内存里那一路并进落盘已有的那一轮（真并行时落盘可能只有 +// 先收尾的那一路，后一路还在内存里）； +// - **同一个图 id 只算一次**：内存里的那一份与刚落盘那一条是同一条路的两次快照， +// 去重后以**落盘的那一份**为准（它才是收尾后的最终快照）； +// - **`before` 口径一致**：传了游标时，比游标新的内存轮不并进这一页（那是**下一页 +// 之后**的内容，并进来会让分页既重复又错序）。 +// +// `limit` ≤0 时不做截断（调用方已由存储层钳过；这是防御性的一支）。 +func mergeLiveRounds(persisted []flowgraph.TimelineRound, live []*trace.FlowGraph, before time.Time, limit int) ([]flowgraph.TimelineRound, bool) { + rounds := make([]flowgraph.TimelineRound, len(persisted)) + copy(rounds, persisted) + index := make(map[string]int, len(rounds)) + for i := range rounds { + index[rounds[i].Key] = i + } + for _, g := range live { + if g == nil || roundHasLane(rounds, g.ID) { + continue + } + if !before.IsZero() && !g.StartedAt.Before(before) { + continue + } + key := flowgraph.RoundKeyOf(g) + i, ok := index[key] + if !ok { + rounds = append(rounds, flowgraph.TimelineRound{Key: key, Round: g.Round}) + i = len(rounds) - 1 + index[key] = i + } + rounds[i].Lanes = append(rounds[i].Lanes, g) + if rounds[i].StartedAt.IsZero() || (!g.StartedAt.IsZero() && g.StartedAt.Before(rounds[i].StartedAt)) { + rounds[i].StartedAt = g.StartedAt + } + } + sortRoundsAsc(rounds) + if limit <= 0 || len(rounds) <= limit { + return rounds, false + } + return rounds[len(rounds)-limit:], true +} + +// roundHasLane 报告某个**图 id**是否已在这一页里（跨轮检查：同一张图只可能出现一次）。 +func roundHasLane(rounds []flowgraph.TimelineRound, id string) bool { + for i := range rounds { + for _, lane := range rounds[i].Lanes { + if lane != nil && lane.ID == id { + return true + } + } + } + return false +} + +// sortRoundsAsc 按（开始时刻、分组键）升序排轮 —— 时间轴从上往下读成时间序。 +// +// 开始时刻**为零**（未知，理论上只可能来自内存里那一路）的轮排在**最后**：它无从定位， +// 与其把它当成最旧（会被 limit 截掉、凭空消失）或编造一个时刻，不如把它放在最新端。 +// 定序含分组键，因此次序不随 map 遍历序抖动。 +func sortRoundsAsc(rounds []flowgraph.TimelineRound) { + sort.SliceStable(rounds, func(i, j int) bool { + return compareRoundStart(rounds[i], rounds[j]) < 0 + }) +} + +// compareRoundStart 是轮次定序的三值比较（<0 / 0 / >0），见 sortRoundsAsc 的口径。 +func compareRoundStart(a, b flowgraph.TimelineRound) int { + switch { + case a.StartedAt.IsZero() && b.StartedAt.IsZero(): + return strings.Compare(a.Key, b.Key) + case a.StartedAt.IsZero(): + return 1 + case b.StartedAt.IsZero(): + return -1 + case a.StartedAt.Equal(b.StartedAt): + return strings.Compare(a.Key, b.Key) + case a.StartedAt.Before(b.StartedAt): + return -1 + default: + return 1 + } +} + +// flowTimelineRoundsOf 把读侧模型翻成线上形状：每一路按开始时刻升序，轮的 `turnId` 取 +// **首路**的图 id，轮的 `endedAt` 取各路结束时刻里**最晚**的一个（都没收尾则为空串）。 +func flowTimelineRoundsOf(rounds []flowgraph.TimelineRound) []flowTimelineRoundDTO { + out := make([]flowTimelineRoundDTO, 0, len(rounds)) + for _, r := range rounds { + lanes := sortLanesByStart(r.Lanes) + dto := flowTimelineRoundDTO{ + StartedAt: flowgraph.CursorStamp(r.StartedAt), + EndedAt: flowgraph.CursorStamp(roundEndedAt(lanes)), + Lanes: make([]flowTimelineLaneDTO, 0, len(lanes)), + Round: timelineRoundNo(r.Round), + } + for _, g := range lanes { + dto.Lanes = append(dto.Lanes, flowTimelineLaneDTO{TurnID: g.TurnID, Graph: g}) + } + if len(lanes) > 0 { + dto.TurnID = lanes[0].ID + } + out = append(out, dto) + } + return out +} + +// timelineRoundNo 把轮次号翻成线上可空值：>0 出数字，否则出 `null`（未知不冒充 0）。 +func timelineRoundNo(round int) *int { + if round <= 0 { + return nil + } + return &round +} + +// roundEndedAt 取一轮的结束时刻：各路 `endedAt` 里最晚的一个；都没有收尾则零值 +// （线上出空串 = 进行中/未收尾）。 +func roundEndedAt(lanes []*trace.FlowGraph) time.Time { + var latest time.Time + for _, g := range lanes { + if g == nil || g.EndedAt.IsZero() { + continue + } + if latest.IsZero() || g.EndedAt.After(latest) { + latest = g.EndedAt + } + } + return latest +} + +// sortLanesByStart 按「开始时刻升序、同刻按图 id」给一路路图定序（**不改入参**）。 +// 口径与 `loop.sortLanes` 逐字相同（同轮多路的展示次序不随 map 遍历序抖动）； +// 那边是未导出的小助手，这里是第二处 —— 出现第三处时应当抽成共享件。 +func sortLanesByStart(lanes []*trace.FlowGraph) []*trace.FlowGraph { + out := make([]*trace.FlowGraph, 0, len(lanes)) + for _, g := range lanes { + if g != nil { + out = append(out, g) + } + } + sort.SliceStable(out, func(i, j int) bool { + if out[i].StartedAt.Equal(out[j].StartedAt) { + return out[i].ID < out[j].ID + } + return out[i].StartedAt.Before(out[j].StartedAt) + }) + return out +} + +// timelineNextBefore 给出下一页要用的游标：本页**最早一轮**的开始时刻（没有轮次时空串）。 +func timelineNextBefore(rounds []flowgraph.TimelineRound) string { + if len(rounds) == 0 { + return "" + } + return flowgraph.CursorStamp(rounds[0].StartedAt) +} + +// timelineSource 归一 `source`：没有任何轮次 ⇒ none；至少一轮来自落盘 ⇒ persisted； +// 其余（本页全来自内存）⇒ live。 +func timelineSource(persisted, hasRounds bool) string { + switch { + case !hasRounds: + return flowSourceNone + case persisted: + return flowSourcePersisted + default: + return flowSourceLive + } +} diff --git a/internal/services/gateway/admin_flow_timeline_test.go b/internal/services/gateway/admin_flow_timeline_test.go new file mode 100644 index 0000000..3622670 --- /dev/null +++ b/internal/services/gateway/admin_flow_timeline_test.go @@ -0,0 +1,454 @@ +package gateway + +// 本文件钉住契约 v0.80 的新出口 `GET /api/v1/admin/flow/timeline`（design/19 §7；切片 W1）—— +// 主人 2026-10-06 的原话：「web 后台的全链路流程图我要的是所有消息按时间顺序串起来的流程图， +// 而不是只有最新一个来回」。 +// +// 走真路径（真发布器 + 真 SQLite + 真迁移 0022/0023 + 真 HTTP handler），断言的是**线上 +// 形状**与**库里的事实**：多轮按时间正序、一轮含该轮每一路、`before` 游标翻页不重不漏、 +// 内存里在飞的那一轮并进同一轮且不重复、保留期两个字段如实、未接线时 500 flow_unavailable。 + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "path/filepath" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/flowgraph" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// flowTimelineGraphProj / flowTimelineLaneProj / flowTimelineRoundProj 是本文件断言的 +// 线上投影（命名类型，使 `roundNos` 之类的助手能直接吃它，不必复写匿名结构）。 +type flowTimelineGraphProj struct { + ID string `json:"id"` + Round int `json:"round"` +} + +type flowTimelineLaneProj struct { + TurnID string `json:"turnId"` + Graph *flowTimelineGraphProj `json:"graph"` +} + +type flowTimelineRoundProj struct { + TurnID string `json:"turnId"` + Round *int `json:"round"` + StartedAt string `json:"startedAt"` + EndedAt string `json:"endedAt"` + Lanes []flowTimelineLaneProj `json:"lanes"` +} + +// flowTimelineResponseDTO 是 v0.80 响应体的线上投影（只解本文件要断言的键）。 +type flowTimelineResponseDTO struct { + SessionID string `json:"sessionId"` + Source string `json:"source"` + Rounds []flowTimelineRoundProj `json:"rounds"` + NextBefore string `json:"nextBefore"` + HasMore bool `json:"hasMore"` + Truncated bool `json:"truncated"` + RetentionHours int `json:"retentionHours"` + RetentionEnabled bool `json:"retentionEnabled"` +} + +func decodeFlowTimelineResp(t *testing.T, body []byte) flowTimelineResponseDTO { + t.Helper() + var out flowTimelineResponseDTO + if err := json.Unmarshal(decodeBody(t, body)["data"], &out); err != nil { + t.Fatalf("decode flow timeline data: %v (body=%s)", err, body) + } + return out +} + +// timelineURL 拼出本端点的 URL（limit / before 为空时省略对应参数）。 +func timelineURL(ts *httptest.Server, session, limit, before string) string { + u := ts.URL + "/api/v1/admin/flow/timeline?sessionId=" + session + if limit != "" { + u += "&limit=" + limit + } + if before != "" { + u += "&before=" + before + } + return u +} + +// getTimeline 发一次请求并把解好的响应交回（非 200 即 Fatal）。 +func getTimeline(t *testing.T, ts *httptest.Server, tok, session, limit, before string) flowTimelineResponseDTO { + t.Helper() + resp, body := doReq(t, http.MethodGet, timelineURL(ts, session, limit, before), nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + return decodeFlowTimelineResp(t, body) +} + +// runParallelRound 跑一轮**真并行两路**（网关那条路径：`WithTurnID` 给权威回合标识、 +// `FlowOwnerKey` 给各自的归属桶）。两路**先都起、再分别收尾** —— 只有这样它们才共享 +// 同一个轮次号（发布器的轮次号只在「上一轮的路全部收尾」后才推进）。 +func runParallelRound(p *loop.FlowPublisher, session, laneA, laneB string) { + for _, lane := range []string{laneA, laneB} { + owner := loop.FlowOwnerKey(session, lane) + p.StartRoundWithTurnID(loop.WithTurnID(context.Background(), lane), owner, "输入-"+lane) + } + for _, lane := range []string{laneA, laneB} { + owner := loop.FlowOwnerKey(session, lane) + sp := &trace.Span{ID: "span-" + lane, Name: "llm.call", Kind: trace.SpanKindLLM, StartedAt: time.Now()} + p.SpanStarted(owner, sp) + sp.Finish(trace.SpanStatusOK) + p.SpanFinished(owner, sp) + p.EndRound(owner, "回复-"+lane) + } +} + +// tickFlow 让相邻两个回合的开始时刻**落在不同的毫秒**（发布器的时刻来自真实时钟， +// 而时间轴的游标精度就是毫秒）。没有它，一串瞬间跑完的回合会挤在同一毫秒里。 +func tickFlow() { time.Sleep(3 * time.Millisecond) } + +// TestAdminFlowTimeline_MultiRoundAscendingWithParallelLanes_正常 是本文件的核心断言： +// 三轮按**时间正序**返回，其中真并行那一轮的**两路**都在同一个轮次元素里（`lanes` 两条， +// 各带自己的路标识与图），最新一轮来自内存的那一份**不重复**成第二条。 +func TestAdminFlowTimeline_MultiRoundAscendingWithParallelLanes_正常(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "cyrene.db") + _, store := openFlowGraphStore(t, path) + pub := flowPublisherWithStore(store, func(string) bool { return true }) + _, ts, jwt := buildEnv(t, cfgDefault(), Deps{Flow: pub, FlowGraphs: store}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + runFlowRound(pub, "web", "第一轮", "答一") + tickFlow() + runParallelRound(pub, "web", "turn-a", "turn-b") + tickFlow() + runFlowRound(pub, "web", "第三轮", "答三") + + got := getTimeline(t, ts, tok, "web", "", "") + if got.Source != "persisted" { + t.Fatalf("source = %q, want persisted（主体来自落盘）", got.Source) + } + if len(got.Rounds) != 3 { + t.Fatalf("轮数 = %d, want 3（真并行的两路属**同一轮**）: %+v", len(got.Rounds), got.Rounds) + } + rounds := roundNos(t, got.Rounds) + if rounds[0] == -1 || rounds[1] == -1 || rounds[2] == -1 { + t.Fatalf("轮次号出现 null（未知），但每个回合都带轮次号: %v", rounds) + } + if rounds[0] >= rounds[1] || rounds[1] >= rounds[2] { + t.Fatalf("轮次号 %v 不是**时间正序**（发布器的轮次号 = 该轮首路的建图序号，允许跳号但必须递增）", rounds) + } + parallel := got.Rounds[1] + if len(parallel.Lanes) != 2 { + t.Fatalf("真并行那一轮的 lanes = %d, want 2: %+v", len(parallel.Lanes), parallel.Lanes) + } + if got.Rounds[0].TurnID != got.Rounds[0].Lanes[0].Graph.ID { + t.Fatalf("轮的 turnId = %q, want 首路的图 id %q", got.Rounds[0].TurnID, got.Rounds[0].Lanes[0].Graph.ID) + } + if parallel.Lanes[0].TurnID != "turn-a" || parallel.Lanes[1].TurnID != "turn-b" { + t.Fatalf("各路的 turnId = %q/%q, want turn-a/turn-b（路标识，不是图 id）", + parallel.Lanes[0].TurnID, parallel.Lanes[1].TurnID) + } + if parallel.Lanes[0].Graph.ID == parallel.Lanes[1].Graph.ID { + t.Fatalf("两路的图 id 撞了：%q", parallel.Lanes[0].Graph.ID) + } + // 最新一轮：落盘一份 + 内存一份是同一条路 ⇒ 必须去重成一条。 + if lanes := len(got.Rounds[2].Lanes); lanes != 1 { + t.Fatalf("最新一轮 lanes = %d, want 1（内存那份与落盘那份是同一条路）", lanes) + } + for i, r := range got.Rounds { + if r.StartedAt == "" || r.EndedAt == "" { + t.Fatalf("rounds[%d] 的 startedAt/endedAt 必须有值（这些轮都已收尾）: %+v", i, r) + } + if r.EndedAt < r.StartedAt { + t.Fatalf("rounds[%d] endedAt(%s) 早于 startedAt(%s)", i, r.EndedAt, r.StartedAt) + } + } +} + +// TestAdminFlowTimeline_InFlightRoundHasNoEndedAt_边界 钉住「在飞轮」的线上形状：还没 +// 收尾的那一轮照常出现在时间轴里（内存是它的唯一来源），但 `endedAt` 是**空串** —— +// 客户端据此显示「进行中」，而绝不被喂一个假的结束时刻。 +func TestAdminFlowTimeline_InFlightRoundHasNoEndedAt_边界(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "cyrene.db") + _, store := openFlowGraphStore(t, path) + // 发布器**不挂 recorder**：这一轮只在内存里，一行都不落盘。 + pub := loop.NewFlowPublisher(func(string, any) {}, func(string) bool { return true }) + _, ts, jwt := buildEnv(t, cfgDefault(), Deps{Flow: pub, FlowGraphs: store}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + pub.StartRound("web", "还在跑的这轮") + sp := &trace.Span{ID: "span-llm", Name: "llm.call", Kind: trace.SpanKindLLM, StartedAt: time.Now()} + pub.SpanStarted("web", sp) + sp.Finish(trace.SpanStatusOK) + pub.SpanFinished("web", sp) + + got := getTimeline(t, ts, tok, "web", "", "") + if got.Source != "live" { + t.Fatalf("source = %q, want live（这一页全部来自内存）", got.Source) + } + if len(got.Rounds) != 1 { + t.Fatalf("轮数 = %d, want 1（在飞的那一轮也要在时间轴上）: %+v", len(got.Rounds), got.Rounds) + } + if got.Rounds[0].EndedAt != "" { + t.Fatalf("endedAt = %q, want 空串（这一轮还没收尾，不许编一个结束时刻）", got.Rounds[0].EndedAt) + } + if got.Rounds[0].StartedAt == "" || got.NextBefore != got.Rounds[0].StartedAt { + t.Fatalf("startedAt/nextBefore = %q/%q, want 同一非空时刻", + got.Rounds[0].StartedAt, got.NextBefore) + } + if got.Truncated { + t.Fatal("truncated = true，但落盘源已接线、库里也确实没有更早的轮次") + } +} + +// TestAdminFlowTimeline_BeforeCursorPagesWithoutGapOrDup_正常 钉住分页：`limit=2` 逐页向 +// 更早翻，各页**不重不漏**、`hasMore` 与 `nextBefore` 如实、翻到底后是空数组。 +func TestAdminFlowTimeline_BeforeCursorPagesWithoutGapOrDup_正常(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "cyrene.db") + _, store := openFlowGraphStore(t, path) + base := time.Now().UTC().Add(-4 * time.Hour) + for round := 1; round <= 4; round++ { + recordTimelineGraph(t, store, "web", round, fmt.Sprintf("turn-%d", round), base.Add(time.Duration(round)*time.Hour)) + } + _, ts, jwt := buildEnv(t, cfgDefault(), Deps{FlowGraphs: store}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + page1 := getTimeline(t, ts, tok, "web", "2", "") + if len(page1.Rounds) != 2 || !page1.HasMore || page1.NextBefore == "" { + t.Fatalf("第 1 页 = %d 轮 / hasMore=%v / nextBefore=%q, want 2 / true / 非空", + len(page1.Rounds), page1.HasMore, page1.NextBefore) + } + if got := fmt.Sprint(roundNos(t, page1.Rounds)); got != "[3 4]" { + t.Fatalf("第 1 页轮次 = %s, want [3 4]（最新两轮、正序）", got) + } + page2 := getTimeline(t, ts, tok, "web", "2", page1.NextBefore) + if got := fmt.Sprint(roundNos(t, page2.Rounds)); got != "[1 2]" { + t.Fatalf("第 2 页轮次 = %s, want [1 2]", got) + } + // 游标也接受 Unix 毫秒（与 `llm/usage` / 渠道事件同一解析口径），且与 RFC3339 同解。 + cursor, err := time.Parse(time.RFC3339, page1.NextBefore) + if err != nil { + t.Fatalf("nextBefore %q 不是 RFC3339: %v", page1.NextBefore, err) + } + byMillis := getTimeline(t, ts, tok, "web", "2", fmt.Sprint(cursor.UnixMilli())) + if got := fmt.Sprint(roundNos(t, byMillis.Rounds)); got != "[1 2]" { + t.Fatalf("毫秒游标那一页轮次 = %s, want [1 2]（与 RFC3339 同一时刻应同解）", got) + } + if page2.HasMore { + t.Fatal("第 2 页 hasMore = true，但库里只有四轮") + } + page3 := getTimeline(t, ts, tok, "web", "2", page2.NextBefore) + if len(page3.Rounds) != 0 || page3.HasMore { + t.Fatalf("翻到底 = %d 轮 / hasMore=%v, want 0 / false", len(page3.Rounds), page3.HasMore) + } + seen := map[int]int{} + for _, page := range []flowTimelineResponseDTO{page1, page2, page3} { + for _, r := range page.Rounds { + seen[*r.Round]++ + } + } + for round := 1; round <= 4; round++ { + if seen[round] != 1 { + t.Fatalf("第 %d 轮出现 %d 次, want 1（翻页不重不漏；全部=%v）", round, seen[round], seen) + } + } +} + +// TestAdminFlowTimeline_LiveOnlyWhenStoreUnwired_正常 钉住降级口径：只接了内存图源 +// （落盘未接线）时仍能给出**最近一轮**的时间轴，但 `source`/`truncated` 如实说明 +// 「更早的取不到」—— 老后端形态下前端据此提示「当前只支持最近一轮」。 +func TestAdminFlowTimeline_LiveOnlyWhenStoreUnwired_正常(t *testing.T) { + t.Parallel() + jwt := mustJWT(t) + pub := loop.NewFlowPublisher(func(string, any) {}, func(string) bool { return true }) + _, ts, _ := buildEnv(t, cfgDefault(), Deps{Auth: jwt, Flow: pub}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + runFlowRound(pub, "web", "问题", "回答") + + got := getTimeline(t, ts, tok, "web", "", "") + if got.Source != "live" { + t.Fatalf("source = %q, want live（落盘未接线）", got.Source) + } + if !got.Truncated { + t.Fatal("truncated = false，但落盘未接线 ⇒ 更早的轮次取不到，必须如实标注") + } + if len(got.Rounds) != 1 || len(got.Rounds[0].Lanes) != 1 { + t.Fatalf("轮/路 = %d/%d, want 1/1: %+v", len(got.Rounds), len(got.Rounds[0].Lanes), got.Rounds) + } + if got.RetentionEnabled || got.RetentionHours != 0 { + t.Fatalf("保留期 = %d h / enabled=%v, want 0 / false（没有落盘源就没有保留期口径）", + got.RetentionHours, got.RetentionEnabled) + } +} + +// TestAdminFlowTimeline_UnwiredIsHonest_异常 钉住未接线的诚实口径：内存与落盘**都没接** +// 时回 `500 flow_unavailable`，**不是**一份空 `rounds` —— 后者会把「图源没接线」伪装成 +// 「这台会话没有轮次」（与老出口同一口径）。 +func TestAdminFlowTimeline_UnwiredIsHonest_异常(t *testing.T) { + env := newEnv(t, nil, false) // newEnv 不注入 Flow / FlowGraphs + + resp, body := doReq(t, http.MethodGet, env.ts.URL+"/api/v1/admin/flow/timeline", nil, env.token(t, auth.RoleOwner)) + if resp.StatusCode != http.StatusInternalServerError { + t.Fatalf("status = %d, want 500 (body=%s)", resp.StatusCode, body) + } + var eobj struct { + Code string `json:"code"` + } + if err := json.Unmarshal(decodeBody(t, body)["error"], &eobj); err != nil { + t.Fatalf("error envelope: %v (body=%s)", err, body) + } + if eobj.Code != flowUnavailableCode { + t.Fatalf("error code = %q, want %q", eobj.Code, flowUnavailableCode) + } +} + +// TestAdminFlowTimeline_BadParamsAre400_边界 钉住参数错误**不静默**：非法 `limit` / +// 非法 `before` 一律 400，绝不当成缺省值继续跑（那会把一次拼错变成一份看似正常的响应）。 +func TestAdminFlowTimeline_BadParamsAre400_边界(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "cyrene.db") + _, store := openFlowGraphStore(t, path) + _, ts, jwt := buildEnv(t, cfgDefault(), Deps{FlowGraphs: store}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + for _, query := range []string{"limit=abc", "before=昨天"} { + resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/flow/timeline?"+query, nil, tok) + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("%s status = %d, want 400 (body=%s)", query, resp.StatusCode, body) + } + } +} + +// TestAdminFlowTimeline_RoleGate 钉住角色门与老出口同档（owner/admin 可读、viewer 403、 +// 匿名 401）：不存在「WS 帧收不到、REST 看得到」的旁路。 +func TestAdminFlowTimeline_RoleGate(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "cyrene.db") + _, store := openFlowGraphStore(t, path) + _, ts, jwt := buildEnv(t, cfgDefault(), Deps{FlowGraphs: store}) + url := ts.URL + "/api/v1/admin/flow/timeline" + + resp, body := doReq(t, http.MethodGet, url, nil, signAny(t, jwt, "acct", auth.RoleAdmin)) + if resp.StatusCode != http.StatusOK { + t.Fatalf("admin status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + resp, body = doReq(t, http.MethodGet, url, nil, signAny(t, jwt, "acct", auth.RoleViewer)) + if resp.StatusCode != http.StatusForbidden { + t.Fatalf("viewer status = %d, want 403 (body=%s)", resp.StatusCode, body) + } + resp, _ = doReq(t, http.MethodGet, url, nil, "") + if resp.StatusCode != http.StatusUnauthorized { + t.Fatalf("anonymous status = %d, want 401", resp.StatusCode) + } +} + +// TestAdminFlowTimeline_ExpiredRowsAbsentRetentionReported_正常 钉住保留期口径： +// 已过保留期、被清扫掉的轮次**不在**返回值里（绝不静默补一个空块），而响应把保留期 +// 两个字段如实给出 —— 前端的「更早的轮次已过保留期」这句话只能由它们推出来。 +func TestAdminFlowTimeline_ExpiredRowsAbsentRetentionReported_正常(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "cyrene.db") + conn, _ := openFlowGraphStore(t, path) + retention := 168 * time.Hour + store, err := flowgraph.NewStore(flowgraph.Config{ + DB: conn, Log: gwTestLogger(), GCInterval: time.Hour, Retention: retention, + }) + if err != nil { + t.Fatalf("new store with retention: %v", err) + } + now := time.Now().UTC() + recordTimelineGraph(t, store, "web", 1, "turn-old", now.Add(-200*time.Hour)) + recordTimelineGraph(t, store, "web", 2, "turn-new", now.Add(-time.Hour)) + if _, err := conn.ExecContext(context.Background(), + `UPDATE flow_graphs SET created_at = ? WHERE turn_id = ?`, + now.Add(-200*time.Hour).Format("2006-01-02T15:04:05.000Z"), "web-1#turn-old"); err != nil { + t.Fatalf("把第 1 轮做旧: %v", err) + } + if stats, err := store.Sweep(context.Background()); err != nil { + t.Fatalf("sweep: %v", err) + } else if stats.Deleted == 0 { + t.Fatal("清扫没有删掉任何行，夹具没做到「已过保留期」这一步") + } + _, ts, jwt := buildEnv(t, cfgDefault(), Deps{FlowGraphs: store}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + got := getTimeline(t, ts, tok, "web", "", "") + if got.RetentionHours != int(retention/time.Hour) || !got.RetentionEnabled { + t.Fatalf("保留期 = %d h / enabled=%v, want %d / true", + got.RetentionHours, got.RetentionEnabled, int(retention/time.Hour)) + } + if rounds := fmt.Sprint(roundNos(t, got.Rounds)); rounds != "[2]" { + t.Fatalf("轮次 = %s, want [2]（过期那轮已被清扫，绝不出现在时间轴里）", rounds) + } +} + +// TestAdminFlowTimeline_UnknownSessionIsHonestEmpty_边界 钉住空态：发布会话范围外的会话 +// （隔离渠道会话）从不建图 ⇒ `rounds:[]` + `source:"none"`，绝不用别的会话顶替。 +func TestAdminFlowTimeline_UnknownSessionIsHonestEmpty_边界(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "cyrene.db") + _, store := openFlowGraphStore(t, path) + pub := flowPublisherWithStore(store, func(session string) bool { return session == "web" }) + _, ts, jwt := buildEnv(t, cfgDefault(), Deps{Flow: pub, FlowGraphs: store}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + runFlowRound(pub, "web", "主人说话", "回应") + + resp, body := doReq(t, http.MethodGet, timelineURL(ts, "sess-im-stranger", "", ""), nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + if !jsonHasEmptyRounds(t, body) { + t.Fatalf("空时间轴的 rounds 必须是 `[]`（客户端不必判空 null）: %s", body) + } + got := decodeFlowTimelineResp(t, body) + if len(got.Rounds) != 0 || got.Source != "none" { + t.Fatalf("白名单外会话 = %d 轮 / source=%q, want 0 / none", len(got.Rounds), got.Source) + } +} + +// recordTimelineGraph 直接经落盘出口写一轮图（不经发布器）：时刻**显式给定**， +// 因此分页/定序类断言完全确定（不依赖真实时钟的毫秒粒度）。 +func recordTimelineGraph(t *testing.T, store *flowgraph.SQLStore, session string, round int, lane string, started time.Time) { + t.Helper() + g := trace.NewRunGraph(session) + g.SetTurnID(lane) + g.SetSeq(round) + g.Start(round, started) + g.SeedInput("输入-" + lane) + g.SeedOutput("回复-" + lane) + if err := store.RecordRound(g.Snapshot()); err != nil { + t.Fatalf("落盘轮 %d: %v", round, err) + } +} + +// roundNos 取出各轮的轮次号（断言用；`null` 记为 -1，让它在断言里显形而不是被当成 0）。 +func roundNos(_ *testing.T, rounds []flowTimelineRoundProj) []int { + out := make([]int, 0, len(rounds)) + for _, r := range rounds { + if r.Round == nil { + out = append(out, -1) + continue + } + out = append(out, *r.Round) + } + return out +} + +// jsonHasEmptyRounds 报告响应体里 `rounds` 是否序列化成空数组（而不是 null）。 +func jsonHasEmptyRounds(t *testing.T, body []byte) bool { + t.Helper() + var envelope struct { + Rounds []json.RawMessage `json:"rounds"` + } + if err := json.Unmarshal(decodeBody(t, body)["data"], &envelope); err != nil { + t.Fatalf("decode data: %v (body=%s)", err, body) + } + return envelope.Rounds != nil && len(envelope.Rounds) == 0 +} diff --git a/internal/services/gateway/routes.go b/internal/services/gateway/routes.go index 0551996..c36bd9d 100644 --- a/internal/services/gateway/routes.go +++ b/internal/services/gateway/routes.go @@ -204,6 +204,11 @@ func (s *Server) registerAdminRoutes( // 只读、不审计；与同一张图的 WS 下行帧同档（只发 owner|admin， // design/19 §3「flow 可视化仅 owner/admin」），因此不存在绕过帧门的旁路。 mux.Handle("GET /api/v1/admin/flow/graph", writeAuth(http.HandlerFunc(s.adminFlowGraph))) + // 多轮时间轴（切片 W1，**v0.80**；design/19 §7）：把**所有轮次按时间正序串起来** + // （每轮含该轮每一路），`before` 游标向更早翻页，响应另带保留期口径供面板如实标注 + // 「更早的轮次已过保留期」。与上面的最近一轮快照**同门同档**（writeAuth = owner|admin）、 + // 同样只读不审计；老出口逐字不变，老客户端零影响。 + mux.Handle("GET /api/v1/admin/flow/timeline", writeAuth(http.HandlerFunc(s.adminFlowTimeline))) // 配置事实面（切片 P3，**v0.68**）：记忆整理 7 键 + 安静时段两键的**进程生效值** // 与声明默认值。只读、不审计、**不读 `.env`**、**不含任何凭据**；与消费它的面板页 // （记忆管理 / 后台思考）同门，故走 owner|admin 档而不是 viewer 监控档。