diff --git a/packages/gateway-client/src/client.ts b/packages/gateway-client/src/client.ts index 8e006b236bda..4130c3052664 100644 --- a/packages/gateway-client/src/client.ts +++ b/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; } diff --git a/packages/sdk/src/index.e2e.test.ts b/packages/sdk/src/index.e2e.test.ts index 18a54b719245..12b42e12f2da 100644 --- a/packages/sdk/src/index.e2e.test.ts +++ b/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 { const server = net.createServer(); await new Promise((resolve) => { @@ -64,11 +51,8 @@ async function createFakeGateway(port = 0): Promise { }); let seq = 1; const requests: FakeGatewayRequest[] = []; - const sockets = new Set(); - 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 { }); 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 { 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((resolve, reject) => { server.close((error) => (error ? reject(error) : resolve())); }); diff --git a/scripts/lib/gateway-ws-client.ts b/scripts/lib/gateway-ws-client.ts index 6ed413a50aa2..8cf0baae2003 100644 --- a/scripts/lib/gateway-ws-client.ts +++ b/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; diff --git a/src/realtime-transcription/websocket-session.ts b/src/realtime-transcription/websocket-session.ts index 97741001d488..d988f2449e7d 100644 --- a/src/realtime-transcription/websocket-session.ts +++ b/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 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 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) {