fix(history): 历史运行复盘读库持久化 —— 杜绝过 Redis TTL 后卡「流式中…」

问题:历史任务超 Redis 流 10min TTL 后,SSE 回放在空流上永久阻塞 → 运行页卡「流式中…」、
轨迹/工具/输出全空。

修复:收尾把最终输出 + 执行轨迹持久化到 PG,历史复盘改读库(不依赖 Redis TTL):
- store:Task 加 output/trace 两列;SaveTaskOutput/SaveTaskTrace/GetRunDetail。
  trace 用 type:text(不是 jsonb)——否则提交时空串 "" 入 jsonb 列会 INSERT 失败、整条任务不落库。
  (已 ALTER 既有 trace 列 jsonb→text。)
- gateway:token/exec 录制器在 done 时把累计的输出/轨迹快照落库。
- 新增 GET /tasks/:id/replay 返回持久化的 {output, exec}。
- RunsView:选中历史运行改 runReplay() 读库(秒回、phase 立即 done/error),不再 SSE 回放。
  即便旧任务无持久化数据,也是 done+空态,绝不再卡「流式中…」。

live:新任务落库 output 305 字(含表格) + 轨迹 5 事件,/replay 正确返回;tsc+vite、gateway 全绿。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Blizzard
2026-06-26 17:52:26 +08:00
parent 594be6a317
commit d2662a1f37
6 changed files with 89 additions and 15 deletions
@@ -9,6 +9,7 @@ import (
"net/http"
"os"
"strconv"
"strings"
"time"
"github.com/gin-contrib/sse"
@@ -85,9 +86,19 @@ func (h *Handler) startTokenRecorder(taskID string) {
return
}
ctx, cancel := context.WithTimeout(context.Background(), 6*time.Minute) // 兜底防泄漏
var out strings.Builder // 收尾时把整段输出快照落库(供过 Redis TTL 的历史复盘)
unsub, err := h.bus.SubscribeTokens(taskID,
func(tok []byte) { _ = h.cache.StreamAppend(ctx, store.ChannelToken, taskID, "token", string(tok)) },
func() { _ = h.cache.StreamAppend(ctx, store.ChannelToken, taskID, "done", ""); cancel() },
func(tok []byte) {
out.Write(tok)
_ = h.cache.StreamAppend(ctx, store.ChannelToken, taskID, "token", string(tok))
},
func() {
_ = h.cache.StreamAppend(ctx, store.ChannelToken, taskID, "done", "")
if out.Len() > 0 {
_ = h.db.SaveTaskOutput(context.Background(), taskID, out.String())
}
cancel()
},
)
if err != nil {
cancel()
@@ -104,9 +115,23 @@ func (h *Handler) startExecRecorder(taskID string) {
return
}
ctx, cancel := context.WithTimeout(context.Background(), 12*time.Minute) // 含 HITL 审批等待,给足
var evs []json.RawMessage // 收尾时把轨迹快照落库(供过 Redis TTL 的历史复盘)
unsub, err := h.bus.SubscribeExec(taskID,
func(data []byte) { _ = h.cache.StreamAppend(ctx, store.ChannelExec, taskID, "exec", string(data)) },
func() { _ = h.cache.StreamAppend(ctx, store.ChannelExec, taskID, "done", ""); cancel() },
func(data []byte) {
cp := make([]byte, len(data))
copy(cp, data)
evs = append(evs, cp)
_ = h.cache.StreamAppend(ctx, store.ChannelExec, taskID, "exec", string(data))
},
func() {
_ = h.cache.StreamAppend(ctx, store.ChannelExec, taskID, "done", "")
if len(evs) > 0 {
if b, err := json.Marshal(evs); err == nil {
_ = h.db.SaveTaskTrace(context.Background(), taskID, string(b))
}
}
cancel()
},
)
if err != nil {
cancel()
@@ -421,6 +446,16 @@ func (h *Handler) StatsOverview(c *gin.Context) {
})
}
// TaskReplay: GET /api/v1/tasks/:id/replay —— 历史运行复盘(持久化的输出 + 轨迹,读库不依赖 Redis TTL)。
func (h *Handler) TaskReplay(c *gin.Context) {
output, trace := h.db.GetRunDetail(c.Request.Context(), c.Param("id"))
var exec []json.RawMessage
if trace != "" {
_ = json.Unmarshal([]byte(trace), &exec)
}
c.JSON(http.StatusOK, gin.H{"output": output, "exec": exec})
}
// Runs: GET /api/v1/runs?limit= —— 运行历史列表(任务 + 评测分级),供「运行 · 观测」复盘。
func (h *Handler) Runs(c *gin.Context) {
limit := 30
@@ -74,6 +74,7 @@ func New(db *store.Postgres, cache *store.Redis, bus *nats.Bus, blobStore *blob.
p.GET("/billing", h.Billing)
p.GET("/stats/overview", h.StatsOverview) // 工作台仪表盘聚合
p.GET("/runs", h.Runs) // 运行历史(复盘)
p.GET("/tasks/:id/replay", h.TaskReplay) // 历史运行复盘(持久化输出+轨迹,免 Redis TTL)
}
// 运维控制面:LLM 模型配置(含 API 密钥管理)—— 必须管理员(RequireAdmin)。
@@ -21,6 +21,9 @@ type Task struct {
Graph string `gorm:"type:jsonb"` // React Flow 导出的 DSL 原文
Status string `gorm:"size:32"` // submitted / running / done / failed / timeout
Detail string `gorm:"type:text"` // 失败/超时原因等(状态机回写)
// 收尾持久化:供「运行历史复盘」永久回放(Redis 流仅 10min TTL,过期后历史任务靠这两列)。
Output string `gorm:"type:text"` // 最终模型输出(收尾时由网关从流快照落库)
Trace string `gorm:"type:text"` // 执行轨迹事件 JSON 数组(存为文本,容忍空串;不在库内查它)
}
// Eval 是一次任务的自动化评测结果(dispatcher 评完经 NATS 回写,每任务一条,按 task_id upsert)。
+28
View File
@@ -145,6 +145,34 @@ func (p *Postgres) GetTaskStatus(ctx context.Context, id string) (status, detail
return t.Status, t.Detail
}
// SaveTaskOutput 收尾时落库最终模型输出(供历史复盘,best-effort)。
func (p *Postgres) SaveTaskOutput(ctx context.Context, id, output string) error {
if p.db == nil {
return nil
}
return p.db.WithContext(ctx).Model(&Task{}).Where("task_id = ?", id).Update("output", output).Error
}
// SaveTaskTrace 收尾时落库执行轨迹 JSON(供历史复盘,best-effort)。
func (p *Postgres) SaveTaskTrace(ctx context.Context, id, traceJSON string) error {
if p.db == nil {
return nil
}
return p.db.WithContext(ctx).Model(&Task{}).Where("task_id = ?", id).Update("trace", traceJSON).Error
}
// GetRunDetail 取一条任务持久化的输出 + 轨迹(历史复盘读库,不依赖 Redis TTL)。
func (p *Postgres) GetRunDetail(ctx context.Context, id string) (output, trace string) {
if p.db == nil {
return "", ""
}
var t Task
if err := p.db.WithContext(ctx).Select("output", "trace").Where("task_id = ?", id).First(&t).Error; err != nil {
return "", ""
}
return t.Output, t.Trace
}
// SaveEval 落库一条评测结果(按 task_id upsert:重评覆盖)。
func (p *Postgres) SaveEval(ctx context.Context, e *Eval) error {
if p.db == nil {