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 271 (264d, #1776, T4): clear context, the bus half. D7 and D8 of
// docs/plans/2026-10-21_t1-sessions.md, the contract in agents/gizmoduck/work/queue-271/BUILD.md.
// The launcher's stop, proof and start are in packages/cli/tests/session-clear.test.mjs.
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: [] } },
};
const GENERATION = '0b6e7f6e-4c1d-4a8e-9a51-6d2f3c4b5a69';
const UUID8 = /^[0-9a-f]{8}-[0-9a-f]{4}-8[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/;

function setup(t, { access = null } = {}) {
  const root = mkdtempSync(join(tmpdir(), 'bus-session-clear-'));
  const store = new Store(root);
  t.after(() => {
    store.close();
    rmSync(root, { recursive: true, force: true });
  });
  const sw = { state: { version: 1, halted: false, generation: GENERATION } };
  const b = new Broker({
    store,
    businesses: {
      demo: {
        id: 'demo',
        human: 'jason',
        arbiters: { technical: 'chief', delivery: 'chief' },
        vars: { 'bus.tester': true },
        roles,
        definitions: { 'chief-of-staff': {}, assistant: { messaging: shipped('assistant').messaging } },
        projects: { stack: { root: '/nonexistent' } },
      },
    },
    processing: { read: () => sw.state },
    ...(access ? { access: () => access.doc } : {}),
  });
  const caps = {
    human: b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true }),
    tester: b.bindTester({ business: 'demo' }),
  };
  const call = (who, verb, args = {}) => b.request(caps[who], { verb, args });
  const code = (fn) => {
    try {
      fn();
    } catch (e) {
      return e.code ?? e.message;
    }
    return null;
  };
  const launch = (name, thread, session) => {
    caps[name] = b.bindLaunch({ business: 'demo', role: 'assistant', run: `${name}-run`, harness: 'pi', thread, ...(session ? { session } : {}) });
    return call(name, 'role.claim');
  };
  caps.chief = b.bindLaunch({ business: 'demo', role: 'chief', run: 'chief-run', harness: 'pi' });
  call('chief', 'role.claim');
  const project = call('human', 'project.create', { name: 'stack work', repo: 'stack' }).id;
  const A = call('human', 'thread.create', { project, name: 'alpha' }).id;
  const list = (thread = A) => call('human', 'session.list', { thread });
  const D = list()[0].id;
  const stages = (op, thread = A) => store.all('SELECT stage FROM session_ops WHERE thread=? AND op=? ORDER BY seq', thread, op).map((r) => r.stage);
  const cancels = (session) => store.all('SELECT message FROM message_cancels WHERE session=? ORDER BY seq', session).map((r) => r.message);
  const count = (table) => store.get(`SELECT count(*) AS n FROM ${table}`).n;
  const clear = (who, args) => call(who, 'session.clear', { thread: A, confirm: true, ...args });
  const where = (op, thread = A) => ({ business: 'demo', thread, op });
  return { b, store, sw, caps, call, code, launch, project, A, D, list, stages, cancels, count, clear, where };
}

const successorOf = (thread, session, op) => {
  const h = createHash('sha256').update(JSON.stringify(['demo', thread, session, op])).digest('hex').split('');
  h[12] = '8';
  h[16] = '89ab'[parseInt(h[16], 16) & 3];
  const x = h.join('');
  return `${x.slice(0, 8)}-${x.slice(8, 12)}-${x.slice(12, 16)}-${x.slice(16, 20)}-${x.slice(20, 32)}`;
};

