jason.woltje
9b7bf11afb
PRD 1.20 REQ-TASK-1, REQ-PLUG-1, REQ-WS-8. Each organization keeps its tasks in its own bus through the new task-tracker plug-in kind. The default module is the built-in store; Vikunja is the optional module and a Vikunja organization keeps today's path. - bus schema 3u to 3v (migration 3u-3v.sql): store_tasks, append-only, one row per write, each citing its events row; seven guard triggers. - broker: the eight role verbs on the store with the slice 1 field table checked twice (module, then a before/after field diff); the person's tasks.create/edit/archive/unarchive/delete as human.input events; the tasks read returns store rows in board order (group, position), filtered and paged like tracker rows. - runtime: each business gets its module at start (organization file plugin, else vikunja with a tracker entry, else store); a store business with a Vikunja tracker refuses the boot (tracker-config). - plugins: task-tracker kind with store and vikunja modules. - cli: the launcher's tracker flag covers store organizations; trail prints store writes. - pins: schema.test counts, the bus migration tests, stack-release #811 and T46 (store_tasks on the always list; COUNTED unchanged), budget-supervisor accepts 3v. Rebased onto SYS1 by the lead: SYS1 took 3t-3u, so this step is 3u-3v.
598 lines
38 KiB
JavaScript
598 lines
38 KiB
JavaScript
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();
|
|
});
|