Agent-System-Backend/cmd/storage-migrate/main.go

// storage-migrate：SQLite → PostgreSQL 受管迁移入口（协作存储方案 U4/R7）。
//
// 纪律（方案 §5 U4 / §6）：
//   - 只处理 Backend 库；不执行 SQLite dump 作为 PG SQL（按行复制 + 类型映射）；
//   - 迁移身份（migration_id + source_snapshot_id + source_schema_version）落
//     目标库 migration_control：仅同一冻结快照可续传；目标已含其他数据即拒绝；
//   - 报告只含计数、校验和与归属检查结论——不含 DSN、密钥与行正文（KTD14）。
//
// 用法：
//
//	storage-migrate inventory  --source agent-system.db [--artifacts dir] [--out report.json]
//	storage-migrate backup     --source agent-system.db --out snapshot.db
//	storage-migrate migrate    --source snapshot.db --target "postgres://…" \
//	                             --migration-id <名> [--artifacts dir] [--out report.json]
package main

import (
	"crypto/sha256"
	"database/sql"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"time"

	_ "modernc.org/sqlite" // SQLite 驱动（源，只读打开）

	"agent-system-backend/internal/store"
)

func main() {
	if len(os.Args) < 2 {
		usage()
		os.Exit(2)
	}
	var err error
	switch os.Args[1] {
	case "inventory":
		err = cmdInventory(os.Args[2:])
	case "backup":
		err = cmdBackup(os.Args[2:])
	case "migrate":
		err = cmdMigrate(os.Args[2:])
	default:
		usage()
		os.Exit(2)
	}
	if err != nil {
		fmt.Fprintln(os.Stderr, "storage-migrate:", err)
		os.Exit(1)
	}
}

func usage() {
	fmt.Fprintln(os.Stderr, "用法：storage-migrate <inventory|backup|migrate> [flags]（-h 看各子命令 flags）")
}

// ---- 公共小件 ----

func openSourceRO(path string) (*sql.DB, error) {
	if _, err := os.Stat(path); err != nil {
		return nil, fmt.Errorf("source %s: %w", path, err)
	}
	// query_only：迁移源只读（冻结快照，任何写入都是缺陷）。
	db, err := sql.Open("sqlite", fmt.Sprintf("file:%s?_pragma=query_only(1)", path))
	if err != nil {
		return nil, err
	}
	if err := db.Ping(); err != nil {
		db.Close()
		return nil, fmt.Errorf("source %s: %w", path, err)
	}
	return db, nil
}

func fileSHA256(path string) (string, int64, error) {
	f, err := os.Open(path)
	if err != nil {
		return "", 0, err
	}
	defer f.Close()
	h := sha256.New()
	n, err := io.Copy(h, f)
	if err != nil {
		return "", 0, err
	}
	return hex.EncodeToString(h.Sum(nil)), n, nil
}

func nowISO() string { return time.Now().UTC().Format(time.RFC3339Nano) }

// ---- inventory：只读盘点（§6.1） ----

func cmdInventory(args []string) error {
	fs := flag.NewFlagSet("inventory", flag.ExitOnError)
	source := fs.String("source", "", "SQLite 库路径（活跃库或冻结快照）")
	artifacts := fs.String("artifacts", "", "工件对象目录（可选；盘点文件清单与哈希）")
	out := fs.String("out", "", "报告输出 JSON 路径（缺省打印到 stdout）")
	fs.Parse(args)
	if *source == "" {
		return fmt.Errorf("--source 必填")
	}
	db, err := openSourceRO(*source)
	if err != nil {
		return err
	}
	defer db.Close()
	rep, err := inventoryReport(db, *source, *artifacts)
	if err != nil {
		return err
	}
	return writeReport(rep, *out)
}

func inventoryReport(db *sql.DB, source, artifactsDir string) (map[string]any, error) {
	version, err := sourceSchemaVersion(db)
	if err != nil {
		return nil, err
	}
	var mode string
	if err := db.QueryRow(`PRAGMA journal_mode`).Scan(&mode); err != nil {
		return nil, fmt.Errorf("journal_mode: %w", err)
	}
	var chk string
	if err := db.QueryRow(`PRAGMA integrity_check`).Scan(&chk); err != nil || chk != "ok" {
		return nil, fmt.Errorf("integrity_check: %v (%q)", err, chk)
	}
	tables, err := sqliteTableCounts(db)
	if err != nil {
		return nil, err
	}
	snap, size, err := fileSHA256(source)
	if err != nil {
		return nil, err
	}
	rep := map[string]any{
		"command": "inventory", "source": source, "source_sha256": snap,
		"source_bytes": size, "source_schema_version": version,
		"journal_mode": mode, "integrity": chk, "table_counts": tables,
		"generated_at": nowISO(),
	}
	if artifactsDir != "" {
		manifest, err := artifactManifest(artifactsDir)
		if err != nil {
			return nil, err
		}
		rep["artifacts"] = manifest
	}
	return rep, nil
}