test('T4 D7: a clear of a held default session fences it, drains at the next receive, retires at the exit proof and moves the default', (t) => {
  const { b, call, code, launch, A, D, list, stages, clear, where, store } = setup(t);
  launch('a1', A);
  call('human', 'thread.send', { thread: A, body: 'first turn' });
  assert.deepEqual(call('a1', 'message.receive').map((m) => m.body), ['first turn']);
  assert.deepEqual(list().map((r) => [r.id, r.state, r.revision, r.successor, r.default]), [[D, 'active', 0, null, true]]);
  const accepted = clear('human', { session: D, revision: 0, op: 'clear-1' });
  const S = successorOf(A, D, 'clear-1');
  assert.match(S, UUID8);
  assert.deepEqual({ ...accepted, at: null }, { op: 'clear-1', thread: A, session: D, successor: S, state: 'clearing', start: null, refusal: null, at: null });
  // Accept fences the session: no send, open or new claim reaches it. The holder keeps its claim.
  assert.equal(code(() => call('human', 'thread.send', { thread: A, body: 'late' })), 'session-clearing');
  assert.equal(code(() => call('human', 'thread.open', { thread: A })), 'session-clearing');
  assert.equal(code(() => launch('a2', A)), 'session-clearing');
  assert.deepEqual(stages('clear-1'), ['requested']);
  assert.deepEqual(list().map((r) => [r.state, r.revision]), [['clearing', 1]]);
  // The run's next receive, at its completed turn, records the drain once and delivers nothing.
  assert.deepEqual(call('a1', 'message.receive'), []);
  assert.deepEqual(call('a1', 'message.receive'), []);
  assert.deepEqual(stages('clear-1'), ['requested', 'drained']);
  assert.deepEqual(
    { ...store.get("SELECT run, generation FROM session_ops WHERE op='clear-1' AND stage='drained'") },
    { run: 'a1-run', generation: 1 },
  );
  assert.deepEqual(b.sessionClears('demo'), [
    { thread: A, instance: 'assistant', session: D, op: 'clear-1', stage: 'drained', run: 'a1-run', generation: 1, successor: S, refusal: null },
  ]);
  // No proof until the run itself has ended: a signal or a dead socket isn't one.
  assert.equal(code(() => b.sessionClearExited({ ...where('clear-1'), run: 'a1-run' })), 'run-not-ended');
  assert.equal(code(() => b.sessionClearExited({ ...where('clear-1'), run: 'other-run' })), 'not-drained-run');
  assert.equal(code(() => b.sessionClearRetire(where('clear-1'))), 'not-exited');
  b.endLaunch({ business: 'demo', run: 'a1-run', reason: 'stopped', exitCode: 0 });
  assert.equal(b.sessionClearExited({ ...where('clear-1'), run: 'a1-run' }).state, 'clearing');
  // A repeat of the proof answers the envelope and appends nothing.
  b.sessionClearExited({ ...where('clear-1'), run: 'a1-run' });
  assert.deepEqual(stages('clear-1'), ['requested', 'drained', 'exited']);
  const retired = b.sessionClearRetire(where('clear-1'));
  assert.deepEqual([retired.state, retired.start, retired.successor], ['retired', null, S]);
  assert.deepEqual(list().map((r) => [r.id, r.name, r.predecessor, r.state, r.successor, r.default]), [
    [D, 'default', null, 'retired', S, false],
    [S, 'default-1', D, 'active', null, true],
  ]);
  // A retired session has no threadsState row; the successor has one and no message.
  assert.deepEqual(b.threadsState('demo').map((r) => [r.session, r.pending, r.holder]), [[S, 0, null]]);
  assert.equal(code(() => call('human', 'thread.send', { thread: A, session: D, body: 'to old' })), 'session-retired');
  assert.equal(code(() => clear('human', { session: D, revision: 4, op: 'clear-2' })), 'session-retired');
  // A thread-only send now goes to the successor, the default.
  const m = call('human', 'thread.send', { thread: A, body: 'fresh' });
  assert.equal(store.get('SELECT session FROM message_threads WHERE message=?', m.id).session, S);
});

test('T4 D7: an unheld session drains at accept, its proof is run null, and a non-default clear leaves the default', (t) => {
  const { b, call, code, A, D, list, stages, clear, where } = setup(t);
  const N = call('human', 'session.create', { thread: A, name: 'side', op: 'op-side' }).id;
  clear('human', { session: N, revision: 0, op: 'clear-side' });
  assert.deepEqual(stages('clear-side'), ['requested', 'drained']);
  assert.deepEqual(b.sessionClears('demo').map((c) => [c.session, c.stage, c.run, c.generation]), [[N, 'drained', null, null]]);
  assert.equal(code(() => b.sessionClearExited({ ...where('clear-side'), run: 'x-run' })), 'not-drained-run');
  b.sessionClearExited({ ...where('clear-side'), run: null });
  b.sessionClearRetire(where('clear-side'));
  const S = successorOf(A, N, 'clear-side');
  assert.deepEqual(list().map((r) => [r.id, r.name, r.default, r.state]), [
    [D, 'default', true, 'active'],
    [N, 'side', false, 'retired'],
    [S, 'side-1', false, 'active'],
  ]);
});

test('T4 D7: a run that ends before its next receive drains at its end, with its generation', (t) => {
  const { b, call, launch, A, D, stages, clear, store } = setup(t);
  launch('a1', A);
  clear('human', { session: D, revision: 0, op: 'clear-end' });
  assert.deepEqual(stages('clear-end'), ['requested']);
  b.endLaunch({ business: 'demo', run: 'a1-run', reason: 'exited', exitCode: 0 });
  assert.deepEqual(stages('clear-end'), ['requested', 'drained']);
  assert.deepEqual({ ...store.get("SELECT run, generation FROM session_ops WHERE op='clear-end' AND stage='drained'") }, { run: 'a1-run', generation: 1 });
});

