From 4504f9ede303ec29d6c047b1698410beef760b6e Mon Sep 17 00:00:00 2001 From: iris Date: Tue, 26 May 2026 01:15:43 +0200 Subject: [PATCH 1/2] dashboard: SharedWorker for SSE multiplexing (closes #448) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit mara on #448: "firefox disconnects bc of too many tabs. needs bg service worker". picked SharedWorker over full Service Worker: smaller change, addresses the actual problem (shared connection across tabs), no offline-cache scope creep. architecture per-tab `new EventSource('/dashboard/stream')` replaced with a SharedWorker-backed facade. one SharedWorker instance per origin holds ONE upstream EventSource and fans every server-sent event out to every connected tab via MessagePort. N hyperhive tabs now share ONE backend connection, immune to Firefox's per-tab SSE throttling under many-open-tabs pressure. wire protocol (port.postMessage): tab → worker { kind: 'subscribe', url: '/dashboard/stream' } { kind: 'unsubscribe', url: '/dashboard/stream' } worker → tab { kind: 'open', url } { kind: 'message', url, data: '' } { kind: 'error', url } subscription tracking is per (port, url). a late subscriber that joins after the upstream is already OPEN gets a synthetic 'open' event so its onStreamOpen handler still runs (triggers the snapshot re-sync that recovers events lost during the join gap). unsubscribing the last port for a URL closes the upstream EventSource so we don't leak idle streams. files - frontend/packages/dashboard/src/stream-worker.js: new — the worker. multi-URL multiplexing via Map. - frontend/packages/dashboard/src/common.js: new exported helper openStream(url) — returns an EventSource-shaped facade backed by the SharedWorker. graceful fallback to direct EventSource when SharedWorker is unavailable. - frontend/packages/dashboard/src/app.js: replaces the inline new EventSource('/dashboard/stream') with openStream. - frontend/packages/dashboard/src/flow.js: passes streamFactory: openStream to termCreate so the broker terminal's SSE goes through the worker too. - frontend/packages/shared/src/terminal.js: accepts an optional streamFactory(url) option. default unchanged — non-dashboard consumers (per-agent UI) keep using direct EventSource. - frontend/packages/dashboard/build.mjs: new esbuild entry for stream-worker.js → dist/static/stream-worker.js (separate bundle because SharedWorker scripts run in a different global scope and can't be inlined into app.js). scope kept tight - per-agent UI's /events/stream stays on direct EventSource. the agent UI's tab count per agent is typically 1; SharedWorker helps when you have N tabs hitting the SAME stream and the per-agent stream URLs differ. if mara wants the agent UI to share its workers too it's a separate small PR. - no offline-cache, no push notifications — those need full Service Worker; explicit non-goal here per the design Q. validation - npm run build --workspace=@hive/dashboard clean. - stream-worker.js bundle: 1.8 kb. - app.js: 154 kb → 158 kb. flow.js: 29.9 kb → 32 kb. - browser smoke test isn't possible from inside iris's container; the EventSource-shaped facade preserves the exact onmessage / onopen / onerror surface the existing IIFE consumers use. --- frontend/packages/dashboard/build.mjs | 18 +++ frontend/packages/dashboard/src/app.js | 10 +- frontend/packages/dashboard/src/common.js | 91 +++++++++++++++ frontend/packages/dashboard/src/flow.js | 5 + .../packages/dashboard/src/stream-worker.js | 110 ++++++++++++++++++ frontend/packages/shared/src/terminal.js | 15 ++- 6 files changed, 246 insertions(+), 3 deletions(-) create mode 100644 frontend/packages/dashboard/src/stream-worker.js diff --git a/frontend/packages/dashboard/build.mjs b/frontend/packages/dashboard/build.mjs index b18ebf06..d2e87153 100644 --- a/frontend/packages/dashboard/build.mjs +++ b/frontend/packages/dashboard/build.mjs @@ -45,6 +45,24 @@ await build({ logLevel: 'info', }); +// Stream-worker entry (#448). Lives in a separate bundle: SharedWorker +// scripts run in a different global (`self` is the worker scope, no +// `window`) so they can't be inlined into app.js / flow.js. Output is +// at `static/stream-worker.js`; common.js's `openStream` references +// `/static/stream-worker.js` as the SharedWorker URL. ES-module-shaped +// (so a future `import` from within the worker can pull in shared +// utilities), browser target — webworker target subset. +await build({ + entryPoints: [src('stream-worker.js')], + outdir: staticDir(''), + bundle: true, + format: 'esm', + platform: 'browser', + target: ['es2022'], + sourcemap: true, + logLevel: 'info', +}); + // Bundle the CSS — esbuild resolves @import including the package // re-exports from @hive/shared. await build({ diff --git a/frontend/packages/dashboard/src/app.js b/frontend/packages/dashboard/src/app.js index 3a597a8d..7f55eb38 100644 --- a/frontend/packages/dashboard/src/app.js +++ b/frontend/packages/dashboard/src/app.js @@ -18,6 +18,7 @@ import { fmtAgeSecs, Panel, NOTIF, makePathLink, appendText, appendLinkified, + openStream, } from './common.js'; // mdNode (in common.js) reads `window.marked` for the markdown side @@ -2069,7 +2070,14 @@ window.marked = marked; rebuild_queue_changed: applyRebuildQueueChanged, }; (function bindDashboardStream() { - const es = new EventSource('/dashboard/stream'); + // #448: route the EventSource through a SharedWorker so all open + // hyperhive tabs share ONE backend SSE connection. Survives + // Firefox's per-tab connection throttling under many-open-tabs + // pressure (the actual mara symptom). `openStream` returns an + // EventSource-shaped facade so the rest of this IIFE is unchanged; + // graceful fallback to direct `new EventSource` when SharedWorker + // isn't supported. + const es = openStream('/dashboard/stream'); es.onmessage = (e) => { let ev; try { ev = JSON.parse(e.data); } catch { return; } diff --git a/frontend/packages/dashboard/src/common.js b/frontend/packages/dashboard/src/common.js index e0870207..09f6f15e 100644 --- a/frontend/packages/dashboard/src/common.js +++ b/frontend/packages/dashboard/src/common.js @@ -58,6 +58,97 @@ export const form = (action, btnClass, btnLabel, confirmMsg, extra = {}, opts = // tied to its caller, so they don't generalise cleanly. We can lift // them when a second consumer needs the same shape. +// ─── shared-worker SSE pipe (#448) ────────────────────────────────────── +// Returns an EventSource-shaped object backed by a SharedWorker that +// holds ONE upstream `new EventSource(url)` and fans events out to +// every connected tab. Replaces direct `new EventSource(url)` at the +// dashboard's two consumer sites (app.js inline + flow.js via +// terminal.js's `streamFactory` option) so N hyperhive tabs share +// ONE backend connection — way under the browser's per-host +// connection cap, immune to per-tab throttling that drops the SSE +// when Firefox suspends background tabs. +// +// Graceful fallback to direct EventSource on environments without +// SharedWorker (some embedded browsers, some Safari versions). The +// per-tab connection cost is the same as today — no regression. +// +// The page consumer uses the returned object like a regular +// EventSource: assign `onmessage` / `onopen` / `onerror`. `.close()` +// tells the worker to drop the subscription; the worker closes the +// upstream EventSource when the last subscriber leaves. +const SHARED_WORKER_PATH = '/static/stream-worker.js'; +const SHARED_WORKER_NAME = 'hyperhive-stream'; + +// One SharedWorker instance per page, reused by all openStream calls +// on that page. Lazy — pages with no streams don't spawn the worker. +let _sharedPort = null; +function getSharedPort() { + if (_sharedPort) return _sharedPort; + if (typeof SharedWorker === 'undefined') return null; + try { + const sw = new SharedWorker(SHARED_WORKER_PATH, SHARED_WORKER_NAME); + _sharedPort = sw.port; + _sharedPort.start(); + return _sharedPort; + } catch (err) { + console.warn('SharedWorker unavailable, falling back to direct EventSource:', err); + return null; + } +} + +export function openStream(url) { + const port = getSharedPort(); + if (!port) return new EventSource(url); + + // Build an EventSource-shaped facade so consumer code is unchanged. + // `target.onmessage` / `onopen` / `onerror` are assigned by the + // consumer; the routing function below forwards events received + // from the worker (filtered by url, since one port can multiplex + // multiple subscriptions). + const target = { + onmessage: null, + onopen: null, + onerror: null, + readyState: 0, // CONNECTING + close() { + try { port.postMessage({ kind: 'unsubscribe', url }); } + catch { /* port dead */ } + port.removeEventListener('message', route); + }, + }; + const route = (e) => { + const m = e.data; + if (!m || m.url !== url) return; + if (m.kind === 'open') { + target.readyState = 1; // OPEN + if (target.onopen) { + try { target.onopen({ target }); } + catch (err) { console.error('openStream onopen threw', err); } + } + } else if (m.kind === 'message') { + if (target.onmessage) { + try { target.onmessage({ data: m.data, target }); } + catch (err) { console.error('openStream onmessage threw', err); } + } + } else if (m.kind === 'error') { + if (target.onerror) { + try { target.onerror({ target }); } + catch (err) { console.error('openStream onerror threw', err); } + } + } + }; + port.addEventListener('message', route); + port.postMessage({ kind: 'subscribe', url }); + // Drop the subscription when the tab unloads so the worker can + // close the upstream EventSource when the last subscriber leaves. + // `pagehide` fires for both real unloads and bfcache transitions. + window.addEventListener('pagehide', () => { + try { port.postMessage({ kind: 'unsubscribe', url }); } + catch { /* port dead — worker side already cleaned up */ } + }); + return target; +} + // ─── side panel ───────────────────────────────────────────────────────── // Singleton drawer that swipes in from the right. Long content // (file previews, approval diffs, journald logs, applied config) diff --git a/frontend/packages/dashboard/src/flow.js b/frontend/packages/dashboard/src/flow.js index 18f0a0f7..41487071 100644 --- a/frontend/packages/dashboard/src/flow.js +++ b/frontend/packages/dashboard/src/flow.js @@ -16,6 +16,7 @@ import { $, el, Panel, NOTIF, appendLinkified, + openStream, } from './common.js'; (() => { @@ -185,6 +186,10 @@ import { pillAnchor: flowMain, historyUrl: '/dashboard/history', streamUrl: '/dashboard/stream', + // #448: route through the SharedWorker so this page's SSE shares + // a single backend connection with /index.html (and any other + // open hyperhive tab). + streamFactory: openStream, renderers: { sent: (ev, api) => renderMsg(ev, api, '→'), delivered: (ev, api) => renderMsg(ev, api, '✓'), diff --git a/frontend/packages/dashboard/src/stream-worker.js b/frontend/packages/dashboard/src/stream-worker.js new file mode 100644 index 00000000..3d7af676 --- /dev/null +++ b/frontend/packages/dashboard/src/stream-worker.js @@ -0,0 +1,110 @@ +// SharedWorker that holds ONE EventSource per stream URL and fans +// every server-sent event out to every connected tab via MessagePort. +// +// Problem this solves (#448 — mara: "firefox disconnects bc of too +// many tabs"): every dashboard / agent tab opens its own +// `EventSource('/dashboard/stream')`. Browsers cap concurrent +// connections per host (~6), and Firefox throttles / disconnects +// background tabs when many are open. The result: tabs silently +// drop the SSE, fall behind, and only catch up on focus. +// +// Centralising the connection in a SharedWorker means N tabs share +// ONE backend EventSource regardless of focus state — way under the +// per-host cap, immune to per-tab throttling, and the worker survives +// any individual tab being suspended. +// +// Wire protocol (port.postMessage payloads): +// +// tab → worker +// { kind: 'subscribe', url: '/dashboard/stream' } +// { kind: 'unsubscribe', url: '/dashboard/stream' } +// +// worker → tab +// { kind: 'open', url: '...' } relayed from EventSource.onopen, +// plus a synthetic open fired to +// a brand-new subscriber when the +// upstream is already OPEN — so +// the page's onStreamOpen still +// runs and triggers a snapshot +// re-sync after a reconnect gap. +// { kind: 'message', url: '...', data: '' } +// { kind: 'error', url: '...' } relayed from EventSource.onerror. +// +// Subscriptions are tracked per (port, url): a single port can +// subscribe to multiple URLs (today only one is in use but the shape +// stays open for the per-agent /events/stream multiplexing follow-up). +// Unsubscribing the last port for a URL closes the EventSource so we +// don't keep idle streams open. + +const streams = new Map(); + +function getOrCreateStream(url) { + let entry = streams.get(url); + if (entry) return entry; + const es = new EventSource(url); + entry = { es, url, ports: new Set() }; + es.onopen = () => { + for (const port of entry.ports) { + try { port.postMessage({ kind: 'open', url }); } + catch { /* port dead — cleanup happens on unsubscribe / next subscribe */ } + } + }; + es.onmessage = (e) => { + for (const port of entry.ports) { + try { port.postMessage({ kind: 'message', url, data: e.data }); } + catch { /* same */ } + } + }; + es.onerror = () => { + for (const port of entry.ports) { + try { port.postMessage({ kind: 'error', url }); } + catch { /* same */ } + } + }; + streams.set(url, entry); + return entry; +} + +function unsubscribe(port, url) { + const entry = streams.get(url); + if (!entry) return; + entry.ports.delete(port); + if (entry.ports.size === 0) { + entry.es.close(); + streams.delete(url); + } +} + +self.onconnect = (connectEvent) => { + const port = connectEvent.ports[0]; + const subscribedUrls = new Set(); + port.onmessage = (e) => { + const msg = e.data; + if (!msg || typeof msg.url !== 'string') return; + if (msg.kind === 'subscribe') { + if (subscribedUrls.has(msg.url)) return; // idempotent + const entry = getOrCreateStream(msg.url); + entry.ports.add(port); + subscribedUrls.add(msg.url); + // Synthetic open for late subscribers — the upstream EventSource + // may already be OPEN when this tab joins, in which case the + // native onopen has long since fired and won't fire again until + // the next reconnect. Hand the new tab the open event explicitly + // so its onStreamOpen handler runs. + if (entry.es.readyState === EventSource.OPEN) { + try { port.postMessage({ kind: 'open', url: msg.url }); } + catch { /* port dead immediately — give up */ } + } + } else if (msg.kind === 'unsubscribe') { + if (!subscribedUrls.has(msg.url)) return; + unsubscribe(port, msg.url); + subscribedUrls.delete(msg.url); + } + }; + // A port has no explicit "disconnect" event in the SharedWorker API + // — tabs close, the GC eventually reclaims the port, but postMessage + // to a dead port throws which the senders above catch. We don't + // proactively prune ports on a timer because the cost is bounded + // (one dead Set entry per stale tab) and the next subscribe / catch + // catches it. +}; diff --git a/frontend/packages/shared/src/terminal.js b/frontend/packages/shared/src/terminal.js index f7677805..b47b08c2 100644 --- a/frontend/packages/shared/src/terminal.js +++ b/frontend/packages/shared/src/terminal.js @@ -318,7 +318,15 @@ export function create(opts) { let live = false; let buffered = []; - const es = new EventSource(opts.streamUrl); + // #448: callers can supply a `streamFactory(url)` that returns an + // EventSource-shaped object (must expose onmessage/onopen/onerror + // + .close()). The dashboard pages pass a SharedWorker-backed + // factory so all open hyperhive tabs share ONE upstream SSE + // connection. Default keeps the direct `new EventSource(url)` + // behaviour so non-dashboard consumers (per-agent UI) are unchanged. + const es = opts.streamFactory + ? opts.streamFactory(opts.streamUrl) + : new EventSource(opts.streamUrl); es.onmessage = (e) => { let ev; try { ev = JSON.parse(e.data); } @@ -331,7 +339,10 @@ export function create(opts) { } }; es.onerror = () => { - if (es.readyState === EventSource.CONNECTING) row('note', '[reconnecting…]'); + // SharedWorker-backed facades expose `readyState` mirroring the + // upstream EventSource state; the native EventSource exposes the + // same. Either way the CONNECTING vs. closed distinction works. + if (es.readyState === 0 /* CONNECTING */) row('note', '[reconnecting…]'); else row('note', '[disconnected]'); }; es.onopen = () => { From b4b7ccf88cf2e0c7199168bfa1bba2579b6c278a Mon Sep 17 00:00:00 2001 From: iris Date: Tue, 26 May 2026 01:25:49 +0200 Subject: [PATCH 2/2] =?UTF-8?q?dashboard:=20address=20argus=20nits=20?= =?UTF-8?q?=E2=80=94=20bfcache=20restore=20+=20worker=20IIFE=20(#448)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit mara on #453: address argus's two yellow nits. bfcache restore gap previously openStream registered a pagehide unsubscribe but never re-subscribed on pageshow, so a bfcache restore left the consumer's onmessage bound but no events flowing. fix: maintain a registry of live subscriptions (Map). bind page lifecycle hooks once: - pagehide: unsubscribe every URL, drop route listeners, invalidate the cached SharedWorker port (it may be collected if all other tabs closed while this page was frozen). - pageshow { persisted: true }: get a fresh port via getSharedPort (creates a new SharedWorker if needed), re-attach every route listener, re-subscribe to every URL. target.readyState resets to CONNECTING so the worker's synthetic open after subscribe fires the consumer's onStreamOpen and triggers a refresh. worker bundle format stream-worker.js was bundled as format: 'esm' but loaded as classic via new SharedWorker(url, name). today's worker has no imports/exports so the ESM bundle is syntactically valid as a classic script; argus's concern was that a future contributor adding an import would silently break things. fix: switched build.mjs to format: 'iife'. esbuild now wraps the worker output in (() => { ... })(); any future import statement would surface as a build error rather than ship broken code. verified output starts with the IIFE wrapper. other - target.close() now reads _sharedPort lazily so close-after-bfcache (port may have been recreated) doesn't try to postMessage on a stale reference. - _activeSubs.delete on close keeps the registry honest if a consumer ever explicitly closes a stream (none do today, but the shape stays correct). validation: npm run build clean. stream-worker.js: 1.8 kb → 1.9 kb (IIFE wrapper). common.js bfcache logic adds ~30 LOC inside the existing module — bundle deltas negligible. --- frontend/packages/dashboard/build.mjs | 13 +++- frontend/packages/dashboard/src/common.js | 91 ++++++++++++++++++----- 2 files changed, 83 insertions(+), 21 deletions(-) diff --git a/frontend/packages/dashboard/build.mjs b/frontend/packages/dashboard/build.mjs index d2e87153..e8cf87b3 100644 --- a/frontend/packages/dashboard/build.mjs +++ b/frontend/packages/dashboard/build.mjs @@ -49,14 +49,19 @@ await build({ // scripts run in a different global (`self` is the worker scope, no // `window`) so they can't be inlined into app.js / flow.js. Output is // at `static/stream-worker.js`; common.js's `openStream` references -// `/static/stream-worker.js` as the SharedWorker URL. ES-module-shaped -// (so a future `import` from within the worker can pull in shared -// utilities), browser target — webworker target subset. +// `/static/stream-worker.js` as the SharedWorker URL. `format: 'iife'` +// matches the classic-script load (`new SharedWorker(url, name)` with +// no `{ type: 'module' }`); Firefox is the #448 target and module +// SharedWorker support there is patchy, so keeping the worker as a +// classic script + IIFE bundle is the compatible default. argus nit +// on #453: if a future contributor adds an `import` to this bundle, +// the IIFE format will surface it as a build error rather than +// silently shipping broken code. await build({ entryPoints: [src('stream-worker.js')], outdir: staticDir(''), bundle: true, - format: 'esm', + format: 'iife', platform: 'browser', target: ['es2022'], sourcemap: true, diff --git a/frontend/packages/dashboard/src/common.js b/frontend/packages/dashboard/src/common.js index 09f6f15e..8998ff46 100644 --- a/frontend/packages/dashboard/src/common.js +++ b/frontend/packages/dashboard/src/common.js @@ -79,26 +79,84 @@ export const form = (action, btnClass, btnLabel, confirmMsg, extra = {}, opts = const SHARED_WORKER_PATH = '/static/stream-worker.js'; const SHARED_WORKER_NAME = 'hyperhive-stream'; -// One SharedWorker instance per page, reused by all openStream calls -// on that page. Lazy — pages with no streams don't spawn the worker. +// One SharedWorker port per page, reused by all openStream calls on +// that page. Invalidated on `pagehide` so a bfcache restore picks up +// a fresh port (the cached one may have been collected if all other +// tabs closed while this page was frozen — argus nit on #453). let _sharedPort = null; -function getSharedPort() { - if (_sharedPort) return _sharedPort; +function makeSharedPort() { if (typeof SharedWorker === 'undefined') return null; try { const sw = new SharedWorker(SHARED_WORKER_PATH, SHARED_WORKER_NAME); - _sharedPort = sw.port; - _sharedPort.start(); - return _sharedPort; + sw.port.start(); + return sw.port; } catch (err) { console.warn('SharedWorker unavailable, falling back to direct EventSource:', err); return null; } } +function getSharedPort() { + if (!_sharedPort) _sharedPort = makeSharedPort(); + return _sharedPort; +} + +// Registry of live subscriptions on this page. Keyed by url so a +// second openStream call for the same URL (would only happen on a +// hypothetical multi-consumer page) attaches to the existing route +// rather than overlapping. Each entry caches the route function so +// bfcache-restore re-bind can re-attach it to the fresh port. +// +// Today's pages only call openStream once with one URL; the registry +// shape just keeps the bfcache-restore path correct if that changes +// (e.g. /index.html later subscribing to two streams). +const _activeSubs = new Map(); + +// One-shot wiring of the page-wide lifecycle hooks: on bfcache +// freeze (`pagehide { persisted: true }`) we unsubscribe so the +// worker can close the upstream when the last live subscriber +// leaves; on bfcache restore (`pageshow { persisted: true }`) we +// invalidate the cached port (it may be dead if all other tabs +// closed during the freeze) and re-attach every active subscription +// to a fresh port. argus nit on #453: without this, the consumer's +// onmessage stays bound but no events flow after a bfcache restore. +let _lifecycleBound = false; +function bindLifecycleOnce() { + if (_lifecycleBound) return; + _lifecycleBound = true; + window.addEventListener('pagehide', () => { + if (!_sharedPort) return; + for (const url of _activeSubs.keys()) { + try { _sharedPort.postMessage({ kind: 'unsubscribe', url }); } + catch { /* port dead — worker side already cleaned up */ } + } + // Drop port routes too; the bfcache-restore path will re-add + // them on a fresh port. Leaving stale routes on a dead port + // would just keep a closure alive without cost, but cleaning + // up keeps the registry shape honest. + for (const sub of _activeSubs.values()) { + try { _sharedPort.removeEventListener('message', sub.route); } + catch { /* same */ } + } + _sharedPort = null; + }); + window.addEventListener('pageshow', (ev) => { + if (!ev.persisted) return; // cold load — openStream just bound listeners + if (!_activeSubs.size) return; + const port = getSharedPort(); + if (!port) return; // SharedWorker really gone; fallback already in place + for (const [url, sub] of _activeSubs) { + sub.target.readyState = 0; // CONNECTING — the worker will fire 'open' + port.addEventListener('message', sub.route); + try { port.postMessage({ kind: 'subscribe', url }); } + catch { /* port dead immediately — skip */ } + } + }); +} export function openStream(url) { const port = getSharedPort(); if (!port) return new EventSource(url); + bindLifecycleOnce(); // Build an EventSource-shaped facade so consumer code is unchanged. // `target.onmessage` / `onopen` / `onerror` are assigned by the @@ -111,9 +169,14 @@ export function openStream(url) { onerror: null, readyState: 0, // CONNECTING close() { - try { port.postMessage({ kind: 'unsubscribe', url }); } - catch { /* port dead */ } - port.removeEventListener('message', route); + const p = _sharedPort; + if (p) { + try { p.postMessage({ kind: 'unsubscribe', url }); } + catch { /* port dead */ } + try { p.removeEventListener('message', route); } + catch { /* same */ } + } + _activeSubs.delete(url); }, }; const route = (e) => { @@ -137,15 +200,9 @@ export function openStream(url) { } } }; + _activeSubs.set(url, { target, route }); port.addEventListener('message', route); port.postMessage({ kind: 'subscribe', url }); - // Drop the subscription when the tab unloads so the worker can - // close the upstream EventSource when the last subscriber leaves. - // `pagehide` fires for both real unloads and bfcache transitions. - window.addEventListener('pagehide', () => { - try { port.postMessage({ kind: 'unsubscribe', url }); } - catch { /* port dead — worker side already cleaned up */ } - }); return target; }