fix(events): carry machine events on the terminal WebSocket, with SSE fallback

changelog.d/multiplex-machine-events.md

@@ -0,0 +1,4 @@
+### Fixed
+- Multiple browser tabs no longer occupy the HTTP/1.1 connection pool with separate machine event
+  streams when the server supports machine notifications on the existing terminal WebSocket.
+  Older servers retain SSE, with a server-side transport rollback switch.

docs/development.md

@@ -71,6 +71,35 @@ still returns `undefined`. Ordinary responses carry the server ID too. The defau
 `X-Content-Type-Options: nosniff` and `Referrer-Policy: no-referrer` fill absent headers only,
 preserving each route's CSP, framing, MIME, caching and compression headers.
 
+## Machine event transport and rollback
+
+New browser clients subscribe to `machine-events` on the existing local `/ws` connection when
+its `features` frame advertises that capability. The `machine-event` frames carry the same
+machine roster, redacted sidebar values, shared preferences, seen finishes and filtered device
+pairings as `/api/machines/events`. The local connection is kept even while a remote PC is
+selected; remote terminal sockets do not subscribe to this document.
+
+Older servers still use SSE. Set `OVIS_MACHINE_EVENTS_WS=0` in the server environment and restart
+only Ovis to disable the capability and return new clients to SSE without changing stored data.
+Removing that override and restarting restores the WebSocket transport. The SSE endpoint remains
+available to older clients and bridge/relay consumers. A reconnect negotiates the transport again;
+no terminal input is replayed by this subscription.
+
+A page takes machine events over SSE whenever its socket does not carry them: a connection that
+has not said whether it does within five seconds, a socket stopped for output (4008, which only
+reconnects on an attach), or a connection whose server sent `machine-events-ended`. Machine events
+never close the terminal or stall its output: one that would put more than `OUTPUT_LOW_BYTES`
+(64 KiB) on the socket, counting what is queued already, ends the subscription with that frame
+instead. A roster carries every PC's snapshot, so a page whose roster is larger than that takes
+machine events over SSE, one stream per tab as before this transport. The next connection that offers the capability takes over
+from SSE again.
+
+This avoids consuming one long-lived HTTP/1.1 request per browser tab. It does not make every tab
+share its rendered conversation, terminal or JavaScript heap, and is not a claim of a measured RAM
+reduction. Compare the same pages, conversation sizes and activity before claiming one. In
+particular, the conversation cache's byte budget is an upper bound on retained serialized answers,
+not a measurement of a tab's resident memory or its parsed objects and rendered nodes.
+
 ## Checks
 
 Production builds create Brotli (`.br`) and gzip (`.gz`) sidecars for JavaScript and CSS under

scripts/ci-browser.sh

@@ -60,6 +60,7 @@ export HERDR_DEMO_BUILD
 run_script scripts/demo-build.ts "$HERDR_DEMO_BUILD"
 run_script scripts/ui-regression.ts
 run_script scripts/ui-contexts-regression.ts
+run_script scripts/machine-events-browser-regression.ts
 run_script scripts/workspace-rows-regression.ts
 run_script scripts/pending-pane-switch-regression.ts
 run_script scripts/lan-exposure-notice-regression.ts

scripts/ci-browser.test.ts