test('T4 D7: a run whose turn is running (delivered, unread) stays clearing; uncertain is recorded once and nothing more', (t) => {
  const { b, call, code, launch, A, D, stages, clear, where, list } = setup(t);
  launch('a1', A);
  call('human', 'thread.send', { thread: A, body: 'long turn' });
  call('a1', 'message.receive');
  clear('human', { session: D, revision: 0, op: 'clear-busy' });
  // Without a receive or an end, it isn't drained, so the launcher has nothing to stop.
  assert.deepEqual(b.sessionClears('demo').map((c) => c.stage), ['requested']);
  assert.equal(code(() => b.sessionClearUncertain(where('clear-busy'))), 'not-drained');
  assert.equal(code(() => b.sessionClearExited({ ...where('clear-busy'), run: 'a1-run' })), 'not-drained');
  assert.deepEqual(call('a1', 'message.receive'), []);
  assert.equal(b.sessionClearUncertain(where('clear-busy')).state, 'uncertain');
  b.sessionClearUncertain(where('clear-busy'));
  assert.deepEqual(stages('clear-busy'), ['requested', 'drained', 'uncertain']);
  assert.deepEqual(list().map((r) => r.state), ['uncertain']);
  assert.equal(code(() => call('human', 'thread.send', { thread: A, body: 'x' })), 'session-clearing');
  // Proof arriving later resumes the clear.
  b.endLaunch({ business: 'demo', run: 'a1-run', reason: 'stopped', exitCode: 0 });
  b.sessionClearExited({ ...where('clear-busy'), run: 'a1-run' });
  assert.equal(b.sessionClearRetire(where('clear-busy')).state, 'retired');
});

test('T4 D7 B1: undelivered inputs are cancelled with receipts at accept, never delivered to the successor, and a routed reply is cancelled at arrival', (t) => {
  const { b, call, launch, A, D, stages, cancels, clear, where, store } = setup(t);
  launch('a1', A);
  const ask = call('a1', 'message.send', { to: 'chief', body: 'need help', class: 'REQUEST' });
  const m1 = call('human', 'thread.send', { thread: A, body: 'waiting one' });
  const m2 = call('human', 'thread.send', { thread: A, body: 'waiting two' });
  clear('human', { session: D, revision: 0, op: 'clear-msgs' });
  assert.deepEqual(cancels(D), [m1.id, m2.id]);
  assert.deepEqual(b.threadsState('demo').map((r) => r.pending), [0]);
  // The chief's reply routes to the session being cleared: kept, cancelled, and the answer says why.
  const back = call('chief', 'message.send', { to: 'assistant', body: 'here', in_reply_to: ask.id });
  assert.equal(back.cancelled, 'session-clearing');
  assert.equal(back.waiting, undefined);
  assert.deepEqual(cancels(D), [m1.id, m2.id, back.id]);
  assert.deepEqual({ ...store.get('SELECT op FROM message_cancels WHERE message=?', back.id) }, { op: 'clear-msgs' });
  assert.deepEqual(call('a1', 'message.receive'), []);
  b.endLaunch({ business: 'demo', run: 'a1-run', reason: 'stopped', exitCode: 0 });
  b.sessionClearExited({ ...where('clear-msgs'), run: 'a1-run' });
  b.sessionClearRetire(where('clear-msgs'));
  const S = successorOf(A, D, 'clear-msgs');
  launch('s1', A, S);
  assert.deepEqual(call('s1', 'message.receive'), []);
  // A reply to the retired session's message is cancelled as retired.
  const late = call('chief', 'message.send', { to: 'assistant', body: 'later', in_reply_to: ask.id });
  assert.equal(late.cancelled, 'session-retired');
  assert.deepEqual(call('s1', 'message.receive'), []);
  assert.deepEqual(stages('clear-msgs'), ['requested', 'drained', 'exited', 'retired']);
});

test('T4 D7: repeat, conflict, a second op, a stale revision, confirm and a halt', (t) => {
  const { sw, call, code, A, D, stages, clear, count } = setup(t);
  // A halt refuses and writes nothing.
  sw.state = { version: 1, halted: true, generation: GENERATION };
  assert.equal(code(() => clear('human', { session: D, revision: 0, op: 'clear-1' })), 'processing-halted');
  assert.equal(count('session_ops'), 0);
  sw.state = { version: 1, halted: false, generation: GENERATION };
  assert.equal(code(() => clear('human', { session: D, revision: 1, op: 'clear-1' })), 'session-revision-stale');
  assert.equal(code(() => call('human', 'session.clear', { thread: A, session: D, revision: 0, op: 'clear-1' })), 'invalid-request');
  assert.equal(code(() => call('human', 'session.clear', { thread: A, session: D, revision: 0, confirm: 'yes', op: 'clear-1' })), 'confirm-required');
  assert.equal(code(() => clear('human', { session: D, revision: -1, op: 'clear-1' })), 'invalid-request');
  assert.equal(code(() => clear('human', { session: 'no-such-session', revision: 0, op: 'clear-1' })), 'session-not-found');
  assert.equal(count('session_ops'), 0);
  const first = clear('human', { session: D, revision: 0, op: 'clear-1' });
  // The same op and payload answers the same envelope, id and name, and appends nothing.
  assert.deepEqual(clear('human', { session: D, revision: 0, op: 'clear-1' }), first);
  assert.deepEqual(stages('clear-1'), ['requested', 'drained']);
  assert.equal(code(() => clear('human', { session: D, revision: 1, op: 'clear-1' })), 'session-op-conflict');
  assert.equal(code(() => clear('human', { session: D, revision: 2, op: 'clear-2' })), 'session-clearing');
  // B3: an op names a session or a clear in its thread, never both.
  assert.equal(code(() => call('human', 'session.create', { thread: A, name: 'other', op: 'clear-1' })), 'session-op-conflict');
  call('human', 'session.create', { thread: A, name: 'side', op: 'op-side' });
  const side = call('human', 'session.list', { thread: A }).find((r) => r.name === 'side').id;
  assert.equal(code(() => clear('human', { session: side, revision: 0, op: 'op-side' })), 'session-op-conflict');
  assert.equal(count('session_ops'), 2);
});

