From a2465c782b1d4420330b25ad623d6e7f42337317 Mon Sep 17 00:00:00 2001 From: quecai-niu <114210016+quecai-niu@users.noreply.github.com> Date: Sat, 11 Jul 2026 10:41:02 +0800 Subject: [PATCH] [codex] fix Qwen realtime compatibility (#977) * fix: improve Qwen realtime compatibility * fix: preserve Qwen endpoint query parameters --- hub/src/web/qwenProxyHandler.test.ts | 38 +++++++++++++++++++++++++++- hub/src/web/qwenProxyHandler.ts | 21 ++++++++++++--- shared/src/voice.ts | 6 +++-- 3 files changed, 59 insertions(+), 6 deletions(-) diff --git a/hub/src/web/qwenProxyHandler.test.ts b/hub/src/web/qwenProxyHandler.test.ts index 337bd069..8816ac91 100644 --- a/hub/src/web/qwenProxyHandler.test.ts +++ b/hub/src/web/qwenProxyHandler.test.ts @@ -53,6 +53,7 @@ class FakeClient { } let lastUpstream: FakeUpstream | null = null +const origQwenRealtimeWsUrl = process.env.QWEN_REALTIME_WS_URL const FakeWebSocket = function FakeWebSocket(url: string, opts?: unknown) { const u = new FakeUpstream(url, opts) lastUpstream = u @@ -65,13 +66,47 @@ beforeEach(() => { afterEach(() => { lastUpstream = null + if (origQwenRealtimeWsUrl === undefined) delete process.env.QWEN_REALTIME_WS_URL + else process.env.QWEN_REALTIME_WS_URL = origQwenRealtimeWsUrl }) function newClient() { - return new FakeClient({ apiKey: 'k', model: 'qwen3-omni-flash-realtime', language: 'en', voiceName: 'Cherry' }) as unknown as ServerWebSocket & FakeClient + return new FakeClient({ apiKey: 'k', model: 'qwen3.5-omni-flash-realtime', language: 'en', voiceName: 'Cherry' }) as unknown as ServerWebSocket & FakeClient } describe('createQwenProxyWebSocketHandler ack-gate', () => { + test('preserves the international DashScope realtime endpoint by default', () => { + delete process.env.QWEN_REALTIME_WS_URL + const handler = createQwenProxyWebSocketHandler(FakeWebSocket) + const client = newClient() + + handler.open(client) + + expect(lastUpstream?.url).toBe('wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime?model=qwen3.5-omni-flash-realtime') + }) + + test('allows overriding the DashScope realtime endpoint with QWEN_REALTIME_WS_URL', () => { + process.env.QWEN_REALTIME_WS_URL = 'wss://example.test/realtime' + const handler = createQwenProxyWebSocketHandler(FakeWebSocket) + const client = newClient() + + handler.open(client) + + expect(lastUpstream?.url).toBe('wss://example.test/realtime?model=qwen3.5-omni-flash-realtime') + }) + + test('preserves existing query parameters in QWEN_REALTIME_WS_URL', () => { + process.env.QWEN_REALTIME_WS_URL = 'wss://example.test/realtime?workspace=abc&model=old' + const handler = createQwenProxyWebSocketHandler(FakeWebSocket) + const client = newClient() + + handler.open(client) + + expect(lastUpstream?.url).toBe( + 'wss://example.test/realtime?workspace=abc&model=qwen3.5-omni-flash-realtime' + ) + }) + test('queues client frames until upstream acks hub-owned session.update with session.updated', () => { const handler = createQwenProxyWebSocketHandler(FakeWebSocket) const client = newClient() @@ -85,6 +120,7 @@ describe('createQwenProxyWebSocketHandler ack-gate', () => { const hubSetup = JSON.parse(upstream.sent[0] as string) as { type: string; session: { instructions: string } } expect(hubSetup.type).toBe('session.update') expect(typeof hubSetup.session.instructions).toBe('string') + expect(hubSetup).not.toHaveProperty('session.tool_choice') handler.message(client, JSON.stringify({ type: 'response.create' })) handler.message(client, JSON.stringify({ type: 'conversation.item.create', item: { type: 'message' } })) diff --git a/hub/src/web/qwenProxyHandler.ts b/hub/src/web/qwenProxyHandler.ts index 8c1cd18b..c0abb66c 100644 --- a/hub/src/web/qwenProxyHandler.ts +++ b/hub/src/web/qwenProxyHandler.ts @@ -53,9 +53,11 @@ export function createQwenProxyWebSocketHandler( return { open(clientWs) { const data = clientWs.data as { apiKey: string; model: string; language?: string; voiceName?: string; systemInstruction?: string } - const upstreamUrl = `${process.env.QWEN_REALTIME_WS_URL || QWEN_WS_BASE}?model=${encodeURIComponent(data.model)}` + const upstreamBase = process.env.QWEN_REALTIME_WS_URL || QWEN_WS_BASE + const upstreamUrl = new URL(upstreamBase) + upstreamUrl.searchParams.set('model', data.model) - const upstream = new WebSocketImpl(upstreamUrl, { + const upstream = new WebSocketImpl(upstreamUrl.toString(), { headers: { 'Authorization': `Bearer ${data.apiKey}` } }) @@ -106,12 +108,17 @@ export function createQwenProxyWebSocketHandler( } } catch { /* client gone */ } } - upstream.onerror = () => { + upstream.onerror = (event) => { pendingSetupMap.delete(clientWs) setupAckedMap.delete(clientWs) pendingClientFrames.delete(clientWs) pendingClientBytes.delete(clientWs) upstreamMap.delete(clientWs) + console.error('[Voice][QwenProxy] Upstream WebSocket error', { + upstreamBase, + model: data.model, + error: event instanceof Error ? event.message : String(event) + }) try { clientWs.close(1011, 'Upstream error') } catch { /* */ } } upstream.onclose = (event) => { @@ -119,6 +126,14 @@ export function createQwenProxyWebSocketHandler( setupAckedMap.delete(clientWs) pendingClientFrames.delete(clientWs) pendingClientBytes.delete(clientWs) + if (event.code !== 1000) { + console.warn('[Voice][QwenProxy] Upstream WebSocket closed', { + upstreamBase, + model: data.model, + code: event.code, + reason: event.reason + }) + } try { clientWs.close(toClientCloseCode(event.code), event.reason || 'Upstream closed') } catch { /* */ } upstreamMap.delete(clientWs) } diff --git a/shared/src/voice.ts b/shared/src/voice.ts index 0f3ac56c..7cc4738c 100644 --- a/shared/src/voice.ts +++ b/shared/src/voice.ts @@ -399,8 +399,10 @@ export function buildQwenSessionUpdateMessage( silence_duration_ms: 800, prefix_padding_ms: 300 }, - tools, - tool_choice: 'auto' + // Qwen-Omni-Realtime decides whether to call tools automatically. + // The official realtime API does not support OpenAI-style tool_choice / + // parallel_tool_calls parameters on session.update. + tools } } }