From 631ec2b1f8b8f92e7eb05a8877fbc22a8dfe7990 Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Sun, 18 Oct 2026 03:57:57 +0000 Subject: [PATCH 01/10] feat(realtime): add opt-in bounded connection recovery --- README.md | 20 +++ src/internal/realtime-recovery.ts | 94 +++++++++++ src/internal/ws-connection-controller.ts | 160 +++++++++++++++--- src/internal/ws.ts | 15 +- src/realtime/internal-base.ts | 19 ++- src/realtime/websocket.ts | 187 +++++++++++++++------ src/realtime/ws.ts | 181 ++++++++++++++------ tests/realtime-wire-lifecycle.test.ts | 202 +++++++++++++++++++++++ 8 files changed, 749 insertions(+), 129 deletions(-) create mode 100644 src/internal/realtime-recovery.ts diff --git a/README.md b/README.md index a498ad8415..0581f9bc28 100644 --- a/README.md +++ b/README.md @@ -529,6 +529,26 @@ rt.on('response.output_text.delta', (event) => process.stdout.write(event.delta) For more information see [docs/realtime.md](docs/realtime.md). +For GA model sessions, `reconnect: true` enables bounded transport recovery on +`OpenAIRealtimeWS` and `OpenAIRealtimeWebSocket`, including their async factories. +Recovery is off by default and does not apply to transcription or existing-call +connections. The logical connection stays the same, but its `.socket` changes. +Use the SDK's `reconnecting`, `reconnected`, and final `close` events when recovery +is enabled; listeners attached directly to an old socket do not move to the new one. + +Recovery creates a new server session. It never restores conversation state or +replays commands already attempted. `maxReconnectAttempts` defaults to five over +the connection's lifetime. Retry delays start at 500 ms and cap at 8 seconds; +`reconnectInitialDelay` and `reconnectMaxDelay` can override them. `close()` cancels +pending retries and credential refresh. Register an `error` listener as usual. + +By default, sends during recovery are rejected. Set `maxQueueBytes` to retain a +bounded number of encoded bytes instead. After restoring application-owned session +state, call `flushPending()` to send those never-attempted events in order, or +`takePending()` to remove them without sending. A final `close` event includes any +remaining unsent events as its third argument. A successful local send does not +establish that the server received or processed a command. + ## Microsoft Azure OpenAI To use this library with [Azure OpenAI](https://learn.microsoft.com/azure/ai-services/openai/overview), use the `AzureOpenAI` diff --git a/src/internal/realtime-recovery.ts b/src/internal/realtime-recovery.ts new file mode 100644 index 0000000000..3aafbe6c26 --- /dev/null +++ b/src/internal/realtime-recovery.ts @@ -0,0 +1,94 @@ +import { WebSocketConnection } from './ws-connection-controller'; +import type { WebSocketLike } from './ws-adapter'; +import { isRecoverableClose } from './ws'; +import type { ReconnectingEvent, UnsentMessage } from './ws'; + +/** 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'); + } + for (const value of [ + options.maxReconnectAttempts, + options.reconnectInitialDelay, + options.reconnectMaxDelay, + options.maxQueueBytes, + ]) { + if (value !== undefined && (!Number.isSafeInteger(value) || value < 0)) { + throw new Error('Realtime recovery limits must be nonnegative safe integers'); + } + } +} + +/** Adapts the existing protocol controller to Realtime's explicit restoration/flush policy. */ +export function createRealtimeRecovery( + owner: { socket: TSocket; url: URL }, + options: RealtimeRecoveryOptions, + hooks: { + buildURL: () => URL; + prepare: () => Promise<(url: URL) => TSocket>; + 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 = owner.socket; + const controller = new WebSocketConnection( + owner, + { + buildURL: hooks.buildURL, + createSocket: () => first, + prepareReconnectSocket: hooks.prepare, + 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-connection-controller.ts b/src/internal/ws-connection-controller.ts index 3f851c99e0..712d36738a 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,59 @@ 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(); + + get reconnecting(): boolean { + return this._isReconnecting; + } + + 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 || this._owner.socket.readyState === ReadyState.CONNECTING) { + 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; + } + } sendWebSocketEvent( this._owner, event, @@ -147,10 +199,17 @@ 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) { + this._emitPermanentClose(this._closeCode, this._closeReason); + } + } } } @@ -391,9 +450,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 +470,9 @@ export class WebSocketConnection< }); socket.on('open', () => { + if (socket !== this._owner.socket) { + return; + } this._flushSendQueue(); }); @@ -426,7 +495,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,7 +507,7 @@ export class WebSocketConnection< if (!this._reconnectOptions.onReconnecting) { return false; } - return isRecoverableClose(code); + return this._hooks.shouldReconnect?.(code) ?? isRecoverableClose(code); } private _abortReconnect(closeCode: number): boolean { @@ -446,13 +515,16 @@ export class WebSocketConnection< 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._hooks.onError(`WebSocket reconnect aborted: non-recoverable close code ${closeCode}`); + } + } 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 +539,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 +561,25 @@ 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 async _reconnect(initialCloseCode: number): Promise { let closeCode = initialCloseCode; if (!this._owner.socket) { @@ -508,7 +596,8 @@ export class WebSocketConnection< const maxDelay = this._reconnectOptions.maxDelay ?? 8000; /* oxlint-disable eslint/no-await-in-loop -- Reconnect attempts and their credential refresh must run sequentially. */ - for (let attempt = 1; attempt <= maxRetries; attempt += 1) { + for (let attempt = 1; this._hasRetryBudget(attempt, maxRetries); attempt += 1) { + this._retryCount += 1; if (this._abortReconnect(closeCode)) { return; } @@ -534,15 +623,20 @@ export class WebSocketConnection< return; } - await sleep(actualDelay); + try { + await sleep(actualDelay, this._backoff.signal); + } catch { + this._abortReconnect(closeCode); + return; + } if (this._abortReconnect(closeCode)) { 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 +668,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) { + 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) { @@ -596,18 +683,33 @@ export class WebSocketConnection< // rather than reading it synchronously. closeCode = await closeCodePromise; } + continue; } + this._isReconnecting = false; + this._internalEvents._emit('socketSwap', oldSocket, this._owner.socket); + this._internalEvents._emit('reconnected'); + this._flushSendQueue(); + this._hooks.emitReconnected(); + 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._hooks.onError( + `WebSocket reconnect failed after ${maxRetries} attempts (close code: ${closeCode})`, + ); + } 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 +723,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 004572575a..db07e1c894 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; } @@ -398,7 +401,10 @@ export class SendQueue { enqueue(event: T): boolean { const data = JSON.stringify(event); 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 +422,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 783de273c6..9e69dde4d7 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 e3ce88ee46..2b48bac2ef 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'; @@ -212,62 +221,124 @@ 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) => { + // SAFETY: OpenAI owns _options; this reads its existing browser-consent flag. + const allowed = + props.dangerouslyAllowBrowser ?? + (client as any)._options?.dangerouslyAllowBrowser ?? + attemptKey?.startsWith('ek_'); + if (!allowed && isRunningInBrowserOrBrowserWorker()) { + throw new OpenAIError('Realtime browser connections require an ephemeral credential.'); + } + const url = buildRealtimeURL(client, props); + validateRealtimeRecovery(props, url); + props.onURL?.(url); + if (url.protocol !== 'wss:') { + throw new OpenAIError('Realtime WebSocket URLs must use the wss: protocol.'); + } + assertBedrockWebSocketOrigin(client, url); - return parsedEvent; - } catch (err) { - this._onError(null, 'could not parse websocket event', err); - return null; + const azure = isAzure(client); + 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) { + 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 (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) { + let adapter = new BrowserWebSocket(initial.socket); + // oxlint-disable-next-line typescript/no-this-alias, unicorn/no-this-assignment -- The accessor receiver is the controller owner, not the public wrapper. + const instance = this; + const owner = { + get socket() { + return adapter; + }, + set socket(value: BrowserWebSocket) { + adapter = value; + // SAFETY: BrowserWebSocket wraps the native socket created by this wrapper. + instance.socket = value.platformSocket as _WebSocket; + }, + get url() { + return instance.url; + }, + set url(value: URL) { + instance.url = value; + }, + }; + this._recovery = createRealtimeRecovery(owner, props, { + buildURL: () => buildRealtimeURL(client, props), + prepare: async () => { + // SAFETY: The branch checks _callApiKey before using the client's credential resolver. + const credential = + '_callApiKey' in client && typeof client._callApiKey === 'function' + ? await resolveRealtimeAPIKey(client as Pick, true) + : { apiKey: client.apiKey, isProvider: false, commit: () => client.apiKey }; + return () => { + credential.commit(); + const next = open(credential.apiKey, credential.isProvider); + instance.url = next.url; + return new BrowserWebSocket(next.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 +411,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 +421,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 +442,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 029401fa0f..4b717f8586 100644 --- a/src/realtime/ws.ts +++ b/src/realtime/ws.ts @@ -1,3 +1,6 @@ +import { createRealtimeRecovery, validateRealtimeRecovery } from '../internal/realtime-recovery'; +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 +31,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,6 +57,10 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { client?: Pick, ) { super(); + validateRealtimeRecovery(props); + if (props.reconnect && !props.model) { + throw new Error('Automatic Realtime recovery requires a model session'); + } client ??= new OpenAI(); const apiKey = props.__apiKey === undefined ? client.apiKey : props.__apiKey; // SAFETY: The supplied OpenAI client owns _options; this read detects its existing API-key provider without changing its public surface. @@ -65,60 +73,123 @@ 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 terminalSockets = new WeakSet(); + const open = (attemptKey: string | null, resolvedApiKey: boolean) => { + const url = buildRealtimeURL(client, props); + validateRealtimeRecovery(props, url); + assertBedrockWebSocketOrigin(client, url); + if (isAzure(client) && props.__apiKey !== undefined && !attemptKey) { + throw new Error('Azure OpenAI Realtime requires an API key'); + } + const azureKey = isAzure(client) && !resolvedApiKey; + const headers = { + 'User-Agent': `${client.constructor.name}/JS ${VERSION}`, + ...props.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({ + ...props.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) { + 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 (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)) { + terminalSockets.add(socket); + } + 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)) + ) { + terminalSockets.add(socket); + } + 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(client, props), + prepare: async () => { + // SAFETY: The branch checks _callApiKey before using the client's credential resolver. + const credential = + '_callApiKey' in client && typeof client._callApiKey === 'function' + ? await resolveRealtimeAPIKey(client as Pick, true) + : { apiKey: client.apiKey, isProvider: false, commit: () => client.apiKey }; + return () => { + credential.commit(); + return open(credential.apiKey, credential.isProvider).socket; + }; + }, + shouldReconnect: () => !terminalSockets.has(this.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), + }); + } } /** @@ -181,7 +252,7 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { headers: { ...props.options?.headers, // A credential must be an own data property without invoking an inherited setter. - ...(isApiKeyProvider ? {} : { 'api-key': apiKey }), + ...(isApiKeyProvider ? {} : { 'api-key': apiKey ?? '' }), }, }, __resolvedApiKey: isApiKeyProvider, @@ -196,6 +267,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 +277,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 +298,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/realtime-wire-lifecycle.test.ts b/tests/realtime-wire-lifecycle.test.ts index 520dada719..d66468f18e 100644 --- a/tests/realtime-wire-lifecycle.test.ts +++ b/tests/realtime-wire-lifecycle.test.ts @@ -313,3 +313,205 @@ describe.each([ }, ); }); + +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)}` }); + 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 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'); + } else { + expect(request.headers['sec-websocket-protocol']).toContain('ek_recovery_2'); + } + expect(realtime.socket).not.toBe(oldSocket); + expect(commands).toEqual([]); + const flushed = once(next, 'message'); + realtime.send({ type: 'session.update', session: { type: 'realtime' } }); + realtime.flushPending(); + await flushed; + // The following server event is a wire barrier after both client commands. + // 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([]); + } 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; + await rawEvent(realtime.socket, 'open'); + const closed = realtime.emitted('close'); + realtime.on('reconnecting', () => realtime.close()); + first.close(1012); + expect(await closed).toEqual([1000, 'OK', []]); +}); + +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); + }); + 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(2); + 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(3); + expect(errors[2]?.message).toContain('1 attempts'); + realtime.close(); + }, +); + +test('failed physical close still ends logical recovery and send admission', async () => { + const connected = once(peers, 'connection'); + const realtime = new StableWS( + { ...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'); + // 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()); + }); + realtime.close({ code: 1, reason: 'invalid physical close' }); + await closed; + realtime.send({ type: 'response.create' }); + expect(errors).toHaveLength(2); + expect(errors[1]?.message).toContain('closed WebSocket'); + first.terminate(); +}); + +test('recovery stops after a permanent authentication rejection', async () => { + const endpoint = createServer({ cert: lab.server.certificate, key: lab.server.privateKey }); + let attempts = 0; + const sockets = new WebSocketServer({ server: endpoint, verifyClient: () => (attempts += 1) === 1 }); + 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 = once(sockets, 'connection'); + const realtime = new StableWS( + { ...settings, reconnect: true, reconnectInitialDelay: 1 }, + new OpenAI({ baseURL: `https://127.0.0.1:${address.port}/v1`, apiKey: 'ek_rejected' }), + ); + realtime.on('error', () => {}); + try { + const [first] = await connected; + await rawEvent(realtime.socket, 'open'); + // oxlint-disable-next-line promise/avoid-new -- Authentication errors must not reject the cleanup wait. + const closed = new Promise((resolve) => { + realtime.on('close', () => resolve()); + }); + first.close(1012); + await closed; + expect(attempts).toBe(2); + expect(realtime.socket.readyState).toBe(WebSocket.CLOSED); + } finally { + realtime.close(); + sockets.close(); + endpoint.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'); +}); From 13159e9ae04b47284c96aba6c752196e55885674 Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Sun, 18 Oct 2026 04:03:39 +0000 Subject: [PATCH 02/10] fix(realtime): remove redundant Azure credential fallback --- src/realtime/ws.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/realtime/ws.ts b/src/realtime/ws.ts index 4b717f8586..47582eebe8 100644 --- a/src/realtime/ws.ts +++ b/src/realtime/ws.ts @@ -252,7 +252,7 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { headers: { ...props.options?.headers, // A credential must be an own data property without invoking an inherited setter. - ...(isApiKeyProvider ? {} : { 'api-key': apiKey ?? '' }), + ...(isApiKeyProvider ? {} : { 'api-key': apiKey }), }, }, __resolvedApiKey: isApiKeyProvider, From afba9a407f014bc42261fbf2f888678bd2762bb2 Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Sun, 18 Oct 2026 04:24:03 +0000 Subject: [PATCH 03/10] docs: explain Realtime reconnect behavior plainly --- README.md | 42 +++++++++++++++++++++++------------------- 1 file changed, 23 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index 0581f9bc28..783567dd1b 100644 --- a/README.md +++ b/README.md @@ -529,25 +529,29 @@ rt.on('response.output_text.delta', (event) => process.stdout.write(event.delta) For more information see [docs/realtime.md](docs/realtime.md). -For GA model sessions, `reconnect: true` enables bounded transport recovery on -`OpenAIRealtimeWS` and `OpenAIRealtimeWebSocket`, including their async factories. -Recovery is off by default and does not apply to transcription or existing-call -connections. The logical connection stays the same, but its `.socket` changes. -Use the SDK's `reconnecting`, `reconnected`, and final `close` events when recovery -is enabled; listeners attached directly to an old socket do not move to the new one. - -Recovery creates a new server session. It never restores conversation state or -replays commands already attempted. `maxReconnectAttempts` defaults to five over -the connection's lifetime. Retry delays start at 500 ms and cap at 8 seconds; -`reconnectInitialDelay` and `reconnectMaxDelay` can override them. `close()` cancels -pending retries and credential refresh. Register an `error` listener as usual. - -By default, sends during recovery are rejected. Set `maxQueueBytes` to retain a -bounded number of encoded bytes instead. After restoring application-owned session -state, call `flushPending()` to send those never-attempted events in order, or -`takePending()` to remove them without sending. A final `close` event includes any -remaining unsent events as its third argument. A successful local send does not -establish that the server received or processed a command. +Set `reconnect: true` to reconnect after a temporary connection failure. +`OpenAIRealtimeWS` and `OpenAIRealtimeWebSocket` support this option, including +when you create them asynchronously. It is off by default. It does not support +transcription sessions or connections to an existing call. + +Reconnecting keeps the same connection object but replaces its `.socket` and +starts a new server session. Your application must restore any session settings +and conversation state it needs. The SDK does not resend commands it already +tried to send. Use the connection's `reconnecting`, `reconnected`, and `close` +events. Listeners attached directly to the old socket do not move to the new one. +Register an `error` listener as usual. + +The SDK allows five reconnect attempts in total by default. Set +`maxReconnectAttempts` to change this limit. Retry delays start at 500 ms and grow +to 8 seconds. Set `reconnectInitialDelay` and `reconnectMaxDelay` to change them. +Calling `close()` cancels pending retries and credential refresh. + +By default, the SDK rejects sends while reconnecting. Set `maxQueueBytes` to hold +unsent events up to that size in bytes. After you restore session state, call +`flushPending()` to send these events in order, or `takePending()` to remove them +without sending. The final `close` event includes any remaining unsent events as +its third argument. A successful send does not confirm that the server received +or processed the command. ## Microsoft Azure OpenAI From c79d6ff5f883efd6848e69647949816f8d4da3e5 Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Sun, 18 Oct 2026 04:27:07 +0000 Subject: [PATCH 04/10] fix(realtime): preserve socket adapters during recovery --- scripts/test-packed-package.ts | 1 + src/internal/realtime-recovery.ts | 35 ++++++++++++++---- src/internal/ws-adapter-browser.ts | 2 +- src/internal/ws-connection-controller.ts | 4 --- src/realtime/websocket.ts | 46 ++++++++---------------- src/realtime/ws.ts | 30 ++++++++-------- tests/realtime-wire-lifecycle.test.ts | 3 ++ 7 files changed, 64 insertions(+), 57 deletions(-) diff --git a/scripts/test-packed-package.ts b/scripts/test-packed-package.ts index 0b29115088..5011130ed8 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 index 3aafbe6c26..c53698fc45 100644 --- a/src/internal/realtime-recovery.ts +++ b/src/internal/realtime-recovery.ts @@ -35,12 +35,16 @@ export function validateRealtimeRecovery(options: RealtimeRecoveryOptions, url?: } /** Adapts the existing protocol controller to Realtime's explicit restoration/flush policy. */ -export function createRealtimeRecovery( - owner: { socket: TSocket; url: URL }, +export function createRealtimeRecovery< + TSocket extends WebSocketLike & { readonly platformSocket: unknown }, + TEvent, +>( + owner: { socket: TSocket['platformSocket']; url: URL }, options: RealtimeRecoveryOptions, hooks: { buildURL: () => URL; - prepare: () => Promise<(url: URL) => TSocket>; + adaptSocket: (socket: TSocket['platformSocket']) => TSocket; + prepare: () => Promise<(url: URL) => TSocket['platformSocket']>; onError: (message: string, cause?: unknown) => void; onReconnecting: (event: ReconnectingEvent) => void; onReconnected: () => void; @@ -48,13 +52,32 @@ export function createRealtimeRecovery( onClose: (code: number, reason: string, unsent: UnsentMessage[]) => void; }, ): WebSocketConnection { - const first = owner.socket; + 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( - owner, + controllerOwner, { buildURL: hooks.buildURL, createSocket: () => first, - prepareReconnectSocket: hooks.prepare, + prepareReconnectSocket: async () => { + const createSocket = await hooks.prepare(); + return (url) => hooks.adaptSocket(createSocket(url)); + }, parameters: undefined, isErrorEvent: () => false, onMessage: () => { diff --git a/src/internal/ws-adapter-browser.ts b/src/internal/ws-adapter-browser.ts index e935ddfc5a..07a18a35fa 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; diff --git a/src/internal/ws-connection-controller.ts b/src/internal/ws-connection-controller.ts index 712d36738a..a3d27a4f5d 100644 --- a/src/internal/ws-connection-controller.ts +++ b/src/internal/ws-connection-controller.ts @@ -128,10 +128,6 @@ export class WebSocketConnection< private _permanentClosed = false; private _backoff = new AbortController(); - get reconnecting(): boolean { - return this._isReconnecting; - } - takePending(): UnsentMessage[] { return this._sendQueue.drain(); } diff --git a/src/realtime/websocket.ts b/src/realtime/websocket.ts index 2b48bac2ef..a5dd111896 100644 --- a/src/realtime/websocket.ts +++ b/src/realtime/websocket.ts @@ -207,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) { @@ -225,20 +225,20 @@ export class OpenAIRealtimeWebSocket extends OpenAIRealtimeEmitter { // SAFETY: OpenAI owns _options; this reads its existing browser-consent flag. const allowed = props.dangerouslyAllowBrowser ?? - (client as any)._options?.dangerouslyAllowBrowser ?? + (connectionClient as any)._options?.dangerouslyAllowBrowser ?? attemptKey?.startsWith('ek_'); if (!allowed && isRunningInBrowserOrBrowserWorker()) { throw new OpenAIError('Realtime browser connections require an ephemeral credential.'); } - const url = buildRealtimeURL(client, props); + const url = buildRealtimeURL(connectionClient, props); validateRealtimeRecovery(props, url); props.onURL?.(url); if (url.protocol !== 'wss:') { throw new OpenAIError('Realtime WebSocket URLs must use the wss: protocol.'); } - assertBedrockWebSocketOrigin(client, url); + assertBedrockWebSocketOrigin(connectionClient, url); - const azure = isAzure(client); + const azure = isAzure(connectionClient); const protocols = ['realtime', ...(azure ? [] : [`openai-insecure-api-key.${attemptKey}`])]; const socket = azure @@ -299,38 +299,20 @@ export class OpenAIRealtimeWebSocket extends OpenAIRealtimeEmitter { this.url = initial.url; this.socket = initial.socket; if (props.reconnect) { - let adapter = new BrowserWebSocket(initial.socket); - // oxlint-disable-next-line typescript/no-this-alias, unicorn/no-this-assignment -- The accessor receiver is the controller owner, not the public wrapper. - const instance = this; - const owner = { - get socket() { - return adapter; - }, - set socket(value: BrowserWebSocket) { - adapter = value; - // SAFETY: BrowserWebSocket wraps the native socket created by this wrapper. - instance.socket = value.platformSocket as _WebSocket; - }, - get url() { - return instance.url; - }, - set url(value: URL) { - instance.url = value; - }, - }; - this._recovery = createRealtimeRecovery(owner, props, { - buildURL: () => buildRealtimeURL(client, props), + this._recovery = createRealtimeRecovery(this, props, { + buildURL: () => buildRealtimeURL(connectionClient, props), + adaptSocket: (socket) => new BrowserWebSocket(socket), prepare: async () => { // SAFETY: The branch checks _callApiKey before using the client's credential resolver. const credential = - '_callApiKey' in client && typeof client._callApiKey === 'function' - ? await resolveRealtimeAPIKey(client as Pick, true) - : { apiKey: client.apiKey, isProvider: false, commit: () => client.apiKey }; + '_callApiKey' in connectionClient && typeof connectionClient._callApiKey === 'function' + ? await resolveRealtimeAPIKey(connectionClient as Pick, true) + : { apiKey: connectionClient.apiKey, isProvider: false, commit: () => connectionClient.apiKey }; return () => { credential.commit(); const next = open(credential.apiKey, credential.isProvider); - instance.url = next.url; - return new BrowserWebSocket(next.socket); + this.url = next.url; + return next.socket; }; }, onError: (message, cause) => this._onError(null, message, cause), diff --git a/src/realtime/ws.ts b/src/realtime/ws.ts index 47582eebe8..9404d6934a 100644 --- a/src/realtime/ws.ts +++ b/src/realtime/ws.ts @@ -1,4 +1,5 @@ import { createRealtimeRecovery, validateRealtimeRecovery } from '../internal/realtime-recovery'; +import { NodeWebSocket } 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'; @@ -31,7 +32,7 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { /** Underlying `ws.WebSocket` instance for connection lifecycle and transport events. */ socket: WS.WebSocket; - private _recovery?: WebSocketConnection; + private _recovery?: WebSocketConnection; /** * Immediately opens a model or transcription session, or attaches to an existing call. @@ -61,10 +62,10 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { if (props.reconnect && !props.model) { throw new Error('Automatic Realtime recovery requires a model session'); } - client ??= new OpenAI(); - const apiKey = props.__apiKey === undefined ? client.apiKey : props.__apiKey; + 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( [ @@ -75,15 +76,15 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { } const terminalSockets = new WeakSet(); const open = (attemptKey: string | null, resolvedApiKey: boolean) => { - const url = buildRealtimeURL(client, props); + const url = buildRealtimeURL(connectionClient, props); validateRealtimeRecovery(props, url); - assertBedrockWebSocketOrigin(client, url); - if (isAzure(client) && props.__apiKey !== undefined && !attemptKey) { + assertBedrockWebSocketOrigin(connectionClient, url); + if (isAzure(connectionClient) && props.__apiKey !== undefined && !attemptKey) { throw new Error('Azure OpenAI Realtime requires an API key'); } - const azureKey = isAzure(client) && !resolvedApiKey; + const azureKey = isAzure(connectionClient) && !resolvedApiKey; const headers = { - 'User-Agent': `${client.constructor.name}/JS ${VERSION}`, + 'User-Agent': `${connectionClient.constructor.name}/JS ${VERSION}`, ...props.options?.headers, // Credentials remain own data properties without invoking inherited setters. ...(azureKey && props.__apiKey !== undefined ? { 'api-key': attemptKey ?? '' } : {}), @@ -170,14 +171,15 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { this.url = initial.url; this.socket = initial.socket; if (props.reconnect) { - this._recovery = createRealtimeRecovery(this, props, { - buildURL: () => buildRealtimeURL(client, props), + this._recovery = createRealtimeRecovery(this, props, { + buildURL: () => buildRealtimeURL(connectionClient, props), + adaptSocket: (socket) => new NodeWebSocket(socket), prepare: async () => { // SAFETY: The branch checks _callApiKey before using the client's credential resolver. const credential = - '_callApiKey' in client && typeof client._callApiKey === 'function' - ? await resolveRealtimeAPIKey(client as Pick, true) - : { apiKey: client.apiKey, isProvider: false, commit: () => client.apiKey }; + '_callApiKey' in connectionClient && typeof connectionClient._callApiKey === 'function' + ? await resolveRealtimeAPIKey(connectionClient as Pick, true) + : { apiKey: connectionClient.apiKey, isProvider: false, commit: () => connectionClient.apiKey }; return () => { credential.commit(); return open(credential.apiKey, credential.isProvider).socket; diff --git a/tests/realtime-wire-lifecycle.test.ts b/tests/realtime-wire-lifecycle.test.ts index d66468f18e..a3917f1a41 100644 --- a/tests/realtime-wire-lifecycle.test.ts +++ b/tests/realtime-wire-lifecycle.test.ts @@ -370,6 +370,9 @@ test.each([StableWS, StableNative])( { 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(); } From 2c1fb728e4b26a1c9bd11c5658e99946ff3aff85 Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Sun, 18 Oct 2026 04:54:06 +0000 Subject: [PATCH 05/10] fix(realtime): preserve reconnect state and queue validity --- src/internal/realtime-recovery.ts | 13 ++-- src/internal/ws-connection-controller.ts | 9 ++- src/internal/ws.ts | 3 + src/realtime/websocket.ts | 12 ++-- src/realtime/ws.ts | 8 +-- tests/realtime-wire-lifecycle.test.ts | 80 +++++++++++++++++++++++- 6 files changed, 103 insertions(+), 22 deletions(-) diff --git a/src/internal/realtime-recovery.ts b/src/internal/realtime-recovery.ts index c53698fc45..7583ec30a0 100644 --- a/src/internal/realtime-recovery.ts +++ b/src/internal/realtime-recovery.ts @@ -3,6 +3,8 @@ 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. */ @@ -22,16 +24,15 @@ export function validateRealtimeRecovery(options: RealtimeRecoveryOptions, url?: if (options.reconnect && (url?.searchParams.has('intent') || url?.searchParams.has('call_id'))) { throw new Error('Automatic Realtime recovery requires a model session'); } - for (const value of [ - options.maxReconnectAttempts, - options.reconnectInitialDelay, - options.reconnectMaxDelay, - options.maxQueueBytes, - ]) { + 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. */ diff --git a/src/internal/ws-connection-controller.ts b/src/internal/ws-connection-controller.ts index a3d27a4f5d..e64a25ff75 100644 --- a/src/internal/ws-connection-controller.ts +++ b/src/internal/ws-connection-controller.ts @@ -576,6 +576,11 @@ export class WebSocketConnection< 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) { @@ -593,7 +598,7 @@ export class WebSocketConnection< /* oxlint-disable eslint/no-await-in-loop -- Reconnect attempts and their credential refresh must run sequentially. */ for (let attempt = 1; this._hasRetryBudget(attempt, maxRetries); attempt += 1) { - this._retryCount += 1; + const reportedAttempt = this._nextRetryAttempt(attempt); if (this._abortReconnect(closeCode)) { return; } @@ -604,7 +609,7 @@ export class WebSocketConnection< const actualDelay = Math.round(baseDelay * jitter); const reconnectingEvent: ReconnectingEvent = { - attempt, + attempt: reportedAttempt, maxAttempts: maxRetries, delay: actualDelay, closeCode, diff --git a/src/internal/ws.ts b/src/internal/ws.ts index db07e1c894..809642fca6 100644 --- a/src/internal/ws.ts +++ b/src/internal/ws.ts @@ -400,6 +400,9 @@ 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 && diff --git a/src/realtime/websocket.ts b/src/realtime/websocket.ts index a5dd111896..cc7f7abd98 100644 --- a/src/realtime/websocket.ts +++ b/src/realtime/websocket.ts @@ -221,7 +221,7 @@ export class OpenAIRealtimeWebSocket extends OpenAIRealtimeEmitter { ); } - const open = (attemptKey: string | null, resolvedApiKey: boolean) => { + const open = (attemptKey: string | null, resolvedApiKey: boolean, capturedURL?: URL) => { // SAFETY: OpenAI owns _options; this reads its existing browser-consent flag. const allowed = props.dangerouslyAllowBrowser ?? @@ -230,9 +230,9 @@ export class OpenAIRealtimeWebSocket extends OpenAIRealtimeEmitter { if (!allowed && isRunningInBrowserOrBrowserWorker()) { throw new OpenAIError('Realtime browser connections require an ephemeral credential.'); } - const url = buildRealtimeURL(connectionClient, props); - validateRealtimeRecovery(props, url); + 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.'); } @@ -308,11 +308,9 @@ export class OpenAIRealtimeWebSocket extends OpenAIRealtimeEmitter { '_callApiKey' in connectionClient && typeof connectionClient._callApiKey === 'function' ? await resolveRealtimeAPIKey(connectionClient as Pick, true) : { apiKey: connectionClient.apiKey, isProvider: false, commit: () => connectionClient.apiKey }; - return () => { + return (url) => { credential.commit(); - const next = open(credential.apiKey, credential.isProvider); - this.url = next.url; - return next.socket; + return open(credential.apiKey, credential.isProvider, url).socket; }; }, onError: (message, cause) => this._onError(null, message, cause), diff --git a/src/realtime/ws.ts b/src/realtime/ws.ts index 9404d6934a..ab9b732c92 100644 --- a/src/realtime/ws.ts +++ b/src/realtime/ws.ts @@ -75,8 +75,8 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { ); } const terminalSockets = new WeakSet(); - const open = (attemptKey: string | null, resolvedApiKey: boolean) => { - const url = buildRealtimeURL(connectionClient, props); + const open = (attemptKey: string | null, resolvedApiKey: boolean, capturedURL?: URL) => { + const url = capturedURL ?? buildRealtimeURL(connectionClient, props); validateRealtimeRecovery(props, url); assertBedrockWebSocketOrigin(connectionClient, url); if (isAzure(connectionClient) && props.__apiKey !== undefined && !attemptKey) { @@ -180,9 +180,9 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { '_callApiKey' in connectionClient && typeof connectionClient._callApiKey === 'function' ? await resolveRealtimeAPIKey(connectionClient as Pick, true) : { apiKey: connectionClient.apiKey, isProvider: false, commit: () => connectionClient.apiKey }; - return () => { + return (url) => { credential.commit(); - return open(credential.apiKey, credential.isProvider).socket; + return open(credential.apiKey, credential.isProvider, url).socket; }; }, shouldReconnect: () => !terminalSockets.has(this.socket), diff --git a/tests/realtime-wire-lifecycle.test.ts b/tests/realtime-wire-lifecycle.test.ts index a3917f1a41..4da268d4f5 100644 --- a/tests/realtime-wire-lifecycle.test.ts +++ b/tests/realtime-wire-lifecycle.test.ts @@ -394,6 +394,58 @@ test('opt-in recovery can be canceled during backoff without opening another soc expect(await closed).toEqual([1000, 'OK', []]); }); +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) => { @@ -426,6 +478,9 @@ test.each([StableWS, StableNative])( }, }; 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. @@ -435,7 +490,7 @@ test.each([StableWS, StableNative])( first.close(1012); const [next] = await replacement; await recovered; - expect(errors).toHaveLength(2); + 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) => { @@ -444,8 +499,8 @@ test.each([StableWS, StableNative])( next.close(1012); await closed; expect(closes).toBe(1); - expect(errors).toHaveLength(3); - expect(errors[2]?.message).toContain('1 attempts'); + expect(errors).toHaveLength(4); + expect(errors[3]?.message).toContain('1 attempts'); realtime.close(); }, ); @@ -518,3 +573,22 @@ test.each([StableWS, StableNative])('recovery rejects sideband selectors in cust ), ).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'); + } +}); From 9750c0d8841f4cb384f9c86e05adbeb0221d6677 Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Sun, 18 Oct 2026 05:25:33 +0000 Subject: [PATCH 06/10] fix(realtime): preserve transport behavior during recovery --- README.md | 3 +- docs/realtime.md | 66 ++++++++----- src/internal/ws-adapter-browser.ts | 10 +- src/internal/ws-adapter-node.ts | 5 + src/internal/ws-adapter.ts | 3 + src/internal/ws-connection-controller.ts | 22 ++++- src/realtime/websocket.ts | 7 +- src/realtime/ws.ts | 21 ++-- tests/realtime-wire-lifecycle.test.ts | 121 +++++++++++++++++++---- 9 files changed, 197 insertions(+), 61 deletions(-) diff --git a/README.md b/README.md index 783567dd1b..ab040b6efc 100644 --- a/README.md +++ b/README.md @@ -544,7 +544,8 @@ Register an `error` listener as usual. The SDK allows five reconnect attempts in total by default. Set `maxReconnectAttempts` to change this limit. Retry delays start at 500 ms and grow to 8 seconds. Set `reconnectInitialDelay` and `reconnectMaxDelay` to change them. -Calling `close()` cancels pending retries and credential refresh. +Calling `close()` stops retries and prevents a pending credential refresh from +opening another socket. By default, the SDK rejects sends while reconnecting. Set `maxQueueBytes` to hold unsent events up to that size in bytes. After you restore session state, call diff --git a/docs/realtime.md b/docs/realtime.md index b39edec62c..b7679b5b9d 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/src/internal/ws-adapter-browser.ts b/src/internal/ws-adapter-browser.ts index 07a18a35fa..828473d796 100644 --- a/src/internal/ws-adapter-browser.ts +++ b/src/internal/ws-adapter-browser.ts @@ -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 666edea1de..fcdff59f21 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 3209cf19e9..7aa084756b 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 e64a25ff75..b476838f4e 100644 --- a/src/internal/ws-connection-controller.ts +++ b/src/internal/ws-connection-controller.ts @@ -128,6 +128,11 @@ export class WebSocketConnection< 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(); } @@ -153,7 +158,7 @@ export class WebSocketConnection< this._hooks.onError('cannot send on a closed WebSocket'); return; } - if (this._isReconnecting || this._owner.socket.readyState === ReadyState.CONNECTING) { + if (this._isReconnecting) { let accepted: boolean; try { accepted = this._sendQueue.enqueue(event); @@ -166,6 +171,10 @@ export class WebSocketConnection< } return; } + if (this._owner.socket.readyState === ReadyState.CONNECTING) { + this._hooks.onError('cannot send before the WebSocket is open'); + return; + } } sendWebSocketEvent( this._owner, @@ -203,7 +212,16 @@ export class WebSocketConnection< this._hooks.onError('could not close the connection', error); } finally { if (this._hooks.manualQueue) { - this._emitPermanentClose(this._closeCode, this._closeReason); + try { + const { socket } = this._owner; + if (socket.terminate) { + socket.terminate(); + } else { + socket.close(); + } + } finally { + this._emitPermanentClose(this._closeCode, this._closeReason); + } } } } diff --git a/src/realtime/websocket.ts b/src/realtime/websocket.ts index cc7f7abd98..2683f5e2a3 100644 --- a/src/realtime/websocket.ts +++ b/src/realtime/websocket.ts @@ -246,7 +246,7 @@ export class OpenAIRealtimeWebSocket extends OpenAIRealtimeEmitter { : new WebSocket(url.toString(), protocols); socket.addEventListener('message', (websocketEvent: MessageEvent) => { - if (this._recovery && socket !== this.socket) { + if (this._recovery && (socket !== this.socket || this._recovery.isClosed)) { return; } const event = (() => { @@ -273,6 +273,9 @@ export class OpenAIRealtimeWebSocket extends OpenAIRealtimeEmitter { if (event) { this._emit('event', event); + if (this._recovery?.isClosed) { + return; + } if (event.type === 'error') { this._onError(event); } else if (!this._recovery || !['close', 'reconnecting', 'reconnected'].includes(event.type)) { @@ -301,7 +304,7 @@ export class OpenAIRealtimeWebSocket extends OpenAIRealtimeEmitter { if (props.reconnect) { this._recovery = createRealtimeRecovery(this, props, { buildURL: () => buildRealtimeURL(connectionClient, props), - adaptSocket: (socket) => new BrowserWebSocket(socket), + adaptSocket: (socket) => new BrowserWebSocket(socket, { preserveBinaryType: true }), prepare: async () => { // SAFETY: The branch checks _callApiKey before using the client's credential resolver. const credential = diff --git a/src/realtime/ws.ts b/src/realtime/ws.ts index ab9b732c92..d511852eee 100644 --- a/src/realtime/ws.ts +++ b/src/realtime/ws.ts @@ -1,5 +1,5 @@ import { createRealtimeRecovery, validateRealtimeRecovery } from '../internal/realtime-recovery'; -import { NodeWebSocket } from '../internal/ws-adapter-node'; +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'; @@ -75,7 +75,12 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { ); } const terminalSockets = new WeakSet(); - const open = (attemptKey: string | null, resolvedApiKey: boolean, capturedURL?: URL) => { + 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); @@ -85,7 +90,7 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { const azureKey = isAzure(connectionClient) && !resolvedApiKey; const headers = { 'User-Agent': `${connectionClient.constructor.name}/JS ${VERSION}`, - ...props.options?.headers, + ...options?.headers, // Credentials remain own data properties without invoking inherited setters. ...(azureKey && props.__apiKey !== undefined ? { 'api-key': attemptKey ?? '' } : {}), ...(azureKey ? {} : { Authorization: `Bearer ${attemptKey}` }), @@ -94,13 +99,13 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { const socket = new WS.WebSocket( url, protectWebSocketOptionsFromCredentialRedirects({ - ...props.options, + ...options, headers, }), ); socket.on('message', (wsEvent) => { - if (this._recovery && socket !== this.socket) { + if (this._recovery && (socket !== this.socket || this._recovery.isClosed)) { return; } const event = (() => { @@ -127,6 +132,9 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { if (event) { this._emit('event', event); + if (this._recovery?.isClosed) { + return; + } if (event.type === 'error') { this._onError(event); } else if (!this._recovery || !['close', 'reconnecting', 'reconnected'].includes(event.type)) { @@ -175,6 +183,7 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { 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' @@ -182,7 +191,7 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { : { apiKey: connectionClient.apiKey, isProvider: false, commit: () => connectionClient.apiKey }; return (url) => { credential.commit(); - return open(credential.apiKey, credential.isProvider, url).socket; + return open(credential.apiKey, credential.isProvider, url, options).socket; }; }, shouldReconnect: () => !terminalSockets.has(this.socket), diff --git a/tests/realtime-wire-lifecycle.test.ts b/tests/realtime-wire-lifecycle.test.ts index 4da268d4f5..77195e0763 100644 --- a/tests/realtime-wire-lifecycle.test.ts +++ b/tests/realtime-wire-lifecycle.test.ts @@ -314,6 +314,28 @@ 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) => { @@ -330,6 +352,7 @@ test.each([StableWS, StableNative])( 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; @@ -347,6 +370,7 @@ test.each([StableWS, StableNative])( expect(request.headers.authorization).toBe('Bearer ek_recovery_2'); } else { expect(request.headers['sec-websocket-protocol']).toContain('ek_recovery_2'); + expect([initialBinaryType, realtime.socket.binaryType]).toEqual(['blob', 'blob']); } expect(realtime.socket).not.toBe(oldSocket); expect(commands).toEqual([]); @@ -379,6 +403,44 @@ test.each([StableWS, StableNative])( }, ); +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( @@ -505,27 +567,44 @@ test.each([StableWS, StableNative])( }, ); -test('failed physical close still ends logical recovery and send admission', async () => { - const connected = once(peers, 'connection'); - const realtime = new StableWS( - { ...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'); - // 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()); - }); - realtime.close({ code: 1, reason: 'invalid physical close' }); - await closed; - realtime.send({ type: 'response.create' }); - expect(errors).toHaveLength(2); - expect(errors[1]?.message).toContain('closed WebSocket'); - first.terminate(); -}); +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'); + }, +); test('recovery stops after a permanent authentication rejection', async () => { const endpoint = createServer({ cert: lab.server.certificate, key: lab.server.privateKey }); From 7731eb456d1330cbc8c9b913ea19e662876f3624 Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Sun, 18 Oct 2026 06:38:56 +0000 Subject: [PATCH 07/10] fix(realtime): retain committed credentials and handshake errors --- src/realtime/websocket.ts | 7 ++--- src/realtime/ws.ts | 28 ++++++++++++-------- tests/realtime-wire-lifecycle.test.ts | 37 ++++++++++++++++++++++----- 3 files changed, 50 insertions(+), 22 deletions(-) diff --git a/src/realtime/websocket.ts b/src/realtime/websocket.ts index 2683f5e2a3..8b759eee16 100644 --- a/src/realtime/websocket.ts +++ b/src/realtime/websocket.ts @@ -310,11 +310,8 @@ export class OpenAIRealtimeWebSocket extends OpenAIRealtimeEmitter { const credential = '_callApiKey' in connectionClient && typeof connectionClient._callApiKey === 'function' ? await resolveRealtimeAPIKey(connectionClient as Pick, true) - : { apiKey: connectionClient.apiKey, isProvider: false, commit: () => connectionClient.apiKey }; - return (url) => { - credential.commit(); - return open(credential.apiKey, credential.isProvider, url).socket; - }; + : { 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), diff --git a/src/realtime/ws.ts b/src/realtime/ws.ts index d511852eee..976706f2fe 100644 --- a/src/realtime/ws.ts +++ b/src/realtime/ws.ts @@ -74,7 +74,7 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { ].join('\n'), ); } - const terminalSockets = new WeakSet(); + const terminalErrors = new WeakMap(); const open = ( attemptKey: string | null, resolvedApiKey: boolean, @@ -147,7 +147,10 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { if (props.reconnect) { socket.on('unexpected-response', (_request, response) => { if (![408, 429, 500, 502, 503, 504].includes(response.statusCode ?? 0)) { - terminalSockets.add(socket); + terminalErrors.set( + socket, + new Error(`Unexpected server response: ${response.statusCode ?? 'unknown'}`), + ); } response.resume(); socket.terminate(); @@ -167,7 +170,7 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { 'UNABLE_TO_GET_ISSUER_CERT_LOCALLY', ].includes(err.code)) ) { - terminalSockets.add(socket); + terminalErrors.set(socket, err); } return; } @@ -188,17 +191,20 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { const credential = '_callApiKey' in connectionClient && typeof connectionClient._callApiKey === 'function' ? await resolveRealtimeAPIKey(connectionClient as Pick, true) - : { apiKey: connectionClient.apiKey, isProvider: false, commit: () => connectionClient.apiKey }; - return (url) => { - credential.commit(); - return open(credential.apiKey, credential.isProvider, url, options).socket; - }; + : { 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 = cause === undefined ? terminalErrors.get(this.socket) : undefined; + this._onError(null, terminal ? `${message}: ${terminal.message}` : message, terminal ?? cause); }, - shouldReconnect: () => !terminalSockets.has(this.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), + onClose: (code, reason, unsent) => { + terminalErrors.delete(this.socket); + this._emit('close', code, reason, unsent); + }, }); } } diff --git a/tests/realtime-wire-lifecycle.test.ts b/tests/realtime-wire-lifecycle.test.ts index 77195e0763..5585b157dc 100644 --- a/tests/realtime-wire-lifecycle.test.ts +++ b/tests/realtime-wire-lifecycle.test.ts @@ -341,6 +341,13 @@ test.each([StableWS, StableNative])( 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, @@ -367,9 +374,9 @@ test.each([StableWS, StableNative])( try { expect(resolutions).toBe(2); if (Realtime === StableWS) { - expect(request.headers.authorization).toBe('Bearer ek_recovery_2'); + expect(request.headers.authorization).toBe('Bearer ek_recovery_2_committed'); } else { - expect(request.headers['sec-websocket-protocol']).toContain('ek_recovery_2'); + 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); @@ -606,7 +613,8 @@ test.each([StableWS, StableNative])( }, ); -test('recovery stops after a permanent authentication rejection', async () => { +test.each(['authentication', 'TLS'])('recovery reports a terminal %s rejection', async (failure) => { + const tls = failure === 'TLS'; const endpoint = createServer({ cert: lab.server.certificate, key: lab.server.privateKey }); let attempts = 0; const sockets = new WebSocketServer({ server: endpoint, verifyClient: () => (attempts += 1) === 1 }); @@ -617,20 +625,37 @@ test('recovery stops after a permanent authentication rejection', async () => { } const connected = once(sockets, 'connection'); const realtime = new StableWS( - { ...settings, reconnect: true, reconnectInitialDelay: 1 }, + { + ...settings, + options: { ...settings.options, agent: false }, + reconnect: true, + reconnectInitialDelay: 1, + maxReconnectAttempts: tls ? 5 : 1, + }, new OpenAI({ baseURL: `https://127.0.0.1:${address.port}/v1`, apiKey: 'ek_rejected' }), ); - realtime.on('error', () => {}); + const errors: Error[] = []; + realtime.on('error', (error) => errors.push(error)); try { const [first] = await connected; await rawEvent(realtime.socket, 'open'); + if (tls) { + endpoint.setSecureContext({ cert: lab.proxyServer.certificate, key: lab.proxyServer.privateKey }); + } // oxlint-disable-next-line promise/avoid-new -- Authentication errors must not reject the cleanup wait. const closed = new Promise((resolve) => { realtime.on('close', () => resolve()); }); first.close(1012); await closed; - expect(attempts).toBe(2); + expect(attempts).toBe(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(); From 6039f103b722da2a149b01e2501111f6cfbfb4f3 Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Sun, 18 Oct 2026 06:42:20 +0000 Subject: [PATCH 08/10] test(realtime): use a typed agent for TLS recovery --- tests/realtime-wire-lifecycle.test.ts | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/tests/realtime-wire-lifecycle.test.ts b/tests/realtime-wire-lifecycle.test.ts index 5585b157dc..5ed32a07ce 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'; @@ -624,10 +624,11 @@ test.each(['authentication', 'TLS'])('recovery reports a terminal %s rejection', throw new Error('Missing server address'); } const connected = once(sockets, 'connection'); + const agent = new Agent({ maxCachedSessions: 0 }); const realtime = new StableWS( { ...settings, - options: { ...settings.options, agent: false }, + options: { ...settings.options, agent }, reconnect: true, reconnectInitialDelay: 1, maxReconnectAttempts: tls ? 5 : 1, @@ -659,6 +660,7 @@ test.each(['authentication', 'TLS'])('recovery reports a terminal %s rejection', expect(realtime.socket.readyState).toBe(WebSocket.CLOSED); } finally { realtime.close(); + agent.destroy(); sockets.close(); endpoint.close(); } From 1735c1704de7bb5ab670cd1f4c900faaa345c472 Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Sun, 18 Oct 2026 06:54:21 +0000 Subject: [PATCH 09/10] fix(realtime): preserve preparation and initial handshake errors --- src/internal/ws-connection-controller.ts | 27 +++++--- src/realtime/ws.ts | 8 ++- tests/realtime-wire-lifecycle.test.ts | 79 ++++++++++++++++++++---- 3 files changed, 91 insertions(+), 23 deletions(-) diff --git a/src/internal/ws-connection-controller.ts b/src/internal/ws-connection-controller.ts index b476838f4e..36b68afc33 100644 --- a/src/internal/ws-connection-controller.ts +++ b/src/internal/ws-connection-controller.ts @@ -524,14 +524,22 @@ export class WebSocketConnection< return this._hooks.shouldReconnect?.(code) ?? isRecoverableClose(code); } - private _abortReconnect(closeCode: number): boolean { + 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, cause?: unknown): boolean { if (this._canReconnect(closeCode)) { return false; } this._isReconnecting = false; try { if (!this._intentionallyClosed && !this._permanentClosed) { - this._hooks.onError(`WebSocket reconnect aborted: non-recoverable close code ${closeCode}`); + this._reportReconnectFailure( + `WebSocket reconnect aborted: non-recoverable close code ${closeCode}`, + cause, + ); } } finally { this._emitPermanentClose( @@ -613,11 +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; this._hasRetryBudget(attempt, maxRetries); attempt += 1) { const reportedAttempt = this._nextRetryAttempt(attempt); - if (this._abortReconnect(closeCode)) { + if (this._abortReconnect(closeCode, lastError)) { return; } @@ -638,18 +647,18 @@ export class WebSocketConnection< return; } - if (this._abortReconnect(closeCode)) { + if (this._abortReconnect(closeCode, lastError)) { return; } try { await sleep(actualDelay, this._backoff.signal); } catch { - this._abortReconnect(closeCode); + this._abortReconnect(closeCode, lastError); return; } - if (this._abortReconnect(closeCode)) { + if (this._abortReconnect(closeCode, lastError)) { return; } @@ -687,7 +696,7 @@ export class WebSocketConnection< }); await waitForWebSocketOpen(this._owner.socket); - } catch { + } 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. @@ -697,6 +706,7 @@ export class WebSocketConnection< } return; } + lastError = error; if (closeCodePromise) { // ws may emit 'error' before 'close', so await the code // rather than reading it synchronously. @@ -717,8 +727,9 @@ export class WebSocketConnection< // distinguish retry failure from a clean close. this._isReconnecting = false; try { - this._hooks.onError( + this._reportReconnectFailure( `WebSocket reconnect failed after ${maxRetries} attempts (close code: ${closeCode})`, + lastError, ); } finally { this._emitPermanentClose(closeCode, `reconnect failed after ${maxRetries} attempts`); diff --git a/src/realtime/ws.ts b/src/realtime/ws.ts index 976706f2fe..9b10431a0a 100644 --- a/src/realtime/ws.ts +++ b/src/realtime/ws.ts @@ -196,8 +196,12 @@ export class OpenAIRealtimeWS extends OpenAIRealtimeEmitter { }, shouldReconnect: () => !terminalErrors.has(this.socket), onError: (message, cause) => { - const terminal = cause === undefined ? terminalErrors.get(this.socket) : undefined; - this._onError(null, terminal ? `${message}: ${terminal.message}` : message, terminal ?? 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'), diff --git a/tests/realtime-wire-lifecycle.test.ts b/tests/realtime-wire-lifecycle.test.ts index 5ed32a07ce..0b76dad83a 100644 --- a/tests/realtime-wire-lifecycle.test.ts +++ b/tests/realtime-wire-lifecycle.test.ts @@ -613,17 +613,22 @@ test.each([StableWS, StableNative])( }, ); -test.each(['authentication', 'TLS'])('recovery reports a terminal %s rejection', async (failure) => { +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 }); + 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 = once(sockets, 'connection'); + const connected = initial ? undefined : once(sockets, 'connection'); const agent = new Agent({ maxCachedSessions: 0 }); const realtime = new StableWS( { @@ -637,19 +642,21 @@ test.each(['authentication', 'TLS'])('recovery reports a terminal %s rejection', ); 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 { - const [first] = await connected; - await rawEvent(realtime.socket, 'open'); - if (tls) { - endpoint.setSecureContext({ cert: lab.proxyServer.certificate, key: lab.proxyServer.privateKey }); + 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); } - // oxlint-disable-next-line promise/avoid-new -- Authentication errors must not reject the cleanup wait. - const closed = new Promise((resolve) => { - realtime.on('close', () => resolve()); - }); - first.close(1012); await closed; - expect(attempts).toBe(tls ? 1 : 2); + expect(attempts).toBe(initial || tls ? 1 : 2); const diagnostic = tls ? /certificate|verify/iu : /401/u; expect(errors).toEqual([ expect.objectContaining({ @@ -666,6 +673,52 @@ test.each(['authentication', 'TLS'])('recovery reports a terminal %s rejection', } }); +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( () => From b44832cf963afdea3c8fbda95485005c39f86dd6 Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Sun, 18 Oct 2026 07:19:29 +0000 Subject: [PATCH 10/10] fix(realtime): preserve reconnect notification order --- README.md | 27 +----------------------- src/internal/ws-connection-controller.ts | 2 +- tests/lib/responsesWebSocket.test.ts | 9 ++++++-- tests/realtime-wire-lifecycle.test.ts | 11 +++++----- 4 files changed, 15 insertions(+), 34 deletions(-) diff --git a/README.md b/README.md index ab040b6efc..6e13423632 100644 --- a/README.md +++ b/README.md @@ -527,32 +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). - -Set `reconnect: true` to reconnect after a temporary connection failure. -`OpenAIRealtimeWS` and `OpenAIRealtimeWebSocket` support this option, including -when you create them asynchronously. It is off by default. It does not support -transcription sessions or connections to an existing call. - -Reconnecting keeps the same connection object but replaces its `.socket` and -starts a new server session. Your application must restore any session settings -and conversation state it needs. The SDK does not resend commands it already -tried to send. Use the connection's `reconnecting`, `reconnected`, and `close` -events. Listeners attached directly to the old socket do not move to the new one. -Register an `error` listener as usual. - -The SDK allows five reconnect attempts in total by default. Set -`maxReconnectAttempts` to change this limit. Retry delays start at 500 ms and grow -to 8 seconds. Set `reconnectInitialDelay` and `reconnectMaxDelay` to change them. -Calling `close()` stops retries and prevents a pending credential refresh from -opening another socket. - -By default, the SDK rejects sends while reconnecting. Set `maxQueueBytes` to hold -unsent events up to that size in bytes. After you restore session state, call -`flushPending()` to send these events in order, or `takePending()` to remove them -without sending. The final `close` event includes any remaining unsent events as -its third argument. A successful send does not confirm that the server received -or processed the command. +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/src/internal/ws-connection-controller.ts b/src/internal/ws-connection-controller.ts index 36b68afc33..fcd2407e89 100644 --- a/src/internal/ws-connection-controller.ts +++ b/src/internal/ws-connection-controller.ts @@ -716,9 +716,9 @@ export class WebSocketConnection< } this._isReconnecting = false; this._internalEvents._emit('socketSwap', oldSocket, this._owner.socket); - this._internalEvents._emit('reconnected'); this._flushSendQueue(); this._hooks.emitReconnected(); + this._internalEvents._emit('reconnected'); return; } diff --git a/tests/lib/responsesWebSocket.test.ts b/tests/lib/responsesWebSocket.test.ts index 5fab1940c0..53e605a5d1 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 0b76dad83a..3bdcb710e9 100644 --- a/tests/realtime-wire-lifecycle.test.ts +++ b/tests/realtime-wire-lifecycle.test.ts @@ -381,11 +381,8 @@ test.each([StableWS, StableNative])( } expect(realtime.socket).not.toBe(oldSocket); expect(commands).toEqual([]); - const flushed = once(next, 'message'); realtime.send({ type: 'session.update', session: { type: 'realtime' } }); realtime.flushPending(); - await flushed; - // The following server event is a wire barrier after both client commands. // oxlint-disable-next-line promise/avoid-new -- Wait for both ordered wire messages, independent of their scheduling. await new Promise((resolve) => { const check = () => { @@ -456,11 +453,15 @@ test('opt-in recovery can be canceled during backoff without opening another soc ); realtime.on('error', () => {}); const [first] = await connected; - await rawEvent(realtime.socket, 'open'); + const originalSocket = realtime.socket; + await rawEvent(originalSocket, 'open'); const closed = realtime.emitted('close'); - realtime.on('reconnecting', () => realtime.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) => {