test('T4 Q4: the successor takes <old>-N, the next N free, and a bound name stays taken', (t) => {
  const { b, call, code, A, D, list, clear, where } = setup(t);
  call('human', 'session.create', { thread: A, name: 'default-1', op: 'op-d1' });
  clear('human', { session: D, revision: 0, op: 'clear-1' });
  assert.equal(code(() => call('human', 'session.create', { thread: A, name: 'default-2', op: 'op-d2' })), 'session-exists');
  // Cancelled before retirement: the name stays bound to the op.
  call('human', 'session.cancel', { thread: A, op: 'clear-1', confirm: true });
  assert.equal(code(() => call('human', 'session.create', { thread: A, name: 'default-2', op: 'op-d2' })), 'session-exists');
  clear('human', { session: D, revision: 3, op: 'clear-2' });
  b.sessionClearExited({ ...where('clear-2'), run: null });
  b.sessionClearRetire(where('clear-2'));
  assert.deepEqual(list().map((r) => [r.name, r.state]), [
    ['default', 'retired'],
    ['default-1', 'active'],
    ['default-3', 'active'],
  ]);
});

test('T4 D8: the successor start is held, refused or started, each report on its own counter', (t) => {
  const { b, code, A, D, stages, clear, where, store } = setup(t);
  clear('human', { session: D, revision: 0, op: 'clear-1' });
  assert.equal(code(() => b.sessionClearStart({ ...where('clear-1'), start: 'held', refusal: 'capacity-full' })), 'not-retired');
  b.sessionClearExited({ ...where('clear-1'), run: null });
  b.sessionClearRetire(where('clear-1'));
  assert.deepEqual(b.sessionClears('demo').map((c) => c.stage), ['retired']);
  assert.equal(code(() => b.sessionClearStart({ ...where('clear-1'), start: 'held' })), 'invalid-request');
  assert.equal(code(() => b.sessionClearStart({ ...where('clear-1'), start: 'started', refusal: 'x' })), 'invalid-request');
  assert.equal(code(() => b.sessionClearStart({ ...where('clear-1'), start: 'begun', refusal: null })), 'invalid-request');
  const held = b.sessionClearStart({ ...where('clear-1'), start: 'held', refusal: 'capacity-full' });
  assert.deepEqual([held.state, held.start, held.refusal], ['retired', 'held', null]);
  const refused = b.sessionClearStart({ ...where('clear-1'), start: 'refused', refusal: 'launch-failed' });
  assert.deepEqual([refused.start, refused.refusal], ['refused', 'launch-failed']);
  assert.deepEqual(b.sessionClears('demo').map((c) => [c.stage, c.refusal]), [['start-refused', 'launch-failed']]);
  b.sessionClearStart({ ...where('clear-1'), start: 'held', refusal: 'limit-reached' });
  const started = b.sessionClearStart({ ...where('clear-1'), start: 'started' });
  assert.deepEqual([started.state, started.start, started.refusal], ['retired', 'started', null]);
  assert.deepEqual(
    store.all("SELECT stage, attempt, work_attempt FROM session_ops WHERE op='clear-1' AND stage GLOB 'start-*' ORDER BY seq").map((r) => ({ ...r })),
    [
      { stage: 'start-held', attempt: 1, work_attempt: 1 },
      { stage: 'start-refused', attempt: 2, work_attempt: 1 },
      { stage: 'start-held', attempt: 3, work_attempt: 2 },
      { stage: 'start-started', attempt: 4, work_attempt: 2 },
    ],
  );
  assert.deepEqual(b.sessionClears('demo'), []);
  assert.equal(code(() => b.sessionClearStart({ ...where('clear-1'), start: 'started' })), 'session-clear-complete');
  assert.equal(stages('clear-1').length, 8);
});

test('T4 D7 N4: a halt at retirement waits and writes nothing; retirement goes once processing resumes', (t) => {
  const { b, sw, code, D, stages, clear, where } = setup(t);
  clear('human', { session: D, revision: 0, op: 'clear-1' });
  b.sessionClearExited({ ...where('clear-1'), run: null });
  sw.state = { version: 1, halted: true, generation: GENERATION };
  assert.equal(code(() => b.sessionClearRetire(where('clear-1'))), 'processing-halted');
  assert.deepEqual(stages('clear-1'), ['requested', 'drained', 'exited']);
  sw.state = { version: 1, halted: false, generation: GENERATION };
  assert.equal(b.sessionClearRetire(where('clear-1')).state, 'retired');
});

