Shell: one stream connection per page

Browsers allow about six HTTP/1.1 connections to a host and an open
EventSource holds one. Sharing a stream per URL was not enough: a
dashboard with six modules still ran out — Safari first — and every later
request (a remounted tile's /state, an avatar) waited in the browser's
queue for ever; a Timers clock tile went blank after any settings change.

The page now holds ONE EventSource, GET /api/stream?s=<keys>, naming the
streams it wants (`slack:/stream`, `shell:` for the shell's own events).
The server relays each module stream into it over loopback — module
routes are untouched and still serve curl — prefixing every event name
with its key, and brackets each stream's life with `<key>|__open` and
`<key>|__down` (re-opened after a pause). sseSubscribe keeps its shape:
subscribers see their events by their own names, a stream joining the
connection is remade with the new list a moment later, and every stream
announces `open` again after it so backfills run as before.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

public/shell.js

@@ -794,80 +794,130 @@ function tileMessage(entry, text) {
   entry.body.appendChild(msg);
 }
 
-/* ── shared event streams ───────────────────────────────────────────── */
+/* ── one stream per page ─────────────────────────────────────────────── */
 
 /* Browsers allow about six HTTP/1.1 connections to one host, and an open
-   EventSource holds one of them for as long as it lives. One stream per
+   EventSource holds one of them for as long as it lives. Sharing one
-   tile plus the shell's own /api/events reaches that ceiling at five or six
+   stream per URL was not enough: a dashboard with six modules still ran
-   tiles, and from then on every further request — the ＋ picker's module
+   out (Safari first), and from then on every further request — a remounted
-   list, a new tile's /state, its stream — sits in the browser's queue
+   tile's /state, an avatar, the ＋ picker's list — sat in the browser's
-   waiting for a connection that is never handed back. So tiles that ask
+   queue waiting for a connection that was never handed back. So the page
-   for the same stream share one connection: the pool keeps one EventSource
+   holds ONE EventSource, GET /api/stream?s=<keys>, and the server relays
-   per URL, fans its events out to every subscriber, and reconnects it (on
+   every stream named in it (a module's, or `shell:` for the shell's own)
-   the same 3-second timer the per-tile version used) while anyone is still
+   with the key prefixed to each event name. Subscribers still see the
-   listening. A tile joining an already-open stream is told `open` straight
+   events they asked for by their own names. Adding or dropping a stream
-   away, so its backfill runs exactly as if the connection were its own. */
+   remakes the connection with the new list, a moment later so several
+   tiles mounting together cause one reconnect; every stream announces
+   `open` again after it, so backfills run as before. */
 
-const ssePool = new Map(); // url -> shared source
+const muxSubs = new Map();     // key -> { subs: Set<{ handlers }> }
+let muxEs = null;              // the page's one EventSource
+let muxKeys = '';              // the key list it was opened with
+let muxTimer = null;           // the pending (re)connect
+let muxRetry = null;           // retry after the connection gave up
+const muxListened = new Set(); // "key|name" pairs with a listener on muxEs
 
-function sseSubscribe(url, handlers = {}) {
+/** The stream key for a URL: `/api/modules/<id><path>` → `<id>:<path>`, `/api/events` → `shell:`. */
-  let shared = ssePool.get(url);
+function muxKeyFor(url) {
-  if (!shared) {
+  if (url === '/api/events') return 'shell:';
-    shared = { subs: new Set(), es: null, retryTimer: null, names: new Set() };
+  const m = /^\/api\/modules\/([^/?#]+)(\/[^?#]*)?$/.exec(url);
-    const fanOut = (kind, e) => {
+  return m ? `${decodeURIComponent(m[1])}:${m[2] || '/'}` : null;
-      for (const sub of [...shared.subs]) {
+}
+
+function muxFanOut(key, kind, e) {
+  for (const sub of [...(muxSubs.get(key)?.subs || [])]) {
     const fn = kind === 'event' ? sub.handlers.events?.[e.type] : sub.handlers[kind];
     try { fn?.(e); } catch (err) { console.error('[shell] stream handler failed:', err); }
   }
-    };
-    shared.listen = (name) => shared.es.addEventListener(name, (e) => fanOut('event', e));
-    shared.connect = () => {
-      clearTimeout(shared.retryTimer);
-      shared.retryTimer = null;
-      try { shared.es?.close(); } catch { /* already closed */ }
-      const es = new EventSource(url);
-      shared.es = es;
-      es.onopen = (e) => fanOut('open', e);
-      es.onmessage = (e) => fanOut('message', e);
-      es.onerror = (e) => {
-        fanOut('error', e);
-        // EventSource retries transient drops itself but gives up for good on
-        // a completed non-SSE response (a 502 while the upstream is down).
-        if (es.readyState === EventSource.CLOSED && shared.subs.size) {
-          clearTimeout(shared.retryTimer);
-          shared.retryTimer = setTimeout(shared.connect, 3000);
-        }
-      };
-      for (const name of shared.names) shared.listen(name);
-    };
-    ssePool.set(url, shared);
 }
 
-  const sub = { handlers };
+/** Listen for `<key>|<name>` on the open connection and hand it to that key's subscribers as `name`. */
-  shared.subs.add(sub);
+function muxListen(key, name) {
-  for (const name of Object.keys(handlers.events || {})) {
+  const full = `${key}|${name}`;
-    if (shared.names.has(name)) continue;
+  if (!muxEs || muxListened.has(full)) return;
-    shared.names.add(name);
+  muxListened.add(full);
-    if (shared.es) shared.listen(name);
+  muxEs.addEventListener(full, (e) => {
+    if (name === '__open') return muxFanOut(key, 'open', new Event('open'));
+    if (name === '__down') return muxFanOut(key, 'error', new Event('error'));
+    const ev = { type: name, data: e.data, lastEventId: e.lastEventId };
+    muxFanOut(key, name === 'message' ? 'message' : 'event', ev);
+  });
 }
-  if (!shared.es || shared.es.readyState === EventSource.CLOSED) {
+
-    // first subscriber, or the stream had given up while nobody listened
+function muxConnect() {
-    shared.connect();
+  clearTimeout(muxTimer);
-  } else if (shared.es.readyState === EventSource.OPEN) {
+  muxTimer = null;
-    // Joining mid-flight: report the open the way a fresh connection would
+  clearTimeout(muxRetry);
-    // have, a moment later, so the caller's own setup finishes first.
+  muxRetry = null;
+  const keys = [...muxSubs.keys()].sort();
+  const wanted = keys.join(',');
+  if (!keys.length) {
+    try { muxEs?.close(); } catch { /* not open */ }
+    muxEs = null;
+    muxKeys = '';
+    return;
+  }
+  if (muxEs && muxKeys === wanted && muxEs.readyState !== EventSource.CLOSED) return;
+  try { muxEs?.close(); } catch { /* not open */ }
+  muxListened.clear();
+  muxKeys = wanted;
+  const es = new EventSource(`/api/stream?s=${encodeURIComponent(wanted)}`);
+  muxEs = es;
+  // Each stream says `open` for itself (the server's __open), so nothing to do on the connection's own open.
+  es.onerror = () => {
+    if (es !== muxEs) return;
+    for (const key of muxSubs.keys()) muxFanOut(key, 'error', new Event('error'));
+    // EventSource retries transient drops itself but gives up for good on a
+    // completed non-SSE response (a 502 while the server restarts).
+    if (es.readyState === EventSource.CLOSED && muxSubs.size) {
+      clearTimeout(muxRetry);
+      muxRetry = setTimeout(muxConnect, 3000);
+    }
+  };
+  for (const [key, entry] of muxSubs) {
+    muxListen(key, '__open');
+    muxListen(key, '__down');
+    muxListen(key, 'message');
+    for (const sub of entry.subs) for (const name of Object.keys(sub.handlers.events || {})) muxListen(key, name);
+  }
+}
+
+function sseSubscribe(url, handlers = {}) {
+  const key = muxKeyFor(url);
+  if (!key) {
+    console.warn('[shell] not a stream this page can carry:', url);
+    return { close() {} };
+  }
+  let entry = muxSubs.get(key);
+  if (!entry) {
+    entry = { subs: new Set() };
+    muxSubs.set(key, entry);
+  }
+  const sub = { handlers };
+  entry.subs.add(sub);
+  const carried = Boolean(muxEs) && muxKeys.split(',').includes(key) && muxEs.readyState !== EventSource.CLOSED;
+  if (carried) {
+    // Joining a stream the connection already carries: listeners for any
+    // new event names, and `open` a moment later so the caller's own setup
+    // finishes first — exactly as a fresh connection would have reported it.
+    for (const name of Object.keys(handlers.events || {})) muxListen(key, name);
+    if (muxEs.readyState === EventSource.OPEN) {
       setTimeout(() => {
-      if (!shared.subs.has(sub)) return;
+        if (!entry.subs.has(sub)) return;
         try { handlers.open?.(new Event('open')); } catch (err) { console.error('[shell] stream handler failed:', err); }
       }, 0);
     }
-
+  } else {
+    clearTimeout(muxTimer);
+    muxTimer = setTimeout(muxConnect, 20);
+  }
   return {
     close() {
-      if (!shared.subs.delete(sub) || shared.subs.size) return;
+      if (!entry.subs.delete(sub) || entry.subs.size) return;
-      clearTimeout(shared.retryTimer);
+      muxSubs.delete(key);
-      try { shared.es?.close(); } catch { /* already closed */ }
+      // A remount closes and reopens within milliseconds: wait before remaking the connection.
-      ssePool.delete(url);
+      clearTimeout(muxTimer);
+      muxTimer = setTimeout(muxConnect, 500);
     },
   };
 }
@@ -1959,20 +2009,14 @@ new ResizeObserver(() => {
    config or enabled flags change. Tiles of a changed module get
    onConfigChange(cfg) if they implement it, otherwise a clean remount. */
 
-let eventsSource = null;
-let eventsRetry = null;
-
 function subscribeEvents() {
-  try { eventsSource?.close(); } catch { /* not open */ }
+  // Rides on the page's one stream connection (see sseSubscribe) as `shell:`.
-  eventsSource = new EventSource('/api/events');
+  sseSubscribe('/api/events', {
-  eventsSource.addEventListener('modules-changed', () => handleModulesChanged());
+    events: {
-  eventsSource.addEventListener('theme', handleThemeEvent);
+      'modules-changed': () => handleModulesChanged(),
-  eventsSource.onerror = () => {
+      theme: handleThemeEvent,
-    if (eventsSource.readyState === EventSource.CLOSED) {
+    },
-      clearTimeout(eventsRetry);
+  });
-      eventsRetry = setTimeout(subscribeEvents, 4000);
-    }
-  };
 }
 
 async function handleModulesChanged() {

server.js

@@ -559,6 +559,142 @@ setInterval(() => {
   }
 }, 25000).unref();
 
+/* ── one stream per page ─────────────────────────────────────────────
+   Browsers allow about six HTTP/1.1 connections to a host and an open
+   EventSource holds one for as long as it lives. A dashboard with six
+   modules ran out — Safari first — and from then on every further request
+   (a remounted tile's /state, an avatar) waited in the browser's queue for
+   ever. So a page opens ONE EventSource, GET /api/stream?s=<keys>, naming
+   the streams it wants: `slack:/stream`, `pco-plan:/timers/stream`, `shell:`
+   for the shell's own events. The server opens each module stream for it
+   over loopback (the module's route is untouched and still serves curl and
+   other tools), and forwards every frame with the event name prefixed by
+   its key — `event: slack:/stream|state`. Two frames of its own bracket a
+   stream's life: `<key>|__open` when the upstream answers, `<key>|__down`
+   when it ends (it is re-opened after a pause while the page stays). */
+
+const MUX_MAX_STREAMS = 32;
+const MUX_RETRY_MS = 3000;
+const MUX_RETRY_REFUSED_MS = 10000;
+const MUX_PING_MS = 15000;
+const MUX_KEY_RE = /^([A-Za-z0-9_-]+):(\/[A-Za-z0-9_\-./]*)?$/;
+
+function parseMuxKeys(search) {
+  const params = new URLSearchParams(search || '');
+  const out = [];
+  const seen = new Set();
+  for (const raw of String(params.get('s') || '').split(',')) {
+    const key = raw.trim();
+    const m = MUX_KEY_RE.exec(key);
+    if (!m || seen.has(key)) continue;
+    seen.add(key);
+    out.push({ key, id: m[1], subPath: m[2] || '' });
+    if (out.length >= MUX_MAX_STREAMS) break;
+  }
+  return out;
+}
+
+function handleMuxStream(req, res, search) {
+  const wanted = parseMuxKeys(search);
+  res.writeHead(200, {
+    'Content-Type': 'text/event-stream; charset=utf-8',
+    'Cache-Control': 'no-cache, no-transform',
+    Connection: 'keep-alive',
+    'X-Accel-Buffering': 'no',
+  });
+  res.write('retry: 3000\n\n');
+  try { req.socket.setKeepAlive(true, 15000); } catch { /* gone */ }
+  let closed = false;
+  const send = (frame) => {
+    if (closed) return;
+    try {
+      res.write(frame);
+    } catch {
+      /* the client is gone; 'close' follows */
+    }
+  };
+
+  // The shell's own events ride along under `shell:` — a stand-in sits in
+  // shellStreams and rewrites the event line of every frame it is handed.
+  let shellProxy = null;
+  const upstreams = [];
+  for (const entry of wanted) {
+    if (entry.id === 'shell') {
+      if (shellProxy) continue;
+      shellProxy = { write(frame) { send(String(frame).replace(/^event: /m, `event: ${entry.key}|`)); } };
+      shellStreams.add(shellProxy);
+      send(`event: ${entry.key}|__open\ndata: {}\n\n`);
+      continue;
+    }
+    upstreams.push({ ...entry, req: null, timer: null, gone: false });
+  }
+
+  function forwardBlock(entry, block) {
+    let name = 'message';
+    const data = [];
+    for (const line of block.split('\n')) {
+      if (line.startsWith(':')) continue; // the upstream's keepalive; this connection pings on its own
+      if (line.startsWith('event:')) name = line.slice(6).trim() || 'message';
+      else if (line.startsWith('data:')) data.push(line.slice(5).replace(/^ /, ''));
+      // id: and retry: belong to the upstream connection, not to the page's
+    }
+    if (!data.length) return;
+    send(`event: ${entry.key}|${name}\n${data.map((d) => `data: ${d}`).join('\n')}\n\n`);
+  }
+
+  function down(entry, why, delay) {
+    if (closed || entry.gone) return;
+    entry.gone = true;
+    send(`event: ${entry.key}|__down\ndata: ${JSON.stringify({ why })}\n\n`);
+    try { entry.req?.destroy(); } catch { /* already gone */ }
+    entry.req = null;
+    entry.timer = setTimeout(() => {
+      entry.gone = false;
+      open(entry);
+    }, delay);
+  }
+
+  function open(entry) {
+    if (closed) return;
+    const path = `/api/modules/${encodeURIComponent(entry.id)}${entry.subPath || '/'}`;
+    const up = http.request({ host: '127.0.0.1', port: PORT, path, method: 'GET', headers: { accept: 'text/event-stream' } }, (upRes) => {
+      if (upRes.statusCode !== 200 || !/text\/event-stream/.test(String(upRes.headers['content-type'] || ''))) {
+        upRes.resume();
+        return down(entry, `HTTP ${upRes.statusCode}`, MUX_RETRY_REFUSED_MS);
+      }
+      send(`event: ${entry.key}|__open\ndata: {}\n\n`);
+      let buf = '';
+      upRes.setEncoding('utf8');
+      upRes.on('data', (chunk) => {
+        buf += chunk;
+        let i;
+        while ((i = buf.indexOf('\n\n')) >= 0) {
+          forwardBlock(entry, buf.slice(0, i));
+          buf = buf.slice(i + 2);
+        }
+        if (buf.length > 1e6) buf = ''; // a frame that never ends is dropped, not kept
+      });
+      upRes.on('end', () => down(entry, 'ended', MUX_RETRY_MS));
+      upRes.on('error', (err) => down(entry, err.message, MUX_RETRY_MS));
+    });
+    up.on('error', (err) => down(entry, err.message, MUX_RETRY_MS));
+    up.end();
+    entry.req = up;
+  }
+
+  for (const entry of upstreams) open(entry);
+  const ping = setInterval(() => send(': ping\n\n'), MUX_PING_MS);
+  req.on('close', () => {
+    closed = true;
+    clearInterval(ping);
+    if (shellProxy) shellStreams.delete(shellProxy);
+    for (const entry of upstreams) {
+      clearTimeout(entry.timer);
+      try { entry.req?.destroy(); } catch { /* already gone */ }
+    }
+  });
+}
+
 /* ── themes ─────────────────────────────────────────────────────────── */
 
 /* Preset palettes for the whole dashboard, chosen in Admin → Theme. The
@@ -1448,6 +1584,11 @@ function handleRequest(req, res) {
     return sendJson(res, 200, themeView());
   }
 
+  /* — every stream a page wants, over one connection — */
+  if (urlPath === '/api/stream' && req.method === 'GET') {
+    return handleMuxStream(req, res, search);
+  }
+
   /* — shell events: open dashboards learn about admin changes live — */
   if (urlPath === '/api/events' && req.method === 'GET') {
     res.writeHead(200, {