fix(core): read product sockets with the Bun.connect WebSocket client

docs/browser.md

@@ -420,9 +420,13 @@ any browser profile.
 
 On Windows Bun cannot open those descriptors (`EBADF`, CI on 2026-10-05), so
 the browser listens on a random port of 127.0.0.1 and the core connects to it.
-That port has no authentication: a process of the same PC that finds it can
-drive the browser, signed-in profiles included. Do not copy sign-ins into the
-agent browser of a Windows machine other people have an account on.
+The core reads that socket with its own WebSocket client on `Bun.connect`
+(`packages/core/src/ws-client.ts`), as it does every socket it opens to
+another core: Bun's client (1.4.2 and 1.4.3) lost frames of sound streams and
+closed them with 1002 or 1011. That port has no authentication: a process of
+the same PC that finds it can drive the browser, signed-in profiles included.
+Do not copy sign-ins into the agent browser of a Windows machine other people
+have an account on.
 
 ## Verification
 

packages/core/src/browser/cdp.ts

@@ -6,10 +6,12 @@
  * pairs carrying one JSON message per NUL, so no port is open for another
  * process to use.
  * On Windows, where Bun cannot open those descriptors, it is a WebSocket to a
- * random loopback port instead.
+ * random loopback port instead, read by `ws-client.ts` rather than Bun's own
+ * client, which loses frames of a sound stream and closes it with 1002.
  */
 
 import { fstatSync } from 'node:fs';
+import { TextSocket, UpgradeError } from '../ws-client.ts';
 
 type Listener = (params: Record<string, unknown>, sessionId: string | undefined) => void;
 interface Waiting { resolve(value: Record<string, unknown>): void; reject(error: Error): void; timer: ReturnType<typeof setTimeout>; method: string }
@@ -92,19 +94,16 @@ export class Cdp {
   }
 
   /** The Windows transport: the browser's WebSocket endpoint on 127.0.0.1. */
