stack/packages/bus/tests/tester.test.mjs

import test from 'node:test';
import assert from 'node:assert/strict';
import { spawn } from 'node:child_process';
import { connect } from 'node:net';
import { mkdtempSync, 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, TASK_VERBS } from '../src/broker.mjs';
import { serve } from '../src/server.mjs';
import { Client } from '../src/client.mjs';
import { startBroker } from '../src/runtime.mjs';
import { readProcess, verifyHuman } from '../src/human.mjs';
import { verifyTester, testerNonce } from '../src/tester.mjs';

// Row 110 (#1604): the bus tester identity for PM tests.
const roles = {
  pm: { authority: { withinRole: ['message.send'], crossRole: [] } },
  cto: { authority: { withinRole: ['message.send'], crossRole: [] } },
};
const business = (vars) => ({
  demo: { id: 'demo', human: 'jason', arbiters: { technical: 'cto', delivery: 'pm' }, roles, ...(vars ? { vars } : {}) },
});
function setup(t, vars = { 'bus.tester': true }) {
  const root = mkdtempSync(join(tmpdir(), 'bus-tester-'));
  const store = new Store(root);
  t.after(() => {
    store.close();
    rmSync(root, { recursive: true, force: true });
  });
  const b = new Broker({ store, businesses: business(vars) });
  const pm = b.bindLaunch({ business: 'demo', role: 'pm', run: 'pm-run', harness: 'pi' });
  b.request(pm, { verb: 'role.claim' });
  const human = b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true });
  const reader = b.bindReader({ business: 'demo' });
  const call = (cap, verb, args = {}) => b.request(cap, { verb, args });
  return { b, store, pm, human, reader, call };
}
const say = (body = 'is the plan current?') => `[mos test] ${body}`;

test('the tester binds only when the business sets bus.tester to exactly true', (t) => {
  for (const vars of [null, {}, { 'bus.tester': false }, { 'bus.tester': 'true' }, { 'bus.tester': 1 }]) {
    const { b } = setup(t, vars);
    assert.throws(() => b.bindTester({ business: 'demo' }), { code: 'tester-off' }, JSON.stringify(vars));
  }
  const { b } = setup(t);
  assert.match(b.bindTester({ business: 'demo' }), /^[0-9a-f]{64}$/);
  assert.throws(() => b.bindTester({ business: 'other' }), { code: 'unknown-business' });
  assert.throws(() => b.bindTester({ business: 'demo', role: 'pm' }), { code: 'invalid-request' });
});

test('a business may not name a role tester, so the tester address is never a role', (t) => {
  const root = mkdtempSync(join(tmpdir(), 'bus-tester-'));
  const store = new Store(root);
  t.after(() => {
    store.close();
    rmSync(root, { recursive: true, force: true });
  });
  const businesses = business();
  businesses.demo.roles = { ...roles, tester: roles.pm };
  assert.throws(() => new Broker({ store, businesses }), { code: 'invalid-business' });
  // No session can claim role tester: a launch must name a business role, and the tester
  // session's own role.claim refuses. So #holder(business, 'tester') is always empty.
  const s = setup(t);
  assert.throws(() => s.b.bindLaunch({ business: 'demo', role: 'tester', run: 'tester-run', harness: 'pi' }), { code: 'unknown-role' });
  assert.throws(() => s.call(s.b.bindTester({ business: 'demo' }), 'role.claim'), { code: 'tester-refused' });
  assert.equal(s.store.get("SELECT count(*) n FROM role_claims WHERE role='tester'").n, 0);
  assert.equal(s.store.get("SELECT count(*) n FROM role_claims WHERE role='pm'").n, 1);
});

