stack/agents/dewey/work/queue-76/demo.mjs

// Row 76 (#1570): `serve --send` end to end against a real broker, driven over HTTP on loopback
// with the headers the sign-in proxy and the browser would send.
// Run: TMPDIR=<scratch> node demo.mjs <worktree with the candidate>.
// The broker is in this process, as in send-browser.test.mjs; the human transport between the
// Console and a bus host is send.test.mjs's. `send` maps the broker's refusal the way humanCall does.
import { request } from 'node:http';
import { mkdtempSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join, resolve } from 'node:path';
import { pathToFileURL } from 'node:url';

const tree = resolve(process.argv[2]);
const load = p => import(pathToFileURL(join(tree, p)).href);
const { Store } = await load('packages/bus/src/store.mjs');
const { Broker } = await load('packages/bus/src/broker.mjs');
const { views } = await load('packages/bus/src/views.mjs');
const { startServer } = await load('packages/webui/src/serve.mjs');
const { BusReadError } = await load('packages/webui/src/bus.mjs');

const NAME = 'mosaic-preview.example.com', HEADER = 'x-authentik-uid', ID = 'jason';
const preview = { host: NAME, header: HEADER, user: ID };
const role = { authority: { withinRole: ['message.send'], crossRole: [] } };
const businesses = { demo: { id: 'demo', human: 'jason', arbiters: { technical: 'cto', delivery: 'pm' }, roles: { pm: role, cto: role, coder: role } } };
const root = mkdtempSync(join(tmpdir(), 'r76-demo-'));
const store = new Store(root), b = new Broker({ store, businesses });
const coder = b.bindLaunch({ business: 'demo', role: 'coder', run: 'coder-run', harness: 'pi', address: 'coder-run' });
b.request(coder, { verb: 'role.claim' });
const human = b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true });
const reader = b.bindReader({ business: 'demo' });
const bus = views({ call: async (verb, args = {}) => structuredClone(b.request(reader, { verb, args })) });
const toBus = fn => async args => { try { return await fn(args); } catch (e) { throw new BusReadError(e.code ?? 'outcome-unknown'); } };
const send = toBus(async args => structuredClone(b.request(human, { verb: 'message.send', args })));
// A transport that loses the answer after the broker wrote: the outcome is unknown.
const lost = toBus(async args => { b.request(human, { verb: 'message.send', args }); throw Object.assign(new Error('lost'), { code: 'outcome-unknown' }); });
const lines = [];
const log = l => lines.push(l);
const servers = {
  send: await startServer({ port: 0, board: 'http://127.0.0.1:9', bus, preview, send, log }),
  lost: await startServer({ port: 0, board: 'http://127.0.0.1:9', bus, preview, send: lost, log }),
  readonly: await startServer({ port: 0, board: 'http://127.0.0.1:9', bus, preview, log }),
};

const signed = { [HEADER]: ID, origin: `https://${NAME}`, 'sec-fetch-site': 'same-origin', 'content-type': 'application/json' };
function call(server, { method = 'POST', headers = signed, body, host = NAME } = {}) {
  return new Promise((done, fail) => {
    const length = body === undefined ? {} : { 'content-length': Buffer.byteLength(body) };
    const req = request({ host: '127.0.0.1', port: servers[server].address().port, method, path: '/api/bus/message', headers: { host, ...length, ...headers } }, res => {
      let text = '';
      res.on('data', c => (text += c));
      res.on('end', () => done(`${res.statusCode}  ${text}`));
    });
    req.on('error', fail);
    req.end(body);
  });
}
const omit = (k, extra = {}) => ({ ...Object.fromEntries(Object.entries(signed).filter(([n]) => n !== k)), ...extra });
const show = async (what, server, options) => console.log(`${what.padEnd(46)} ${await call(server, options)}`);
const BODY = 'Please look at row 76 before the cutover.';
const sent = [];

