refactor(ws): use nodebuffer socket contracts

packages/gateway-client/src/client.ts

@@ -119,22 +119,6 @@ function normalizeLowercaseStringOrEmpty(value: unknown): string {
   return typeof value === "string" ? value.trim().toLowerCase() : "";
 }
 
-function rawDataToString(data: unknown): string {
-  if (typeof data === "string") {
-    return data;
-  }
-  if (Buffer.isBuffer(data)) {
-    return data.toString("utf8");
-  }
-  if (data instanceof ArrayBuffer) {
-    return Buffer.from(data).toString("utf8");
-  }
-  if (Array.isArray(data)) {
-    return Buffer.concat(data.map((entry) => Buffer.from(entry))).toString("utf8");
-  }
-  return String(data);
-}
-
 function isSensitiveUrlQueryParamName(key: string): boolean {
   return /(?:token|password|secret|key|auth|credential)/iu.test(key);
 }
@@ -610,6 +594,7 @@ export class GatewayClient {
     }
     try {
       ws = new WebSocket(url, wsOptions as ClientOptions);
+      ws.binaryType = "nodebuffer";
     } catch (error) {
       throw error instanceof Error ? error : new Error(String(error));
     } finally {
@@ -629,9 +614,9 @@ export class GatewayClient {
       }
       this.transportValidated = true;
     });
