// 一次性幂等回填:把历史 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 }