feat(storage): transactional schema migration history

README.md

@@ -417,3 +417,7 @@ The reviewer uses the configured, approved **free OpenRouter Nemotron** model an
 The production semantic corroboration search currently performs an exact quadratic pair comparison. The database already has a pgvector HNSW index, but an index existing does not establish that the planner actually uses it. Run `npm run diagnostics:knn` to compare a bounded LATERAL nearest-neighbor candidate query against a bounded exact reference on **current source-grounded problems and distinct source threads**. The report gives exact-top-K candidate coverage and missing pairs, without any model requests, vector writes or production query changes. Configurations: `PIE_KNN_AUDIT_ANCHORS` (1–200), `PIE_KNN_AUDIT_NEIGHBORS` (1–100), `PIE_KNN_AUDIT_THRESHOLD` (0–1) and `PIE_KNN_AUDIT_TOPK` (1–100).
 
 Local preflight with 44 embeddings returned 20/20 exact top candidates from 70 proposed neighbors. EXPLAIN chose a **sequential scan**, not HNSW, on this tiny local dataset. A representative larger dataset with measured query plans and acceptable candidate recall is required before replacing production with ANN retrieval.
+
+### Versioned and atomic PostgreSQL migration baseline
+
+`npm run db:migrate` now runs the idempotent schema baseline within one PostgreSQL transaction protected by an advisory lock. It writes a SHA-256 checksum of the applied schema to `pie_schema_migrations`, safely handles databases created before the migration ledger existed and becomes a no-op when the same version is already applied. Modifying the SQL file without creating a **new forward-only migration version** raises an explicit checksum mismatch; never silently overwrite history or drop live tables to correct it. This is a baseline integrity guard, not a complete multi-version migration framework.

packages/storage/migrate.mjs

@@ -1,11 +1,42 @@
 import {readFile} from 'node:fs/promises';
+import {createHash} from 'node:crypto';
 import {createPool} from './postgres.mjs';
 
-export async function migrate(pool){
- const ddl=await readFile(new URL('./schema.sql',import.meta.url),'utf8');
- await pool.query('CREATE EXTENSION IF NOT EXISTS vector');
- await pool.query(ddl);
- return {schema:'schema.sql',idempotent:true};
+// Increment this identifier only with an audited, forward-only schema change.
+// The baseline is intentionally compatible with previously initialized DBs.
+export const SCHEMA_MIGRATION_VERSION='baseline-2026-10-13';
+const lockSql='SELECT pg_advisory_xact_lock(18222026,1013)';
+const tableSql=`CREATE TABLE IF NOT EXISTS pie_schema_migrations (
+ version TEXT PRIMARY KEY,
+ checksum CHAR(64) NOT NULL,
+ applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+)`;
+export function schemaChecksum(ddl){
+ if(typeof ddl!=='string'||!ddl.trim())throw Error('Migration requires nonempty schema');
+ return createHash('sha256').update(ddl).digest('hex');
+}
+export async function migrate(pool,{schemaUrl=new URL('./schema.sql',import.meta.url),version=SCHEMA_MIGRATION_VERSION}={}){
+ if(typeof version!=='string'||!/^[-a-z0-9]{4,80}$/.test(version))throw Error('Invalid schema migration version');
+ const ddl=await readFile(schemaUrl,'utf8'),checksum=schemaChecksum(ddl);
+ const client=await pool.connect();
+ try{
+  await client.query('BEGIN');
+  await client.query(lockSql);
+  await client.query('CREATE EXTENSION IF NOT EXISTS vector');
+  await client.query(tableSql);
+  const previous=await client.query('SELECT checksum FROM pie_schema_migrations WHERE version=$1',[version]);
+  if(previous.rows.length){
+   if(previous.rows[0].checksum.trim()!==checksum)throw Error('Schema checksum mismatch; create a NEW forward-only migration version instead of editing an applied version');
+   await client.query('COMMIT');
+   return {schema:'schema.sql',version,checksum,applied:false,idempotent:true};
+  }
+  // Baseline DDL remains idempotent for machines upgraded from early PIE.
+  await client.query(ddl);
+  await client.query('INSERT INTO pie_schema_migrations(version,checksum) VALUES ($1,$2)',[version,checksum]);
+  await client.query('COMMIT');
+  return {schema:'schema.sql',version,checksum,applied:true,idempotent:true};
+ }catch(error){await client.query('ROLLBACK');throw error;}
+ finally{client.release();}
 }
 if(process.argv[1]?.endsWith('/packages/storage/migrate.mjs')){
  const pool=createPool();

tests/versioned-migrations.test.mjs

@@ -0,0 +1,51 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import {migrate,schemaChecksum,SCHEMA_MIGRATION_VERSION} from '../packages/storage/migrate.mjs';
+import {createPool} from '../packages/storage/postgres.mjs';
+
+test('baseline has a stable identifier and checksum protects schema changes',()=>{
+ assert.equal(SCHEMA_MIGRATION_VERSION,'baseline-2026-10-13');
+ assert.match(schemaChecksum('CREATE TABLE a()'),/^[0-9a-f]{64}$/);
+ assert.notEqual(schemaChecksum('CREATE TABLE a()'),schemaChecksum('CREATE TABLE b()'));
+ assert.throws(()=>schemaChecksum(''),/nonempty/);
+});
+test('migration starts transaction, locks, records hash, and returns idempotent on re-run',async()=>{
+ const executed=[];let applied=false;
+ const client={release(){executed.push('release');},query:async(sql,args)=>{
+  executed.push(sql);
+  if(sql.startsWith('SELECT checksum FROM pie_schema_migrations'))return {rows:applied?[{checksum:recorded}]:[]};
+  if(sql.startsWith('INSERT INTO pie_schema_migrations')){recorded=args[1];applied=true;}
+  return {rows:[]};
+ }};
+ let recorded;
+ const pool={connect:async()=>client};
+ const first=await migrate(pool);
+ assert.equal(first.applied,true);
+ assert.ok(executed.includes('COMMIT'));
+ assert.ok(executed.some(s=>s.includes('pg_advisory_xact_lock')));
+ const second=await migrate(pool);
+ assert.equal(second.applied,false);
+ assert.equal(first.checksum,second.checksum);
+ assert.equal(executed.filter(x=>x==='release').length,2);
+});
+test('checksum conflict triggers rollback rather than reapplying modified schema',async()=>{
+ const commands=[];
+ const client={release(){},query:async(sql)=>{
+  commands.push(sql);
+  if(sql.startsWith('SELECT checksum FROM pie_schema_migrations'))return {rows:[{checksum:'0'.repeat(64)}]};
+  return {rows:[]};
+ }};
+ await assert.rejects(migrate({connect:async()=>client}),/checksum mismatch/);
+ assert.equal(commands.at(-1),'ROLLBACK');
+ assert.ok(!commands.some(s=>s.startsWith('INSERT INTO pie_schema_migrations')));
+});
+test('real PostgreSQL migration is idempotent',async t=>{
+ if(process.env.PIE_INTEGRATION!=='1')return t.skip('PIE_INTEGRATION=1 required');
+ const pool=createPool();try{
+  const first=await migrate(pool);const second=await migrate(pool);
+  assert.equal(second.applied,false);
+  assert.equal(first.checksum,second.checksum);
+  const {rows}=await pool.query('SELECT checksum FROM pie_schema_migrations WHERE version=$1',[SCHEMA_MIGRATION_VERSION]);
+  assert.equal(rows[0].checksum.trim(),first.checksum);
+ }finally{await pool.end();}
+});