| 1 | #!/usr/bin/env node |
| 2 | // End-to-end optional-agent check. All state, keys and processes are local. |
| 3 | import {finalizeEvent, getPublicKey} from 'nostr-tools/pure' ; |
| 4 | import {createHash, randomUUID} from 'node:crypto' ; |
| 5 | import {mkdtemp, rm} from 'node:fs/promises' ; |
| 6 | import {createServer} from 'node:net' ; |
| 7 | import {tmpdir} from 'node:os' ; |
| 8 | import {dirname, resolve} from 'node:path' ; |
| 9 | import {spawn, spawnSync} from 'node:child_process' ; |
| 10 | |
| 11 | const root = resolve(dirname(new URL(import.meta.url).pathname), '../..' ); |
| 12 | const binary = process.env.TINY_QA_BINARY || '/tmp/tiny-buzz-qa' ; |
| 13 | const reservation = createServer(); |
| 14 | await new Promise((resolve,reject)=>{reservation.once('error' ,reject);reservation.listen(0,'127.0.0.1' ,resolve);}); |
| 15 | const port = reservation.address().port; |
| 16 | await new Promise(resolve=>reservation.close(resolve)); |
| 17 | const base = `http://127.0.0.1:${port}` ; |
| 18 | const ws = `ws://127.0.0.1:${port}` ; |
| 19 | const keys = Object.fromEntries(['owner' , 'agent' ].map((name, i) => [name, Uint8Array.from(Buffer.from((i + 1).toString(16).padStart(64, '0' ), 'hex' ))])); |
| 20 | const pub = Object.fromEntries(Object.entries(keys).map(([name, key]) => [name, getPublicKey(key)])); |
| 21 | const data = await mkdtemp(`${tmpdir()}/tiny-buzz-runner-` ); |
| 22 | let relay, runner; |
| 23 | try { |
| 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 | } |
| 76 | |
| 77 | function 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 | } |
| 80 | async function stopChild(child) { |
| 81 | child.kill('SIGTERM' ); |
| 82 | await exitAfter(child); |
| 83 | } |
| 84 | |
| 85 | function requireBinary() { |
| 86 | const result = spawnSync(binary, ['version' ], {encoding: 'utf8' }); |
| 87 | if (result.status !== 0) throw new Error(`build ${binary} first` ); |
| 88 | } |
| 89 | function 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 | } |
| 93 | async 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 | } |
| 100 | async 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 | } |
| 108 | async 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 | } |
| 113 | async function query(filter) { return await signed('/query' , 'POST' , [filter], 'owner' ); } |
| 114 | async 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 | } |
| 123 | function assert(condition, message) { if (!condition) throw new Error(message); } |
| 124 | function 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 | } |