From f25738b49c756b52660473da052dd3840f1f7ff8 Mon Sep 17 00:00:00 2001
From: meetsu <96637888+klNuno@users.noreply.github.com>
Date: Mon, 19 Oct 2026 22:21:23 +0200
Subject: [PATCH 1/4] fix(core): read product sockets with the Bun.connect
 WebSocket client

---
 docs/browser.md                               |  10 +-
 packages/core/src/browser/cdp.ts              |  25 ++-
 packages/core/src/client.ts                   |  35 ++--
 packages/core/src/group/relay.ts              |  71 ++------
 .../lib => packages/core/src}/ws-client.ts    | 159 +++++++++++++---
 packages/core/test/browser/fixture.ts         |   4 +-
 .../core/test/fixtures/journal-rpc-trace.ts   |  37 +++-
 packages/core/test/product-sockets.test.ts    | 159 ++++++++++++++++
 .../core/test}/ws-client.test.ts              | 170 +++++++++++++++++-
 tests/e2e/lib/cdp.ts                          |   4 +-
 10 files changed, 540 insertions(+), 134 deletions(-)
 rename {tests/e2e/lib => packages/core/src}/ws-client.ts (60%)
 create mode 100644 packages/core/test/product-sockets.test.ts
 rename {tests/e2e => packages/core/test}/ws-client.test.ts (55%)

diff --git a/docs/browser.md b/docs/browser.md
index 27f6effaa..8c01bee75 100644
--- a/docs/browser.md
+++ b/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
 
diff --git a/packages/core/src/browser/cdp.ts b/packages/core/src/browser/cdp.ts
index c955eadd6..fd2c396d2 100644
--- a/packages/core/src/browser/cdp.ts
+++ b/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;
   }
 
   /**
diff --git a/packages/core/src/client.ts b/packages/core/src/client.ts
index 8b2567b58..56bfe8787 100644
--- a/packages/core/src/client.ts
+++ b/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,13 @@ 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.send(JSON.stringify({ jsonrpc: '2.0', id, method, params }));
     });
   };
 
-  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 +143,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 +154,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 }) }),
diff --git a/packages/core/src/group/relay.ts b/packages/core/src/group/relay.ts
index afd1b2c9f..248460390 100644
--- a/packages/core/src/group/relay.ts
+++ b/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);
diff --git a/tests/e2e/lib/ws-client.ts b/packages/core/src/ws-client.ts
similarity index 60%
rename from tests/e2e/lib/ws-client.ts
rename to packages/core/src/ws-client.ts
index c97fbce92..2acf17336 100644
--- a/tests/e2e/lib/ws-client.ts
+++ b/packages/core/src/ws-client.ts
@@ -1,18 +1,25 @@
 /*
- * A WebSocket client for devtools sockets, on `Bun.connect`. Bun's own client
- * (1.4.2 and 1.4.3) closes sound devtools streams with 1002 ("control frame is
- * fragmented", "unsupported control frame", "unexpected opcode") or 1011
- * ("Compression not implemented yet"), with or without permessage-deflate:
- * #613's relay checked every frame Chrome sent on those runs and found them
- * well-formed (#623, #635). The relay read the same bytes through
+ * The WebSocket client of every socket the core opens: the browser's DevTools
+ * on Windows (`browser/cdp.ts`), another core for the CLI (`client.ts`) and
+ * for the group relay (`group/relay.ts`). It runs on `Bun.connect`, because
+ * Bun's own client (1.4.2 and 1.4.3) closes sound streams with 1002 ("control
+ * frame is fragmented", "unsupported control frame", "unexpected opcode") or
+ * 1011 ("Compression not implemented yet"), with or without permessage-deflate:
+ * against Chrome, whose frames #613's relay checked one by one and found
+ * well-formed (#623, #635), and against the core's own server (core tests on
+ * macOS arm64, through `client.ts`). The relay read the same bytes through
  * `Bun.connect` without a fault, so this client parses them itself.
  *
- * RFC 6455 for text messages only: no extension is offered, server frames
- * must be unmasked with their reserved bits clear, fragmented messages are
- * joined, pings are answered and a close is echoed.
+ * RFC 6455 for text messages only: server frames must be unmasked, fragmented
+ * messages are joined, pings are answered and a close is echoed. Reserved bits
+ * stay clear unless `deflate` negotiated permessage-deflate (RFC 7692), and
+ * then only without context takeover, so each message inflates on its own;
+ * this side sends uncompressed.
  */
 
 import { createHash, randomBytes } from 'node:crypto';
+import { constants as zlib, inflateRawSync } from 'node:zlib';
+import { drainMicrotasks } from 'bun:jsc';
 
 const GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11';
 /** A devtools reply carrying a screenshot is a few MB; past this the stream is taken as corrupt. */
@@ -28,6 +35,11 @@ export type Frame =
   /** The stream broke here: what came before it stands, nothing after it is read. */
   | { kind: 'invalid'; reason: string };
 