test('every verb outside the tester list refuses tester-refused and is recorded against the tester', async (t) => {
  const { b, store, call } = setup(t);
  const tester = b.bindTester({ business: 'demo' });
  const refused = [
    ['role.claim', {}],
    ['role.release', {}],
    ['role.revoke', { role: 'pm' }],
    ['decision.raise', { action: 'message.send', question: 'q', options: [], recommendation: 'no', blocking: false }],
    ['decision.resolve', { id: 'd-1', choice: 'yes' }],
    ['decision.seen', { id: 'd-1' }],
    ['decision.withdraw', { id: 'd-1' }],
    ['review.request', { task_ref: 'vikunja:1/1', role: 'cto' }],
    ['review.verdict', { task_ref: 'vikunja:1/1', verdict: 'approve' }],
    ['launch.revoke', {}],
    ['launch.restore', {}],
    ['event.emit', { kind: 'action.allowed', body: { action: 'routine' } }],
    ['state', {}],
    ['credentials', {}],
    ['requirements', {}],
  ];
  for (const [verb, args] of refused) assert.throws(() => call(tester, verb, args), { code: 'tester-refused' }, verb);
  let handled = 0;
  for (const verb of TASK_VERBS)
    await assert.rejects(
      b.requestTask(tester, { verb, args: {} }, async () => handled++),
      { code: 'tester-refused' },
      verb,
    );
  assert.equal(handled, 0);
  const rows = store.all("SELECT actor_role, actor_run, body FROM events WHERE kind='action.refused'");
  assert.equal(rows.length, refused.length + TASK_VERBS.length);
  assert.ok(rows.every((r) => r.actor_role === 'tester' && r.actor_run === null && JSON.parse(r.body).code === 'tester-refused'));
  assert.equal(store.get("SELECT count(*) n FROM events WHERE kind IN ('launch.revoked','launch.restored','human.input')").n, 0);
  // The tester gate runs before any argument check, so malformed arguments refuse tester-refused too.
  for (const [verb] of refused) assert.throws(() => call(tester, verb, { bogus: 1 }), { code: 'tester-refused' }, `${verb} bogus`);
});

test('a tester message is from tester, starts [mos test], and writes no human.input or authority', (t) => {
  const { b, store, pm, call } = setup(t);
  const tester = b.bindTester({ business: 'demo' });
  assert.throws(() => call(tester, 'message.send', { to: 'pm', body: 'is the plan current?' }), { code: 'tester-prefix' });
  assert.throws(() => call(tester, 'message.send', { to: 'pm', body: ' [mos test] late prefix' }), { code: 'tester-prefix' });
  for (const to of ['human', 'tester', 'nobody'])
    assert.throws(() => call(tester, 'message.send', { to, body: say() }), { code: 'unknown-role' }, to);
  for (const extra of [{ corrects: 'm-1' }, { decision: 'd-1' }])
    assert.throws(() => call(tester, 'message.send', { to: 'pm', body: say(), ...extra }), { code: 'invalid-request' });
  const events = store.get('SELECT count(*) n FROM events WHERE kind<>?', 'action.refused').n;
  const sent = call(tester, 'message.send', { to: 'pm', body: say(), class: 'REQUEST' });
  assert.equal(sent.request, null);
  const row = store.get('SELECT * FROM messages WHERE id=?', sent.id);
  assert.equal(row.from_role, 'tester');
  assert.equal(row.from_run, 'tester-cli');
  assert.equal(row.to_role, 'pm');
  assert.equal(store.get('SELECT count(*) n FROM events WHERE kind<>?', 'action.refused').n, events, 'no event at all');
  const got = call(pm, 'message.receive');
  assert.equal(got.length, 1);
  assert.equal(got[0].from_role, 'tester');
  assert.equal(got[0].request, null);
  // task.create needs a human.input event as its request (packages/tasks/src/verbs.mjs); a tester
  // message gives none, so it can't back one.
  assert.equal(b.taskView('demo', 'input', sent.id), false);
  assert.equal(store.get("SELECT count(*) n FROM events WHERE kind='human.input'").n, 0);
});