func sourceSchemaVersion(db *sql.DB) (int, error) {
	var v int
	err := db.QueryRow(`SELECT COALESCE(MAX(version),0) FROM schema_migrations`).Scan(&v)
	if err != nil {
		if strings.Contains(err.Error(), "no such table") {
			return 0, fmt.Errorf("源库缺 schema_migrations（不是本工具管理的库）：%w", err)
		}
		return 0, err
	}
	return v, nil
}

// sqliteTableCounts：仅盘点本工具认识的业务表（见 copyOrder）。
func sqliteTableCounts(db *sql.DB) (map[string]int64, error) {
	out := map[string]int64{}
	for _, t := range copyOrder {
		var n int64
		if err := db.QueryRow(`SELECT COUNT(*) FROM "` + t + `"`).Scan(&n); err != nil {
			if strings.Contains(err.Error(), "no such table") {
				continue // 旧库可能没有新表
			}
			return nil, fmt.Errorf("count %s: %w", t, err)
		}
		out[t] = n
	}
	return out, nil
}

func artifactManifest(dir string) (map[string]any, error) {
	type entry struct {
		Name, SHA string
		Size      int64
	}
	var entries []entry
	hAll := sha256.New()
	err := filepath.Walk(dir, func(p string, info os.FileInfo, err error) error {
		if err != nil || info.IsDir() {
			return err
		}
		sum, _, err := fileSHA256(p)
		if err != nil {
			return err
		}
		rel, _ := filepath.Rel(dir, p)
		entries = append(entries, entry{rel, sum, info.Size()})
		fmt.Fprintf(hAll, "%s %d %s\n", rel, info.Size(), sum)
		return nil
	})
	if err != nil {
		return nil, fmt.Errorf("artifacts %s: %w", dir, err)
	}
	sort.Slice(entries, func(i, j int) bool { return entries[i].Name < entries[j].Name })
	return map[string]any{
		"dir": dir, "files": entries, "file_count": len(entries),
		"manifest_sha256": hex.EncodeToString(hAll.Sum(nil)),
	}, nil
}

// ---- backup：一致性快照（§6.2：VACUUM INTO + 可打开校验 + 计数对账） ----

func cmdBackup(args []string) error {
	fs := flag.NewFlagSet("backup", flag.ExitOnError)
	source := fs.String("source", "", "活跃 SQLite 库路径")
	out := fs.String("out", "", "快照输出路径（不得与源同文件；已存在则拒绝）")
	fs.Parse(args)
	if *source == "" || *out == "" {
		return fmt.Errorf("--source 与 --out 必填")
	}
	if same, _ := filepath.EvalSymlinks(*source); same != "" {
		if abs1, _ := filepath.Abs(same); abs1 != "" {
			if abs2, _ := filepath.Abs(*out); abs2 == abs1 {
				return fmt.Errorf("快照输出不得与源同文件")
			}
		}
	}
	if _, err := os.Stat(*out); err == nil {
		return fmt.Errorf("快照输出已存在（不覆盖）：%s", *out)
	}
	if err := os.MkdirAll(filepath.Dir(*out), 0o755); err != nil {
		return err
	}
	// VACUUM INTO 在线安全（含 WAL 内容）；query_only 会禁写，需普通打开。
	db, err := sql.Open("sqlite", fmt.Sprintf("file:%s?_pragma=busy_timeout(30000)", *source))
	if err != nil {
		return err
	}
	defer db.Close()
	if err := refuseCandidateRawContextScope(db); err != nil {
		return err
	}
	if _, err := db.Exec(`VACUUM INTO ?`, *out); err != nil {
		return fmt.Errorf("vacuum into: %w", err)
	}
	// 校验：快照可打开、完整性、逐表计数与源一致。
	snap, err := openSourceRO(*out)
	if err != nil {
		return err
	}
	defer snap.Close()
	var chk string
	if err := snap.QueryRow(`PRAGMA integrity_check`).Scan(&chk); err != nil || chk != "ok" {
		return fmt.Errorf("快照完整性校验失败：%v (%q)", err, chk)
	}
	srcCounts, err := sqliteTableCounts(db)
	if err != nil {
		return err
	}
	snapCounts, err := sqliteTableCounts(snap)
	if err != nil {
		return err
	}
	for t, n := range srcCounts {
		if snapCounts[t] != n {
			return fmt.Errorf("快照计数不符：%s 源=%d 快照=%d", t, n, snapCounts[t])
		}
	}
	sum, size, err := fileSHA256(*out)
	if err != nil {
		return err
	}
	fmt.Printf("backup ok: %s（%d 字节，sha256 %s…，%d 表计数一致）\n", *out, size, sum[:12], len(srcCounts))
	return nil
}

// ---- migrate：受管迁移（§6.3 + U4） ----

