From 77efe1391b94b444e565715992cc41ef06d3027c Mon Sep 17 00:00:00 2001 From: Suleiman Shahbari Date: Sun, 5 Jul 2026 22:47:32 +0300 Subject: [PATCH] feat(framework): hosted run relay to watch one run from multiple browsers (#230) --- .changeset/hosted-run-relay.md | 9 + packages/framework/src/cli.test.ts | 12 + packages/framework/src/cli.ts | 52 +++- packages/framework/src/dashboard/page.ts | 6 +- .../framework/src/dashboard/server.test.ts | 4 +- packages/framework/src/dashboard/server.ts | 36 +-- packages/framework/src/dashboard/sse.ts | 41 ++++ packages/framework/src/index.ts | 7 + packages/framework/src/relay.test.ts | 137 +++++++++++ packages/framework/src/relay.ts | 227 ++++++++++++++++++ 10 files changed, 491 insertions(+), 40 deletions(-) create mode 100644 .changeset/hosted-run-relay.md create mode 100644 packages/framework/src/dashboard/sse.ts create mode 100644 packages/framework/src/relay.test.ts create mode 100644 packages/framework/src/relay.ts diff --git a/.changeset/hosted-run-relay.md b/.changeset/hosted-run-relay.md new file mode 100644 index 0000000..ed5062e --- /dev/null +++ b/.changeset/hosted-run-relay.md @@ -0,0 +1,9 @@ +--- +'@gemstack/framework': minor +--- + +feat(framework): hosted run relay — watch one run from multiple browsers (#230) + +The first slice toward shared team sessions: a run can now be watched live from more than one machine. `framework relay` hosts a relay; a run started with `framework "..." --share ` publishes its event stream to it and prints a shareable URL. Anyone who opens that URL gets the same dashboard over SSE, replaying the run's full history and then following live — so two teammates watch one build together. + +Reuses the existing dashboard: the SSE serving is factored into a shared helper and the page's stream/stop paths are now relative so they resolve both on the localhost dashboard and under the relay's `/r//`. New exports: `startRelay`, `relayPublisher`. Deliberately unauthenticated — accounts, teams, RBAC, and authorized steering layer on later; the relay only projects the stream, it never runs an agent. diff --git a/packages/framework/src/cli.test.ts b/packages/framework/src/cli.test.ts index 379092e..0b7bc94 100644 --- a/packages/framework/src/cli.test.ts +++ b/packages/framework/src/cli.test.ts @@ -95,6 +95,18 @@ test('parseArgs reads --sandbox and rejects an unknown value (#229)', () => { assert.match(parseArgs(['--sandbox', 'vm', 'x']).error!, /invalid --sandbox/) }) +test('parseArgs reads the relay subcommand and --share (#230)', () => { + const relay = parseArgs(['relay', '--port', '5000']) + assert.equal(relay.relayServe, true) + assert.equal(relay.intent, '') // 'relay' is a subcommand, not an intent word + assert.equal(relay.port, 5000) + const share = parseArgs(['--share', 'http://host:4488', 'a', 'blog']) + assert.equal(share.share, 'http://host:4488') + assert.equal(share.intent, 'a blog') + assert.equal(parseArgs(['x']).relayServe, false) + assert.equal(parseArgs(['x']).share, undefined) +}) + test('workspaceSummary describes an empty vs an existing workspace', async () => { const empty = await mkdtemp(join(tmpdir(), 'framework-ws-')) try { diff --git a/packages/framework/src/cli.ts b/packages/framework/src/cli.ts index 26f327e..a275790 100644 --- a/packages/framework/src/cli.ts +++ b/packages/framework/src/cli.ts @@ -14,6 +14,8 @@ import { import { ClaudeCodeDriver, type ClaudeCodeDriverOptions, type Driver, type DriverSession, type PermissionMode } from './driver/index.js' import { hostExecutor } from './host-exec.js' import { startDashboard, type Dashboard } from './dashboard/index.js' +import { startRelay, relayPublisher, type RelayPublisher } from './relay.js' +import { randomUUID } from 'node:crypto' import { formatFrameworkEvent, CLAUDE_CODE_SESSION_LINK, type FrameworkEvent } from './events.js' import { runFramework, @@ -74,6 +76,7 @@ Usage: framework [intent...] Build what you describe, from scratch. framework --fake Run the offline demo (no CLI, no model, deterministic). framework doctor Check prerequisites (Claude Code installed, etc.). + framework relay Host a run relay so teammates can watch a run (#230). Options: --fake Use the fake driver + scripted run (offline / CI). @@ -115,8 +118,10 @@ Options: --cf-project Cloudflare Pages project name (for a Pages deploy). --dokploy-url Dokploy instance URL (required for --deploy dokploy). --dokploy-app Dokploy application id (required for --deploy dokploy). - --port Dashboard port (default: 4477). + --port Dashboard port (default: 4477); with the relay, the relay port (4488). --no-dashboard Do not start the localhost dashboard. + --share Publish this run to a relay (from "framework relay") so + teammates can watch it live; prints the shareable URL. --resume Reopen the last run's dashboard from .framework/ in --cwd (read-only replay; no new agent run). Survives a restart. --no-persist Do not write the orchestration state to .framework/. @@ -164,6 +169,8 @@ export interface CliOptions { sandbox?: 'local' | 'docker' | undefined port?: number dashboard: boolean + relayServe: boolean + share?: string | undefined composeExtensions: boolean sessionLink?: string | undefined permissionMode?: PermissionMode | undefined @@ -187,6 +194,7 @@ export function parseArgs(argv: string[]): CliOptions { autopilot: false, technical: false, dashboard: true, + relayServe: false, composeExtensions: false, skipPermissions: false, resume: false, @@ -292,6 +300,9 @@ export function parseArgs(argv: string[]): CliOptions { else opts.sandbox = where break } + case '--share': + opts.share = argv[++i] + break case '--session-link': opts.sessionLink = argv[++i] break @@ -318,10 +329,13 @@ export function parseArgs(argv: string[]): CliOptions { else words.push(arg) } } - // `framework doctor` is a subcommand, not an intent. + // `framework doctor` / `framework relay` are subcommands, not an intent. if (words[0] === 'doctor') { opts.doctor = true words.shift() + } else if (words[0] === 'relay') { + opts.relayServe = true + words.shift() } opts.intent = words.join(' ').trim() return opts @@ -477,6 +491,10 @@ export async function runCli(argv: string[], io: CliIO = defaultIO): Promise + io.err(`relay publish failed (${err instanceof Error ? err.message : String(err)})`), + ) + io.out(`◆ shared run: ${publisher.url}`) + } + const onEvent = (event: FrameworkEvent) => { io.out(formatFrameworkEvent(event)) dashboard?.push(event) void store?.append(event) + publisher?.publish(event) } // Discover installed `framework-*` capability packages (#190) from the signals @@ -720,9 +749,28 @@ export async function runCli(argv: string[], io: CliIO = defaultIO): Promise`) and watch it live with full + * history replay. Runs until interrupted. Unauthenticated by design — anyone with + * a run URL can watch; accounts/teams/steering come later. + */ +async function runRelayServer(opts: CliOptions, io: CliIO): Promise { + const relay = await startRelay(opts.port !== undefined ? { port: opts.port } : {}) + io.out(`◆ relay listening at ${relay.url}`) + io.out(` Runs published with \`framework "..." --share ${relay.url}\` are watchable at ${relay.url}/r//`) + io.out(` Press Ctrl+C to stop.`) + await waitForInterrupt() + await relay.close() + return 0 +} + /** * Reopen the last run from its persisted `.framework/` log and replay it into a * fresh dashboard (#211). No agent runs; the dashboard rehydrates from the saved diff --git a/packages/framework/src/dashboard/page.ts b/packages/framework/src/dashboard/page.ts index d7cbf10..f118315 100644 --- a/packages/framework/src/dashboard/page.ts +++ b/packages/framework/src/dashboard/page.ts @@ -2,7 +2,7 @@ import { CLAUDE_CODE_SESSION_LINK } from '../events.js' /** * The single self-contained dashboard page: HTML + inline CSS + inline JS, no - * assets, no build step. The client opens an `EventSource` to `/events` and + * assets, no build step. The client opens an `EventSource` to `events` and * projects the {@link import('../events.js').FrameworkEvent} stream into panels * that foreground the orchestration (stack rationale, loop status, decisions) * beside a tail of the wrapped agent's own activity. @@ -261,11 +261,11 @@ function stopRun() { const btn = $('stop'); btn.disabled = true; btn.textContent = 'stopping\\u2026'; - fetch('/stop', { method: 'POST' }).catch(() => {}); + fetch('stop', { method: 'POST' }).catch(() => {}); } function esc(s) { const d = document.createElement('div'); d.textContent = String(s); return d.innerHTML; } $('stop').addEventListener('click', stopRun); -const src = new EventSource('/events'); +const src = new EventSource('events'); src.onmessage = ev => { try { onEvent(JSON.parse(ev.data)); } catch {} }; src.onerror = () => { $('status').textContent = '\\u25cb offline'; }; ` diff --git a/packages/framework/src/dashboard/server.test.ts b/packages/framework/src/dashboard/server.test.ts index 97c4d44..2c39d28 100644 --- a/packages/framework/src/dashboard/server.test.ts +++ b/packages/framework/src/dashboard/server.test.ts @@ -59,7 +59,9 @@ test('dashboard serves the HTML page with the title', async () => { const { status, body } = await fetchText(dash.url + '/') assert.equal(status, 200) assert.match(body, /My Framework/) - assert.match(body, /new EventSource\('\/events'\)/) + // Relative path so it resolves against the page's base URL — /events on the + // localhost dashboard, /r//events when re-served by the relay (#230). + assert.match(body, /new EventSource\('events'\)/) // The Modes panel + its renderer ship in the page (#272), hidden until a modes event. assert.match(body, /id="modes-panel" hidden/) assert.match(body, /function renderModes/) diff --git a/packages/framework/src/dashboard/server.ts b/packages/framework/src/dashboard/server.ts index 9121aa1..d73bd6a 100644 --- a/packages/framework/src/dashboard/server.ts +++ b/packages/framework/src/dashboard/server.ts @@ -3,6 +3,7 @@ import type { AddressInfo } from 'node:net' import { EventStream } from '@gemstack/ai-autopilot' import type { FrameworkEvent } from '../events.js' import { dashboardHtml } from './page.js' +import { serveSSE } from './sse.js' /** Options for {@link startDashboard}. */ export interface DashboardOptions { @@ -83,7 +84,7 @@ function handle( return } if (url === '/events') { - streamEvents(req, res, stream, clients) + serveSSE(req, res, stream, clients) return } if (url === '/stop') { @@ -109,39 +110,6 @@ function handle( res.end('not found') } -function streamEvents( - req: IncomingMessage, - res: ServerResponse, - stream: EventStream, - clients: Set, -): void { - res.writeHead(200, { - 'content-type': 'text/event-stream', - 'cache-control': 'no-cache', - connection: 'keep-alive', - }) - clients.add(res) - - const send = (event: FrameworkEvent) => res.write(`data: ${JSON.stringify(event)}\n\n`) - // Replay history, then follow live. A fresh iterator gives this client its own - // cursor, so a late browser still sees the whole run from the start. - void (async () => { - try { - for await (const event of stream[Symbol.asyncIterator]()) send(event) - } catch { - // client went away - } finally { - clients.delete(res) - res.end() - } - })() - - req.on('close', () => { - clients.delete(res) - res.end() - }) -} - function closeServer( server: Server, clients: Set, diff --git a/packages/framework/src/dashboard/sse.ts b/packages/framework/src/dashboard/sse.ts new file mode 100644 index 0000000..6d77c43 --- /dev/null +++ b/packages/framework/src/dashboard/sse.ts @@ -0,0 +1,41 @@ +import type { IncomingMessage, ServerResponse } from 'node:http' +import type { EventStream } from '@gemstack/ai-autopilot' +import type { FrameworkEvent } from '../events.js' + +/** + * Serve one client the {@link FrameworkEvent} stream over Server-Sent Events: + * replay the whole run's history, then follow live. A fresh async iterator gives + * this client its own cursor from the start, so a late browser still sees the run + * from the beginning. Shared by the localhost dashboard and the hosted relay (#230) + * so both project the identical stream. + */ +export function serveSSE( + req: IncomingMessage, + res: ServerResponse, + stream: EventStream, + clients: Set, +): void { + res.writeHead(200, { + 'content-type': 'text/event-stream', + 'cache-control': 'no-cache', + connection: 'keep-alive', + }) + clients.add(res) + + const send = (event: FrameworkEvent) => res.write(`data: ${JSON.stringify(event)}\n\n`) + void (async () => { + try { + for await (const event of stream[Symbol.asyncIterator]()) send(event) + } catch { + // client went away + } finally { + clients.delete(res) + res.end() + } + })() + + req.on('close', () => { + clients.delete(res) + res.end() + }) +} diff --git a/packages/framework/src/index.ts b/packages/framework/src/index.ts index 145890c..612ab8e 100644 --- a/packages/framework/src/index.ts +++ b/packages/framework/src/index.ts @@ -58,6 +58,13 @@ export { type AppPreview, } from './run.js' export { snapshotWorkspace, SANDBOX_IGNORE, type SnapshotOptions } from './sandbox.js' +export { + startRelay, + relayPublisher, + type Relay, + type RelayOptions, + type RelayPublisher, +} from './relay.js' export { discoverExtensions, readProjectSignals, diff --git a/packages/framework/src/relay.test.ts b/packages/framework/src/relay.test.ts new file mode 100644 index 0000000..1fb2551 --- /dev/null +++ b/packages/framework/src/relay.test.ts @@ -0,0 +1,137 @@ +import { strict as assert } from 'node:assert' +import { test } from 'node:test' +import { get, request } from 'node:http' +import { startRelay, relayPublisher } from './relay.js' +import type { FrameworkEvent } from './events.js' + +function fetchFull(url: string): Promise<{ status: number; headers: Record; body: string }> { + return new Promise((resolvePromise, rejectPromise) => { + get(url, res => { + let body = '' + res.on('data', c => (body += c)) + res.on('end', () => resolvePromise({ status: res.statusCode ?? 0, headers: res.headers, body })) + }).on('error', rejectPromise) + }) +} + +function send(url: string, method: string, body?: string): Promise<{ status: number; body: string }> { + return new Promise((resolvePromise, rejectPromise) => { + const req = request(url, { method }, res => { + let b = '' + res.on('data', c => (b += c)) + res.on('end', () => resolvePromise({ status: res.statusCode ?? 0, body: b })) + }) + req.on('error', rejectPromise) + if (body !== undefined) req.write(body) + req.end() + }) +} + +// Read SSE `data:` frames until `count` have arrived, then disconnect. +function readSse(url: string, count: number): Promise { + return new Promise((resolvePromise, rejectPromise) => { + const req = get(url, res => { + let buffer = '' + const collected: FrameworkEvent[] = [] + res.on('data', chunk => { + buffer += chunk + let nl: number + while ((nl = buffer.indexOf('\n\n')) !== -1) { + const frame = buffer.slice(0, nl) + buffer = buffer.slice(nl + 2) + const line = frame.split('\n').find(l => l.startsWith('data: ')) + if (line) collected.push(JSON.parse(line.slice(6)) as FrameworkEvent) + if (collected.length >= count) { + req.destroy() + resolvePromise(collected) + return + } + } + }) + }) + req.on('error', rejectPromise) + }) +} + +const local = { host: '127.0.0.1', port: 0 as const } + +test('relay re-serves one run to two browsers, each replaying full history (#230)', async () => { + const relay = await startRelay(local) + try { + relay.ingest('run-1', { kind: 'log', message: 'a' }) + relay.ingest('run-1', { kind: 'log', message: 'b' }) + relay.ingest('run-1', { kind: 'log', message: 'c' }) + + // Two independent viewers connect after the fact; both replay the whole run. + const [one, two] = await Promise.all([ + readSse(`${relay.url}/r/run-1/events`, 3), + readSse(`${relay.url}/r/run-1/events`, 3), + ]) + assert.deepEqual(one.map(e => (e as { message: string }).message), ['a', 'b', 'c']) + assert.deepEqual(two.map(e => (e as { message: string }).message), ['a', 'b', 'c']) + } finally { + await relay.close() + } +}) + +test('relayPublisher POSTs a run to the relay, which re-serves it live', async () => { + const relay = await startRelay(local) + try { + // A viewer connects first; then the run publishes over HTTP. + const seen = readSse(`${relay.url}/r/pub/events`, 2) + const pub = relayPublisher(relay.url, 'pub') + assert.equal(pub.url, `${relay.url}/r/pub/`) + pub.publish({ kind: 'log', message: 'from-cli-1' }) + pub.publish({ kind: 'end', ok: true } as FrameworkEvent) + await pub.flush() + const events = await seen + assert.equal((events[0] as { message: string }).message, 'from-cli-1') + assert.equal(events[1]!.kind, 'end') + } finally { + await relay.close() + } +}) + +test('runs are isolated by id', async () => { + const relay = await startRelay(local) + try { + relay.ingest('run-a', { kind: 'log', message: 'only-a' }) + relay.ingest('run-b', { kind: 'log', message: 'only-b' }) + const a = await readSse(`${relay.url}/r/run-a/events`, 1) + assert.deepEqual( + a.map(e => (e as { message: string }).message), + ['only-a'], + ) + assert.deepEqual(relay.runIds().sort(), ['run-a', 'run-b']) + } finally { + await relay.close() + } +}) + +test('GET /r/:id redirects to the trailing-slash page; the page loads relative SSE', async () => { + const relay = await startRelay(local) + try { + const redirect = await fetchFull(`${relay.url}/r/xyz`) + assert.equal(redirect.status, 302) + assert.equal(redirect.headers.location, '/r/xyz/') + + const page = await fetchFull(`${relay.url}/r/xyz/`) + assert.equal(page.status, 200) + assert.match(page.body, /new EventSource\('events'\)/) // resolves under /r/xyz/ + } finally { + await relay.close() + } +}) + +test('healthz is 200; a bad publish is rejected without crashing the run', async () => { + const relay = await startRelay(local) + try { + assert.equal((await fetchFull(`${relay.url}/healthz`)).status, 200) + assert.equal((await send(`${relay.url}/r/x/publish`, 'GET')).status, 405) // must POST + assert.equal((await send(`${relay.url}/r/x/publish`, 'POST', 'not json')).status, 400) + const ok = await send(`${relay.url}/r/x/publish`, 'POST', JSON.stringify({ kind: 'log', message: 'ok' })) + assert.equal(ok.status, 202) + } finally { + await relay.close() + } +}) diff --git a/packages/framework/src/relay.ts b/packages/framework/src/relay.ts new file mode 100644 index 0000000..d22e0dd --- /dev/null +++ b/packages/framework/src/relay.ts @@ -0,0 +1,227 @@ +import { createServer, type IncomingMessage, type Server, type ServerResponse } from 'node:http' +import type { AddressInfo } from 'node:net' +import { EventStream } from '@gemstack/ai-autopilot' +import type { FrameworkEvent } from './events.js' +import { dashboardHtml } from './dashboard/page.js' +import { serveSSE } from './dashboard/sse.js' + +/** + * The hosted run relay (#230): the first slice toward shared team sessions. It + * ingests a run's {@link FrameworkEvent} stream over HTTP and re-serves the exact + * same dashboard (SSE + history replay) to N remote browsers, keyed by run id. So + * two people on different machines open one run URL and both watch it live. + * + * Deliberately unauthenticated: anyone with a run's URL can watch it. Accounts, + * teams, RBAC, and authorized steering layer on later (via vike-auth/-rbac). The + * relay only projects the stream — it never runs an agent. + * + * Endpoints (per run id): + * - `POST /r/:id/publish` — ingest one event (JSON object) or a batch (JSON array) + * - `GET /r/:id/` — the dashboard page (read-only: no Stop button) + * - `GET /r/:id/events` — the SSE stream (replays history, then follows live) + * - `GET /r/:id` — redirects to `/r/:id/` so the page's relative paths resolve + * - `GET /healthz` — liveness probe for the host + */ +export interface RelayOptions { + /** Port to bind. Default `4488`; pass `0` for an ephemeral port. */ + port?: number + /** + * Host to bind. Default `0.0.0.0` — the relay exists to be reached from other + * machines. Bind `127.0.0.1` to keep it local (e.g. tests). + */ + host?: string + /** Page title. Default `"The Framework"`. */ + title?: string + /** Max bytes accepted per publish request body. Default 256 KiB. */ + maxBodyBytes?: number +} + +/** A running relay. Ingest events programmatically or over HTTP; browsers watch by run id. */ +export interface Relay { + /** The base URL of the relay (e.g. `http://0.0.0.0:4488`). */ + readonly url: string + /** The viewer URL for a run id (`/r//`). */ + viewerUrl(runId: string): string + /** Push one event into a run's stream, creating the run on first use. */ + ingest(runId: string, event: FrameworkEvent): void + /** The run ids seen so far. */ + runIds(): string[] + /** Close every stream and stop the server. Idempotent. */ + close(): Promise +} + +interface Run { + stream: EventStream + clients: Set +} + +const RUN_PATH = /^\/r\/([^/]+)(\/[^?]*)?/ + +/** Start the hosted run relay. See {@link Relay}. */ +export function startRelay(opts: RelayOptions = {}): Promise { + const host = opts.host ?? '0.0.0.0' + const port = opts.port ?? 4488 + const title = opts.title ?? 'The Framework' + const maxBody = opts.maxBodyBytes ?? 256 * 1024 + const runs = new Map() + + const run = (id: string): Run => { + let r = runs.get(id) + if (!r) { + r = { stream: new EventStream(), clients: new Set() } + runs.set(id, r) + } + return r + } + + const server = createServer((req, res) => handle(req, res, { runs, run, title, maxBody })) + + return new Promise((resolvePromise, rejectPromise) => { + server.once('error', rejectPromise) + server.listen(port, host, () => { + server.removeListener('error', rejectPromise) + const address = server.address() as AddressInfo + const url = `http://${host}:${address.port}` + resolvePromise({ + url, + viewerUrl: id => `${url}/r/${encodeURIComponent(id)}/`, + ingest: (id, event) => run(id).stream.push(event), + runIds: () => [...runs.keys()], + close: () => closeRelay(server, runs), + }) + }) + }) +} + +interface HandleCtx { + runs: Map + run: (id: string) => Run + title: string + maxBody: number +} + +function handle(req: IncomingMessage, res: ServerResponse, ctx: HandleCtx): void { + const url = req.url ?? '/' + if (url === '/healthz') { + res.writeHead(200, { 'content-type': 'text/plain' }) + res.end('ok') + return + } + const m = RUN_PATH.exec(url) + if (!m) { + res.writeHead(404, { 'content-type': 'text/plain' }) + res.end('not found') + return + } + const id = decodeURIComponent(m[1]!) + const rest = m[2] ?? '' + + // No trailing slash: redirect so the page's relative `events`/`stop` resolve under /r/:id/. + if (rest === '') { + res.writeHead(302, { location: `/r/${encodeURIComponent(id)}/` }) + res.end() + return + } + if (rest === '/') { + res.writeHead(200, { 'content-type': 'text/html; charset=utf-8' }) + res.end(dashboardHtml(ctx.title, false)) // read-only: steering is out of scope for the keystone + return + } + if (rest === '/events') { + const r = ctx.run(id) + serveSSE(req, res, r.stream, r.clients) + return + } + if (rest === '/publish') { + if (req.method !== 'POST') { + res.writeHead(405, { 'content-type': 'text/plain', allow: 'POST' }) + res.end('method not allowed') + return + } + ingestBody(req, res, ctx.run(id), ctx.maxBody) + return + } + res.writeHead(404, { 'content-type': 'text/plain' }) + res.end('not found') +} + +/** Read a JSON body (one event or an array) and push each event into the run's stream. */ +function ingestBody(req: IncomingMessage, res: ServerResponse, r: Run, maxBody: number): void { + let body = '' + let tooBig = false + req.on('data', (chunk: Buffer) => { + if (tooBig) return + body += chunk + if (body.length > maxBody) { + tooBig = true + res.writeHead(413, { 'content-type': 'text/plain' }) + res.end('payload too large') + req.destroy() + } + }) + req.on('end', () => { + if (tooBig) return + let parsed: unknown + try { + parsed = JSON.parse(body) + } catch { + res.writeHead(400, { 'content-type': 'text/plain' }) + res.end('invalid json') + return + } + const events = (Array.isArray(parsed) ? parsed : [parsed]) as FrameworkEvent[] + for (const event of events) r.stream.push(event) + res.writeHead(202, { 'content-type': 'application/json' }) + res.end(`{"ok":true,"received":${events.length}}`) + }) +} + +function closeRelay(server: Server, runs: Map): Promise { + for (const { stream, clients } of runs.values()) { + stream.close() + for (const res of clients) res.end() + clients.clear() + } + runs.clear() + return new Promise(resolvePromise => server.close(() => resolvePromise())) +} + +/** A publisher that forwards a run's events to a relay. Best-effort and ordered. */ +export interface RelayPublisher { + /** The viewer URL to share (`/r//`). */ + readonly url: string + /** Queue one event to POST to the relay (serialized, so the relay replays in order). */ + publish(event: FrameworkEvent): void + /** Resolve once every queued POST has been sent (or failed), for a clean shutdown. */ + flush(): Promise +} + +/** + * Forward a live run's {@link FrameworkEvent}s to a {@link startRelay} relay so + * remote browsers can watch it. POSTs are serialized (chained) so the relay's + * replay order matches the run, and best-effort: a failed POST is reported via + * `onError` but never interrupts the run. + */ +export function relayPublisher(base: string, runId: string, onError?: (err: unknown) => void): RelayPublisher { + const root = `${base.replace(/\/+$/, '')}/r/${encodeURIComponent(runId)}` + let chain: Promise = Promise.resolve() + return { + url: `${root}/`, + publish(event) { + chain = chain.then(async () => { + try { + await fetch(`${root}/publish`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify(event), + }) + } catch (err) { + onError?.(err) + } + }) + }, + flush() { + return chain + }, + } +}