test('T4 sage-2828: a cancel before retirement makes the session active again, keeps the claim and the cancelled messages, and never makes the successor', (t) => {
  const { b, call, code, launch, A, D, list, stages, cancels, clear, where, store } = setup(t);
  launch('a1', A);
  // Routed to a1's claim but not delivered: the accept cancels it.
  const m = call('human', 'thread.send', { thread: A, body: 'waiting' });
  clear('human', { session: D, revision: 0, op: 'clear-1' });
  const S = successorOf(A, D, 'clear-1');
  assert.equal(code(() => call('human', 'session.cancel', { thread: A, op: 'clear-1' })), 'invalid-request');
  assert.equal(code(() => call('human', 'session.cancel', { thread: A, op: 'clear-1', confirm: false })), 'confirm-required');
  assert.equal(code(() => call('human', 'session.cancel', { thread: A, op: 'no-such-op', confirm: true })), 'operation-not-found');
  const cancelled = call('human', 'session.cancel', { thread: A, op: 'clear-1', confirm: true });
  assert.deepEqual([cancelled.state, cancelled.retired, cancelled.start, cancelled.successor], ['cancelled', null, null, S]);
  assert.deepEqual(call('human', 'session.cancel', { thread: A, op: 'clear-1', confirm: true }), cancelled);
  assert.deepEqual(call('human', 'session.operation', { thread: A, op: 'clear-1' }), cancelled);
  assert.deepEqual(stages('clear-1'), ['requested', 'cancelled']);
  assert.deepEqual(list().map((r) => [r.id, r.state, r.successor]), [[D, 'active', null]]);
  assert.equal(store.get('SELECT 1 AS x FROM sessions WHERE id=?', S), undefined);
  assert.deepEqual(b.sessionClears('demo'), []);
  assert.deepEqual(cancels(D), [m.id]);
  // The holder kept its claim: a new send reaches it, and the cancelled message never does.
  call('human', 'thread.send', { thread: A, body: 'after cancel' });
  assert.deepEqual(call('a1', 'message.receive').map((r) => r.body), ['after cancel']);
  // The launcher's late reports refuse.
  assert.equal(code(() => b.sessionClearExited({ ...where('clear-1'), run: null })), 'operation-not-open');
  assert.equal(code(() => b.sessionClearUncertain(where('clear-1'))), 'operation-not-open');
  assert.equal(code(() => b.sessionClearRetire(where('clear-1'))), 'not-exited');
  // A new clear may follow; its revision counts the cancelled op's rows.
  assert.deepEqual(list().map((r) => r.revision), [2]);
  assert.equal(clear('human', { session: D, revision: 2, op: 'clear-2' }).state, 'clearing');
});

test('T4 sage-2828 sage-2830: a cancel after retirement ends the startup only, and a later thread-only send makes the successor pending as new input', (t) => {
  const { b, call, code, A, D, list, stages, clear, where } = setup(t);
  clear('human', { session: D, revision: 0, op: 'clear-1' });
  b.sessionClearExited({ ...where('clear-1'), run: null });
  b.sessionClearRetire(where('clear-1'));
  b.sessionClearStart({ ...where('clear-1'), start: 'held', refusal: 'capacity-full' });
  const S = successorOf(A, D, 'clear-1');
  const cancelled = call('human', 'session.cancel', { thread: A, op: 'clear-1', confirm: true });
  assert.deepEqual([cancelled.state, cancelled.retired, cancelled.start], ['cancelled', true, 'held']);
  assert.deepEqual(b.sessionClears('demo'), []);
  assert.equal(code(() => b.sessionClearStart({ ...where('clear-1'), start: 'started' })), 'session-clear-cancelled');
  assert.deepEqual(list().map((r) => [r.id, r.state, r.default]), [
    [D, 'retired', false],
    [S, 'active', true],
  ]);
  call('human', 'thread.send', { thread: A, body: 'new input' });
  assert.deepEqual(b.threadsState('demo').map((r) => [r.session, r.pending]), [[S, 1]]);
  assert.deepEqual(stages('clear-1'), ['requested', 'drained', 'exited', 'retired', 'start-held', 'cancelled']);
});

// N4: retirement rechecks the owner. Only a user's thread can fail it (a lost grant), and no verb
// clears a user's thread in 271: the R3 gate refuses all three clear verbs to a user, and #thread
// refuses them to anyone else. So the op's first two rows are written here as the accept writes
// them, and the launcher's ops run as they would. Row 264e carry (duckworth-0201): once a user
// clear verb lands, these tests take the verb path instead of the raw session_ops inserts.
const userClearExited = (t) => {
  const access = { doc: { accessVersion: 1, revision: 1, users: [], grants: [] } };
  const ctx = setup(t, { access });
  const { b, call, code, project, store, where } = ctx;
  access.doc.grants = [{ id: 'g-1', user: 'alice', business: 'demo', projects: [project], instances: ['assistant'], level: 'collaborator' }];
  const alice = b.bindUser({ business: 'demo', user: 'alice' });
  const U = b.request(alice, { verb: 'thread.create', args: { project, name: 'alice work' } }).id;
  const UD = store.get('SELECT id FROM sessions WHERE thread=?', U).id;
  assert.equal(code(() => b.request(alice, { verb: 'session.clear', args: { thread: U, session: UD, revision: 0, confirm: true, op: 'u1' } })), 'access-refused');
  assert.equal(code(() => call('human', 'session.clear', { thread: U, session: UD, revision: 0, confirm: true, op: 'u1' })), 'thread-not-owned');
  const S = successorOf(U, UD, 'u1');
  for (const stage of ['requested', 'drained'])
    store.run(
      "INSERT INTO session_ops(at,business,thread,session,op,stage,attempt,successor,successor_name) VALUES(?,'demo',?,?,'u1',?,0,?,'default-1')",
      new Date().toISOString(),
      U,
      UD,
      stage,
      S,
    );
  b.sessionClearExited({ ...where('u1', U), run: null });
  return { ...ctx, access, U, UD, S };
};