// copyOrder：按外键依赖排序的业务表全集（projects 最先，引用方在后）。
// admission_leases 是运行态执行许可（迁移无意义），明确排除。
var copyOrder = []string{
	"projects", "users", "account_directory_revision", "profiles", "settings", "plugins",
	"conversations", "runs", "messages", "project_work_outbox", "events", "artifacts",
	"document_render_deliveries",
	"hitl_cards", "human_decisions", "plans", "todos", "knowledge", "knowledge_vectors",
	"activity_cards", "project_event_shares", "command_audit", "project_audit_links", "prompt_modules",
	"sessions", "api_keys",
	"plugin_grants", "plugin_schema", "plugin_kv", "background_tasks", "work_task_bindings",
	"memory_entries", "audit_events", "context_manifests", "workflow_step_sources", "pending_confirms", "llm_usage_daily",
	"secret_keys", "settings_snapshots", "regen_pending", "context_entries",
	"user_commands", "pending_notices", "recovery_holds",
	"surfaces", "surface_revisions", "surface_actions", "plugin_activation",
	// v26 login_keys（#171 合入 main 时漏登记；本 PR 一并补齐）
	"login_keys",
	// U1 I3：服务实例 DB 身份（无外键；迁移须保留以免目标库丢锚）
	"agent_system_identity",
	"account_events",
	"redeem_attempts",
	"recovery_proposals", "recovery_proposal_proofs",
	// v35 分层 Python 环境（无外键；申请与会话绑定要随库迁走）
	"exec_env_requests", "exec_env_bindings",
	// U8：活动服务锁的库内标记（无外键；迁移须保留以免目标库丢掉锁根）
	"instance_guard",
}

const controlDDL = `
CREATE TABLE IF NOT EXISTS migration_control (
	migration_id TEXT PRIMARY KEY,
	source_snapshot_id TEXT NOT NULL,
	source_schema_version INTEGER NOT NULL,
	started_at TEXT NOT NULL,
	finished_at TEXT,
	verified_at TEXT,
	report TEXT
);
CREATE TABLE IF NOT EXISTS migration_table_state (
	migration_id TEXT NOT NULL,
	table_name TEXT NOT NULL,
	source_count INTEGER NOT NULL,
	target_count INTEGER NOT NULL,
	checksum TEXT NOT NULL,
	state TEXT NOT NULL,
	finished_at TEXT,
	PRIMARY KEY (migration_id, table_name)
);`

func cmdMigrate(args []string) error {
	fs := flag.NewFlagSet("migrate", flag.ExitOnError)
	source := fs.String("source", "", "冻结快照路径（backup 子命令产物；不接受活跃库）")
	target := fs.String("target", "", "目标 PostgreSQL DSN（须为空库或同身份续传库）")
	migrationID := fs.String("migration-id", "", "迁移实例名（人工指定，报告与续传身份的一部分）")
	artifacts := fs.String("artifacts", "", "工件对象目录（可选；仅盘点清单入报告，不复制）")
	out := fs.String("out", "", "报告输出 JSON 路径（缺省打印到 stdout）")
	fs.Parse(args)
	if *source == "" || *target == "" || *migrationID == "" {
		return fmt.Errorf("--source、--target、--migration-id 必填")
	}
	src, err := openSourceRO(*source)
	if err != nil {
		return err
	}
	defer src.Close()
	if err := refuseCandidateRawContextScope(src); err != nil {
		return err
	}
	snapID, _, err := fileSHA256(*source)
	if err != nil {
		return err
	}
	srcVersion, err := sourceSchemaVersion(src)
	if err != nil {
		return err
	}
	if err := validateSourceMigrationLedger(src); err != nil {
		return err
	}

	// 目标：先用 store 同款迁移器建 schema/扩展（OpenPostgres 会跑迁移与
	// 未知版本门禁），再落控制表。DSN 不回显、不入报告。
	st, err := store.OpenPostgres(*target, filepath.Join(filepath.Dir(*source), "migrate-artifacts"))
	if err != nil {
		return fmt.Errorf("open target: %w", err)
	}
	defer st.Close()
	db := st.DB()
	if _, err := db.Exec(controlDDL); err != nil {
		return fmt.Errorf("control tables: %w", err)
	}
	if err := guardIdentity(db, *migrationID, snapID, srcVersion); err != nil {
		return err
	}
	if err := ensureControlRow(db, *migrationID, snapID, srcVersion); err != nil {
		return err
	}

	rep := map[string]any{
		"command": "migrate", "migration_id": *migrationID,
		"source_snapshot_id": snapID, "source_schema_version": srcVersion,
		"started_at": nowISO(),
	}
	var tables []map[string]any
	for _, t := range copyOrder {
		if !sqliteHasTable(src, t) {
			continue // 旧库无此表
		}
		entry, err := migrateTable(db, src, *migrationID, t)
		if err != nil {
			return fmt.Errorf("table %s: %w", t, err)
		}
		tables = append(tables, entry)
	}
	// 未知 owner 检查（U4：阻塞，不静默）。'' 与 'local' 是单租户默认身份。
	if err := checkOwners(db, src); err != nil {
		return err
	}
	// 校验：目标逐表重算计数与校验和，全部一致才落 verified。
	if err := verifyTables(db, *migrationID, tables); err != nil {
		return err
	}
	var totalSrc, totalDst int64
	for _, e := range tables {
		totalSrc += e["source_count"].(int64)
		totalDst += e["target_count"].(int64)
	}
	rep["tables"] = tables
	rep["rows_total"] = totalSrc
	rep["verified_at"] = nowISO()
	if *artifacts != "" {
		manifest, err := artifactManifest(*artifacts)
		if err != nil {
			return err
		}
		rep["artifacts"] = manifest
	}
	repJSON, _ := json.Marshal(rep)
	if _, err := db.Exec(`UPDATE migration_control SET finished_at=$1, verified_at=$2, report=$3 WHERE migration_id=$4`,
		nowISO(), rep["verified_at"], string(repJSON), *migrationID); err != nil {
		return err
	}
	return writeReport(rep, *out)
}

