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

import test from 'node:test';
import assert from 'node:assert/strict';
import { fork } from 'node:child_process';
import { once } from 'node:events';
import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { Store } from '../src/store.mjs';
import { Broker } from '../src/broker.mjs';
import { Client } from '../src/client.mjs';
// Row 264a (#1769): a thread made by hand gets the default session the 3k-3l migration gives a
// legacy thread, with the thread's id.
const DEFAULT_SESSION = [
  'INSERT INTO sessions(at,business,thread,id,instance,owner,name,op) SELECT at,business,id,id,instance,owner,\'default\',\'legacy-3l\' FROM threads WHERE id=?',
  'INSERT INTO session_defaults(at,thread,session,by) SELECT at,id,id,\'migration-3l\' FROM threads WHERE id=?',
];

const businesses = {
  demo: {
    id: 'demo',
    human: 'jason',
    // Row 65: runtime launches bind only on a harness the business switches on.
    plugins: { harness: ['pi'] },
    arbiters: { technical: 'cto', delivery: 'pm' },
    launch: { by: 'pm', instances: ['coder'], max: { opus: 2, sonnet: 2 } },
    roles: {
      pm: { authority: { withinRole: ['message.send', 'role.launch'], crossRole: [] } },
      cto: { authority: { withinRole: ['message.send'], crossRole: [] } },
      coder: { authority: { withinRole: ['message.send'], crossRole: [] } },
    },
  },
};
function setup(t) {
  const root = mkdtempSync(join(tmpdir(), 'bus-end-'));
  const store = new Store(root);
  const b = new Broker({ store, businesses });
  t.after(() => {
    store.close();
    rmSync(root, { recursive: true, force: true });
  });
  const agent = (role, run = role + '-run') => b.bindLaunch({ business: 'demo', role, run, harness: 'pi' });
  return { b, store, agent };
}
const call = (b, cap, verb, args = {}) => b.request(cap, { verb, args });

test('endLaunch writes session.ended, releases the run claim and kills its capabilities', (t) => {
  const { b, store, agent } = setup(t);
  const c = agent('coder');
  call(b, c, 'role.claim');
  assert.deepEqual(b.endLaunch({ business: 'demo', run: 'coder-run', reason: 'stopped', exitCode: 0 }), {
    released: true,
  });
  const ended = store.get("SELECT * FROM events WHERE kind='session.ended'");
  assert.equal(ended.subject, 'coder-run');
  assert.equal(ended.actor_role, 'coder');
  assert.equal(ended.actor_run, 'coder-run');
  assert.deepEqual(JSON.parse(ended.body), { reason: 'stopped', exitCode: 0, released: true });
  const claim = store.get("SELECT * FROM role_claims WHERE role='coder' ORDER BY seq DESC LIMIT 1");
  assert.equal(claim.op, 'release');
  assert.equal(claim.by, 'coder-run');
  assert.throws(() => call(b, c, 'message.send', { to: 'pm', body: 'still here?' }), /unauthenticated/);
  assert.throws(() => b.identity(c), /unauthenticated/);
});

test('endLaunch refuses an unknown run, a second end and a rebind of the ended run', (t) => {
  const { b, agent } = setup(t);
  agent('coder');
  assert.throws(() => b.endLaunch({ business: 'demo', run: 'nope', reason: 'exited' }), /unknown-run/);
  assert.throws(() => b.endLaunch({ business: 'demo', run: 'coder-run', reason: 'bad reason' }), /invalid-request/);
  assert.throws(
    () => b.endLaunch({ business: 'demo', run: 'coder-run', reason: 'exited', exitCode: 1.5 }),
    /invalid-request/,
  );
  assert.deepEqual(b.endLaunch({ business: 'demo', run: 'coder-run', reason: 'exited', exitCode: null }), {
    released: false,
  });
  assert.throws(() => b.endLaunch({ business: 'demo', run: 'coder-run', reason: 'exited' }), /run-ended/);
  assert.throws(() => agent('coder'), /duplicate-run/);
});

