From 2bf3b536dde9b0417f6b6dfa411a568f05b1a0a6 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Tue, 6 Oct 2026 21:16:21 +0800 Subject: [PATCH] =?UTF-8?q?fix(flow):=20=E6=97=B6=E9=97=B4=E8=BD=B4?= =?UTF-8?q?=E4=B8=A4=E5=A4=84=E8=AF=9A=E5=AE=9E=E6=80=A7=E7=BC=BA=E9=99=B7?= =?UTF-8?q?=EF=BC=88X1=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ① round 号跨重启被复用 ⇒ 时间轴把多个真实轮次并成一块（真机实测 web 的 r:1 挤了 4 路、 startedAt 跨 13:52→21:05）。修法：发号前从落盘播种（MaxRound，json_extract + json_valid 挡坏行）， 每会话每进程只播一次、失败只 WARN 且不重试、并发首轮共享一次读；接口形状零变化 ② truncated 一个字段兼两个意思 ⇒ 一页没拉完时前端会说「更早的轮次已过保留期」（假话）。 新增 retentionBoundaryReached（= 保留期启用 且 已翻到底 且 现存最早一轮 round>1 ⇒ 确有轮次被清扫）， truncated 语义一字未改；盲区如实登记（整段被清扫会说少、最早一轮落盘失败可能多说） 3 个变异自检（不落账/floor>0/去掉 hasMore 守卫） --- internal/runtime/flowgraph/roundseed.go | 63 +++++ internal/runtime/flowgraph/roundseed_test.go | 114 ++++++++++ internal/runtime/flowgraph/timeline.go | 51 +++++ internal/runtime/flowgraph/timeline_test.go | 48 ++++ internal/runtime/loop/flow.go | 17 +- internal/runtime/loop/flow_roundseed.go | 118 ++++++++++ internal/runtime/loop/flow_roundseed_test.go | 215 ++++++++++++++++++ .../services/gateway/admin_flow_timeline.go | 61 ++++- .../gateway/admin_flow_timeline_test.go | 145 ++++++++++++ 9 files changed, 820 insertions(+), 12 deletions(-) create mode 100644 internal/runtime/flowgraph/roundseed.go create mode 100644 internal/runtime/flowgraph/roundseed_test.go create mode 100644 internal/runtime/loop/flow_roundseed.go create mode 100644 internal/runtime/loop/flow_roundseed_test.go diff --git a/internal/runtime/flowgraph/roundseed.go b/internal/runtime/flowgraph/roundseed.go new file mode 100644 index 0000000..07810cf --- /dev/null +++ b/internal/runtime/flowgraph/roundseed.go @@ -0,0 +1,63 @@ +package flowgraph + +// 本文件是**轮次计数器的播种读侧**（design/19 §6.1「`graph.round`：每会话单调」； +// 切片 X1，2026-10-06）。 +// +// 它补的是真机挖出的第一处真缺陷：发布器的轮次号是**进程内**计数 +// （`loop.FlowPublisher` 的 `rounds`/`seq`），后端一重启就从 1 重来，而落盘的 +// `flow_graphs` 保留 168h ⇒ 时间轴（`Timeline`）按轮次号分组时把**三个真实轮次并成 +// 一块**。真机实测（`session=web`）：分组键 `r:1` 下有 13:52:53 / 15:42:27 / 18:40:44 +// 三次真实的轮，块标题「第 1 轮 · 3 路」是失真的。 +// +// 本文件只提供**一个只读出口**（`MaxRound`）：把一个会话**已经落盘**的最大轮次号交回 +// 调用方，让它把进程内的计数器播到那个值之上。播种与发号在 `loop` 侧 +// （`internal/runtime/loop/flow_roundseed.go`）—— 本包不认识发布器、不判会话、也绝不 +// 改任何会话的状态（与 `Timeline` 同一条只读纪律）。 + +import ( + "context" + "fmt" +) + +// maxRoundSQL 取该会话已落盘的最大轮次号（一轮一行，因此聚合面有界）。 +// +// 三条口径与 `timeline.go` 的 `timelineRoundKeySQL` **逐字同源**： +// +// 1. `json_valid` 是**保护**：JSON 破损的行根本不参与解析 —— 否则 `json_extract` +// 会让整条查询报错，把「一行坏数据」升级成「播种整体失败」（坏行不该拖垮重启后的 +// 计数器）。用嵌套 `CASE`（而不是 `json_valid(...) AND ...`）是因为 SQLite 不保证 +// AND 的操作数求值次序，而 CASE 只求值命中的那一支。 +// 2. **非正数不算真实轮次**：`round` 键缺席 / 为 0 / 为负 / 非数字（`CAST` 归 0）的行 +// 都给不出一个「更大的号」；一个会话若只剩下这类行，聚合结果就是 0 或负数 ⇒ Go 侧 +// 归 0 —— 绝不冒充一个真实轮次号（design/19 §6.1 的缺席语义）。 +// 3. 没有可用行时 `MAX` 是 NULL ⇒ `COALESCE` 归 **0**（=「库里没有该会话的轮次」）， +// 调用方据此从 1 起发号（与引入播种之前的行为一致）。 +const maxRoundSQL = `SELECT COALESCE(MAX(CASE WHEN json_valid(graph_json) + THEN CAST(json_extract(graph_json,'$.round') AS INTEGER) END), 0) + FROM flow_graphs + WHERE session_id = ?` + +// MaxRound 返回该会话**已落盘**的最大轮次号；库里没有该会话的可用行时返回 0。 +// +// 语义与边界： +// - **只读**：不写库、不删行、不碰保留期（与 `Graph` / `Latest` / `Timeline` 同档）； +// - **保留期决定它能看多远**：已被 `Sweep` 清掉的轮次不在这里的账上 ⇒ 一个会话若整段 +// 历史都被清扫过，返回值会回落到 0，调用方随即从 1 起发号 —— 那**不会**造成时间轴 +// 上的轮次合并（旧行已经不在了），只是轮次号从头开始（design/19 §7.4 的诚实边界）； +// - `ctx` 的上界由调用方给（发布器侧用一个有界的读取上界，绝不把一次库读挂在会话收尾 +// 或轮次开始上无限等）； +// - 失败**如实上抛**（调用方记 WARN 后退回进程内计数）：宁可让计数器退回修复前的行为， +// 也绝不把「读不到」当成「库里没有轮次」而悄悄发一个会撞旧行的号。 +func (s *SQLStore) MaxRound(ctx context.Context, sessionID string) (int, error) { + if s == nil || s.db == nil { + return 0, fmt.Errorf("flowgraph: query max round: %w", ErrNilDB) + } + var n int + if err := s.db.QueryRowContext(ctx, maxRoundSQL, sessionID).Scan(&n); err != nil { + return 0, fmt.Errorf("flowgraph: 查最大轮次号(%s): %w", sessionID, err) + } + if n < 0 { + return 0, nil + } + return n, nil +} diff --git a/internal/runtime/flowgraph/roundseed_test.go b/internal/runtime/flowgraph/roundseed_test.go new file mode 100644 index 0000000..308d67c --- /dev/null +++ b/internal/runtime/flowgraph/roundseed_test.go @@ -0,0 +1,114 @@ +package flowgraph + +// 本文件钉住**轮次计数器的播种读侧**（design/19 §6.1「`graph.round`：每会话单调」； +// 切片 X1）：`MaxRound` 如实交回该会话**已落盘**的最大轮次号，坏行/未知轮次/别的会话 +// 的行都不参与取值，读不到时如实报错而不是回 0。 +// +// 断言的是**库里读回来的事实**（数值本身），不按日志断言 —— 与 flowgraph_test.go / +// timeline_test.go 同一口径。发布器侧「拿这个值播种之后轮次号跨重启不重用」的判据在 +// `internal/runtime/loop/flow_roundseed_test.go`（同一切片的另一半）。 + +import ( + "context" + "errors" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// maxRoundOf 取某会话的最大轮次号（失败即 Fatal）。 +func maxRoundOf(t *testing.T, st *SQLStore, session string) int { + t.Helper() + n, err := st.MaxRound(context.Background(), session) + if err != nil { + t.Fatalf("MaxRound(%s): %v", session, err) + } + return n +} + +// timelineRoundNoKey 造一行**轮次号缺席**的老图（`round` 键根本不写入 —— Round 为 0 +// 时 `trace` 的 JSON 编码不带该键，与真机上那批旧行同形）。 +func timelineRoundNoKey(id, session string) *trace.FlowGraph { + return &trace.FlowGraph{ + ID: id, + SessionID: session, + StartedAt: testNow().Add(-time.Minute), + EndedAt: testNow(), + Nodes: []trace.FlowNode{{ID: "input", Label: "你好", Kind: "input"}}, + } +} + +// TestMaxRound_PicksLargestPersistedRound_正常 钉住取值本身：多轮取最大，缺失轮次号的 +// 老行（Round 0）与**别的会话**的行都不干扰。 +func TestMaxRound_PicksLargestPersistedRound_正常(t *testing.T) { + t.Parallel() + st, _ := newTestStore(t, nil) + + if got := maxRoundOf(t, st, "web"); got != 0 { + t.Fatalf("空会话 MaxRound = %d, want 0（库里没有该会话的行）", got) + } + recordTimelineRound(t, st, timelineGraph("web", 3, 3, "turn-c", testNow())) + recordTimelineRound(t, st, timelineRoundNoKey("web-old", "web")) + recordTimelineRound(t, st, timelineGraph("web", 7, 9, "turn-g", testNow())) + recordTimelineRound(t, st, timelineGraph("other", 42, 42, "turn-x", testNow())) + + if got := maxRoundOf(t, st, "web"); got != 7 { + t.Fatalf("MaxRound(web) = %d, want 7（最大值；老行与别的会话都不算）", got) + } + if got := maxRoundOf(t, st, "other"); got != 42 { + t.Fatalf("MaxRound(other) = %d, want 42（按会话隔离，绝不串会话）", got) + } + if got := maxRoundOf(t, st, "sess-im-stranger"); got != 0 { + t.Fatalf("MaxRound(白名单外会话) = %d, want 0（它从不落盘）", got) + } +} + +// TestMaxRound_BrokenJSONRowIsIgnored_边界 钉住坏行**不拖垮**播种：`graph_json` 破损 +// 的那一行从不参与解析（SQL 侧 `json_valid` 保护），查询照常给出其余行的最大值。 +func TestMaxRound_BrokenJSONRowIsIgnored_边界(t *testing.T) { + t.Parallel() + st, conn := newTestStore(t, nil) + recordTimelineRound(t, st, timelineGraph("web", 5, 5, "turn-e", testNow())) + recordTimelineRound(t, st, timelineGraph("web", 6, 6, "turn-f", testNow())) + if _, err := conn.ExecContext(context.Background(), + `UPDATE flow_graphs SET graph_json = '{ oops' WHERE turn_id = ?`, "web-6#turn-f"); err != nil { + t.Fatalf("把一行做坏: %v", err) + } + + if got := maxRoundOf(t, st, "web"); got != 5 { + t.Fatalf("MaxRound = %d, want 5（坏行被忽略；它绝不能让整条查询报错）", got) + } +} + +// TestMaxRound_NonPositiveRoundIsUnknown_边界 钉住缺席语义：`round` 为 0 / 负数 / +// 非数字（`CAST` 归 0）的行都算「未知轮次」，绝不冒充一个真实轮次号 —— 因此一个只有 +// 这类行的会话回 0，而不是回一个假的「第 -5 轮」。 +func TestMaxRound_NonPositiveRoundIsUnknown_边界(t *testing.T) { + t.Parallel() + st, conn := newTestStore(t, nil) + recordTimelineRound(t, st, timelineGraph("web", 4, 4, "turn-d", testNow())) + recordTimelineRound(t, st, timelineGraph("neg", 9, 9, "turn-n", testNow())) + if _, err := conn.ExecContext(context.Background(), + `UPDATE flow_graphs SET graph_json = json_set(graph_json, '$.round', -5) WHERE turn_id = ?`, + "neg-9#turn-n"); err != nil { + t.Fatalf("把 round 改成负数: %v", err) + } + + if got := maxRoundOf(t, st, "neg"); got != 0 { + t.Fatalf("MaxRound(只有负轮次号) = %d, want 0（未知不冒充真实轮次）", got) + } + if got := maxRoundOf(t, st, "web"); got != 4 { + t.Fatalf("MaxRound(web) = %d, want 4（别的会话的负值不该影响它）", got) + } +} + +// TestMaxRound_NilDBIsHonest_异常 钉住未接数据库时如实报 `ErrNilDB`：绝不静默回 0 +// —— 那会让发布器把「读不到」当成「库里没有轮次」而发一个会撞旧行的号。 +func TestMaxRound_NilDBIsHonest_异常(t *testing.T) { + t.Parallel() + var nilStore *SQLStore + if _, err := nilStore.MaxRound(context.Background(), "web"); !errors.Is(err, ErrNilDB) { + t.Fatalf("nil store 的 err = %v, want ErrNilDB", err) + } +} diff --git a/internal/runtime/flowgraph/timeline.go b/internal/runtime/flowgraph/timeline.go index 5a6cad4..f063251 100644 --- a/internal/runtime/flowgraph/timeline.go +++ b/internal/runtime/flowgraph/timeline.go @@ -20,6 +20,9 @@ package flowgraph // 3. **查不到就是查不到**：已过保留期的行已被 `gc.go` 清扫、从未落盘的行本就不存在 —— // 两者都不出现在返回值里，本包**不猜**、绝不用别的轮次顶替。保留期口径由调用方经 // `Retention()` 如实转达给客户端（「更早的轮次已过保留期」这句话不能靠前端编）。 +// 4. **清扫事实单独判**（切片 X1 新增，见 `EarlierRoundsSwept`）：`Timeline` 的第二个 +// 返回值只回答「本次是不是全部」（`hasMore`），它**不**回答「更早的轮次是不是被 +// 保留期清扫掉了」—— 那是另一件事、另一个判据。 import ( "context" @@ -284,6 +287,54 @@ func mergeLane(byKey map[string]*TimelineRound, g *trace.FlowGraph, createdAt st 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 { diff --git a/internal/runtime/flowgraph/timeline_test.go b/internal/runtime/flowgraph/timeline_test.go index acd0523..018922e 100644 --- a/internal/runtime/flowgraph/timeline_test.go +++ b/internal/runtime/flowgraph/timeline_test.go @@ -240,6 +240,54 @@ func TestTimeline_EmptyAndNilDB_边界(t *testing.T) { } } +// TestEarlierRoundsSwept_RoundFloorIsTheEvidence_正常 钉住**清扫信号**的判据（切片 X1； +// design/19 §7.1）：该会话**完整**的轮次列表里，最早一轮的轮次号 > 1 ⇒ 号更小的轮次曾 +// 发过号且已不在库里 ⇒ 「更早的轮次已被保留期清扫删掉」成立；最早一轮就是 1 ⇒ 没有更早 +// 的被删，绝不说「已过保留期」。 +func TestEarlierRoundsSwept_RoundFloorIsTheEvidence_正常(t *testing.T) { + t.Parallel() + roundsOf := func(rounds ...int) []TimelineRound { + out := make([]TimelineRound, 0, len(rounds)) + for _, r := range rounds { + out = append(out, TimelineRound{Key: fmt.Sprintf("r:%d", r), Round: r}) + } + return out + } + + if EarlierRoundsSwept(roundsOf(1, 2, 3)) { + t.Fatal("最早一轮是第 1 轮 ⇒ 没有更早的被清扫，绝不能说「已过保留期」") + } + if !EarlierRoundsSwept(roundsOf(5, 6)) { + t.Fatal("最早一轮是第 5 轮 ⇒ 号更小的轮次确实发过号、已不在库里 ⇒ 必须为 true") + } + if !EarlierRoundsSwept(roundsOf(3, 7, 12)) { + t.Fatal("乱序/跳号不影响判据：取最小轮次号 > 1 即为 true") + } + // 同一轮的多路（真并行）不改变最早轮次号。 + parallel := append(roundsOf(2, 3), TimelineRound{Key: "r:2", Round: 2, Lanes: nil}) + if !EarlierRoundsSwept(parallel) { + t.Fatal("同一轮两路时最早的仍是第 2 轮 ⇒ true（多路不该把判据翻成 false）") + } +} + +// TestEarlierRoundsSwept_UnknownRoundAndEmptyAreConservative_边界 钉住两条保守口径： +// 轮次号未知的轮次在场（老行 / 在飞路没有号）时判据**不成立**（返回 false，宁少说不多说）， +// 空列表同样为 false —— 别把「不知道」说成「被清扫了」。 +func TestEarlierRoundsSwept_UnknownRoundAndEmptyAreConservative_边界(t *testing.T) { + t.Parallel() + + if EarlierRoundsSwept(nil) || EarlierRoundsSwept([]TimelineRound{}) { + t.Fatal("没有轮次 ⇒ 判据不成立（绝不是「已过保留期」）") + } + mixed := []TimelineRound{ + {Key: "t:web-old-1", Round: 0}, + {Key: "r:5", Round: 5}, + } + if EarlierRoundsSwept(mixed) { + t.Fatal("有轮次号未知的轮次在场 ⇒ 号不可信 ⇒ 必须保守为 false") + } +} + // roundKeys 取出轮次键序列（断言用）。 func roundKeys(rounds []TimelineRound) []string { out := make([]string, 0, len(rounds)) diff --git a/internal/runtime/loop/flow.go b/internal/runtime/loop/flow.go index b5bb100..78a6d0a 100644 --- a/internal/runtime/loop/flow.go +++ b/internal/runtime/loop/flow.go @@ -82,9 +82,13 @@ type FlowPublisher struct { // （`{会话}-{序号}[#{路}]`）。用序号而不是轮次号做 id 的第二个分量，是为了让 // 「同一轮里的第二条路」也拿到一个全新的 id —— 否则两路会撞同一个 // `flow_graphs.turn_id`（唯一键），后收尾的那一路把前一路整行顶掉。 + // + // **跨重启**：本进程第一次给某会话发号之前，`ensureRoundSeeded` 会把它播到该会话 + // **已落盘**的最大轮次号之上（见 flow_roundseed.go），因此重启不会重用轮次号。 seq map[string]int // rounds 是每会话当前的**轮次号**（= 该轮第一条路的建图序号）。同一轮的各条路 - // 共享它，客户端据此只保留最新一轮的泳道。 + // 共享它，客户端据此只保留最新一轮的泳道。它**跨重启单调**（同上：从落盘侧播种）， + // 因此时间轴按它分组时不会把重启前后的两轮并成一块。 rounds map[string]int // roundOf 是「归属桶 → 它属于哪一轮」。 roundOf map[string]int @@ -92,6 +96,10 @@ type FlowPublisher struct { // （集合为空），因此它是轮次号推进的唯一依据，也是内存不随路数无界增长的 // 依据（新一轮起时丢掉更早轮次的桶）。 inflight map[string]map[string]bool + // seedDone 是「该会话的播种动作已完成」的标记（成功或放弃，见 flow_roundseed.go）； + // seeding 是正在播种的会话 → 完成信号，使同一会话的并发首轮等这一次播种。 + seedDone map[string]bool + seeding map[string]chan struct{} // record 是每轮收尾的落盘出口（design/19 §4 P3；切片 F2）。nil = 不落盘 // （保持 F1 之前的纯内存行为，且测试可只测发布）。 record RoundRecorder @@ -114,6 +122,8 @@ func NewFlowPublisher(publish func(subject string, payload any), allow func(sess rounds: make(map[string]int), roundOf: make(map[string]int), inflight: make(map[string]map[string]bool), + seedDone: make(map[string]bool), + seeding: make(map[string]chan struct{}), } for _, opt := range opts { if opt != nil { @@ -161,11 +171,15 @@ func (p *FlowPublisher) StartRoundWithTurnID(ctx context.Context, owner, inputTe // 轮次号只在**上一轮的路全部收尾**之后推进（inflight 为空），因此「同一条消息 // 引出的两路」共享同一个轮次号；新一轮开始时丢掉更早轮次的桶，内存因此不随路数 // 无界增长（图的历史在落盘里，见 flowgraph 包）。 +// +// 发号之前先 `ensureRoundSeeded`：本进程第一次给该会话发号时从落盘侧播种，使轮次号 +// **跨重启单调**（否则重启后从 1 重来，会把库里的旧轮次并进同一块，见 flow_roundseed.go）。 func (p *FlowPublisher) startRound(owner, turnID, inputText string) { session, _ := splitFlowOwner(owner) if session == "" || owner == "" || !p.allowed(session) { return } + p.ensureRoundSeeded(session) p.mu.Lock() defer p.mu.Unlock() round := p.beginRoundLocked(session, owner) @@ -184,6 +198,7 @@ func (p *FlowPublisher) startRound(owner, turnID, inputText string) { } // beginRoundLocked 分配（或沿用）本轮次号并清掉更早轮次的桶。调用方必须持有 p.mu。 +// 计数器由 `ensureRoundSeeded` 在发号之前播种（跨重启单调，见 flow_roundseed.go）。 func (p *FlowPublisher) beginRoundLocked(session, owner string) int { p.seq[session]++ if len(p.inflight[session]) == 0 { diff --git a/internal/runtime/loop/flow_roundseed.go b/internal/runtime/loop/flow_roundseed.go new file mode 100644 index 0000000..7d61e5b --- /dev/null +++ b/internal/runtime/loop/flow_roundseed.go @@ -0,0 +1,118 @@ +package loop + +// 本文件把**轮次号跨重启不重用**这件事修好（切片 X1；design/19 §6.1「`graph.round`： +// 每会话单调」）。它补的是真机挖出的第一处真缺陷：轮次号原本只是**进程内**计数 +// （`flow.go` 的 `seq` / `rounds`），后端一重启就从 1 重来，而落盘的 `flow_graphs` +// 保留 168h ⇒ 时间轴（`flowgraph.Timeline`）按轮次号分组时把**三个真实轮次并成一块** +// （真机实测 `session=web` 的 `r:1` 三路 `startedAt` = 13:52:53 / 15:42:27 / 18:40:44， +// 块标题「第 1 轮 · 3 路」失真）。 +// +// 修法：**本进程第一次**给某会话发号之前，从落盘侧读一次该会话**已落盘的最大轮次号**， +// 把计数器播到它之上 —— 于是重启后的第一条路拿到 `max+1`，时间轴上「重启前那一轮」与 +// 「重启后那一轮」是**两块**、轮次号不重复。 +// +// 三条边界（都刻意如此，不美化）： +// +// 1. **播种是一次尽力而为的动作**：读失败只记一条 WARN，本进程对该会话退回**进程内 +// 计数**（= 修复之前的行为，轮次号可能重用）；失败后**不再重试** —— 绝不把一次库读 +// 挂在每个轮次开始上。 +// 2. **只在第一次调用时阻塞**：同一会话的并发首轮（真并行两路）会**等这一次播种结束**， +// 而不是各自用未播种的计数器发号；播种查询在**放锁之后**做，因此它绝不挡住别的会话 +// 的 span 发布。 +// 3. **不猜、不造**：读侧只回一个「已落盘的最大轮次号」，库为空（或该会话整段历史已被 +// 保留期清扫）时它就是 0 ⇒ 计数器从 1 起，与引入播种之前逐字相同。 +// +// 能力来自哪里：落盘出口（`WithRoundRecorder`）与播种来源是**同一个存储的两种能力** +// （生产实现都是 `*flowgraph.SQLStore`），因此发布器直接对注入的落盘出口做**可选接口** +// 断言，不为它另开一条接线。实现方没有该能力（测试桩 / 别的实现）⇒ 不播种，行为与引入 +// 本切片之前逐字相同。 + +import ( + "context" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" +) + +// RoundSeedReader 报告某会话**已落盘**的最大轮次号（0 = 库里没有该会话的可用行）。 +// 生产实现是 `*flowgraph.SQLStore`（`MaxRound`）。 +// +// 它是**可选能力**：`RoundRecorder` 的实现方不必实现它 —— 不实现就等于不播种（发布器 +// 保持进程内计数），因此引入本接口不破坏既有落盘出口。 +type RoundSeedReader interface { + MaxRound(ctx context.Context, sessionID string) (int, error) +} + +// roundSeedTimeout 是单次播种读取的时长上界：轮次开始路径上的一次库读必须**有界**， +// 否则一次库卡死会把主人的第一条回复挂住。（5s 的 `flowgraph` 落盘上界不适用：那在 +// 会话收尾路径上，而这里在开轮路径上。） +const roundSeedTimeout = 2 * time.Second + +// ensureRoundSeeded 在**本进程第一次**为某会话分配轮次号之前，把该会话的计数器播到 +// 「落盘侧的最大轮次号」之上（见本文件头注释的三条边界）。 +// +// 调用方：`startRound`（在 `beginRoundLocked` **之前**）—— 只有这样，重启后的第一条路 +// 才拿到 `max+1` 而不是 1。发号出口不止一个时，每个出口都必须走这一步。 +func (p *FlowPublisher) ensureRoundSeeded(sessionID string) { + reader, ok := p.record.(RoundSeedReader) + if !ok { + return // 落盘出口没有播种能力（未接线 / 测试桩 / 别的实现）⇒ 保持进程内计数 + } + p.mu.Lock() + if p.seedDone[sessionID] { + p.mu.Unlock() + return + } + if wait, busy := p.seeding[sessionID]; busy { + // 同一会话的并发首轮：等那一次播种结束再发号，绝不各自用一个未播种的计数器。 + p.mu.Unlock() + <-wait + return + } + wait := make(chan struct{}) + p.seeding[sessionID] = wait + p.mu.Unlock() + + maxRound, err := p.readMaxRound(reader, sessionID) + + p.mu.Lock() + delete(p.seeding, sessionID) + p.seedDone[sessionID] = true // 无论成败都只尝试一次（失败即本进程退回进程内计数） + if err == nil && maxRound > p.seq[sessionID] { + p.seq[sessionID] = maxRound + } + p.mu.Unlock() + close(wait) + if err != nil { + p.logSeedError(sessionID, err) + } +} + +// readMaxRound 带上界地读一次「已落盘的最大轮次号」，并把负数归 0（未知轮次不冒充 +// 真实轮次号：读侧只回正整数或 0，这里是第二道防线）。 +func (p *FlowPublisher) readMaxRound(reader RoundSeedReader, sessionID string) (int, error) { + ctx, cancel := context.WithTimeout(context.Background(), roundSeedTimeout) + defer cancel() + n, err := reader.MaxRound(ctx, sessionID) + if err != nil { + return 0, err + } + if n < 0 { + return 0, nil + } + return n, nil +} + +// logSeedError 把一次播种失败记成带来源的 WARN。它只记会话标识，**绝不打印图内容** +// （与 `logPersistError` 同一条纪律）；logger 未接线时静默。 +// +// 这条日志是「轮次号可能又重用了」的唯一可见信号：主人的面板若看到时间轴把两轮并成 +// 一块，先查这里。 +func (p *FlowPublisher) logSeedError(sessionID string, err error) { + if p.log == nil { + return + } + logger.WithSource(p.log, flowLogSource).Warn("flow round seed read failed", + "session", sessionID, "err", err, + "effect", "round numbers fall back to in-process counting (may be reused)") +} diff --git a/internal/runtime/loop/flow_roundseed_test.go b/internal/runtime/loop/flow_roundseed_test.go new file mode 100644 index 0000000..b5eb61a --- /dev/null +++ b/internal/runtime/loop/flow_roundseed_test.go @@ -0,0 +1,215 @@ +package loop + +// 本文件钉住**轮次计数器跨重启不重用**（切片 X1；design/19 §6.1「`graph.round`：每会话 +// 单调」）。真机缺陷：轮次号原本只是进程内计数，后端一重启就从 1 重来，而落盘保留 168h +// ⇒ 时间轴把**三个真实轮次并成一块**（`session=web` 的 `r:1` 三路 13:52:53 / 15:42:27 / +// 18:40:44）。 +// +// 判据（本文件逐条钉住）： +// - **重启前后各跑一轮 ⇒ 两个轮次号不同**（第二个「进程」的发布器从落盘侧播种）； +// - 播种**每会话只读一次**、失败**只退化为进程内计数**（绝不重试、绝不 panic）； +// - 同一会话的**并发首轮**只读一次，且两路共享同一个轮次号； +// - 落盘出口**没有**播种能力时，行为与引入本切片之前逐字相同。 + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// seededRecorder 是**同时**满足 `RoundRecorder` 与 `RoundSeedReader` 的假存储： +// 落盘（记下每一轮的轮次号）与播种（交回「已落盘的最大轮次号」）两件能力都在同一个 +// 对象上 —— 与生产里同一个 `*flowgraph.SQLStore` 担这两个角色同形。 +type seededRecorder struct { + mu sync.Mutex + rounds []int + reads int + // deadline 记下最近一次 MaxRound 的 ctx 是否**带期限**（钉「读必须有界」）。 + deadline bool + // err 非 nil 时 MaxRound 报错（模拟库读失败）。 + err error + // delay 非零时把 MaxRound 拖住（并发用例用它制造真实的交错）。 + delay time.Duration +} + +func (r *seededRecorder) RecordRound(g *trace.FlowGraph) error { + r.mu.Lock() + defer r.mu.Unlock() + r.rounds = append(r.rounds, g.Round) + return nil +} + +func (r *seededRecorder) MaxRound(ctx context.Context, _ string) (int, error) { + _, hasDeadline := ctx.Deadline() + r.mu.Lock() + r.reads++ + r.deadline = hasDeadline + delay, err := r.delay, r.err + max := 0 + for _, round := range r.rounds { + if round > max { + max = round + } + } + r.mu.Unlock() + if delay > 0 { + select { + case <-time.After(delay): + case <-ctx.Done(): + return 0, ctx.Err() + } + } + if err != nil { + return 0, err + } + return max, nil +} + +// seedReads 返回播种读取的次数（断言「每会话只读一次」）。 +func (r *seededRecorder) seedReads() int { + r.mu.Lock() + defer r.mu.Unlock() + return r.reads +} + +// recordedRounds 返回落盘次序的轮次号序列。 +func (r *seededRecorder) recordedRounds() []int { + r.mu.Lock() + defer r.mu.Unlock() + return append([]int(nil), r.rounds...) +} + +// seededPublisher 造一个把每轮落进 rec 的发布器（生产那条路径：recorder = 播种来源）。 +func seededPublisher(rec RoundRecorder) *FlowPublisher { + return NewFlowPublisher(func(string, any) {}, func(string) bool { return true }, + WithRoundRecorder(rec)) +} + +// TestFlowPublisher_RoundSeedSurvivesRestart_正常 是本切片的**核心判据**（主人给的那条 +// 原话）：「重启前后各跑一轮 ⇒ 时间轴里是两块、标题轮次不重复」。这里用**同一个存储、 +// 两个「进程」**（换发布器 = 换内存；store 保留 = 落盘跨进程）跑两轮，断言两个轮次号 +// **不同**且后者更大。 +func TestFlowPublisher_RoundSeedSurvivesRestart_正常(t *testing.T) { + t.Parallel() + store := &seededRecorder{} + + // 第一个「进程」：空计数器，第一轮拿到的就是 1（与修复前一致）。 + before := seededPublisher(store) + runRoundWithSpans(before, "web") + + // 第二个「进程」：新发布器（空内存）+ 同一个 store（落盘还在）。 + after := seededPublisher(store) + runRoundWithSpans(after, "web") + + got := store.recordedRounds() + if len(got) != 2 { + t.Fatalf("落盘轮数 = %d, want 2: %v", len(got), got) + } + if got[0] != 1 { + t.Fatalf("重启前的轮次号 = %d, want 1（新会话从 1 起）", got[0]) + } + if got[1] <= got[0] { + t.Fatalf("重启后的轮次号 = %d, want > %d（否则时间轴会把两轮并成一块、标题轮次重复）", + got[1], got[0]) + } + if store.seedReads() != 2 { + t.Fatalf("播种读取次数 = %d, want 2（两个「进程」各播一次）", store.seedReads()) + } +} + +// TestFlowPublisher_RoundSeedReadsOncePerSession_正常 钉住播种的**次数**：同一会话跑多轮 +// 只读一次（第二轮起走内存计数），且后续轮次号继续递增（不回到 1）。 +func TestFlowPublisher_RoundSeedReadsOncePerSession_正常(t *testing.T) { + t.Parallel() + store := &seededRecorder{} + p := seededPublisher(store) + + for i := 0; i < 3; i++ { + runRoundWithSpans(p, "web") + } + if got := store.seedReads(); got != 1 { + t.Fatalf("播种读取次数 = %d, want 1（每一轮都去读库是纯浪费）", got) + } + if got := store.recordedRounds(); len(got) != 3 || got[0] >= got[1] || got[1] >= got[2] { + t.Fatalf("轮次号 = %v, want 严格递增的三轮", got) + } + if !store.deadline { + t.Fatal("播种读取必须带**期限**（否则一次库卡死会挂住主人的第一条回复）") + } +} + +// TestFlowPublisher_RoundSeedFailureFallsBack_异常 钉住失败口径：落盘侧读不出来时**只 +// 退化**（本进程退回进程内计数、轮次号照常发），绝不 panic、绝不每轮重试 —— 一次 +// WARN 之后就不再打扰库。 +func TestFlowPublisher_RoundSeedFailureFallsBack_异常(t *testing.T) { + t.Parallel() + store := &seededRecorder{err: errors.New("db down")} + p := seededPublisher(store) + + runRoundWithSpans(p, "web") + runRoundWithSpans(p, "web") + + if got := store.recordedRounds(); len(got) != 2 || got[0] != 1 || got[1] != 2 { + t.Fatalf("轮次号 = %v, want [1 2]（播种失败 ⇒ 退回进程内计数，行为与修复前一致）", got) + } + if got := store.seedReads(); got != 1 { + t.Fatalf("播种读取次数 = %d, want 1（失败后不再重试，绝不把库读挂在每个轮次开始上）", got) + } +} + +// TestFlowPublisher_RoundSeedSkipsRecorderWithoutCapability_边界 钉住可选接口的**向后 +// 兼容**：落盘出口没有播种能力（只实现 `RoundRecorder`）时行为与引入本切片之前逐字 +// 相同 —— 两个「进程」的首轮都从 1 起（这正是修复前的那条行为，断言它没被悄悄改坏）。 +func TestFlowPublisher_RoundSeedSkipsRecorderWithoutCapability_边界(t *testing.T) { + t.Parallel() + rec := &fakeRecorder{} // 只实现 RoundRecorder（flow_persist_test.go 的既有桩） + + first := NewFlowPublisher(func(string, any) {}, func(string) bool { return true }, + WithRoundRecorder(rec)) + runRoundWithSpans(first, "web") + second := NewFlowPublisher(func(string, any) {}, func(string) bool { return true }, + WithRoundRecorder(rec)) + runRoundWithSpans(second, "web") + + if got := rec.count(); got != 2 { + t.Fatalf("落盘次数 = %d, want 2", got) + } + if got := rec.last().Round; got != 1 { + t.Fatalf("第二轮轮次号 = %d, want 1（没有播种能力 ⇒ 与引入播种之前一致）", got) + } +} + +// TestFlowPublisher_ConcurrentFirstLegsShareOneSeedRead_边界 钉住并发首轮：真并行两路 +// **同时**开轮时，播种只读一次，且两路共享同一个轮次号（绝不让第二路用未播种的计数器 +// 单独发号，那是「重启后又并成一块」的另一条路径）。 +func TestFlowPublisher_ConcurrentFirstLegsShareOneSeedRead_边界(t *testing.T) { + t.Parallel() + store := &seededRecorder{delay: 30 * time.Millisecond} + p := seededPublisher(store) + + var wg sync.WaitGroup + for _, lane := range []string{"turn-a", "turn-b"} { + wg.Add(1) + go func(lane string) { + defer wg.Done() + p.StartRoundWithTurnID(WithTurnID(context.Background(), lane), FlowOwnerKey("web", lane), "输入-"+lane) + }(lane) + } + wg.Wait() + + rounds := p.LatestLanes("web") + if len(rounds) != 2 { + t.Fatalf("在飞路数 = %d, want 2（真并行两路）", len(rounds)) + } + if rounds[0].Round != rounds[1].Round { + t.Fatalf("两路的轮次号 = %d/%d, want 相同（同一轮的各路共享轮次号）", + rounds[0].Round, rounds[1].Round) + } + if got := store.seedReads(); got != 1 { + t.Fatalf("播种读取次数 = %d, want 1（并发首轮只播一次）", got) + } +} diff --git a/internal/services/gateway/admin_flow_timeline.go b/internal/services/gateway/admin_flow_timeline.go index d7f02dd..872b842 100644 --- a/internal/services/gateway/admin_flow_timeline.go +++ b/internal/services/gateway/admin_flow_timeline.go @@ -19,8 +19,13 @@ package gateway // 严格早于它的轮次」，响应带 `nextBefore` 供客户端原样回带（分页不重不漏的口径由 // `flowgraph.SQLStore.Timeline` 保证并测试）。 // 4. **保留期如实**：过期轮次已被清扫，不在返回值里；响应给 `retentionHours` / -// `retentionEnabled` / `truncated`，由前端据此标注「更早的轮次已过保留期」—— -// 绝不静默少一块（design/19 §7.1 的原话）。 +// `retentionEnabled` / `truncated` / `retentionBoundaryReached`，由前端据此标注 +// 「更早的轮次已过保留期」——绝不静默少一块（design/19 §7.1 的原话）。 +// 🔴 **两个字段各说一件事**（切片 X1 修的一处真缺陷）：`truncated` 只回答「本次是不是 +// 全部」（`hasMore` 或落盘源未接线），`retentionBoundaryReached` 才回答「更早的轮次 +// 是不是**确实被保留期清扫掉了**」——后者由 `flowgraph.EarlierRoundsSwept` 给出判据 +// （该会话现存最早一轮的轮次号 > 1 ⇒ 号更小的轮次曾落盘、且已被清扫）。混用会让面板 +// 对着**仍然可翻**的历史说「已过保留期」，那是假话。 // // 未接线（内存与落盘**都没接**）时回 `500 flow_unavailable`，与老出口同一码、同一口径： // 绝不回一份空的 `rounds` 把「图源没接线」伪装成「这台会话没有轮次」。 @@ -103,8 +108,25 @@ type adminFlowTimelineResponse struct { // HasMore 报告库里**还有更早的轮次**没被本次返回（用 NextBefore 继续取）。 HasMore bool `json:"hasMore"` // Truncated 报告本次**不是全部**：`hasMore`（分页未到底）或落盘源未接线（只剩内存 - // 最近的这一轮）。前端据此如实标注「更早的轮次未显示」，**绝不静默少一块**。 + // 最近的这一轮）。它是**分页口径**，不是保留期口径 —— 面板据此标注「更早的轮次未 + // 显示」，**绝不静默少一块**。要问「更早的是不是被清扫掉了」请看 + // `retentionBoundaryReached`（两个字段各说一件事，切片 X1 修的正是这处混用）。 Truncated bool `json:"truncated"` + // RetentionBoundaryReached 报告**保留期边界已到达**：该会话**确有更早的轮次已被 + // 保留期清扫删掉**，且本次已取到它现存最早的一端（切片 X1 新增；契约版本待主会话 + // 登记 —— 本字段需一次版本号，建议 v0.81）。 + // + // 判据（三条一起成立，逐条可复核）：① 保留期清扫已启用（`retentionEnabled`）； + // ② `hasMore:false`（本次不是半页 —— 只看半页就说「已过保留期」正是修掉的那句假话）； + // ③ `flowgraph.EarlierRoundsSwept`：该会话现存最早一轮的轮次号 > 1（轮次号自 1 起、 + // 每会话跨重启单调 ⇒ 号更小的轮次曾落盘，而 `flow_graphs` 的唯一删除者就是保留期 + // 清扫）。⇒ **客户端只有看到 `true` 才可以说「更早的轮次已过保留期」**，其余情形一律 + // 说「更早的轮次未显示」（`truncated`）或「只拿到最近一轮」（`source:"live"`）。 + // + // 诚实的盲区（写在契约里，不掩盖）：一个会话若**整段历史都被清扫**（一条可用行都不 + // 剩），发号从 1 重新开始 ⇒ 本字段为 false（**少说**，绝不多说）；有轮次号未知的老行 + // 在场时同样为 false。 + RetentionBoundaryReached bool `json:"retentionBoundaryReached"` // RetentionHours 是落盘保留期（小时，下取整）；`RetentionEnabled` 为 false 时保留期 // 清扫**未启用** ⇒ "更早的轮次" 不是被清扫掉的，别把它说成过期。 // `FLOW_GRAPH_RETENTION` 是既有旋钮（默认 168h），本片**不新增任何配置键**。 @@ -169,17 +191,34 @@ func (s *Server) adminFlowTimeline(w http.ResponseWriter, r *http.Request) { 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, + SessionID: sessionID, + Source: timelineSource(persisted, len(merged) > 0), + Rounds: flowTimelineRoundsOf(merged), + NextBefore: timelineNextBefore(merged), + HasMore: hasMore, + Truncated: hasMore || !storeWired, + // 保留期边界：只有「清扫已启用 + 已到最早一端 + 确有更早的轮次被清扫删掉」三条 + // 同时成立才为 true（判据见 DTO 注释与 design/19 §7.1）。 + RetentionBoundaryReached: retainedEarlierRoundsSwept(merged, hasMore, retention), + RetentionHours: int(retention / time.Hour), + RetentionEnabled: retention > 0, }) } +// retainedEarlierRoundsSwept 是 `retentionBoundaryReached` 的判据本体：清扫已启用 +// （retention > 0）、本次已到该会话现存最早的一端（!hasMore）、且该会话**确有**更早的 +// 轮次已被清扫删掉（`flowgraph.EarlierRoundsSwept` 的轮次号下界判据）。 +// +// 三条缺一不可：少了第一条会说「保留期」却根本没开清扫；少了第二条会对着**仍然可翻**的 +// 历史说「已过保留期」（切片 X1 修掉的那句假话）；少了第三条会把一个刚开张、什么都没被 +// 清掉的会话说成「更早的轮次已过保留期」。 +func retainedEarlierRoundsSwept(merged []flowgraph.TimelineRound, hasMore bool, retention time.Duration) bool { + if retention <= 0 || hasMore { + return false + } + return flowgraph.EarlierRoundsSwept(merged) +} + // timelineLimitParam 解析 `?limit=`：空串 = 未给（0 ⇒ 存储层用缺省 `DefaultTurnsLimit`）； // 非法整数报错（400，绝不静默当缺省）；负值归一成 0；超过硬上限 `MaxTurnsLimit` 即钳住。 func timelineLimitParam(raw string) (int, error) { diff --git a/internal/services/gateway/admin_flow_timeline_test.go b/internal/services/gateway/admin_flow_timeline_test.go index 3622670..5a3a0b0 100644 --- a/internal/services/gateway/admin_flow_timeline_test.go +++ b/internal/services/gateway/admin_flow_timeline_test.go @@ -52,6 +52,7 @@ type flowTimelineResponseDTO struct { NextBefore string `json:"nextBefore"` HasMore bool `json:"hasMore"` Truncated bool `json:"truncated"` + BoundaryReached bool `json:"retentionBoundaryReached"` RetentionHours int `json:"retentionHours"` RetentionEnabled bool `json:"retentionEnabled"` } @@ -386,6 +387,150 @@ func TestAdminFlowTimeline_ExpiredRowsAbsentRetentionReported_正常(t *testing. if rounds := fmt.Sprint(roundNos(t, got.Rounds)); rounds != "[2]" { t.Fatalf("轮次 = %s, want [2]（过期那轮已被清扫，绝不出现在时间轴里）", rounds) } + // 🔴 清扫信号（切片 X1）：一轮**真被清扫删掉**（上面的 Sweep）⇒ 现存最早一轮是第 2 轮 + // ⇒ 保留期边界已到达 ⇒ 面板**可以**说「更早的轮次已过保留期」。这是本字段唯一正当的 + // 出处：不是「返回的轮次比想象中少」，而是「确有轮次被清扫删掉」。 + if !got.BoundaryReached { + t.Fatal("retentionBoundaryReached = false，但第 1 轮确实被 Sweep 删掉、且本次已翻到最早一端") + } + if got.Truncated { + t.Fatal("truncated = true，但本次已经是全部（已翻到最早一端、落盘源已接线）") + } +} + +// TestAdminFlowTimeline_RoundNumbersSurviveRestart_正常 是本切片缺陷①的**真判据** +// （主人给的原话）：「重启前后各跑一轮 ⇒ 时间轴里是两块、标题轮次不重复」。 +// +// 走**生产形状**：真 SQLite 文件库 + 真 store 作落盘出口（发布器的播种能力正是从这个 +// 出口上探到的 ⇒ 这条用例同时钉住「接线零改动，但能力真接上了」）+ 真 HTTP 出口。 +// 「重启」用换发布器（空内存）+ 同一个文件库的新连接/新 store 模拟（与 +// `TestAdminFlowGraph_SurvivesRestart_正常` 同一手法）。 +func TestAdminFlowTimeline_RoundNumbersSurviveRestart_正常(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "cyrene.db") + + // 「重启前」那个进程：跑一轮（轮次号从 1 起，与修复前一致）。 + conn1, store1 := openFlowGraphStore(t, path) + pub1 := flowPublisherWithStore(store1, func(string) bool { return true }) + _, ts1, _ := buildEnv(t, cfgDefault(), Deps{Flow: pub1, FlowGraphs: store1}) + runFlowRound(pub1, "web", "重启前那一轮", "答一") + ts1.Close() + _ = conn1.Close() // 模拟进程退出 + + // 「重启后」那个进程：空内存的新发布器 + 同一个文件库的新连接 / 新 store。 + _, store2 := openFlowGraphStore(t, path) + pub2 := flowPublisherWithStore(store2, func(string) bool { return true }) + _, ts2, jwt2 := buildEnv(t, cfgDefault(), Deps{Flow: pub2, FlowGraphs: store2}) + tok := signAny(t, jwt2, "acct", auth.RoleOwner) + runFlowRound(pub2, "web", "重启后那一轮", "答二") + + got := getTimeline(t, ts2, tok, "web", "", "") + if len(got.Rounds) != 2 { + t.Fatalf("时间轴轮数 = %d, want 2（重启前后各一轮是**两块**，绝不并成一块）: %+v", + len(got.Rounds), got.Rounds) + } + rounds := roundNos(t, got.Rounds) + if rounds[0] <= 0 || rounds[1] <= 0 { + t.Fatalf("轮次号出现 null/0，但两轮都有轮次号: %v", rounds) + } + if rounds[0] == rounds[1] { + t.Fatalf("两轮轮次号相同（%v）⇒ 时间轴会把它们并成一块、标题轮次重复（本切片要治的缺陷）", rounds) + } + if rounds[0] >= rounds[1] { + t.Fatalf("轮次号 %v 不是递增（时间正序）", rounds) + } +} + +// TestAdminFlowTimeline_TruncatedIsNotRetention_正常 钉住**两个字段各说一件事** +// （切片 X1 修的那处真缺陷）：`truncated` 只说「本次不是全部」，`retentionBoundaryReached` +// 才说「更早的轮次确实被保留期清扫掉了」。三步都在同一个会话上走： +// +// ① 只拉一页（还有更早的）⇒ truncated=true 而 boundaryReached=**false**（历史仍可翻， +// 「已过保留期」是假话）； +// ② 翻到底但最早一轮就是第 1 轮 ⇒ 两个都是 false（没有更早的轮次被删，别栽赃给保留期）； +// ③ 同上但落盘源未接线 ⇒ truncated=true（分页口径）而 boundaryReached=false。 +func TestAdminFlowTimeline_TruncatedIsNotRetention_正常(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "cyrene.db") + conn, _ := openFlowGraphStore(t, path) + store, err := flowgraph.NewStore(flowgraph.Config{ + DB: conn, Log: gwTestLogger(), GCInterval: time.Hour, Retention: 168 * time.Hour, + }) + if err != nil { + t.Fatalf("new store with retention: %v", err) + } + base := time.Now().UTC().Add(-3 * time.Hour) + for round := 1; round <= 3; 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) + + // ① 最新那一页（limit=1）：还有更早的可翻。 + page1 := getTimeline(t, ts, tok, "web", "1", "") + if got := fmt.Sprint(roundNos(t, page1.Rounds)); got != "[3]" { + t.Fatalf("第 1 页轮次 = %s, want [3]", got) + } + if !page1.HasMore || !page1.Truncated { + t.Fatalf("第 1 页 hasMore/truncated = %v/%v, want true/true（本次不是全部）", + page1.HasMore, page1.Truncated) + } + if page1.BoundaryReached { + t.Fatal("还有更早的可翻 ⇒ **绝不**能说「更早的轮次已过保留期」（那是对着能翻的历史说假话）") + } + + // ② 翻到最早那一页：本次是全部，但现存最早一轮就是第 1 轮 ⇒ 没有被清扫的轮次。 + page2 := getTimeline(t, ts, tok, "web", "1", page1.NextBefore) + if !page2.HasMore { + t.Fatal("第 2 页 hasMore = false，但后面还有第 1 轮") + } + last := getTimeline(t, ts, tok, "web", "1", page2.NextBefore) + if got := fmt.Sprint(roundNos(t, last.Rounds)); got != "[1]" { + t.Fatalf("最后一页轮次 = %s, want [1]", got) + } + if last.HasMore || last.Truncated { + t.Fatalf("最后一页 hasMore/truncated = %v/%v, want false/false（已翻到底）", + last.HasMore, last.Truncated) + } + if last.BoundaryReached { + t.Fatal("现存最早一轮是第 1 轮 ⇒ 没有更早的轮次被清扫 ⇒ 不得说「已过保留期」") + } + + // ③ 落盘源未接线：truncated 仍如实为 true（分页口径），保留期信号不越权。 + pubOnly := loop.NewFlowPublisher(func(string, any) {}, func(string) bool { return true }) + _, ts2, jwt2 := buildEnv(t, cfgDefault(), Deps{Flow: pubOnly}) + runFlowRound(pubOnly, "web", "问题", "回答") + live := getTimeline(t, ts2, signAny(t, jwt2, "acct", auth.RoleOwner), "web", "", "") + if !live.Truncated { + t.Fatal("落盘源未接线 ⇒ truncated = true（本次不是全部），这条口径一字未改") + } + if live.BoundaryReached { + t.Fatal("落盘源未接线 ⇒ 保留期边界信号必须为 false（连库都没接，无从谈清扫）") + } +} + +// TestAdminFlowTimeline_RetentionDisabledNeverClaimsExpiry_边界 钉住第一条合成判据： +// **清扫没启用就不存在「保留期」这回事**。库里现存最早的是第 5 轮（本该能推出「更早的 +// 被删过」），但保留期关闭 ⇒ `retentionBoundaryReached` 必须为 false，且响应如实标注 +// `retentionEnabled:false` —— 那种情况下的「更早的轮次」不是被清扫掉的，不能说成过期。 +func TestAdminFlowTimeline_RetentionDisabledNeverClaimsExpiry_边界(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "cyrene.db") + _, store := openFlowGraphStore(t, path) // GCInterval/Retention 均为 0 ⇒ 清扫未启用 + base := time.Now().UTC().Add(-time.Hour) + recordTimelineGraph(t, store, "web", 5, "turn-5", base) + recordTimelineGraph(t, store, "web", 6, "turn-6", base.Add(time.Minute)) + _, ts, jwt := buildEnv(t, cfgDefault(), Deps{FlowGraphs: store}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + got := getTimeline(t, ts, tok, "web", "", "") + if got.RetentionEnabled || got.RetentionHours != 0 { + t.Fatalf("保留期 = %d h / enabled=%v, want 0 / false", got.RetentionHours, got.RetentionEnabled) + } + if got.BoundaryReached { + t.Fatal("保留期清扫未启用 ⇒ 不得断言「更早的轮次已过保留期」（它们不是被清扫掉的）") + } } // TestAdminFlowTimeline_UnknownSessionIsHonestEmpty_边界 钉住空态：发布会话范围外的会话