test('T4 N4: a user\'s thread whose grant is gone by retirement ends the clear failed, writes no successor, and leaves the session as it was', (t) => {
  const { b, code, stages, store, where, access, U, UD, S } = userClearExited(t);
  access.doc.grants = [];
  const failed = b.sessionClearRetire(where('u1', U));
  assert.deepEqual([failed.state, failed.refusal, failed.start], ['failed', 'access-refused', null]);
  assert.deepEqual(stages('u1', U), ['requested', 'drained', 'exited', 'failed']);
  assert.equal(store.get('SELECT 1 AS x FROM sessions WHERE id=?', S), undefined);
  assert.deepEqual(b.sessionClears('demo'), []);
  // The session is active again, with no successor; the default stays.
  assert.deepEqual(
    store.all('SELECT session FROM session_defaults WHERE thread=? ORDER BY seq', U).map((r) => r.session),
    [UD],
  );
  assert.equal(code(() => b.sessionClearStart({ ...where('u1', U), start: 'started' })), 'not-retired');
});

// duckworth-0201 B1: the recheck passes while the user's collaborator grant is live.
test('T4 N4: a user\'s thread whose grant is live at retirement retires the session and makes its successor', (t) => {
  const { b, stages, store, where, U, UD, S } = userClearExited(t);
  const retired = b.sessionClearRetire(where('u1', U));
  assert.deepEqual([retired.state, retired.refusal ?? null, retired.start, retired.successor], ['retired', null, null, S]);
  assert.deepEqual(stages('u1', U), ['requested', 'drained', 'exited', 'retired']);
  assert.equal(store.get('SELECT predecessor FROM sessions WHERE id=?', S).predecessor, UD);
  assert.deepEqual(
    store.all('SELECT session FROM session_defaults WHERE thread=? ORDER BY seq', U).map((r) => r.session),
    [UD, S],
  );
});

// This one covers the started successor.
test('T4 sage-2828: a cancel after the successor started refuses session-clear-complete', (t) => {
  const { b, call, code, A, D, clear, where } = setup(t);
  clear('human', { session: D, revision: 0, op: 'clear-1' });
  b.sessionClearExited({ ...where('clear-1'), run: null });
  b.sessionClearRetire(where('clear-1'));
  b.sessionClearStart({ ...where('clear-1'), start: 'started' });
  assert.equal(code(() => call('human', 'session.cancel', { thread: A, op: 'clear-1', confirm: true })), 'session-clear-complete');
});

test('T4 B2: a run bound to a clearing session refuses its claim, and claims again once the clear is cancelled', (t) => {
  const { call, code, launch, A, D, clear } = setup(t);
  clear('human', { session: D, revision: 0, op: 'clear-1' });
  assert.equal(code(() => launch('a1', A)), 'session-clearing');
  assert.equal(code(() => launch('a2', A, D)), 'session-clearing');
  call('human', 'session.cancel', { thread: A, op: 'clear-1', confirm: true });
  assert.deepEqual(launch('a3', A), { role: 'assistant', run: 'a3-run', thread: A });
});

test('T4 mos-1325: the tester clears its own thread; its clear, cancel and operation on a human-owned thread refuse thread-not-owned and write nothing', (t) => {
  const { b, call, code, project, A, D, count } = setup(t);
  const T = call('tester', 'thread.create', { project, name: 'tester work' }).id;
  const [{ id: TD }] = call('tester', 'session.list', { thread: T });
  const own = call('tester', 'session.clear', { thread: T, session: TD, revision: 0, confirm: true, op: 'tclear' });
  assert.equal(own.state, 'clearing');
  assert.deepEqual(call('tester', 'session.operation', { thread: T, op: 'tclear' }), own);
  call('human', 'thread.send', { thread: A, body: 'human waiting' });
  call('human', 'session.clear', { thread: A, session: D, revision: 0, confirm: true, op: 'hclear' });
  const before = [count('session_ops'), count('message_cancels')];
  assert.equal(code(() => call('tester', 'session.clear', { thread: A, session: D, revision: 2, confirm: true, op: 'tx' })), 'thread-not-owned');
  assert.equal(code(() => call('tester', 'session.cancel', { thread: A, op: 'hclear', confirm: true })), 'thread-not-owned');
  assert.equal(code(() => call('tester', 'session.operation', { thread: A, op: 'hclear' })), 'thread-not-owned');
  assert.deepEqual([count('session_ops'), count('message_cancels')], before);
  // And the human can't reach the tester's.
  assert.equal(code(() => call('human', 'session.cancel', { thread: T, op: 'tclear', confirm: true })), 'thread-not-owned');
  // An agent refuses before any check.
  assert.equal(code(() => call('chief', 'session.clear', { thread: A, session: D, revision: 2, confirm: true, op: 'ax' })), 'human-required');
  assert.equal(code(() => call('chief', 'session.operation', { thread: A, op: 'hclear' })), 'human-required');
  assert.deepEqual(b.sessionClears('demo').map((c) => [c.thread, c.op]), [
    [T, 'tclear'],
    [A, 'hclear'],
  ]);
});

