diff --git a/README.md b/README.md index a498ad841..6e1342363 100644 --- a/README.md +++ b/README.md @@ -527,7 +527,7 @@ const rt = new OpenAIRealtimeWebSocket({ model: 'gpt-realtime-2' }); rt.on('response.output_text.delta', (event) => process.stdout.write(event.delta)); ``` -For more information see [docs/realtime.md](docs/realtime.md). +Realtime model sessions can opt in to reconnection with `reconnect: true`. Each reconnection starts a new server session, so your application must restore its state. See [the Realtime guide](docs/realtime.md) for connection options and session restoration. ## Microsoft Azure OpenAI diff --git a/docs/realtime.md b/docs/realtime.md index b39edec62..b7679b5b9 100644 --- a/docs/realtime.md +++ b/docs/realtime.md @@ -158,31 +158,47 @@ rt.on('error', (err) => { ### Disconnection and restoring a session -Realtime does not automatically retry or reconnect. That retry bound is zero: -an SDK `error` from a protocol rejection can leave the same socket usable, -while a raw socket `close` (including a transport failure) ends that connection. -Do not keep retrying permission or authentication failures. Check -`response.done` to learn the response's outcome; a close alone cannot tell -you whether a request completed. - -Only your application can decide whether to create a new session. Construct a -new client socket, or use `OpenAIRealtimeWS.create(client, options)` (and -`OpenAIRealtimeWebSocket.create` for native WebSockets) for a function-based -credential. Each factory call resolves the current credential anew. For Azure, -call the appropriate `.azure(client, options)` factory again. Register new -listeners and wait for the new socket's `open` before sending new session -configuration. Neither session settings nor conversation context are restored -automatically. Remove unwanted listeners with `rt.off(event, originalListener)`; -it detaches that listener only. - -`send()` returns `void` and writes to the current raw socket: it is not a -durable pre-open queue. Wait for `open`; report send/serialization errors via -an `error` listener. If your application queues commands, it owns that queue. -It may choose what to do with commands it never attempted. A command already -passed to `send()` is delivery-uncertain after a disconnect and must not be -replayed automatically. The SDK does not retry or transfer it to the new -socket. These rules apply to stable and beta Realtime; other WebSocket -endpoints can have different contracts. +By default, Realtime does not retry or reconnect. A protocol `error` can leave +the same socket usable, while a socket `close` ends that connection. Do not keep +retrying permission or authentication failures. Check `response.done` to learn +whether a response completed; a close alone cannot tell you its outcome. + +To create a new connection yourself, call `OpenAIRealtimeWS.create(client, options)` +or `OpenAIRealtimeWebSocket.create(client, options)`. Each call resolves the current +credential again. For Azure, call the appropriate `.azure(client, options)` factory. +Register listeners on the new connection and wait for `open` before sending. +Remove unwanted listeners with `rt.off(event, originalListener)`. + +The helpers imported from `openai/realtime/ws` and `openai/realtime/websocket` also +support `reconnect: true` for model sessions. This option does not support +transcription sessions or connections to an existing call. Helpers imported from +`openai/beta/realtime/ws` and `openai/beta/realtime/websocket` keep their existing +behavior and do not reconnect. + +With reconnection enabled, the SDK keeps the same connection object but replaces +its `.socket`. Each replacement starts a new server session and refreshes +credentials. Listeners attached directly to the old socket do not move to the +new one. Use the connection's `reconnecting`, `reconnected`, and `close` events. + +The SDK allows five reconnect attempts in total by default. Set +`maxReconnectAttempts` to change the limit. Delays start at 500 ms and grow to +8 seconds; use `reconnectInitialDelay` and `reconnectMaxDelay` to change them. +Calling `close()` stops retries and prevents a pending credential refresh from +opening another socket. + +Wait for the initial `open` before sending. During reconnection, the SDK rejects +sends by default. Set `maxQueueBytes` to retain unsent commands up to that byte +limit. After `reconnected`, restore any session settings and conversation state +your application needs, then call `flushPending()` to send the retained commands +in order. Pause other application sends during restoration. To remove retained +commands without sending them, call `takePending()`. The final `close` event +includes remaining unsent commands as its third argument. + +`send()` returns `void`; handle send and serialization errors with an `error` +listener. The SDK never resends a command it already tried to send. A successful +send does not confirm receipt or processing by the server, and a disconnect can +leave that outcome unknown. The pending queue stays in memory; it is not durable +storage. The SDK does not restore server-side state automatically. ## Realtime translation diff --git a/scripts/test-packed-package.ts b/scripts/test-packed-package.ts index 0b2911508..5011130ed 100644 --- a/scripts/test-packed-package.ts +++ b/scripts/test-packed-package.ts @@ -222,6 +222,7 @@ const packedPackagePath = require('node:path'); " on(event: 'message', listener: (data: ArrayBuffer | Uint8Array | Uint8Array[], isBinary: boolean) => void): this;", " on(event: 'error', listener: (error: Error) => void): this;", " on(event: 'redirect', listener: (url: string, request: { getHeaders(): Record }) => void): this;", + " on(event: 'unexpected-response', listener: (request: unknown, response: { statusCode?: number; resume(): void }) => void): this;", ' on(event: string, listener: (...args: unknown[]) => void): this;', ' removeListener(event: string, listener: (...args: unknown[]) => void): this;', ' }', diff --git a/src/internal/realtime-recovery.ts b/src/internal/realtime-recovery.ts new file mode 100644 index 000000000..7583ec30a --- /dev/null +++ b/src/internal/realtime-recovery.ts @@ -0,0 +1,118 @@ +import { WebSocketConnection } from './ws-connection-controller'; +import type { WebSocketLike } from './ws-adapter'; +import { isRecoverableClose } from './ws'; +import type { ReconnectingEvent, UnsentMessage } from './ws'; + +const MAX_TIMER_DELAY_MS = 2_147_483_647; + +/** Optional transport recovery. Reconnecting creates a new session; it does not restore server state. */ +export interface RealtimeRecoveryOptions { + /** Enable automatic transport recovery. Default: false. */ + reconnect?: boolean | undefined; + /** Additional attempts over the connection's entire lifetime. Default: 5. */ + maxReconnectAttempts?: number | undefined; + /** Initial retry delay in milliseconds. Default: 500. */ + reconnectInitialDelay?: number | undefined; + /** Maximum retry delay in milliseconds. Default: 8000. */ + reconnectMaxDelay?: number | undefined; + /** Bytes retained while disconnected. Default: 0 (reject instead of retaining). */ + maxQueueBytes?: number | undefined; +} + +/** Validate opt-in settings before opening a socket. */ +export function validateRealtimeRecovery(options: RealtimeRecoveryOptions, url?: URL): void { + if (options.reconnect && (url?.searchParams.has('intent') || url?.searchParams.has('call_id'))) { + throw new Error('Automatic Realtime recovery requires a model session'); + } + const delays = [options.reconnectInitialDelay, options.reconnectMaxDelay]; + for (const value of [options.maxReconnectAttempts, ...delays, options.maxQueueBytes]) { + if (value !== undefined && (!Number.isSafeInteger(value) || value < 0)) { + throw new Error('Realtime recovery limits must be nonnegative safe integers'); + } + } + if (delays.some((delay) => delay !== undefined && delay > MAX_TIMER_DELAY_MS)) { + throw new Error(`Realtime reconnect delays must not exceed ${MAX_TIMER_DELAY_MS} ms`); + } +} + +/** Adapts the existing protocol controller to Realtime's explicit restoration/flush policy. */ +export function createRealtimeRecovery< + TSocket extends WebSocketLike & { readonly platformSocket: unknown }, + TEvent, +>( + owner: { socket: TSocket['platformSocket']; url: URL }, + options: RealtimeRecoveryOptions, + hooks: { + buildURL: () => URL; + adaptSocket: (socket: TSocket['platformSocket']) => TSocket; + prepare: () => Promise<(url: URL) => TSocket['platformSocket']>; + onError: (message: string, cause?: unknown) => void; + onReconnecting: (event: ReconnectingEvent) => void; + onReconnected: () => void; + shouldReconnect?: (code: number) => boolean; + onClose: (code: number, reason: string, unsent: UnsentMessage[]) => void; + }, +): WebSocketConnection { + const first = hooks.adaptSocket(owner.socket); + let currentSocket = first; + const controllerOwner = { + get socket() { + return currentSocket; + }, + set socket(socket: TSocket) { + currentSocket = socket; + owner.socket = socket.platformSocket; + }, + get url() { + return owner.url; + }, + set url(url: URL) { + owner.url = url; + }, + }; + const controller = new WebSocketConnection( + controllerOwner, + { + buildURL: hooks.buildURL, + createSocket: () => first, + prepareReconnectSocket: async () => { + const createSocket = await hooks.prepare(); + return (url) => hooks.adaptSocket(createSocket(url)); + }, + parameters: undefined, + isErrorEvent: () => false, + onMessage: () => { + /* The Realtime wrapper owns decoding and event dispatch. */ + }, + onError: hooks.onError, + createError: (message) => new Error(message), + subscribe: () => { + /* Realtime exposes its existing event emitter, not the controller iterator. */ + }, + unsubscribe: () => { + /* Realtime exposes its existing event emitter, not the controller iterator. */ + }, + emitReconnecting: hooks.onReconnecting, + emitReconnected: hooks.onReconnected, + emitClose: hooks.onClose, + requeueFailed: false, + manualQueue: true, + lifetimeRetries: true, + shouldReconnect: (code) => + code !== 1015 && isRecoverableClose(code) && (hooks.shouldReconnect?.(code) ?? true), + }, + { + maxQueueSize: options.maxQueueBytes ?? 0, + reconnect: { + onReconnecting: () => { + /* The boolean enables recovery without a required user callback. */ + }, + maxRetries: options.maxReconnectAttempts ?? 5, + initialDelay: options.reconnectInitialDelay ?? 500, + maxDelay: options.reconnectMaxDelay ?? 8000, + }, + }, + ); + controller.connect(() => first, owner.url); + return controller; +} diff --git a/src/internal/ws-adapter-browser.ts b/src/internal/ws-adapter-browser.ts index e935ddfc5..828473d79 100644 --- a/src/internal/ws-adapter-browser.ts +++ b/src/internal/ws-adapter-browser.ts @@ -15,7 +15,7 @@ interface WebSocket { binaryType: string; /** Sends a text or binary WebSocket frame. */ - send(data: string | ArrayBufferLike | ArrayBufferView): void; + send(data: string | ArrayBufferLike | ArrayBufferView | Blob): void; /** Initiates a closing handshake with an optional status code and reason. */ close(code?: number, reason?: string): void; @@ -39,7 +39,7 @@ interface CloseEvent { /** * Adapts a browser-native WebSocket to the SDK's platform-neutral event API. * - * Binary messages are received as `ArrayBuffer` values, browser event objects + * Binary messages are received as `ArrayBuffer` values by default, browser event objects * are converted into positional listener arguments, and browser errors are * normalized to `Error` instances. */ @@ -47,10 +47,12 @@ export class BrowserWebSocket implements WebSocketLike { private _ws: WebSocket; private _listenerMap = new Map>(); - /** Wraps an existing browser socket and configures binary frames as array buffers. */ - constructor(ws: WebSocket) { + /** Wraps an existing browser socket, preserving its binary mode for lifecycle-only use if requested. */ + constructor(ws: WebSocket, options?: { preserveBinaryType?: boolean }) { this._ws = ws; - this._ws.binaryType = 'arraybuffer'; + if (!options?.preserveBinaryType) { + this._ws.binaryType = 'arraybuffer'; + } } /** The browser-native socket; accessing it makes calling code platform-specific. */ diff --git a/src/internal/ws-adapter-node.ts b/src/internal/ws-adapter-node.ts index 666edea1d..fcdff59f2 100644 --- a/src/internal/ws-adapter-node.ts +++ b/src/internal/ws-adapter-node.ts @@ -124,6 +124,11 @@ export class NodeWebSocket implements WebSocketLike { this._ws.close(code, reason); } + /** Forcibly closes the underlying transport. */ + terminate(): void { + this._ws.terminate(); + } + /** Registers a listener with normalized message payloads and close reasons. */ on(event: string, listener: Listener): void { const wrapped = NodeWebSocket._wrapListener(event, listener); diff --git a/src/internal/ws-adapter.ts b/src/internal/ws-adapter.ts index 3209cf19e..7aa084756 100644 --- a/src/internal/ws-adapter.ts +++ b/src/internal/ws-adapter.ts @@ -12,6 +12,9 @@ export interface WebSocketLike { /** Initiates the closing handshake, optionally providing a close code and reason. */ close(code?: number, reason?: string): void; + /** Forcibly closes the transport when a graceful close fails, if the platform supports it. */ + terminate?(): void; + /** Registers a listener that runs when the socket connection opens. */ on(event: 'open', listener: () => void): void; diff --git a/src/internal/ws-connection-controller.ts b/src/internal/ws-connection-controller.ts index 3f851c99e..fcd2407e8 100644 --- a/src/internal/ws-connection-controller.ts +++ b/src/internal/ws-connection-controller.ts @@ -73,6 +73,11 @@ export interface WebSocketConnectionHooks Promise<(url: URL) => TSocket>; /** Preserve whether a failed queued write remains in the legacy unsent list. */ requeueFailed?: boolean; + /** Retain unsent events until the application explicitly restores state and flushes. */ + manualQueue?: boolean; + /** Bound retries across this logical connection, rather than per disconnect. */ + lifetimeRetries?: boolean; + shouldReconnect?: (code: number) => boolean; } type WebSocketParameters = NonNullable; @@ -113,12 +118,64 @@ export class WebSocketConnection< this._owner = owner; this._hooks = hooks; this._reconnectOptions = options?.reconnect ?? null; - this._sendQueue = new SendQueue(options?.maxQueueSize); + this._sendQueue = new SendQueue( + options?.maxQueueSize ?? (hooks.manualQueue ? 0 : undefined), + !hooks.manualQueue, + ); + } + + private _retryCount = 0; + private _permanentClosed = false; + private _backoff = new AbortController(); + + /** Whether the logical connection has emitted its terminal close. */ + get isClosed(): boolean { + return this._permanentClosed; + } + + takePending(): UnsentMessage[] { + return this._sendQueue.drain(); + } + + flushPending(): void { + if ( + this._intentionallyClosed || + this._permanentClosed || + this._isReconnecting || + this._owner.socket.readyState !== ReadyState.OPEN + ) { + this._hooks.onError('cannot flush before the WebSocket is open'); + return; + } + this._flushSendQueue(true); } private _cancelRefresh: (() => void) | undefined; send(event: TClientEvent) { + if (this._hooks.manualQueue) { + if (this._intentionallyClosed || this._permanentClosed) { + this._hooks.onError('cannot send on a closed WebSocket'); + return; + } + if (this._isReconnecting) { + let accepted: boolean; + try { + accepted = this._sendQueue.enqueue(event); + } catch (error) { + this._hooks.onError('could not send data', error); + return; + } + if (!accepted) { + this._hooks.onError('send queue is full, message discarded'); + } + return; + } + if (this._owner.socket.readyState === ReadyState.CONNECTING) { + this._hooks.onError('cannot send before the WebSocket is open'); + return; + } + } sendWebSocketEvent( this._owner, event, @@ -147,10 +204,26 @@ export class WebSocketConnection< this._closeCode = props?.code ?? 1000; this._closeReason = props?.reason ?? 'OK'; this._cancelRefresh?.(); + this._backoff.abort(); try { this._owner.socket.close(this._closeCode, this._closeReason); } catch (error) { - this._hooks.onError('could not close the connection', error); + try { + this._hooks.onError('could not close the connection', error); + } finally { + if (this._hooks.manualQueue) { + try { + const { socket } = this._owner; + if (socket.terminate) { + socket.terminate(); + } else { + socket.close(); + } + } finally { + this._emitPermanentClose(this._closeCode, this._closeReason); + } + } + } } } @@ -391,9 +464,16 @@ export class WebSocketConnection< const socket = createSocket(this._owner.url); - socket.on('message', (data, isBinary) => this._hooks.onMessage(data, isBinary)); + socket.on('message', (data, isBinary) => { + if (socket === this._owner.socket) { + this._hooks.onMessage(data, isBinary); + } + }); socket.on('error', (err: Error) => { + if (socket !== this._owner.socket) { + return; + } this._hooks.onSocketError?.(socket, err); // Suppress transient errors during reconnection — the retry loop // already handles them and will surface a close if retries exhaust. @@ -404,6 +484,9 @@ export class WebSocketConnection< }); socket.on('open', () => { + if (socket !== this._owner.socket) { + return; + } this._flushSendQueue(); }); @@ -426,7 +509,7 @@ export class WebSocketConnection< // Reconnect is opt-in via onReconnecting so callers can pass // state (e.g. session IDs) into the new connection. private _canReconnect(code: number): boolean { - if (this._intentionallyClosed) { + if (this._intentionallyClosed || this._permanentClosed) { return false; } if (!this._reconnectOptions) { @@ -438,21 +521,32 @@ export class WebSocketConnection< if (!this._reconnectOptions.onReconnecting) { return false; } - return isRecoverableClose(code); + return this._hooks.shouldReconnect?.(code) ?? isRecoverableClose(code); + } + + private _reportReconnectFailure(message: string, cause?: unknown): void { + // Preserve the legacy automatic-queue clients' reconnect error contract. + this._hooks.onError(message, this._hooks.manualQueue ? cause : undefined); } - private _abortReconnect(closeCode: number): boolean { + private _abortReconnect(closeCode: number, cause?: unknown): boolean { if (this._canReconnect(closeCode)) { return false; } this._isReconnecting = false; - if (!this._intentionallyClosed) { - this._hooks.onError(`WebSocket reconnect aborted: non-recoverable close code ${closeCode}`); + try { + if (!this._intentionallyClosed && !this._permanentClosed) { + this._reportReconnectFailure( + `WebSocket reconnect aborted: non-recoverable close code ${closeCode}`, + cause, + ); + } + } finally { + this._emitPermanentClose( + this._intentionallyClosed ? this._closeCode : closeCode, + this._intentionallyClosed ? this._closeReason : 'reconnect aborted', + ); } - this._emitPermanentClose( - this._intentionallyClosed ? this._closeCode : closeCode, - this._intentionallyClosed ? this._closeReason : 'reconnect aborted', - ); return true; } @@ -467,8 +561,11 @@ export class WebSocketConnection< overrides = options.onReconnecting(reconnectingEvent); } catch (error) { this._isReconnecting = false; - this._hooks.onError('onReconnecting callback threw', error); - this._emitPermanentClose(closeCode, 'onReconnecting callback threw'); + try { + this._hooks.onError('onReconnecting callback threw', error); + } finally { + this._emitPermanentClose(closeCode, 'onReconnecting callback threw'); + } return false; } @@ -486,12 +583,30 @@ export class WebSocketConnection< try { this._hooks.emitReconnecting(reconnectingEvent); } catch (error) { + if (this._hooks.manualQueue) { + this._isReconnecting = false; + try { + this._hooks.onError('onReconnecting callback threw', error); + } finally { + this._emitPermanentClose(closeCode, 'onReconnecting callback threw'); + } + return false; + } this._hooks.onError('onReconnecting callback threw', error); } this._internalEvents._emit('reconnecting', reconnectingEvent); return true; } + private _hasRetryBudget(attempt: number, maximum: number): boolean { + return attempt <= maximum && (!this._hooks.lifetimeRetries || this._retryCount < maximum); + } + + private _nextRetryAttempt(attempt: number): number { + this._retryCount += 1; + return this._hooks.lifetimeRetries ? this._retryCount : attempt; + } + private async _reconnect(initialCloseCode: number): Promise { let closeCode = initialCloseCode; if (!this._owner.socket) { @@ -506,10 +621,12 @@ export class WebSocketConnection< const maxRetries = this._reconnectOptions.maxRetries ?? 5; const initialDelay = this._reconnectOptions.initialDelay ?? 500; const maxDelay = this._reconnectOptions.maxDelay ?? 8000; + let lastError: unknown; /* oxlint-disable eslint/no-await-in-loop -- Reconnect attempts and their credential refresh must run sequentially. */ - for (let attempt = 1; attempt <= maxRetries; attempt += 1) { - if (this._abortReconnect(closeCode)) { + for (let attempt = 1; this._hasRetryBudget(attempt, maxRetries); attempt += 1) { + const reportedAttempt = this._nextRetryAttempt(attempt); + if (this._abortReconnect(closeCode, lastError)) { return; } @@ -519,7 +636,7 @@ export class WebSocketConnection< const actualDelay = Math.round(baseDelay * jitter); const reconnectingEvent: ReconnectingEvent = { - attempt, + attempt: reportedAttempt, maxAttempts: maxRetries, delay: actualDelay, closeCode, @@ -530,19 +647,24 @@ export class WebSocketConnection< return; } - if (this._abortReconnect(closeCode)) { + if (this._abortReconnect(closeCode, lastError)) { return; } - await sleep(actualDelay); + try { + await sleep(actualDelay, this._backoff.signal); + } catch { + this._abortReconnect(closeCode, lastError); + return; + } - if (this._abortReconnect(closeCode)) { + if (this._abortReconnect(closeCode, lastError)) { return; } let closeCodePromise: Promise | undefined; + const oldSocket = this._owner.socket; try { - const oldSocket = this._owner.socket; if (this._hooks.prepareReconnectSocket) { const url = this._hooks.buildURL(); // oxlint-disable-next-line promise/avoid-new -- Public close cancels the pending credential refresh through this callback. @@ -574,15 +696,8 @@ export class WebSocketConnection< }); await waitForWebSocketOpen(this._owner.socket); - - this._internalEvents._emit('socketSwap', oldSocket, this._owner.socket); - this._isReconnecting = false; - this._flushSendQueue(); - this._hooks.emitReconnected(); - this._internalEvents._emit('reconnected'); - return; - } catch { - if (this._hooks.prepareReconnectSocket && this._intentionallyClosed) { + } catch (error) { + if (this._intentionallyClosed) { // A public close observer may have thrown after lifecycle cleanup. // Only emit here if the reconnect itself had not already closed. if (this._isReconnecting) { @@ -591,23 +706,40 @@ export class WebSocketConnection< } return; } + lastError = error; if (closeCodePromise) { // ws may emit 'error' before 'close', so await the code // rather than reading it synchronously. closeCode = await closeCodePromise; } + continue; } + this._isReconnecting = false; + this._internalEvents._emit('socketSwap', oldSocket, this._owner.socket); + this._flushSendQueue(); + this._hooks.emitReconnected(); + this._internalEvents._emit('reconnected'); + return; } /* oxlint-enable eslint/no-await-in-loop */ // All retries exhausted — surface an error so consumers can // distinguish retry failure from a clean close. this._isReconnecting = false; - this._hooks.onError(`WebSocket reconnect failed after ${maxRetries} attempts (close code: ${closeCode})`); - this._emitPermanentClose(closeCode, `reconnect failed after ${maxRetries} attempts`); + try { + this._reportReconnectFailure( + `WebSocket reconnect failed after ${maxRetries} attempts (close code: ${closeCode})`, + lastError, + ); + } finally { + this._emitPermanentClose(closeCode, `reconnect failed after ${maxRetries} attempts`); + } } - private _flushSendQueue(): void { + private _flushSendQueue(explicit = false): void { + if (this._hooks.manualQueue && !explicit) { + return; + } flushWebSocketSendQueue( this._owner, this._sendQueue, @@ -621,6 +753,10 @@ export class WebSocketConnection< * `close` event used by the async iterator. */ private _emitPermanentClose(code: number, reason: string): void { + if (this._permanentClosed) { + return; + } + this._permanentClosed = true; this._lastCloseCode = code; this._lastCloseReason = reason; const unsent = this._sendQueue.drain(); diff --git a/src/internal/ws.ts b/src/internal/ws.ts index 004572575..809642fca 100644 --- a/src/internal/ws.ts +++ b/src/internal/ws.ts @@ -384,8 +384,11 @@ export class SendQueue { private _bytes = 0; private _maxBytes: number; + private readonly allowOversizedFirstMessage: boolean; + /** Creates a queue with a one-mebibyte default byte budget. */ - constructor(maxBytes = 1_048_576) { + constructor(maxBytes = 1_048_576, allowOversizedFirstMessage = true) { + this.allowOversizedFirstMessage = allowOversizedFirstMessage; this._maxBytes = maxBytes; } @@ -397,8 +400,14 @@ export class SendQueue { */ enqueue(event: T): boolean { const data = JSON.stringify(event); + if (data === undefined) { + throw new TypeError('WebSocket event could not be serialized as JSON'); + } const byteLength = encodeUTF8(data).byteLength; - if (this._bytes + byteLength > this._maxBytes && this._queue.length > 0) { + if ( + this._bytes + byteLength > this._maxBytes && + (!this.allowOversizedFirstMessage || this._queue.length > 0) + ) { return false; } this._queue.push({ kind: 'json', data, byteLength }); @@ -416,7 +425,10 @@ export class SendQueue { enqueueRaw(data: RawWebSocketData): boolean { const snapshot = snapshotRawData(data); const byteLength = rawByteLength(snapshot); - if (this._bytes + byteLength > this._maxBytes && this._queue.length > 0) { + if ( + this._bytes + byteLength > this._maxBytes && + (!this.allowOversizedFirstMessage || this._queue.length > 0) + ) { return false; } this._queue.push({ kind: 'raw', data: snapshot, byteLength }); diff --git a/src/realtime/internal-base.ts b/src/realtime/internal-base.ts index 783de273c..9e69dde4d 100644 --- a/src/realtime/internal-base.ts +++ b/src/realtime/internal-base.ts @@ -1,3 +1,5 @@ +import type { RealtimeRecoveryOptions } from '../internal/realtime-recovery'; +import type { ReconnectingEvent, UnsentMessage } from '../internal/ws'; import type { RealtimeClientEvent, RealtimeServerEvent, @@ -75,6 +77,12 @@ type RealtimeEvents = Simplify< { /** Receives every server event before its event-specific listener is notified. */ event: (event: RealtimeServerEvent) => void; + /** A physical replacement is about to be attempted. */ + reconnecting: (event: ReconnectingEvent) => void; + /** The new socket is open. Restore session state before flushing retained events. */ + reconnected: () => void; + /** Permanent closure, including never-attempted retained events. */ + close: (code: number, reason: string, unsent: UnsentMessage[]) => void; /** Receives API-reported errors and client-side WebSocket failures. */ error: (error: OpenAIRealtimeError) => void; @@ -154,7 +162,7 @@ export function isAzure(client: Pick): client is A return client instanceof AzureOpenAI; } -interface RealtimeURLBuilderOptions { +interface RealtimeURLBuilderOptions extends RealtimeRecoveryOptions { /** * Builds the exact WebSocket URL after the connection target has been validated. * @@ -319,7 +327,14 @@ export function getAzureRealtimeConnection( const hasCallID = connection.callID !== undefined; const hasIntent = connection.intent !== undefined; const customURLBuilder = connection.buildRealtimeURL; - const normalizedOptions = customURLBuilder ? { buildRealtimeURL: customURLBuilder } : {}; + const normalizedOptions = { + ...(customURLBuilder ? { buildRealtimeURL: customURLBuilder } : {}), + reconnect: connection.reconnect, + maxReconnectAttempts: connection.maxReconnectAttempts, + reconnectInitialDelay: connection.reconnectInitialDelay, + reconnectMaxDelay: connection.reconnectMaxDelay, + maxQueueBytes: connection.maxQueueBytes, + }; if ( Number(hasDeploymentName) + Number(hasCallID) + Number(hasIntent) > 1 || diff --git a/src/realtime/websocket.ts b/src/realtime/websocket.ts index e3ce88ee4..8b759eee1 100644 --- a/src/realtime/websocket.ts +++ b/src/realtime/websocket.ts @@ -1,3 +1,7 @@ +import { BrowserWebSocket } from '../internal/ws-adapter-browser'; +import { createRealtimeRecovery, validateRealtimeRecovery } from '../internal/realtime-recovery'; +import type { WebSocketConnection } from '../internal/ws-connection-controller'; +import type { UnsentMessage } from '../internal/ws'; import { getRealtimeAPIKey, resolveRealtimeAPIKey } from '../internal/realtime-credentials'; import type { AzureOpenAI } from '../index'; import { assertBedrockWebSocketOrigin } from '../internal/bedrock'; @@ -153,6 +157,7 @@ export class OpenAIRealtimeWebSocket extends OpenAIRealtimeEmitter { /** Underlying runtime-native WebSocket instance for connection lifecycle events. */ socket: _WebSocket; + private _recovery?: WebSocketConnection; /** * Immediately opens a native WebSocket session or attaches to an existing call. @@ -184,6 +189,10 @@ export class OpenAIRealtimeWebSocket extends OpenAIRealtimeEmitter { client?: Pick, ) { super(); + validateRealtimeRecovery(props); + if (props.reconnect && !props.model) { + throw new Error('Automatic Realtime recovery requires a model session'); + } let apiKey = getRealtimeAPIKey(client, props.__apiKey); // SAFETY: The supplied OpenAI client owns _options; these compatibility reads preserve its provider and browser-consent settings. const hasProvider = typeof (client as any)?._options?.apiKey === 'function'; @@ -198,9 +207,9 @@ export class OpenAIRealtimeWebSocket extends OpenAIRealtimeEmitter { ); } - client ??= new OpenAI({ dangerouslyAllowBrowser }); + const connectionClient = client ?? new OpenAI({ dangerouslyAllowBrowser }); if (apiKey === undefined) { - apiKey = client.apiKey; + apiKey = connectionClient.apiKey; } if (hasProvider && !props?.__resolvedApiKey) { @@ -212,62 +221,104 @@ export class OpenAIRealtimeWebSocket extends OpenAIRealtimeEmitter { ); } - this.url = buildRealtimeURL(client, props); - props.onURL?.(this.url); - if (this.url.protocol !== 'wss:') { - throw new OpenAIError('Realtime WebSocket URLs must use the wss: protocol.'); - } - assertBedrockWebSocketOrigin(client, this.url); - - const azure = isAzure(client); - const protocols = ['realtime', ...(azure ? [] : [`openai-insecure-api-key.${apiKey}`])]; - - this.socket = azure - ? createAzureWebSocket(this.url, apiKey, props.__resolvedApiKey === true, protocols) - : new WebSocket(this.url.toString(), protocols); - - this.socket.addEventListener('message', (websocketEvent: MessageEvent) => { - const event = (() => { - try { - const parsedEvent = parseRealtimeEvent(websocketEvent.data.toString()); - - if ( - typeof parsedEvent !== 'object' || - parsedEvent === null || - Array.isArray(parsedEvent) || - !Object.getOwnPropertyDescriptor(parsedEvent, 'type') || - typeof parsedEvent.type !== 'string' - ) { - throw new TypeError('Realtime WebSocket event must be an object with a string type.'); - } + const open = (attemptKey: string | null, resolvedApiKey: boolean, capturedURL?: URL) => { + // SAFETY: OpenAI owns _options; this reads its existing browser-consent flag. + const allowed = + props.dangerouslyAllowBrowser ?? + (connectionClient as any)._options?.dangerouslyAllowBrowser ?? + attemptKey?.startsWith('ek_'); + if (!allowed && isRunningInBrowserOrBrowserWorker()) { + throw new OpenAIError('Realtime browser connections require an ephemeral credential.'); + } + const url = capturedURL ?? buildRealtimeURL(connectionClient, props); + props.onURL?.(url); + validateRealtimeRecovery(props, url); + if (url.protocol !== 'wss:') { + throw new OpenAIError('Realtime WebSocket URLs must use the wss: protocol.'); + } + assertBedrockWebSocketOrigin(connectionClient, url); - return parsedEvent; - } catch (err) { - this._onError(null, 'could not parse websocket event', err); - return null; + const azure = isAzure(connectionClient); + const protocols = ['realtime', ...(azure ? [] : [`openai-insecure-api-key.${attemptKey}`])]; + + const socket = azure + ? createAzureWebSocket(url, attemptKey, resolvedApiKey === true, protocols) + : new WebSocket(url.toString(), protocols); + + socket.addEventListener('message', (websocketEvent: MessageEvent) => { + if (this._recovery && (socket !== this.socket || this._recovery.isClosed)) { + return; } - })(); + const event = (() => { + try { + const parsedEvent = parseRealtimeEvent(websocketEvent.data.toString()); + + if ( + typeof parsedEvent !== 'object' || + parsedEvent === null || + Array.isArray(parsedEvent) || + !Object.getOwnPropertyDescriptor(parsedEvent, 'type') || + typeof parsedEvent.type !== 'string' + ) { + throw new TypeError('Realtime WebSocket event must be an object with a string type.'); + } + + return parsedEvent; + } catch (err) { + this._onError(null, 'could not parse websocket event', err); + return null; + } + })(); - if (event) { - this._emit('event', event); + if (event) { + this._emit('event', event); - if (event.type === 'error') { - this._onError(event); - } else { - // @ts-expect-error TS isn't smart enough to get the relationship right here - this._emit(event.type, event); + if (this._recovery?.isClosed) { + return; + } + if (event.type === 'error') { + this._onError(event); + } else if (!this._recovery || !['close', 'reconnecting', 'reconnected'].includes(event.type)) { + // @ts-expect-error TS cannot relate the event discriminant to the mapped callback type. + this._emit(event.type, event); + } } - } - }); + }); - this.socket.addEventListener('error', (event: any) => { - // Native ErrorEvents can carry an empty message while `error` still holds the failure. - const cause = event.error ?? null; - const message = [event.message, cause?.message].find( - (value) => typeof value === 'string' && value !== '', - ); - this._onError(null, message ?? 'unknown error', cause); - }); + socket.addEventListener('error', (event: any) => { + if (this._recovery) { + return; + } + // Native ErrorEvents can carry an empty message while `error` still holds the failure. + const cause = event.error ?? null; + const message = [event.message, cause?.message].find( + (value) => typeof value === 'string' && value !== '', + ); + this._onError(null, message ?? 'unknown error', cause); + }); + return { url, socket }; + }; + const initial = open(apiKey, props.__resolvedApiKey === true); + this.url = initial.url; + this.socket = initial.socket; + if (props.reconnect) { + this._recovery = createRealtimeRecovery(this, props, { + buildURL: () => buildRealtimeURL(connectionClient, props), + adaptSocket: (socket) => new BrowserWebSocket(socket, { preserveBinaryType: true }), + prepare: async () => { + // SAFETY: The branch checks _callApiKey before using the client's credential resolver. + const credential = + '_callApiKey' in connectionClient && typeof connectionClient._callApiKey === 'function' + ? await resolveRealtimeAPIKey(connectionClient as Pick, true) + : { isProvider: false, commit: () => connectionClient.apiKey }; + return (url) => open(credential.commit(), credential.isProvider, url).socket; + }, + onError: (message, cause) => this._onError(null, message, cause), + onReconnecting: (event) => this._emit('reconnecting', event), + onReconnected: () => this._emit('reconnected'), + onClose: (code, reason, unsent) => this._emit('close', code, reason, unsent), + }); + } } /** @@ -340,6 +391,9 @@ export class OpenAIRealtimeWebSocket extends OpenAIRealtimeEmitter { * Serialization and transport failures are delivered to the SDK `error` event. */ send(event: RealtimeClientEvent): void { + if (this._recovery) { + return this._recovery.send(event); + } try { this.socket.send(JSON.stringify(event)); } catch (err) { @@ -347,6 +401,16 @@ export class OpenAIRealtimeWebSocket extends OpenAIRealtimeEmitter { } } + /** Explicitly flush never-attempted events after restoring the replacement session. */ + flushPending(): void { + this._recovery?.flushPending(); + } + + /** Remove retained events without transmitting them. */ + takePending(): UnsentMessage[] { + return this._recovery?.takePending() ?? []; + } + /** * Closes the WebSocket with status code `1000` and reason `OK` by default. * Connection-closing failures are delivered to the SDK `error` event. @@ -358,6 +422,9 @@ export class OpenAIRealtimeWebSocket extends OpenAIRealtimeEmitter { /** WebSocket close reason; defaults to `OK`. */ reason: string; }): void { + if (this._recovery) { + return this._recovery.close(props); + } try { this.socket.close(props?.code ?? 1000, props?.reason ?? 'OK'); } catch (err) { diff --git a/src/realtime/ws.ts b/src/realtime/ws.ts index 029401fa0..9b10431a0 100644 --- a/src/realtime/ws.ts +++ b/src/realtime/ws.ts @@ -1,3 +1,7 @@ +import { createRealtimeRecovery, validateRealtimeRecovery } from '../internal/realtime-recovery'; +import { NodeWebSocket, snapshotNodeWebSocketOptions } from '../internal/ws-adapter-node'; +import type { WebSocketConnection } from '../internal/ws-connection-controller'; +import type { UnsentMessage } from '../internal/ws'; import { resolveRealtimeAPIKey } from '../internal/realtime-credentials'; import * as WS from 'ws'; import { assertBedrockWebSocketOrigin } from '../internal/bedrock'; @@ -28,6 +32,7 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { /** Underlying `ws.WebSocket` instance for connection lifecycle and transport events. */ socket: WS.WebSocket; + private _recovery?: WebSocketConnection; /** * Immediately opens a model or transcription session, or attaches to an existing call. @@ -53,10 +58,14 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { client?: Pick, ) { super(); - client ??= new OpenAI(); - const apiKey = props.__apiKey === undefined ? client.apiKey : props.__apiKey; + validateRealtimeRecovery(props); + if (props.reconnect && !props.model) { + throw new Error('Automatic Realtime recovery requires a model session'); + } + const connectionClient = client ?? new OpenAI(); + const apiKey = props.__apiKey === undefined ? connectionClient.apiKey : props.__apiKey; // SAFETY: The supplied OpenAI client owns _options; this read detects its existing API-key provider without changing its public surface. - const hasProvider = typeof (client as any)?._options?.apiKey === 'function'; + const hasProvider = typeof (connectionClient as any)?._options?.apiKey === 'function'; if (hasProvider && !props.__resolvedApiKey) { throw new Error( [ @@ -65,60 +74,143 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { ].join('\n'), ); } - this.url = buildRealtimeURL(client, props); - assertBedrockWebSocketOrigin(client, this.url); - const headers = { - 'User-Agent': `${client.constructor.name}/JS ${VERSION}`, - ...props.options?.headers, - // A credential must be an own data property without invoking an inherited setter. - ...(isAzure(client) && !props.__resolvedApiKey ? {} : { Authorization: `Bearer ${apiKey}` }), - }; + const terminalErrors = new WeakMap(); + const open = ( + attemptKey: string | null, + resolvedApiKey: boolean, + capturedURL?: URL, + options: typeof props.options | ReturnType = props.options, + ) => { + const url = capturedURL ?? buildRealtimeURL(connectionClient, props); + validateRealtimeRecovery(props, url); + assertBedrockWebSocketOrigin(connectionClient, url); + if (isAzure(connectionClient) && props.__apiKey !== undefined && !attemptKey) { + throw new Error('Azure OpenAI Realtime requires an API key'); + } + const azureKey = isAzure(connectionClient) && !resolvedApiKey; + const headers = { + 'User-Agent': `${connectionClient.constructor.name}/JS ${VERSION}`, + ...options?.headers, + // Credentials remain own data properties without invoking inherited setters. + ...(azureKey && props.__apiKey !== undefined ? { 'api-key': attemptKey ?? '' } : {}), + ...(azureKey ? {} : { Authorization: `Bearer ${attemptKey}` }), + }; - this.socket = new WS.WebSocket( - this.url, - protectWebSocketOptionsFromCredentialRedirects({ - ...props.options, - headers, - }), - ); + const socket = new WS.WebSocket( + url, + protectWebSocketOptionsFromCredentialRedirects({ + ...options, + headers, + }), + ); - this.socket.on('message', (wsEvent) => { - const event = (() => { - try { - const parsedEvent = parseRealtimeEvent(wsEvent.toString()); + socket.on('message', (wsEvent) => { + if (this._recovery && (socket !== this.socket || this._recovery.isClosed)) { + return; + } + const event = (() => { + try { + const parsedEvent = parseRealtimeEvent(wsEvent.toString()); - if ( - typeof parsedEvent !== 'object' || - parsedEvent === null || - Array.isArray(parsedEvent) || - !Object.getOwnPropertyDescriptor(parsedEvent, 'type') || - typeof parsedEvent.type !== 'string' - ) { - throw new TypeError('Realtime WebSocket event must be an object with a string type.'); - } + if ( + typeof parsedEvent !== 'object' || + parsedEvent === null || + Array.isArray(parsedEvent) || + !Object.getOwnPropertyDescriptor(parsedEvent, 'type') || + typeof parsedEvent.type !== 'string' + ) { + throw new TypeError('Realtime WebSocket event must be an object with a string type.'); + } - return parsedEvent; - } catch (err) { - this._onError(null, 'could not parse websocket event', err); - return null; - } - })(); + return parsedEvent; + } catch (err) { + this._onError(null, 'could not parse websocket event', err); + return null; + } + })(); - if (event) { - this._emit('event', event); + if (event) { + this._emit('event', event); - if (event.type === 'error') { - this._onError(event); - } else { - // @ts-expect-error TS isn't smart enough to get the relationship right here - this._emit(event.type, event); + if (this._recovery?.isClosed) { + return; + } + if (event.type === 'error') { + this._onError(event); + } else if (!this._recovery || !['close', 'reconnecting', 'reconnected'].includes(event.type)) { + // @ts-expect-error TS cannot relate the event discriminant to the mapped callback type. + this._emit(event.type, event); + } } - } - }); + }); - this.socket.on('error', (err) => { - this._onError(null, err.message, err); - }); + if (props.reconnect) { + socket.on('unexpected-response', (_request, response) => { + if (![408, 429, 500, 502, 503, 504].includes(response.statusCode ?? 0)) { + terminalErrors.set( + socket, + new Error(`Unexpected server response: ${response.statusCode ?? 'unknown'}`), + ); + } + response.resume(); + socket.terminate(); + }); + } + socket.on('error', (err) => { + if (this._recovery) { + if ( + 'code' in err && + typeof err.code === 'string' && + (err.code.startsWith('ERR_TLS_') || + err.code.startsWith('CERT_') || + [ + 'DEPTH_ZERO_SELF_SIGNED_CERT', + 'SELF_SIGNED_CERT_IN_CHAIN', + 'UNABLE_TO_VERIFY_LEAF_SIGNATURE', + 'UNABLE_TO_GET_ISSUER_CERT_LOCALLY', + ].includes(err.code)) + ) { + terminalErrors.set(socket, err); + } + return; + } + this._onError(null, err.message, err); + }); + return { url, socket }; + }; + const initial = open(apiKey, props.__resolvedApiKey === true); + this.url = initial.url; + this.socket = initial.socket; + if (props.reconnect) { + this._recovery = createRealtimeRecovery(this, props, { + buildURL: () => buildRealtimeURL(connectionClient, props), + adaptSocket: (socket) => new NodeWebSocket(socket), + prepare: async () => { + const options = snapshotNodeWebSocketOptions(props.options); + // SAFETY: The branch checks _callApiKey before using the client's credential resolver. + const credential = + '_callApiKey' in connectionClient && typeof connectionClient._callApiKey === 'function' + ? await resolveRealtimeAPIKey(connectionClient as Pick, true) + : { isProvider: false, commit: () => connectionClient.apiKey }; + return (url) => open(credential.commit(), credential.isProvider, url, options).socket; + }, + shouldReconnect: () => !terminalErrors.has(this.socket), + onError: (message, cause) => { + const terminal = terminalErrors.get(this.socket); + this._onError( + null, + terminal && terminal.message !== message ? `${message}: ${terminal.message}` : message, + terminal ?? cause, + ); + }, + onReconnecting: (event) => this._emit('reconnecting', event), + onReconnected: () => this._emit('reconnected'), + onClose: (code, reason, unsent) => { + terminalErrors.delete(this.socket); + this._emit('close', code, reason, unsent); + }, + }); + } } /** @@ -196,6 +288,9 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { * Serialization and transport failures are delivered to the SDK `error` event. */ send(event: RealtimeClientEvent): void { + if (this._recovery) { + return this._recovery.send(event); + } try { this.socket.send(JSON.stringify(event)); } catch (err) { @@ -203,6 +298,16 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { } } + /** Explicitly flush never-attempted events after restoring the replacement session. */ + flushPending(): void { + this._recovery?.flushPending(); + } + + /** Remove retained events without transmitting them. */ + takePending(): UnsentMessage[] { + return this._recovery?.takePending() ?? []; + } + /** * Closes the WebSocket with status code `1000` and reason `OK` by default. * Connection-closing failures are delivered to the SDK `error` event. @@ -214,6 +319,9 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { /** WebSocket close reason; defaults to `OK`. */ reason: string; }): void { + if (this._recovery) { + return this._recovery.close(props); + } try { this.socket.close(props?.code ?? 1000, props?.reason ?? 'OK'); } catch (err) { diff --git a/tests/lib/responsesWebSocket.test.ts b/tests/lib/responsesWebSocket.test.ts index 5fab1940c..53e605a5d 100644 --- a/tests/lib/responsesWebSocket.test.ts +++ b/tests/lib/responsesWebSocket.test.ts @@ -398,14 +398,16 @@ describe.each(variants)('%s Responses WebSocket', (_version, Base, WebSocketErro expect(listener).toHaveBeenCalledWith(1001, 'going away', []); }); - test('reconnects after recoverable closes and flushes queued messages', async () => { + test('reconnects, flushes queued messages, and honors a close in the reconnected listener', async () => { const reconnect = vi.fn(() => ({ parameters: { starting_after: 'event_123' } })); const websocket = createWebSocket({ reconnect: { onReconnecting: reconnect, maxRetries: 1, initialDelay: 0, maxDelay: 0 }, }); - const reconnected = vi.fn(); + const reconnected = vi.fn(() => websocket.close()); websocket.on('error', vi.fn()); websocket.on('reconnected', reconnected); + const iterator = websocket.stream(); + await iterator.next(); const original = websocket.socket; original.readyState = ReadyState.CLOSED; @@ -423,6 +425,9 @@ describe.each(variants)('%s Responses WebSocket', (_version, Base, WebSocketErro expect(websocket.url.toString()).toBe('wss://example.com/v1/responses?starting_after=event_123'); expect(replacement.send).toHaveBeenCalledWith('{"type":"queued-during-reconnect"}'); expect(reconnected).toHaveBeenCalledTimes(1); + await expect(iterator.next()).resolves.toMatchObject({ value: { type: 'reconnecting' } }); + await expect(iterator.next()).resolves.toMatchObject({ value: { type: 'close' } }); + await expect(iterator.next()).resolves.toEqual({ value: undefined, done: true }); }); test('emits a permanent close when a reconnect callback aborts', async () => { diff --git a/tests/realtime-wire-lifecycle.test.ts b/tests/realtime-wire-lifecycle.test.ts index 520dada71..3bdcb710e 100644 --- a/tests/realtime-wire-lifecycle.test.ts +++ b/tests/realtime-wire-lifecycle.test.ts @@ -1,5 +1,5 @@ import { once } from 'node:events'; -import { createServer } from 'node:https'; +import { Agent, createServer } from 'node:https'; import { getCACertificates, setDefaultCACertificates } from 'node:tls'; import OpenAI, { AzureOpenAI } from 'openai'; @@ -313,3 +313,442 @@ describe.each([ }, ); }); + +test.each([StableWS, StableNative])( + 'recovery only retains sends after a disconnect (%s)', + async (Realtime) => { + const connected = once(peers, 'connection'); + const realtime = new Realtime( + { ...settings, reconnect: true, maxQueueBytes: 1024 }, + new OpenAI({ baseURL, apiKey: 'ek_initial' }), + ); + const errors: Error[] = []; + realtime.on('error', (error) => errors.push(error)); + const opened = rawEvent(realtime.socket, 'open'); + realtime.send({ type: 'response.create' }); + try { + await connected; + await opened; + expect({ errors: errors.length, pending: realtime.takePending() }).toEqual({ errors: 1, pending: [] }); + } finally { + realtime.close(); + } + }, +); + +test.each([StableWS, StableNative])( + 'opt-in model recovery refreshes credentials and holds unsent events until explicit flush (%s)', + async (Realtime) => { + let resolutions = 0; + const client = new OpenAI({ baseURL, apiKey: async () => `ek_recovery_${(resolutions += 1)}` }); + let cachedKey: string | null = null; + Object.defineProperty(client, 'apiKey', { + get: () => (cachedKey === null ? null : `${cachedKey}_committed`), + set: (value: string | null) => { + cachedKey = value; + }, + }); + const connected = once(peers, 'connection'); + const realtime = await Realtime.create(client, { + ...settings, + reconnect: true, + maxQueueBytes: 1024, + reconnectInitialDelay: 1, + }); + realtime.on('error', () => {}); + const [first] = await connected; + await rawEvent(realtime.socket, 'open'); + const oldSocket = realtime.socket; + const initialBinaryType = oldSocket.binaryType; + const sent = once(first, 'message'); + realtime.send({ type: 'response.create', event_id: 'already_sent' }); + await sent; + realtime.on('reconnecting', () => realtime.send({ type: 'response.create', event_id: 'held' })); + const replacement = once(peers, 'connection'); + const recovered = realtime.emitted('reconnected'); + first.close(1012); + const [next, request] = await replacement; + const commands: unknown[] = []; + next.on('message', (data: Buffer) => commands.push(JSON.parse(String(data)))); + await recovered; + try { + expect(resolutions).toBe(2); + if (Realtime === StableWS) { + expect(request.headers.authorization).toBe('Bearer ek_recovery_2_committed'); + } else { + expect(request.headers['sec-websocket-protocol']).toContain('ek_recovery_2_committed'); + expect([initialBinaryType, realtime.socket.binaryType]).toEqual(['blob', 'blob']); + } + expect(realtime.socket).not.toBe(oldSocket); + expect(commands).toEqual([]); + realtime.send({ type: 'session.update', session: { type: 'realtime' } }); + realtime.flushPending(); + // oxlint-disable-next-line promise/avoid-new -- Wait for both ordered wire messages, independent of their scheduling. + await new Promise((resolve) => { + const check = () => { + if (commands.length === 2) { + resolve(); + } + }; + next.on('message', check); + check(); + }); + expect(commands).toEqual([ + { type: 'session.update', session: { type: 'realtime' } }, + { type: 'response.create', event_id: 'held' }, + ]); + expect(realtime.takePending()).toEqual([]); + const closed = realtime.emitted('close'); + next.close(1000, 'peer done'); + await expect(closed).resolves.toEqual([1000, 'peer done', []]); + } finally { + realtime.close(); + } + }, +); + +test('recovery captures ws options before refreshing credentials', async () => { + const options = { + ca: Buffer.from(lab.certificateAuthority), + headers: { 'x-recovery-snapshot': 'original' }, + }; + let resolutions = 0; + const client = new OpenAI({ + baseURL, + apiKey: async () => { + if ((resolutions += 1) === 2) { + options.ca.fill(0); + options.headers['x-recovery-snapshot'] = 'mutated'; + } + return `ek_snapshot_${resolutions}`; + }, + }); + const connected = once(peers, 'connection'); + const realtime = await StableWS.create(client, { + model: settings.model, + options, + reconnect: true, + maxReconnectAttempts: 1, + reconnectInitialDelay: 1, + }); + realtime.on('error', () => {}); + const [first] = await connected; + await rawEvent(realtime.socket, 'open'); + const incoming = once(peers, 'connection', { signal: AbortSignal.timeout(5000) }); + const recovered = realtime.emitted('reconnected'); + first.close(1012); + try { + const [[, request]] = await Promise.all([incoming, recovered]); + expect(request.headers['x-recovery-snapshot']).toBe('original'); + } finally { + realtime.close(); + } +}); + +test('opt-in recovery can be canceled during backoff without opening another socket', async () => { + const connected = once(peers, 'connection'); + const realtime = new StableWS( + { ...settings, reconnect: true, reconnectInitialDelay: 60_000 }, + new OpenAI({ baseURL, apiKey: 'ek_cancel' }), + ); + realtime.on('error', () => {}); + const [first] = await connected; + const originalSocket = realtime.socket; + await rawEvent(originalSocket, 'open'); + const closed = realtime.emitted('close'); + const reconnecting = realtime.emitted('reconnecting'); + first.close(1012); + await reconnecting; + realtime.close(); + expect(await closed).toEqual([1000, 'OK', []]); + expect(realtime.socket).toBe(originalSocket); +}); + +test.each([StableWS, StableNative])('recovery uses one URL per lifetime attempt (%s)', async (Realtime) => { + let builtURLs = 0; + const connected = once(peers, 'connection'); + const realtime = new Realtime( + { + ...settings, + reconnect: true, + maxReconnectAttempts: 2, + reconnectInitialDelay: 1, + buildRealtimeURL: () => { + const url = new URL('realtime', baseURL); + url.protocol = 'wss:'; + url.searchParams.set('target', String((builtURLs += 1))); + return url; + }, + }, + new OpenAI({ baseURL, apiKey: 'ek_attempt' }), + ); + realtime.on('error', () => {}); + const attempts: number[] = []; + realtime.on('reconnecting', ({ attempt }) => attempts.push(attempt)); + const reconnect = async (peer: WebSocket) => { + const incoming = once(peers, 'connection'); + const recovered = realtime.emitted('reconnected'); + peer.close(1012); + const [replacement, request] = await incoming; + await recovered; + return { + peer: replacement, + target: new URL(request.url, baseURL).searchParams.get('target'), + reported: realtime.url.searchParams.get('target'), + }; + }; + try { + const [first] = await connected; + await rawEvent(realtime.socket, 'open'); + const second = await reconnect(first); + const third = await reconnect(second.peer); + expect({ + attempts, + targets: [second.target, third.target], + reported: [second.reported, third.reported], + }).toEqual({ + attempts: [1, 2], + targets: ['2', '3'], + reported: ['2', '3'], + }); + } finally { + realtime.close(); + } +}); + +test.each([StableWS, StableNative])( + 'recovery rejects overflow and keeps wire events separate from lifecycle (%s)', + async (Realtime) => { + const connected = once(peers, 'connection'); + const realtime = new Realtime( + { ...settings, reconnect: true, maxReconnectAttempts: 1, reconnectInitialDelay: 1 }, + new OpenAI({ baseURL, apiKey: 'ek_limits' }), + ); + const errors: Error[] = []; + realtime.on('error', (error) => errors.push(error)); + const [first] = await connected; + await rawEvent(realtime.socket, 'open'); + let closes = 0; + realtime.on('close', () => { + closes += 1; + }); + for (const type of ['close', 'reconnected', 'reconnecting']) { + const event = realtime.emitted('event'); + first.send(JSON.stringify({ type })); + // oxlint-disable-next-line eslint/no-await-in-loop -- Each event acts as a barrier before checking the next wire type. + expect(await event).toEqual({ type }); + } + expect(closes).toBe(0); + realtime.on('reconnecting', () => { + realtime.send({ type: 'response.create' }); + const unserializable = { + type: 'response.create' as const, + toJSON() { + throw new Error('serialization failed'); + }, + }; + realtime.send(unserializable); + // oxlint-disable-next-line unicorn/no-useless-undefined -- Exercise JSON.stringify returning undefined. + const omitted = { type: 'response.create' as const, toJSON: () => undefined }; + realtime.send(omitted); + }); + const replacement = once(peers, 'connection'); + // oxlint-disable-next-line promise/avoid-new -- Expected admission errors must not reject this lifecycle wait. + const recovered = new Promise((resolve) => { + realtime.on('reconnected', resolve); + }); + first.close(1012); + const [next] = await replacement; + await recovered; + expect(errors).toHaveLength(3); + expect(realtime.takePending()).toEqual([]); + // oxlint-disable-next-line promise/avoid-new -- Wait for terminal cleanup while also observing expected error events. + const closed = new Promise((resolve) => { + realtime.on('close', () => resolve()); + }); + next.close(1012); + await closed; + expect(closes).toBe(1); + expect(errors).toHaveLength(4); + expect(errors[3]?.message).toContain('1 attempts'); + realtime.close(); + }, +); + +test.each([StableWS, StableNative])( + 'failed physical close ends transport and logical recovery (%s)', + async (Realtime) => { + const connected = once(peers, 'connection'); + const realtime = new Realtime( + { ...settings, reconnect: true }, + new OpenAI({ baseURL, apiKey: 'ek_close' }), + ); + const errors: Error[] = []; + realtime.on('error', (error) => errors.push(error)); + const [first] = await connected; + await rawEvent(realtime.socket, 'open'); + const physicalClose = once(first, 'close', { signal: AbortSignal.timeout(5000) }); + const rawClose = rawEvent(realtime.socket, 'close'); + const notifications: string[] = []; + realtime.on('event', ({ type }) => { + notifications.push(type); + if (type === 'session.created') { + realtime.close({ code: 1, reason: 'invalid physical close' }); + } + }); + realtime.on('session.created', () => notifications.push('session.created callback')); + // oxlint-disable-next-line promise/avoid-new -- Wait for terminal cleanup while also observing expected error events. + const closed = new Promise((resolve) => { + realtime.on('close', () => { + notifications.push('close'); + resolve(); + }); + }); + first.send(JSON.stringify({ type: 'session.created' })); + first.send(JSON.stringify({ type: 'session.updated' })); + await Promise.all([closed, physicalClose, rawClose]); + expect(notifications).toEqual(['session.created', 'close']); + realtime.send({ type: 'response.create' }); + expect(errors).toHaveLength(2); + expect(errors[1]?.message).toContain('closed WebSocket'); + }, +); + +const terminalFailures = ['initial authentication', 'authentication', 'TLS']; +test.each(terminalFailures)('recovery reports a terminal %s rejection', async (failure) => { + const tls = failure === 'TLS'; + const initial = failure === 'initial authentication'; + const endpoint = createServer({ cert: lab.server.certificate, key: lab.server.privateKey }); + let attempts = 0; + const sockets = new WebSocketServer({ + server: endpoint, + verifyClient: () => (attempts += 1) === 1 && !initial, + }); + await once(endpoint.listen(0, '127.0.0.1'), 'listening'); + const address = endpoint.address(); + if (!address || typeof address === 'string') { + throw new Error('Missing server address'); + } + const connected = initial ? undefined : once(sockets, 'connection'); + const agent = new Agent({ maxCachedSessions: 0 }); + const realtime = new StableWS( + { + ...settings, + options: { ...settings.options, agent }, + reconnect: true, + reconnectInitialDelay: 1, + maxReconnectAttempts: tls ? 5 : 1, + }, + new OpenAI({ baseURL: `https://127.0.0.1:${address.port}/v1`, apiKey: 'ek_rejected' }), + ); + const errors: Error[] = []; + realtime.on('error', (error) => errors.push(error)); + // oxlint-disable-next-line promise/avoid-new -- Authentication errors must not reject the cleanup wait. + const closed = new Promise((resolve) => { + realtime.on('close', () => resolve()); + }); + try { + if (connected) { + const [first] = await connected; + await rawEvent(realtime.socket, 'open'); + if (tls) { + endpoint.setSecureContext({ cert: lab.proxyServer.certificate, key: lab.proxyServer.privateKey }); + } + first.close(1012); + } + await closed; + expect(attempts).toBe(initial || tls ? 1 : 2); + const diagnostic = tls ? /certificate|verify/iu : /401/u; + expect(errors).toEqual([ + expect.objectContaining({ + message: expect.stringMatching(diagnostic), + cause: expect.objectContaining({ message: expect.stringMatching(diagnostic) }), + }), + ]); + expect(realtime.socket.readyState).toBe(WebSocket.CLOSED); + } finally { + realtime.close(); + agent.destroy(); + sockets.close(); + endpoint.close(); + } +}); + +test.each([StableWS, StableNative])( + 'recovery reports the latest credential failure (%s)', + async (Realtime) => { + const failures = [new Error('first refresh failed'), new Error('latest refresh failed')]; + let resolutions = 0; + const client = new OpenAI({ + baseURL, + apiKey: async () => { + resolutions += 1; + if (resolutions === 1) { + return 'ek_refresh'; + } + throw failures[resolutions - 2]; + }, + }); + const connected = once(peers, 'connection'); + const realtime = await Realtime.create(client, { + ...settings, + reconnect: true, + maxReconnectAttempts: 2, + reconnectInitialDelay: 1, + }); + const errors: Error[] = []; + realtime.on('error', (error) => errors.push(error)); + // oxlint-disable-next-line promise/avoid-new -- Expected credential errors must not reject the cleanup wait. + const closed = new Promise((resolve) => { + realtime.on('close', () => resolve()); + }); + try { + const [first] = await connected; + await rawEvent(realtime.socket, 'open'); + first.close(1012); + await closed; + expect(resolutions).toBe(3); + expect(errors).toEqual([ + expect.objectContaining({ + message: expect.stringContaining('2 attempts'), + cause: expect.objectContaining({ cause: failures[1] }), + }), + ]); + } finally { + realtime.close(); + } + }, +); + +test.each([StableWS, StableNative])('recovery rejects sideband selectors in custom URLs (%s)', (Realtime) => { + expect( + () => + new Realtime( + { + ...settings, + reconnect: true, + buildRealtimeURL: () => new URL('wss://localhost/realtime?call_id=existing'), + }, + new OpenAI({ apiKey: 'ek_sideband' }), + ), + ).toThrow('requires a model session'); +}); + +test.each([StableWS, StableNative])('recovery rejects delays outside the timer range (%s)', (Realtime) => { + for (const delay of [{ reconnectInitialDelay: 2_147_483_648 }, { reconnectMaxDelay: 2_147_483_648 }]) { + expect( + () => + new Realtime( + { + ...settings, + ...delay, + reconnect: true, + buildRealtimeURL: () => { + throw new Error('URL should not be built'); + }, + }, + new OpenAI({ apiKey: 'ek_timer' }), + ), + ).toThrow('2147483647'); + } +});