feat(client): add support for binary messages

Author: stainless-app[bot]

src/internal/ws.ts

@@ -32,14 +32,49 @@ export type ReconnectingOverrides<Parameters = Record<string, unknown>> =
     };
 
 /**
- * A bounded queue for outgoing WebSocket messages. Messages are serialized
- * on enqueue and held until the connection is ready to flush them. The queue
- * enforces a configurable byte-size limit (measured in UTF-8 bytes) and can
- * return the original deserialized messages via {@link drain} when the
- * connection permanently closes.
+ * Raw data types that can be sent over a WebSocket without serialization.
+ */
+export type RawWebSocketData = string | ArrayBufferLike | ArrayBufferView | Buffer[];
+
+export type UnsentMessage<T> = { type: 'message'; message: T } | { type: 'raw'; data: RawWebSocketData };
+
+type QueueEntry =
+  | { kind: 'json'; data: string; byteLength: number }
+  | { kind: 'raw'; data: RawWebSocketData; byteLength: number };
+
+/**
+ * Flatten `Buffer[]` fragments into a single `Buffer` so that `ws.send()`
+ * transmits the correct bytes.
+ */
+export function flattenRawData(data: RawWebSocketData): Exclude<RawWebSocketData, Buffer[]> {
+  if (Array.isArray(data)) return Buffer.concat(data);
+  return data;
+}
+
+function snapshotRawData(data: RawWebSocketData): Exclude<RawWebSocketData, Buffer[]> {
+  if (typeof data === 'string') return data;
+  if (Array.isArray(data)) return Buffer.concat(data);
+  if (ArrayBuffer.isView(data)) {
+    return Buffer.from(data.buffer.slice(data.byteOffset, data.byteOffset + data.byteLength));
+  }
+  return data.slice(0);
+}
+
+function rawByteLength(data: RawWebSocketData): number {
+  if (typeof data === 'string') return Buffer.byteLength(data, 'utf8');
+  if (Array.isArray(data)) return data.reduce((sum, buf) => sum + buf.byteLength, 0);
+  if ('byteLength' in data) return data.byteLength;
+  return 0;
+}
+
+/**
+ * A bounded queue for outgoing WebSocket messages. JSON messages are
+ * serialized on enqueue; raw messages are stored as-is. The queue enforces
+ * a configurable byte-size limit and can return the original messages via
+ * {@link drain} when the connection permanently closes.
  */
 export class SendQueue<T = unknown> {
-  private _queue: { data: string; byteLength: number }[] = [];
+  private _queue: QueueEntry[] = [];
   private _bytes: number = 0;
   private _maxBytes: number;
 
@@ -48,7 +83,7 @@ export class SendQueue<T = unknown> {
   }
 
   /**
-   * Serialize and enqueue a message. Returns `true` if the message was
+   * Serialize and enqueue a JSON message. Returns `true` if the message was
    * accepted, `false` if it would exceed the byte-size limit.
    */
   enqueue(event: T): boolean {
@@ -57,7 +92,22 @@ export class SendQueue<T = unknown> {
     if (this._bytes + byteLength > this._maxBytes) {
       return false;
     }
-    this._queue.push({ data, byteLength });
+    this._queue.push({ kind: 'json', data, byteLength });
+    this._bytes += byteLength;
+    return true;
+  }
+
+  /**
+   * Enqueue raw data without serialization. Returns `true` if the data was
+   * accepted, `false` if it would exceed the byte-size limit.
+   */
+  enqueueRaw(data: RawWebSocketData): boolean {
+    const snapshot = snapshotRawData(data);
+    const byteLength = rawByteLength(snapshot);
+    if (this._bytes + byteLength > this._maxBytes) {
+      return false;
+    }
+    this._queue.push({ kind: 'raw', data: snapshot, byteLength });
     this._bytes += byteLength;
     return true;
   }
@@ -67,7 +117,7 @@ export class SendQueue<T = unknown> {
    * message and all subsequent messages are re-queued and the error is
    * re-thrown so the caller can report it.
    */
-  flush(send: (data: string) => void): void {
+  flush(send: (data: string | RawWebSocketData) => void): void {
     const pending = this._queue.splice(0);
     this._bytes = 0;
     for (let i = 0; i < pending.length; i++) {
@@ -83,11 +133,14 @@ export class SendQueue<T = unknown> {
   }
 
   /**
-   * Drain the queue and return the deserialized messages. Resets byte
-   * tracking to zero.
+   * Drain the queue and return the unsent messages. JSON messages are
+   * deserialized back to their original form. Resets byte tracking to zero.
    */
-  drain(): T[] {
-    const unsent = this._queue.map((item) => JSON.parse(item.data) as T);
+  drain(): UnsentMessage<T>[] {
+    const unsent = this._queue.map((entry): UnsentMessage<T> => {
+      if (entry.kind === 'raw') return { type: 'raw', data: entry.data };
+      return { type: 'message', message: JSON.parse(entry.data) as T };
+    });
     this._queue = [];
     this._bytes = 0;
     return unsent;

src/resources/responses/internal-base.ts

@@ -6,14 +6,20 @@ import { EventEmitter } from '../../core/EventEmitter';
 import { OpenAIError } from '../../core/error';
 import { stringifyQuery } from '../../internal/utils';
 
-import type { ReconnectingEvent } from '../../internal/ws';
+import type { RawWebSocketData, ReconnectingEvent, UnsentMessage } from '../../internal/ws';
 
 export type ResponsesStreamMessage =
   | { type: 'connecting' | 'open' | 'closing' }
-  | { type: 'close'; code: number; reason: string; unsent: ResponsesAPI.ResponsesClientEvent[] }
+  | {
+      type: 'close';
+      code: number;
+      reason: string;
+      unsent: UnsentMessage<ResponsesAPI.ResponsesClientEvent>[];
+    }
   | { type: 'reconnecting'; reconnect: ReconnectingEvent }
   | { type: 'reconnected' }
   | { type: 'message'; message: ResponsesAPI.ResponsesServerEvent }
+  | { type: 'raw'; data: RawWebSocketData }
   | { type: 'error'; error: WebSocketError };
 
 export class WebSocketError extends OpenAIError {
@@ -34,8 +40,9 @@ type Simplify<T> = { [KeyType in keyof T]: T[KeyType] } & {};
 type WebSocketEvents = Simplify<
   {
     event: (event: ResponsesAPI.ResponsesServerEvent) => void;
+    raw: (data: RawWebSocketData) => void;
     error: (error: WebSocketError) => void;
-    close: (code: number, reason: string, unsent: ResponsesAPI.ResponsesClientEvent[]) => void;
+    close: (code: number, reason: string, unsent: UnsentMessage<ResponsesAPI.ResponsesClientEvent>[]) => void;
     reconnecting: (event: ReconnectingEvent) => void;
     reconnected: () => void;
   } & {
@@ -51,6 +58,11 @@ export abstract class ResponsesEmitter extends EventEmitter<WebSocketEvents> {
    */
   abstract send(event: ResponsesAPI.ResponsesClientEvent): void;
 
+  /**
+   * Send raw data over the WebSocket without JSON serialization.
+   */
+  abstract sendRaw(data: RawWebSocketData): void;
+
   /**
    * Close the WebSocket connection.
    */

src/resources/responses/ws.ts

@@ -6,9 +6,12 @@ import { InternalEventEmitter } from '../../core/EventEmitter';
 import { sleep } from '../../internal/utils/sleep';
 import {
   SendQueue,
+  flattenRawData,
   isRecoverableClose,
+  type RawWebSocketData,
   type ReconnectingEvent,
   type ReconnectingOverrides,
+  type UnsentMessage,
 } from '../../internal/ws';
 import * as ResponsesAPI from './responses';
 import { OpenAI } from '../../client';
@@ -77,7 +80,7 @@ export class ResponsesWS extends ResponsesEmitter {
     socketSwap: (oldSocket: WS.WebSocket, newSocket: WS.WebSocket) => void;
     reconnecting: (event: ReconnectingEvent<Record<string, unknown>>) => void;
     reconnected: () => void;
-    close: (code: number, reason: string, unsent: ResponsesAPI.ResponsesClientEvent[]) => void;
+    close: (code: number, reason: string, unsent: UnsentMessage<ResponsesAPI.ResponsesClientEvent>[]) => void;
   }>();
 
   constructor(client: OpenAI, options?: ResponsesWSClientOptions | null | undefined) {
@@ -110,6 +113,24 @@ export class ResponsesWS extends ResponsesEmitter {
     }
   }
 
+  sendRaw(data: RawWebSocketData) {
+    if (this._isReconnecting || this.socket.readyState === WS.WebSocket.CONNECTING) {
+      if (!this._sendQueue.enqueueRaw(data)) {
+        this._onError(null, 'send queue is full, message discarded', undefined);
+      }
+      return;
+    }
+    if (this.socket.readyState !== WS.WebSocket.OPEN) {
+      this._onError(null, 'cannot send on a closed WebSocket', undefined);
+      return;
+    }
+    try {
+      this.socket.send(flattenRawData(data));
+    } catch (err) {
+      this._onError(null, 'could not send data', err);
+    }
+  }
+
   close(props?: { code: number; reason: string }) {
     this._intentionallyClosed = true;
     this._closeCode = props?.code ?? 1000;
@@ -167,6 +188,10 @@ export class ResponsesWS extends ResponsesEmitter {
       push({ type: 'message', message: event });
     };
 
+    const onRaw = (data: RawWebSocketData) => {
+      push({ type: 'raw', data });
+    };
+
     // All errors (API + socket) funnel through _onError → 'error' event
     const onEmitterError = (err: WebSocketError) => {
       push({ type: 'error', error: err });
@@ -190,7 +215,11 @@ export class ResponsesWS extends ResponsesEmitter {
       }
     };
 
-    const onClose = (code: number, reason: string, unsent: ResponsesAPI.ResponsesClientEvent[]) => {
+    const onClose = (
+      code: number,
+      reason: string,
+      unsent: UnsentMessage<ResponsesAPI.ResponsesClientEvent>[],
+    ) => {
       push({ type: 'close', code, reason, unsent });
       done = true;
       flushResolvers();
@@ -205,6 +234,7 @@ export class ResponsesWS extends ResponsesEmitter {
 
     const cleanup = () => {
       this.off('event', onEvent);
+      this.off('raw', onRaw);
       this.off('error', onEmitterError);
       currentSocket.off('open', onOpen);
       this._internalEvents.off('close', onClose);
@@ -214,6 +244,7 @@ export class ResponsesWS extends ResponsesEmitter {
     };
 
     this.on('event', onEvent);
+    this.on('raw', onRaw);
     this.on('error', onEmitterError);
     this.socket.on('open', onOpen);
     this._internalEvents.on('close', onClose);
@@ -297,25 +328,27 @@ export class ResponsesWS extends ResponsesEmitter {
       },
     });
 
-    socket.on('message', (wsEvent) => {
-      const event = (() => {
-        try {
-          return JSON.parse(wsEvent.toString()) as ResponsesAPI.ResponsesServerEvent;
-        } catch (err) {
-          this._onError(null, 'could not parse websocket event', err);
-          return null;
-        }
-      })();
+    socket.on('message', (wsEvent, isBinary) => {
+      if (isBinary) {
+        this._emit('raw', wsEvent as RawWebSocketData);
+        return;
+      }
 
-      if (event) {
-        this._emit('event', event);
+      let event: ResponsesAPI.ResponsesServerEvent;
+      try {
+        event = JSON.parse(wsEvent.toString()) as ResponsesAPI.ResponsesServerEvent;
+      } catch {
+        this._emit('raw', wsEvent as RawWebSocketData);
+        return;
+      }
 
-        if (event.type === 'error') {
-          this._onError(event);
-        } else {
-          // @ts-ignore TS isn't smart enough to get the relationship right here
-          this._emit(event.type, event);
-        }
+      this._emit('event', event);
+
+      if (event.type === 'error') {
+        this._onError(event);
+      } else {
+        // @ts-ignore TS isn't smart enough to get the relationship right here
+        this._emit(event.type, event);
       }
     });
 
@@ -523,7 +556,9 @@ export class ResponsesWS extends ResponsesEmitter {
 
   private _flushSendQueue(): void {
     try {
-      this._sendQueue.flush((data) => this.socket.send(data));
+      this._sendQueue.flush((data) =>
+        this.socket.send(typeof data === 'string' ? data : flattenRawData(data)),
+      );
     } catch (err) {
       this._onError(null, 'could not send queued data', err);
     }