import test from 'node:test';
import assert from 'node:assert/strict';
import { DatabaseSync } from 'node:sqlite';
import { chmodSync, copyFileSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { createHash } from 'node:crypto';
import { fileURLToPath, pathToFileURL } from 'node:url';
import { Store, schemaDigest } from '../src/store.mjs';
import { Broker } from '../src/broker.mjs';

// Row 264a (#1769, T1 item 7): sessions below a thread, in the bus. T1's bus half, T2 and T7 of
// docs/plans/2026-10-21_t1-sessions.md. The launcher, seat and Console halves are 264b-264e's.
const shipped = (name) =>
  JSON.parse(readFileSync(fileURLToPath(new URL(`../../../roles/${name}.json`, import.meta.url)), 'utf8'));
const roles = {
  chief: { definition: 'chief-of-staff', authority: { withinRole: ['message.send'], crossRole: [] } },
  assistant: { definition: 'assistant', authority: { withinRole: ['message.send'], crossRole: [] } },
  pm: { definition: 'pm', authority: { withinRole: ['message.send'], crossRole: [] } },
};
const businesses = {
  demo: {
    id: 'demo',
    human: 'jason',
    arbiters: { technical: 'pm', delivery: 'chief' },
    roles,
    definitions: { 'chief-of-staff': {}, assistant: { messaging: shipped('assistant').messaging }, pm: {} },
    projects: { stack: { root: '/nonexistent' } },
  },
};
const AT = '2026-10-21T00:00:00.000Z';
const now = () => new Date().toISOString();
const here = (p) => new URL(p, import.meta.url);
const ddl3k = readFileSync(here('./fixtures/schema-3k.sql'), 'utf8');

function broker(t, store) {
  const b = new Broker({ store, businesses });
  const human = b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true });
  const caps = { human };
  const call = (who, verb, args = {}) => b.request(caps[who], { verb, args });
  const launch = (name, role, extra = {}) => {
    caps[name] = b.bindLaunch({ business: 'demo', role, run: `${name}-run`, harness: 'pi', ...extra });
    return caps[name];
  };
  const claim = (name) => call(name, 'role.claim');
  const sessionOf = (message) => store.get('SELECT session FROM message_threads WHERE message=?', message)?.session ?? null;
  const claims = (session) => store.all('SELECT op, holder_run, generation FROM thread_claims WHERE session=? ORDER BY seq', session).map((r) => ({ ...r }));
  return { b, store, caps, call, launch, claim, sessionOf, claims };
}

function setup(t) {
  const root = mkdtempSync(join(tmpdir(), 'bus-sessions-'));
  const store = new Store(root);
  t.after(() => {
    store.close();
    rmSync(root, { recursive: true, force: true });
  });
  const ctx = broker(t, store);
  const project = ctx.call('human', 'project.create', { name: 'stack work', repo: 'stack' }).id;
  const A = ctx.call('human', 'thread.create', { project, name: 'alpha' }).id;
  for (const r of ['chief', 'pm']) {
    ctx.launch(r, r);
    ctx.claim(r);
  }
  const [first] = ctx.call('human', 'session.list', { thread: A });
  return { ...ctx, root, project, A, D: first.id };
}

