diff --git a/cmd/storage-migrate/main.go b/cmd/storage-migrate/main.go index 93d9463..3a60187 100644 --- a/cmd/storage-migrate/main.go +++ b/cmd/storage-migrate/main.go @@ -430,6 +430,21 @@ func refuseCandidateRawContextScope(db *sql.DB) error { 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 >= 34 { + 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 } diff --git a/cmd/storage-migrate/main_test.go b/cmd/storage-migrate/main_test.go index 24b9170..de09be6 100644 --- a/cmd/storage-migrate/main_test.go +++ b/cmd/storage-migrate/main_test.go @@ -173,7 +173,8 @@ func TestCopyOrderCoversAllFixtureTables(t *testing.T) { for _, t := range copyOrder { known[t] = true } - known["schema_migrations"] = 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"} { @@ -248,3 +249,58 @@ func TestBackupAndMigrateRejectCandidateRawContextScope(t *testing.T) { 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=35`); err != nil { + t.Fatal(err) + } + if err := db.Close(); err != nil { + t.Fatal(err) + } + assertRejected("v34-missing") +} diff --git a/internal/store/context_manifest_test.go b/internal/store/context_manifest_test.go index 2259897..2200a42 100644 --- a/internal/store/context_manifest_test.go +++ b/internal/store/context_manifest_test.go @@ -128,6 +128,95 @@ func TestCandidateV34RawScopeColumnsAreScrubbedOnUpgrade(t *testing.T) { } } +func TestCandidateV34CommittedDropKeepsPendingScrubAcrossRestart(t *testing.T) { + path := filepath.Join(t.TempDir(), "candidate-crash.db") + st, err := OpenSQLite(path) + if err != nil { + t.Fatal(err) + } + m := extension.ProjectContextManifest{ProjectID: "p1", AttemptID: "crash-model", InputSHA256: "input", Delivery: "attempted"} + if err := st.AppendContextManifestWithScope("alice", "run-1", m, ContextManifestScope{Bound: true}); err != nil { + t.Fatal(err) + } + for _, query := range []string{ + `DELETE FROM schema_migrations WHERE version=35`, + `ALTER TABLE context_manifests ADD COLUMN resolve_goal TEXT NOT NULL DEFAULT ''`, + `ALTER TABLE context_manifests ADD COLUMN resolve_target_id TEXT NOT NULL DEFAULT ''`, + `UPDATE context_manifests SET resolve_goal='OLD_RESTRICTED_BODY',resolve_target_id='doc-private'`, + } { + if _, err := st.db.Exec(query); err != nil { + t.Fatal(err) + } + } + // Apply only the transactional DDL and ledger. A process crash at this + // boundary must not make the next open mistake the file for scrubbed. + if err := st.migratePass(); err != nil { + t.Fatal(err) + } + var pending int + if err := st.db.QueryRow(`SELECT pending FROM context_scope_scrub_state WHERE id=1`).Scan(&pending); err != nil || pending != 1 { + t.Fatalf("committed v35 lost physical scrub obligation: pending=%d err=%v", pending, err) + } + if err := st.Close(); err != nil { + t.Fatal(err) + } + restarted, err := OpenSQLite(path) + if err != nil { + t.Fatal(err) + } + defer restarted.Close() + if err := restarted.db.QueryRow(`SELECT pending FROM context_scope_scrub_state WHERE id=1`).Scan(&pending); err != nil || pending != 0 { + t.Fatalf("restart did not complete pending scrub: pending=%d err=%v", pending, err) + } + for _, suffix := range []string{"", "-wal"} { + data, err := os.ReadFile(path + suffix) + if err == nil && bytes.Contains(data, []byte("OLD_RESTRICTED_BODY")) { + t.Fatalf("restart left candidate raw bytes in SQLite %s", suffix) + } + } +} + +func TestEarlierV35WithoutMarkerIsConservativelyRescrubbed(t *testing.T) { + path := filepath.Join(t.TempDir(), "old-v35.db") + st, err := OpenSQLite(path) + if err != nil { + t.Fatal(err) + } + m := extension.ProjectContextManifest{ProjectID: "p1", AttemptID: "old-v35-model", InputSHA256: "input", Delivery: "attempted"} + if err := st.AppendContextManifestWithScope("alice", "run-1", m, ContextManifestScope{Bound: true}); err != nil { + t.Fatal(err) + } + for _, query := range []string{ + `ALTER TABLE context_manifests ADD COLUMN resolve_goal TEXT NOT NULL DEFAULT ''`, + `UPDATE context_manifests SET resolve_goal='OLD_RESTRICTED_BODY'`, + `UPDATE context_manifests SET resolve_goal=''`, + `ALTER TABLE context_manifests DROP COLUMN resolve_goal`, + `DROP TABLE context_scope_scrub_state`, + } { + if _, err := st.db.Exec(query); err != nil { + t.Fatal(err) + } + } + if err := st.Close(); err != nil { + t.Fatal(err) + } + restarted, err := OpenSQLite(path) + if err != nil { + t.Fatal(err) + } + defer restarted.Close() + var pending int + if err := restarted.db.QueryRow(`SELECT pending FROM context_scope_scrub_state WHERE id=1`).Scan(&pending); err != nil || pending != 0 { + t.Fatalf("earlier v35 without marker was not re-scrubbed: pending=%d err=%v", pending, err) + } + for _, suffix := range []string{"", "-wal"} { + data, err := os.ReadFile(path + suffix) + if err == nil && bytes.Contains(data, []byte("OLD_RESTRICTED_BODY")) { + t.Fatalf("old v35 raw bytes survived re-scrub in SQLite %s", suffix) + } + } +} + func TestCandidateV34PostgresRawScopeScrubAndFreshSchema(t *testing.T) { st := openTestStore(t) if st.Dialect() != "postgres" { diff --git a/internal/store/migrate.go b/internal/store/migrate.go index b9bb9af..5ca9f0d 100644 --- a/internal/store/migrate.go +++ b/internal/store/migrate.go @@ -881,8 +881,11 @@ func init() { }, }, migration{version: 35, name: "scrub_candidate_context_scope_text", - up: map[string][]string{"sqlite": {`SELECT 1`}, "postgres": {`SELECT 1`}}, - down: map[string][]string{"sqlite": {`SELECT 1`}, "postgres": {`SELECT 1`}}, + up: map[string][]string{ + "sqlite": {`CREATE TABLE IF NOT EXISTS context_scope_scrub_state(id INTEGER PRIMARY KEY CHECK(id=1),pending INTEGER NOT NULL CHECK(pending IN (0,1)))`}, + "postgres": {`SELECT 1`}, + }, + down: map[string][]string{"sqlite": {`DROP TABLE IF EXISTS context_scope_scrub_state`}, "postgres": {`SELECT 1`}}, }, ) } @@ -976,17 +979,17 @@ func (s *Store) migrate() error { if s.dialect.name() == "postgres" { return s.migratePG() } - var candidateRawColumns int - if err := s.db.QueryRow(`SELECT COUNT(*) FROM pragma_table_info('context_manifests') WHERE name IN ('resolve_goal','resolve_target_id')`).Scan(&candidateRawColumns); err != nil { - return err - } // SQLite：_txlock=immediate（OpenSQLite DSN）保证迁移事务在 BEGIN 即取写锁， // 跨进程迁移者按 busy_timeout 排队；后续版本读取发生在同一事务内， // 前一个迁移者提交后本进程看到的已是完整账本（无部分版本可见）。 if err := s.migratePass(); err != nil { return err } - if candidateRawColumns > 0 { + pending, err := s.ensureContextScrubState() + if err != nil { + return err + } + if pending { // The prior candidate schema may have stored full prompts. VACUUM is // outside the DDL transaction; it compacts the isolated SQLite file // after v35 drops those columns. Refuse to serve if compaction fails. @@ -1003,10 +1006,38 @@ func (s *Store) migrate() error { if busy != 0 { return errors.New("candidate context scrub checkpoint remains busy") } + if _, err := s.db.Exec(`UPDATE context_scope_scrub_state SET pending=0 WHERE id=1 AND pending=1`); err != nil { + return fmt.Errorf("candidate context scrub completion: %w", err) + } } return nil } +// ensureContextScrubState handles databases that ran the earlier candidate +// v35 before a durable marker existed. They are conservatively re-scrubbed. +func (s *Store) ensureContextScrubState() (bool, error) { + var rawColumns int + if err := s.db.QueryRow(`SELECT COUNT(*) FROM pragma_table_info('context_manifests') WHERE name IN ('resolve_goal','resolve_target_id')`).Scan(&rawColumns); err != nil { + return false, err + } + if rawColumns != 0 { + return false, errors.New("candidate context raw scope columns remain after migration") + } + if _, err := s.db.Exec(`CREATE TABLE IF NOT EXISTS context_scope_scrub_state(id INTEGER PRIMARY KEY CHECK(id=1),pending INTEGER NOT NULL CHECK(pending IN (0,1)))`); err != nil { + return false, err + } + // An absent row may be a partially initialized earlier candidate; mark it + // pending before any physical cleanup. A fresh v35 inserted pending=0. + if _, err := s.db.Exec(`INSERT OR IGNORE INTO context_scope_scrub_state(id,pending) VALUES(1,1)`); err != nil { + return false, err + } + var pending int + if err := s.db.QueryRow(`SELECT pending FROM context_scope_scrub_state WHERE id=1`).Scan(&pending); err != nil { + return false, err + } + return pending != 0, nil +} + // migratePG 以会话级咨询锁串行化并发迁移者（KTD8）：锁在专用连接上获取， // 等锁者在轮到自己后重新读取账本——前一个迁移者已完成时本轮为空操作。 // 锁与迁移事务分离：迁移语句走连接池，锁只负责让 migrator 全局互斥。 @@ -1082,8 +1113,10 @@ func (s *Store) migratePass() error { if applied[m.version] { continue } + candidateRawScope := false if m.version == 35 { - if err := scrubCandidateContextScope(tx, s.dialect.name()); err != nil { + candidateRawScope, err = scrubCandidateContextScope(tx, s.dialect.name()) + if err != nil { return fmt.Errorf("migrate v35: %w", err) } } @@ -1096,6 +1129,15 @@ func (s *Store) migratePass() error { return fmt.Errorf("migrate v%d (%s): %w", m.version, m.name, err) } } + if m.version == 35 && s.dialect.name() == "sqlite" { + pending := 0 + if candidateRawScope { + pending = 1 + } + if _, err := tx.Exec(`INSERT OR REPLACE INTO context_scope_scrub_state(id,pending) VALUES(1,?)`, pending); err != nil { + return fmt.Errorf("migrate v35 scrub marker: %w", err) + } + } // 故障注入点：DDL 已执行、版本未登记——历史实现在此处崩溃会留下 // 「DDL 生效而账本缺失」的中间态；同事务设计下这里失败必须整体回滚。 if fault := s.takeMigrateFault(); fault != nil { @@ -1111,7 +1153,8 @@ func (s *Store) migratePass() error { // Candidate v34 was never released, but isolated test databases may already // contain raw Resolve goals. Fresh v34 creates no such columns; v35 removes // them conditionally from the earlier candidate schema before serving. -func scrubCandidateContextScope(tx *sql.Tx, dialect string) error { +func scrubCandidateContextScope(tx *sql.Tx, dialect string) (bool, error) { + found := false for _, column := range []string{"resolve_goal", "resolve_target_id"} { var exists int var err error @@ -1121,19 +1164,20 @@ func scrubCandidateContextScope(tx *sql.Tx, dialect string) error { err = tx.QueryRow(`SELECT COUNT(*) FROM information_schema.columns WHERE table_schema=current_schema() AND table_name='context_manifests' AND column_name=$1`, column).Scan(&exists) } if err != nil { - return err + return false, err } if exists == 0 { continue } + found = true if _, err := tx.Exec(`UPDATE context_manifests SET ` + column + `=''`); err != nil { - return err + return false, err } if _, err := tx.Exec(`ALTER TABLE context_manifests DROP COLUMN ` + column); err != nil { - return err + return false, err } } - return nil + return found, nil } func latestMigrationVersion() int {