test('a role answers a tester only in reply to a tester message, and the answer never reaches the human', (t) => {
  const { b, store, pm, human, reader, call } = setup(t);
  const tester = b.bindTester({ business: 'demo' });
  const ask = call(tester, 'message.send', { to: 'pm', body: say() }).id;
  const jason = call(human, 'message.send', { to: 'pm', body: 'a real instruction' }).id;
  call(pm, 'message.receive');
  assert.throws(() => call(pm, 'message.send', { to: 'tester', body: 'answer' }), { code: 'tester-reply-required' });
  assert.throws(() => call(pm, 'message.send', { to: 'tester', body: 'answer', in_reply_to: jason }), { code: 'tester-reply-required' });
  assert.throws(() => call(pm, 'message.send', { to: 'tester', body: 'answer', in_reply_to: 'missing' }), { code: 'tester-reply-required' });
  assert.throws(() => call(human, 'message.send', { to: 'tester', body: 'hi', in_reply_to: ask }), { code: 'unknown-role' });
  const answer = call(pm, 'message.send', { to: 'tester', body: 'the plan is current', in_reply_to: ask }).id;
  assert.equal(store.get('SELECT to_role FROM messages WHERE id=?', answer).to_role, 'tester');
  // Not in the human's inbox or the Console's Messages page.
  assert.deepEqual(call(human, 'message.receive'), []);
  for (const cap of [human, reader]) assert.equal(call(cap, 'messages').total, 0);
  const got = call(tester, 'message.receive');
  assert.deepEqual(got.map((m) => [m.id, m.from_role, m.in_reply_to]), [[answer, 'pm', ask]]);
  assert.deepEqual(call(tester, 'message.receive'), []);
  assert.throws(() => call(tester, 'message.read', { id: jason }), { code: 'message-not-found' });
  assert.deepEqual(call(tester, 'message.read', { id: answer }), { read: true });
  const delivered = store.all('SELECT op, holder_run, transport FROM deliveries WHERE message=? ORDER BY seq', answer);
  assert.deepEqual(delivered.map((d) => [d.op, d.holder_run, d.transport]), [
    ['routed', null, null],
    ['delivered', null, 'cli'],
    ['read', null, null],
  ]);
  // A role's own inbox is not the tester's.
  assert.throws(() => call(pm, 'message.read', { id: answer }), { code: 'message-not-found' });
});

test('the tester reads the human views a reader reads', (t) => {
  const { b, pm, reader, call } = setup(t);
  const tester = b.bindTester({ business: 'demo' });
  const ask = call(tester, 'message.send', { to: 'pm', body: say() }).id;
  call(pm, 'message.receive');
  call(pm, 'message.send', { to: 'tester', body: 'answer', in_reply_to: ask });
  call(pm, 'message.send', { to: 'human', body: 'for jason' });
  for (const [verb, args] of [
    ['inbox', {}],
    ['agents', {}],
    ['tasks', {}],
    ['trail', { subject: ask }],
    ['decisions', { list: 'roles' }],
    ['decisions', { list: 'recent' }],
    ['messages', {}],
    ['replies', { id: ask }],
  ])
    assert.deepEqual(call(tester, verb, args), call(reader, verb, args), verb);
  assert.equal(call(tester, 'replies', { id: ask }).length, 1);
  assert.equal(call(tester, 'messages').total, 1);
});

// The process proof, on a process table like human.test.mjs's.
const testerCli = '/opt/mosaic/tester-cli.mjs';
const humanCli = '/opt/mosaic/human-cli.mjs';
const nonce = 'b'.repeat(64);
function table(cli = testerCli, env = { MOSAIC_BUS_TESTER_NONCE: nonce }) {
  return new Map([
    [50, { pid: 50, ppid: 40, startTime: '100', uid: process.getuid(), argv: ['node', cli], env }],
    [40, { pid: 40, ppid: 1, startTime: '90', uid: process.getuid(), argv: ['bash'], env: {} }],
  ]);
}
const proof = { pid: 50, startTime: '100', nonce, business: 'demo' };
const reader = (t) => (pid) => {
  if (!t.has(pid)) throw Error('gone');
  return t.get(pid);
};
const tester = (t, extra = {}) =>
  verifyTester(proof, {
    cliPath: testerCli,
    readProcess: reader(t),
    readNonce: (pid) => t.get(pid)?.env?.MOSAIC_BUS_TESTER_NONCE ?? null,
    ...extra,
  });

