stack/packages/bus/tests/thread-migration.test.mjs

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 = '3r';", "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, '3r');
  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 3r ready', ''].join('\n'));
});