stack/packages/bus/tests/launch-notice.test.mjs

import test from 'node:test';
import assert from 'node:assert/strict';
import { mkdtempSync, readFileSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { fileURLToPath } from 'node:url';
import { Store } from '../src/store.mjs';
import { Broker } from '../src/broker.mjs';
import { createContinuations } from '../src/request-continuation.mjs';
import { ID_PATTERN } from '../../organization/src/vocabulary.mjs';

// Row 146 (#1640): a message.send to a role nobody holds tells the business's launch.by
// instance once, and the result says so. The assistant's lists are the shipped ones, as in
// threads.test.mjs, so the thread case runs under the lists the stack runs.
const shipped = (name) =>
  JSON.parse(readFileSync(fileURLToPath(new URL(`../../../roles/${name}.json`, import.meta.url)), 'utf8'));

const roles = {
  chief: { definition: 'chief-of-staff', authority: { withinRole: ['message.send'], crossRole: [] } },
  assistant: { definition: 'assistant', authority: { withinRole: ['message.send'], crossRole: [] } },
  pm: { definition: 'pm', authority: { withinRole: ['message.send'], crossRole: [] } },
  coder: { definition: 'coder', authority: { withinRole: ['message.send'], crossRole: [] } },
};
const definitions = { 'chief-of-staff': {}, assistant: { messaging: shipped('assistant').messaging }, pm: {}, coder: {} };
const demo = (extra = {}) => ({
  demo: {
    id: 'demo',
    human: 'jason',
    arbiters: { technical: 'pm', delivery: 'chief' },
    roles,
    definitions,
    projects: { stack: { root: '/nonexistent' } },
    launch: { by: 'chief', instances: ['pm', 'coder'], models: {}, limits: {} },
    vars: { 'bus.tester': true },
    ...extra,
  },
});

test('continuation notices ignore unrelated lifecycle events, halts, and unclaimed launches', async t => {
  const root = mkdtempSync(join(tmpdir(), 'continuation-notice-'));
  const store = new Store(root);
  t.after(() => { store.close(); rmSync(root, { recursive: true, force: true }); });
  const extraRoles = { ...roles, reviewer: { definition: 'reviewer', authority: { withinRole: ['message.send'], crossRole: [] } }, helper: { definition: 'coder', authority: { withinRole: ['message.send'], crossRole: [] } } };
  const businesses = demo({ roles: extraRoles, definitions: { ...definitions, reviewer: {} }, launch: { by: 'chief', instances: ['pm', 'coder', 'reviewer'], models: {}, limits: {} } });
  const states = ['pm', 'coder', 'reviewer'].map((role, i) => ({ business: 'demo', request: `request-${i}`, revision: 1, phase: 'selection', pm: role, source_message: `source-${i}`, tasks: [] }));
  const rows = [];
  const journal = {
    effects: (business, request) => rows.filter(r => r.business === business && r.request === request),
    transaction: fn => fn(),
    prepareEffect: e => { rows.push(structuredClone(e)); return e; },
    finishEffect: e => { rows.push(structuredClone(e)); return e; },
  };
  let halted = false, continuations;
  const processing = { read: () => ({ version: 1, generation: '00000000-0000-4000-8000-000000000000', halted }) };
  const b = new Broker({ store, businesses, processing, requestLifecycle: services => {
    continuations = createContinuations({ journal, services });
    const noop = () => {};
    return { open: noop, get: noop, projects: noop, propose: noop, question: noop, answer: noop, attach: noop, cancel: noop, modeConsole: noop, draftConsole: noop, resume: noop, recover: noop };
  } });
  const resume = () => { for (const state of states) { try { continuations.resume(state); } catch (e) { if (e.code !== 'processing-halted') throw e; } } };
  const notices = id => store.get("SELECT count(*) n FROM messages WHERE from_role='_bus' AND to_role='chief' AND in_reply_to=?", id).n;
  const launch = (role, run) => b.bindLaunch({ business: 'demo', role, run, harness: 'pi' });
  launch('helper', 'helper-old');
  resume();
  const waiting = store.all("SELECT id,to_role FROM messages WHERE from_role='_bus' AND body LIKE 'Request continuation %' ORDER BY seq");
  assert.equal(waiting.length, 3);
  for (const m of waiting) assert.equal(notices(m.id), 1);

  halted = true;
  b.endLaunch({ business: 'demo', run: 'helper-old', reason: 'exit' });
  resume();
  for (const m of waiting) assert.equal(notices(m.id), 1, 'halt sends no notices');
  halted = false;
  launch('helper', 'helper-new');
  resume();
  for (const m of waiting) assert.equal(notices(m.id), 1, 'unrelated launch sends no notices');

  const pm = launch('pm', 'pm-new');
  resume();
  for (const m of waiting) assert.equal(notices(m.id), 1, 'unclaimed target launch sends no notices');
  b.request(pm, { verb: 'role.claim', args: {} });
  resume();
  const pmMessage = waiting.find(m => m.to_role === 'pm');
  assert.equal(store.get("SELECT holder_run FROM deliveries WHERE message=? AND op='routed' ORDER BY seq DESC LIMIT 1", pmMessage.id).holder_run, 'pm-new');
  launch('coder', 'coder-new');
  resume();
  for (const m of waiting) assert.equal(notices(m.id), 1, 'staggered launches do not duplicate notices');
  states.push({ business: 'demo', request: 'request-starting', revision: 1, phase: 'selection', pm: 'coder', source_message: 'source-starting', tasks: [] });
  resume();
  const starting = store.get("SELECT id FROM messages WHERE to_role='coder' AND body LIKE '%request-starting%'");
  assert.equal(notices(starting.id), 0, 'a new continuation does not announce an unclaimed launch');
  halted = true;
  b.endLaunch({ business: 'demo', run: 'coder-new', reason: 'exit' });
  resume();
  assert.equal(notices(starting.id), 0, 'target end while halted sends no notice');
  halted = false;
  resume();
  assert.equal(notices(starting.id), 1, 'target end is noticed after processing resumes');
  const replacement = launch('coder', 'coder-replacement');
  b.request(replacement, { verb: 'role.claim', args: {} });
  resume();
  assert.equal(store.get("SELECT holder_run FROM deliveries WHERE message=? AND op='routed' ORDER BY seq DESC LIMIT 1", starting.id).holder_run, 'coder-replacement');
});

function setup(t, businesses = demo()) {
  const root = mkdtempSync(join(tmpdir(), 'bus-notice-'));
  let store = new Store(root);
  t.after(() => {
    store.close();
    rmSync(root, { recursive: true, force: true });
  });
  const ctx = { root };
  const open = () => {
    ctx.b = new Broker({ store, businesses });
    ctx.caps = { human: ctx.b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true }) };
  };
  open();
  ctx.store = () => store;
  // A broker restart: close the store and open it again under a new Broker.
  ctx.restart = () => {
    store.close();
    store = new Store(root);
    open();
  };
  ctx.call = (who, verb, args = {}) => ctx.b.request(ctx.caps[who], { verb, args });
  ctx.launch = (name, role, extra = {}) => {
    ctx.caps[name] = ctx.b.bindLaunch({ business: 'demo', role, run: `${name}-run`, harness: 'pi', ...extra });
    return ctx.caps[name];
  };
  ctx.claim = (name) => ctx.call(name, 'role.claim');
  ctx.notices = () => store.all("SELECT * FROM messages WHERE from_role='_bus' ORDER BY seq");
  return ctx;
}