console.log('== a send, and a reply to the coder');
const first = await call('send', { body: JSON.stringify({ to: 'coder', body: BODY }) });
console.log(`${'to coder'.padEnd(46)} ${first}`);
sent.push(JSON.parse(first.slice(5)).id);
const asked = b.request(coder, { verb: 'message.send', args: { to: 'human', body: 'Which branch?' } });
const reply = await call('send', { body: JSON.stringify({ to: 'coder', body: 'refactor', in_reply_to: asked.id }) });
console.log(`${'to coder, in_reply_to the coder\'s message'.padEnd(46)} ${reply}`);
sent.push(JSON.parse(reply.slice(5)).id);

console.log('== refused before the bus');
await show('no sign-in header (401, no log line)', 'send', { headers: omit(HEADER), body: JSON.stringify({ to: 'coder', body: BODY }) });
await show('loopback Host with the header', 'send', { host: '127.0.0.1', body: JSON.stringify({ to: 'coder', body: BODY }) });
await show('another Origin (shared check, no log line)', 'send', { headers: omit('origin', { origin: 'https://evil.example' }), body: JSON.stringify({ to: 'coder', body: BODY }) });
await show('no Origin', 'send', { headers: omit('origin'), body: JSON.stringify({ to: 'coder', body: BODY }) });
await show('Sec-Fetch-Site: same-site', 'send', { headers: omit('sec-fetch-site', { 'sec-fetch-site': 'same-site' }), body: JSON.stringify({ to: 'coder', body: BODY }) });
await show('Content-Type: text/plain', 'send', { headers: omit('content-type', { 'content-type': 'text/plain' }), body: JSON.stringify({ to: 'coder', body: BODY }) });
await show('16385 bytes', 'send', { body: JSON.stringify({ to: 'coder', body: 'x'.repeat(16385 - 23) }) });
await show('an extra key', 'send', { body: JSON.stringify({ to: 'coder', body: BODY, class: 'BLOCKING' }) });
await show('GET', 'send', { method: 'GET', body: undefined });
console.log('== refused by the broker');
await show('a role the business lacks', 'send', { body: JSON.stringify({ to: 'auditor', body: BODY }) });
await show('an empty body', 'send', { body: JSON.stringify({ to: 'coder', body: '' }) });
await show('in_reply_to a missing message', 'send', { body: JSON.stringify({ to: 'coder', body: BODY, in_reply_to: 'msg-none' }) });
console.log('== the answer lost after the broker wrote');
await show('unknown, not retried', 'lost', { body: JSON.stringify({ to: 'pm', body: 'Lost answer test' }) });
console.log('== without --send');
await show('POST is read-only', 'readonly', { body: JSON.stringify({ to: 'coder', body: BODY }) });

console.log('== what the broker holds');
for (const id of sent) {
  const [m] = b.request(human, { verb: 'trail', args: { subject: id } }).filter(r => r.table === 'messages' && r.id === id);
  console.log(`${id.slice(0, 8)}  from=${m.from_role}/${m.from_run} to=${m.to_role} class=${m.class} in_reply_to=${m.in_reply_to ?? '-'} body=${JSON.stringify(m.body)}`);
}
const pm = b.bindLaunch({ business: 'demo', role: 'pm', run: 'pm-run', harness: 'pi', address: 'pm-run' });
b.request(pm, { verb: 'role.claim' });
const lostRows = b.request(pm, { verb: 'message.receive', args: {} });
console.log(`pm receives ${lostRows.length} message(s) from the lost-answer send: ${JSON.stringify(lostRows.map(m => m.body))}`);

console.log('== the Console log, one line per send past sign-in');
for (const l of lines) console.log(l);
console.log(`a body in the log: ${lines.some(l => l.includes(BODY) || l.includes('refactor') || l.includes('Lost answer'))}`);
for (const s of Object.values(servers)) await new Promise(r => s.close(r));
store.close();
rmSync(root, { recursive: true, force: true });