-    ws.on("message", (data) => handlers.message(rawDataToString(data)));
+    ws.on("message", (data) => handlers.message(data.toString()));
     ws.on("close", (code, reason) => {
-      const reasonText = rawDataToString(reason);
+      const reasonText = reason.toString();
       if (this.ws === ws) {
         this.ws = null;
       }

packages/sdk/src/index.e2e.test.ts

@@ -2,7 +2,7 @@
 import type { AddressInfo } from "node:net";
 import net from "node:net";
 import { afterEach, describe, expect, it } from "vitest";
-import { WebSocketServer, type RawData, type WebSocket } from "ws";
+import { WebSocketServer, type WebSocket } from "ws";
 import { installGatewayTestHooks, startServer } from "../../../src/gateway/test-helpers.js";
 import { emitAgentEvent, registerAgentRunContext } from "../../../src/infra/agent-events.js";
 import { withTimeout } from "../../../src/utils/with-timeout.js";
@@ -31,19 +31,6 @@ function sendJson(socket: WebSocket, payload: JsonObject): void {
   socket.send(JSON.stringify(payload));
 }
 
-function readRawMessage(raw: RawData): string {
-  if (typeof raw === "string") {
-    return raw;
-  }
-  if (Buffer.isBuffer(raw)) {
-    return raw.toString("utf8");
-  }
-  if (raw instanceof ArrayBuffer) {
-    return Buffer.from(raw).toString("utf8");
-  }
-  return Buffer.concat(raw).toString("utf8");
-}
-
 async function reservePort(): Promise<number> {
   const server = net.createServer();
   await new Promise<void>((resolve) => {
@@ -64,11 +51,8 @@ async function createFakeGateway(port = 0): Promise<FakeGateway> {
   });
   let seq = 1;
   const requests: FakeGatewayRequest[] = [];
-  const sockets = new Set<WebSocket>();
-
   server.on("connection", (socket) => {
-    sockets.add(socket);
-    socket.once("close", () => sockets.delete(socket));
+    socket.binaryType = "nodebuffer";
     sendJson(socket, {
       type: "event",
       event: "connect.challenge",
@@ -77,7 +61,7 @@ async function createFakeGateway(port = 0): Promise<FakeGateway> {
     });
 
     socket.on("message", (raw) => {
-      const frame = JSON.parse(readRawMessage(raw)) as FakeGatewayRequest;
+      const frame = JSON.parse(raw.toString()) as FakeGatewayRequest;
       requests.push(frame);
       const reply = (payload: JsonObject): void => {
         sendJson(socket, { type: "res", id: frame.id, ok: true, payload });
@@ -340,10 +324,9 @@ async function createFakeGateway(port = 0): Promise<FakeGateway> {
       if (index >= 0) {
         servers.splice(index, 1);
       }
-      for (const socket of sockets) {
+      for (const socket of server.clients) {
         socket.terminate();
       }
-      sockets.clear();
       return new Promise<void>((resolve, reject) => {
         server.close((error) => (error ? reject(error) : resolve()));
       });

scripts/lib/gateway-ws-client.ts

@@ -37,19 +37,6 @@ export function resolveGatewayUrl(urlRaw: string): URL {
   return url;
 }
 
-function toText(data: WebSocket.RawData): string {
-  if (typeof data === "string") {
-    return data;
-  }
-  if (data instanceof ArrayBuffer) {
-    return Buffer.from(data).toString("utf8");
-  }
-  if (Array.isArray(data)) {
-    return Buffer.concat(data.map((chunk) => Buffer.from(chunk))).toString("utf8");
-  }
-  return Buffer.from(data as Buffer).toString("utf8");
-}
-
 export function createGatewayWsClient(params: {
   url: string;
   handshakeTimeoutMs?: number;
@@ -58,6 +45,7 @@ export function createGatewayWsClient(params: {
   onEvent?: (evt: GatewayEventFrame) => void;
 }) {
   const ws = new WebSocket(params.url, { handshakeTimeout: params.handshakeTimeoutMs ?? 8000 });
+  ws.binaryType = "nodebuffer";
   const pending = new Map<
     string,
     {
@@ -140,7 +128,7 @@ export function createGatewayWsClient(params: {
     });
 
   ws.on("message", (data) => {
-    const text = toText(data);
+    const text = data.toString();
     let frame: GatewayFrame | null;
     try {
       frame = JSON.parse(text) as GatewayFrame;

src/realtime-transcription/websocket-session.ts

@@ -1,6 +1,6 @@
 // Realtime transcription websocket session streams audio to transcription providers.
 import { randomUUID } from "node:crypto";
-import WebSocket, { type RawData } from "ws";
+import WebSocket from "ws";
 import { RetrySupervisor } from "../../packages/retry/src/index.js";
 import { sleepWithAbort } from "../infra/backoff.js";
 import { createDebugProxyWebSocketAgent, resolveDebugProxySettings } from "../proxy-capture/env.js";
@@ -59,16 +59,6 @@ const RECONNECT_STABLE_RESET_MS = 30_000;
 // they reach onMessage, replacing its 100 MiB client default.
 export const REALTIME_TRANSCRIPTION_WS_MAX_PAYLOAD_BYTES = 16 * 1024 * 1024;
 
-function rawWsDataToBuffer(data: RawData): Buffer {
-  if (Buffer.isBuffer(data)) {
-    return data;
-  }
-  if (Array.isArray(data)) {
-    return Buffer.concat(data);
-  }
-  return Buffer.from(data);
-}
-
 function defaultParseMessage(payload: Buffer): unknown {
   try {
     return JSON.parse(payload.toString()) as unknown;
@@ -266,6 +256,7 @@ class WebSocketRealtimeTranscriptionSession<Event> implements RealtimeTranscript
             maxPayload: REALTIME_TRANSCRIPTION_WS_MAX_PAYLOAD_BYTES,
             ...(proxyAgent ? { agent: proxyAgent } : {}),
           });
+          this.ws.binaryType = "nodebuffer";
         } catch (error) {
           failConnect(normalizeError(error));
           return;
@@ -286,7 +277,7 @@ class WebSocketRealtimeTranscriptionSession<Event> implements RealtimeTranscript
         });
 
         this.ws.on("message", (data) => {
-          const payload = rawWsDataToBuffer(data);
+          const payload = data as Buffer;
           this.captureFrame("inbound", payload);
           try {
             if (!this.options.onMessage) {