test('a message to a stopped role sends launch.by one notice, and the result says so', (t) => {
  const ctx = setup(t);
  const sent = ctx.call('human', 'message.send', { to: 'pm', body: 'is the plan current?', class: 'REQUEST' });
  assert.deepEqual(sent.waiting, { role: 'pm', launcher: 'chief', told: true, launcherRunning: false, launchable: true });
  const [n, ...more] = ctx.notices();
  assert.deepEqual(more, []);
  assert.equal(n.from_run, 'bus');
  assert.equal(n.to_role, 'chief');
  assert.equal(n.class, 'RESULT');
  assert.equal(n.in_reply_to, sent.id);
  assert.equal(n.decision, null);
  assert.equal(
    n.body,
    `Message ${sent.id} from human to pm waits: pm has no running session. Launch pm within the launch block's limits, or tell human why not with message.send in_reply_to ${sent.id}.`,
  );
  assert.equal(ctx.store().get("SELECT holder_run FROM deliveries WHERE message=? AND op='routed'", n.id).holder_run, null);
  // chief comes up and receives it; with chief running, the next result says so.
  ctx.launch('chief', 'chief');
  ctx.claim('chief');
  assert.deepEqual(ctx.call('chief', 'message.receive').map((m) => m.id), [n.id]);
  assert.deepEqual(ctx.call('human', 'message.send', { to: 'coder', body: 'and you?' }).waiting, {
    role: 'coder',
    launcher: 'chief',
    told: true,
    launcherRunning: true,
    launchable: true,
  });
  assert.equal(ctx.notices().length, 2);
  // Row 151 (#1645): the notice is routed to launch.by's holder, which the Console's waiting lists read.
  // message.receive selects on to_role, so only this row shows a notice routed anywhere else.
  assert.equal(
    ctx.store().get("SELECT holder_run FROM deliveries WHERE message=? AND op='routed'", ctx.notices().at(-1).id).holder_run,
    'chief-run',
  );
  // A role's message to a stopped role tells launch.by too.
  ctx.launch('coder', 'coder');
  ctx.claim('coder');
  assert.equal(ctx.call('coder', 'message.send', { to: 'pm', body: 'blocked on you' }).waiting.told, true);
  assert.equal(ctx.notices().at(-1).body.startsWith('Message '), true);
  assert.match(ctx.notices().at(-1).body, / from coder to pm waits: /);
});