test('T1 REQ-ROLE-2: two sessions of one thread keep their own history, delivery and replies', (t) => {
  const { call, launch, claim, sessionOf, store, project, A, D } = setup(t);
  // thread.create made one default session, with a random id that isn't the thread's.
  assert.deepEqual(call('human', 'session.list', { thread: A }).map((r) => [r.name, r.predecessor, r.default]), [['default', null, true]]);
  assert.notEqual(D, A);
  assert.match(D, /^[a-f0-9-]{36}$/);
  const S = call('human', 'session.create', { thread: A, name: 'second', op: 'op-second' }).id;
  assert.deepEqual(call('human', 'session.list', { thread: A }).map((r) => [r.id, r.name, r.default]), [
    [D, 'default', true],
    [S, 'second', false],
  ]);
  // A run with no session named works the default (D3). Each session has its own holder.
  launch('a1', 'assistant', { thread: A });
  launch('a2', 'assistant', { thread: A, session: S });
  assert.deepEqual(claim('a1'), { role: 'assistant', run: 'a1-run', thread: A });
  assert.deepEqual(claim('a2'), { role: 'assistant', run: 'a2-run', thread: A });
  const toDefault = call('human', 'thread.send', { thread: A, body: 'to the default' });
  const toSecond = call('human', 'thread.send', { thread: A, session: S, body: 'to the second' });
  assert.equal(sessionOf(toDefault.id), D);
  assert.equal(sessionOf(toSecond.id), S);
  assert.deepEqual(call('a1', 'message.receive').map((m) => m.body), ['to the default']);
  assert.deepEqual(call('a2', 'message.receive').map((m) => m.body), ['to the second']);
  assert.deepEqual(call('a1', 'message.receive'), []);
  // A reply stays in its run's session, and each history reads only its own.
  const reply = call('a2', 'message.send', { to: 'human', body: 'second answers', in_reply_to: toSecond.id });
  assert.equal(sessionOf(reply.id), S);
  assert.deepEqual(call('human', 'thread.messages', { thread: A }).map((m) => m.body), ['to the default']);
  assert.deepEqual(call('human', 'thread.messages', { thread: A, session: S }).map((m) => m.body), ['to the second', 'second answers']);
  // The chief's reply to the second session's work comes back to that session.
  const ask = call('a2', 'message.send', { to: 'chief', body: 'need help', class: 'REQUEST' });
  const back = call('chief', 'message.send', { to: 'assistant', body: 'here', in_reply_to: ask.id });
  assert.equal(sessionOf(back.id), S);
  assert.deepEqual(call('a2', 'message.receive').map((m) => m.body), ['here']);
  assert.deepEqual(call('a1', 'message.receive'), []);
  // An open names its session too.
  call('human', 'thread.open', { thread: A, session: S });
  assert.equal(store.get('SELECT session FROM thread_opens ORDER BY seq DESC LIMIT 1').session, S);
  // A non-thread role names no session, and a session of another thread is refused.
  assert.throws(() => launch('p2', 'pm', { session: S }), { code: 'thread-not-allowed' });
  const B = call('human', 'thread.create', { project, name: 'beta' }).id;
  assert.throws(() => launch('a3', 'assistant', { thread: B, session: S }), { code: 'session-not-found' });
  assert.throws(() => call('human', 'thread.send', { thread: B, session: S, body: 'x' }), { code: 'session-not-found' });
  assert.throws(() => call('human', 'thread.messages', { thread: B, session: 'no-such-session' }), { code: 'session-not-found' });
});

