Agent-System/scripts/backfill_task_phase_snapshot_labels.go

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