-  static connect(url: string, timeoutMs = 10_000): Promise<Cdp> {
-    return new Promise((resolve, reject) => {
-      const socket = new WebSocket(url);
-      const timer = setTimeout(() => { socket.close(); reject(new CdpError(`the browser did not accept a DevTools connection within ${timeoutMs / 1000} seconds`)); }, timeoutMs);
-      socket.addEventListener('error', () => { clearTimeout(timer); reject(new CdpError(`could not connect to the browser's DevTools at ${url}`)); }, { once: true });
-      socket.addEventListener('open', () => {
-        clearTimeout(timer);
-        const cdp = new Cdp(text => socket.send(text), () => { try { socket.close(); } catch { /* already gone */ } });
-        socket.addEventListener('message', event => cdp.#receive(String(event.data)));
-        socket.addEventListener('close', () => cdp.#ended());
-        resolve(cdp);
-      }, { once: true });
+  static async connect(url: string, timeoutMs = 10_000): Promise<Cdp> {
+    const socket = await TextSocket.connect(url, timeoutMs).catch((error: unknown) => {
+      throw new CdpError(error instanceof UpgradeError && error.timedOut
+        ? `the browser did not accept a DevTools connection within ${timeoutMs / 1000} seconds`
+        : `could not connect to the browser's DevTools at ${url}: ${error instanceof Error ? error.message : String(error)}`);
     });
+    const cdp = new Cdp(text => socket.send(text), () => socket.close());
+    socket.onMessage(text => cdp.#receive(text));
+    socket.onClose(() => cdp.#ended());
+    return cdp;
   }
 
   /**

packages/core/src/client.ts

@@ -1,4 +1,5 @@
 import { PROTOCOL_VERSION, RPC_PATH } from '@boite/contracts';
+import { TextSocket, UpgradeError } from './ws-client.ts';
 import type {
   CoreInfo,
   HelloAccepted,
@@ -84,9 +85,12 @@ export class SocketClosedError extends Error {
 export async function connect(url: string, token: string, options: ConnectOptions = {}): Promise<CoreClient> {
   const timeoutMs = options.timeoutMs ?? 5000;
   const deadline = Date.now() + timeoutMs;
-  const socket = options.headers === undefined
-    ? new WebSocket(socketUrl(url))
-    : new WebSocket(socketUrl(url), { headers: options.headers } as unknown as string[]);
+  // Not Bun's own WebSocket, which closes sound streams from this core's server with 1002 or 1011 (`ws-client.ts`).
+  const socket = await TextSocket.connect(socketUrl(url), timeoutMs, { deflate: true, ...(options.headers === undefined ? {} : { headers: options.headers }) })
+    .catch((error: unknown) => {
+      if (error instanceof UpgradeError && error.timedOut) throw new Error('timed out opening the socket', { cause: error });
+      throw new Error(`the socket failed to open: ${error instanceof Error ? error.message : String(error)}`, { cause: error });
+    });
   const pending = new Map<number, Pending>();
   const listeners = new Map<string, Set<(payload: unknown) => void>>();
   let nextId = 1;
@@ -104,14 +108,14 @@ export async function connect(url: string, token: string, options: ConnectOption
         resolve(value) { clearTimeout(timer); resolve(value); },
         reject(error) { clearTimeout(timer); reject(error); },
       };
+      if (!socket.open) { waiter.reject(new Error('the client is closed')); return; }
       pending.set(id, waiter);
       try { socket.send(JSON.stringify({ jsonrpc: '2.0', id, method, params })); }
       catch (error) { pending.delete(id); waiter.reject(error as Error); }
     });
   };
 
-  socket.addEventListener('message', (event: MessageEvent) => {
-    const raw = typeof event.data === 'string' ? event.data : '';
+  socket.onMessage((raw) => {
     let frame: { id?: unknown; method?: unknown; params?: unknown; result?: unknown; error?: RpcError };
     try {
       frame = JSON.parse(raw) as typeof frame;
@@ -140,7 +144,7 @@ export async function connect(url: string, token: string, options: ConnectOption
   const watchers = new Set<{ cancel(): void; fail(error: Error): void }>();
   let ownerClosed = false;
 
-  socket.addEventListener('close', (event: CloseEvent) => {
+  socket.onClose((event) => {
     closed = true;
     const error = new SocketClosedError(event.code, event.reason);
     for (const waiter of pending.values()) waiter.reject(error);
@@ -151,22 +155,6 @@ export async function connect(url: string, token: string, options: ConnectOption
     }
   });
 
-  await new Promise<void>((resolve, reject) => {
-    const timer = setTimeout(() => reject(new Error('timed out opening the socket')), timeoutMs);
-    socket.addEventListener('open', () => {
-      clearTimeout(timer);
-      resolve();
-    });
-    socket.addEventListener('error', () => {
-      clearTimeout(timer);
-      reject(new Error('the socket failed to open'));
-    });
-    socket.addEventListener('close', (event: CloseEvent) => {
-      clearTimeout(timer);
-      reject(new SocketClosedError(event.code, event.reason, 'the socket closed before opening'));
-    }, { once: true });
-  }).catch(error => { socket.close(); throw error; });
-
   const hello = (await send('hello', {
     ...(options.ticket !== undefined ? { ticket: options.ticket }
       : options.grant === undefined ? (options.device === undefined ? { token } : {}) : { grant: options.grant, ...(options.nonce === undefined ? {} : { nonce: options.nonce }) }),

packages/core/src/group/relay.ts

@@ -31,6 +31,7 @@ import type { ServerWebSocket } from 'bun';
 import { FILE_ROUTE, GROUP_RELAY_ROUTE, PROTOCOL_VERSION, RPC_PATH, RpcErrorCode, VIEW_ROUTE } from '@boite/contracts';
 import type { Core } from '../core.ts';
 import { messageOf } from '../errors.ts';
+import { TextSocket } from '../ws-client.ts';
 import { OUTBOUND_CONGESTION_TIMEOUT_MS, OUTBOUND_MAX_BUFFERED_BYTES } from '../server/connection.ts';
 import { ticketAddresses } from './addresses.ts';
 
@@ -75,7 +76,8 @@ export interface RelaySocketData {
   relay: RelayPipe;
 }
 
-type Upstream = WebSocket;
+/** Not Bun's own WebSocket, which closes sound streams from a core's server with 1002 or 1011 (`ws-client.ts`). */
+type Upstream = TextSocket;
 
 /**
  * Who may be carried: a key this machine issued, or its own owner's token,
@@ -258,13 +260,8 @@ export class RelayPipe {
       void this.open(text);
       return;
     }
-    if (this.upstream !== null && this.upstream.readyState === WebSocket.OPEN) {
-      // A throw here would leave the server's handler and this pipe would stay in the hub: both sides close instead.
-      try {
-        this.upstream.send(text);
-      } catch {
-        this.close(1011, 'the other machine dropped the connection');
-      }
+    if (this.upstream !== null && this.upstream.open) {
+      this.upstream.send(text);
       return;
     }
     this.queued += Buffer.byteLength(text);
@@ -319,24 +316,12 @@ export class RelayPipe {
       return;
     }
     this.upstream = upstream;
-    upstream.onmessage = (event) => {
-      const data = event.data;
-      this.forward(typeof data === 'string' ? data : Buffer.from(data as ArrayBuffer));
-    };
-    upstream.onclose = (event) => this.close(SENDABLE(event.code) ? event.code : 1011, event.reason || 'the other machine closed the connection');
-    upstream.onerror = () => this.close(1011, 'the other machine dropped the connection');
-    // Closed before its handlers were in place: nothing would ever say so.
-    if (upstream.readyState !== WebSocket.OPEN) {
-      this.close(1011, 'the other machine dropped the connection');
-      return;
-    }
-    try {
-      upstream.send(JSON.stringify({ ...frame, params: rest }));
-      for (const queued of this.queue) upstream.send(queued);
-    } catch {
-      this.close(1011, 'the other machine dropped the connection');
-      return;
-    }
+    // What came before these were in place waits for them, a close included.
+    upstream.onMessage((text) => this.forward(text));
+    upstream.onClose((event) => this.close(SENDABLE(event.code) ? event.code : 1011, event.reason || 'the other machine closed the connection'));
+    if (this.closed) return;
+    upstream.send(JSON.stringify({ ...frame, params: rest }));
+    for (const queued of this.queue) upstream.send(queued);
     this.queue = [];
     this.queued = 0;
   }
@@ -349,10 +334,10 @@ export class RelayPipe {
    * reconnect, and loads a fresh snapshot, rather than missing frames on a
    * socket that stays open.
    */
-  private forward(data: string | Buffer): void {
+  private forward(data: string): void {
     const socket = this.socket;
     if (this.closed || socket === null) return;
-    const bytes = typeof data === 'string' ? Buffer.byteLength(data) : data.length;
+    const bytes = Buffer.byteLength(data);
     try {
       if (socket.getBufferedAmount() + bytes > OUTBOUND_MAX_BUFFERED_BYTES) {
         this.close(1013, 'connection exceeded outgoing buffer limit; reconnect');
@@ -385,30 +370,8 @@ export class RelayPipe {
   private async dial(targets: string[]): Promise<Upstream | null> {
     for (const target of targets) {
       if (this.closed) return null;
-      const opened = await new Promise<Upstream | null>((resolve) => {
-        let socket: Upstream;
-        try {
-          socket = new WebSocket(`${target.replace(/^http/, 'ws')}${RPC_PATH}`);
-        } catch {
-          resolve(null);
-          return;
-        }
-        const timer = setTimeout(() => {
-          socket.onopen = socket.onerror = socket.onclose = null;
-          try { socket.close(); } catch { /* never opened */ }
-          resolve(null);
-        }, DIAL_MS);
-        socket.onopen = () => {
-          clearTimeout(timer);
-          socket.onopen = socket.onerror = socket.onclose = null;
-          resolve(socket);
-        };
-        socket.onerror = socket.onclose = () => {
-          clearTimeout(timer);
-          socket.onopen = socket.onerror = socket.onclose = null;
-          resolve(null);
-        };
-      });
+      // The other core deflates what it sends a client that is not on its loopback: offered, as Bun's client did.
+      const opened = await TextSocket.connect(`${target.replace(/^http/, 'ws')}${RPC_PATH}`, DIAL_MS, { deflate: true }).catch(() => null);
       if (opened !== null) {
         this.core.group.reached(this.coreId, target);
         return opened;
@@ -441,8 +404,8 @@ export class RelayPipe {
     const upstream = this.upstream;
     this.upstream = null;
     if (upstream !== null) {
-      upstream.onmessage = upstream.onclose = upstream.onerror = null;
-      try { upstream.close(SENDABLE(code) ? code : 1000, reason.slice(0, 120)); } catch { /* already gone */ }
+      // Its close comes back to `close`, which has already run.
+      upstream.close(SENDABLE(code) ? code : 1000, reason.slice(0, 120));
     }
     try { this.socket?.close(SENDABLE(code) ? code : 1011, reason.slice(0, 120)); } catch { /* already gone */ }
     this.hub.ended(this);

packages/core/src/projects.ts

@@ -76,6 +76,8 @@ interface GitFlag {
   /** The folder itself is gone. Undefined before a check gave a clear answer. */
   missing?: boolean | undefined;
   inflight: boolean;
+  /** The check in flight; `overdue` once a list gave up waiting for it. */
+  pending?: { done: Promise<void>; overdue: boolean; again?: Promise<void> | undefined } | undefined;
 }
 
 export class ProjectStore {
@@ -155,14 +157,16 @@ export class ProjectStore {
    * The list a client asks for, with a fresh `.git` answer per folder: a
    * `git init` done in a terminal shows on the very next answer. Each check
    * runs off the event loop and gets `gitWaitMs`; past it, or when a folder's
-   * previous check is still pending (a share whose host is gone), that folder
-   * answers from its last check instead of holding the list.
+   * previous check already went past it (a share whose host is gone), that
+   * folder answers from its last check instead of holding the list. A check
+   * still within its time may have read the disk before this call: the list
+   * waits for it and checks once more.
    */
   async listFresh(): Promise<Project[]> {
     const projects = this.core.journal.listProjects();
     const checks: Promise<void>[] = [];
     for (const project of projects) {
-      const check = this.refreshGit(project.path);
+      const check = this.freshGit(project.path);
       if (check !== null) checks.push(check);
     }
     if (checks.length > 0) {
@@ -173,6 +177,11 @@ export class ProjectStore {
       });
       await Promise.race([Promise.all(checks), deadline]);
       clearTimeout(timer);
+      // A check still pending was waited for once: the next lists answer without it.
+      for (const project of projects) {
+        const pending = this.git.get(project.path)?.pending;
+        if (pending !== undefined) pending.overdue = true;
+      }
     }
     // Every folder was just checked or is still being checked: no second probe.
     const counts = this.core.journal.archivedThreadCounts();
@@ -503,6 +512,19 @@ export class ProjectStore {
     };
   }
 
+  /**
+   * A check of `path` that started after this call: a new one, or the pending
+   * one followed by another. Null when the pending check is overdue.
+   */
+  private freshGit(path: string): Promise<void> | null {
+    const pending = this.git.get(path)?.pending;
+    if (pending === undefined) return this.refreshGit(path);
+    if (pending.overdue) return null;
+    // One follow-up per pending check, however many lists ask meanwhile.
+    pending.again ??= pending.done.then(() => this.refreshGit(path) ?? this.git.get(path)?.pending?.done);
+    return pending.again;
+  }
+
   /** Starts a check of `path`, or returns null when one is already pending. */
   private refreshGit(path: string): Promise<void> | null {
     const flag = this.git.get(path) ?? { value: undefined, inflight: false };
@@ -511,19 +533,23 @@ export class ProjectStore {
     this.git.set(path, flag);
     // One check per folder at a time, held until the disk answers: a dead share
     // costs one pending check, never a pile of them.
-    return this.gitProbe(path).then(
+    const done = this.gitProbe(path).then(
       async (found) => {
         // No clear answer keeps the last one: a sleeping share is not a lost repository.
         if (found !== null) flag.value = found;
         // A `.git` that answered sits in a folder that is there; without one the folder itself is asked.
         const present = found === true ? true : found === false ? await this.folderProbe(path).catch(() => null) : null;
         flag.inflight = false;
+        flag.pending = undefined;
         if (present !== null && this.git.get(path) === flag) this.noteFolder(path, !present);
       },
       () => {
         flag.inflight = false;
+        flag.pending = undefined;
       },
     );
+    flag.pending = { done, overdue: false };
+    return done;
   }
 }