import test from 'node:test'; import assert from 'node:assert/strict'; import { chmodSync, copyFileSync, mkdtempSync, mkdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { createHash } from 'node:crypto'; import { spawnSync } from 'node:child_process'; import { DatabaseSync } from 'node:sqlite'; import { pathToFileURL } from 'node:url'; import { Store, schemaDigest } from '../src/store.mjs'; import { Broker } from '../src/broker.mjs'; const here = (path) => new URL(path, import.meta.url); const AT = '2026-10-15T00:00:00.000Z'; // [id, business, to_role, delivery ops] const MESSAGES = [ ['m-asst-waiting', 'demo', 'assistant', ['routed']], ['m-asst-delivered', 'demo', 'assistant', ['routed', 'delivered']], ['m-asst-read', 'demo', 'assistant', ['routed', 'delivered', 'read']], ['m-chief-waiting', 'demo', 'chief', []], ['m-human-waiting', 'demo', 'human', ['routed']], ['m-other-1', 'other', 'assistant', ['routed']], ['m-other-2', 'other', 'assistant', ['failed']], ]; const REPORT = [ 'kept 1 undelivered message(s) for demo/assistant without a thread; if that role is a thread instance, its thread runs never receive them, so the human re-sends them with thread.send', 'kept 1 undelivered message(s) for demo/chief without a thread; if that role is a thread instance, its thread runs never receive them, so the human re-sends them with thread.send', 'kept 1 undelivered message(s) for demo/human without a thread; if that role is a thread instance, its thread runs never receive them, so the human re-sends them with thread.send', 'kept 2 undelivered message(s) for other/assistant without a thread; if that role is a thread instance, its thread runs never receive them, so the human re-sends them with thread.send', ]; // A 3d bus with messages in every delivery state, made by the engine pinned at 3d, plus an untouched copy. async function database3d(t) { const root = mkdtempSync(join(tmpdir(), 'bus-thread-migrate-')); t.after(() => rmSync(root, { recursive: true, force: true })); mkdirSync(join(root, 'bus'), { mode: 0o700 }); const path = join(root, 'bus/bus.sqlite'); const db = new DatabaseSync(path); db.exec(readFileSync(here('./fixtures/schema-3b.sql'), 'utf8')); db.prepare('INSERT INTO meta VALUES (?,?)').run('schema_version', '3b'); db.prepare('INSERT INTO meta VALUES (?,?)').run('schema_digest', schemaDigest(db)); db.close(); chmodSync(path, 0o600); // The engine pinned at 3d: current store.mjs with the 3d→3e step removed and the 3d schema. const ddl3d = readFileSync(here('./fixtures/schema-3d.sql'), 'utf8'); assert.equal(createHash('sha256').update(ddl3d).digest('hex'), 'e4f1de9e16d2fab5e051fd3810189b28cc76a670cfe573b752d9f53bf9ca39b5'); const moduleRoot = join(root, 'engine-3d'); mkdirSync(join(moduleRoot, 'src/migrations'), { recursive: true }); const source = readFileSync(here('../src/store.mjs'), 'utf8'); const pinned = source .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3d).digest('hex')}';`) .replace("export const SCHEMA_VERSION = '3w';", "export const SCHEMA_VERSION = '3d';") .replace(/\}, \{\n {2}from: '3d',[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables'); assert.ok(!pinned.includes('3d-3e'), 'the 3d engine has no 3d→3e step'); writeFileSync(join(moduleRoot, 'src/store.mjs'), pinned); writeFileSync(join(moduleRoot, 'schema.sql'), ddl3d); for (const name of ['3b.sql', '3b-3c.sql', '3c-3d.sql']) copyFileSync(here('../src/migrations/' + name), join(moduleRoot, 'src/migrations', name)); const { Store: Store3d } = await import(pathToFileURL(join(moduleRoot, 'src/store.mjs'))); const old = new Store3d(root, { migrateSchema: true }); assert.equal(old.get("SELECT value FROM meta WHERE key='schema_version'").value, '3d'); old.transaction(() => { for (const [id, business, to, ops] of MESSAGES) { old.run( 'INSERT INTO messages(id,at,business,from_role,from_run,to_role,class,body) VALUES(?,?,?,?,?,?,?,?)', id, AT, business, 'human', 'human', to, 'info', 'before threads', ); for (const op of ops) old.run('INSERT INTO deliveries(message,at,op,holder_run) VALUES(?,?,?,?)', id, AT, op, op === 'routed' ? null : 'old-run'); } }); old.close(); const copy = join(root, 'bus-3d-copy.sqlite'); copyFileSync(path, copy); return { root, path, copy }; } const rows = (path) => { const db = new DatabaseSync(path, { readOnly: true }); try { return { messages: db.prepare('SELECT * FROM messages ORDER BY seq').all(), deliveries: db.prepare('SELECT * FROM deliveries ORDER BY seq').all(), }; } finally { db.close(); } }; // Row 141: 3f→3g adds messages.to_name, which reads null on a message from before names. Rows from // node:sqlite have a null prototype, so the expected rows keep one. const named = ({ messages, deliveries }) => ({ messages: messages.map((m) => Object.assign(Object.create(null), m, { to_name: null })), deliveries, }); test('REQ-THREAD-1 REQ-ORG-2: the 3d→3e migration adds the thread tables and leaves messages from before threads as written', async (t) => { const { root, path, copy } = await database3d(t); const before = rows(path); assert.equal(before.messages.length, MESSAGES.length); // A 3d bus does not open as 3e until the explicit step runs. assert.throws(() => new Store(root), /schema-mismatch/); const store = new Store(root, { migrateSchema: true }); t.after(() => store.close()); assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3w'); for (const table of ['projects', 'threads', 'thread_claims', 'message_threads', 'thread_opens']) assert.equal(store.get(`SELECT count(*) n FROM ${table}`).n, 0, table); // Parked: the undelivered ones are counted per business and role, and none is changed or given a thread. assert.deepEqual(store.migrationReport, REPORT); store.close(); assert.deepEqual(rows(path), named(before)); assert.deepEqual(rows(copy), before); // Opening again needs no step and reports nothing. const again = new Store(root, { migrateSchema: true }); assert.deepEqual(again.migrationReport, []); again.close(); // A thread run of the instance never receives the old message, and the thread does not count it. const current = new Store(root); const b = new Broker({ store: current, businesses: { demo: { id: 'demo', human: 'jason', arbiters: { technical: 'chief', delivery: 'chief' }, roles: { chief: { definition: 'chief-of-staff', authority: { withinRole: [], crossRole: [] } }, assistant: { definition: 'assistant', authority: { withinRole: ['message.send'], crossRole: [] } }, }, definitions: { 'chief-of-staff': {}, assistant: {} }, projects: { stack: { root: '/nonexistent' } }, }, }, }); const human = b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true }); const project = b.request(human, { verb: 'project.create', args: { name: 'stack work', repo: 'stack' } }).id; const thread = b.request(human, { verb: 'thread.create', args: { project, name: 'after 3e' } }).id; const run = b.bindLaunch({ business: 'demo', role: 'assistant', run: 'assistant-run', harness: 'pi', thread }); b.request(run, { verb: 'role.claim', args: {} }); assert.deepEqual(b.threadsState('demo').map((r) => [r.thread, r.holder, r.pending, r.unread]), [ [thread, 'assistant-run', 0, 0], ]); assert.deepEqual(b.request(run, { verb: 'message.receive', args: {} }), []); current.close(); assert.deepEqual(rows(path), named(before)); }); test('REQ-THREAD-1: a 3d bus with a changed schema is refused and keeps its version and tables', async (t) => { const { root, path } = await database3d(t); const db = new DatabaseSync(path); db.exec('DROP TRIGGER messages_no_delete'); db.close(); assert.throws(() => new Store(root, { migrateSchema: true }), /schema-mismatch/); const check = new DatabaseSync(path, { readOnly: true }); assert.equal(check.prepare("SELECT value FROM meta WHERE key='schema_version'").get().value, '3d'); assert.equal(check.prepare("SELECT count(*) n FROM sqlite_master WHERE name='threads'").get().n, 0); check.close(); }); test('REQ-THREAD-1: the offline command prints the kept-message counts before the ready line', async (t) => { const { root } = await database3d(t); const command = new URL('../src/migrate-schema.mjs', import.meta.url).pathname; const migrated = spawnSync(process.execPath, [command, root], { encoding: 'utf8' }); assert.equal(migrated.status, 0, migrated.stderr); assert.equal(migrated.stdout, [...REPORT, 'bus schema 3w ready', ''].join('\n')); });