test('T2 REQ-ROLE-2: one holder per session, generations count its claims, and a sibling session\'s message never reaches the other holder', (t) => {
  const { b, call, launch, claim, claims, store, project, A, D } = setup(t);
  const S = call('human', 'session.create', { thread: A, name: 'second', op: 'op-second' }).id;
  launch('a1', 'assistant', { thread: A });
  claim('a1');
  launch('a1b', 'assistant', { thread: A });
  assert.throws(() => claim('a1b'), { code: 'thread-already-held' });
  // A sibling session is free to claim while the default is held.
  launch('a2', 'assistant', { thread: A, session: S });
  claim('a2');
  // Release and claim again: the second claim is generation 2, and its release carries it.
  call('a2', 'role.release');
  launch('a3', 'assistant', { thread: A, session: S });
  claim('a3');
  b.endLaunch({ business: 'demo', run: 'a3-run', reason: 'idle' });
  assert.deepEqual(claims(S), [
    { op: 'claim', holder_run: 'a2-run', generation: 1 },
    { op: 'release', holder_run: 'a2-run', generation: 1 },
    { op: 'claim', holder_run: 'a3-run', generation: 2 },
    { op: 'release', holder_run: 'a3-run', generation: 2 },
  ]);
  // The human's stop revokes each held session, with that claim's generation.
  launch('a4', 'assistant', { thread: A, session: S });
  claim('a4');
  assert.deepEqual(call('human', 'assistant.stop', { instance: 'assistant' }), {
    stopped: [
      { thread: A, run: 'a1-run' },
      { thread: A, run: 'a4-run' },
    ],
  });
  assert.deepEqual(claims(D).at(-1), { op: 'revoke', holder_run: 'a1-run', generation: 1 });
  assert.deepEqual(claims(S).at(-1), { op: 'revoke', holder_run: 'a4-run', generation: 3 });
  // The triggers hold these for a write that skips the broker.
  const raw = (thread, session, generation, run) =>
    store.run(
      "INSERT INTO thread_claims(at,business,thread,session,generation,op,holder_run,harness,by) VALUES(?,'demo',?,?,?,'claim',?,'pi',?)",
      now(),
      thread,
      session,
      generation,
      run,
      run,
    );
  assert.throws(() => raw(A, S, 5, 'x-run'), /a claim row names its session's generation/);
  assert.throws(() => raw(A, S, 3, 'x-run'), /a claim row names its session's generation/);
  const B = call('human', 'thread.create', { project, name: 'beta' }).id;
  const [{ id: BD }] = call('human', 'session.list', { thread: B });
  assert.throws(() => raw(A, BD, 1, 'x-run'), /a claim's session is in its thread/);
  raw(A, S, 4, 'x-run');
  assert.throws(() => raw(B, BD, 1, 'x-run'), /run already holds a session/);
  // A threaded message goes only to its own session's holder.
  launch('a5', 'assistant', { thread: A });
  claim('a5');
  const m = call('human', 'thread.send', { thread: A, session: S, body: 'for x' });
  assert.throws(
    () => store.run("INSERT INTO deliveries(message,at,op,holder_run) VALUES(?,?,'delivered',?)", m.id, now(), 'a5-run'),
    /a threaded message goes only to its session's holder/,
  );
  store.run("INSERT INTO deliveries(message,at,op,holder_run) VALUES(?,?,'delivered',?)", m.id, now(), 'x-run');
});

test('T2 sage-2257: a released or revoked run\'s retired generation refuses its approval, delivery and read, and writes nothing', (t) => {
  const { call, launch, claim, claims, store, A, D } = setup(t);
  const tables = store.all("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'").map((r) => r.name);
  const counts = () => Object.fromEntries(tables.map((n) => [n, store.get(`SELECT count(*) AS n FROM "${n}"`).n]));
  // The run raises a decision, takes one message and leaves a second waiting.
  const prepare = (name) => {
    launch(name, 'assistant', { thread: A });
    claim(name);
    const d = call(name, 'decision.raise', {
      action: 'spend',
      target: 'demo',
      question: 'spend?',
      options: [
        { key: 'yes', text: 'Yes' },
        { key: 'no', text: 'No' },
      ],
      recommendation: 'no',
      blocking: false,
    });
    call('human', 'thread.send', { thread: A, body: `for ${name}` });
    const [taken] = call(name, 'message.receive');
    call('human', 'thread.send', { thread: A, body: `waiting after ${name}` });
    return { d, taken };
  };
  // Each refusal leaves its one action.refused event (request()'s catch, #refused); nothing else is
  // written: no approval, delivery, read or claim state (mos-1277).
  const lastSeq = () => store.get('SELECT max(seq) AS s FROM events').s;
  const refusedOnce = (fn, label, run) => {
    const from = lastSeq();
    assert.throws(fn, { code: 'not-holder' }, label);
    const added = store.all('SELECT kind, actor_run, body FROM events WHERE seq > ? ORDER BY seq', from);
    assert.deepEqual(
      added.map((e) => [e.kind, e.actor_run, JSON.parse(e.body)]),
      [['action.refused', run, { code: 'not-holder' }]],
      label,
    );
  };
  const refused = (name, { d, taken }) => {
    const { events, ...before } = counts();
    refusedOnce(() => call(name, 'decision.withdraw', { id: d.id }), 'approval', `${name}-run`);
    refusedOnce(() => call(name, 'message.receive'), 'delivery', `${name}-run`);
    refusedOnce(() => call(name, 'message.read', { id: taken.id }), 'read', `${name}-run`);
    const { events: after, ...rest } = counts();
    assert.deepEqual(rest, before);
    assert.equal(after, events + 3);
  };
  // Released: generation 1 is retired once a2 claims generation 2.
  const one = prepare('a1');
  call('a1', 'role.release');
  const two = prepare('a2');
  refused('a1', one);
  // Revoked by the human's stop: generation 2 is retired once a3 claims generation 3.
  call('human', 'assistant.stop', { instance: 'assistant' });
  prepare('a3');
  refused('a2', two);
  assert.deepEqual(claims(D).map((r) => [r.op, r.holder_run, r.generation]), [
    ['claim', 'a1-run', 1],
    ['release', 'a1-run', 1],
    ['claim', 'a2-run', 2],
    ['revoke', 'a2-run', 2],
    ['claim', 'a3-run', 3],
  ]);
});

test('session.create is idempotent on its op, refuses a reused op or a taken name, and reaches only the owner\'s threads', (t) => {
  const { call, store, project, A, D } = setup(t);
  const first = call('human', 'session.create', { thread: A, name: 'second', op: 'op-1' });
  assert.deepEqual(call('human', 'session.create', { thread: A, name: 'second', op: 'op-1' }), first);
  assert.throws(() => call('human', 'session.create', { thread: A, name: 'third', op: 'op-1' }), { code: 'session-op-conflict' });
  assert.throws(() => call('human', 'session.create', { thread: A, name: 'second', op: 'op-2' }), { code: 'session-exists' });
  assert.throws(() => call('human', 'session.create', { thread: A, name: 'default', op: 'op-3' }), { code: 'session-exists' });
  // thread.create's op answers a retry that names the default.
  assert.deepEqual(call('human', 'session.create', { thread: A, name: 'default', op: 'default' }), { id: D, thread: A, name: 'default' });
  for (const args of [{ thread: A, name: 'x\ny', op: 'op-4' }, { thread: A, name: 'x' }, { thread: A, name: 'x', op: 'op-5', extra: 1 }])
    assert.throws(() => call('human', 'session.create', args), { code: 'invalid-request' });
  assert.equal(store.get('SELECT count(*) n FROM sessions WHERE thread=?', A).n, 2);
  // Another owner's thread, made by hand with the session the broker would write.
  const other = 'thread-of-someone-else';
  store.run('INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES(?,?,?,?,?,?,?)', other, now(), 'demo', project, 'assistant', 'someone-else', 'theirs');
  store.run("INSERT INTO sessions(id,at,business,thread,instance,owner,name,op) VALUES('s-theirs',?,'demo',?,'assistant','someone-else','default','default')", now(), other);
  store.run("INSERT INTO session_defaults(at,thread,session,by) VALUES(?,?,'s-theirs','someone-else')", now(), other);
  for (const [verb, args] of [
    ['session.list', { thread: other }],
    ['session.create', { thread: other, name: 'mine', op: 'op-6' }],
    ['thread.send', { thread: other, session: 's-theirs', body: 'hi' }],
  ])
    assert.throws(() => call('human', verb, args), { code: 'thread-not-owned' }, verb);
  assert.throws(() => call('human', 'session.list', { thread: 'nope' }), { code: 'thread-not-found' });
  // Agents don't call the session verbs.
  assert.throws(() => call('pm', 'session.list', { thread: A }), { code: 'human-required' });
});

test('the 3l triggers keep sessions in their thread and session ids apart from thread ids', (t) => {
  // A bare database from schema.sql, so only the triggers stand between a write and the tables.
  const root = mkdtempSync(join(tmpdir(), 'bus-sessions-sql-'));
  const db = new DatabaseSync(join(root, 'bus.sqlite'));
  t.after(() => {
    db.close();
    rmSync(root, { recursive: true, force: true });
  });
  db.exec(readFileSync(here('../schema.sql'), 'utf8'));
  const run = (sql, ...args) => db.prepare(sql).run(...args);
  const session = (id, thread, extra = {}) =>
    run(
      'INSERT INTO sessions(id,at,business,thread,instance,owner,name,op,predecessor) VALUES(?,?,?,?,?,?,?,?,?)',
      id,
      now(),
      extra.business ?? 'demo',
      thread,
      extra.instance ?? 'assistant',
      extra.owner ?? 'jason',
      extra.name ?? id,
      extra.op ?? id,
      extra.predecessor ?? null,
    );
  run("INSERT INTO projects(id,at,business,name,repo,created_by) VALUES('p-1',?,'demo','work',NULL,'jason')", now());
  for (const id of ['th-1', 'th-2'])
    run("INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES(?,?,'demo','p-1','assistant','jason',?)", id, now(), id);
  session('s-1', 'th-1');
  // The legacy shape: a session whose id is its own thread's.
  session('th-1', 'th-1', { name: 'default', op: 'legacy-3l' });
  // A session id that names another thread, and a thread id that names a session, are refused.
  assert.throws(() => session('th-2', 'th-1'), /a session id names another thread/);
  assert.throws(
    () => run("INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES('s-1',?,'demo','p-1','assistant','jason','x')", now()),
    /a thread id names a session/,
  );
  // A session has its thread's business, instance and owner, and its predecessor is in its thread.
  for (const extra of [{ business: 'other' }, { instance: 'pm' }, { owner: 'someone-else' }])
    assert.throws(() => session('s-x', 'th-1', extra), /a session has its thread's business, instance and owner/, JSON.stringify(extra));
  assert.throws(() => session('s-x', 'th-2', { predecessor: 's-1' }), /a session's predecessor is in its thread/);
  session('s-2', 'th-1', { predecessor: 's-1' });
  // The names and op keys are unique in a thread.
  assert.throws(() => session('s-3', 'th-1', { name: 's-1' }), /UNIQUE/);
  assert.throws(() => session('s-3', 'th-1', { op: 's-1' }), /UNIQUE/);
  // A default is a session of its thread.
  assert.throws(() => run("INSERT INTO session_defaults(at,thread,session,by) VALUES(?,'th-2','s-1','jason')", now()), /a default session is in its thread/);
  run("INSERT INTO session_defaults(at,thread,session,by) VALUES(?,'th-1','s-1','jason')", now());
  // A message's and an open's session is in their thread.
  run("INSERT INTO messages(id,at,business,from_role,from_run,to_role,class,body) VALUES('m-1',?,'demo','human','human-cli','assistant','INFO','hi')", now());
  assert.throws(() => run("INSERT INTO message_threads(message,thread,session) VALUES('m-1','th-2','s-1')"), /a threaded message's session is in its thread/);
  assert.throws(() => run("INSERT INTO thread_opens(at,thread,session,by) VALUES(?,'th-2','s-1','jason')", now()), /an open's session is in its thread/);
});

// T7: a 3k store from the pinned 3k schema (8f51900b5's, the parent of this change's schema.sql)
// and fixture rows. No live store is read or copied.
function source3k(t) {
  const root = mkdtempSync(join(tmpdir(), 'sessions-migrate-'));
  t.after(() => rmSync(root, { recursive: true, force: true }));
  mkdirSync(join(root, 'bus'), { mode: 0o700 });
  const path = join(root, 'bus/bus.sqlite'), db = new DatabaseSync(path);
  assert.equal(createHash('sha256').update(ddl3k).digest('hex'), 'aba9240825a35fdcc741439d0708efa33afcf519d11cac9952c7207ad99250f0');
  db.exec(ddl3k);
  const run = (sql, ...args) => db.prepare(sql).run(...args);
  run('INSERT INTO meta VALUES (?,?)', 'schema_version', '3k');
  run('INSERT INTO meta VALUES (?,?)', 'schema_digest', schemaDigest(db));
  run('INSERT INTO meta VALUES (?,?)', 'comment_cursor_epoch', AT);
  run('INSERT INTO projects(id,at,business,name,repo,created_by) VALUES(?,?,?,?,?,?)', 'p-1', AT, 'demo', 'stack work', 'stack', 'jason');
  for (const id of ['t1', 't2'])
    run('INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES(?,?,?,?,?,?,?)', id, AT, 'demo', 'p-1', 'assistant', 'jason', id);
  // Duckworth's fixture: claim, release, claim, revoke on t1, with t2's claim between, at seqs
  // with gaps so a renumbered copy would show.
  const claimRow = (seq, thread, op, holder, extra = {}) =>
    run(
      'INSERT INTO thread_claims(seq,at,business,thread,op,holder_run,harness,by,reason) VALUES(?,?,?,?,?,?,?,?,?)',
      seq, AT, 'demo', thread, op, holder, 'pi', extra.by ?? holder, extra.reason ?? null,
    );
  claimRow(10, 't1', 'claim', 'a-run');
  claimRow(11, 't2', 'claim', 'b-run');
  claimRow(12, 't1', 'release', 'a-run');
  claimRow(20, 't1', 'claim', 'c-run');
  claimRow(40, 't1', 'revoke', 'c-run', { by: 'jason', reason: 'stuck' });
  for (const [id, thread] of [['m-1', 't1'], ['m-2', 't2'], ['m-3', 't1']]) {
    run('INSERT INTO messages(id,at,business,from_role,from_run,to_role,class,body) VALUES(?,?,?,?,?,?,?,?)', id, AT, 'demo', 'human', 'human-cli', 'assistant', 'INFO', id);
    run('INSERT INTO message_threads VALUES(?,?)', id, thread);
  }
  run('INSERT INTO thread_opens(seq,at,thread,by) VALUES(?,?,?,?)', 7, AT, 't1', 'jason');
  db.close(); chmodSync(path, 0o600);
  return { root, path };
}
const master = (path) => {
  const db = new DatabaseSync(path, { readOnly: true });
  try {
    return db.prepare('SELECT type,name,tbl_name,sql FROM sqlite_master ORDER BY type,name').all().map((r) => ({ ...r }));
  } finally { db.close(); }
};
const rows = (path) => {
  const db = new DatabaseSync(path, { readOnly: true });
  try {
    return Object.fromEntries(['sessions', 'session_defaults', 'thread_claims', 'message_threads', 'thread_opens', 'meta']
      .map((name) => [name, db.prepare(`SELECT * FROM ${name} ORDER BY rowid`).all().map((r) => ({ ...r }))]));
  } finally { db.close(); }
};
const version = (path) => {
  const db = new DatabaseSync(path, { readOnly: true });
  try { return db.prepare("SELECT value FROM meta WHERE key='schema_version'").get().value; } finally { db.close(); }
};

test('T7: a 3k store migrates to 3l, each thread\'s default session has the thread\'s id, and old rows keep their seqs', (t) => {
  const { root, path } = source3k(t);
  assert.throws(() => new Store(root), /schema-mismatch/);
  new Store(root, { migrateSchema: true }).close();
  assert.equal(version(path), '3v');
  // The migrated schema is the fresh one, object for object.
  const freshRoot = mkdtempSync(join(tmpdir(), 'sessions-fresh-'));
  t.after(() => rmSync(freshRoot, { recursive: true, force: true }));
  new Store(freshRoot).close();
  assert.deepEqual(master(path), master(join(freshRoot, 'bus/bus.sqlite')));
  const store = new Store(root);
  t.after(() => store.close());
  assert.deepEqual(store.all('SELECT id,at,business,thread,instance,owner,name,op,predecessor FROM sessions ORDER BY seq').map((r) => ({ ...r })), ['t1', 't2'].map((id) => (
    { id, at: AT, business: 'demo', thread: id, instance: 'assistant', owner: 'jason', name: 'default', op: 'legacy-3l', predecessor: null })));
  assert.deepEqual(store.all('SELECT at,thread,session,by FROM session_defaults ORDER BY seq').map((r) => ({ ...r })), ['t1', 't2'].map((id) => (
    { at: AT, thread: id, session: id, by: 'migration-3l' })));
  // Generation counts claim rows only: claim, release, claim, revoke on t1 is 1, 1, 2, 2.
  assert.deepEqual(store.all('SELECT seq,thread,session,op,generation FROM thread_claims ORDER BY seq').map((r) => ({ ...r })), [
    { seq: 10, thread: 't1', session: 't1', op: 'claim', generation: 1 },
    { seq: 11, thread: 't2', session: 't2', op: 'claim', generation: 1 },
    { seq: 12, thread: 't1', session: 't1', op: 'release', generation: 1 },
    { seq: 20, thread: 't1', session: 't1', op: 'claim', generation: 2 },
    { seq: 40, thread: 't1', session: 't1', op: 'revoke', generation: 2 },
  ]);
  assert.deepEqual(store.all('SELECT message,thread,session FROM message_threads ORDER BY rowid').map((r) => ({ ...r })), [
    { message: 'm-1', thread: 't1', session: 't1' },
    { message: 'm-2', thread: 't2', session: 't2' },
    { message: 'm-3', thread: 't1', session: 't1' },
  ]);
  assert.deepEqual({ ...store.get('SELECT seq,thread,session,by FROM thread_opens') }, { seq: 7, thread: 't1', session: 't1', by: 'jason' });
  for (const [name, seq] of [['thread_claims', 40], ['thread_opens', 7]])
    assert.equal(store.get('SELECT seq FROM sqlite_sequence WHERE name=?', name).seq, seq, name);
  assert.equal(store.get('PRAGMA quick_check').quick_check, 'ok');
  assert.deepEqual(store.all('PRAGMA foreign_key_check'), []);
  // A legacy client names no session and gets the default: t2's holder releases at generation 1,
  // and t1's next claim is its third.
  const { call, launch, claim, claims, sessionOf } = broker(t, store);
  for (const r of ['chief', 'pm']) {
    launch(r, r);
    claim(r);
  }
  assert.throws(() => call('human', 'session.create', { thread: 't1', name: 'default', op: 'op-new' }), { code: 'session-exists' });
  store.run("INSERT INTO thread_claims(at,business,thread,session,generation,op,holder_run,harness,by) VALUES(?,'demo','t2','t2',1,'release','b-run','pi','b-run')", now());
  assert.equal(store.get("SELECT seq FROM thread_claims WHERE thread='t2' ORDER BY seq DESC LIMIT 1").seq, 41);
  launch('d', 'assistant', { thread: 't1' });
  claim('d');
  assert.deepEqual(claims('t1').at(-1), { op: 'claim', holder_run: 'd-run', generation: 3 });
  const sent = call('human', 'thread.send', { thread: 't1', body: 'after the step' });
  assert.equal(sessionOf(sent.id), 't1');
  assert.deepEqual(call('d', 'message.receive').map((m) => m.body), ['m-1', 'm-3', 'after the step']);
  assert.deepEqual(call('human', 'thread.messages', { thread: 't1' }).map((m) => m.body), ['m-1', 'm-3', 'after the step']);
});

// The 3k engine is store.mjs at this change with the 3k schema and version and no 3k→3l step.
async function engine3k(root) {
  const moduleRoot = join(root, 'engine-3k');
  mkdirSync(join(moduleRoot, 'src/migrations'), { recursive: true });
  const source = readFileSync(here('../src/store.mjs'), 'utf8');
  const edited = source
    .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3k).digest('hex')}';`)
    .replace("export const SCHEMA_VERSION = '3v';", "export const SCHEMA_VERSION = '3k';")
    .replace(/\}, \{\n {2}\/\/ Row 264a \(#1769\)[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables');
  assert.ok(edited.includes("export const SCHEMA_VERSION = '3k';") && edited.includes('3j-3k') && !edited.includes('3k-3l'), 'the 3k engine has no 3k→3l step');
  writeFileSync(join(moduleRoot, 'src/store.mjs'), edited);
  writeFileSync(join(moduleRoot, 'schema.sql'), ddl3k);
  for (const file of ['3b.sql', '3b-3c.sql', '3c-3d.sql', '3d-3e.sql', '3e-3f.sql', '3f-3g.sql', '3g-3h.sql', '3h-3i.sql', '3i-3j.sql', '3j-3k.sql'])
    copyFileSync(here('../src/migrations/' + file), join(moduleRoot, 'src/migrations', file));
  return (await import(pathToFileURL(join(moduleRoot, 'src/store.mjs')))).Store;
}

test('T7 D9: the 3k engine refuses a 3l store, migrating or not, and leaves it as it was', async (t) => {
  const { root, path } = source3k(t);
  new Store(root, { migrateSchema: true }).close();
  const was = master(path), before = rows(path);
  const Store3k = await engine3k(root);
  assert.throws(() => new Store3k(root), /schema-mismatch/);
  assert.throws(() => new Store3k(root, { migrateSchema: true }), /schema-mismatch/);
  assert.equal(version(path), '3v');
  assert.deepEqual(master(path), was);
  assert.deepEqual(rows(path), before);
  // The same engine opens a fresh 3k store, so the refusal is the version's and not a broken copy.
  const { root: old } = source3k(t);
  new Store3k(old).close();
});