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

package main

import (
	"bytes"
	"database/sql"
	"encoding/json"
	"os"
	"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 TestMigrateRejectsSourceLedgerNameCollisionBeforeTargetOpen(t *testing.T) {
	source := buildFixture(t)
	db, err := sql.Open("sqlite", source)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`UPDATE schema_migrations SET name='work_task_bindings' WHERE version=35`); err != nil {
		db.Close()
		t.Fatal(err)
	}
	db.Close()
	err = cmdMigrate([]string{"--source", source, "--target", "invalid://target", "--migration-id", "name-collision"})
	if err == nil || !strings.Contains(err.Error(), "source migration identity mismatch") || !strings.Contains(err.Error(), "legacy feature migration numbering") {
		t.Fatalf("candidate branch ledger was not rejected before opening target: %v", err)
	}
}

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, project_id, actor_name) VALUES
		(7, '2026-10-11T01:00:00Z', 'cmd-1', 'http', 'chat', 'committed', 'op-1', 'c1', '{"actor":"alice","commandId":"cmd-1","operationId":"op-1","state":"committed","conversationId":"c1"}', 'alice', 'p1', '爱丽丝')`)
	mustExec(t, st, `INSERT INTO project_audit_links (id,host_audit_offset,project_id,owner,source_kind,source_id,operation_id,resource_kind,resource_id,run_id,disclosure_class,idempotency_key,created_at) VALUES
		(3,7,'p1','alice','domain-commit','commit-1','op-1','document','doc-1','','metadata','outbox-1','2026-10-11T01:00:01Z')`)
	// 设置（'' 默认租户 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["context_scope_scrub_state"] = true // SQLite 候选原文物理清理状态，不随迁
	// 运行态表不随迁：执行许可是进程瞬态；执行租约与跨实例命令属当前
	// 实例的调度现场（接管语义在 U5 内由 lease/fencing 处理，不复制历史）。
	for _, rt := range []string{"admission_leases", "conversation_executions", "run_commands"} {
		known[rt] = true
	}
	// 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)
	}
}

func TestBackupAndMigrateRejectCandidateRawContextScope(t *testing.T) {
	path := buildFixture(t)
	db, err := sql.Open("sqlite", path)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`ALTER TABLE context_manifests ADD COLUMN resolve_goal TEXT NOT NULL DEFAULT ''`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`ALTER TABLE context_manifests ADD COLUMN resolve_target_id TEXT NOT NULL DEFAULT ''`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`DELETE FROM schema_migrations WHERE version=41`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`INSERT INTO context_manifests(id,attempt_id,owner,target_id,project_id,input_sha256,delivery,manifest_json,created_at,scope_bound,resolve_goal,resolve_target_id) VALUES('candidate-ctx','candidate-attempt','alice','r1','p1','input-hash','attempted','{}','2026-10-14T00:00:00Z',1,'OLD_RESTRICTED_BODY','doc-private')`); err != nil {
		t.Fatal(err)
	}
	if err := db.Close(); err != nil {
		t.Fatal(err)
	}
	out := filepath.Join(t.TempDir(), "unsafe-backup.db")
	if err := cmdBackup([]string{"--source", path, "--out", out}); err == nil || !strings.Contains(err.Error(), "candidate context scope text") {
		t.Fatalf("backup accepted candidate v38 raw scope: %v", err)
	}
	if _, err := os.Stat(out); !os.IsNotExist(err) {
		t.Fatalf("rejected backup still created output: %v", err)
	}
	if err := cmdMigrate([]string{"--source", path, "--target", "postgres://invalid", "--migration-id", "unsafe"}); err == nil || !strings.Contains(err.Error(), "candidate context scope text") {
		t.Fatalf("migration accepted candidate v38 raw source: %v", err)
	}
	scrubbed, err := store.OpenSQLite(path)
	if err != nil {
		t.Fatal(err)
	}
	if err := scrubbed.Close(); err != nil {
		t.Fatal(err)
	}
	if err := cmdBackup([]string{"--source", path, "--out", out}); err != nil {
		t.Fatalf("scrubbed source rejected: %v", err)
	}
	data, err := os.ReadFile(out)
	if err != nil {
		t.Fatal(err)
	}
	if bytes.Contains(data, []byte("OLD_RESTRICTED_BODY")) {
		t.Fatal("backup retained candidate raw scope body")
	}
}

func TestBackupAndMigrateRejectPendingOrMissingContextScrubMarker(t *testing.T) {
	path := buildFixture(t)
	db, err := sql.Open("sqlite", path)
	if err != nil {
		t.Fatal(err)
	}
	for _, query := range []string{
		`CREATE TABLE IF NOT EXISTS context_scope_scrub_state(id INTEGER PRIMARY KEY,pending INTEGER NOT NULL)`,
		`INSERT OR REPLACE INTO context_scope_scrub_state(id,pending) VALUES(1,1)`,
	} {
		if _, err := db.Exec(query); err != nil {
			t.Fatal(err)
		}
	}
	if err := db.Close(); err != nil {
		t.Fatal(err)
	}
	assertRejected := func(name string) {
		t.Helper()
		out := filepath.Join(t.TempDir(), name+".db")
		if err := cmdBackup([]string{"--source", path, "--out", out}); err == nil {
			t.Fatalf("backup accepted %s context scrub state", name)
		}
		if _, err := os.Stat(out); !os.IsNotExist(err) {
			t.Fatalf("rejected %s backup created output: %v", name, err)
		}
		if err := cmdMigrate([]string{"--source", path, "--target", "postgres://invalid", "--migration-id", name}); err == nil {
			t.Fatalf("migration accepted %s context scrub state", name)
		}
	}
	assertRejected("pending")
	db, err = sql.Open("sqlite", path)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`DROP TABLE context_scope_scrub_state`); err != nil {
		t.Fatal(err)
	}
	if err := db.Close(); err != nil {
		t.Fatal(err)
	}
	assertRejected("missing")
	db, err = sql.Open("sqlite", path)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`DELETE FROM schema_migrations WHERE version=41`); err != nil {
		t.Fatal(err)
	}
	if err := db.Close(); err != nil {
		t.Fatal(err)
	}
	assertRejected("v38-missing")
}