test('T4 schema: session_ops and message_cancels are append-only, and a second open clear of a session is refused below the broker', (t) => {
  const { call, A, D, clear, store } = setup(t);
  const m = call('human', 'thread.send', { thread: A, body: 'w' });
  clear('human', { session: D, revision: 0, op: 'clear-1' });
  assert.throws(() => store.run("UPDATE session_ops SET stage='retired'"), /session_ops is append-only/);
  assert.throws(() => store.run('DELETE FROM session_ops'), /session_ops is append-only/);
  assert.throws(() => store.run("UPDATE message_cancels SET op='x'"), /message_cancels is append-only/);
  assert.throws(() => store.run('DELETE FROM message_cancels'), /message_cancels is append-only/);
  const raw = (op, stage, attempt = 0) =>
    store.run(
      "INSERT INTO session_ops(at,business,thread,session,op,stage,attempt,successor,successor_name) VALUES(?,'demo',?,?,?,?,?,'s','n')",
      new Date().toISOString(),
      A,
      D,
      op,
      stage,
      attempt,
    );
  assert.throws(() => raw('clear-2', 'requested'), /session has an open clear/);
  assert.throws(() => raw('clear-1', 'start-held', 0), /CHECK constraint failed/);
  assert.throws(() => raw('clear-1', 'retired', 1), /CHECK constraint failed/);
  assert.throws(() => raw('clear-1', 'bogus'), /CHECK constraint failed/);
  assert.throws(() => raw('clear-1', 'drained'), /UNIQUE constraint failed/);
  assert.throws(
    () => store.run("INSERT INTO message_cancels(at,message,session,op) VALUES(?,?,?,'again')", new Date().toISOString(), m.id, D),
    /UNIQUE constraint failed/,
  );
});

// The 3m→3n step (sage-2968: row 206 holds 3l→3m). A 3m store from fixtures/schema-3m.sql, which is
// refactor's schema.sql at 0cbe2e9b3, with one row in each thread table and a share.
const AT = '2026-10-22T00:00:00.000Z';
const here = (p) => new URL(p, import.meta.url);
const ddl3m = readFileSync(here('./fixtures/schema-3m.sql'), 'utf8');
function source3m(t) {
  const root = mkdtempSync(join(tmpdir(), 'session-clear-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(ddl3m).digest('hex'), 'd5ca9bb9268994b175ac79d38939be4bb7f2e80bc0d9ea273eedaae50c29b8a9');
  db.exec(ddl3m);
  const run = (sql, ...args) => db.prepare(sql).run(...args);
  run('INSERT INTO meta VALUES (?,?)', 'schema_version', '3m');
  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, owner] of [['t1', 'jason'], ['t2', 'user:alice']]) {
    run('INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES(?,?,?,?,?,?,?)', id, AT, 'demo', 'p-1', 'assistant', owner, id);
    run("INSERT INTO sessions(at,business,thread,id,instance,owner,name,op) VALUES(?,?,?,?,?,?,'default','legacy-3l')", AT, 'demo', id, id, 'assistant', owner);
    run("INSERT INTO session_defaults(at,thread,session,by) VALUES(?,?,?,'migration-3l')", AT, id, id);
  }
  run('INSERT INTO messages(id,at,business,from_role,from_run,to_role,class,body) VALUES(?,?,?,?,?,?,?,?)', 'm-1', AT, 'demo', 'user', 'user:alice', 'assistant', 'INFO', 'm-1');
  run('INSERT INTO message_threads VALUES(?,?,?)', 'm-1', 't2', 't2');
  run('INSERT INTO thread_opens(seq,at,thread,session,by) VALUES(?,?,?,?,?)', 7, AT, 't2', 't2', 'user:alice');
  run('INSERT INTO thread_shares(at,business,thread,op,target,by) VALUES(?,?,?,?,?,?)', AT, 'demo', 't2', 'share', 'jason', 'user:alice');
  db.close();
  chmodSync(path, 0o600);
  return { root, path };
}
const open = (path, fn) => {
  const db = new DatabaseSync(path, { readOnly: true });
  try { return fn(db); } finally { db.close(); }
};
const master = (path) => open(path, (db) => db.prepare('SELECT type,name,tbl_name,sql FROM sqlite_master ORDER BY type,name').all().map((r) => ({ ...r })));
// Every table's rows but meta's, whose version and digest the step changes.
const tables = (path) => open(path, (db) => Object.fromEntries(
  db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name NOT IN ('meta') ORDER BY name").all()
    .map(({ name }) => [name, db.prepare(`SELECT * FROM "${name}" ORDER BY rowid`).all().map((r) => ({ ...r }))]),
));
const version = (path) => open(path, (db) => db.prepare("SELECT value FROM meta WHERE key='schema_version'").get().value);