// Row 152 (#1646): a target outside launch.instances. The result says so at once, the message is
// stored for a later claim, and launch.by is still told, in words that say no launch can follow.
test('a message to a role outside launch.instances says so in its result and in the notice', (t) => {
  const ctx = setup(t, demo({ launch: { by: 'chief', instances: ['pm'], models: {}, limits: {} } }));
  const sent = ctx.call('human', 'message.send', { to: 'coder', body: 'can you take this?', class: 'REQUEST' });
  assert.deepEqual(sent.waiting, { role: 'coder', launcher: 'chief', told: true, launcherRunning: false, launchable: false });
  assert.equal(ctx.store().get('SELECT to_role FROM messages WHERE id=?', sent.id).to_role, 'coder');
  const [n, ...more] = ctx.notices();
  assert.deepEqual(more, []);
  assert.equal(n.to_role, 'chief');
  assert.equal(n.in_reply_to, sent.id);
  assert.equal(
    n.body,
    `Message ${sent.id} from human to coder waits: coder has no running session and isn't in the launch block's instances, so a launch of coder is refused. Tell human why not with message.send in_reply_to ${sent.id}.`,
  );
  // The same business's launchable target keeps row 146's text.
  assert.equal(ctx.call('human', 'message.send', { to: 'pm', body: 'and you?' }).waiting.launchable, true);
  assert.match(ctx.notices().at(-1).body, / waits: pm has no running session\. Launch pm within /);
  // A later claim still delivers the stored message.
  ctx.launch('coder', 'coder');
  ctx.claim('coder');
  assert.deepEqual(ctx.call('coder', 'message.receive').map((m) => m.id), [sent.id]);
  assert.equal(ctx.notices().length, 2);
});

test('a message to a running role sends no notice and its result is unchanged', (t) => {
  const ctx = setup(t);
  ctx.launch('pm', 'pm');
  ctx.claim('pm');
  const sent = ctx.call('human', 'message.send', { to: 'pm', body: 'hello' });
  assert.deepEqual(Object.keys(sent).sort(), ['id', 'request']);
  assert.deepEqual(ctx.notices(), []);
});

test('a re-route at a claim, a release and claim, and a broker restart send no second notice', (t) => {
  const ctx = setup(t);
  const sent = ctx.call('human', 'message.send', { to: 'pm', body: 'waits' });
  assert.equal(ctx.notices().length, 1);
  ctx.launch('pm', 'pm');
  ctx.claim('pm');
  assert.equal(ctx.notices().length, 1, 'role.claim re-routes the waiting message only');
  assert.equal(
    ctx.store().get("SELECT holder_run FROM deliveries WHERE message=? AND op='routed' ORDER BY seq DESC LIMIT 1", sent.id).holder_run,
    'pm-run',
  );
  ctx.call('pm', 'role.release');
  ctx.restart();
  ctx.launch('pm2', 'pm');
  ctx.claim('pm2');
  assert.deepEqual(ctx.call('pm2', 'message.receive').map((m) => m.id), [sent.id]);
  assert.equal(ctx.notices().length, 1, 'a restarted broker and a new claim send none');
});

test('a run launched but not yet claimed counts as no holder', (t) => {
  const ctx = setup(t);
  ctx.launch('pm', 'pm');
  assert.equal(ctx.call('human', 'message.send', { to: 'pm', body: 'early' }).waiting.told, true);
  assert.equal(ctx.notices().length, 1);
});