@@ -142,6 +142,7 @@ describe("browser lane", () => {
       { script: "scripts/demo-build.ts", seconds: expect.any(Number), code: 0 },
       { script: "scripts/ui-regression.ts", seconds: expect.any(Number), code: 0 },
       { script: "scripts/ui-contexts-regression.ts", seconds: expect.any(Number), code: 0 },
+      { script: "scripts/machine-events-browser-regression.ts", seconds: expect.any(Number), code: 0 },
       { script: "scripts/workspace-rows-regression.ts", seconds: expect.any(Number), code: 0 },
       { script: "scripts/pending-pane-switch-regression.ts", seconds: expect.any(Number), code: 0 },
       { script: "scripts/lan-exposure-notice-regression.ts", seconds: expect.any(Number), code: 0 },

scripts/file-viewer-regression.ts

@@ -25,6 +25,8 @@ import { runMoreItem } from "./header-more.ts";
 import type { Machine } from "../shared/protocol.ts";
 import { chromePath } from "./chrome-path.ts";
 import { routeDrained } from "./drained-route.ts";
+// machine events stay on SSE here: these checks hold or watch /api/machines/events (the socket transport has its own checks)
+process.env["OVIS_MACHINE_EVENTS_WS"] = "0";
 
 const openFiles = async (page: import("playwright-core").Page) => {
   const headerButton = page.locator(".header-files-button");

scripts/machine-events-browser-regression.ts

@@ -0,0 +1,65 @@
+/** More tabs than a browser's HTTP/1.1 connections per host take machine events over their terminal socket; tests use isolated app state. */
+import "./test-herdr.ts";
+import assert from "node:assert/strict";
+import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { chromium } from "playwright-core";
+import { chromePath } from "./chrome-path.ts";
+import { createServer } from "../server/index.ts";
+
+const TABS = 8;
+const stateDir = mkdtempSync(join(tmpdir(), "ovis-machine-browser-"));
+const previous = process.env["OVIS_MACHINE_EVENTS_WS"];
+delete process.env["OVIS_MACHINE_EVENTS_WS"];
+const server = createServer({ port: 0, hostname: "127.0.0.1", stateDir, token: "", tailscaleOwner: null, sidecar: false });
+const base = `http://127.0.0.1:${server.port}`;
+let browser: Awaited<ReturnType<typeof chromium.launch>> | undefined;
+try {
+  browser = await chromium.launch({ executablePath: chromePath(), headless: true });
+  const context = await browser.newContext();
+  let sse = 0;
+  context.on("request", (request) => { if (new URL(request.url()).pathname === "/api/machines/events") sse++; });
+  await context.addInitScript(() => {
+    const status = { documents: 0, preference: -1 };
+    Object.assign(window, { machineEventTest: status });
+    const NativeSocket = window.WebSocket;
+    window.WebSocket = class extends NativeSocket {
+      constructor(url: string | URL, protocols?: string | string[]) {
+        super(url, protocols);
+        this.addEventListener("message", (event) => {
+          const frame = JSON.parse(String(event.data));
+          if (frame.type !== "machine-event") return;
+          if (frame.event.type === "machines") status.documents++;
+          if (frame.event.type === "preferences") status.preference = frame.event.preferences.revision;
+        });
+      }
+    };
+  });
+  const pages = [];
+  for (let i = 0; i < TABS; i++) {
+    const page = await context.newPage(); pages.push(page);
+    await page.goto(base, { waitUntil: "domcontentloaded" });
+    await page.waitForFunction(() => (window as unknown as { machineEventTest: { documents: number } }).machineEventTest.documents > 0);
+  }
+  assert.equal(sse, 0, "machine events must not open any HTTP event stream");
+  // every tab can still make a plain request: no long-lived stream holds a connection slot
+  const latencies = await Promise.all(pages.map((page) => page.evaluate(async () => {
+    const start = performance.now();
+    const response = await fetch("/api/health", { signal: AbortSignal.timeout(10_000) });
+    if (!response.ok) throw new Error(`health answered ${response.status}`);
+    return performance.now() - start;
+  })));
+  const change = await context.request.put(base + "/api/preferences", { data: { settings: { language: "ko" } }, headers: { origin: base } });
+  assert.equal(change.status(), 200);
+  const revision = (await change.json()).revision;
+  await Promise.all(pages.map((page) => page.waitForFunction((revision) =>
+    (window as unknown as { machineEventTest: { preference: number } }).machineEventTest.preference === revision, revision)));
+  const report = { tabs: pages.length, sseRequests: sse, preferenceRevisionDeliveredToAll: revision, healthMaxMs: Math.round(Math.max(...latencies)) };
+  console.log(JSON.stringify(report));
+  writeFileSync(process.env["MACHINE_EVENTS_REPORT"] ?? join(tmpdir(), "ovis-machine-events-browser-report.json"), JSON.stringify(report, null, 2));
+  await context.close();
+} finally {
+  await browser?.close(); server.stop(); rmSync(stateDir, { recursive: true, force: true });
+  if (previous === undefined) delete process.env["OVIS_MACHINE_EVENTS_WS"]; else process.env["OVIS_MACHINE_EVENTS_WS"] = previous;
+}

scripts/preferences-sync-regression.ts

@@ -8,6 +8,8 @@ import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { chromium, type BrowserContext, type Page, type Response } from "playwright-core";
 import { chromePath } from "./chrome-path.ts";
+// machine events stay on SSE here: these checks hold or watch /api/machines/events (the socket transport has its own checks)
+process.env["OVIS_MACHINE_EVENTS_WS"] = "0";
 
 await import("./test-herdr.ts");
 const { createServer } = await import("../server/index.ts");

scripts/recent-folders-regression.ts

@@ -9,6 +9,8 @@ import type { WorkspaceCreated } from "../shared/protocol.ts";
 import { sessionSnapshot, workspaceClose, workspaceCreate } from "../server/herdr/client.ts";
 import { passingFolder } from "../src/lib/recent-directories.ts";
 import { chromePath } from "./chrome-path.ts";
+// machine events stay on SSE here: these checks hold or watch /api/machines/events (the socket transport has its own checks)
+process.env["OVIS_MACHINE_EVENTS_WS"] = "0";
 
 await import("./test-herdr.ts");
 const { createServer } = await import("../server/index.ts");

scripts/ui-contexts-checks.ts

@@ -37,6 +37,8 @@ import { checkBackgroundTabKeepsTerminalSize, checkChatKeepsTerminalSize, checkF
 import { checkUpdateNotice } from "./update-notice-regression.ts";
 import { checkTelemetrySettings } from "./telemetry-regression.ts";
 import { chromePath } from "./chrome-path.ts";
+// machine events stay on SSE here: these checks hold or watch /api/machines/events (the socket transport has its own checks)
+process.env["OVIS_MACHINE_EVENTS_WS"] = "0";
 
 async function timed(name: string, check: () => Promise<void>): Promise<void> {
   const started = performance.now();

server/core/connections.test.ts

@@ -5,11 +5,14 @@ import { join } from "node:path";
 import { OUTPUT_STALLED_CLOSE_CODE } from "../../shared/terminal-flow.ts";
 import { DeviceStore } from "../devices.ts";
 import type { MachineRelay } from "../machine-relay.ts";
-import { OUTPUT_HARD_BYTES } from "../output-window.ts";
+import { OUTPUT_HARD_BYTES, OUTPUT_HIGH_BYTES, OUTPUT_LOW_BYTES } from "../output-window.ts";
 import type { PendingInputs } from "../pending-input.ts";
-import { backlog, type Client, connectionData, connectionOf, createConnections, send } from "./connections.ts";
+import type { MachineEventDocument } from "../machine-events.ts";
+import type { ServerMessage } from "../../shared/protocol.ts";
+import type { SharedPreferences } from "../../shared/preferences.ts";
+import { backlog, type Client, connectionData, connectionOf, createConnections, send, setMachineEvents } from "./connections.ts";
 
-function harness({ deviceId, readOnly = false, relay, revokedAlready = false, store }: { deviceId?: string; readOnly?: boolean; relay?: MachineRelay; revokedAlready?: boolean; store?: DeviceStore } = {}) {
+function harness({ deviceId, readOnly = false, relay, revokedAlready = false, store, machineEvents }: { deviceId?: string; readOnly?: boolean; relay?: MachineRelay; revokedAlready?: boolean; store?: DeviceStore; machineEvents?: MachineEventDocument } = {}) {
   const log: string[] = [];
   const revocations: (() => void)[] = [];
   const devices = store ?? {
@@ -20,7 +23,7 @@ function harness({ deviceId, readOnly = false, relay, revokedAlready = false, st
   } as unknown as DeviceStore;
   const pending = { close: () => log.push("pending closed") } as unknown as PendingInputs<Client, never>;
   const client = {
-    data: connectionData({ deviceId, readOnly, ...(relay ? { relay } : {}) }),
+    data: connectionData({ deviceId, readOnly, ...(relay ? { relay } : {}), ...(machineEvents ? { machineEvents } : {}) }),
     readyState: WebSocket.OPEN,
     close: (code?: number, reason?: string) => log.push(`closed ${code} ${reason}`),
     getBufferedAmount: () => 0,
@@ -92,6 +95,64 @@ it("closes a client whose backlog of ordinary frames would pass the transport bu
   expect(log).toEqual([`closed ${OUTPUT_STALLED_CLOSE_CODE} terminal output transport stalled`]);
 });
 
+it("ends the machine-event subscription, never the terminal connection, rather than put more than OUTPUT_LOW_BYTES of machine events on the socket", () => {
+  const listeners = new Set<(preferences: SharedPreferences) => void>();
+  const preferences = (bytes: number) => ({ revision: 1, settings: { note: "x".repeat(bytes) } }) as unknown as SharedPreferences;
+  let initial = 200;
+  const machineEvents: MachineEventDocument = {
+    manager: { subscribe: () => () => {} },
+    preferences: { get: () => preferences(initial), subscribe: (listener) => { listeners.add(listener); return () => { listeners.delete(listener); }; } },
+  };
+  const subscribe = (queued: number) => {
+    const { client, log } = harness({ machineEvents });
+    let buffered = queued;
+    const sent: string[] = [];
+    Object.assign(client, { getBufferedAmount: () => buffered, send: (raw: string) => { sent.push((JSON.parse(raw) as ServerMessage).type); buffered += Buffer.byteLength(raw); return -1; } });
+    setMachineEvents(client, true);
+    return { client, log, sent, queue: (bytes: number) => { buffered = bytes; } };
+  };
+  // a backlog with no room left even for the ending notice: still told, and the terminal kept
+  for (const queued of [OUTPUT_HARD_BYTES - 10, OUTPUT_LOW_BYTES - 100]) {
+    const { log, sent } = subscribe(queued);
+    expect(log).toEqual([]);
+    expect(sent).toEqual(["machine-events-ended"]);
+    expect(listeners.size).toBe(0);
+  }
+  // a small document on an empty queue goes out; a live event that would pass the cap ends it
+  const small = subscribe(0);
+  expect(small.sent).toEqual(["machine-event"]);
+  expect(listeners.size).toBe(1);
+  small.queue(OUTPUT_LOW_BYTES - 100);
+  for (const listener of [...listeners]) listener(preferences(200));
+  expect(small.log).toEqual([]);
+  expect(small.sent).toEqual(["machine-event", "machine-events-ended"]);
+  expect(listeners.size).toBe(0);
+  // a roster of 768 KiB-1 MiB on an empty queue never rides the socket: the page moves to SSE, and
+  // the 256 KiB of terminal output that follows is sent with nothing of it in front
+  initial = 900 * 1024;
+  const large = subscribe(0);
+  expect(large.sent).toEqual(["machine-events-ended"]);
+  expect(listeners.size).toBe(0);
+  const output = "x".repeat(256 * 1024);
+  expect(send(large.client, { type: "pty-data", pane_id: "w1:p1", data: output } as unknown as ServerMessage)).toBe(-1);
+  expect(backlog(large.client)).toBeLessThan(OUTPUT_HIGH_BYTES + 1024);
+  expect(large.log).toEqual([]);
+  expect(large.sent).toEqual(["machine-events-ended", "pty-data"]);
+});
+
+it("ends the machine-event subscription, never the terminal connection, when the socket refuses a machine event", () => {
+  const machineEvents: MachineEventDocument = {
+    manager: { subscribe: () => () => {} },
+    preferences: { get: () => ({ revision: 1, settings: {} }) as unknown as SharedPreferences, subscribe: () => () => {} },
+  };
+  const { client, log } = harness({ machineEvents });
+  const sent: string[] = [];
+  Object.assign(client, { send: (raw: string) => { const type = (JSON.parse(raw) as ServerMessage).type; sent.push(type); return type === "machine-event" ? 0 : 1; } });
+  setMachineEvents(client, true);
+  expect(log).toEqual([]);
+  expect(sent).toEqual(["machine-event", "machine-events-ended"]);
+});
+
 it("a revoke ends the connection with 1008 and leaves its pending input to the socket's close", () => {
   const { client, clients, present, log, admit, revoke, end, mayType, authorizeSocket } = harness({ deviceId: "phone" });
   expect(admit(client)).toBe(true);