Merge remote-tracking branch 'origin/main' into feature/generative-ui-u2-coredep

cmd/storage-migrate/integration_test.go

@@ -0,0 +1,188 @@
+package main
+
+import (
+	"database/sql"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"agent-system-backend/internal/store"
+)
+
+// U4 集成（环境门控 AGENT_PG_TEST_DSN：CI test-pg job / 本机隔离实例）：
+// 夹具 → 受管迁移 → 校验和验证 → 幂等重跑 → 断点续传 → 非空目标拒绝 →
+// 未知 owner 阻塞。每用例独立 PG schema（与 store 测试同隔离约定）。
+
+func pgSchemaDSN(t *testing.T) (string, func()) {
+	t.Helper()
+	dsn := os.Getenv("AGENT_PG_TEST_DSN")
+	if dsn == "" {
+		t.Skip("AGENT_PG_TEST_DSN 未设置：真库集成由 CI 的必跑 job 覆盖")
+	}
+	schema := "mig_" + strings.ToLower(store.NewID("t")) + "_" + fmt.Sprint(os.Getpid())
+	admin, err := sql.Open("pgx", dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := admin.Exec(`CREATE EXTENSION IF NOT EXISTS pg_trgm`); err != nil {
+		admin.Close()
+		t.Fatalf("preinstall pg_trgm: %v", err)
+	}
+	if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil {
+		admin.Close()
+		t.Fatalf("create schema: %v", err)
+	}
+	cleanup := func() {
+		admin.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE")
+		admin.Close()
+	}
+	return store.AppendPGRuntimeParam(dsn, "search_path", schema+",public"), cleanup
+}
+
+func TestMigrateEndToEndAndIdempotent(t *testing.T) {
+	target, cleanup := pgSchemaDSN(t)
+	defer cleanup()
+	fixture := buildFixture(t)
+
+	if err := cmdMigrate([]string{"--source", fixture, "--target", target, "--migration-id", "mig-e2e"}); err != nil {
+		t.Fatalf("首次迁移：%v", err)
+	}
+	// 幂等重跑（同身份）：表级 done 对账，零重复行。
+	if err := cmdMigrate([]string{"--source", fixture, "--target", target, "--migration-id", "mig-e2e"}); err != nil {
+		t.Fatalf("幂等重跑：%v", err)
+	}
+	// 目标侧关键断言：布尔映射、NULL 保留、Unicode、BLOB、SERIAL 序列推进、
+	// 复合主键事件序。
+	st, err := store.OpenPostgres(target, filepath.Join(t.TempDir(), "artifacts"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer st.Close()
+	db := st.DB()
+	var pinned, archived bool
+	if err := db.QueryRow(`SELECT pinned, archived FROM conversations WHERE id='c1'`).Scan(&pinned, &archived); err != nil || !pinned || archived {
+		t.Fatalf("布尔映射：pinned=%v archived=%v err=%v", pinned, archived, err)
+	}
+	var title, note string
+	if err := db.QueryRow(`SELECT title, session_note FROM conversations WHERE id='c1'`).Scan(&title, &note); err != nil || !strings.Contains(note, "注") {
+		t.Fatalf("Unicode：title=%q note=%q err=%v", title, note, err)
+	}
+	var vec []byte
+	if err := db.QueryRow(`SELECT vec FROM knowledge_vectors WHERE doc_id='k1'`).Scan(&vec); err != nil || len(vec) != 12 {
+		t.Fatalf("BLOB 向量：len=%d err=%v", len(vec), err)
+	}
+	var ended sql.NullString
+	if err := db.QueryRow(`SELECT ended_at FROM runs WHERE id='r1'`).Scan(&ended); err != nil || ended.Valid {
+		t.Fatalf("NULL 保留：ended=%v err=%v", ended, err)
+	}
+	var seqMax int
+	if err := db.QueryRow(`SELECT MAX(seq) FROM events WHERE conversation_id='c1'`).Scan(&seqMax); err != nil || seqMax != 2 {
+		t.Fatalf("事件复合主键序：max=%d err=%v", seqMax, err)
+	}
+	// 复制的显式 id 之后，应用侧插入不撞序列（command_audit.id=7 已复制）。
+	if _, err := db.Exec(`INSERT INTO command_audit (time, command_id, source, entrance, state, operation_id, conversation_id, payload, owner) VALUES ('2026-10-11T03:00:00Z','cmd-2','http','chat','done','op-2','c1','{}','alice')`); err != nil {
+		t.Fatalf("序列推进失败（复制后插入撞主键）：%v", err)
+	}
+	// 报告不含秘密值。
+	var report string
+	if err := db.QueryRow(`SELECT report FROM migration_control WHERE migration_id='mig-e2e'`).Scan(&report); err != nil || strings.Contains(report, "sk-secret-value") {
+		t.Fatalf("报告泄密：err=%v", err)
+	}
+}
+
+func TestMigrateResumeAfterInterrupt(t *testing.T) {
+	target, cleanup := pgSchemaDSN(t)
+	defer cleanup()
+	fixture := buildFixture(t)
+
+	if err := cmdMigrate([]string{"--source", fixture, "--target", target, "--migration-id", "mig-resume"}); err != nil {
+		t.Fatal(err)
+	}
+	// 模拟断点：messages 已 done 后被部分回退（行删一半、状态回 pending），
+	// 重跑只重做该表，终态校验仍全绿。
+	db, err := sql.Open("pgx", target)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	if _, err := db.Exec(`DELETE FROM messages WHERE id='m3'`); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.Exec(`UPDATE migration_table_state SET state='pending' WHERE migration_id='mig-resume' AND table_name='messages'`); err != nil {
+		t.Fatal(err)
+	}
+	if err := cmdMigrate([]string{"--source", fixture, "--target", target, "--migration-id", "mig-resume"}); err != nil {
+		t.Fatalf("断点续跑：%v", err)
+	}
+	var n int
+	if err := db.QueryRow(`SELECT COUNT(*) FROM messages`).Scan(&n); err != nil || n != 3 {
+		t.Fatalf("续跑后 messages=%d：重做该表应恢复 3 行", n)
+	}
+}
+
+func TestMigrateRefusesForeignDataAndWrongSnapshot(t *testing.T) {
+	target, cleanup := pgSchemaDSN(t)
+	defer cleanup()
+	fixture := buildFixture(t)
+
+	// 非空目标（人工数据、无迁移身份）：拒绝。
+	st, err := store.OpenPostgres(target, filepath.Join(t.TempDir(), "artifacts"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	st.Close()
+	db, err := sql.Open("pgx", target)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	if _, err := db.Exec(`INSERT INTO projects (id, name, settings, created_at, updated_at, owner) VALUES ('foreign', '外来', '{}', '2026-10-11T00:00:00Z', '2026-10-11T00:00:00Z', 'alice')`); err != nil {
+		t.Fatal(err)
+	}
+	err = cmdMigrate([]string{"--source", fixture, "--target", target, "--migration-id", "mig-x"})
+	if err == nil || !strings.Contains(err.Error(), "非空目标") {
+		t.Fatalf("非空目标应拒绝：%v", err)
+	}
+	if _, err := db.Exec(`DELETE FROM projects WHERE id='foreign'`); err != nil {
+		t.Fatal(err)
+	}
+
+	// 同 migration_id 换快照：拒绝（仅同一冻结快照可续传）。
+	if err := cmdMigrate([]string{"--source", fixture, "--target", target, "--migration-id", "mig-snap"}); err != nil {
+		t.Fatal(err)
+	}
+	other := buildFixture(t) // 不同文件 ⇒ 不同 snapshot_id
+	err = cmdMigrate([]string{"--source", other, "--target", target, "--migration-id", "mig-snap"})
+	if err == nil || !strings.Contains(err.Error(), "快照不同") {
+		t.Fatalf("换快照应拒绝：%v", err)
+	}
+
+	// 其他 migration_id 已存在：拒绝。
+	err = cmdMigrate([]string{"--source", other, "--target", target, "--migration-id", "mig-other"})
+	if err == nil || !strings.Contains(err.Error(), "其他迁移记录") {
+		t.Fatalf("他身份应拒绝：%v", err)
+	}
+}
+
+func TestMigrateBlocksUnknownOwner(t *testing.T) {
+	target, cleanup := pgSchemaDSN(t)
+	defer cleanup()
+	fixture := buildFixture(t)
+	// 在夹具里造一个 users 之外的 owner 行（直接写源库）。
+	srcPath := fixture
+	st, err := store.OpenSQLite(srcPath) // 普通打开（夹具可写）
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := st.DB().Exec(`INSERT INTO knowledge (id, project_id, source, title, ref, chunk, created_at, owner) VALUES
+		('k_ghost', 'p1', 'upload', '幽灵', 'g.md', '…', '2026-10-11T01:00:00Z', 'ghost-user')`); err != nil {
+		t.Fatal(err)
+	}
+	st.Close()
+	err = cmdMigrate([]string{"--source", srcPath, "--target", target, "--migration-id", "mig-ghost"})
+	if err == nil || !strings.Contains(err.Error(), "未知 owner") {
+		t.Fatalf("未知 owner 应阻塞：%v", err)
+	}
+}

cmd/storage-migrate/main.go

@@ -0,0 +1,853 @@
+// 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 := 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", "profiles", "settings", "plugins",
+	"conversations", "runs", "messages", "events", "artifacts",
+	"hitl_cards", "plans", "todos", "knowledge", "knowledge_vectors",
+	"activity_cards", "command_audit", "prompt_modules",
+	"sessions", "api_keys",
+	"plugin_grants", "plugin_schema", "plugin_kv", "background_tasks",
+	"memory_entries", "audit_events", "pending_confirms", "llm_usage_daily",
+	"secret_keys", "settings_snapshots", "regen_pending", "context_entries",
+	"user_commands", "pending_notices", "recovery_holds",
+	"surfaces", "surface_revisions", "plugin_activation",
+}
+
+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()
+	snapID, _, err := fileSHA256(*source)
+	if err != nil {
+		return err
+	}
+	srcVersion, err := sourceSchemaVersion(src)
+	if 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)
+}
+
+// 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 {
+		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])
+		}
+		entry := map[string]any{"table": table, "source_count": countNow, "target_count": dstCountStored, "checksum": sumNow, "state": "done(idempotent)"}
+		return entry, nil
+	}
+	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）不复制不校验：PG 从
+	// 业务列自动派生，源库也没有该列。
+	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' 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)
+}
+
+func targetScan(db *sql.DB, 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
+	}
+	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]
+	}
+	tx, err := db.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+	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 tx.Commit()
+}
+
+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 pg_get_serial_sequence($1, column_name)
+		FROM information_schema.columns
+		WHERE table_schema=current_schema() AND table_name=$1
+		  AND (column_default LIKE 'nextval%' OR is_identity='YES')`, table)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+	var seqs []string
+	for rows.Next() {
+		var s any
+		if err := rows.Scan(&s); err != nil {
+			return err
+		}
+		if s != nil {
+			seqs = append(seqs, s.(string))
+		}
+	}
+	if err := rows.Err(); err != nil {
+		return err
+	}
+	for _, s := range seqs {
+		if _, err := db.Exec(fmt.Sprintf(`SELECT setval('%s', COALESCE((SELECT MAX(id) FROM "%s"), 1))`, s, table)); err != nil {
+			return fmt.Errorf("setval %s: %w", s, 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", "command_audit"}
+	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)
+}

cmd/storage-migrate/main_test.go

@@ -0,0 +1,193 @@
+package main
+
+import (
+	"encoding/json"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"agent-system-backend/internal/store"
+)
+
+// U4 迁移工具测试：夹具覆盖全部当前 schema 的边值（布尔/空值/时间/JSON/
+// Unicode/二进制向量/秘密字段），以及 inventory/backup 的行为面。
+// 真库端到端（幂等/续传/非空拒/未知 owner）在 integration_test.go（PG 门控）。
+
+// buildFixture：建一个全 schema 的 SQLite 库并填充边值行，返回路径。
+func buildFixture(t *testing.T) string {
+	t.Helper()
+	dir := t.TempDir()
+	path := filepath.Join(dir, "fixture.db")
+	st, err := store.OpenSQLite(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer st.Close()
+	seedEdgeRows(t, st)
+	return path
+}
+
+func seedEdgeRows(t *testing.T, st *store.Store) {
+	t.Helper()
+	// users（含 is_admin/must_change 布尔）与单租户默认身份。
+	mustExec(t, st, `INSERT INTO users (id, owner, name, created_at, password_hash, password_set_at, must_change_password, is_admin) VALUES
+		('u1', 'alice', '爱丽丝', '2026-10-11T01:00:00Z', 'sha256:…', '2026-10-11T01:00:00Z', false, true),
+		('u2', 'bob',   'Bob 😊',  '2026-10-11T01:00:00Z', '', '', true, false)`)
+	// 项目与会话（布尔 pinned/archived/allow_cross_read/cross_private、空 JSON）。
+	mustExec(t, st, `INSERT INTO projects (id, name, settings, created_at, updated_at, owner) VALUES
+		('p1', '项目一', '{}', '2026-10-11T01:00:00Z', '2026-10-11T01:00:00Z', 'alice'),
+		('p2', 'Emoji 项目 🚀', '{"a":null,"b":[1,2,"三"]}', '2026-10-11T02:00:00Z', '2026-10-11T02:00:00Z', '')`)
+	mustExec(t, st, `INSERT INTO conversations (id, project_id, title, profile_id, created_at, updated_at, owner, pinned, archived, allow_cross_read, cross_private, permission_mode, session_note) VALUES
+		('c1', 'p1', '会话', '', '2026-10-11T01:00:00Z', '2026-10-11T01:00:00Z', 'alice', true, false, false, true, '', 'note—注'),
+		('c2', 'p2', '', '', '2026-10-11T02:00:00Z', '2026-10-11T02:00:00Z', '', false, true, true, false, 'strict', '')`)
+	// 消息（request_id 幂等、parts JSON 含空对象/空数组/Unicode）与 seq 顺序。
+	mustExec(t, st, `INSERT INTO messages (id, conversation_id, run_id, role, parts, seq, created_at, request_id) VALUES
+		('m1', 'c1', '', 'user', '[{"type":"text","text":"你好"}]', 1, '2026-10-11T01:00:01Z', 'req-1'),
+		('m2', 'c1', '', 'assistant', '[{"type":"kv","kv":[]},{"type":"text","text":""}]', 2, '2026-10-11T01:00:02Z', ''),
+		('m3', 'c2', '', 'system', '[]', 1, '2026-10-11T02:00:01Z', 'req-2')`)
+	// runs（NULL ended_at/usage/error）。
+	mustExec(t, st, `INSERT INTO runs (id, parent_run_id, conversation_id, project_id, owner, kind, profile_id, status, phase, agent_name, created_at, instance_id) VALUES
+		('r1', '', 'c1', 'p1', 'alice', 'conversation', '', 'completed', '', '', '2026-10-11T01:00:00Z', 'host:1'),
+		('r2', 'r1', 'c1', 'p1', 'alice', 'subagent', '', 'running', 'thinking', '检索', '2026-10-11T01:00:03Z', 'host:2')`)
+	// events（复合主键 (conversation_id, seq)）。
+	mustExec(t, st, `INSERT INTO events (conversation_id, seq, run_id, ts, v, type, payload) VALUES
+		('c1', 1, 'r1', '2026-10-11T01:00:01Z', 1, 'run.phase', '{"phase":"thinking"}'),
+		('c1', 2, 'r1', '2026-10-11T01:00:02Z', 1, 'message.appended', NULL),
+		('c2', 1, '', '2026-10-11T02:00:01Z', 1, 'x', '{}')`)
+	// 工件（size 0 与大值）。
+	mustExec(t, st, `INSERT INTO artifacts (id, project_id, name, kind, mime, size, source_run, content_path, created_at, updated_at, owner) VALUES
+		('a1', 'p1', '报告.md', 'doc', 'text/markdown', 0, 'r1', 'art/a1', '2026-10-11T01:00:00Z', '2026-10-11T01:00:00Z', 'alice'),
+		('a2', 'p2', '数据.xlsx', 'doc', '', 2147483647, '', '', '2026-10-11T02:00:00Z', '2026-10-11T02:00:00Z', '')`)
+	// HITL 卡（resolved=0、resolution NULL）。
+	mustExec(t, st, `INSERT INTO hitl_cards (id, run_id, conversation_id, payload, created_at, resolved, resolution, owner) VALUES
+		('h1', 'r2', 'c1', '{"q":"继续？"}', '2026-10-11T01:00:04Z', 0, NULL, 'alice')`)
+	// 知识 + 向量（BLOB 二进制：float32 little-endian）。
+	mustExec(t, st, `INSERT INTO knowledge (id, project_id, source, title, ref, chunk, created_at, owner) VALUES
+		('k1', 'p1', 'upload', '标题', 'doc.md', '正文内容…', '2026-10-11T01:00:00Z', 'alice')`)
+	mustExec(t, st, `INSERT INTO knowledge_vectors (doc_id, owner, ref, dim, vec, created_at) VALUES
+		('k1', 'alice', 'doc.md', 3, x'0000803F0000003F000080BF', '2026-10-11T01:00:00Z')`)
+	// 活动卡（pinned 布尔）。
+	mustExec(t, st, `INSERT INTO activity_cards (id, project_id, conversation_id, ref, owner, payload, status, pinned, created_at, updated_at, version) VALUES
+		('ac1', 'p1', 'c1', '', 'alice', '{"title":"卡"}', 'active', 1, '2026-10-11T01:00:00Z', '2026-10-11T01:00:00Z', 2)`)
+	// 命令审计（SERIAL 主键：显式 id 插入）。
+	mustExec(t, st, `INSERT INTO command_audit (id, time, command_id, source, entrance, state, operation_id, conversation_id, payload, owner) VALUES
+		(7, '2026-10-11T01:00:00Z', 'cmd-1', 'http', 'chat', 'done', 'op-1', 'c1', '{"actor":"alice"}', 'alice')`)
+	// 设置（'' 默认租户 scope）。
+	mustExec(t, st, `INSERT INTO settings (owner, scope, "values") VALUES ('', 'runtime', '{"rateLimitPerMin":0}')`)
+	// 秘密字段（值不外泄：迁移按行复制，报告无正文）。
+	mustExec(t, st, `INSERT INTO secret_keys (id, owner, service, label, value, active, created_at) VALUES
+		('sk1', 'alice', 'openai', 'main', 'sk-secret-value-不打印', 1, '2026-10-11T01:00:00Z')`)
+	// 恢复隔离（业务态，应随迁）。
+	mustExec(t, st, `INSERT INTO recovery_holds (run_id, owner, conversation_id, reason, created_at) VALUES
+		('rh1', 'alice', 'c1', '副作用未知', '2026-10-11T01:00:00Z')`)
+}
+
+func must(t *testing.T, err error) {
+	t.Helper()
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+// mustExec：夹具直接走底层连接写行（业务代码不得如此；见 store.DB 注释）。
+func mustExec(t *testing.T, st *store.Store, q string) {
+	t.Helper()
+	if _, err := st.DB().Exec(q); err != nil {
+		t.Fatal(err)
+	}
+}
+
+// ---- inventory ----
+
+func TestInventoryReport(t *testing.T) {
+	path := buildFixture(t)
+	db, err := openSourceRO(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	rep, err := inventoryReport(db, path, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if rep["source_schema_version"].(int) < 20 {
+		t.Fatalf("schema version: %v", rep["source_schema_version"])
+	}
+	counts := rep["table_counts"].(map[string]int64)
+	if counts["messages"] != 3 || counts["events"] != 3 || counts["users"] != 2 {
+		t.Fatalf("counts: %v", counts)
+	}
+	b, _ := json.Marshal(rep)
+	if strings.Contains(string(b), "sk-secret-value") {
+		t.Fatal("inventory 报告不得包含秘密值")
+	}
+}
+
+func TestInventoryMissingFile(t *testing.T) {
+	if _, err := openSourceRO(filepath.Join(t.TempDir(), "nope.db")); err == nil {
+		t.Fatal("缺文件应报错")
+	}
+}
+
+// ---- backup ----
+
+func TestBackupVerifiesAndSnapshotsConsistent(t *testing.T) {
+	path := buildFixture(t)
+	out := filepath.Join(t.TempDir(), "snap.db")
+	if err := cmdBackup([]string{"--source", path, "--out", out}); err != nil {
+		t.Fatal(err)
+	}
+	snap, err := openSourceRO(out)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer snap.Close()
+	var n int
+	if err := snap.QueryRow(`SELECT COUNT(*) FROM messages`).Scan(&n); err != nil || n != 3 {
+		t.Fatalf("snap messages: %d %v", n, err)
+	}
+	// 已存在的输出拒绝覆盖。
+	if err := cmdBackup([]string{"--source", path, "--out", out}); err == nil {
+		t.Fatal("已存在快照应拒绝")
+	}
+}
+
+// ---- 夹具即文档：确认全部业务表在 copyOrder 内（防新表漏迁） ----
+
+func TestCopyOrderCoversAllFixtureTables(t *testing.T) {
+	path := buildFixture(t)
+	db, err := openSourceRO(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	rows, err := db.Query(`SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer rows.Close()
+	known := map[string]bool{}
+	for _, t := range copyOrder {
+		known[t] = true
+	}
+	known["schema_migrations"] = true // 迁移器管理，不随迁
+	known["admission_leases"] = true  // 运行态，明确排除（见 copyOrder 注释）
+	// FTS5 虚表影子表：SQLite 内部结构，PG 侧由 v4 生成列/索引从 knowledge
+	// 行自动重建，不随迁。
+	for _, shadow := range []string{"knowledge_fts", "knowledge_fts_data", "knowledge_fts_idx", "knowledge_fts_docsize", "knowledge_fts_config"} {
+		known[shadow] = true
+	}
+	var missing []string
+	for rows.Next() {
+		var name string
+		if err := rows.Scan(&name); err != nil {
+			t.Fatal(err)
+		}
+		if !known[name] {
+			missing = append(missing, name)
+		}
+	}
+	if len(missing) > 0 {
+		t.Fatalf("copyOrder 漏表（新表加入 schema 后须同步工具清单）：%v", missing)
+	}
+}

internal/store/store.go

@@ -209,6 +209,10 @@ func (s *Store) Dialect() string { return s.dialect.name() }
 // Pool 返回生效的连接预算（启动诊断；脱敏，不含 DSN）。
 func (s *Store) Pool() PoolOptions { return s.pool }
 
+// DB 暴露底层连接（运维工具专用：storage-migrate 的控制表与
+// information_schema 查询）。业务代码一律走本包公开方法。
+func (s *Store) DB() *sql.DB { return s.db }
+
 // DataDir 返回工件对象目录。
 func (s *Store) DataDir() string { return s.dataDir }