diff --git a/scripts/backfill_task_phase_snapshot_labels.go b/scripts/backfill_task_phase_snapshot_labels.go new file mode 100644 index 00000000..3d71599a --- /dev/null +++ b/scripts/backfill_task_phase_snapshot_labels.go @@ -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 +}