func validateSourceMigrationLedger(src *sql.DB) error {
	rows, err := src.Query(`SELECT version,name FROM schema_migrations`)
	if err != nil {
		return err
	}
	defer rows.Close()
	for rows.Next() {
		var version int
		var name string
		if err := rows.Scan(&version, &name); err != nil {
			return err
		}
		if err := store.ValidateMigrationIdentity(version, name); err != nil {
			return fmt.Errorf("source %w", err)
		}
	}
	return rows.Err()
}

// Candidate context-scope v35 may contain full model prompts in private columns.
// Never copy or back up such a source implicitly; it must be migrated and
// reviewed in an isolated Backend store before this tool accepts it.
func refuseCandidateRawContextScope(db *sql.DB) error {
	if !sqliteHasTable(db, "context_manifests") {
		return nil
	}
	var columns int
	if err := db.QueryRow(`SELECT COUNT(*) FROM pragma_table_info('context_manifests') WHERE name IN ('resolve_goal','resolve_target_id')`).Scan(&columns); err != nil {
		return err
	}
	if columns > 0 {
		return fmt.Errorf("candidate context scope text columns remain; isolate and scrub source before backup or migration")
	}
	var version int
	if sqliteHasTable(db, "schema_migrations") {
		if err := db.QueryRow(`SELECT COALESCE(MAX(version),0) FROM schema_migrations`).Scan(&version); err != nil {
			return err
		}
	}
	if version >= 35 {
		if !sqliteHasTable(db, "context_scope_scrub_state") {
			return fmt.Errorf("candidate context physical scrub marker missing; isolate and scrub source before backup or migration")
		}
		var pending int
		if err := db.QueryRow(`SELECT pending FROM context_scope_scrub_state WHERE id=1`).Scan(&pending); err != nil || pending != 0 {
			return fmt.Errorf("candidate context physical scrub pending or unverified; isolate and scrub source before backup or migration")
		}
	}
	return nil
}

// guardIdentity：目标非空且身份不同即拒绝（R7：不覆盖他人数据）。
//   - 存在其他 migration_id 的完成记录 → 拒绝；
//   - 同 migration_id 但 source_snapshot_id 不同（换了快照续传）→ 拒绝；
//   - 业务表已有数据而控制表为空（人工造过数据）→ 拒绝。
func guardIdentity(db *sql.DB, migrationID, snapID string, srcVersion int) error {
	var n int
	if err := db.QueryRow(`SELECT COUNT(*) FROM migration_control WHERE migration_id<>$1`, migrationID).Scan(&n); err != nil {
		return err
	}
	if n > 0 {
		return fmt.Errorf("目标库已有其他迁移记录（%d 条）：拒绝覆盖，换空库或专项回迁方案", n)
	}
	var snap, started string
	var sv int
	err := db.QueryRow(`SELECT source_snapshot_id, source_schema_version, started_at FROM migration_control WHERE migration_id=$1`, migrationID).Scan(&snap, &sv, &started)
	switch {
	case err == sql.ErrNoRows:
		// 新迁移：业务表必须为空。
		busy, err := businessRows(db)
		if err != nil {
			return err
		}
		if busy > 0 {
			return fmt.Errorf("目标库业务表已有 %d 行且无迁移身份：拒绝（非空目标不得作为迁移目标）", busy)
		}
	case err != nil:
		return err
	default:
		if snap != snapID {
			return fmt.Errorf("同 migration_id 但快照不同（库=%s… 本次=%s…）：仅同一冻结快照可续传", snap[:12], snapID[:12])
		}
		_ = sv
		_ = started
	}
	return nil
}

func businessRows(db *sql.DB) (int64, error) {
	var total int64
	for _, t := range copyOrder {
		// Schema setup seeds this singleton before a migration starts. It is
		// replaced from the source during copying and is not target business data.
		if t == "account_directory_revision" {
			continue
		}
		var n int64
		if err := db.QueryRow(`SELECT COUNT(*) FROM "` + t + `"`).Scan(&n); err != nil {
			if strings.Contains(err.Error(), "does not exist") {
				continue
			}
			return 0, err
		}
		total += n
	}
	return total, nil
}