test('tester proof binds its own CLI, start time and nonce, refuses a managed run and allows an agent', () => {
  assert.equal(tester(table()).business, 'demo');
  for (const change of [{ startTime: '101' }, { uid: process.getuid() + 1 }, { argv: ['node', 'other.mjs'] }, { env: { MOSAIC_BUS_TESTER_NONCE: 'c'.repeat(64) } }]) {
    const t = table();
    Object.assign(t.get(50), change);
    assert.throws(() => tester(t), { code: 'tester-required' }, JSON.stringify(change));
  }
  for (const env of [{ MOSAIC_RUN_ID: 'run-x' }, { MOSAIC_BUS_CAP: 'cap' }, { MOSAIC_AGENT_RUN: '1' }]) {
    const t = table();
    t.get(40).env = env;
    assert.throws(() => tester(t), { code: 'tester-required' }, Object.keys(env)[0]);
    const self = table();
    Object.assign(self.get(50).env, env);
    assert.throws(() => tester(self), { code: 'tester-required' }, `${Object.keys(env)[0]} on the CLI`);
  }
  assert.throws(() => tester(table(), { launches: [{ pid: 40, startTime: '90' }] }), { code: 'tester-required' });
  assert.equal(tester(table(), { launches: [{ pid: 40, startTime: '89' }] }).business, 'demo');
  // An agent above the tester is the intended case: Mos or a T3 seat.
  for (const [argv, env] of [
    [['bash'], { CLAUDECODE: '1', CLAUDE_CODE_ENTRYPOINT: 'sdk-ts' }],
    [['bash'], { CODEX_THREAD_ID: 'thread', PI_AGENT_DIR: '/x' }],
    [['claude'], {}],
    [['node', '/usr/lib/node_modules/codex/bin/codex.js'], {}],
    [['pi'], {}],
  ]) {
    const t = table();
    Object.assign(t.get(40), { argv, env });
    assert.equal(tester(t).business, 'demo', argv.join(' '));
  }
  const broken = table();
  broken.delete(40);
  assert.throws(() => tester(broken), { code: 'tester-required' });
  // A loop refuses when it comes round, not at the depth bound.
  const loop = table();
  loop.get(40).ppid = 50;
  let reads = 0;
  assert.throws(() => tester(loop, { readProcess: (pid) => (reads++, reader(loop)(pid)) }), { code: 'tester-required' });
  assert.ok(reads < 5, `${reads} reads`);
  // A parent pid under 2 refuses even when /proc would answer for it.
  const gone = table();
  gone.get(40).ppid = 0;
  gone.set(0, { pid: 0, ppid: 1, startTime: '1', uid: process.getuid(), argv: ['init'], env: {} });
  assert.throws(() => tester(gone), { code: 'tester-required' });
  // The CLI is read again at the top: a changed start time or parent refuses.
  for (const change of [{ startTime: '101' }, { ppid: 1 }]) {
    const t = table();
    let first = true;
    const again = (pid) => (pid === 50 && !first ? { ...t.get(50), ...change } : ((first = pid === 50 ? false : first), t.get(pid)));
    assert.throws(() => tester(t, { readProcess: again }), { code: 'tester-required' }, JSON.stringify(change));
  }
  // The first read's start time is checked on its own, not only by the re-read.
  const once = table();
  let fresh = true;
  const first = (pid) => (pid === 50 && fresh ? ((fresh = false), { ...once.get(50), startTime: '101' }) : once.get(pid));
  assert.throws(() => tester(once, { readProcess: first }), { code: 'tester-required' });
  // Each bad shape sits on a table where every other check passes, so only the shape check refuses.
  const pid1 = table();
  pid1.set(1, { ...pid1.get(50), pid: 1 });
  const nonHex = table(testerCli, { MOSAIC_BUS_TESTER_NONCE: 'x' });
  for (const [bad, t] of [
    [null, table()],
    [{ ...proof, extra: 1 }, table()],
    [{ ...proof, pid: 1 }, pid1],
    [{ ...proof, nonce: 'x' }, nonHex],
    [{ ...proof, business: 3 }, table()],
  ])
    assert.throws(
      () => verifyTester(bad, { cliPath: testerCli, readProcess: reader(t), readNonce: (pid) => t.get(pid)?.env?.MOSAIC_BUS_TESTER_NONCE ?? null }),
      { code: 'tester-required' },
      JSON.stringify(bad),
    );
});