test('a restarted broker refuses to rebind an ended run; a refused rebind leaves the run unbound', (t) => {
  const { b, store, agent } = setup(t);
  agent('coder');
  agent('pm');
  b.endLaunch({ business: 'demo', run: 'coder-run', reason: 'stopped', exitCode: 0 });
  const again = new Broker({ store, businesses });
  assert.throws(() => again.bindLaunch({ business: 'demo', role: 'coder', run: 'coder-run', harness: 'pi' }), /run-ended/);
  assert.throws(() => again.endLaunch({ business: 'demo', run: 'coder-run', reason: 'host-lost' }), /unknown-run/);
  assert.throws(() => again.bindLaunch({ business: 'demo', role: 'cto', run: 'pm-run', harness: 'pi' }), /launch-record-mismatch/);
  again.bindLaunch({ business: 'demo', role: 'pm', run: 'pm-run', harness: 'pi' });
  assert.deepEqual(again.endLaunch({ business: 'demo', run: 'pm-run', reason: 'host-lost' }), { released: false });
  assert.equal(store.get("SELECT count(*) n FROM events WHERE kind='session.ended'").n, 2);
});

test('endLaunch leaves a claim another run took alone', (t) => {
  const { b, store, agent } = setup(t);
  const first = agent('coder', 'c1');
  call(b, first, 'role.claim');
  call(b, first, 'role.release');
  call(b, agent('coder', 'c2'), 'role.claim');
  assert.deepEqual(b.endLaunch({ business: 'demo', run: 'c1', reason: 'exited', exitCode: 0 }), { released: false });
  assert.equal(store.get("SELECT holder_run FROM role_claims ORDER BY seq DESC LIMIT 1").holder_run, 'c2');
});

test('refuse records action.refused against the caller with the code only', (t) => {
  const { b, store, agent } = setup(t);
  const pm = agent('pm');
  assert.equal(b.refuse(pm, 'capacity-full'), 'capacity-full');
  const e = store.get("SELECT * FROM events WHERE kind='action.refused'");
  assert.equal(e.actor_role, 'pm');
  assert.deepEqual(JSON.parse(e.body), { code: 'capacity-full' });
  assert.throws(() => b.refuse(pm, 'Not A Code'), /invalid-request/);
  assert.throws(() => b.refuse('f'.repeat(64), 'capacity-full'), /unauthenticated/);
});

function launch(t) {
  const child = fork(new URL('../src/process.mjs', import.meta.url), [], { stdio: ['ignore', 'pipe', 'pipe', 'ipc'] });
  t.after(() => {
    if (child.exitCode === null) child.kill('SIGKILL');
  });
  return child;
}
async function ask(child, message) {
  const reply = once(child, 'message');
  child.send(message);
  return (await reply)[0];
}

test('launches off refuses role.launch with launch-revoked until launches on', (t) => {
  const { b, agent } = setup(t);
  const pm = agent('pm');
  call(b, pm, 'role.claim');
  const human = b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true });
  assert.equal(b.authorize(pm, 'role.launch', { target: 'coder' }).class, 'within-role');
  call(b, human, 'launch.revoke');
  assert.throws(() => b.authorize(pm, 'role.launch', { target: 'coder' }), { code: 'launch-revoked' });
  // Other actions are untouched; only launching stops.
  assert.equal(b.authorize(pm, 'message.send').class, 'within-role');
  call(b, human, 'launch.restore');
  assert.equal(b.authorize(pm, 'role.launch', { target: 'coder' }).class, 'within-role');
});