func ensureControlRow(db *sql.DB, migrationID, snapID string, srcVersion int) error {
	_, err := db.Exec(`INSERT INTO migration_control (migration_id, source_snapshot_id, source_schema_version, started_at)
		VALUES ($1,$2,$3,$4)
		ON CONFLICT(migration_id) DO NOTHING`, migrationID, snapID, srcVersion, nowISO())
	return err
}

func sqliteHasTable(db *sql.DB, name string) bool {
	var n int
	if err := db.QueryRow(`SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name=?`, name).Scan(&n); err != nil {
		return false
	}
	return n > 0
}

// migrateTable：done 则对账（幂等重跑）；pending 则按行复制（布尔映射 +
// 批量 INSERT + 序列推进）后落 done 与源侧校验和。
func migrateTable(db *sql.DB, src *sql.DB, migrationID, table string) (map[string]any, error) {
	cols, pk, err := targetColumns(db, table)
	if err != nil {
		return nil, err
	}
	var state string
	var srcCount, dstCountStored int64
	var srcChecksum string
	err = db.QueryRow(`SELECT state, source_count, target_count, checksum FROM migration_table_state WHERE migration_id=$1 AND table_name=$2`,
		migrationID, table).Scan(&state, &srcCount, &dstCountStored, &srcChecksum)
	if err != nil && err != sql.ErrNoRows {
		return nil, err
	}
	if err == sql.ErrNoRows {
		state = "pending"
	}
	// 源侧计数与校验和始终重算（冻结快照上必然稳定；差异=快照被换，拒绝）。
	bools, err := boolColumns(db, table, cols)
	if err != nil {
		return nil, err
	}
	countNow, sumNow, err := sourceScan(src, table, cols, pk, bools)
	if err != nil {
		return nil, err
	}
	if state == "done" {
		if countNow != srcCount || sumNow != srcChecksum {
			return nil, fmt.Errorf("源与登记不符（疑似换快照）：count %d/%d checksum %s…/%s…", countNow, srcCount, sumNow[:12], srcChecksum[:12])
		}
		if table == "command_audit" {
			var guarded bool
			if err := db.QueryRow(`SELECT project_audit_cutoff_safe('command_audit'::regclass) AND project_audit_writers_guarded('command_audit'::regclass)`).Scan(&guarded); err != nil || !guarded {
				return nil, fmt.Errorf("command_audit idempotent guard is unsafe: safe=%v err=%v", guarded, err)
			}
		}
		entry := map[string]any{"table": table, "source_count": countNow, "target_count": dstCountStored, "checksum": sumNow, "state": "done(idempotent)"}
		return entry, nil
	}
	if table == "command_audit" {
		return migrateCommandAudit(db, src, migrationID, cols, pk, bools, countNow, sumNow)
	}
	if err := copyRows(db, src, table, cols); err != nil {
		return nil, err
	}
	dstCount, _, err := targetScan(db, table, cols, pk, bools)
	if err != nil {
		return nil, err
	}
	if dstCount != countNow {
		return nil, fmt.Errorf("复制后计数不符：源=%d 目标=%d", countNow, dstCount)
	}
	if err := advanceSequences(db, table); err != nil {
		return nil, err
	}
	if _, err := db.Exec(`INSERT INTO migration_table_state (migration_id, table_name, source_count, target_count, checksum, state, finished_at)
		VALUES ($1,$2,$3,$4,$5,$6,$7)
		ON CONFLICT(migration_id, table_name) DO UPDATE SET source_count=excluded.source_count, target_count=excluded.target_count, checksum=excluded.checksum, state=excluded.state, finished_at=excluded.finished_at`,
		migrationID, table, countNow, dstCount, sumNow, "done", nowISO()); err != nil {
		return nil, err
	}
	return map[string]any{"table": table, "source_count": countNow, "target_count": dstCount, "checksum": sumNow, "state": "done"}, nil
}