test('no notice for a message to launch.by, from it, a refused one from a tester, to the human, or in a thread to a thread instance', (t) => {
  const ctx = setup(t);
  // To launch.by while it is stopped: the result says so, and nobody is told.
  assert.deepEqual(ctx.call('human', 'message.send', { to: 'chief', body: 'are you there?' }).waiting, {
    role: 'chief',
    launcher: 'chief',
    told: false,
    launcherRunning: false,
    launchable: false,
  });
  // From launch.by: the result tells it.
  ctx.launch('chief', 'chief');
  ctx.claim('chief');
  assert.deepEqual(ctx.call('chief', 'message.send', { to: 'pm', body: 'start on it' }).waiting, {
    role: 'pm',
    launcher: 'chief',
    told: false,
    launcherRunning: true,
    launchable: true,
  });
  // From a tester to a role nobody holds: refused before it is stored (row 170, #1667), so a PM
  // test wakes no launcher.
  ctx.caps.tester = ctx.b.bindTester({ business: 'demo' });
  const stored = ctx.store().get('SELECT count(*) n FROM messages').n;
  assert.throws(() => ctx.call('tester', 'message.send', { to: 'pm', body: '[mos test] ping' }), { code: 'tester-target-not-running' });
  assert.equal(ctx.store().get('SELECT count(*) n FROM messages').n, stored);
  // To the human or the tester, which no run holds: no waiting field at all.
  ctx.launch('coder', 'coder');
  ctx.claim('coder');
  assert.equal(ctx.call('coder', 'message.send', { to: 'human', body: 'fyi' }).waiting, undefined);
  const ping = ctx.call('tester', 'message.send', { to: 'coder', body: '[mos test] are you up?' });
  assert.equal(ping.waiting, undefined);
  assert.equal(ctx.call('coder', 'message.send', { to: 'tester', body: 'up', in_reply_to: ping.id }).waiting, undefined);
  // In a thread to a thread instance: row 121's launcher starts that session.
  const project = ctx.call('human', 'project.create', { name: 'work', repo: 'stack' }).id;
  const A = ctx.call('human', 'thread.create', { project, name: 'alpha' }).id;
  const asked = ctx.call('human', 'thread.send', { thread: A, body: 'what is left?' });
  const answer = ctx.call('chief', 'message.send', { to: 'assistant', body: 'two rows', in_reply_to: asked.id });
  assert.equal(answer.waiting, undefined);
  assert.equal(ctx.store().get('SELECT thread FROM message_threads WHERE message=?', answer.id).thread, A);
  assert.deepEqual(ctx.notices(), []);
});

test('a business with no launch block tells nobody and says there is no launcher', (t) => {
  const ctx = setup(t, demo({ launch: undefined }));
  assert.deepEqual(ctx.call('human', 'message.send', { to: 'pm', body: 'hello' }).waiting, {
    role: 'pm',
    launcher: null,
    told: false,
    launcherRunning: null,
    launchable: null,
  });
  assert.deepEqual(ctx.notices(), []);
});

test('the notice sender stays internal: no role is named _bus and no caller can send as it or to it', (t) => {
  assert.equal(ID_PATTERN.test('_bus'), false, 'a business file cannot declare an instance named _bus');
  const ctx = setup(t);
  ctx.launch('coder', 'coder');
  ctx.claim('coder');
  ctx.call('human', 'message.send', { to: 'pm', body: 'waits' });
  const [n] = ctx.notices();
  const refused = (who, args, code) =>
    assert.throws(() => ctx.call(who, 'message.send', { body: 'x', ...args }), { code }, JSON.stringify(args));
  for (const who of ['human', 'coder']) {
    refused(who, { to: '_bus' }, 'unknown-role');
    refused(who, { to: 'pm', from_role: '_bus' }, 'invalid-request');
    refused(who, { to: 'pm', from_run: 'bus' }, 'invalid-request');
    refused(who, { to: 'pm', class: '_bus' }, 'invalid-class');
    refused(who, { to: 'pm', in_reply_to: '_bus' }, 'invalid-request');
    refused(who, { to: 'pm', corrects: '_bus' }, 'invalid-request');
    refused(who, { to: 'pm', decision: '_bus' }, 'decision-not-found');
  }
  // Answering the notice goes to the sender it names, never back to _bus.
  ctx.launch('chief', 'chief');
  ctx.claim('chief');
  refused('chief', { to: '_bus', in_reply_to: n.id }, 'unknown-role');
  const why = ctx.call('chief', 'message.send', { to: 'human', body: 'pm is over its limit', in_reply_to: n.in_reply_to });
  assert.equal(ctx.store().get('SELECT from_role FROM messages WHERE id=?', why.id).from_role, 'chief');
  assert.equal(ctx.notices().length, 1);
});