chore(ops): 历史 task_phase_snapshot 冻结 label 一次性回填脚本
扫描 chat_messages.metadata.task_phase_snapshot,对 completed 步骤复用 taskrun.NormalizeSettledStep 归一化 label(幂等,可重复执行)。 支持 -dry-run 与 -id 定位;仅面向当前环境 DB(默认 ContextDSN)。 dev 实测:4424/4508 两条修复,二次 dry-run 为 0。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -0,0 +1,158 @@
|
||||
// 一次性幂等回填:把历史 chat_messages.metadata.task_phase_snapshot 中
|
||||
// 已收口为 completed 的步骤 label,从冻结的"正在…(NN%)"进度中措辞
|
||||
// 归一化为静态短标签(复用 taskrun.NormalizeSettledStep,避免逻辑漂移)。
|
||||
//
|
||||
// 缺陷来源见 .doc/ops/short-term/2026-09-21-taskrun-status-card.md 与消息 #4424。
|
||||
//
|
||||
// 用法(dev 环境,需 ContextDB 可达;在仓库根目录执行以加载 .env):
|
||||
//
|
||||
// go run ./scripts/backfill_task_phase_snapshot_labels.go -dry-run
|
||||
// go run ./scripts/backfill_task_phase_snapshot_labels.go # 实际写入
|
||||
// go run ./scripts/backfill_task_phase_snapshot_labels.go -id 4424 # 仅处理指定消息
|
||||
//
|
||||
// 环境变量 ENV / APP_DB_DSN 等覆盖同 config.Load。默认连接 config.ContextDSN()。
|
||||
//
|
||||
//go:build ignore
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"flag"
|
||||
"fmt"
|
||||
"os"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"chat-agent-system/internal/config"
|
||||
"chat-agent-system/internal/taskrun"
|
||||
)
|
||||
|
||||
func main() {
|
||||
dryRun := flag.Bool("dry-run", false, "仅打印将被变更的 message_id,不写库")
|
||||
onlyID := flag.Int64("id", 0, "仅处理指定 chat_messages.id(0 表示全表扫描)")
|
||||
flag.Parse()
|
||||
|
||||
cfg, err := config.Load("")
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "config load: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
pool, err := pgxpool.New(ctx, cfg.ContextDSN())
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "pool connect: %v (dsn=%s)\n", err, config.RedactPostgresDSN(cfg.ContextDSN()))
|
||||
os.Exit(1)
|
||||
}
|
||||
defer pool.Close()
|
||||
if err := pool.Ping(ctx); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "db ping: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
query := `SELECT id, metadata FROM chat_messages WHERE metadata ? 'task_phase_snapshot'`
|
||||
args := []any{}
|
||||
if *onlyID > 0 {
|
||||
query += ` AND id = $1`
|
||||
args = append(args, *onlyID)
|
||||
}
|
||||
query += ` ORDER BY id`
|
||||
|
||||
rows, err := pool.Query(ctx, query, args...)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "query: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
type row struct {
|
||||
id int64
|
||||
metadata []byte
|
||||
}
|
||||
var pending []row
|
||||
for rows.Next() {
|
||||
var (
|
||||
r row
|
||||
md []byte
|
||||
)
|
||||
if err := rows.Scan(&r.id, &md); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "scan: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
newMD, changed, err := rewriteSnapshot(md)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "message %d: %v\n", r.id, err)
|
||||
continue
|
||||
}
|
||||
if !changed {
|
||||
continue
|
||||
}
|
||||
r.metadata = newMD
|
||||
pending = append(pending, r)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "rows: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
if *dryRun {
|
||||
fmt.Printf("[dry-run] 将归一化 %d 条消息:\n", len(pending))
|
||||
for _, r := range pending {
|
||||
fmt.Printf(" id=%d\n", r.id)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
updated := 0
|
||||
for _, r := range pending {
|
||||
tag, err := pool.Exec(ctx,
|
||||
`UPDATE chat_messages SET metadata = $1 WHERE id = $2`,
|
||||
r.metadata, r.id)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "update id=%d: %v\n", r.id, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
updated += int(tag.RowsAffected())
|
||||
}
|
||||
fmt.Printf("backfill complete: scanned_candidates=%d updated=%d dry_run=false\n", len(pending), updated)
|
||||
}
|
||||
|
||||
// rewriteSnapshot 对 metadata 中的 task_phase_snapshot.steps 逐项归一化。
|
||||
// 返回(新 metadata JSON, 是否发生变化, error)。幂等:已归一化则 changed=false。
|
||||
func rewriteSnapshot(raw []byte) ([]byte, bool, error) {
|
||||
var meta map[string]any
|
||||
if err := json.Unmarshal(raw, &meta); err != nil {
|
||||
return nil, false, fmt.Errorf("unmarshal metadata: %w", err)
|
||||
}
|
||||
snapRaw, ok := meta["task_phase_snapshot"]
|
||||
if !ok {
|
||||
return raw, false, nil
|
||||
}
|
||||
buf, err := json.Marshal(snapRaw)
|
||||
if err != nil {
|
||||
return nil, false, fmt.Errorf("re-marshal snapshot: %w", err)
|
||||
}
|
||||
var snap taskrun.PhaseSnapshot
|
||||
if err := json.Unmarshal(buf, &snap); err != nil {
|
||||
return nil, false, fmt.Errorf("unmarshal snapshot: %w", err)
|
||||
}
|
||||
changed := false
|
||||
for i := range snap.Steps {
|
||||
before := snap.Steps[i].Label
|
||||
snap.Steps[i] = taskrun.NormalizeSettledStep(snap.Steps[i])
|
||||
if snap.Steps[i].Label != before {
|
||||
changed = true
|
||||
}
|
||||
}
|
||||
if !changed {
|
||||
return raw, false, nil
|
||||
}
|
||||
meta["task_phase_snapshot"] = snap
|
||||
out, err := json.Marshal(meta)
|
||||
if err != nil {
|
||||
return nil, false, fmt.Errorf("marshal metadata: %w", err)
|
||||
}
|
||||
return out, true, nil
|
||||
}
|
||||
Reference in New Issue
Block a user