// targetColumns：目标侧列清单（按名排序）+ 主键列。布尔列集合同取自目标
// information_schema（类型映射以 PG schema 为准——SQLite 侧 0/1 → bool）。
func targetColumns(db *sql.DB, table string) (cols []string, pk []string, err error) {
	// 生成列（PG GENERATED ALWAYS，如 knowledge.tsv）与库内向量副本列
	// （vec_pg：BLOB 权威的派生表示，U6-3）不复制不校验：PG 从业务列
	// 派生或由 Upsert 双写，源库（SQLite）没有该列。
	rows, err := db.Query(`
		SELECT column_name, data_type FROM information_schema.columns
		WHERE table_schema=current_schema() AND table_name=$1 AND is_generated='NEVER'
		  AND column_name NOT IN ('vec_pg') ORDER BY column_name`, table)
	if err != nil {
		return nil, nil, err
	}
	defer rows.Close()
	exists := false
	for rows.Next() {
		var name, dtype string
		if err := rows.Scan(&name, &dtype); err != nil {
			return nil, nil, err
		}
		exists = true
		cols = append(cols, name)
	}
	if !exists {
		return nil, nil, fmt.Errorf("目标缺表 %s（迁移器未建？）", table)
	}
	prows, err := db.Query(`
		SELECT a.attname FROM pg_index i
		JOIN pg_attribute a ON a.attrelid=i.indrelid AND a.attnum=ANY(i.indkey)
		WHERE i.indisprimary AND i.indrelid=$1::regclass`, table)
	if err != nil {
		return nil, nil, err
	}
	defer prows.Close()
	for prows.Next() {
		var p string
		if err := prows.Scan(&p); err != nil {
			return nil, nil, err
		}
		pk = append(pk, p)
	}
	if len(pk) == 0 {
		pk = cols // 无主键表按全列序校验
	}
	return cols, pk, nil
}

func boolColumns(db *sql.DB, table string, cols []string) (map[string]bool, error) {
	out := map[string]bool{}
	rows, err := db.Query(`
		SELECT column_name FROM information_schema.columns
		WHERE table_schema=current_schema() AND table_name=$1 AND data_type='boolean'`, table)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	for rows.Next() {
		var c string
		if err := rows.Scan(&c); err != nil {
			return nil, err
		}
		out[c] = true
	}
	return out, nil
}

// canonicalRow：稳定行编码（校验和用）：列名排序、值规范化
// （bool→true/false、[]byte→base64、NULL→null、数值/字符串原样）。
func canonicalRow(cols []string, vals []any, bools map[string]bool) string {
	type kv struct {
		K string
		V any
	}
	pairs := make([]kv, len(cols))
	for i := range cols {
		pairs[i] = kv{cols[i], normalize(vals[i], bools[cols[i]])}
	}
	b, _ := json.Marshal(pairs)
	return string(b)
}

// normalize：布尔列按目标 schema 对称归一（源 int64 0/1、目标 bool 编码一致，
// 否则校验和必然不符——真库实测）；[]byte→base64。
func normalize(v any, isBool bool) any {
	if isBool {
		return sqliteBool(v)
	}
	switch t := v.(type) {
	case []byte:
		return base64.StdEncoding.EncodeToString(t)
	default:
		return v
	}
}

// hashChain：按主键序哈希链（前后衔接，顺序敏感）。
func hashChain(prev string, rowJSON string) string {
	h := sha256.Sum256([]byte(prev + "|" + rowJSON))
	return hex.EncodeToString(h[:])
}

// scanChecksum：源/目标共用的扫描对账（列清单一致、主键序、行哈希链）。
func scanChecksum(rows *sql.Rows, cols []string, bools map[string]bool) (int64, string, error) {
	vals := make([]any, len(cols))
	ptrs := make([]any, len(cols))
	for i := range vals {
		ptrs[i] = &vals[i]
	}
	var count int64
	sum := ""
	for rows.Next() {
		if err := rows.Scan(ptrs...); err != nil {
			return 0, "", err
		}
		sum = hashChain(sum, canonicalRow(cols, vals, bools))
		count++
	}
	return count, sum, rows.Err()
}

func sourceScan(src *sql.DB, table string, cols, pk []string, bools map[string]bool) (int64, string, error) {
	orderBy := "ORDER BY " + quoteJoin(pk, `"`)
	rows, err := src.Query(`SELECT ` + quoteJoin(cols, `"`) + ` FROM "` + table + `" ` + orderBy)
	if err != nil {
		return 0, "", err
	}
	defer rows.Close()
	return scanChecksum(rows, cols, bools)
}

type rowsQuerier interface {
	Query(string, ...any) (*sql.Rows, error)
}

func targetScan(db rowsQuerier, table string, cols, pk []string, bools map[string]bool) (int64, string, error) {
	rows, err := db.Query(`SELECT ` + quoteJoin(cols, `"`) + ` FROM "` + table + `" ORDER BY ` + quoteJoin(pk, `"`))
	if err != nil {
		return 0, "", err
	}
	defer rows.Close()
	return scanChecksum(rows, cols, bools)
}

func quoteJoin(names []string, q string) string {
	out := make([]string, len(names))
	for i, n := range names {
		out[i] = q + n + q
	}
	return strings.Join(out, ",")
}

// copyRows：按行复制。布尔列 0/1→false/true；其余原样（TEXT/INT/REAL/BLOB）。
func copyRows(db *sql.DB, src *sql.DB, table string, cols []string) error {
	bools, err := boolColumns(db, table, cols)
	if err != nil {
		return err
	}
	tx, err := db.Begin()
	if err != nil {
		return err
	}
	defer tx.Rollback()
	if err := copyRowsInTx(tx, src, table, cols, bools); err != nil {
		return err
	}
	return tx.Commit()
}