test("neither CLI's nonce proves the other: each proof is bound to its own CLI and nonce variable", () => {
  // A human CLI process, offered as a tester proof.
  const human = table(humanCli, { MOSAIC_BUS_CLI_NONCE: nonce });
  assert.throws(() => tester(human), { code: 'tester-required' });
  assert.throws(() => tester(human, { cliPath: humanCli }), { code: 'tester-required' }, 'right path, wrong variable');
  // A tester CLI process, offered as a human proof.
  const t = table();
  const human_ = (extra = {}) => verifyHuman(proof, { cliPath: humanCli, readProcess: reader(t), ...extra });
  assert.throws(() => human_(), { code: 'human-required' });
  assert.throws(() => human_({ cliPath: testerCli }), { code: 'human-required' }, 'right path, wrong variable');
  // Both variables set: the path still decides.
  const both = table(testerCli, { MOSAIC_BUS_TESTER_NONCE: nonce, MOSAIC_BUS_CLI_NONCE: nonce });
  assert.throws(() => verifyHuman(proof, { cliPath: humanCli, readProcess: reader(both) }), { code: 'human-required' });
  assert.throws(() => tester(both, { cliPath: humanCli }), { code: 'tester-required' });
});

test("testerNonce reads only the tester's variable and gives null on any read error", () => {
  const io = (text) => ({ readFile: () => text });
  assert.equal(testerNonce(5, io(`A=1\0MOSAIC_BUS_TESTER_NONCE=${nonce}\0`)), nonce);
  assert.equal(testerNonce(5, io(`MOSAIC_BUS_CLI_NONCE=${nonce}\0`)), null);
  assert.equal(testerNonce(5, { readFile: () => { throw Object.assign(Error(), { code: 'EACCES' }); } }), null);
  assert.equal(testerNonce(process.pid), null);
});

test('an envelope carries exactly one of cap, human and tester; no tester authenticator refuses', async (t) => {
  const root = mkdtempSync(join(tmpdir(), 'bus-tester-'));
  const store = new Store(root);
  const b = new Broker({ store, businesses: business({ 'bus.tester': true }) });
  const path = join(store.directory, 'broker.sock');
  let calls = 0;
  let bound = null;
  const server = await serve({ broker: b, path, authenticateHuman: () => (calls++, 'x'), authenticateTester: () => (calls++, (bound = b.bindTester({ business: 'demo' }))) });
  const bare = join(store.directory, 'bare.sock');
  const bareServer = await serve({ broker: b, path: bare });
  t.after(async () => {
    await server.close();
    await bareServer.close();
    store.close();
    rmSync(root, { recursive: true, force: true });
  });
  const raw = (at, request) =>
    new Promise((resolve, reject) => {
      const s = connect(at);
      let out = '';
      s.on('data', (d) => (out += d));
      s.on('end', () => resolve(JSON.parse(out)));
      s.on('error', reject);
      s.end(JSON.stringify(request) + '\n');
    });
  const p = { business: 'demo', pid: 2, startTime: '1', nonce };
  const cap = b.bindReader({ business: 'demo' });
  const role = b.bindLaunch({ business: 'demo', role: 'pm', run: 'pm-run', harness: 'pi' });
  for (const r of [{ tester: p, cap }, { tester: p, cap: role }, { tester: p, human: p }, { tester: p, human: p, cap }])
    assert.deepEqual(await raw(path, { ...r, verb: 'agents' }), { ok: false, error: 'invalid-envelope' }, Object.keys(r).join('+'));
  assert.equal(calls, 0);
  assert.deepEqual(await raw(bare, { tester: p, verb: 'agents' }), { ok: false, error: 'tester-required' });
  assert.deepEqual(await raw(path, { tester: p, verb: 'agents' }), { ok: true, result: [] });
  assert.equal(calls, 1);
  // The tester's capability lasts one request, as the human's does.
  assert.throws(() => b.request(bound, { verb: 'agents' }), { code: 'unauthenticated' });
  assert.deepEqual(await new Client({ path, tester: p }).call('agents'), []);
});

