248 lines
13 KiB
JavaScript
248 lines
13 KiB
JavaScript
import test from 'node:test';
|
|
import assert from 'node:assert/strict';
|
|
import { randomUUID } from 'node:crypto';
|
|
import { mkdtempSync, readFileSync, rmSync } from 'node:fs';
|
|
import { tmpdir } from 'node:os';
|
|
import { join } from 'node:path';
|
|
import { Store } from '../src/store.mjs';
|
|
import { Broker, THREAD_PAGE, THREAD_PAGE_MAX, TESTER_VERBS } from '../src/broker.mjs';
|
|
|
|
const MEMBERS = ['bean', 'chet', 'continuity'];
|
|
const shipped = (name) => JSON.parse(readFileSync(new URL(`../../../roles/${name}.json`, import.meta.url), 'utf8'));
|
|
function setup(t, { deny = false, empty = false } = {}) {
|
|
const root = mkdtempSync(join(tmpdir(), 'bus-rooms-'));
|
|
const store = new Store(root);
|
|
t.after(() => { store.close(); rmSync(root, { recursive: true, force: true }); });
|
|
const role = (definition) => ({ definition, authority: { withinRole: ['message.send'], crossRole: [] } });
|
|
// Deliberately unsorted, with two instances of one definition and a non-member chief.
|
|
const roles = { continuity: role('continuity'), chet: role('companion'), bean: role('companion'), chief: role('chief-of-staff') };
|
|
const definitions = {
|
|
'chief-of-staff': {}, companion: { messaging: shipped('companion').messaging },
|
|
continuity: { messaging: deny ? { sendTo: ['human'], receiveFrom: [] } : shipped('continuity').messaging },
|
|
};
|
|
const b = new Broker({ store, businesses: { demo: {
|
|
id: 'demo', human: 'jason', arbiters: { technical: 'chief', delivery: 'chief' },
|
|
roles: empty ? { chief: roles.chief } : roles, definitions,
|
|
projects: { stack: { root: '/nonexistent' } }, vars: { 'bus.tester': true },
|
|
} } });
|
|
const caps = {
|
|
human: b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true }),
|
|
tester: b.bindTester({ business: 'demo' }),
|
|
};
|
|
const call = (verb, args = {}, who = 'human') => b.request(caps[who], { verb, args });
|
|
caps.chief = b.bindLaunch({ business: 'demo', role: 'chief', run: 'chief-run', harness: 'pi' });
|
|
call('role.claim', {}, 'chief');
|
|
const project = call('project.create', { name: 'room', repo: 'stack' }).id;
|
|
const open = () => call('room.open', { project });
|
|
const send = (body, extra = {}) => call('room.send', { project, body, ...extra });
|
|
const read = (extra = {}) => call('room.read', { project, ...extra });
|
|
const effects = () => ['messages', 'message_threads', 'deliveries'].map((table) => store.get(`SELECT count(*) n FROM ${table}`).n)
|
|
.concat(store.get("SELECT count(*) n FROM events WHERE kind='human.input'").n);
|
|
const refuse = (code, fn) => {
|
|
const before = effects();
|
|
assert.throws(fn, { code });
|
|
assert.deepEqual(effects(), before, `${code}: no message, route, delivery or human.input`);
|
|
};
|
|
return { b, caps, store, call, project, open, send, read, effects, refuse };
|
|
}
|
|
|
|
test('room.open sorts all threaded instances, creates defaults, and reuses caller-owned threads', (t) => {
|
|
const { open, call, project, store } = setup(t);
|
|
const existing = call('thread.create', { project, name: 'chet', instance: 'chet' }).id;
|
|
const result = open();
|
|
assert.equal(result.project, project);
|
|
assert.deepEqual(result.threads.map(({ instance, name }) => [instance, name]), MEMBERS.map((x) => [x, x]));
|
|
assert.equal(result.threads[1].id, existing);
|
|
for (const thread of result.threads) {
|
|
assert.equal(store.get('SELECT owner FROM threads WHERE id=?', thread.id).owner, 'jason');
|
|
assert.equal(call('session.list', { thread: thread.id }).filter((s) => s.default).length, 1);
|
|
}
|
|
assert.deepEqual(open(), result);
|
|
assert.equal(store.get('SELECT count(*) n FROM threads').n, 3);
|
|
});
|
|
|
|
for (const collision of ['instance', 'owner']) {
|
|
test(`room.open preflights ${collision} collisions atomically`, (t) => {
|
|
const { call, project, open, store } = setup(t);
|
|
call('thread.create', { project, name: 'continuity', instance: collision === 'instance' ? 'bean' : 'continuity' },
|
|
collision === 'owner' ? 'tester' : 'human');
|
|
assert.throws(open, { code: 'room-thread-name-taken' });
|
|
assert.deepEqual(store.all('SELECT name FROM threads').map((r) => r.name), ['continuity']);
|
|
assert.equal(store.get('SELECT count(*) n FROM sessions').n, 1);
|
|
});
|
|
}
|
|
|
|
test('room verbs require the human, and none is a tester verb', (t) => {
|
|
const { call, project, open, refuse } = setup(t);
|
|
open();
|
|
for (const verb of ['room.open', 'room.send', 'room.read']) {
|
|
const args = { project, ...(verb === 'room.send' ? { body: 'hello' } : {}) };
|
|
assert.ok(!TESTER_VERBS.includes(verb));
|
|
refuse('tester-refused', () => call(verb, args, 'tester'));
|
|
refuse('human-required', () => call(verb, args, 'chief'));
|
|
}
|
|
});
|
|
|
|
test('room verbs refuse missing projects and zero member businesses', (t) => {
|
|
const { call, refuse } = setup(t);
|
|
for (const verb of ['room.open', 'room.send', 'room.read']) {
|
|
refuse('project-not-found', () => call(verb, { project: randomUUID(), ...(verb === 'room.send' ? { body: 'hello' } : {}) }));
|
|
}
|
|
const empty = setup(t, { empty: true });
|
|
assert.throws(empty.open, { code: 'thread-instance-missing' });
|
|
});
|
|
|
|
for (const state of ['missing', 'foreign-owner', 'wrong-instance']) {
|
|
test(`send/read refuse a ${state} canonical member without effects`, (t) => {
|
|
const { call, project, send, read, refuse } = setup(t);
|
|
for (const instance of MEMBERS) {
|
|
if (instance === 'continuity' && state === 'missing') continue;
|
|
call('thread.create', { project, name: instance, instance: instance === 'continuity' && state === 'wrong-instance' ? 'bean' : instance },
|
|
instance === 'continuity' && state === 'foreign-owner' ? 'tester' : 'human');
|
|
}
|
|
refuse('room-not-open', () => send('hello'));
|
|
refuse('room-not-open', read);
|
|
});
|
|
}
|
|
|
|
test('room.send targets a key, broadcasts without tags, and preserves original text and ordinary route/input rows', (t) => {
|
|
const { open, send, store, read } = setup(t);
|
|
const threads = open().threads;
|
|
const one = send('@continuity x');
|
|
assert.deepEqual(one.to, ['continuity']);
|
|
assert.equal(one.ids.length, 1);
|
|
assert.equal(store.get('SELECT thread FROM message_threads WHERE message=?', one.ids[0]).thread, threads[2].id);
|
|
const body = ' \t@bean @chet hello @continuity';
|
|
const two = send(body);
|
|
assert.deepEqual(two.to, ['bean', 'chet']);
|
|
const all = send('hello @bean');
|
|
assert.deepEqual(all.to, MEMBERS);
|
|
const rows = store.all('SELECT * FROM messages ORDER BY seq');
|
|
assert.deepEqual(rows.map((r) => r.body), ['@continuity x', body, body, ...MEMBERS.map(() => 'hello @bean')]);
|
|
assert.equal(new Set(rows.slice(-3).map((r) => r.at)).size, 1);
|
|
assert.ok(rows.every((r) => r.from_role === 'human' && r.from_run === 'human-cli' && r.class === 'INFO'));
|
|
assert.equal(store.get('SELECT count(*) n FROM message_threads').n, 6);
|
|
assert.equal(store.get("SELECT count(*) n FROM deliveries WHERE op='routed'").n, 6);
|
|
assert.equal(store.get("SELECT count(*) n FROM events WHERE kind='human.input'").n, 6);
|
|
assert.deepEqual(read().map((r) => r.to), [['continuity'], ['bean', 'chet'], MEMBERS]);
|
|
});
|
|
|
|
test('current display names and repeated tags resolve once; a replaced display name is unknown', (t) => {
|
|
const { open, call, send, refuse } = setup(t);
|
|
open();
|
|
call('name.set', { instance: 'bean', name: 'beanie' });
|
|
assert.deepEqual(send('@beanie @bean @beanie hi').to, ['bean']);
|
|
call('name.set', { instance: 'bean', name: 'newbeanie' });
|
|
refuse('unknown-addressee', () => send('@beanie hi'));
|
|
assert.deepEqual(send('@newbeanie hi').to, ['bean']);
|
|
});
|
|
|
|
test('a legacy name colliding with another member key is ambiguous; its own key resolves once', (t) => {
|
|
const { open, send, store, refuse } = setup(t);
|
|
open();
|
|
// Naming policy now forbids key collisions; legacy data can still contain them. Append real rows.
|
|
store.run('INSERT INTO instance_names(at,business,instance,name,by) VALUES(?,?,?,?,?)', new Date().toISOString(), 'demo', 'bean', 'chet', 'fixture');
|
|
refuse('addressee-ambiguous', () => send('@chet hi'));
|
|
store.run('INSERT INTO instance_names(at,business,instance,name,by) VALUES(?,?,?,?,?)', new Date().toISOString(), 'demo', 'bean', 'bean', 'fixture');
|
|
assert.deepEqual(send('@bean @bean hi').to, ['bean']);
|
|
});
|
|
|
|
test('unknown or non-member addressees and tags-only bodies leave no message effects', (t) => {
|
|
const { open, send, refuse } = setup(t);
|
|
open();
|
|
for (const body of ['@nobody x', '@chief x', '@bean @nobody x']) refuse('unknown-addressee', () => send(body));
|
|
for (const body of ['@continuity', ' \t@bean\n@chet ', '', ' \t']) refuse('invalid-request', () => send(body));
|
|
});
|
|
|
|
test('messaging denial on the last target rolls back the entire broadcast; allowed subset still works', (t) => {
|
|
const { open, send, refuse } = setup(t, { deny: true });
|
|
open();
|
|
refuse('message-sender-refused', () => send('all'));
|
|
assert.deepEqual(send('@bean @chet okay').to, ['bean', 'chet']);
|
|
});
|
|
|
|
test('a clearing last member fences the whole send; another active target still accepts', (t) => {
|
|
const { b, caps, open, call, send, refuse } = setup(t);
|
|
const thread = open().threads[2].id;
|
|
caps.worker = b.bindLaunch({ business: 'demo', role: 'continuity', run: 'continuity-run', harness: 'pi', thread });
|
|
call('role.claim', {}, 'worker');
|
|
const session = call('session.list', { thread })[0].id;
|
|
call('session.clear', { thread, session, revision: 0, op: 'room-clear', confirm: true });
|
|
refuse('session-clearing', () => send('all'));
|
|
assert.deepEqual(send('@bean okay').to, ['bean']);
|
|
});
|
|
|
|
test('room.read groups one send, pages by the last grouped id, and rejects foreign or absent cursors', (t) => {
|
|
const { open, send, read, call, project, refuse } = setup(t);
|
|
const threads = open().threads;
|
|
const first = send('first');
|
|
const second = send('@continuity second');
|
|
const page = read({ limit: 1 });
|
|
assert.equal(page.length, 1);
|
|
assert.deepEqual(page[0].ids, first.ids);
|
|
assert.deepEqual(page[0].to, MEMBERS);
|
|
assert.equal(page[0].thread, threads[0].id);
|
|
assert.equal(page[0].instance, 'bean');
|
|
assert.deepEqual(read({ after: first.ids.at(-1), limit: 1 }).map((r) => r.ids), [second.ids]);
|
|
assert.deepEqual(read({ after: second.ids[0] }), []);
|
|
const other = call('project.create', { name: 'other', repo: 'stack' }).id;
|
|
const thread = call('thread.create', { project: other, name: 'bean', instance: 'bean' }).id;
|
|
const foreign = call('thread.send', { thread, body: 'elsewhere' }).id;
|
|
const noncanonical = call('thread.create', { project, name: 'side', instance: 'bean' }).id;
|
|
const side = call('thread.send', { thread: noncanonical, body: 'side' }).id;
|
|
for (const after of [foreign, side, randomUUID()]) refuse('message-not-found', () => read({ after }));
|
|
for (const limit of [0, -1, 1.1, THREAD_PAGE_MAX + 1, '1']) refuse('invalid-request', () => read({ limit }));
|
|
});
|
|
|
|
test('room.read keeps identical sends at the same timestamp separate and includes every session', (t) => {
|
|
const { open, call, project, store, read } = setup(t);
|
|
const threads = open().threads;
|
|
const at = '2026-10-24T00:00:00.000Z';
|
|
const batches = [[], []];
|
|
// Broker clocks are monotonic. Append two independently valid transaction-shaped batches to
|
|
// exercise the historical same-at boundary without altering production clocks or existing rows.
|
|
for (const batch of batches) for (const thread of threads) {
|
|
const id = randomUUID();
|
|
batch.push(id);
|
|
const session = call('session.list', { thread: thread.id })[0].id;
|
|
store.run('INSERT INTO messages(id,at,business,from_role,from_run,to_role,class,body) VALUES(?,?,?,?,?,?,?,?)',
|
|
id, at, 'demo', 'human', 'human-cli', thread.instance, 'INFO', 'same');
|
|
store.run('INSERT INTO message_threads(message,thread,session) VALUES(?,?,?)', id, thread.id, session);
|
|
}
|
|
assert.deepEqual(read().map((r) => r.ids), batches);
|
|
assert.deepEqual(read({ limit: 1 })[0].ids, batches[0]);
|
|
assert.deepEqual(read({ after: batches[0].at(-1) }).map((r) => r.ids), [batches[1]]);
|
|
const created = call('session.create', { thread: threads[0].id, name: 'later', op: 'new-later' });
|
|
call('thread.send', { thread: threads[0].id, session: created.id, body: 'later session' });
|
|
assert.equal(read().at(-1).body, 'later session');
|
|
assert.equal(read().at(-1).session, created.id);
|
|
assert.equal(read().at(-1).thread, threads[0].id);
|
|
assert.equal(store.get('SELECT count(*) n FROM threads WHERE project=?', project).n, 3);
|
|
});
|
|
|
|
test('room.read defaults and caps the number of merged entries, not recipient rows', (t) => {
|
|
const { open, send, read } = setup(t);
|
|
open();
|
|
for (let i = 0; i <= THREAD_PAGE_MAX; i++) send(`message ${i}`);
|
|
assert.equal(read().length, THREAD_PAGE);
|
|
assert.equal(read({ limit: THREAD_PAGE_MAX }).length, THREAD_PAGE_MAX);
|
|
assert.ok(read({ limit: 1 })[0].to.length === 3);
|
|
});
|
|
|
|
test('room.read includes an agent reply with its thread and instance without merging it with human rows', (t) => {
|
|
const { b, caps, open, call, send, read, refuse, project } = setup(t);
|
|
const thread = open().threads[0].id;
|
|
send('question');
|
|
caps.worker = b.bindLaunch({ business: 'demo', role: 'bean', run: 'bean-run', harness: 'pi', thread });
|
|
call('role.claim', {}, 'worker');
|
|
const reply = call('message.send', { to: 'human', body: 'answer' }, 'worker');
|
|
const entries = read();
|
|
assert.equal(entries.length, 2);
|
|
assert.deepEqual(entries[0].to, MEMBERS);
|
|
assert.equal(entries[1].id, reply.id);
|
|
assert.equal(entries[1].from_role, 'bean');
|
|
assert.equal(entries[1].thread, thread);
|
|
assert.equal(entries[1].instance, 'bean');
|
|
refuse('human-required', () => call('room.read', { project }, 'worker'));
|
|
});
|