AskaEth
2bf3b536dd
① 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 守卫）
600 lines
27 KiB
Go
600 lines
27 KiB
Go
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"`
|
||
BoundaryReached bool `json:"retentionBoundaryReached"`
|
||
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)
|
||
}
|
||
// 🔴 清扫信号（切片 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_边界 钉住空态：发布会话范围外的会话
|
||
// （隔离渠道会话）从不建图 ⇒ `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
|
||
}
|