Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331// A tab multiplexes its live follows over one WebSocket. Older daemons fall// back to SSE; each transport resumes after the last fully handled batch.
export interface StreamEvent { type: string; sequence?: number; data: any;}
export interface Batch { generation: string; cursor: number; events: StreamEvent[]; snapshot?: any;}
const RETRY = { first: 500, most: 5000 };type Follow = { key: string; kind: 'session' | 'list'; background: boolean; sessionId?: string; onBatch: (batch: any) => void; onState: (connected: boolean) => void; after: Record<string, string>; subscribed: boolean; delay: number; retry?: ReturnType<typeof setTimeout>; fallback?: () => void;};
let nextKey = 0;let socket: WebSocket | null = null;let socketDelay = RETRY.first;let reconnect: ReturnType<typeof setTimeout> | undefined;let websocketUnavailable = false;const follows = new Map<string, Follow>();
function disconnected(follow: Follow) { clearTimeout(follow.retry); follow.retry = undefined; follow.subscribed = false; follow.onState(false);}
function closeSocket() { clearTimeout(reconnect); reconnect = undefined; const previous = socket; socket = null; previous?.close(); for (const follow of follows.values()) disconnected(follow);}
function active(follow: Follow) { return !document.hidden || follow.background;}
function hasActive() { return [...follows.values()].some(active);}
document.addEventListener('visibilitychange', () => { if (!hasActive()) closeSocket(); else { for (const follow of follows.values()) { if (!active(follow) && follow.subscribed) { if (socket?.readyState === WebSocket.OPEN) socket.send(JSON.stringify({ key: follow.key, type: 'stop' })); disconnected(follow); } } ensureSocket(); for (const follow of follows.values()) if (active(follow)) sendFollow(follow); }});
function startFallback(follow: Follow) { if (follow.fallback) return; const path = follow.kind === 'list' ? '/sessions?scope=roots' : `/sessions/${follow.sessionId}?tail=0`; follow.fallback = sseSubscribe(path, () => follow.after, (batch: any) => consume(follow, batch), follow.onState, follow.after, follow.background);}
function ensureSocket() { if (!hasActive()) return; if (websocketUnavailable) { for (const follow of follows.values()) if (active(follow)) startFallback(follow); return; } if (socket) return; clearTimeout(reconnect); reconnect = undefined; const scheme = location.protocol === 'https:' ? 'wss:' : 'ws:'; const ws = new WebSocket(`${scheme}//${location.host}/api/events`); socket = ws; let opened = false; let lastMessage = Date.now(); let heartbeat: ReturnType<typeof setInterval> | undefined; const timeout = setTimeout(() => ws.close(), 5000);
ws.onopen = () => { clearTimeout(timeout); if (socket !== ws || !hasActive()) { ws.close(); return; } opened = true; socketDelay = RETRY.first; heartbeat = setInterval(() => { if (Date.now() - lastMessage > 45000) ws.close(); else if (ws.readyState === WebSocket.OPEN) ws.send(JSON.stringify({ type: 'ping' })); }, 15000); for (const follow of follows.values()) sendFollow(follow); }; ws.onmessage = (event) => { if (socket !== ws) return; lastMessage = Date.now(); try { receive(String(event.data)); } catch { ws.close(); } }; ws.onerror = () => ws.close(); ws.onclose = () => { clearTimeout(timeout); clearInterval(heartbeat); if (socket !== ws) return; socket = null; for (const follow of follows.values()) disconnected(follow); if (!opened) { websocketUnavailable = true; ensureSocket(); } else if (hasActive()) { const wait = socketDelay; socketDelay = Math.min(RETRY.most, socketDelay * 2); reconnect = setTimeout(ensureSocket, wait); } };}
function sendFollow(follow: Follow) { if (!active(follow) || follow.subscribed || !socket || socket.readyState !== WebSocket.OPEN) return; clearTimeout(follow.retry); follow.retry = undefined; const value = follow.kind === 'list' ? { key: follow.key, type: 'sessions', scope: 'roots' } : { key: follow.key, type: 'session', session_id: follow.sessionId, tail: 0, ...follow.after }; socket.send(JSON.stringify(value)); follow.subscribed = true;}
function consume(follow: Follow, batch: any) { if (!Array.isArray(batch.events)) return; follow.onBatch(batch); if (follow.kind === 'session' && batch.generation) { follow.after = { after_generation: batch.generation, after_seq: String(batch.cursor) }; } follow.delay = RETRY.first; follow.onState(true);}
function receive(raw: string) { let value: any; try { value = JSON.parse(raw); } catch { return; } const follow = typeof value?.key === 'string' ? follows.get(value.key) : undefined; if (!follow) return; if (value.batch && typeof value.batch === 'object') { consume(follow, value.batch); } else if (value.end || value.failure) { disconnected(follow); if (!active(follow)) return; const wait = follow.delay; follow.delay = Math.min(RETRY.most, follow.delay * 2); follow.retry = setTimeout(() => { follow.retry = undefined; if (!follows.has(follow.key) || !active(follow)) return; ensureSocket(); sendFollow(follow); }, wait); }}
function subscribe(kind: Follow['kind'], sessionId: string | undefined, onBatch: (batch: any) => void, onState: (connected: boolean) => void, background = false): () => void { const key = `tab-${++nextKey}`; const follow: Follow = { key, kind, background, sessionId, onBatch, onState, after: {}, subscribed: false, delay: RETRY.first }; follows.set(key, follow); ensureSocket(); // onopen handles pending follows; an already-open socket needs this one now. sendFollow(follow); return () => { follows.delete(key); clearTimeout(follow.retry); follow.fallback?.(); if (follow.subscribed && socket?.readyState === WebSocket.OPEN) { socket.send(JSON.stringify({ key, type: 'stop' })); } if (!hasActive()) closeSocket(); };}
export function follow(sessionId: string, onBatch: (batch: Batch) => void, onState: (connected: boolean) => void): () => void { return subscribe('session', sessionId, onBatch, onState);}
export function followSessions(onEvents: (events: StreamEvent[]) => void, onState: (connected: boolean) => void, background = false): () => void { return subscribe('list', undefined, (batch) => onEvents(batch.events), onState, background);}
/** * Streams a tab holds at once. A browser keeps at most six connections to * the daemon, shared by every tab, and a stream holds one for as long as it * runs; past this many, a stream waits for one to end, and what it would * have carried is polled meanwhile (a chat reads its history every few * seconds while it is not streaming). Half are left for ordinary reads. */const MOST_STREAMS = 3;let streams = 0;const queued: (() => void)[] = [];
/** Waits for a stream's turn to connect; false when it was stopped first. */function take(signal: AbortSignal): Promise<boolean> { if (streams < MOST_STREAMS) { streams += 1; return Promise.resolve(true); } return new Promise((resolve) => { const turn = () => { signal.removeEventListener('abort', stop); resolve(true); }; const stop = () => { queued.splice(queued.indexOf(turn), 1); resolve(false); }; queued.push(turn); signal.addEventListener('abort', stop, { once: true }); });}
/** Hands a stream's connection to the next one waiting. */function release() { const next = queued.shift(); if (next) next(); else streams -= 1;}
/** * Reads server-sent events from a daemon resource until stopped, asking * again after an ordinary end or a failure with a growing wait. `resume` * names the query that continues after a handled batch. Returns the stop * function. * * Chat streams pause while hidden. The root-list follow can stay live * for background completion alerts. */function sseSubscribe<T>(path: string, resume: (batch: T) => Record<string, string> | null, onBatch: (batch: T) => void, onState: (connected: boolean) => void, initial: Record<string, string> = {}, background = false): () => void { const controller = new AbortController(); let after: Record<string, string> | null = initial; let delay = RETRY.first;
const run = async () => { while (!controller.signal.aborted) { if (document.hidden && !background) { await shown(controller.signal); continue; } const [base, given = ''] = path.split('?'); const query = new URLSearchParams(given); for (const [key, value] of Object.entries(after ?? {})) query.set(key, value); if (!(await take(controller.signal))) return; const pause = new AbortController(); const hide = () => document.hidden && !background && pause.abort(); document.addEventListener('visibilitychange', hide); try { const response = await fetch(`/api${base}?${query}`, { headers: { accept: 'text/event-stream' }, signal: AbortSignal.any([controller.signal, pause.signal]) }); if (!response.ok || !response.body) throw new Error(`stream answered ${response.status}`); onState(true); delay = RETRY.first; for await (const data of frames(response.body)) { const batch = JSON.parse(data) as T; onBatch(batch); after = resume(batch) ?? after; } } catch { if (controller.signal.aborted) return; } finally { document.removeEventListener('visibilitychange', hide); release(); } onState(false); // Hidden, it waits for the tab to show; an ordinary end is a restart // or a deletion, and either way it asks again. if (document.hidden && !background) continue; await new Promise((resolve) => setTimeout(resolve, delay)); delay = Math.min(RETRY.most, delay * 2); } }; run(); return () => controller.abort();}
/** Resolves once the tab shows, or the stream is stopped. */function shown(signal: AbortSignal): Promise<void> { return new Promise((resolve) => { const done = () => { if (document.hidden && !signal.aborted) return; document.removeEventListener('visibilitychange', done); signal.removeEventListener('abort', done); resolve(); }; document.addEventListener('visibilitychange', done); signal.addEventListener('abort', done, { once: true }); });}
/** The data of each server-sent event in a body: lines after `data:`, joined, up to a blank line. */async function* frames(body: ReadableStream<Uint8Array>): AsyncGenerator<string> { const reader = body.getReader(); const decoder = new TextDecoder(); let pending = ''; let data: string[] = []; for (;;) { const { value, done } = await reader.read(); if (done) return; pending += decoder.decode(value, { stream: true }); let end: number; while ((end = pending.indexOf('\n')) >= 0) { const line = pending.slice(0, end).replace(/\r$/, ''); pending = pending.slice(end + 1); if (line === '') { if (data.length) yield data.join('\n'); data = []; } else if (line.startsWith('data:')) data.push(line.slice(line.startsWith('data: ') ? 6 : 5)); } }}