diff --git a/extensions/xai/realtime-voice-bridge.ts b/extensions/xai/realtime-voice-bridge.ts index 873432bcf4b9..f25f5d75c3d1 100644 --- a/extensions/xai/realtime-voice-bridge.ts +++ b/extensions/xai/realtime-voice-bridge.ts @@ -8,6 +8,7 @@ import type { RealtimeVoiceBridge, RealtimeVoiceToolResultOptions, } from "openclaw/plugin-sdk/realtime-voice"; +import { sleepWithAbort } from "openclaw/plugin-sdk/runtime-env"; import WebSocket from "ws"; import { XAI_REALTIME_BASE_RECONNECT_DELAY_MS, @@ -43,9 +44,13 @@ export class XaiRealtimeVoiceBridge extends XaiRealtimeVoiceEvents implements Re private connectionUrl = ""; private readonly flowId = randomUUID(); private sessionReadyFired = false; + private reconnectAbortController = new AbortController(); async connect(): Promise { this.intentionallyClosed = false; + if (this.reconnectAbortController.signal.aborted) { + this.reconnectAbortController = new AbortController(); + } this.reconnectAttempts = 0; await this.doConnect(); } @@ -107,6 +112,9 @@ export class XaiRealtimeVoiceBridge extends XaiRealtimeVoiceEvents implements Re close(): void { this.intentionallyClosed = true; + // The bridge owns both its active socket and reconnect delay; canceling + // both keeps terminal close from retaining callbacks for the full backoff. + this.reconnectAbortController.abort(); this.connected = false; this.sessionConfigured = false; this.pendingToolResultAcks.clear(); @@ -315,9 +323,15 @@ export class XaiRealtimeVoiceBridge extends XaiRealtimeVoiceEvents implements Re type: "session.reconnect.scheduled", detail: `reason=${reason} attempt=${attempt} delayMs=${delay}`, }); - await new Promise((resolve) => { - setTimeout(resolve, delay); - }); + const reconnectSignal = this.reconnectAbortController.signal; + try { + await sleepWithAbort(delay, reconnectSignal); + } catch (error) { + if (!reconnectSignal.aborted) { + throw error; + } + return; + } if (this.intentionallyClosed) { return; } diff --git a/extensions/xai/realtime-voice-provider.test.ts b/extensions/xai/realtime-voice-provider.test.ts index f928acef0562..9f326bb8d278 100644 --- a/extensions/xai/realtime-voice-provider.test.ts +++ b/extensions/xai/realtime-voice-provider.test.ts @@ -1580,6 +1580,57 @@ describe("buildXaiRealtimeVoiceProvider", () => { expect(onReady).toHaveBeenCalledOnce(); }); + it("cancels a pending reconnect and allows a later explicit connect", async () => { + vi.useFakeTimers(); + resolveApiKeyForProviderMock.mockResolvedValue({ apiKey: ["xai", "test"].join("-") }); + const provider = buildXaiRealtimeVoiceProvider(); + const onError = vi.fn(); + const bridge = provider.createBridge({ + providerConfig: { sessionResumption: true }, + onAudio: vi.fn(), + onClearAudio: vi.fn(), + onError, + }); + + const connecting = bridge.connect(); + await vi.waitFor(() => expect(FakeWebSocket.instances.length).toBe(1)); + const firstSocket = requireSocket(); + firstSocket.readyState = FakeWebSocket.OPEN; + firstSocket.emit("open"); + firstSocket.emit( + "message", + Buffer.from( + JSON.stringify({ type: "conversation.created", conversation: { id: "conv_close" } }), + ), + ); + firstSocket.emit("message", Buffer.from(JSON.stringify({ type: "session.updated" }))); + await connecting; + + firstSocket.close(1006, "connection lost"); + await vi.advanceTimersByTimeAsync(0); + expect(vi.getTimerCount()).toBe(1); + + bridge.close(); + await vi.advanceTimersByTimeAsync(0); + + expect(vi.getTimerCount()).toBe(0); + expect(FakeWebSocket.instances).toHaveLength(1); + expect(onError).not.toHaveBeenCalled(); + + const reconnecting = bridge.connect(); + await vi.waitFor(() => expect(FakeWebSocket.instances.length).toBe(2)); + const reconnectedSocket = requireSocket(1); + reconnectedSocket.readyState = FakeWebSocket.OPEN; + reconnectedSocket.emit("open"); + reconnectedSocket.emit("message", Buffer.from(JSON.stringify({ type: "session.updated" }))); + await reconnecting; + + expect(bridge.isConnected()).toBe(true); + expect(FakeWebSocket.instances).toHaveLength(2); + expect(onError).not.toHaveBeenCalled(); + bridge.close(); + }); + it("enables xAI session resumption and reconnects with the created conversation id", async () => { vi.useFakeTimers(); vi.stubEnv("XAI_API_KEY", "xai-env"); // pragma: allowlist secret