func copyRowsInTx(tx *sql.Tx, src *sql.DB, table string, cols []string, bools map[string]bool) error {
	colList := quoteJoin(cols, `"`)
	ph := make([]string, len(cols))
	for i := range ph {
		ph[i] = fmt.Sprintf("$%d", i+1)
	}
	placeholders := strings.Join(ph, ",")
	srcRows, err := src.Query(`SELECT ` + colList + ` FROM "` + table + `"`)
	if err != nil {
		return err
	}
	defer srcRows.Close()
	vals := make([]any, len(cols))
	ptrs := make([]any, len(cols))
	for i := range vals {
		ptrs[i] = &vals[i]
	}
	// The destination migrator creates a fresh singleton directory revision at
	// schema startup. Replace it with the source's opaque revision so existing
	// cursors/visible-change history remain tied to the migrated account data.
	if table == "account_directory_revision" {
		if _, err := tx.Exec(`DELETE FROM account_directory_revision`); err != nil {
			return err
		}
	}
	stmt, err := tx.Prepare(fmt.Sprintf(`INSERT INTO "%s" (%s) VALUES (%s) ON CONFLICT DO NOTHING`, table, colList, placeholders))
	if err != nil {
		return err
	}
	defer stmt.Close()
	var batch int
	for srcRows.Next() {
		if err := srcRows.Scan(ptrs...); err != nil {
			return err
		}
		args := make([]any, len(cols))
		for i, c := range cols {
			v := vals[i]
			if bools[c] {
				args[i] = sqliteBool(v)
			} else {
				args[i] = v
			}
		}
		if _, err := stmt.Exec(args...); err != nil {
			return fmt.Errorf("insert: %w", err)
		}
		batch++
	}
	if err := srcRows.Err(); err != nil {
		return err
	}
	return nil
}

// command_audit alone needs an offline transaction: explicit source IDs do not
// pass the online monotone guard, and disabling it must never become visible
// separately from the copied rows, sequence advance, or done marker.
func migrateCommandAudit(db *sql.DB, src *sql.DB, migrationID string, cols, pk []string, bools map[string]bool, sourceCount int64, sourceChecksum string) (map[string]any, error) {
	tx, err := db.Begin()
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()
	if _, err := tx.Exec(`LOCK TABLE command_audit IN ACCESS EXCLUSIVE MODE`); err != nil {
		return nil, err
	}
	if _, err := tx.Exec(`ALTER TABLE command_audit DISABLE TRIGGER project_audit_insert_guard_v1`); err != nil {
		return nil, err
	}
	if err := copyRowsInTx(tx, src, "command_audit", cols, bools); err != nil {
		return nil, err
	}
	targetCount, targetChecksum, err := targetScan(tx, "command_audit", cols, pk, bools)
	if err != nil {
		return nil, err
	}
	if targetCount != sourceCount || targetChecksum != sourceChecksum {
		return nil, fmt.Errorf("command_audit copy mismatch: count %d/%d checksum %s/%s", targetCount, sourceCount, targetChecksum, sourceChecksum)
	}
	if err := advanceCommandAuditSequence(tx); err != nil {
		return nil, err
	}
	if _, err := tx.Exec(`ALTER TABLE command_audit ENABLE ALWAYS TRIGGER project_audit_insert_guard_v1`); err != nil {
		return nil, err
	}
	var guarded bool
	if err := tx.QueryRow(`SELECT project_audit_cutoff_safe('command_audit'::regclass) AND project_audit_writers_guarded('command_audit'::regclass)`).Scan(&guarded); err != nil || !guarded {
		return nil, fmt.Errorf("command_audit guard not restored: safe=%v err=%v", guarded, err)
	}
	if _, err := tx.Exec(`INSERT INTO migration_table_state (migration_id, table_name, source_count, target_count, checksum, state, finished_at)
		VALUES ($1,'command_audit',$2,$3,$4,'done',$5)
		ON CONFLICT(migration_id, table_name) DO UPDATE SET source_count=excluded.source_count, target_count=excluded.target_count, checksum=excluded.checksum, state=excluded.state, finished_at=excluded.finished_at`,
		migrationID, sourceCount, targetCount, sourceChecksum, nowISO()); err != nil {
		return nil, err
	}
	if err := tx.Commit(); err != nil {
		return nil, err
	}
	return map[string]any{"table": "command_audit", "source_count": sourceCount, "target_count": targetCount, "checksum": sourceChecksum, "state": "done"}, nil
}

func advanceCommandAuditSequence(tx *sql.Tx) error {
	var seq string
	if err := tx.QueryRow(`SELECT pg_get_serial_sequence('command_audit','id')`).Scan(&seq); err != nil || seq == "" {
		return fmt.Errorf("command_audit sequence missing: %v", err)
	}
	var last int64
	var called bool
	if err := tx.QueryRow(fmt.Sprintf(`SELECT last_value,is_called FROM %s`, seq)).Scan(&last, &called); err != nil {
		return err
	}
	var maxID sql.NullInt64
	if err := tx.QueryRow(`SELECT MAX(id) FROM command_audit`).Scan(&maxID); err != nil {
		return err
	}
	if maxID.Valid && (maxID.Int64 > last || (maxID.Int64 == last && !called)) {
		if _, err := tx.Exec(`SELECT setval($1::regclass,$2,true)`, seq, maxID.Int64); err != nil {
			return err
		}
	}
	return nil
}