test('broker process: launch ops authorize role.launch, record refusals and end runs', async (t) => {
  const root = mkdtempSync(join(tmpdir(), 'bus-launch-ops-'));
  t.after(() => rmSync(root, { recursive: true, force: true }));
  const token = join(root, 'token');
  writeFileSync(token, 'fixture-token-6c1d', { mode: 0o600 });
  const config = structuredClone(businesses);
  config.demo.roles.coder.credentials = { vikunja: { file: token, expires: '2099-01-01' } };
  const child = launch(t);
  const data = join(root, 'data');
  mkdirSync(data);
  // Row 121 (#1615, REQ-LAUNCH-1): a thread for the launcher's thread view below, written
  // before the child takes the store's writer lock. The human verbs need an assistant instance.
  const side = new Store(data);
  side.run('INSERT INTO projects(id,at,business,name,repo,created_by) VALUES(?,?,?,?,?,?)', 'p-1', '2026-10-15T00:00:00.000Z', 'demo', 'work', null, 'jason');
  side.run('INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES(?,?,?,?,?,?,?)', 'th-1', '2026-10-15T00:00:00.000Z', 'demo', 'p-1', 'coder', 'jason', 'one');
  for (const sql of DEFAULT_SESSION) side.run(sql, 'th-1');
  side.run('INSERT INTO thread_opens(at,thread,session,by) VALUES(?,?,?,?)', '2026-10-15T00:00:01.000Z', 'th-1', 'th-1', 'jason');
  side.close();
  const ready = await ask(child, {
    op: 'boot',
    config: { dataRoot: data, businesses: config, launches: [], readers: [] },
  });
  assert.equal(ready.ok, true, ready.error);
  const record = (role, run) => ({ business: 'demo', role, run, harness: 'pi', pid: process.pid, startTime: '1' });
  const pm = (await ask(child, { op: 'bindLaunch', record: record('pm', 'pm-1') })).launch.cap;
  const cto = (await ask(child, { op: 'bindLaunch', record: record('cto', 'cto-1') })).launch.cap;
  assert.deepEqual((await ask(child, { op: 'identity', cap: pm })).result, {
    business: 'demo',
    role: 'pm',
    run: 'pm-1',
    human: null,
  });
  assert.equal((await ask(child, { op: 'identity', cap: 'f'.repeat(64) })).error, 'unauthenticated');
  // A launch is authorized for the role's holder only, so both runs claim first, as the runner does.
  const reader = new Client({ path: ready.path, cap: cto });
  await new Client({ path: ready.path, cap: pm }).call('role.claim');
  await reader.call('role.claim');
  const ok = await ask(child, { op: 'authorizeLaunch', cap: pm, instance: 'coder' });
  assert.equal(ok.ok, true);
  assert.equal(ok.result.class, 'within-role');
  // The cto is not the business's launcher, so role.launch is gated for it and the refusal is evidence.
  const no = await ask(child, { op: 'authorizeLaunch', cap: cto, instance: 'coder' });
  assert.equal(no.ok, false);
  assert.equal(no.error, 'decision-required');
  assert.equal((await ask(child, { op: 'refuse', cap: pm, code: 'capacity-full' })).result, 'capacity-full');
  assert.deepEqual((await ask(child, { op: 'credentialStatus', business: 'demo', role: 'coder' })).result, [
    { service: 'vikunja', date: '2099-01-01', state: 'valid' },
  ]);
  assert.deepEqual((await ask(child, { op: 'credentialStatus', business: 'demo', role: 'pm' })).result, []);
  // The launcher's thread view, through the same child.
  assert.deepEqual((await ask(child, { op: 'threadsState', business: 'demo' })).result, [
    { thread: 'th-1', instance: 'coder', pending: 0, unread: 0, lastMessage: null, lastOpen: '2026-10-15T00:00:01.000Z', lastClaim: null, holder: null, session: 'th-1', legacy: true },
  ]);
  // And its idle release (Dewey's F3, 28591): the op reaches the broker, which refuses a run
  // that doesn't hold the thread.
  assert.deepEqual(await ask(child, { op: 'threadIdle', record: { business: 'demo', thread: 'th-1', run: 'pm-1' } }), { ok: false, error: 'not-holder' });
  const end = await ask(child, { op: 'endLaunch', record: { business: 'demo', run: 'pm-1', reason: 'stopped', exitCode: 0 } });
  assert.deepEqual(end, { ok: true, result: { released: true } });
  // The host's request id comes back on every launch-op and bind reply, refusals too.
  assert.equal((await ask(child, { op: 'identity', cap: cto, id: 7 })).id, 7);
  assert.deepEqual(await ask(child, { op: 'identity', cap: 'f'.repeat(64), id: 8 }), { ok: false, error: 'unauthenticated', id: 8 });
  assert.equal((await ask(child, { op: 'bindLaunch', record: record('coder', 'coder-1'), id: 9 })).id, 9);
  assert.deepEqual(await ask(child, { op: 'bindLaunch', record: record('coder', 'coder-1'), id: 10 }), { ok: false, error: 'duplicate-run', id: 10 });
  await assert.rejects(new Client({ path: ready.path, cap: pm }).call('agents'), /unauthenticated/);
  const trail = await reader.call('trail', { subject: 'pm-1' });
  assert.deepEqual(
    trail.filter((r) => r.table === 'events').map((r) => r.kind),
    ['session.launched', 'session.ended'],
  );
  const exit = once(child, 'exit');
  child.send({ op: 'close' });
  assert.equal((await exit)[0], 0);
});