Something went wrong. Try again.
small gleam coding and (not yet) persistent agent daemon with a detachable cli forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315// Browser lifecycle races cannot be reached through the daemon E2E suite.import assert from 'node:assert/strict';import { readFile } from 'node:fs/promises';import { stripTypeScriptTypes } from 'node:module';import test from 'node:test';
const source = stripTypeScriptTypes(await readFile(new URL('../src/lib/stream.ts', import.meta.url), 'utf8'));let imports = 0;
async function browser(t) { const saved = new Map(); const sockets = []; const requests = []; const timers = new Map(); const listeners = new Set(); let nextTimer = 0; const document = { hidden: false, addEventListener(name, callback) { assert.equal(name, 'visibilitychange'); listeners.add(callback); }, removeEventListener(name, callback) { assert.equal(name, 'visibilitychange'); listeners.delete(callback); }, }; class WebSocket { static CONNECTING = 0; static OPEN = 1; static CLOSED = 3; readyState = WebSocket.CONNECTING; sent = []; closes = 0; constructor(url) { this.url = url; sockets.push(this); } send(value) { assert.equal(this.readyState, WebSocket.OPEN); this.sent.push(JSON.parse(value)); } open() { this.readyState = WebSocket.OPEN; this.onopen?.(); } close() { this.closes++; this.readyState = WebSocket.CLOSED; this.onclose?.(); } message(value) { this.onmessage?.({ data: JSON.stringify(value) }); } } const install = (name, value) => { saved.set(name, Object.getOwnPropertyDescriptor(globalThis, name)); Object.defineProperty(globalThis, name, { configurable: true, writable: true, value }); }; t.after(() => { for (const [name, descriptor] of saved) { if (descriptor) Object.defineProperty(globalThis, name, descriptor); else delete globalThis[name]; } }); const schedule = (kind) => (callback, delay) => { const id = ++nextTimer; timers.set(id, { kind, callback, delay }); return id; }; install('document', document); install('location', { protocol: 'https:', host: 'albedo.example' }); install('WebSocket', WebSocket); install('fetch', (url, options) => { requests.push({ url, ...options }); return new Promise((_, reject) => { const abort = () => reject(options.signal.reason); if (options.signal.aborted) abort(); else options.signal.addEventListener('abort', abort, { once: true }); }); }); install('setTimeout', schedule('timeout')); install('setInterval', schedule('interval')); install('clearTimeout', (id) => timers.delete(id)); install('clearInterval', (id) => timers.delete(id)); const api = await import(`data:text/javascript;base64,${Buffer.from(source).toString('base64')}#${++imports}`); return { api, sockets, timers, requests, visibility(hidden) { document.hidden = hidden; for (const callback of [...listeners]) callback(); }, fireTimeout(delay) { const matches = [...timers].filter(([, timer]) => timer.kind === 'timeout' && timer.delay === delay); assert.equal(matches.length, 1, `expected one ${delay}ms timeout`); const [id, timer] = matches[0]; timers.delete(id); timer.callback(); }, };}
function follow(api, id) { const batches = []; const states = []; const stop = api.follow(id, (batch) => batches.push(batch), (state) => states.push(state)); return { batches, states, stop };}
function batch(cursor) { return { generation: 'saved-generation', cursor, events: [] };}
test('stopping a pending follow sends nothing and excludes it when the socket opens', async (t) => { const { api, sockets } = await browser(t); const stopped = follow(api, 'stopped'); follow(api, 'live'); const ws = sockets[0]; assert.equal(ws.url, 'wss://albedo.example/api/events'); stopped.stop(); assert.deepEqual(ws.sent, []); ws.open(); assert.deepEqual(ws.sent.map((command) => command.session_id), ['live']); assert.equal(sockets.length, 1);});
test('adding a follow to an open socket subscribes exactly once', async (t) => { const { api, sockets } = await browser(t); follow(api, 'first'); const ws = sockets[0]; ws.open(); follow(api, 'second'); api.followSessions(() => {}, () => {}); assert.deepEqual(ws.sent.map((command) => command.type), ['session', 'session', 'sessions']); assert.deepEqual(ws.sent.slice(0, 2).map((command) => command.session_id), ['first', 'second']); assert.equal(ws.sent[2].scope, 'roots'); assert.equal(new Set(ws.sent.map((command) => command.key)).size, 3); assert.equal(sockets.length, 1);});
test('reconnecting resumes every live follow with its handled cursor and excludes stopped keys', async (t) => { const { api, sockets, fireTimeout } = await browser(t); follow(api, 'alpha'); follow(api, 'beta'); const stopped = follow(api, 'stopped'); const ws = sockets[0]; ws.open(); const [alpha, beta, removed] = ws.sent; ws.message({ key: alpha.key, batch: batch(11) }); ws.message({ key: beta.key, batch: batch(23) }); stopped.stop(); assert.deepEqual(ws.sent.at(-1), { key: removed.key, type: 'stop' }); ws.close(); fireTimeout(500); assert.equal(sockets.length, 2); sockets[1].open(); assert.deepEqual(sockets[1].sent, [ { ...alpha, after_generation: 'saved-generation', after_seq: '11' }, { ...beta, after_generation: 'saved-generation', after_seq: '23' }, ]);});
test('hiding a connecting socket closes it and ignores its stale open callback', async (t) => { const { api, sockets, timers, visibility } = await browser(t); follow(api, 'live'); const stale = sockets[0]; visibility(true); assert.equal(stale.closes, 1); assert.equal(timers.size, 0); visibility(false); assert.equal(sockets.length, 2); stale.open(); assert.deepEqual(stale.sent, []); assert.equal(stale.closes, 2); sockets[1].open(); assert.deepEqual(sockets[1].sent.map((command) => command.session_id), ['live']);});
test('hiding an open socket clears heartbeat and retry timers and resumes on return', async (t) => { const { api, sockets, timers, visibility } = await browser(t); const observed = follow(api, 'live'); const ws = sockets[0]; ws.open(); const command = ws.sent[0]; ws.message({ key: command.key, batch: batch(31) }); ws.message({ key: command.key, end: true }); assert.equal(timers.size, 2); visibility(true); assert.equal(ws.closes, 1); assert.equal(timers.size, 0); assert.equal(observed.states.at(-1), false); visibility(false); sockets[1].open(); assert.deepEqual(sockets[1].sent, [ { ...command, after_generation: 'saved-generation', after_seq: '31' }, ]);});
for (const outcome of [{ end: true }, { failure: { code: 'unavailable' } }]) { test(`a per-key ${Object.keys(outcome)[0]} retries only that follow and a batch restores connected state`, async (t) => { const { api, sockets, fireTimeout } = await browser(t); const alpha = follow(api, 'alpha'); const beta = follow(api, 'beta'); const ws = sockets[0]; ws.open(); const [first, second] = ws.sent; ws.message({ key: first.key, batch: batch(7) }); ws.message({ key: second.key, batch: batch(9) }); ws.message({ key: first.key, ...outcome }); assert.deepEqual(alpha.states, [true, false]); assert.deepEqual(beta.states, [true]); fireTimeout(500); assert.equal(sockets.length, 1); assert.deepEqual(ws.sent.slice(2), [ { ...first, after_generation: 'saved-generation', after_seq: '7' }, ]); ws.message({ key: first.key, batch: batch(8) }); assert.deepEqual(alpha.states, [true, false, true]); assert.deepEqual(alpha.batches, [batch(7), batch(8)]); assert.deepEqual(beta.states, [true]); });}
test('an initial WebSocket failure falls back once per follow and SSE pauses and stops without socket commands', async (t) => { const { api, sockets, requests, visibility } = await browser(t); const alpha = follow(api, 'alpha'); const beta = follow(api, 'beta'); let stopList = () => {}; try { const ws = sockets[0]; ws.close(); await new Promise(setImmediate); assert.deepEqual(requests.map((request) => request.url), [ '/api/sessions/alpha?tail=0', '/api/sessions/beta?tail=0', ]); stopList = api.followSessions(() => {}, () => {}); await new Promise(setImmediate); assert.equal(requests.length, 3); assert.equal(requests[2].url, '/api/sessions?scope=roots'); assert.ok(requests.every((request) => request.headers.accept === 'text/event-stream')); assert.ok(requests.every((request) => !request.signal.aborted)); assert.equal(sockets.length, 1);
visibility(true); assert.ok(requests.every((request) => request.signal.aborted)); await new Promise(setImmediate); assert.equal(requests.length, 3); visibility(false); await new Promise(setImmediate); assert.deepEqual(requests.slice(3).map((request) => request.url), requests.slice(0, 3).map((request) => request.url)); assert.equal(requests.length, 6); assert.ok(requests.slice(3).every((request) => !request.signal.aborted));
alpha.stop(); assert.equal(requests[3].signal.aborted, true); assert.equal(requests[4].signal.aborted, false); assert.equal(requests[5].signal.aborted, false); await new Promise(setImmediate); assert.equal(requests.length, 6); assert.deepEqual(ws.sent, []); assert.equal(sockets.length, 1); } finally { alpha.stop(); beta.stop(); stopList(); await new Promise(setImmediate); }});
test('background list keeps the socket alive while chat follows stop and resume with their cursors', async (t) => { const { api, sockets, visibility } = await browser(t); const chat = follow(api, 'chat'); const events = []; const stop = api.followSessions((batch) => events.push(...batch), () => {}, true); const ws = sockets[0]; ws.open(); const [session, list] = ws.sent; ws.message({ key: session.key, batch: batch(12) }); visibility(true); assert.equal(ws.closes, 0); assert.deepEqual(ws.sent.at(-1), { key: session.key, type: 'stop' }); ws.message({ key: list.key, batch: { events: [{ type: 'mail', data: { mail_id: 'reply' } }] } }); assert.equal(events[0].data.mail_id, 'reply'); visibility(false); assert.deepEqual(ws.sent.at(-1), { ...session, after_generation: 'saved-generation', after_seq: '12' }); chat.stop(); stop(); assert.equal(ws.closes, 1);});
test('a background list can connect and reconnect while hidden, including SSE fallback', async (t) => { const { api, sockets, requests, visibility, fireTimeout } = await browser(t); visibility(true); const stop = api.followSessions(() => {}, () => {}, true); const ws = sockets[0]; ws.open(); assert.equal(ws.sent[0].type, 'sessions'); ws.close(); fireTimeout(500); assert.equal(sockets.length, 2); sockets[1].close(); await new Promise(setImmediate); assert.equal(requests.length, 1); assert.equal(requests[0].url, '/api/sessions?scope=roots'); assert.equal(requests[0].signal.aborted, false); visibility(false); visibility(true); assert.equal(requests[0].signal.aborted, false); stop(); assert.equal(requests[0].signal.aborted, true); await new Promise(setImmediate);});