12// End-to-end optional-agent check. All state, keys and processes are local.3import {finalizeEvent, getPublicKey} from 'nostr-tools/pure';4import {createHash, randomUUID} from 'node:crypto';5import {mkdtemp, rm} from 'node:fs/promises';6import {createServer} from 'node:net';7import {tmpdir} from 'node:os';8import {dirname, resolve} from 'node:path';9import {spawn, spawnSync} from 'node:child_process';1011const root = resolve(dirname(new URL(import.meta.url).pathname), '../..');12const binary = process.env.TINY_QA_BINARY || '/tmp/tiny-buzz-qa';13const reservation = createServer();14await new Promise((resolve,reject)=>{reservation.once('error',reject);reservation.listen(0,'127.0.0.1',resolve);});15const port = reservation.address().port;16await new Promise(resolve=>reservation.close(resolve));17const base = `http://127.0.0.1:${port}`;18const ws = `ws://127.0.0.1:${port}`;19const keys = Object.fromEntries(['owner', 'agent'].map((name, i) => [name, Uint8Array.from(Buffer.from((i + 1).toString(16).padStart(64, '0'), 'hex'))]));20const pub = Object.fromEntries(Object.entries(keys).map(([name, key]) => [name, getPublicKey(key)]));21const data = await mkdtemp(`${tmpdir()}/tiny-buzz-runner-`);22let relay, runner;23try {24 requireBinary();25 run([binary, 'tenant', 'create', '--data-dir', data, '--name', 'main', '--owner', pub.owner]);26 relay = spawn(binary, ['serve', '--data-dir', data, '--default-tenant', 'main', '--listen', `127.0.0.1:${port}`, '--allow-private-relays'], {stdio: ['ignore', 'ignore', 'pipe']});27 await waitFor(`${base}/readyz`);28 await publish('owner', 0, JSON.stringify({name: 'QA owner'}));29 await publish('owner', 30392, '', [['d', pub.agent], ['p', pub.agent], ['name', 'QA agent'], ['room', 'work'], ['room', 'cancel'], ['room', 'crash'], ['k', '0'], ['k', '9'], ['k', '5000'], ['k', '6000'], ['k', '7000'], ['k', '7'], ['k', '1111'], ['k', '20001'], ['jobs', 'both'], ['expiration', String(Math.floor(Date.now() / 1000) + 3600)]]);30 await publish('owner', 9007, '', [['h', 'work'], ['name', 'QA room'], ['visibility', 'open']]);31 await publish('owner', 9000, '', [['h', 'work'], ['p', pub.agent, 'member']]);32 for (const room of ['cancel', 'crash']) {33 await publish('owner', 9007, '', [['h', room], ['name', `QA ${room} room`], ['visibility', 'open']]);34 await publish('owner', 9000, '', [['h', room], ['p', pub.agent, 'member']]);35 }36 const acp = resolve(root, 'scripts/qa/buzz-runner-acp.mjs');37 runner = startRunner(acp, `${data}/queue.json`);38 const first = await publish('owner', 9, 'Please publish the preview.', [['h', 'work'], ['p', pub.agent]]);39 const request = await waitForEvent({kinds: [5000], '#h': ['work'], '#p': [pub.agent]});40 await waitForEvent({kinds:[7000], '#h':['work'], '#e':[request.id]}, event=>event.tags.some(tag=>tag[0]==='status'&&tag[1]==='processing'));41 const approval = await waitForEvent({kinds: [9], '#h': ['work'], '#p': [pub.owner]}, event => event.tags.some(tag => tag[0] === 'request' && tag[1] === 'approve'));42 assert(approval.tags.some(tag=>tag[0]==='expiration'&&Number(tag[1])>Math.floor(Date.now()/1000)), 'approval has no bounded expiry');43 await publish('owner', 7, '+', [['e', approval.id], ['p', pub.agent], ['k', '9'], ['h', 'work']]);44 const result = await waitForEvent({kinds: [6000], '#h': ['work'], '#e': [request.id]});45 const reply = await waitForEvent({kinds: [9], '#h': ['work'], '#e': [first.id], '#p': [pub.owner]}, event => event.pubkey === pub.agent && event.content === 'Preview published.');46 assert(result.content === 'Preview published.', `unexpected result: ${result.content}`);47 assert(reply.content === 'Preview published.', `unexpected reply: ${reply.content}`);48 await stopChild(runner);49 runner = startRunner(acp, `${data}/cancel-queue.json`, {FAKE_ACP_MODE: 'wait'}, 'cancel');50 const cancelMention = await publish('owner', 9, 'Start a task I will cancel.', [['h', 'cancel'], ['p', pub.agent]]);51 const cancelRequest = await waitForEvent({kinds: [5000], '#h': ['cancel'], '#p': [pub.agent]}, event => event.content.includes('Start a task'));52 const cancelApproval = await waitForEvent({kinds: [9], '#h': ['cancel'], '#p': [pub.owner]}, event => event.created_at >= cancelRequest.created_at && event.tags.some(tag => tag[0] === 'request' && tag[1] === 'approve'));53 await publish('owner', 9, '/cancel', [['h', 'cancel'], ['p', pub.agent]]);54 const canceled = await waitForEvent({kinds: [7000], '#h': ['cancel'], '#e': [cancelRequest.id]}, event => event.tags.some(tag => tag[0] === 'status' && tag[1] === 'error'));55 assert(canceled.content === 'Canceled', `unexpected cancellation: ${canceled.content}`);56 await stopChild(runner);57 const marker = `${data}/crash-once.marker`;58 runner = startRunner(acp, `${data}/crash-queue.json`, {FAKE_ACP_MODE: 'crash-once', FAKE_ACP_MARKER: marker}, 'crash');59 const crashedMention = await publish('owner', 9, 'Crash once.', [['h', 'crash'], ['p', pub.agent]]);60 const crashedRequest = await waitForEvent({kinds: [5000], '#h': ['crash'], '#p': [pub.agent]}, event => event.content.includes('Crash once'));61 const crashed = await waitForEvent({kinds: [7000], '#h': ['crash'], '#e': [crashedRequest.id]}, event => event.tags.some(tag => tag[0] === 'status' && tag[1] === 'error'));62 assert(crashed.content.includes('Agent stopped'), `missing crash report: ${crashed.content}`);63 const retryMention = await publish('owner', 9, 'Recover after crash.', [['h', 'crash'], ['p', pub.agent]]);64 const retryRequest = await waitForEvent({kinds: [5000], '#h': ['crash'], '#p': [pub.agent]}, event => event.content.includes('Recover after crash'));65 const retryApproval = await waitForEvent({kinds: [9], '#h': ['crash'], '#p': [pub.owner]}, event => event.created_at >= retryRequest.created_at && event.tags.some(tag => tag[0] === 'request' && tag[1] === 'approve'));66 await publish('owner', 7, '+', [['e', retryApproval.id], ['p', pub.agent], ['k', '9'], ['h', 'crash']]);67 const retryResult = await waitForEvent({kinds: [6000], '#h': ['crash'], '#e': [retryRequest.id]});68 assert(retryResult.content === 'Preview published.', `unexpected recovery result: ${retryResult.content}`);69 console.log(JSON.stringify({ok: true, request: request.id, approval: approval.id, result: result.id, reply: reply.id, canceled: cancelMention.id, crash: crashedMention.id, recovered: retryMention.id}));70} finally {71 if (runner) runner.kill('SIGTERM');72 if (relay) relay.kill('SIGTERM');73 await Promise.all([exitAfter(runner), exitAfter(relay)]);74 await rm(data, {recursive: true, force: true});75}7677function startRunner(acp, state, extra = {}, room = 'work') {78 return spawn(binary, ['agent', '--relay', ws, '--room', room, '--command', process.execPath, '--arg', acp, '--state', state, '--cwd', root], {env: {...process.env, TINY_AGENT_KEY: Buffer.from(keys.agent).toString('hex'), ...extra}, stdio: ['ignore', 'pipe', 'pipe']});79}80async function stopChild(child) {81 child.kill('SIGTERM');82 await exitAfter(child);83}8485function requireBinary() {86 const result = spawnSync(binary, ['version'], {encoding: 'utf8'});87 if (result.status !== 0) throw new Error(`build ${binary} first`);88}89function run(args) {90 const result = spawnSync(args[0], args.slice(1), {cwd: root, encoding: 'utf8'});91 if (result.status !== 0) throw new Error(result.stderr || result.stdout || `command failed: ${args.join(' ')}`);92}93async function waitFor(url) {94 for (let i = 0; i < 100; i++) {95 try { const response = await fetch(url); if (response.ok) return; } catch {}96 await new Promise(resolve => setTimeout(resolve, 50));97 }98 throw new Error(`timed out waiting for ${url}`);99}100async function signed(path, method, value, who) {101 const body = JSON.stringify(value);102 const auth = finalizeEvent({kind: 27235, created_at: Math.floor(Date.now() / 1000), tags: [['u', base + path], ['method', method], ['payload', createHash('sha256').update(body).digest('hex')], ['nonce', randomUUID()]], content: ''}, keys[who]);103 const response = await fetch(base + path, {method, headers: {'content-type': 'application/json', authorization: `Nostr ${Buffer.from(JSON.stringify(auth)).toString('base64')}`}, body});104 const result = await response.json();105 if (!response.ok || result.error || result.accepted === false) throw new Error(JSON.stringify(result));106 return result;107}108async function publish(who, kind, content, tags = []) {109 const event = finalizeEvent({kind, content, tags, created_at: Math.floor(Date.now() / 1000)}, keys[who]);110 await signed('/events', 'POST', event, who);111 return event;112}113async function query(filter) { return await signed('/query', 'POST', [filter], 'owner'); }114async function waitForEvent(filter, predicate = () => true) {115 for (let i = 0; i < 160; i++) {116 const events = await query(filter);117 const match = events.filter(predicate);118 if (match.length) return match.sort((a, b) => b.created_at - a.created_at || b.id.localeCompare(a.id))[0];119 await new Promise(resolve => setTimeout(resolve, 100));120 }121 throw new Error(`timed out waiting for event ${JSON.stringify(filter)}`);122}123function assert(condition, message) { if (!condition) throw new Error(message); }124function exitAfter(child) {125 if(!child || child.exitCode!==null || child.signalCode!==null) return Promise.resolve();126 return new Promise(resolve=>{const timeout=setTimeout(()=>child.kill('SIGKILL'),3000); child.once('exit',()=>{clearTimeout(timeout);resolve();});});127}