+/** What permessage-deflate leaves off the end of each message (RFC 7692 section 7.2.1). */
+const DEFLATE_TAIL = new Uint8Array([0x00, 0x00, 0xff, 0xff]);
+/** Each message on its own: the server's compressor and this side's inflater keep no window between messages. */
+const DEFLATE_OFFER = 'permessage-deflate; server_no_context_takeover; client_no_context_takeover';
+
 /** Turns the bytes a server sent, cut anywhere, into messages; a malformed stream ends in one `invalid` frame. */
 export class FrameReader {
   /** Bytes of a frame not complete yet, kept as they came, and how many the next parse needs. */
@@ -38,8 +50,14 @@ export class FrameReader {
   #parts: Uint8Array[] = [];
   #partBytes = 0;
   #fragmented = false;
+  /** Whether the message being read set RSV1: its joined payload is deflated. */
+  #compressed = false;
   #decoder = new TextDecoder('utf-8', { fatal: true });
   #broken = false;
+  /** Whether permessage-deflate was negotiated, so the first frame of a message may set RSV1. */
+  readonly #inflate: boolean;
+
+  constructor(options: { inflate?: boolean } = {}) { this.#inflate = options.inflate === true; }
 
   push(bytes: Uint8Array): Frame[] {
     if (this.#broken) return [];
@@ -71,7 +89,10 @@ export class FrameReader {
       if (rest < 2) break;
       const first = joined[at]!, second = joined[at + 1]!;
       const fin = (first & 0x80) !== 0, opcode = first & 0x0f;
-      if (first & 0x70) throw new ProtocolError(`reserved bits ${(first >> 4) & 7} set on opcode ${opcode} with no extension negotiated`);
+      const rsv = (first >> 4) & 7;
+      if (rsv !== 0 && !(rsv === 4 && this.#inflate && (opcode === 0x1 || opcode === 0x2))) {
+        throw new ProtocolError(`reserved bits ${rsv} set on opcode ${opcode}${this.#inflate ? '' : ' with no extension negotiated'}`);
+      }
       if (second & 0x80) throw new ProtocolError('the server masked a frame');
       let length = second & 0x7f, header = 2;
       if (length === 126) {
@@ -93,22 +114,23 @@ export class FrameReader {
       if (rest < this.#need) break;
       const payload = joined.subarray(at + header, at + header + length);
       at += header + length;
-      const frame = this.#frame(fin, opcode, payload);
+      const frame = this.#frame(fin, opcode, payload, rsv === 4);
       if (frame !== null) frames.push(frame);
     }
     if (at < joined.length) { this.#held = [joined.slice(at)]; this.#heldBytes = joined.length - at; }
   }
 
-  #frame(fin: boolean, opcode: number, payload: Uint8Array): Frame | null {
+  #frame(fin: boolean, opcode: number, payload: Uint8Array, compressed: boolean): Frame | null {
     switch (opcode) {
       case 0x0:
         if (!this.#fragmented) throw new ProtocolError('a continuation frame with no message to continue');
         return this.#data(fin, payload);
       case 0x1:
         if (this.#fragmented) throw new ProtocolError('a text frame inside a fragmented message');
+        this.#compressed = compressed;
         return this.#data(fin, payload);
       case 0x2:
-        throw new ProtocolError('a binary message on a devtools socket');
+        throw new ProtocolError('a binary message on a text socket');
       case 0x8: {
         if (payload.length === 1) throw new ProtocolError('a close frame of 1 B');
         const code = payload.length >= 2 ? (payload[0]! << 8) | payload[1]! : 1005;
@@ -133,7 +155,9 @@ export class FrameReader {
     }
     const whole = this.#parts.length === 0 ? payload : concat([...this.#parts, payload]);
     this.#parts = []; this.#partBytes = 0; this.#fragmented = false;
-    return { kind: 'text', text: this.#text(whole) };
+    if (!this.#compressed) return { kind: 'text', text: this.#text(whole) };
+    this.#compressed = false;
+    return { kind: 'text', text: this.#text(inflated(whole)) };
   }
 
   #text(bytes: Uint8Array): string {
@@ -147,6 +171,27 @@ export class FrameReader {
 
 class ProtocolError extends Error {}
 
+function inflated(deflated: Uint8Array): Uint8Array {
+  try {
+    // A message ends on a sync flush, not a final block: the stream is not over.
+    return inflateRawSync(concat([deflated, DEFLATE_TAIL]), { finishFlush: zlib.Z_SYNC_FLUSH, maxOutputLength: MAX_MESSAGE_BYTES });
+  } catch (error) {
+    throw new ProtocolError(`a compressed message that does not inflate (${error instanceof Error ? error.message : String(error)})`);
+  }
+}
+
+/** Why `TextSocket.connect` failed; `timedOut` when the server never answered the upgrade. */
+export class UpgradeError extends Error {
+  constructor(message: string, readonly timedOut = false) { super(message); }
+}
+
+export interface ConnectOptions {
+  /** Upgrade request headers; a `Host` here replaces the URL's. */
+  headers?: Record<string, string>;
+  /** Offer permessage-deflate without context takeover, for a server that compresses what it sends. */
+  deflate?: boolean;
+}
+
 /** One client frame, masked as RFC 6455 requires of a client. */
 export function encodeFrame(opcode: number, payload: Uint8Array, mask: Uint8Array = randomBytes(4)): Uint8Array {
   const n = payload.length;
@@ -208,10 +253,10 @@ export class TextSocket {
   }
 
   /** Sends a close and hangs up once the server answers, or after `CLOSE_GRACE_MS`. */
-  close(code = 1000): void {
+  close(code = 1000, reason = ''): void {
     if (!this.#open) return;
     this.#open = false;
-    this.#write(encodeFrame(0x8, new Uint8Array([code >> 8, code & 0xff])));
+    this.#write(encodeFrame(0x8, closePayload(code, reason)));
     this.#graceTimer = setTimeout(() => this.#ended({ code, reason: 'the server did not answer the close' }), CLOSE_GRACE_MS);
   }
 
@@ -244,28 +289,43 @@ export class TextSocket {
    * Connects, upgrades and resolves once the server accepted. Rejects with the
    * server's answer when it refuses, and after `timeoutMs` when it says nothing.
    */
-  static connect(url: string, timeoutMs: number): Promise<TextSocket> {
-    const target = new URL(url);
-    if (target.protocol !== 'ws:') return Promise.reject(new Error(`only ws:// devtools urls are supported, got ${target.protocol}`));
+  static connect(url: string, timeoutMs: number, options: ConnectOptions = {}): Promise<TextSocket> {
+    let target: URL, request: string;
+    try {
+      target = new URL(url);
+      if (target.protocol !== 'ws:' && target.protocol !== 'wss:') throw new UpgradeError(`a WebSocket url starts with ws:// or wss://, got ${target.protocol}`);
+      request = upgradeRequest(target, options);
+    } catch (error) {
+      return Promise.reject(error instanceof UpgradeError ? error : new UpgradeError(`not a WebSocket url: ${url}`));
+    }
     const key = randomBytes(16).toString('base64');
     const accept = createHash('sha1').update(key + GUID).digest('base64');
+    const secure = target.protocol === 'wss:';
+    // `[::1]` in a URL is `::1` to connect to.
+    const hostname = target.hostname.replace(/^\[(.*)\]$/, '$1');
     return new Promise<TextSocket>((resolve, reject) => {
       let client: TextSocket | null = null;
       let handshake: Uint8Array = new Uint8Array(0);
-      const reader = new FrameReader();
+      let reader = new FrameReader();
       let settled = false;
       /** The connection before it upgraded: a failed upgrade ends it, including one that connects after the timeout. */
       let raw: Bun.Socket | null = null;
-      const fail = (message: string) => {
+      const fail = (message: string, timedOut = false) => {
         if (settled) return;
         settled = true;
         clearTimeout(timer);
         raw?.end();
-        reject(new Error(message));
+        reject(new UpgradeError(message, timedOut));
       };
       const receive = (bytes: Uint8Array) => {
+        let delivered = false;
         for (const frame of reader.push(bytes)) {
           if (client!.#closed) return;
+          // As with a WebSocket's message events: what the last message started, an awaited answer
+          // included, runs before the next one arrives, though both came in one read.
+          if (delivered) drainMicrotasks();
+          if (client!.#closed) return;
+          delivered = frame.kind === 'text';
           if (frame.kind === 'text') client!.#deliver(frame.text);
           else if (frame.kind === 'ping') { if (client!.#open) client!.#write(encodeFrame(0xa, frame.payload)); }
           else if (frame.kind === 'close') {
@@ -281,15 +341,18 @@ export class TextSocket {
           }
         }
       };
-      const timer = setTimeout(() => fail(`no WebSocket upgrade from ${target.host} within ${timeoutMs} ms`), timeoutMs);
+      const timer = setTimeout(() => fail(`no WebSocket upgrade from ${target.host} within ${timeoutMs} ms`, true), timeoutMs);
       Bun.connect({
-        hostname: target.hostname,
-        port: Number(target.port || 80),
+        hostname,
+        port: Number(target.port || (secure ? 443 : 80)),
+        // Checked against the system's authorities, for the name the URL gives.
+        ...(secure ? { tls: { serverName: hostname } } : {}),
         socket: {
           open(socket) {
             raw = socket;
             if (settled) { socket.end(); return; }
-            socket.write(`GET ${target.pathname}${target.search} HTTP/1.1\r\nHost: ${target.host}\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: ${key}\r\nSec-WebSocket-Version: 13\r\n\r\n`);
+            // Over TLS this runs before the handshake, and a write here is dropped: the request goes once it succeeded.
+            if (!secure) socket.write(`${request}Sec-WebSocket-Key: ${key}\r\n\r\n`);
           },
           data(socket, bytes) {
             if (client !== null) { receive(bytes); return; }
@@ -305,7 +368,11 @@ export class TextSocket {
             if (!/^HTTP\/1\.1 101\b/.test(status)) { fail(`the server refused the WebSocket upgrade: ${status}`); return; }
             if (/^sec-websocket-accept:\s*(\S+)/im.exec(head)?.[1] !== accept) { fail('the server answered the upgrade with a wrong Sec-WebSocket-Accept'); return; }
             const extensions = /^sec-websocket-extensions:\s*(.*)$/im.exec(head)?.[1]?.trim();
-            if (extensions) { fail(`the server turned on extensions nobody offered: ${extensions}`); return; }
+            if (extensions) {
+              const refusal = deflateRefusal(extensions, options.deflate === true);
+              if (refusal !== null) { fail(refusal); return; }
+              reader = new FrameReader({ inflate: true });
+            }
             settled = true;
             clearTimeout(timer);
             client = new TextSocket(socket);
@@ -313,6 +380,11 @@ export class TextSocket {
             if (end < handshake.length) receive(handshake.subarray(end));
           },
           drain() { if (client !== null) client.#flush(); },
+          handshake(socket, success, authorizationError) {
+            if (!success || authorizationError) { fail(`the TLS certificate of ${target.host} was refused: ${authorizationError?.message ?? 'the handshake failed'}`); return; }
+            if (settled) { socket.end(); return; }
+            socket.write(`${request}Sec-WebSocket-Key: ${key}\r\n\r\n`);
+          },
           end() { if (client) client.#ended({ code: 1006, reason: 'the server hung up' }); else fail('the server hung up during the WebSocket upgrade'); },
           close() { if (client) client.#ended({ code: 1006, reason: 'the connection closed' }); else fail('the connection closed during the WebSocket upgrade'); },
           error(_socket, error) { if (client) client.#ended({ code: 1006, reason: error.message }); else fail(`the WebSocket connection failed: ${error.message}`); },
@@ -323,6 +395,39 @@ export class TextSocket {
   }
 }
 
+/** The request line and headers of the upgrade, up to the key. */
+function upgradeRequest(target: URL, options: ConnectOptions): string {
+  const headers = new Map<string, [string, string]>([['host', ['Host', target.host]]]);
+  for (const [name, value] of Object.entries(options.headers ?? {})) {
+    if (!/^[!#$%&'*+.^`|~\w-]+$/.test(name) || /[\r\n]/.test(value)) throw new UpgradeError(`upgrade header ${JSON.stringify(name)} must be a token with a value on one line`);
+    if (/^(upgrade|connection|sec-websocket-.*)$/i.test(name)) throw new UpgradeError(`upgrade header ${name} is the client's own`);
+    headers.set(name.toLowerCase(), [name, value]);
+  }
+  if (options.deflate === true) headers.set('sec-websocket-extensions', ['Sec-WebSocket-Extensions', DEFLATE_OFFER]);
+  const lines = [...headers.values()].map(([name, value]) => `${name}: ${value}\r\n`).join('');
+  return `GET ${target.pathname}${target.search} HTTP/1.1\r\n${lines}Upgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Version: 13\r\n`;
+}
+
+/**
+ * Null when the server's extensions are the deflate this side offered, with
+ * its compressor keeping no window between messages; else why not.
+ */
+function deflateRefusal(extensions: string, offered: boolean): string | null {
+  if (!offered) return `the server turned on extensions nobody offered: ${extensions}`;
+  const [name, ...params] = extensions.split(';').map(part => part.trim().toLowerCase());
+  if (name !== 'permessage-deflate' || extensions.includes(',')) return `the server answered the deflate offer with ${extensions}`;
+  if (!params.includes('server_no_context_takeover')) return `the server keeps a deflate window between messages, which this client did not offer: ${extensions}`;
+  const unknown = params.find(param => !/^(server_no_context_takeover|client_no_context_takeover|(server|client)_max_window_bits(=\d+)?)$/.test(param));
+  return unknown === undefined ? null : `the server set a deflate parameter this client does not know: ${unknown}`;
+}
+
+/** A close frame's payload: the code, then as much of the reason as fits in 125 bytes without splitting a character. */
+function closePayload(code: number, reason: string): Uint8Array {
+  let text = new TextEncoder().encode(reason);
+  if (text.length > 123) text = new TextEncoder().encode(new TextDecoder().decode(text.subarray(0, 123)).replace(/\uFFFD$/, ''));
+  return concat([new Uint8Array([code >> 8, code & 0xff]), text]);
+}
+
 function concat(parts: Uint8Array[]): Uint8Array {
   let size = 0;
   for (const part of parts) size += part.length;
diff --git a/packages/core/test/browser/fixture.ts b/packages/core/test/browser/fixture.ts
index 4ddcf287d..c4a8c051a 100644
--- a/packages/core/test/browser/fixture.ts
+++ b/packages/core/test/browser/fixture.ts
@@ -6,7 +6,7 @@ import type { RemoteBrowserFrame, RemoteBrowserUnchanged } from '@boite/contract
 import { connect, type CoreClient } from '../../src/client.ts';
 import { chromiumArgs, findChromium, waitForEndpoint } from '../../src/browser/chromium.ts';
 import { echoThread, startTestCore, type TestCore } from '../harness.ts';
-import { TextSocket } from '../../../../tests/e2e/lib/ws-client.ts';
+import { TextSocket } from '../../src/ws-client.ts';
 
 const PAGE = `<!doctype html><title>Fixture</title>
 <style>body{margin:0;font:16px sans-serif} button{position:absolute;left:100px;top:100px;width:200px;height:80px}</style>
@@ -150,7 +150,7 @@ export function createBrowserFixture() {
     try {
       const args = chromiumArgs(dir, 'linux').filter(arg => arg !== '--remote-debugging-pipe');
       browser = Bun.spawn([findChromium().path!, ...args, '--remote-debugging-port=0', '--remote-debugging-address=127.0.0.1', ...extraArgs], { stdout: 'ignore', stderr: 'ignore' });
-      // Bun's own WebSocket client closes sound devtools streams with 1002 (tests/e2e/lib/ws-client.ts).
+      // Bun's own WebSocket client closes sound devtools streams with 1002 (src/ws-client.ts).
       const connectedSocket = socket = await TextSocket.connect(await waitForEndpoint(dir, browser.exited), 10_000)
         .catch((error: unknown) => { throw new Error(`could not open the desktop browser DevTools socket: ${error instanceof Error ? error.message : String(error)}`, { cause: error }); });
       const connectedApp = app = await connect(harness.url, harness.token, { client: { name, version: 'test' } });
diff --git a/packages/core/test/fixtures/journal-rpc-trace.ts b/packages/core/test/fixtures/journal-rpc-trace.ts
index 670c6f17b..90052c6de 100644
--- a/packages/core/test/fixtures/journal-rpc-trace.ts
+++ b/packages/core/test/fixtures/journal-rpc-trace.ts
@@ -2,6 +2,7 @@ import { spyOn } from 'bun:test';
 import { FrameQueue } from '../../src/server/frame-queue.ts';
 import { ServerConnection } from '../../src/server/connection.ts';
 import { Router } from '../../src/router.ts';
+import { TextSocket } from '../../src/ws-client.ts';
 
 /** Test-only bounded metadata: never retain a request, response body or credential. */
 export function rpcTrace({ label = 'journal.inspect', methods = ['hello', 'pairing.grant', 'journal.inspect'] }: { label?: string; methods?: readonly string[] } = {}) {
@@ -39,6 +40,20 @@ export function rpcTrace({ label = 'journal.inspect', methods = ['hello', 'pairi
   const close = ServerConnection.prototype.close;
   const send = WebSocket.prototype.send;
   const addListener = WebSocket.prototype.addEventListener;
+  const textSend = TextSocket.prototype.send;
+  const onMessage = TextSocket.prototype.onMessage;
+  const onClose = TextSocket.prototype.onClose;
+  /** A message a client read: its frame, and the role a hello answer gives the connection. */
+  const received = (socket: object, raw: unknown): void => {
+    try {
+      const principal = JSON.parse(String(raw)).result?.principal;
+      if (['owner', 'session', 'agent'].includes(principal)) roles.set(socket, principal === 'session' ? 'phone' : principal);
+    } catch { /* Metadata only; invalid frames remain classified by frame(). */ }
+    record('client.receive', socket, frame(raw));
+  };
+  const closed = (socket: object, code: number, reason: string): void => {
+    record('client.close', socket, { code, classification: reason === 'Protocol error - unexpected opcode' ? 'unexpected-opcode' : undefined });
+  };
   const spies: { mockRestore(): void }[] = [
     spyOn(FrameQueue.prototype, 'enqueue').mockImplementation(function (this: FrameQueue, connection, raw) {
       serverConnections.set(connection.id, connection);
@@ -67,19 +82,25 @@ export function rpcTrace({ label = 'journal.inspect', methods = ['hello', 'pairi
     spyOn(WebSocket.prototype, 'addEventListener').mockImplementation(function (this: WebSocket, type: string, listener: EventListener | { handleEvent(event: Event): void }, options?: boolean | AddEventListenerOptions) {
       if (type !== 'message' && type !== 'close') return addListener.call(this, type, listener, options);
       const observed: EventListener = event => {
-        if (type === 'message') {
-          try {
-            const principal = JSON.parse((event as MessageEvent).data).result?.principal;
-            if (['owner', 'session', 'agent'].includes(principal)) roles.set(this, principal === 'session' ? 'phone' : principal);
-          } catch { /* Metadata only; invalid frames remain classified by frame(). */ }
-        }
-        record(type === 'message' ? 'client.receive' : 'client.close', this,
-          type === 'message' ? frame((event as MessageEvent).data) : { code: (event as CloseEvent).code, classification: (event as CloseEvent).reason === 'Protocol error - unexpected opcode' ? 'unexpected-opcode' : undefined });
+        if (type === 'message') received(this, (event as MessageEvent).data);
+        else closed(this, (event as CloseEvent).code, (event as CloseEvent).reason);
         if (typeof listener === 'function') listener.call(this, event);
         else listener?.handleEvent(event);
       };
       return addListener.call(this, type, observed, options);
     }),
+    // The core client reads its socket through `ws-client.ts`.
+    spyOn(TextSocket.prototype, 'send').mockImplementation(function (this: TextSocket, text) {
+      record('client.send', this, frame(text));
+      textSend.call(this, text);
+      record('client.send.returned', this);
+    }),
+    spyOn(TextSocket.prototype, 'onMessage').mockImplementation(function (this: TextSocket, listener) {
+      onMessage.call(this, text => { received(this, text); listener(text); });
+    }),
+    spyOn(TextSocket.prototype, 'onClose').mockImplementation(function (this: TextSocket, listener) {
+      onClose.call(this, event => { closed(this, event.code, event.reason); listener(event); });
+    }),
   ];
   let restored = false;
   return {
diff --git a/packages/core/test/product-sockets.test.ts b/packages/core/test/product-sockets.test.ts
new file mode 100644
index 000000000..2a64d1ddc
--- /dev/null
+++ b/packages/core/test/product-sockets.test.ts
@@ -0,0 +1,159 @@
+import { expect, test } from 'bun:test';
+import { createHash } from 'node:crypto';
+import { readdirSync, readFileSync } from 'node:fs';
+import { join } from 'node:path';
+import { PROTOCOL_VERSION } from '@boite/contracts';
+import { Cdp } from '../src/browser/cdp.ts';
+import { connect } from '../src/client.ts';
+
+/*
+ * The sockets the core opens to the browser's DevTools and to another core
+ * read through `ws-client.ts`: Bun's own WebSocket client closed sound streams
+ * from Chrome and from the core's server with 1002 and 1011 on CI.
+ */
+
+const SOURCE = join(import.meta.dir, '../src');
+
+test('no product module opens a socket with Bun\'s own WebSocket client', () => {
+  const found: string[] = [];
+  const walk = (directory: string) => {
+    for (const entry of readdirSync(directory, { withFileTypes: true })) {
+      const path = join(directory, entry.name);
+      if (entry.isDirectory()) walk(path);
+      else if (/\.ts$/.test(entry.name) && /\bnew WebSocket\(/.test(readFileSync(path, 'utf8'))) found.push(path.slice(SOURCE.length + 1).replaceAll('\\', '/'));
+    }
+  };
+  walk(SOURCE);
+  expect(found).toEqual([]);
+});
+
+function serverFrame(opcode: number, text: string | Uint8Array, fin = true): Uint8Array {
+  const body = typeof text === 'string' ? new TextEncoder().encode(text) : text;
+  const n = body.length;
+  const head = n < 126 ? [n] : n < 65_536 ? [126, n >> 8, n & 0xff] : [127, 0, 0, 0, 0, (n >>> 24) & 0xff, (n >> 16) & 0xff, (n >> 8) & 0xff, n & 0xff];
+  const out = new Uint8Array(1 + head.length + n);
+  out[0] = (fin ? 0x80 : 0) | opcode;
+  out.set(head, 1);
+  out.set(body, 1 + head.length);
+  return out;
+}
+
+/** A message as the frames that broke Bun's client: split in fragments, a ping between two of them. */
+function fragmented(text: string, pieces: number, ping: boolean): Uint8Array[] {
+  const body = new TextEncoder().encode(text);
+  const frames: Uint8Array[] = [];
+  const step = Math.ceil(body.length / pieces);
+  for (let at = 0, i = 0; i < pieces; i++, at += step) {
+    frames.push(serverFrame(i === 0 ? 1 : 0, body.subarray(at, Math.min(body.length, at + step)), i === pieces - 1));
+    if (ping && i === 0 && pieces > 1) frames.push(serverFrame(9, 'ping'));
+  }
+  return frames;
+}
+
+function seeded(seed: number): () => number {
+  return () => { seed = (seed * 1_103_515_245 + 12_345) & 0x7fffffff; return seed / 0x7fffffff; };
+}
+
+/** Messages of every length class, sent back to back: whole, fragmented, with pings, some of 300 KB. */
+function burst(random: () => number, make: (i: number, pad: string) => string, count: number): { texts: string[]; frames: Uint8Array[] } {
+  const texts: string[] = [], frames: Uint8Array[] = [];
+  for (let i = 0; i < count; i++) {
+    const size = [0, 100, 125, 126, 65_535, 65_536, 300_000][Math.floor(random() * 7)]!;
+    const text = make(i, 'é'.repeat(Math.floor(size / 2)));
+    texts.push(text);
+    frames.push(...fragmented(text, 1 + Math.floor(random() * 3), random() < 0.5));
+  }
+  return { texts, frames };
+}
+
+/**
+ * A server that upgrades by hand and writes its frames cut anywhere, down to
+ * one byte; `reply` gets each text message the client sent.
+ */
+function rawServer(reply: (text: string, write: (frames: Uint8Array[]) => void) => void, random: () => number) {
+  return Bun.listen({
+    hostname: '127.0.0.1', port: 0,
+    socket: {
+      data(socket, bytes) {
+        const state = socket as unknown as { upgraded?: boolean; held?: Uint8Array };
+        const write = (frames: Uint8Array[]) => {
+          const all = Buffer.concat(frames);
+          void (async () => {
+            for (let at = 0; at < all.length;) {
+              const part = all.subarray(at, at + 1 + Math.floor(random() * (random() < 0.3 ? 3 : 9_000)));
+              at += part.length;
+              for (let off = 0; off < part.length;) {
+                const written = socket.write(part.subarray(off));
+                if (written <= 0) await Bun.sleep(1); else off += written;
+              }
+              if (random() < 0.1) await Bun.sleep(0);
+            }
+          })();
+        };
+        if (!state.upgraded) {
+          const key = /sec-websocket-key:\s*(\S+)/i.exec(new TextDecoder().decode(bytes))?.[1];
+          if (key === undefined) return;
+          state.upgraded = true;
+          const accept = createHash('sha1').update(`${key}258EAFA5-E914-47DA-95CA-C5AB0DC85B11`).digest('base64');
+          socket.write(`HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: ${accept}\r\n\r\n`);
+          return;
+        }
+        // Client frames: masked, whole messages, at most 64 KB here.
+        let buffer: Uint8Array = state.held ? Buffer.concat([state.held, bytes]) : bytes.slice();
+        for (;;) {
+          if (buffer.length < 2) break;
+          let length = buffer[1]! & 0x7f, header = 2;
+          if (length === 126) { length = (buffer[2]! << 8) | buffer[3]!; header = 4; }
+          if (buffer.length < header + 4 + length) break;
+          const mask = buffer.subarray(header, header + 4);
+          const payload = buffer.slice(header + 4, header + 4 + length).map((byte, i) => byte ^ mask[i & 3]!);
+          if ((buffer[0]! & 0x0f) === 1) reply(new TextDecoder().decode(payload), write);
+          buffer = buffer.subarray(header + 4 + length);
+        }
+        state.held = buffer.slice();
+      },
+    },
+  });
+}
+
+test('the Windows DevTools connection reads fragmented and large frames sent back to back, cut anywhere', async () => {
+  const random = seeded(11);
+  const events = burst(random, (i, pad) => JSON.stringify({ method: 'Test.event', params: { i, pad } }), 120);
+  const reply = serverFrame(1, JSON.stringify({ id: 1, result: { product: 'Chrome' } }));
+  const server = rawServer((text, write) => {
+    if ((JSON.parse(text) as { method?: string }).method !== 'Browser.getVersion') return;
+    // The answer lands in the middle of the burst.
+    write([...events.frames.slice(0, 40), reply, ...events.frames.slice(40)]);
+  }, random);
+  try {
+    const cdp = await Cdp.connect(`ws://127.0.0.1:${server.port}/devtools/browser/x`, 5_000);
+    const seen: string[] = [];
+    const all = new Promise<void>((resolve) => cdp.on('Test.event', (params) => {
+      seen.push(JSON.stringify({ method: 'Test.event', params }));
+      if (seen.length === events.texts.length) resolve();
+    }));
+    expect(await cdp.send('Browser.getVersion', {}, undefined, 20_000)).toEqual({ product: 'Chrome' });
+    await all;
+    expect(seen).toEqual(events.texts);
+    expect(cdp.open).toBe(true);
+    cdp.close();
+  } finally { server.stop(true); }
+}, 60_000);
+
+test('the core client reads fragmented and large frames sent back to back, cut anywhere', async () => {
+  const random = seeded(13);
+  const events = burst(random, (i, pad) => JSON.stringify({ jsonrpc: '2.0', method: 'thread.updated', params: { i, pad } }), 120);
+  const server = rawServer((text, write) => {
+    const frame = JSON.parse(text) as { id: number; method: string };
+    if (frame.method === 'hello') write(fragmented(JSON.stringify({ jsonrpc: '2.0', id: frame.id, result: { core: { protocolVersion: PROTOCOL_VERSION }, principal: 'owner' } }), 3, true));
+    else write([...events.frames, serverFrame(1, JSON.stringify({ jsonrpc: '2.0', id: frame.id, result: { ok: true } }))]);
+  }, random);
+  try {
+    const client = await connect(`http://127.0.0.1:${server.port}`, 'token', { timeoutMs: 5_000 });
+    const seen: string[] = [];
+    client.onAny((event, payload) => seen.push(JSON.stringify({ jsonrpc: '2.0', method: event, params: payload })));
+    expect(await client.call('settings.get', {})).toEqual({ ok: true } as never);
+    expect(seen).toEqual(events.texts);
+    client.close();
+  } finally { server.stop(true); }
+}, 60_000);
diff --git a/tests/e2e/ws-client.test.ts b/packages/core/test/ws-client.test.ts
similarity index 55%
rename from tests/e2e/ws-client.test.ts
rename to packages/core/test/ws-client.test.ts
index e1f3d44bd..98c38165a 100644
--- a/tests/e2e/ws-client.test.ts
+++ b/packages/core/test/ws-client.test.ts
@@ -1,6 +1,7 @@
 import { expect, test } from 'bun:test';
 import { createHash } from 'node:crypto';
-import { encodeFrame, FrameReader, TextSocket, type CloseEvent, type Frame } from './lib/ws-client.ts';
+import { constants as zlibConstants, deflateRawSync } from 'node:zlib';
+import { encodeFrame, FrameReader, TextSocket, type CloseEvent, type Frame } from '../src/ws-client.ts';
 
 /** A server frame: unmasked, final unless told otherwise. */
 function serverFrame(opcode: number, payload: Uint8Array | string, fin = true, rsv = 0): Uint8Array {
@@ -216,6 +217,37 @@ test('a raw server whose frames arrive cut anywhere is read whole, and a malform
   } finally { broken.stop(true); }
 }, 30_000);
 
+test('what a message starts, an awaited answer included, runs before the next message of the same read', async () => {
+  let upgraded = false;
+  const server = Bun.listen({
+    hostname: '127.0.0.1', port: 0,
+    socket: {
+      data(socket, bytes) {
+        if (upgraded) { socket.write(join([serverFrame(1, 'answer'), serverFrame(1, 'event')])); return; }
+        const key = /sec-websocket-key:\s*(\S+)/i.exec(new TextDecoder().decode(bytes))?.[1];
+        const accept = createHash('sha1').update(`${key}258EAFA5-E914-47DA-95CA-C5AB0DC85B11`).digest('base64');
+        upgraded = true;
+        socket.write(`HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: ${accept}\r\n\r\n`);
+      },
+    },
+  });
+  try {
+    const socket = await TextSocket.connect(`ws://127.0.0.1:${server.port}/`, 5_000);
+    const order: string[] = [];
+    let answered!: () => void;
+    const answer = new Promise<void>((resolve) => { answered = resolve; });
+    const done = new Promise<void>((resolve) => socket.onMessage((text) => {
+      order.push(text);
+      if (text === 'answer') answered(); else resolve();
+    }));
+    void answer.then(() => order.push('the caller resumed'));
+    socket.send('go');
+    await done;
+    expect(order).toEqual(['answer', 'the caller resumed', 'event']);
+    socket.close();
+  } finally { server.stop(true); }
+});
+
 test('an upgrade the server refuses or never answers rejects with what happened', async () => {
   const refusing = Bun.serve({ hostname: '127.0.0.1', port: 0, fetch: () => new Response('no page', { status: 404 }) });
   try {
@@ -230,3 +262,139 @@ test('an upgrade the server refuses or never answers rejects with what happened'
     expect(hungUp).toBe(true);
   } finally { silent.stop(true); }
 });
+
+/** A deflated message as permessage-deflate sends it: raw deflate without its empty final block. */
+function deflated(text: string): Uint8Array {
+  const raw = deflateRawSync(new TextEncoder().encode(text), { finishFlush: zlibConstants.Z_SYNC_FLUSH });
+  return raw.subarray(0, raw.length - 4);
+}
+
+test('with deflate negotiated, a compressed message is read whole whatever its fragments and the reads cut', () => {
+  const text = JSON.stringify({ method: 'thread.updated', params: { pad: 'é'.repeat(50_000) } });
+  const body = deflated(text);
+  const third = Math.floor(body.length / 3);
+  const bytes = join([
+    serverFrame(1, deflated('{"id":1}'), true, 4),
+    serverFrame(1, '{"plain":true}'),
+    // RSV1 on the first fragment only, a ping between fragments.
+    serverFrame(1, body.subarray(0, third), false, 4),
+    serverFrame(9, 'p'),
+    serverFrame(0, body.subarray(third, 2 * third), false),
+    serverFrame(0, body.subarray(2 * third)),
+  ]);
+  const want: Frame[] = [
+    { kind: 'text', text: '{"id":1}' },
+    { kind: 'text', text: '{"plain":true}' },
+    { kind: 'ping', payload: new TextEncoder().encode('p') },
+    { kind: 'text', text },
+  ];
+  const read = (cut: () => number) => {
+    const reader = new FrameReader({ inflate: true });
+    const frames: Frame[] = [];
+    for (let at = 0; at < bytes.length;) {
+      const n = Math.max(1, Math.min(cut(), bytes.length - at));
+      frames.push(...reader.push(bytes.slice(at, at + n)));
+      at += n;
+    }
+    return frames;
+  };
+  expect(read(() => bytes.length)).toEqual(want);
+  expect(read(() => 1)).toEqual(want);
+  const random = seeded(3);
+  expect(read(() => 1 + Math.floor(random() * 700))).toEqual(want);
+
+  const refusal = (frames: Uint8Array[]) => new FrameReader({ inflate: true }).push(join(frames)).at(-1);
+  expect(refusal([serverFrame(1, 'a', false), serverFrame(0, 'b', true, 4)])).toEqual({ kind: 'invalid', reason: 'reserved bits 4 set on opcode 0' });
+  expect(refusal([serverFrame(9, 'p', true, 4)])).toEqual({ kind: 'invalid', reason: 'reserved bits 4 set on opcode 9' });
+  expect(refusal([serverFrame(1, '{}', true, 2)])).toEqual({ kind: 'invalid', reason: 'reserved bits 2 set on opcode 1' });
+  expect(refusal([serverFrame(1, new Uint8Array([0xff, 0xff, 0xff]), true, 4)])).toMatchObject({ kind: 'invalid', reason: expect.stringContaining('does not inflate') });
+});
+
+test('a core server that deflates for remote clients is read whole once deflate is offered, and plain without it', async () => {
+  const messages = Array.from({ length: 50 }, (_, id) => JSON.stringify({ id, pad: 'é'.repeat((id * 7_919) % 90_000) }));
+  const offers: Array<string | null> = [];
+  const server = Bun.serve({
+    hostname: '127.0.0.1', port: 0,
+    fetch(request, self) {
+      offers.push(request.headers.get('sec-websocket-extensions'));
+      return self.upgrade(request) ? undefined : new Response('upgrade required', { status: 426 });
+    },
+    // The core's own setting, and every send asking for deflate as one to a remote client does.
+    websocket: { perMessageDeflate: true, open(socket) { for (const text of messages) socket.send(text, true); }, message() {} },
+  });
+  try {
+    for (const deflate of [true, false]) {
+      const socket = await TextSocket.connect(`ws://127.0.0.1:${server.port}/rpc`, 5_000, { deflate });
+      const received: string[] = [];
+      await new Promise<void>((resolve) => socket.onMessage((text) => { received.push(text); if (received.length === messages.length) resolve(); }));
+      expect(received).toEqual(messages);
+      socket.close();
+    }
+  } finally { await server.stop(true); }
+  expect(offers).toEqual(['permessage-deflate; server_no_context_takeover; client_no_context_takeover', null]);
+});
+
+test('an upgrade answered with extensions this client cannot read is refused', async () => {
+  const answering = (extensions: string) => Bun.listen({
+    hostname: '127.0.0.1', port: 0,
+    socket: {
+      data(socket, bytes) {
+        const key = /sec-websocket-key:\s*(\S+)/i.exec(new TextDecoder().decode(bytes))?.[1];
+        if (key === undefined) return;
+        const accept = createHash('sha1').update(`${key}258EAFA5-E914-47DA-95CA-C5AB0DC85B11`).digest('base64');
+        socket.write(`HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: ${accept}\r\nSec-WebSocket-Extensions: ${extensions}\r\n\r\n`);
+      },
+    },
+  });
+  const cases: Array<[string, boolean, string]> = [
+    ['permessage-deflate; server_no_context_takeover', false, 'extensions nobody offered'],
+    ['permessage-deflate', true, 'keeps a deflate window'],
+    ['permessage-deflate; server_no_context_takeover; mystery', true, 'does not know: mystery'],
+    ['x-webkit-deflate-frame', true, 'answered the deflate offer with'],
+  ];
+  for (const [extensions, deflate, error] of cases) {
+    const server = answering(extensions);
+    try {
+      await expect(TextSocket.connect(`ws://127.0.0.1:${server.port}/`, 5_000, { deflate })).rejects.toThrow(error);
+    } finally { server.stop(true); }
+  }
+});
+
+test('upgrade headers reach the server, a Host among them replacing the url one, and unsafe ones are refused', async () => {
+  let seen: { host: string | null; origin: string | null } | undefined;
+  const server = Bun.serve({
+    hostname: '127.0.0.1', port: 0,
+    fetch(request, self) {
+      seen = { host: request.headers.get('host'), origin: request.headers.get('origin') };
+      return self.upgrade(request) ? undefined : new Response('upgrade required', { status: 426 });
+    },
+    websocket: { message() {} },
+  });
+  try {
+    const socket = await TextSocket.connect(`ws://127.0.0.1:${server.port}/`, 5_000, { headers: { host: 'boite.example:7000', Origin: 'http://boite.example' } });
+    expect(seen).toEqual({ host: 'boite.example:7000', origin: 'http://boite.example' });
+    socket.close();
+    await expect(TextSocket.connect(`ws://127.0.0.1:${server.port}/`, 5_000, { headers: { 'x-a': 'b\r\nx-injected: 1' } })).rejects.toThrow('on one line');
+    await expect(TextSocket.connect(`ws://127.0.0.1:${server.port}/`, 5_000, { headers: { 'Sec-WebSocket-Key': 'x' } })).rejects.toThrow('the client\'s own');
+    await expect(TextSocket.connect(`http://127.0.0.1:${server.port}/`, 5_000)).rejects.toThrow('ws:// or wss://');
+  } finally { await server.stop(true); }
+});
+
+test('a close carries its code and reason, the reason cut to what a close frame holds without splitting a character', async () => {
+  const closes: Array<[number, string]> = [];
+  const server = Bun.serve({
+    hostname: '127.0.0.1', port: 0,
+    fetch: (request, self) => (self.upgrade(request) ? undefined : new Response('no', { status: 426 })),
+    websocket: { message() {}, close(_socket, code, reason) { closes.push([code, reason]); } },
+  });
+  try {
+    for (const reason of ['the other machine dropped the connection', 'é'.repeat(100)]) {
+      const socket = await TextSocket.connect(`ws://127.0.0.1:${server.port}/`, 5_000);
+      const closed = new Promise<CloseEvent>((resolve) => socket.onClose(resolve));
+      socket.close(4001, reason);
+      await closed;
+    }
+    for (let waited = 0; closes.length < 2 && waited < 2_000; waited += 20) await Bun.sleep(20);
+  } finally { await server.stop(true); }
+  expect(closes).toEqual([[4001, 'the other machine dropped the connection'], [4001, 'é'.repeat(61)]]);
+});
diff --git a/tests/e2e/lib/cdp.ts b/tests/e2e/lib/cdp.ts
index 1d00627b1..41d4df804 100644
--- a/tests/e2e/lib/cdp.ts
+++ b/tests/e2e/lib/cdp.ts
@@ -6,7 +6,7 @@ import { ONBOARDING_STORAGE_KEY, ONBOARDING_VERSION } from '../../../packages/ui
 import type { Subprocess } from 'bun';
 import { E2E_DIR_PREFIX, killProcessTreeAsync, removeDirectory, watchParentDeath, type ParentDeathWatch } from './cleanup.ts';
 import { uiServerSide } from './ui-requests.ts';
-import { TextSocket } from './ws-client.ts';
+import { TextSocket } from '../../../packages/core/src/ws-client.ts';
 
 /**
  * Marks the tour seen before any script of the page runs, and only where no
@@ -905,7 +905,7 @@ export async function probeDebuggingPort(port: number): Promise<string> {
   }
 }
 
-/** Bun's own WebSocket client closes sound devtools streams with 1002: see `ws-client.ts`. */
+/** Bun's own WebSocket client closes sound devtools streams with 1002: see `packages/core/src/ws-client.ts`. */
 async function openSocket(target: TargetInfo): Promise<TextSocket> {
   const url = target.webSocketDebuggerUrl;
   if (url === undefined) throw new Error('the page target has no debugger url');

From 6a028a83cbec4de6b74e48a13701d4f96fb274d8 Mon Sep 17 00:00:00 2001
From: meetsu <96637888+klNuno@users.noreply.github.com>
Date: Mon, 19 Oct 2026 22:42:48 +0200
Subject: [PATCH 2/4] fix(projects): check again a folder whose pending check
 read the disk before the list

---
 packages/core/src/projects.ts           | 34 ++++++++++++++++++++++---
 packages/core/test/projects-git.test.ts | 27 +++++++++++++++++++-
 2 files changed, 56 insertions(+), 5 deletions(-)

diff --git a/packages/core/src/projects.ts b/packages/core/src/projects.ts
index 14f12200f..f01a5578c 100644
--- a/packages/core/src/projects.ts
+++ b/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;
   }
 }
 
diff --git a/packages/core/test/projects-git.test.ts b/packages/core/test/projects-git.test.ts
index 303beb456..12e7d76aa 100644
--- a/packages/core/test/projects-git.test.ts
+++ b/packages/core/test/projects-git.test.ts
@@ -1,7 +1,7 @@
 import { mkdirSync, rmSync } from 'node:fs';
 import { join } from 'node:path';
 import { afterEach, beforeEach, expect, test } from 'bun:test';
-import { ProjectStore, hasGitMarker } from '../src/projects.ts';
+import { ProjectStore, hasGitMarker, probeGit } from '../src/projects.ts';
 import { startTestCore, waitFor } from './harness.ts';
 import type { TestCore } from './harness.ts';
 
@@ -128,6 +128,31 @@ test('a deleted project folder is named by every refusal, and a folder put back
   await client.call('threads.create', { projectId: project.id, providerId: 'echo', accountId: account?.id ?? '', title: 'back' });
 });
 
+test('a list asked after a folder came back is not answered by a check that read the disk before', async () => {
+  const client = await harness.connect();
+  const path = join(harness.dataDir, 'returning');
+  mkdirSync(path, { recursive: true });
+  const project = await client.call('projects.add', { path });
+  rmSync(path, { recursive: true, force: true });
+  expect((await client.call('projects.list', {})).find((entry) => entry.id === project.id)?.missing).toBe(true);
+  // The request below starts a check in the background, as any request naming a
+  // project does. It reads the disk while the folder is gone; on a loaded runner
+  // its answer is not in yet when the folder comes back.
+  const stale = Promise.withResolvers<boolean | null>();
+  const store = harness.core.projects;
+  let started = 0;
+  store.gitProbe = () => { started += 1; return stale.promise; };
+  await expect(client.call('worktrees.list', { projectId: project.id })).rejects.toThrow('does not exist any more');
+  expect(started).toBe(1);
+  // Every other folder answers at once; the stale answer lands once the list has started.
+  store.gitProbe = async (folder) => (folder === path ? probeGit(folder) : null);
+  mkdirSync(path, { recursive: true });
+  const listFresh = store.listFresh.bind(store);
+  store.listFresh = () => { const answer = listFresh(); setTimeout(() => stale.resolve(false), 0); return answer; };
+  const back = (await client.call('projects.list', {})).find((entry) => entry.id === project.id);
+  expect(back?.missing).toBeUndefined();
+});
+
 test('a disk that gives no clear answer does not mark a project missing', async () => {
   const client = await harness.connect();
   const path = join(harness.dataDir, 'asleep');

From 8c5129197fd30d3e39229347324715348073a64e Mon Sep 17 00:00:00 2001
From: meetsu <96637888+klNuno@users.noreply.github.com>
Date: Mon, 19 Oct 2026 23:17:42 +0200
Subject: [PATCH 3/4] fix(core): report listener errors, drop the send queue on
 close and check upgrade answers and close codes

---
 packages/core/src/client.ts               |   3 +-
 packages/core/src/ws-client.ts            |  76 ++++++++++---
 packages/core/test/client-timeout.test.ts |  54 ++++++++++
 packages/core/test/ws-client.test.ts      | 124 +++++++++++++++++++++-
 4 files changed, 239 insertions(+), 18 deletions(-)

diff --git a/packages/core/src/client.ts b/packages/core/src/client.ts
index 56bfe8787..1b433a989 100644
--- a/packages/core/src/client.ts
+++ b/packages/core/src/client.ts
@@ -110,7 +110,8 @@ export async function connect(url: string, token: string, options: ConnectOption
       };
       if (!socket.open) { waiter.reject(new Error('the client is closed')); return; }
       pending.set(id, waiter);
-      socket.send(JSON.stringify({ jsonrpc: '2.0', id, method, params }));
+      try { socket.send(JSON.stringify({ jsonrpc: '2.0', id, method, params })); }
+      catch (error) { pending.delete(id); waiter.reject(error as Error); }
     });
   };
 
diff --git a/packages/core/src/ws-client.ts b/packages/core/src/ws-client.ts
index 2acf17336..5b9a7098f 100644
--- a/packages/core/src/ws-client.ts
+++ b/packages/core/src/ws-client.ts
@@ -32,8 +32,8 @@ export type Frame =
   | { kind: 'ping'; payload: Uint8Array }
   | { kind: 'pong' }
   | { kind: 'close'; code: number; reason: string }
-  /** The stream broke here: what came before it stands, nothing after it is read. */
-  | { kind: 'invalid'; reason: string };
+  /** The stream broke here: what came before it stands, nothing after it is read. 1007 for text that is not UTF-8, else 1002. */
+  | { kind: 'invalid'; code: number; reason: string };
 
 /** What permessage-deflate leaves off the end of each message (RFC 7692 section 7.2.1). */
 const DEFLATE_TAIL = new Uint8Array([0x00, 0x00, 0xff, 0xff]);
@@ -68,7 +68,7 @@ export class FrameReader {
       if (!(error instanceof ProtocolError)) throw error;
       this.#broken = true;
       this.#held = []; this.#heldBytes = 0;
-      frames.push({ kind: 'invalid', reason: error.message });
+      frames.push({ kind: 'invalid', code: error.code, reason: error.message });
     }
     return frames;
   }
@@ -134,6 +134,7 @@ export class FrameReader {
       case 0x8: {
         if (payload.length === 1) throw new ProtocolError('a close frame of 1 B');
         const code = payload.length >= 2 ? (payload[0]! << 8) | payload[1]! : 1005;
+        if (payload.length >= 2 && !sendableCloseCode(code)) throw new ProtocolError(`a close frame with code ${code}`);
         return { kind: 'close', code, reason: this.#text(payload.subarray(2)) };
       }
       case 0x9:
@@ -164,12 +165,20 @@ export class FrameReader {
     try {
       return this.#decoder.decode(bytes);
     } catch {
-      throw new ProtocolError('a text message that is not UTF-8');
+      throw new ProtocolError('a text message that is not UTF-8', 1007);
     }
   }
 }
 
-class ProtocolError extends Error {}
+class ProtocolError extends Error {
+  /** The close code the stream ends with. */
+  constructor(message: string, readonly code = 1002) { super(message); }
+}
+
+/** Whether a peer may send `code` in a close frame (RFC 6455 section 7.4 and the IANA registry). */
+function sendableCloseCode(code: number): boolean {
+  return (code >= 1000 && code <= 1003) || (code >= 1007 && code <= 1014) || (code >= 3000 && code <= 4999);
+}
 
 function inflated(deflated: Uint8Array): Uint8Array {
   try {
@@ -230,21 +239,36 @@ export class TextSocket {
   /** Whether `send` still reaches the server. */
   get open(): boolean { return this.#open; }
 
-  /** Only one listener: a devtools socket has one reader. It gets what arrived before it first. */
+  /** Bytes written but not yet taken by the connection; none once it ended. */
+  get bufferedAmount(): number {
+    let bytes = 0;
+    for (const chunk of this.#queue) bytes += chunk.length;
+    return bytes;
+  }
+
+  /**
+   * Only one listener: a devtools socket has one reader. It gets what arrived
+   * before it first, each message's microtasks run before the next, as in a read.
+   */
   onMessage(listener: (text: string) => void): void {
     this.#onMessage = listener;
-    for (const text of this.#inbox.splice(0)) listener(text);
+    for (let first = true; this.#inbox.length > 0 && this.#onMessage === listener; first = false) {
+      if (!first) drainMicrotasks();
+      const text = this.#inbox.shift()!;
+      reported(() => listener(text));
+    }
   }
 
   /** Called once; at once when the socket already closed. */
   onClose(listener: (event: CloseEvent) => void): void {
-    if (this.#closeEvent !== null) listener(this.#closeEvent);
+    if (this.#closeEvent !== null) reported(() => listener(this.#closeEvent!));
     else this.#onClose.push(listener);
   }
 
   #deliver(text: string): void {
-    if (this.#onMessage === null) this.#inbox.push(text);
-    else this.#onMessage(text);
+    const listener = this.#onMessage;
+    if (listener === null) this.#inbox.push(text);
+    else reported(() => listener(text));
   }
 
   send(text: string): void {
@@ -281,8 +305,13 @@ export class TextSocket {
     this.#closeEvent = event;
     this.#open = false;
     clearTimeout(this.#graceTimer);
+    // Nothing waiting leaves any more: a reader that stopped reading must not keep it alive.
+    this.#queue = [];
     this.#socket.end();
-    for (const listener of this.#onClose) listener(event);
+    const listeners = this.#onClose;
+    this.#onClose = [];
+    this.#onMessage = null;
+    for (const listener of listeners) reported(() => listener(event));
   }
 
   /**
@@ -336,8 +365,8 @@ export class TextSocket {
             }
             client!.#ended({ code: frame.code, reason: frame.reason });
           } else if (frame.kind === 'invalid') {
-            client!.#write(encodeFrame(0x8, new Uint8Array([1002 >> 8, 1002 & 0xff])));
-            client!.#ended({ code: 1002, reason: frame.reason });
+            client!.#write(encodeFrame(0x8, new Uint8Array([frame.code >> 8, frame.code & 0xff])));
+            client!.#ended({ code: frame.code, reason: frame.reason });
           }
         }
       };
@@ -367,6 +396,8 @@ export class TextSocket {
             const status = head.split('\r\n', 1)[0] ?? '';
             if (!/^HTTP\/1\.1 101\b/.test(status)) { fail(`the server refused the WebSocket upgrade: ${status}`); return; }
             if (/^sec-websocket-accept:\s*(\S+)/im.exec(head)?.[1] !== accept) { fail('the server answered the upgrade with a wrong Sec-WebSocket-Accept'); return; }
+            const refused = upgradeRefusal(head);
+            if (refused !== null) { fail(refused); return; }
             const extensions = /^sec-websocket-extensions:\s*(.*)$/im.exec(head)?.[1]?.trim();
             if (extensions) {
               const refusal = deflateRefusal(extensions, options.deflate === true);
@@ -395,6 +426,25 @@ export class TextSocket {
   }
 }
 
+/** Null when an upgrade answer says `Upgrade: websocket` and `Connection: Upgrade` and picks no subprotocol; else why not. */
+function upgradeRefusal(head: string): string | null {
+  const tokens = (name: string) => [...head.matchAll(new RegExp(`^${name}:(.*)$`, 'gim'))].flatMap(match => match[1]!.split(',').map(token => token.trim().toLowerCase()));
+  if (!tokens('upgrade').includes('websocket')) return 'the server answered the upgrade without Upgrade: websocket';
+  if (!tokens('connection').includes('upgrade')) return 'the server answered the upgrade without Connection: Upgrade';
+  // This client offers no subprotocol, so the server may not pick one.
+  const protocol = /^sec-websocket-protocol:\s*(.*)$/im.exec(head)?.[1]?.trim();
+  return protocol ? `the server picked a subprotocol nobody offered: ${protocol}` : null;
+}
+
+/** Runs a listener; what it throws goes where an uncaught error goes, as from a WebSocket's listener, and the reading goes on. */
+function reported(listener: () => void): void {
+  try {
+    listener();
+  } catch (error) {
+    reportError(error);
+  }
+}
+
 /** The request line and headers of the upgrade, up to the key. */
 function upgradeRequest(target: URL, options: ConnectOptions): string {
   const headers = new Map<string, [string, string]>([['host', ['Host', target.host]]]);
diff --git a/packages/core/test/client-timeout.test.ts b/packages/core/test/client-timeout.test.ts
index 9971081e0..c5f51ba16 100644
--- a/packages/core/test/client-timeout.test.ts
+++ b/packages/core/test/client-timeout.test.ts
@@ -56,3 +56,57 @@ test.each(['hello', 'call'] as const)('client reports peer close code and bounde
     expect(error.message).not.toContain('\n');
   } finally { client?.close(); void server.stop(true); }
 });
+
+/** A core that answers hello, and every call with `{ ok: true }` after a `core.log` event. */
+function answeringCore() {
+  return Bun.serve({ hostname: '127.0.0.1', port: 0,
+    fetch(request, self) { return self.upgrade(request) ? undefined : new Response('upgrade required'); },
+    websocket: { message(socket, raw) {
+      const frame = JSON.parse(String(raw));
+      if (frame.method !== 'hello') socket.send(JSON.stringify({ jsonrpc: '2.0', method: 'core.log', params: {} }));
+      socket.send(JSON.stringify({ jsonrpc: '2.0', id: frame.id,
+        result: frame.method === 'hello' ? { core: { protocolVersion: PROTOCOL_VERSION }, principal: 'owner' } : { ok: true } }));
+    } },
+  });
+}
+
+test('a call whose params do not serialize rejects and leaves no timer behind', async () => {
+  const server = answeringCore();
+  const client = await connect(`http://127.0.0.1:${server.port}`, 'test', { requestTimeoutMs: 123_456 });
+  const live = new Set<unknown>();
+  const set = globalThis.setTimeout, clear = globalThis.clearTimeout;
+  globalThis.setTimeout = ((handler: Parameters<typeof setTimeout>[0], timeout?: number, ...args: unknown[]) => {
+    const timer = set(handler, timeout, ...args);
+    if (timeout === 123_456) live.add(timer);
+    return timer;
+  }) as typeof setTimeout;
+  globalThis.clearTimeout = ((timer: ReturnType<typeof setTimeout>) => { live.delete(timer); clear(timer); }) as typeof clearTimeout;
+  try {
+    const cyclic: Record<string, unknown> = {};
+    cyclic.self = cyclic;
+    await expect(client.call('settings.get', cyclic as never)).rejects.toThrow();
+    expect(live.size).toBe(0);
+  } finally {
+    globalThis.setTimeout = set;
+    globalThis.clearTimeout = clear;
+    client.close();
+    await server.stop(true);
+  }
+});
+
+test('an event handler that throws is reported and the answer behind it still arrives', async () => {
+  const server = answeringCore();
+  const client = await connect(`http://127.0.0.1:${server.port}`, 'test', { requestTimeoutMs: 2_000 });
+  const report = globalThis.reportError;
+  const reported: unknown[] = [];
+  globalThis.reportError = (error) => { reported.push(error); };
+  try {
+    client.on('core.log', () => { throw new Error('a subscriber failed'); });
+    expect(await client.call('settings.get', {} as never) as unknown).toEqual({ ok: true });
+    expect(reported.map((error) => (error as Error).message)).toEqual(['a subscriber failed']);
+  } finally {
+    globalThis.reportError = report;
+    client.close();
+    await server.stop(true);
+  }
+});
diff --git a/packages/core/test/ws-client.test.ts b/packages/core/test/ws-client.test.ts
index 98c38165a..822fb8ed7 100644
--- a/packages/core/test/ws-client.test.ts
+++ b/packages/core/test/ws-client.test.ts
@@ -1,5 +1,6 @@
 import { expect, test } from 'bun:test';
 import { createHash } from 'node:crypto';
+import { createServer, type Socket } from 'node:net';
 import { constants as zlibConstants, deflateRawSync } from 'node:zlib';
 import { encodeFrame, FrameReader, TextSocket, type CloseEvent, type Frame } from '../src/ws-client.ts';
 
@@ -91,7 +92,7 @@ test('the frame reader refuses what Bun mistook for frames: reserved bits, masks
   const reader = new FrameReader();
   expect(reader.push(join([serverFrame(1, 'kept'), serverFrame(1, '{}', true, 4), serverFrame(1, 'lost')]))).toEqual([
     { kind: 'text', text: 'kept' },
-    { kind: 'invalid', reason: 'reserved bits 4 set on opcode 1 with no extension negotiated' },
+    { kind: 'invalid', code: 1002, reason: 'reserved bits 4 set on opcode 1 with no extension negotiated' },
   ]);
   expect(reader.push(serverFrame(1, 'later'))).toEqual([]);
 });
@@ -304,9 +305,9 @@ test('with deflate negotiated, a compressed message is read whole whatever its f
   expect(read(() => 1 + Math.floor(random() * 700))).toEqual(want);
 
   const refusal = (frames: Uint8Array[]) => new FrameReader({ inflate: true }).push(join(frames)).at(-1);
-  expect(refusal([serverFrame(1, 'a', false), serverFrame(0, 'b', true, 4)])).toEqual({ kind: 'invalid', reason: 'reserved bits 4 set on opcode 0' });
-  expect(refusal([serverFrame(9, 'p', true, 4)])).toEqual({ kind: 'invalid', reason: 'reserved bits 4 set on opcode 9' });
-  expect(refusal([serverFrame(1, '{}', true, 2)])).toEqual({ kind: 'invalid', reason: 'reserved bits 2 set on opcode 1' });
+  expect(refusal([serverFrame(1, 'a', false), serverFrame(0, 'b', true, 4)])).toEqual({ kind: 'invalid', code: 1002, reason: 'reserved bits 4 set on opcode 0' });
+  expect(refusal([serverFrame(9, 'p', true, 4)])).toEqual({ kind: 'invalid', code: 1002, reason: 'reserved bits 4 set on opcode 9' });
+  expect(refusal([serverFrame(1, '{}', true, 2)])).toEqual({ kind: 'invalid', code: 1002, reason: 'reserved bits 2 set on opcode 1' });
   expect(refusal([serverFrame(1, new Uint8Array([0xff, 0xff, 0xff]), true, 4)])).toMatchObject({ kind: 'invalid', reason: expect.stringContaining('does not inflate') });
 });
 
@@ -398,3 +399,118 @@ test('a close carries its code and reason, the reason cut to what a close frame
   } finally { await server.stop(true); }
   expect(closes).toEqual([[4001, 'the other machine dropped the connection'], [4001, 'é'.repeat(61)]]);
 });
+
+/** A server that answers the upgrade with `head` and sends `body` in the same write. */
+function answering(body: Uint8Array = new Uint8Array(0), head = 'Upgrade: websocket\r\nConnection: Upgrade\r\n') {
+  return Bun.listen({
+    hostname: '127.0.0.1', port: 0,
+    socket: {
+      data(socket, bytes) {
+        const key = /sec-websocket-key:\s*(\S+)/i.exec(new TextDecoder().decode(bytes))?.[1];
+        if (key === undefined) return;
+        const accept = createHash('sha1').update(`${key}258EAFA5-E914-47DA-95CA-C5AB0DC85B11`).digest('base64');
+        socket.write(join([new TextEncoder().encode(`HTTP/1.1 101 Switching Protocols\r\n${head}Sec-WebSocket-Accept: ${accept}\r\n\r\n`), body]));
+      },
+    },
+  });
+}
+
+test('a listener that throws is reported, and the socket reads on', async () => {
+  const server = answering(join([serverFrame(1, 'A'), serverFrame(1, 'B')]));
+  const report = globalThis.reportError;
+  const reported: unknown[] = [];
+  globalThis.reportError = (error) => { reported.push(error); };
+  try {
+    const socket = await TextSocket.connect(`ws://127.0.0.1:${server.port}/`, 5_000);
+    const received: string[] = [];
+    await new Promise<void>((resolve) => socket.onMessage((text) => {
+      received.push(text);
+      if (text === 'B') resolve();
+      else throw new Error('a subscriber failed');
+    }));
+    expect(received).toEqual(['A', 'B']);
+    expect(reported.map((error) => (error as Error).message)).toEqual(['a subscriber failed']);
+    expect(socket.open).toBe(true);
+    socket.close();
+  } finally {
+    globalThis.reportError = report;
+    server.stop(true);
+  }
+});
+
+test('a socket that ended keeps nothing of what it could not send', async () => {
+  // A server that stops reading once upgraded.
+  const peers: Socket[] = [];
+  const server = createServer((peer) => {
+    peers.push(peer);
+    peer.once('data', (bytes) => {
+      const key = /sec-websocket-key:\s*(\S+)/i.exec(bytes.toString())![1];
+      const accept = createHash('sha1').update(`${key}258EAFA5-E914-47DA-95CA-C5AB0DC85B11`).digest('base64');
+      peer.write(`HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: ${accept}\r\n\r\n`);
+      peer.pause();
+    });
+  });
+  await new Promise<void>((resolve) => server.listen(0, '127.0.0.1', resolve));
+  try {
+    const socket = await TextSocket.connect(`ws://127.0.0.1:${(server.address() as { port: number }).port}/`, 5_000);
+    const text = 'x'.repeat(4 << 20);
+    for (let i = 0; i < 4; i++) socket.send(text);
+    expect(socket.bufferedAmount).toBeGreaterThan(0);
+    const closed = new Promise<CloseEvent>((resolve) => socket.onClose(resolve));
+    socket.close();
+    expect((await closed).reason).toBe('the server did not answer the close');
+    expect(socket.bufferedAmount).toBe(0);
+  } finally {
+    for (const peer of peers) peer.destroy();
+    await new Promise<void>((resolve) => server.close(() => resolve()));
+  }
+}, 15_000);
+
+test('an upgrade answer without Upgrade and Connection, or with a subprotocol nobody offered, is refused', async () => {
+  const cases: Array<[string, string]> = [
+    ['Connection: Upgrade\r\n', 'without Upgrade: websocket'],
+    ['Upgrade: websocket\r\n', 'without Connection: Upgrade'],
+    ['Upgrade: websocket\r\nConnection: keep-alive, Upgrade\r\nSec-WebSocket-Protocol: surprise\r\n', 'subprotocol nobody offered: surprise'],
+  ];
+  for (const [head, error] of cases) {
+    const server = answering(undefined, head);
+    try {
+      await expect(TextSocket.connect(`ws://127.0.0.1:${server.port}/`, 5_000)).rejects.toThrow(error);
+    } finally { server.stop(true); }
+  }
+});
+
+test('a close code no peer may send ends in 1002, and text that is not UTF-8 in 1007', async () => {
+  const closeWith = (code: number) => serverFrame(8, new Uint8Array([code >> 8, code & 0xff]));
+  const cases: Array<[Uint8Array, number]> = [
+    ...[0, 999, 1004, 1005, 1006, 1015, 2000].map((code): [Uint8Array, number] => [closeWith(code), 1002]),
+    [closeWith(4001), 4001],
+    [serverFrame(1, new Uint8Array([0xc3, 0x28])), 1007],
+  ];
+  const ended: number[] = [];
+  for (const [frame] of cases) {
+    const server = answering(frame);
+    try {
+      const socket = await TextSocket.connect(`ws://127.0.0.1:${server.port}/`, 5_000);
+      ended.push((await new Promise<CloseEvent>((resolve) => socket.onClose(resolve))).code);
+    } finally { server.stop(true); }
+  }
+  expect(ended).toEqual(cases.map(([, code]) => code));
+});
+
+test('messages that came with the upgrade answer reach a late reader in order, each one\'s microtasks first', async () => {
+  const server = answering(join([serverFrame(1, 'answer'), serverFrame(1, 'event')]));
+  try {
+    const socket = await TextSocket.connect(`ws://127.0.0.1:${server.port}/`, 5_000);
+    // Both messages are in before anyone reads.
+    await Bun.sleep(50);
+    const order: string[] = [];
+    socket.onMessage((text) => {
+      order.push(text);
+      if (text === 'answer') void Promise.resolve().then(() => order.push('the caller resumed'));
+    });
+    await Bun.sleep(0);
+    expect(order).toEqual(['answer', 'the caller resumed', 'event']);
+    socket.close();
+  } finally { server.stop(true); }
+});

From 787c116f8adbf69d3c0ae424af9667080369895a Mon Sep 17 00:00:00 2001
From: meetsu <96637888+klNuno@users.noreply.github.com>
Date: Mon, 19 Oct 2026 23:17:43 +0200
Subject: [PATCH 4/4] test(projects): hold the stale folder read so the list
 must check again

---
 packages/core/test/projects-git.test.ts | 19 ++++++++++---------
 1 file changed, 10 insertions(+), 9 deletions(-)

diff --git a/packages/core/test/projects-git.test.ts b/packages/core/test/projects-git.test.ts
index 12e7d76aa..2ff18c19e 100644
--- a/packages/core/test/projects-git.test.ts
+++ b/packages/core/test/projects-git.test.ts
@@ -1,7 +1,7 @@
 import { mkdirSync, rmSync } from 'node:fs';
 import { join } from 'node:path';
 import { afterEach, beforeEach, expect, test } from 'bun:test';
-import { ProjectStore, hasGitMarker, probeGit } from '../src/projects.ts';
+import { ProjectStore, hasGitMarker, probeFolder, probeGit } from '../src/projects.ts';
 import { startTestCore, waitFor } from './harness.ts';
 import type { TestCore } from './harness.ts';
 
@@ -135,18 +135,19 @@ test('a list asked after a folder came back is not answered by a check that read
   const project = await client.call('projects.add', { path });
   rmSync(path, { recursive: true, force: true });
   expect((await client.call('projects.list', {})).find((entry) => entry.id === project.id)?.missing).toBe(true);
-  // The request below starts a check in the background, as any request naming a
-  // project does. It reads the disk while the folder is gone; on a loaded runner
-  // its answer is not in yet when the folder comes back.
+  // Any request naming the project starts a check in the background. Its folder
+  // read saw the folder gone, and on a loaded runner its answer is not in yet
+  // when the folder comes back.
   const stale = Promise.withResolvers<boolean | null>();
   const store = harness.core.projects;
-  let started = 0;
-  store.gitProbe = () => { started += 1; return stale.promise; };
-  await expect(client.call('worktrees.list', { projectId: project.id })).rejects.toThrow('does not exist any more');
-  expect(started).toBe(1);
-  // Every other folder answers at once; the stale answer lands once the list has started.
+  let reads = 0;
+  // Every other folder answers at once.
   store.gitProbe = async (folder) => (folder === path ? probeGit(folder) : null);
+  store.folderProbe = (folder) => (folder === path && ++reads === 1 ? stale.promise : probeFolder(folder));
+  store.require(project.id);
+  await waitFor(() => reads === 1);
   mkdirSync(path, { recursive: true });
+  // The stale answer lands once the list has started.
   const listFresh = store.listFresh.bind(store);
   store.listFresh = () => { const answer = listFresh(); setTimeout(() => stale.resolve(false), 0); return answer; };
   const back = (await client.call('projects.list', {})).find((entry) => entry.id === project.id);