test('T4 schema: a 3m store migrates to 3n, session_ops and message_cancels are new and empty, and every other row stays as it was', (t) => {
  const { root, path } = source3m(t);
  const before = tables(path);
  assert.throws(() => new Store(root), /schema-mismatch/);
  new Store(root, { migrateSchema: true }).close();
  assert.equal(version(path), '3v');
  const after = tables(path);
  for (const name of ['session_ops', 'message_cancels', 'request_events', 'request_effects', 'request_snapshots', 'request_snapshot_events', 'request_project_events', 'project_archives', 'project_system_types', 'store_tasks']) {
    assert.deepEqual(after[name], [], name);
    delete after[name];
  }
  assert.deepEqual(after, before);
  // The migrated schema is the fresh one, object for object.
  const freshRoot = mkdtempSync(join(tmpdir(), 'session-clear-fresh-'));
  t.after(() => rmSync(freshRoot, { recursive: true, force: true }));
  new Store(freshRoot).close();
  assert.deepEqual(master(path), master(join(freshRoot, 'bus/bus.sqlite')));
  // The new tables are append-only on the migrated store, and a session holds one open clear.
  const store = new Store(root);
  t.after(() => store.close());
  const op = (stage, over = {}) => ['INSERT INTO session_ops(at,business,thread,session,op,stage,attempt,successor,successor_name) VALUES(?,?,?,?,?,?,?,?,?)',
    AT, 'demo', 't2', 't2', over.op ?? 'c-1', stage, 0, over.successor ?? 's-1', 'default-1'];
  const [insert, ...args] = op('requested');
  store.run(insert, ...args);
  const [again, ...againArgs] = op('requested', { op: 'c-2', successor: 's-2' });
  assert.throws(() => store.run(again, ...againArgs), /session has an open clear/);
  store.run('INSERT INTO message_cancels(at,message,session,op) VALUES(?,?,?,?)', AT, 'm-1', 't2', 'c-1');
  for (const [table, sql, sqlArgs] of [
    ['session_ops', "UPDATE session_ops SET stage='retired'", []],
    ['session_ops', 'DELETE FROM session_ops', []],
    ['session_ops', 'INSERT OR REPLACE INTO session_ops(seq,at,business,thread,session,op,stage,attempt,successor,successor_name) VALUES(?,?,?,?,?,?,?,?,?,?)', [1, AT, 'demo', 't2', 't2', 'c-1', 'drained', 0, 's-1', 'default-1']],
    ['message_cancels', "UPDATE message_cancels SET op='c-9'", []],
    ['message_cancels', 'DELETE FROM message_cancels', []],
    ['message_cancels', 'INSERT OR REPLACE INTO message_cancels(seq,at,message,session,op) VALUES(?,?,?,?,?)', [1, AT, 'm-1', 't2', 'c-9']],
  ]) assert.throws(() => store.run(sql, ...sqlArgs), new RegExp(`${table} is append-only`), sql);
  assert.throws(() => store.run('INSERT INTO message_cancels(at,message,session,op) VALUES(?,?,?,?)', AT, 'no-such', 't2', 'c-1'), /FOREIGN KEY/);
  assert.deepEqual(store.all('SELECT seq,thread,session,op,stage FROM session_ops').map((r) => ({ ...r })), [{ seq: 1, thread: 't2', session: 't2', op: 'c-1', stage: 'requested' }]);
});

// The 3m engine is store.mjs at this change with the 3m schema and version and no 3m→3n step.
async function engine3m(root) {
  const moduleRoot = join(root, 'engine-3m');
  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(ddl3m).digest('hex')}';`)
    .replace("export const SCHEMA_VERSION = '3v';", "export const SCHEMA_VERSION = '3m';")
    .replace(/\}, \{\n {2}\/\/ Row 271 \(#1776\)[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables');
  assert.ok(edited.includes("export const SCHEMA_VERSION = '3m';") && edited.includes('3l-3m') && !edited.includes('3m-3n'), 'the 3m engine has no 3m→3n step');
  writeFileSync(join(moduleRoot, 'src/store.mjs'), edited);
  writeFileSync(join(moduleRoot, 'schema.sql'), ddl3m);
  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', '3k-3l.sql', '3l-3m.sql'])
    copyFileSync(here('../src/migrations/' + file), join(moduleRoot, 'src/migrations', file));
  return (await import(pathToFileURL(join(moduleRoot, 'src/store.mjs')))).Store;
}

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