// The real shim against a real runtime. The test process is not a managed run, so the proof
// passes here; a managed run's marker on the shim itself refuses.
const shim = fileURLToPath(new URL('../src/tester-cli.mjs', import.meta.url));
// Async: the broker runs in this process, so a synchronous spawn would block its answer.
function run(socket, request, env = {}) {
  const clean = Object.fromEntries(Object.entries(process.env).filter(([k]) => !/^MOSAIC_BUS_(TESTER|CLI)_NONCE$/.test(k)));
  return new Promise((resolve, reject) => {
    const child = spawn(process.execPath, [shim, socket], { env: { ...clean, ...env }, timeout: 20000 });
    let out = '';
    let err = '';
    child.stdout.on('data', (d) => (out += d));
    child.stderr.on('data', (d) => (err += d));
    child.on('error', reject);
    child.on('close', (status) => resolve({ status, out: out.trim(), err: err.trim() }));
    child.stdin.end(JSON.stringify(request));
  });
}
async function runtime(t, vars, launches = []) {
  const root = mkdtempSync(join(tmpdir(), 'bus-tester-rt-'));
  const rt = await startBroker({ dataRoot: root, businesses: business(vars), readers: ['demo'], launches });
  t.after(async () => {
    await rt.close();
    rmSync(root, { recursive: true, force: true });
  });
  return rt;
}

test('tester-cli.mjs binds through the runtime when the switch is on, and refuses off or under a run marker', async (t) => {
  const on = await runtime(t, { 'bus.tester': true });
  const reader = new Client({ path: on.path, cap: on.readers[0].cap });
  const sent = await run(on.path, { business: 'demo', verb: 'message.send', args: { to: 'pm', body: say() } });
  assert.equal(sent.status, 0, sent.err);
  const id = JSON.parse(sent.out).id;
  assert.equal((await reader.call('trail', { subject: id })).find((r) => r.table === 'messages').from_role, 'tester');
  assert.deepEqual(await run(on.path, { business: 'demo', verb: 'state' }), { status: 2, out: '', err: 'tester-refused' });
  for (const marker of ['MOSAIC_RUN_ID', 'MOSAIC_BUS_CAP', 'MOSAIC_AGENT_RUN'])
    assert.deepEqual(await run(on.path, { business: 'demo', verb: 'agents' }, { [marker]: 'x' }), { status: 2, out: '', err: 'tester-required' }, marker);
  assert.equal((await run(on.path, { business: 'demo', verb: 'agents' }, { CLAUDECODE: '1' })).status, 0);
  assert.deepEqual(await run(on.path, { business: 'demo', verb: 'agents', cap: 'x' }), { status: 2, out: '', err: 'invalid-request' });
  // This test process as a managed run's launch record: the shim it spawns has it as an ancestor.
  const me = readProcess(process.pid);
  const under = await runtime(t, { 'bus.tester': true }, [{ business: 'demo', role: 'cto', run: 'cto-run', harness: 'pi', pid: me.pid, startTime: me.startTime }]);
  assert.deepEqual(await run(under.path, { business: 'demo', verb: 'agents' }), { status: 2, out: '', err: 'tester-required' });
  const off = await runtime(t, null);
  assert.deepEqual(await run(off.path, { business: 'demo', verb: 'agents' }), { status: 2, out: '', err: 'tester-off' });
});