func sqliteBool(v any) bool {
	switch t := v.(type) {
	case int64:
		return t != 0
	case bool:
		return t
	default:
		return false
	}
}

// advanceSequences：SERIAL/identity 列复制后推进序列（显式插入不推进，
// 后续应用插入会撞主键）。无序列表为空操作。
func advanceSequences(db *sql.DB, table string) error {
	rows, err := db.Query(`
		SELECT column_name, pg_get_serial_sequence(format('%I.%I',table_schema,table_name), column_name)
		FROM information_schema.columns
		WHERE table_schema=current_schema() AND table_name=$1`, table)
	if err != nil {
		return err
	}
	defer rows.Close()
	var seqs []struct{ name, column string }
	for rows.Next() {
		var column string
		var seq sql.NullString
		if err := rows.Scan(&column, &seq); err != nil {
			return err
		}
		if seq.Valid {
			seqs = append(seqs, struct{ name, column string }{seq.String, column})
		}
	}
	if err := rows.Err(); err != nil {
		return err
	}
	for _, s := range seqs {
		// Sequence ownership remains authoritative when a column's default is a
		// guarded function rather than a literal nextval expression.
		column := `"` + strings.ReplaceAll(s.column, `"`, `""`) + `"`
		tableName := `"` + strings.ReplaceAll(table, `"`, `""`) + `"`
		if _, err := db.Exec(fmt.Sprintf(`SELECT setval($1::regclass, COALESCE((SELECT MAX(%s) FROM %s), 1))`, column, tableName), s.name); err != nil {
			return fmt.Errorf("setval %s (%s): %w", s.name, s.column, err)
		}
	}
	return nil
}

// checkOwners：业务行引用的 owner 必须存在于 users（单租户默认 ”/'local'
// 除外）。未知 owner 阻塞迁移（U4/R7：不静默带过未知归属）。
func checkOwners(db *sql.DB, src *sql.DB) error {
	ownerTables := []string{"projects", "conversations", "runs", "artifacts", "knowledge", "hitl_cards", "profiles", "activity_cards",
		"project_event_shares", "command_audit", "project_audit_links", "project_work_outbox", "work_task_bindings", "context_manifests", "workflow_step_sources",
		"sessions", "api_keys", "login_keys", "account_events", "recovery_proposals"}
	for _, t := range ownerTables {
		if !sqliteHasTable(src, t) {
			continue
		}
		rows, err := src.Query(`SELECT DISTINCT owner FROM "` + t + `" WHERE owner<>'' AND owner<>'local'`)
		if err != nil {
			return err
		}
		var unknown []string
		for rows.Next() {
			var o string
			if err := rows.Scan(&o); err != nil {
				rows.Close()
				return err
			}
			var n int
			if err := db.QueryRow(`SELECT COUNT(*) FROM users WHERE owner=$1`, o).Scan(&n); err != nil {
				rows.Close()
				return err
			}
			if n == 0 {
				unknown = append(unknown, o)
			}
		}
		rows.Close()
		if err := rows.Err(); err != nil {
			return err
		}
		if len(unknown) > 0 {
			return fmt.Errorf("表 %s 引用未知 owner（users 无此行）：%v——先在目标补建用户或修正归属，再迁移", t, unknown)
		}
	}
	return nil
}

// verifyTables：目标逐表重算，与登记的源计数/校验和全部一致才通过。
func verifyTables(db *sql.DB, migrationID string, tables []map[string]any) error {
	for _, e := range tables {
		table := e["table"].(string)
		wantCount := e["source_count"].(int64)
		wantSum := e["checksum"].(string)
		cols, pk, err := targetColumns(db, table)
		if err != nil {
			return err
		}
		bools, err := boolColumns(db, table, cols)
		if err != nil {
			return err
		}
		gotCount, gotSum, err := targetScan(db, table, cols, pk, bools)
		if err != nil {
			return err
		}
		if gotCount != wantCount || gotSum != wantSum {
			return fmt.Errorf("表 %s 校验不符：count %d/%d checksum %s…/%s…", table, gotCount, wantCount, gotSum[:12], wantSum[:12])
		}
	}
	return nil
}

func writeReport(rep map[string]any, out string) error {
	b, err := json.MarshalIndent(rep, "", "  ")
	if err != nil {
		return err
	}
	if out == "" {
		fmt.Println(string(b))
		return nil
	}
	if err := os.MkdirAll(filepath.Dir(out), 0o755); err != nil {
		return err
	}
	return os.WriteFile(out, append(b, '\n'), 0o644)
}