feat(flow): 链路图时间轴出口（W1）—— 所有轮次按时间正序
- 新增 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 静默）
This commit is contained in:
AskaEth committed
1 parent
78e99d9f4c
commit
1efe3a6a79
5 files changed
+1401
No files matched your search
@@ -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
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user