diff --git a/.gitignore b/.gitignore index 78903200ce..e25c32fcd5 100644 --- a/.gitignore +++ b/.gitignore @@ -110,3 +110,4 @@ tmp/ .venv .codegraph .qwen/computer-use/installed.json +.playwright-mcp/ diff --git a/integration-tests/cli/qwen-serve-routes.test.ts b/integration-tests/cli/qwen-serve-routes.test.ts index bcfa22f764..30dad436c6 100644 --- a/integration-tests/cli/qwen-serve-routes.test.ts +++ b/integration-tests/cli/qwen-serve-routes.test.ts @@ -221,9 +221,9 @@ describe('qwen serve — capabilities envelope', () => { // `packages/cli/src/serve/capabilities.ts` and the unit-level // baseline features in `packages/cli/src/serve/server.test.ts`. // - // Conditional tags absent under this suite's spawn flags (no - // `--require-auth` / `--allow-origin` / deadline env vars / - // rate-limit opt-in): `require_auth`, `allow_origin`, + // Conditional tags absent under this suite's spawn flags (token auth / + // no `--require-auth` / no `--allow-origin` / no deadline env vars / + // no rate-limit opt-in): `require_auth`, `allow_origin`, // `prompt_absolute_deadline`, `writer_idle_timeout`, `rate_limit`. // Pool tags (`mcp_workspace_pool`, `mcp_pool_restart`) ARE present // because the workspace MCP pool is on by default, as are @@ -293,6 +293,7 @@ describe('qwen serve — capabilities envelope', () => { 'workspace_extensions', 'session_branch', 'workspace_reload', + 'voice_transcribe', ]); }); }); diff --git a/packages/cli/src/serve/acp-http/index.ts b/packages/cli/src/serve/acp-http/index.ts index 1959cc04cf..885acd56a7 100644 --- a/packages/cli/src/serve/acp-http/index.ts +++ b/packages/cli/src/serve/acp-http/index.ts @@ -27,6 +27,46 @@ import { RPC, error as rpcError, isRequest, parseInbound } from './json-rpc.js'; export const ACP_CONNECTION_HEADER = 'acp-connection-id'; export const ACP_SESSION_HEADER = 'acp-session-id'; +/** + * Browsers cannot set an `Authorization` header on a WebSocket, so the Web + * Shell authenticates the `/voice/stream` (and `/acp`) upgrade by offering the + * bearer token as a `Sec-WebSocket-Protocol` subprotocol of the form + * `qwen-bearer.`. Kept in sync with the encoder in + * `packages/web-shell/client/voice/useVoiceCapture.ts`. + */ +export const WS_BEARER_SUBPROTOCOL_PREFIX = 'qwen-bearer.'; + +/** + * Pull the bearer credential off a WS upgrade request. Prefer the standard + * `Authorization: Bearer ` header (non-browser clients); fall back to + * the `qwen-bearer.*` subprotocol (browser clients). Returns `undefined` when + * neither is present or parseable. + */ +function extractUpgradeBearer(req: IncomingMessage): string | undefined { + const authHeader = req.headers['authorization']; + if (authHeader && authHeader.includes(' ')) { + const scheme = authHeader.slice(0, authHeader.indexOf(' ')).toLowerCase(); + if (scheme === 'bearer') { + const credentials = authHeader.slice(authHeader.indexOf(' ') + 1).trim(); + if (credentials) return credentials; + } + } + const offered = req.headers['sec-websocket-protocol']; + if (offered) { + for (const raw of offered.split(',')) { + const entry = raw.trim(); + if (!entry.startsWith(WS_BEARER_SUBPROTOCOL_PREFIX)) continue; + const encoded = entry.slice(WS_BEARER_SUBPROTOCOL_PREFIX.length); + // `Buffer.from(_, 'base64url')` never throws — malformed input just + // decodes to garbage bytes, which fail the constant-time hash compare + // at the call site. An empty decode means "no credential offered". + const decoded = Buffer.from(encoded, 'base64url').toString('utf8'); + if (decoded) return decoded; + } + } + return undefined; +} + /** * Grace window after the connection-scoped SSE stream closes before the * connection is reaped (if not reconnected and no session stream is live). @@ -80,6 +120,22 @@ export interface MountAcpHttpOptions { sessionShellCommandEnabled?: boolean; /** Rate limit checker for WS messages (WS bypasses Express middleware). */ checkRate?: (key: string, tier: RateLimitTier) => boolean; + /** + * Additional non-ACP WebSocket routes (e.g. `/voice/stream`) that reuse this + * upgrade listener's security checks. Matched paths skip the ACP init flow. + */ + extraWsRoutes?: readonly ExtraWsRoute[]; +} + +/** + * A non-ACP WebSocket route that shares the daemon's single upgrade listener + * (and therefore its loopback / host-allowlist / CSRF / bearer-token checks) + * instead of attaching a second `'upgrade'` listener — the ACP listener + * `socket.destroy()`s unknown paths, so a competing listener can't coexist. + */ +export interface ExtraWsRoute { + path: string; + onConnection: (ws: WebSocket, req: IncomingMessage) => void; } export interface AcpHttpHandle { @@ -428,13 +484,39 @@ export function mountAcpHttp( function setupWebSocket(httpServer: import('node:http').Server): void { if (wss) return; - wss = new WebSocketServer({ noServer: true, maxPayload: 10 * 1024 * 1024 }); + wss = new WebSocketServer({ + noServer: true, + maxPayload: 10 * 1024 * 1024, + // Browsers authenticate the upgrade by offering the bearer token as a + // `qwen-bearer.*` subprotocol (see extractUpgradeBearer). Never echo that + // secret-bearing value back in the handshake response — select the first + // non-secret subprotocol instead. The web-shell offers a non-secret + // marker (`qwen-ws`) alongside the bearer one precisely so there is always + // a safe value to select: selecting none would make strict WS clients + // (e.g. the `ws` library) reject the handshake with "Server sent no + // subprotocol". ACP clients offer no subprotocol, so this is a no-op for + // them. + handleProtocols: (protocols) => { + for (const proto of protocols) { + if (!proto.startsWith(WS_BEARER_SUBPROTOCOL_PREFIX)) return proto; + } + return false; + }, + }); upgradeServer = httpServer; const expectedTokenHash = opts.token ? createHash('sha256').update(opts.token).digest() : undefined; upgradeListener = (req: IncomingMessage, socket: Duplex, head: Buffer) => { + const rawAddr = + (socket as unknown as { remoteAddress?: string }).remoteAddress ?? + 'ws-unknown'; + const logReject = (reason: string) => { + writeStderrLine( + `qwen serve: WebSocket upgrade rejected (${reason}) from ${rawAddr}`, + ); + }; let url: URL; try { url = new URL( @@ -442,10 +524,15 @@ export function mountAcpHttp( `http://${req.headers.host ?? 'localhost'}`, ); } catch { + logReject('invalid-url'); socket.destroy(); return; } - if (url.pathname !== path) { + const extraRoute = opts.extraWsRoutes?.find( + (route) => route.path === url.pathname, + ); + if (url.pathname !== path && !extraRoute) { + logReject(`unknown-path ${url.pathname}`); socket.destroy(); return; } @@ -467,6 +554,7 @@ export function mountAcpHttp( `host.docker.internal:${localPort}`, ]); if (!allowed.has(host)) { + logReject(`host-not-allowed ${host || '(missing)'}`); socket.write('HTTP/1.1 403 Forbidden\r\n\r\n'); socket.destroy(); return; @@ -485,11 +573,13 @@ export function mountAcpHttp( originHost !== 'localhost' && originHost !== '::1' ) { + logReject(`origin-not-allowed ${originHost}`); socket.write('HTTP/1.1 403 Forbidden\r\n\r\n'); socket.destroy(); return; } } catch { + logReject('invalid-origin'); socket.write('HTTP/1.1 403 Forbidden\r\n\r\n'); socket.destroy(); return; @@ -499,35 +589,38 @@ export function mountAcpHttp( // Auth: WS bypasses Express middleware. Same posture as REST: // loopback without token = allow; non-loopback/token-mismatch = reject. if (opts.token) { - const authHeader = req.headers['authorization']; - if (!authHeader || !authHeader.includes(' ')) { - socket.write('HTTP/1.1 401 Unauthorized\r\n\r\n'); - socket.destroy(); - return; - } - const scheme = authHeader - .slice(0, authHeader.indexOf(' ')) - .toLowerCase(); - const credentials = authHeader - .slice(authHeader.indexOf(' ') + 1) - .trim(); - const actual = createHash('sha256').update(credentials).digest(); + // Accept the token from `Authorization` (non-browser clients) or the + // `qwen-bearer.*` subprotocol (browsers, which can't set Authorization + // on a WebSocket). Hash-compare in constant time, same posture as REST. + const credentials = extractUpgradeBearer(req); + const actual = credentials + ? createHash('sha256').update(credentials).digest() + : undefined; if ( - scheme !== 'bearer' || + !actual || !expectedTokenHash || + actual.length !== expectedTokenHash.length || !timingSafeEqual(expectedTokenHash, actual) ) { + logReject('auth-mismatch'); socket.write('HTTP/1.1 401 Unauthorized\r\n\r\n'); socket.destroy(); return; } } else if (!fromLoopback) { + logReject('non-loopback-without-token'); socket.write('HTTP/1.1 403 Forbidden\r\n\r\n'); socket.destroy(); return; } wss!.handleUpgrade(req, socket, head, (ws: WebSocket) => { + // Non-ACP routes (e.g. voice) own their own protocol — hand the + // upgraded socket off and skip the ACP initialize handshake. + if (extraRoute) { + extraRoute.onConnection(ws, req); + return; + } let initialized = false; const initTimer = setTimeout(() => { if (!initialized) { @@ -540,9 +633,6 @@ export function mountAcpHttp( initTimer.unref?.(); let connRef: AcpConnection | undefined; let messageQueue = Promise.resolve(); - const rawAddr = - (socket as unknown as { remoteAddress?: string }).remoteAddress ?? - 'ws-unknown'; const wsKey = rawAddr.startsWith('::ffff:') ? rawAddr.slice(7) : rawAddr; diff --git a/packages/cli/src/serve/acp-http/transport.test.ts b/packages/cli/src/serve/acp-http/transport.test.ts index 0719c60b15..da85afd840 100644 --- a/packages/cli/src/serve/acp-http/transport.test.ts +++ b/packages/cli/src/serve/acp-http/transport.test.ts @@ -2970,6 +2970,119 @@ describe('ACP WebSocket transport security', () => { ws.close(); }); + // ── Bearer token via Sec-WebSocket-Protocol (browser clients) ────── + // Browsers can't set an Authorization header on a WebSocket, so the token + // rides in a `qwen-bearer.` subprotocol that the upgrade + // listener decodes (extractUpgradeBearer). Matches the web-shell encoder. + function bearerProto(token: string): string { + return `qwen-bearer.${Buffer.from(token).toString('base64url')}`; + } + // Non-secret marker the web-shell offers alongside the bearer subprotocol so + // the daemon can select it (never the secret) and the handshake completes. + const WS_AUTH_SUBPROTOCOL = 'qwen-ws'; + + function wsConnectWithSubprotocols( + protocols: string[], + ): Promise<{ code: number; protocol: string }> { + return new Promise((resolve) => { + const ws = new WebSocket(`ws://127.0.0.1:${port}/acp`, protocols, { + handshakeTimeout: 2000, + }); + ws.once('open', () => { + const { protocol } = ws; + ws.close(); + resolve({ code: 101, protocol }); + }); + ws.once('unexpected-response', (_req, res) => + resolve({ code: res.statusCode ?? 0, protocol: '' }), + ); + ws.once('error', () => resolve({ code: 0, protocol: '' })); + }); + } + + it('accepts WS upgrade with a valid token in the subprotocol', async () => { + await startServer({ token: 'secret-token-123' }); + const result = await wsConnectWithSubprotocols([ + WS_AUTH_SUBPROTOCOL, + bearerProto('secret-token-123'), + ]); + expect(result.code).toBe(101); + }); + + it('falls back to bearer subprotocol when Authorization bearer is empty', async () => { + await startServer({ token: 'secret-token-123' }); + const result = await new Promise<{ code: number }>((resolve) => { + const ws = new WebSocket( + `ws://127.0.0.1:${port}/acp`, + [WS_AUTH_SUBPROTOCOL, bearerProto('secret-token-123')], + { + headers: { Authorization: 'Bearer ' }, + handshakeTimeout: 2000, + }, + ); + ws.once('open', () => { + ws.close(); + resolve({ code: 101 }); + }); + ws.once('unexpected-response', (_req, res) => + resolve({ code: res.statusCode ?? 0 }), + ); + ws.once('error', () => resolve({ code: 0 })); + }); + expect(result.code).toBe(101); + }); + + it('never echoes the secret subprotocol back in the handshake', async () => { + await startServer({ token: 'secret-token-123' }); + const result = await wsConnectWithSubprotocols([ + WS_AUTH_SUBPROTOCOL, + bearerProto('secret-token-123'), + ]); + expect(result.code).toBe(101); + // The daemon selects the non-secret marker, never the bearer value. + expect(result.protocol).toBe(WS_AUTH_SUBPROTOCOL); + expect(result.protocol).not.toContain('qwen-bearer.'); + }); + + it('selects a non-secret subprotocol, never the bearer one', async () => { + await startServer({ token: 'secret-token-123' }); + const result = await wsConnectWithSubprotocols([ + 'acp.v1', + bearerProto('secret-token-123'), + ]); + expect(result.code).toBe(101); + expect(result.protocol).toBe('acp.v1'); + }); + + it('rejects WS upgrade with a wrong token in the subprotocol', async () => { + await startServer({ token: 'secret-token-123' }); + const result = await wsConnectWithSubprotocols([ + WS_AUTH_SUBPROTOCOL, + bearerProto('wrong-token'), + ]); + expect(result.code).toBe(401); + }); + + it('rejects WS upgrade with a malformed bearer subprotocol', async () => { + await startServer({ token: 'secret-token-123' }); + // `----` is a valid subprotocol token but decodes to garbage bytes (not the + // token) — exercises the non-throwing decode + constant-time mismatch path. + const result = await wsConnectWithSubprotocols([ + WS_AUTH_SUBPROTOCOL, + 'qwen-bearer.----', + ]); + expect(result.code).toBe(401); + }); + + it('ignores the subprotocol on a no-token loopback daemon', async () => { + await startServer(); + const result = await wsConnectWithSubprotocols([ + WS_AUTH_SUBPROTOCOL, + bearerProto('anything'), + ]); + expect(result.code).toBe(101); + }); + // ── maxPayload ───────────────────────────────────────────────────── it('closes WS on oversized frame (>10MB)', async () => { await startServer(); diff --git a/packages/cli/src/serve/capabilities.ts b/packages/cli/src/serve/capabilities.ts index 4759d9412f..770325a34e 100644 --- a/packages/cli/src/serve/capabilities.ts +++ b/packages/cli/src/serve/capabilities.ts @@ -232,6 +232,16 @@ export const SERVE_CAPABILITY_REGISTRY = { session_branch: { since: 'v1' }, rate_limit: { since: 'v1' }, workspace_reload: { since: 'v1' }, + // Daemon hosts the `/voice/stream` WebSocket: the browser captures audio and + // streams raw PCM, the daemon transcribes server-side via the configured + // `voiceModel` (credentials never reach the client). Advertised + // UNCONDITIONALLY (like `auth_device_flow`): presence means the endpoint + // exists, not that a voice model is configured. The WS returns an `error` + // frame when no transcribable `voiceModel` is set, so clients probe by + // connecting rather than reading ambient settings into `/capabilities` (which + // would make the envelope depend on the user's home config). `modes` + // enumerates the two transcription paths (realtime vs. on-stop batch). + voice_transcribe: { since: 'v1', modes: ['streaming', 'batch'] }, } as const satisfies Record; export type ServeFeature = keyof typeof SERVE_CAPABILITY_REGISTRY; @@ -251,6 +261,7 @@ export interface AdvertiseFeatureToggles { sessionShellCommandEnabled?: boolean; rateLimit?: boolean; reloadAvailable?: boolean; + voiceWsAvailable?: boolean; } /** @@ -316,6 +327,15 @@ export const CONDITIONAL_SERVE_FEATURES: ReadonlyMap< ], ['rate_limit', (toggles) => toggles.rateLimit === true], ['workspace_reload', (toggles) => toggles.reloadAvailable === true], + [ + // Advertised whenever the `/voice/stream` WS endpoint exists. A configured + // token (or `--require-auth`) no longer suppresses it: browsers can't set + // an `Authorization` header on a WebSocket, so the Web Shell carries the + // bearer token in the `Sec-WebSocket-Protocol` subprotocol, which the ACP + // upgrade listener verifies (see acp-http/index.ts). + 'voice_transcribe', + (toggles) => toggles.voiceWsAvailable !== false, + ], ]); export const SERVE_FEATURES = Object.freeze( diff --git a/packages/cli/src/serve/routes/workspace-settings.ts b/packages/cli/src/serve/routes/workspace-settings.ts index af5fdc4e89..c053cf0ef7 100644 --- a/packages/cli/src/serve/routes/workspace-settings.ts +++ b/packages/cli/src/serve/routes/workspace-settings.ts @@ -33,7 +33,10 @@ const TUI_ONLY_SETTINGS = new Set([ 'ui.enableWelcomeBack', ]); -const WEB_SHELL_SETTINGS = new Set(['ui.compactMode']); +// `voiceModel` is `showInDialog: false` (so not in the dialog allowlist), but +// the Web Shell `/model --voice` picker needs to read + persist it; the daemon +// `/voice/stream` then reads it back via `loadSettings`. +const WEB_SHELL_SETTINGS = new Set(['ui.compactMode', 'voiceModel']); const VALID_WRITE_SCOPES = new Set(['workspace']); diff --git a/packages/cli/src/serve/server.test.ts b/packages/cli/src/serve/server.test.ts index ddd8fa6e6b..2cbd49e4fd 100644 --- a/packages/cli/src/serve/server.test.ts +++ b/packages/cli/src/serve/server.test.ts @@ -225,6 +225,9 @@ const EXPECTED_STAGE1_FEATURES = [ 'session_hooks', 'workspace_extensions', 'session_branch', + // Baseline (always advertised) — presence means the `/voice/stream` + // endpoint exists; the WS errors if no voice model is configured. + 'voice_transcribe', ] as const; // Issue #4175 PR 15. `require_auth` is registered but conditionally @@ -258,7 +261,8 @@ const EXPECTED_REGISTERED_FEATURES = [ f !== 'workspace_hooks' && f !== 'session_hooks' && f !== 'workspace_extensions' && - f !== 'session_branch', + f !== 'session_branch' && + f !== 'voice_transcribe', ), 'workspace_settings', 'workspace_permissions', @@ -284,6 +288,7 @@ const EXPECTED_REGISTERED_FEATURES = [ 'session_branch', 'rate_limit', 'workspace_reload', + 'voice_transcribe', ] as const; interface FakeBridgeOpts { @@ -1509,6 +1514,24 @@ describe('createServeApp', () => { ); }); + it('advertises `voice_transcribe` only when the voice WebSocket route is active', () => { + expect( + getAdvertisedServeFeatures(undefined, { voiceWsAvailable: true }), + ).toContain('voice_transcribe'); + expect( + getAdvertisedServeFeatures(undefined, { voiceWsAvailable: false }), + ).not.toContain('voice_transcribe'); + // A configured token / `--require-auth` no longer suppresses voice: the + // browser carries the bearer token via the WS subprotocol, which the + // upgrade listener verifies. + expect( + getAdvertisedServeFeatures(undefined, { + requireAuth: true, + voiceWsAvailable: true, + }), + ).toContain('voice_transcribe'); + }); + it('honors every entry in CONDITIONAL_SERVE_FEATURES (PR #4236 review #3254467192 — drift insurance)', () => { // Iterate the Map so any future conditional tag added here whose // predicate isn't honored by `getAdvertisedServeFeatures` fails @@ -1658,6 +1681,27 @@ describe('createServeApp', () => { ); continue; } + if (feature === 'voice_transcribe') { + expect(predicate({ voiceWsAvailable: true })).toBe(true); + expect(predicate({ voiceWsAvailable: false })).toBe(false); + // requireAuth no longer suppresses voice (token rides the WS + // subprotocol), so the predicate ignores it. + expect(predicate({ requireAuth: true, voiceWsAvailable: true })).toBe( + true, + ); + expect(predicate({})).toBe(true); + expect( + getAdvertisedServeFeatures(undefined, { + voiceWsAvailable: true, + }), + ).toContain(feature); + expect( + getAdvertisedServeFeatures(undefined, { + voiceWsAvailable: false, + }), + ).not.toContain(feature); + continue; + } // Future conditional tag. Authors must add a branch above with // the toggle field that drives this predicate. Failing here is // intentional: it forces the new conditional tag to ship with a @@ -8706,6 +8750,10 @@ describe('createServeApp', () => { v: 1, detail: 'summary', }); + // Voice is advertised even with a token configured: browsers authenticate + // the WS via the `qwen-bearer.*` subprotocol, so the token no longer + // suppresses the capability. + expect(withAuth.body.capabilities.features).toContain('voice_transcribe'); }); it('returns summary diagnostics without querying workspace status', async () => { diff --git a/packages/cli/src/serve/server.ts b/packages/cli/src/serve/server.ts index d3f4c0d862..50a6c0e0bb 100644 --- a/packages/cli/src/serve/server.ts +++ b/packages/cli/src/serve/server.ts @@ -67,6 +67,7 @@ import { loadSettings } from '../config/settings.js'; import { isWorkspaceTrusted } from '../config/trustedFolders.js'; import { isLoopbackBind } from './loopback-binds.js'; import { mountAcpHttp, type AcpHttpHandle } from './acp-http/index.js'; +import { createVoiceWsConnectionHandler } from './voice/voice-ws.js'; import { buildDaemonStatusResponse, parseDaemonStatusDetail, @@ -2003,6 +2004,11 @@ export function createServeApp( sessionShellCommandEnabled, rateLimit: opts.rateLimit === true, reloadAvailable: deps.workspace !== undefined, + // Advertised whenever the `/voice/stream` WS endpoint exists (ACP HTTP + // on). A configured token no longer suppresses it — the browser carries + // the bearer token via the WS subprotocol, which the upgrade listener + // verifies (acp-http/index.ts). + voiceWsAvailable: process.env['QWEN_SERVE_ACP_HTTP'] !== '0', }); const acpHandleRef: { current?: AcpHttpHandle } = {}; @@ -4907,6 +4913,15 @@ export function createServeApp( token: opts.token, sessionShellCommandEnabled, checkRate: rateLimiter?.checkRate, + // Browser captures audio and streams raw PCM here; the daemon transcribes + // server-side via the reused CLI voice pipeline. Shares the ACP upgrade + // listener's loopback/CSRF/bearer checks. + extraWsRoutes: [ + { + path: '/voice/stream', + onConnection: createVoiceWsConnectionHandler(boundWorkspace), + }, + ], }); if (acpHandleRef.current) { app.locals['acpHandle'] = acpHandleRef.current; diff --git a/packages/cli/src/serve/voice/resolve-voice-config.ts b/packages/cli/src/serve/voice/resolve-voice-config.ts new file mode 100644 index 0000000000..51153c5229 --- /dev/null +++ b/packages/cli/src/serve/voice/resolve-voice-config.ts @@ -0,0 +1,87 @@ +/** + * @license + * Copyright 2025 Qwen + * SPDX-License-Identifier: Apache-2.0 + */ + +import { ModelsConfig } from '@qwen-code/qwen-code-core'; +import { loadSettings } from '../../config/settings.js'; +import type { LoadedSettings } from '../../config/settings.js'; +import { + getAuthTypeFromEnv, + resolveCliGenerationConfig, +} from '../../utils/modelConfigUtils.js'; +import { + isStreamingVoiceModel, + resolveVoiceTranscriptionConfig, + type VoiceModelLookup, +} from '../../ui/voice/voice-transcriber.js'; + +/** + * Fully-validated voice context for a daemon workspace. The browser captures + * audio and streams raw PCM to `/voice/stream`; the daemon resolves the + * configured voice model here (reusing the CLI voice resolver) and transcribes + * server-side so provider credentials never reach the client. + */ +export interface DaemonVoiceContext { + settings: LoadedSettings; + /** A `ModelsConfig` — satisfies the resolver's structural `getAllConfiguredModels`. */ + models: VoiceModelLookup; + voiceModel: string; + /** True for realtime models (open an upstream WS); false → batch on stop. */ + streaming: boolean; +} + +function readVoiceModel(settings: LoadedSettings): string | undefined { + const raw = (settings.merged as { voiceModel?: unknown }).voiceModel; + if (typeof raw !== 'string') return undefined; + const trimmed = raw.trim(); + return trimmed.length > 0 ? trimmed : undefined; +} + +/** + * Build a `ModelsConfig` from workspace settings, mirroring + * `workspace-providers-status.ts` so the daemon resolves the same configured + * models the CLI would — without constructing a full CLI `Config`. + */ +function buildModelsConfig(settings: LoadedSettings): ModelsConfig { + const merged = settings.merged; + const selectedAuthType = + merged.security?.auth?.selectedType ?? getAuthTypeFromEnv(); + const resolvedCliConfig = resolveCliGenerationConfig({ + argv: {}, + settings: merged, + selectedAuthType, + env: process.env as Record, + }); + return new ModelsConfig({ + initialAuthType: selectedAuthType, + modelProvidersConfig: merged.modelProviders, + generationConfig: resolvedCliConfig.generationConfig, + generationConfigSources: resolvedCliConfig.sources, + }); +} + +/** + * Load and validate the workspace's voice configuration. Throws when voice is + * not usable (no `voiceModel` configured, model not transcribable, missing + * baseUrl/apiKey) — the throw message is a safe, user-facing reason. + */ +export function loadDaemonVoiceContext( + workspaceCwd: string, +): DaemonVoiceContext { + const settings = loadSettings(workspaceCwd); + const voiceModel = readVoiceModel(settings); + if (!voiceModel) { + throw new Error('No voice model is configured for this workspace.'); + } + const models = buildModelsConfig(settings); + // Validates transcribable + baseUrl + apiKey presence (throws otherwise). + resolveVoiceTranscriptionConfig({ config: models, settings, voiceModel }); + return { + settings, + models, + voiceModel, + streaming: isStreamingVoiceModel(voiceModel), + }; +} diff --git a/packages/cli/src/serve/voice/voice-ws.test.ts b/packages/cli/src/serve/voice/voice-ws.test.ts new file mode 100644 index 0000000000..616f9d2fda --- /dev/null +++ b/packages/cli/src/serve/voice/voice-ws.test.ts @@ -0,0 +1,475 @@ +/** + * @license + * Copyright 2025 Qwen + * SPDX-License-Identifier: Apache-2.0 + */ + +// @vitest-environment node + +import { afterEach, describe, it, expect, vi } from 'vitest'; +import { createVoiceWsConnectionHandler } from './voice-ws.js'; +import type { DaemonVoiceContext } from './resolve-voice-config.js'; +import type { VoiceStreamSession } from '../../ui/voice/voice-stream-session.js'; + +/** Minimal stand-in for a `ws` WebSocket the handler attaches to. */ +class FakeWs { + readonly OPEN = 1; + readyState = 1; + readonly sent: Array = []; + closeCode: number | undefined; + private handlers: Record void>> = {}; + + constructor(private readonly emitCloseOnClose = true) {} + + on(event: string, cb: (...args: unknown[]) => void): this { + (this.handlers[event] ??= []).push(cb); + return this; + } + send(data: string | Uint8Array): void { + this.sent.push(typeof data === 'string' ? data : 'binary'); + } + close(code?: number): void { + this.closeCode = code; + this.readyState = 3; + if (this.emitCloseOnClose) this.emit('close'); + } + emit(event: string, ...args: unknown[]): void { + (this.handlers[event] ?? []).forEach((cb) => cb(...args)); + } + + // ── test drivers ── + text(obj: unknown): void { + this.emit('message', Buffer.from(JSON.stringify(obj)), false); + } + binary(bytes: number[]): void { + this.emit('message', Buffer.from(bytes), true); + } + binaryBuffer(buffer: Buffer): void { + this.emit('message', buffer, true); + } + frames(): Array> { + return this.sent + .filter((s): s is string => s !== 'binary') + .map((s) => JSON.parse(s) as Record); + } +} + +const tick = () => new Promise((resolve) => setImmediate(resolve)); + +function deferred() { + let resolve!: (value: T) => void; + const promise = new Promise((res) => { + resolve = res; + }); + return { promise, resolve }; +} + +function streamingCtx(): DaemonVoiceContext { + return { + settings: {} as DaemonVoiceContext['settings'], + models: { getAllConfiguredModels: () => [] }, + voiceModel: 'paraformer-realtime-v2', + streaming: true, + }; +} + +function batchCtx(): DaemonVoiceContext { + return { + settings: {} as DaemonVoiceContext['settings'], + models: { getAllConfiguredModels: () => [] }, + voiceModel: 'qwen3-asr-flash', + streaming: false, + }; +} + +describe('createVoiceWsConnectionHandler', () => { + afterEach(() => { + vi.useRealTimers(); + }); + + it('streams audio to the upstream session and returns the final transcript', async () => { + const pushed: Uint8Array[] = []; + let onInterim: ((t: string) => void) | undefined; + const session: VoiceStreamSession = { + pushAudio: (pcm) => pushed.push(pcm), + finish: vi.fn(async () => 'hello world'), + abort: vi.fn(), + }; + const openStream = vi.fn(async (_ctx, callbacks) => { + onInterim = callbacks.onInterim; + return session; + }); + const ws = new FakeWs(); + const handler = createVoiceWsConnectionHandler('/ws', { + loadContext: () => streamingCtx(), + openStream, + }); + handler(ws as never, {} as never); + + ws.text({ type: 'start' }); + await tick(); + expect(openStream).toHaveBeenCalledOnce(); + expect(ws.frames()[0]).toMatchObject({ type: 'ready', streaming: true }); + + ws.binary([1, 2, 3, 4]); + await tick(); + expect(pushed).toHaveLength(1); + + onInterim?.('hel'); + expect( + ws.frames().some((f) => f['type'] === 'interim' && f['text'] === 'hel'), + ).toBe(true); + + ws.text({ type: 'stop' }); + await tick(); + expect(session.finish).toHaveBeenCalledOnce(); + expect(ws.frames().at(-1)).toMatchObject({ + type: 'final', + text: 'hello world', + }); + expect(ws.closeCode).toBe(1000); + }); + + it('lazily starts on the first audio frame', async () => { + const loadContext = vi.fn(() => streamingCtx()); + const session: VoiceStreamSession = { + pushAudio: vi.fn(), + finish: vi.fn(async () => ''), + abort: vi.fn(), + }; + const ws = new FakeWs(); + const handler = createVoiceWsConnectionHandler('/ws', { + loadContext, + openStream: async () => session, + }); + handler(ws as never, {} as never); + + ws.binary([9, 9]); + await tick(); + expect(loadContext).toHaveBeenCalledOnce(); + expect(session.pushAudio).toHaveBeenCalledOnce(); + }); + + it('buffers audio and batch-transcribes non-streaming models on stop', async () => { + const transcribe = vi.fn( + async (_ctx, pcm: Uint8Array) => `batched:${pcm.byteLength}`, + ); + const ws = new FakeWs(); + const handler = createVoiceWsConnectionHandler('/ws', { + loadContext: () => batchCtx(), + transcribe, + }); + handler(ws as never, {} as never); + + ws.text({ type: 'start' }); + await tick(); + ws.binary([1, 2, 3]); + ws.binary([4, 5]); + await tick(); + ws.text({ type: 'stop' }); + await tick(); + + expect(transcribe).toHaveBeenCalledOnce(); + // The two 3- and 2-byte frames concatenate to 5 bytes. + expect(ws.frames().at(-1)).toMatchObject({ + type: 'final', + text: 'batched:5', + }); + }); + + it('reports no-model voice config errors to the client', async () => { + const ws = new FakeWs(); + const handler = createVoiceWsConnectionHandler('/ws', { + loadContext: () => { + throw new Error('No voice model is configured for this workspace.'); + }, + }); + handler(ws as never, {} as never); + + ws.text({ type: 'start' }); + await tick(); + expect(ws.frames().at(-1)).toMatchObject({ + type: 'error', + message: 'No voice model is configured for this workspace.', + }); + expect(ws.closeCode).toBe(1011); + }); + + it('keeps unexpected voice config errors generic', async () => { + const ws = new FakeWs(); + const handler = createVoiceWsConnectionHandler('/ws', { + loadContext: () => { + throw new Error('DASHSCOPE_API_KEY from /private/config is invalid'); + }, + }); + handler(ws as never, {} as never); + + ws.text({ type: 'start' }); + await tick(); + expect(ws.frames().at(-1)).toMatchObject({ + type: 'error', + message: 'Voice transcription failed. Please try again.', + }); + expect(ws.closeCode).toBe(1011); + }); + + it('reports a generic error frame when streaming finalization fails', async () => { + const session: VoiceStreamSession = { + pushAudio: vi.fn(), + finish: vi.fn(async () => { + throw new Error('upstream private endpoint failed'); + }), + abort: vi.fn(), + }; + const ws = new FakeWs(); + const handler = createVoiceWsConnectionHandler('/ws', { + loadContext: () => streamingCtx(), + openStream: async () => session, + }); + handler(ws as never, {} as never); + + ws.text({ type: 'start' }); + await tick(); + ws.text({ type: 'stop' }); + await tick(); + + expect(ws.frames().at(-1)).toMatchObject({ + type: 'error', + message: 'Voice transcription failed. Please try again.', + }); + expect(ws.closeCode).toBe(1011); + }); + + it('aborts the upstream session on abort', async () => { + const session: VoiceStreamSession = { + pushAudio: vi.fn(), + finish: vi.fn(async () => ''), + abort: vi.fn(), + }; + const ws = new FakeWs(); + const handler = createVoiceWsConnectionHandler('/ws', { + loadContext: () => streamingCtx(), + openStream: async () => session, + }); + handler(ws as never, {} as never); + + ws.text({ type: 'start' }); + await tick(); + ws.text({ type: 'abort' }); + await tick(); + expect(session.abort).toHaveBeenCalledOnce(); + expect(ws.closeCode).toBe(1000); + }); + + it('lets abort preempt a pending streaming start', async () => { + const sessionReady = deferred(); + const session: VoiceStreamSession = { + pushAudio: vi.fn(), + finish: vi.fn(async () => ''), + abort: vi.fn(), + }; + const ws = new FakeWs(); + const handler = createVoiceWsConnectionHandler('/ws', { + loadContext: () => streamingCtx(), + openStream: async () => sessionReady.promise, + }); + handler(ws as never, {} as never); + + ws.text({ type: 'start' }); + await tick(); + ws.text({ type: 'abort' }); + await tick(); + sessionReady.resolve(session); + await tick(); + + expect(ws.closeCode).toBe(1000); + expect(session.abort).toHaveBeenCalledOnce(); + }); + + it('does not finalize after abort closes a pending streaming start', async () => { + const sessionReady = deferred(); + const session: VoiceStreamSession = { + pushAudio: vi.fn(), + finish: vi.fn(async () => 'late final'), + abort: vi.fn(), + }; + const ws = new FakeWs(); + const handler = createVoiceWsConnectionHandler('/ws', { + loadContext: () => streamingCtx(), + openStream: async () => sessionReady.promise, + }); + handler(ws as never, {} as never); + + ws.text({ type: 'stop' }); + await tick(); + ws.text({ type: 'abort' }); + await tick(); + sessionReady.resolve(session); + await tick(); + + expect(session.abort).toHaveBeenCalledOnce(); + expect(session.finish).not.toHaveBeenCalled(); + expect(ws.frames().some((frame) => frame['type'] === 'final')).toBe(false); + }); + + it('aborts a streaming session that resolves after the socket closed', async () => { + const sessionReady = deferred(); + const session: VoiceStreamSession = { + pushAudio: vi.fn(), + finish: vi.fn(async () => ''), + abort: vi.fn(), + }; + const ws = new FakeWs(); + const handler = createVoiceWsConnectionHandler('/ws', { + loadContext: () => streamingCtx(), + openStream: async () => sessionReady.promise, + }); + handler(ws as never, {} as never); + + ws.text({ type: 'start' }); + await tick(); + ws.close(); + sessionReady.resolve(session); + await tick(); + + expect(session.abort).toHaveBeenCalledOnce(); + }); + + it('aborts the streaming session if finalization times out', async () => { + vi.useFakeTimers(); + const session: VoiceStreamSession = { + pushAudio: vi.fn(), + finish: vi.fn(() => new Promise(() => {})), + abort: vi.fn(), + }; + const ws = new FakeWs(); + const handler = createVoiceWsConnectionHandler('/ws', { + loadContext: () => streamingCtx(), + openStream: async () => session, + }); + handler(ws as never, {} as never); + + ws.text({ type: 'start' }); + await vi.runAllTicks(); + ws.text({ type: 'stop' }); + await vi.runAllTicks(); + await vi.advanceTimersByTimeAsync(6 * 60_000); + + expect(session.abort).toHaveBeenCalledOnce(); + expect(ws.frames().at(-1)).toMatchObject({ + type: 'error', + message: 'Voice session exceeded the time limit.', + }); + expect(ws.closeCode).toBe(1011); + }); + + it('fails oversized batch audio before buffering it', async () => { + const ws = new FakeWs(); + const handler = createVoiceWsConnectionHandler('/ws', { + loadContext: () => batchCtx(), + }); + handler(ws as never, {} as never); + + ws.text({ type: 'start' }); + await tick(); + ws.binaryBuffer(Buffer.alloc(10 * 1024 * 1024 + 1)); + await tick(); + + expect(ws.frames().at(-1)).toMatchObject({ + type: 'error', + message: 'Recording is too long for transcription (max ~5 minutes).', + }); + expect(ws.closeCode).toBe(1011); + }); + + it('rejects queued audio while streaming start is pending', async () => { + const sessionReady = deferred(); + const ws = new FakeWs(); + const handler = createVoiceWsConnectionHandler('/ws', { + loadContext: () => streamingCtx(), + openStream: async () => sessionReady.promise, + }); + handler(ws as never, {} as never); + + ws.text({ type: 'start' }); + await tick(); + ws.binaryBuffer(Buffer.alloc(20 * 1024 * 1024 + 1)); + await tick(); + + expect(ws.frames().at(-1)).toMatchObject({ + type: 'error', + message: 'Queued voice audio exceeded the memory limit.', + }); + expect(ws.closeCode).toBe(1011); + }); + + it('fails and cleans up when the hard timer fires', async () => { + vi.useFakeTimers(); + const session: VoiceStreamSession = { + pushAudio: vi.fn(), + finish: vi.fn(async () => ''), + abort: vi.fn(), + }; + const ws = new FakeWs(); + const handler = createVoiceWsConnectionHandler('/ws', { + loadContext: () => streamingCtx(), + openStream: async () => session, + }); + handler(ws as never, {} as never); + + ws.text({ type: 'start' }); + await vi.runAllTicks(); + await vi.advanceTimersByTimeAsync(6 * 60_000); + + expect(session.abort).toHaveBeenCalledOnce(); + expect(ws.frames().at(-1)).toMatchObject({ + type: 'error', + message: 'Voice session exceeded the time limit.', + }); + expect(ws.closeCode).toBe(1011); + }); + + it('frees a voice slot when a failed socket ignores close', async () => { + vi.useFakeTimers(); + const handler = createVoiceWsConnectionHandler('/ws', { + loadContext: () => streamingCtx(), + openStream: async () => ({ + pushAudio: vi.fn(), + finish: vi.fn(async () => ''), + abort: vi.fn(), + }), + }); + const open = Array.from({ length: 8 }, () => new FakeWs(false)); + for (const ws of open) handler(ws as never, {} as never); + + await vi.advanceTimersByTimeAsync(6 * 60_000); + + const next = new FakeWs(); + handler(next as never, {} as never); + expect(next.closeCode).not.toBe(1013); + }); + + it('rejects connections past the concurrency cap and frees slots on close', async () => { + const handler = createVoiceWsConnectionHandler('/ws', { + loadContext: () => streamingCtx(), + openStream: async () => ({ + pushAudio: vi.fn(), + finish: vi.fn(async () => ''), + abort: vi.fn(), + }), + }); + // Open the cap (8) and hold them; the 9th must be refused with 1013. + const open = Array.from({ length: 8 }, () => new FakeWs()); + for (const ws of open) handler(ws as never, {} as never); + const overflow = new FakeWs(); + handler(overflow as never, {} as never); + expect(overflow.closeCode).toBe(1013); + expect(overflow.frames().at(-1)).toMatchObject({ type: 'error' }); + + // Closing one frees a slot for a new connection. + open[0].close(); + const next = new FakeWs(); + handler(next as never, {} as never); + expect(next.closeCode).not.toBe(1013); + }); +}); diff --git a/packages/cli/src/serve/voice/voice-ws.ts b/packages/cli/src/serve/voice/voice-ws.ts new file mode 100644 index 0000000000..7ac47b8e24 --- /dev/null +++ b/packages/cli/src/serve/voice/voice-ws.ts @@ -0,0 +1,442 @@ +/** + * @license + * Copyright 2025 Qwen + * SPDX-License-Identifier: Apache-2.0 + */ + +import type { IncomingMessage } from 'node:http'; +import type { RawData, WebSocket } from 'ws'; +import { createDebugLogger } from '@qwen-code/qwen-code-core'; +import { + loadDaemonVoiceContext, + type DaemonVoiceContext, +} from './resolve-voice-config.js'; +import { + assertVoiceBaseUrlNetworkAllowed, + resolveVoiceStreamConfig, + transcribeVoiceAudio, +} from '../../ui/voice/voice-transcriber.js'; +import { openVoiceStream } from '../../ui/voice/voice-stream-session.js'; +import { openQwenAsrRealtimeStream } from '../../ui/voice/qwen-asr-realtime-session.js'; +import { openVoiceStreamWithRetry } from '../../ui/voice/voice-stream-retry.js'; +import { writeStderrLine } from '../../utils/stdioHelpers.js'; +import type { + VoiceStreamCallbacks, + VoiceStreamSession, +} from '../../ui/voice/voice-stream-session.js'; + +const debugLogger = createDebugLogger('VOICE_WS'); + +// Qwen-ASR caps each audio file at 10 MB / ~5 minutes; guard the batch buffer +// before WAV-encoding so an overlong stream fails with a clear message. +const MAX_BATCH_AUDIO_BYTES = 10 * 1024 * 1024; +const MAX_QUEUED_AUDIO_BYTES = MAX_BATCH_AUDIO_BYTES * 2; +// Hard cap on a single voice connection so a client that opens the socket and +// never sends `stop` can't pin an upstream ASR session indefinitely. +const MAX_CONNECTION_MS = 6 * 60_000; +// Voice WS bypasses the ACP connection registry; cap concurrent sessions so a +// client can't open unbounded sockets (each opens an upstream ASR connection). +// Generous for one interactive user across a few tabs. +const MAX_CONCURRENT_VOICE_SESSIONS = 8; +const GENERIC_TRANSCRIPTION_ERROR = + 'Voice transcription failed. Please try again.'; +const NO_VOICE_MODEL_ERROR = 'No voice model is configured for this workspace.'; + +// Audio is 16 kHz mono signed-16-bit PCM. Browser capture sends raw frames; the +// batch transcription path (non-streaming models) wants a WAV container. +const SAMPLE_RATE = 16_000; + +function encodeWav(pcm: Uint8Array): Uint8Array { + const header = Buffer.alloc(44); + const dataLen = pcm.byteLength; + header.write('RIFF', 0); + header.writeUInt32LE(36 + dataLen, 4); + header.write('WAVE', 8); + header.write('fmt ', 12); + header.writeUInt32LE(16, 16); // PCM fmt chunk size + header.writeUInt16LE(1, 20); // audio format = PCM + header.writeUInt16LE(1, 22); // channels = mono + header.writeUInt32LE(SAMPLE_RATE, 24); + header.writeUInt32LE(SAMPLE_RATE * 2, 28); // byte rate (mono, 16-bit) + header.writeUInt16LE(2, 32); // block align + header.writeUInt16LE(16, 34); // bits per sample + header.write('data', 36); + header.writeUInt32LE(dataLen, 40); + return Buffer.concat([header, Buffer.from(pcm)]); +} + +/** Injection seams for unit tests; production uses the reused CLI pipeline. */ +export interface VoiceWsDeps { + loadContext?: (workspaceCwd: string) => DaemonVoiceContext; + openStream?: ( + ctx: DaemonVoiceContext, + callbacks: VoiceStreamCallbacks, + ) => Promise; + transcribe?: (ctx: DaemonVoiceContext, pcm: Uint8Array) => Promise; +} + +async function defaultOpenStream( + ctx: DaemonVoiceContext, + callbacks: VoiceStreamCallbacks, +): Promise { + try { + const cfg = resolveVoiceStreamConfig({ + config: ctx.models, + settings: ctx.settings, + voiceModel: ctx.voiceModel, + }); + await assertVoiceBaseUrlNetworkAllowed(cfg); + return await openVoiceStreamWithRetry(() => + cfg.transport === 'qwen-asr-realtime' + ? openQwenAsrRealtimeStream(cfg, callbacks) + : openVoiceStream(cfg, callbacks), + ); + } catch (error) { + debugLogger.debug(`[voice-ws] stream open error: ${errMessage(error)}`); + throw new Error(GENERIC_TRANSCRIPTION_ERROR); + } +} + +function defaultTranscribe( + ctx: DaemonVoiceContext, + pcm: Uint8Array, +): Promise { + return transcribeVoiceAudio( + { data: encodeWav(pcm), mimeType: 'audio/wav' }, + { config: ctx.models, settings: ctx.settings, voiceModel: ctx.voiceModel }, + ).catch((error: unknown) => { + debugLogger.debug( + `[voice-ws] batch transcription error: ${errMessage(error)}`, + ); + throw new Error(GENERIC_TRANSCRIPTION_ERROR); + }); +} + +function errMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +function voiceConfigErrorMessage(error: unknown): string { + const message = errMessage(error); + return message === NO_VOICE_MODEL_ERROR + ? message + : GENERIC_TRANSCRIPTION_ERROR; +} + +/** Normalize a `ws` frame (Buffer | ArrayBuffer | Buffer[]) to a Buffer. */ +function toBuffer(data: RawData): Buffer { + if (Buffer.isBuffer(data)) return data; + if (Array.isArray(data)) return Buffer.concat(data); + return Buffer.from(data as ArrayBuffer); +} + +interface ControlMessage { + type: 'start' | 'stop' | 'abort'; +} + +function parseControl(text: string): ControlMessage | undefined { + let parsed: unknown; + try { + parsed = JSON.parse(text); + } catch { + return undefined; + } + const type = (parsed as { type?: unknown })?.type; + if (type === 'start' || type === 'stop' || type === 'abort') { + return { type }; + } + return undefined; +} + +/** + * Build the per-connection handler for the daemon `/voice/stream` WebSocket. + * + * Protocol — client → server: + * - text `{"type":"start"}` open the upstream session (optional; lazily + * opened on first audio frame otherwise) + * - binary raw s16le/16 kHz/mono PCM frames + * - text `{"type":"stop"}` finalize and return the transcript + * - text `{"type":"abort"}` discard and close + * + * server → client: + * - `{"type":"ready","streaming":bool,"model":string}` + * - `{"type":"interim","text":string}` (streaming models only) + * - `{"type":"final","text":string}` + * - `{"type":"error","message":string}` + * + * Capture happens in the browser; the daemon reuses the CLI transcription + * pipeline so provider credentials never leave the server. + */ +export function createVoiceWsConnectionHandler( + boundWorkspace: string, + deps: VoiceWsDeps = {}, +): (ws: WebSocket, req: IncomingMessage) => void { + const loadContext = deps.loadContext ?? loadDaemonVoiceContext; + const openStream = deps.openStream ?? defaultOpenStream; + const transcribe = deps.transcribe ?? defaultTranscribe; + // Shared across all connections from this daemon (factory closure). + let activeSessions = 0; + + return (ws: WebSocket) => { + if (activeSessions >= MAX_CONCURRENT_VOICE_SESSIONS) { + writeStderrLine( + `qwen serve: voice websocket rejected; activeSessions=${activeSessions}`, + ); + try { + ws.send( + JSON.stringify({ + type: 'error', + message: 'Too many voice sessions in progress; try again shortly.', + }), + ); + ws.close(1013, 'busy'); + } catch { + // ignore + } + return; + } + activeSessions++; + writeStderrLine( + `qwen serve: voice websocket accepted; activeSessions=${activeSessions}`, + ); + let released = false; + const releaseSlot = () => { + if (!released) { + released = true; + activeSessions--; + writeStderrLine( + `qwen serve: voice websocket slot released; activeSessions=${activeSessions}`, + ); + } + }; + + let state: 'idle' | 'active' | 'finalizing' | 'closed' = 'idle'; + let ctx: DaemonVoiceContext | undefined; + let session: VoiceStreamSession | undefined; + let sessionPromise: Promise | undefined; + const pcmChunks: Uint8Array[] = []; + let bufferedBytes = 0; + let queuedBytes = 0; + let pendingOperations = 0; + // Serialize message handling so async start/push/finalize never interleave. + let chain: Promise = Promise.resolve(); + + const hardTimer = setTimeout(() => { + if (state !== 'closed') fail('Voice session exceeded the time limit.'); + }, MAX_CONNECTION_MS); + hardTimer.unref?.(); + + // Read `state` through a helper so an async error path (e.g. a failed + // `ensureStarted`) that flips it to 'closed' isn't flow-narrowed away by + // an earlier guard. + const isClosed = (): boolean => state === 'closed'; + + const releaseSlotWhenIdle = (): void => { + if (state === 'closed' && pendingOperations === 0) releaseSlot(); + }; + + const sendJson = (obj: unknown): void => { + if (ws.readyState === ws.OPEN) { + try { + ws.send(JSON.stringify(obj)); + } catch { + // socket already going away — nothing to do + } + } + }; + + function cleanup(): void { + state = 'closed'; + clearTimeout(hardTimer); + if (session) { + try { + session.abort(); + } catch { + // best effort + } + session = undefined; + } + sessionPromise = undefined; + pcmChunks.length = 0; + bufferedBytes = 0; + queuedBytes = 0; + } + + function fail(message: string): void { + if (state === 'closed') return; + writeStderrLine(`qwen serve: voice websocket failed: ${message}`); + sendJson({ type: 'error', message }); + cleanup(); + releaseSlotWhenIdle(); + try { + ws.close(1011, 'voice error'); + } catch { + // ignore + } + } + + async function ensureStarted(): Promise { + if (ctx) return; + try { + ctx = loadContext(boundWorkspace); + } catch (error) { + debugLogger.debug( + `[voice-ws] load context error: ${errMessage(error)}`, + ); + fail(voiceConfigErrorMessage(error)); + return; + } + sendJson({ + type: 'ready', + streaming: ctx.streaming, + model: ctx.voiceModel, + }); + if (ctx.streaming) { + const callbacks: VoiceStreamCallbacks = { + onInterim: (text) => sendJson({ type: 'interim', text }), + onError: (error) => { + debugLogger.debug( + `[voice-ws] upstream error: ${errMessage(error)}`, + ); + fail(GENERIC_TRANSCRIPTION_ERROR); + }, + }; + const opening = openStream(ctx, callbacks); + sessionPromise = opening; + const opened = await opening; + if (state === 'closed') { + try { + opened.abort(); + } catch { + // best effort + } + return; + } + session = opened; + } + if (state === 'idle') state = 'active'; + } + + async function finalize(): Promise { + if (state === 'closed' || state === 'finalizing') return; + state = 'finalizing'; + await ensureStarted(); + if (isClosed()) return; + let transcript = ''; + if (ctx!.streaming) { + const active = + session ?? (sessionPromise ? await sessionPromise : undefined); + if (isClosed()) return; + if (active) { + try { + transcript = await active.finish(); + } finally { + session = undefined; + } + } + } else if (pcmChunks.length > 0) { + transcript = await transcribe(ctx!, Buffer.concat(pcmChunks)); + } + sendJson({ type: 'final', text: transcript }); + writeStderrLine('qwen serve: voice websocket finalized successfully'); + cleanup(); + try { + ws.close(1000, 'done'); + } catch { + // ignore + } + } + + async function handleMessage( + data: Buffer, + isBinary: boolean, + ): Promise { + if (state === 'closed' || state === 'finalizing') return; + if (isBinary) { + await ensureStarted(); + if (isClosed()) return; + if (ctx!.streaming) { + const active = + session ?? (sessionPromise ? await sessionPromise : undefined); + active?.pushAudio(data); + } else { + bufferedBytes += data.byteLength; + if (bufferedBytes > MAX_BATCH_AUDIO_BYTES) { + fail('Recording is too long for transcription (max ~5 minutes).'); + return; + } + pcmChunks.push(data); + } + return; + } + const control = parseControl(data.toString('utf8')); + if (!control) return; + switch (control.type) { + case 'start': + writeStderrLine('qwen serve: voice websocket start received'); + await ensureStarted(); + return; + case 'stop': + writeStderrLine('qwen serve: voice websocket stop received'); + await finalize(); + return; + default: + return; + } + } + + ws.on('message', (data: RawData, isBinary: boolean) => { + const buf = toBuffer(data); + if (!isBinary) { + const control = parseControl(buf.toString('utf8')); + if (control?.type === 'abort') { + writeStderrLine('qwen serve: voice websocket abort received'); + cleanup(); + try { + ws.close(1000, 'aborted'); + } catch { + // ignore + } + releaseSlotWhenIdle(); + return; + } + } + const queuedSize = isBinary ? buf.byteLength : 0; + if (queuedSize > 0) { + queuedBytes += queuedSize; + if (queuedBytes > MAX_QUEUED_AUDIO_BYTES) { + fail('Queued voice audio exceeded the memory limit.'); + releaseSlotWhenIdle(); + return; + } + } + chain = chain + .then(async () => { + pendingOperations++; + try { + await handleMessage(buf, isBinary); + } finally { + if (queuedSize > 0) { + queuedBytes = Math.max(0, queuedBytes - queuedSize); + } + pendingOperations--; + releaseSlotWhenIdle(); + } + }) + .catch((error: unknown) => { + debugLogger.debug(`[voice-ws] ${errMessage(error)}`); + fail(GENERIC_TRANSCRIPTION_ERROR); + releaseSlotWhenIdle(); + }); + }); + ws.on('close', () => { + if (state !== 'closed') cleanup(); + releaseSlotWhenIdle(); + }); + ws.on('error', (error: Error) => { + debugLogger.debug(`[voice-ws] socket error: ${error.message}`); + if (state !== 'closed') cleanup(); + releaseSlotWhenIdle(); + }); + }; +} diff --git a/packages/cli/src/serve/web-shell-static.ts b/packages/cli/src/serve/web-shell-static.ts index f1c63275e8..3f431bcf40 100644 --- a/packages/cli/src/serve/web-shell-static.ts +++ b/packages/cli/src/serve/web-shell-static.ts @@ -116,8 +116,12 @@ function createSendIndex(webShellDir: string): (res: Response) => void { .set('X-Content-Type-Options', 'nosniff') .set('Referrer-Policy', 'no-referrer') .set( + // `microphone=(self)` lets the same-origin Web Shell document request + // the mic for voice dictation (the prompt won't even appear under an + // empty `microphone=()` allowlist). Still blocks cross-origin iframes; + // camera/geolocation/payment stay disabled (unused). 'Permissions-Policy', - 'camera=(), microphone=(), geolocation=(), payment=()', + 'camera=(), microphone=(self), geolocation=(), payment=()', ) .set('Cache-Control', 'no-cache'); // `dotfiles: 'allow'` is required because the resolved path may pass diff --git a/packages/cli/src/ui/voice/voice-transcriber.ts b/packages/cli/src/ui/voice/voice-transcriber.ts index ea33f9d4c2..bbfa6888ca 100644 --- a/packages/cli/src/ui/voice/voice-transcriber.ts +++ b/packages/cli/src/ui/voice/voice-transcriber.ts @@ -8,7 +8,7 @@ import process from 'node:process'; import { lookup as dnsLookup } from 'node:dns/promises'; import { isIP } from 'node:net'; import { createDebugLogger } from '@qwen-code/qwen-code-core'; -import type { AvailableModel, Config } from '@qwen-code/qwen-code-core'; +import type { AvailableModel } from '@qwen-code/qwen-code-core'; import type { LoadedSettings } from '../../config/settings.js'; import type { RecordedVoiceAudio } from '../hooks/use-voice-input.js'; import { buildVoiceKeyterms } from './voice-keyterms.js'; @@ -57,8 +57,17 @@ export interface ResolvedVoiceStreamConfig extends VoiceStreamConfig { transport: VoiceStreamingTransport; } +/** + * Minimal structural view of the model registry the voice resolver needs. + * Both the CLI `Config` and core's `ModelsConfig` satisfy this, so the daemon + * can resolve a voice model from settings without building a full CLI `Config`. + */ +export interface VoiceModelLookup { + getAllConfiguredModels(): AvailableModel[]; +} + interface ResolveVoiceTranscriptionConfigArgs { - config: Config; + config: VoiceModelLookup; settings: LoadedSettings; voiceModel: string; } diff --git a/packages/web-shell/client/App.tsx b/packages/web-shell/client/App.tsx index eef9b936be..e577529309 100644 --- a/packages/web-shell/client/App.tsx +++ b/packages/web-shell/client/App.tsx @@ -32,6 +32,7 @@ import type { import { extractPendingPermission } from './adapters/transcriptAdapter'; import { removeInjectedFromQueue } from './midTurnDedup'; import { MessageList, type MessageListHandle } from './components/MessageList'; +import { extractVoiceModels, type VoiceModelOption } from './voice/voiceModels'; import { ChatEditor, type ComposerToolbarAction, @@ -1190,6 +1191,7 @@ export function App({ const [modelDialogMode, setModelDialogMode] = useState(null); + const [voiceModels, setVoiceModels] = useState([]); const [showApprovalModeDialog, setShowApprovalModeDialog] = useState(false); const [showResumeDialog, setShowResumeDialog] = useState(false); const [showDeleteDialog, setShowDeleteDialog] = useState(false); @@ -1717,6 +1719,12 @@ export function App({ const languageSetting = workspaceSettings.find( (setting) => setting.key === LANGUAGE_SETTING_KEY, ); + const currentVoiceModel = (() => { + const value = workspaceSettings.find( + (setting) => setting.key === 'voiceModel', + )?.values.effective; + return typeof value === 'string' && value.trim() ? value.trim() : undefined; + })(); const shellOutputMaxLines = resolveShellOutputMaxLines(workspaceSettings); const [compactMode, setCompactMode] = useState(false); const compactModeRef = useRef(compactMode); @@ -2383,6 +2391,26 @@ export function App({ ); return true; } + if (modelArg === '--voice') { + store.appendLocalUserMessage(text); + workspaceActions + .loadProviders() + .then((status) => { + setVoiceModels(extractVoiceModels(status)); + setModelDialogMode('voice'); + }) + .catch((error: unknown) => + reportError(error, t('model.setVoice')), + ); + return true; + } + if (modelArg.startsWith('--voice ')) { + if (promptBlocked) return enqueuePrompt(text, images); + sendPrompt(text, images).catch((error: unknown) => + reportError(error, 'Failed to send /model --voice'), + ); + return true; + } if (modelArg) { sessionActions .setModel(modelArg) @@ -3191,12 +3219,32 @@ export function App({ const handleFastModelSelect = useCallback( (modelId: string) => { - if (streamingState !== 'idle') return; + if (streamingState !== 'idle') { + enqueuePrompt(`/model --fast ${modelId}`); + return; + } sendPrompt(`/model --fast ${modelId}`).catch((error: unknown) => { reportError(error, 'Failed to switch fast model'); }); }, - [sendPrompt, streamingState, reportError], + [enqueuePrompt, sendPrompt, streamingState, reportError], + ); + + // Persist via the prompt channel (like `/model --fast`): the daemon's command + // processor writes `voiceModel` to settings. The `/workspace/settings` route + // is token-gated, but browser voice runs on loopback-no-token — so this is + // the path that actually works there. The daemon's /voice/stream reads it back. + const handleVoiceModelSelect = useCallback( + (modelId: string) => { + if (streamingState !== 'idle') { + enqueuePrompt(`/model --voice ${modelId}`); + return; + } + sendPrompt(`/model --voice ${modelId}`).catch((error: unknown) => { + reportError(error, t('model.setVoice')); + }); + }, + [enqueuePrompt, sendPrompt, streamingState, reportError, t], ); const commands = useMemo(() => { @@ -3330,16 +3378,24 @@ export function App({ title={ modelDialogMode === 'fast' ? t('model.setFast') - : t('model.select') + : modelDialogMode === 'voice' + ? t('model.setVoice') + : t('model.select') } size="lg" onClose={() => setModelDialogMode(null)} > { if (modelDialogMode === 'fast') { handleFastModelSelect(modelId); + } else if (modelDialogMode === 'voice') { + handleVoiceModelSelect(modelId); } else { handleModelSelect(modelId); } diff --git a/packages/web-shell/client/components/ChatEditor.tsx b/packages/web-shell/client/components/ChatEditor.tsx index 3a95b63a89..5c5a7ec530 100644 --- a/packages/web-shell/client/components/ChatEditor.tsx +++ b/packages/web-shell/client/components/ChatEditor.tsx @@ -31,6 +31,7 @@ import { getComposerTagValue, } from '../hooks/useComposerCore'; import { ModeIcon } from './ModeIcon'; +import { VoiceButton } from '../voice/VoiceButton'; import styles from './ChatEditor.module.css'; export type ComposerToolbarAction = @@ -642,11 +643,7 @@ function SlashCommandPanel({ } as CSSProperties; return createPortal( -
+
new Set(commands.map((command) => command.name.replace(/^\/+/, ''))), + () => + new Set(commands.map((command) => command.name.replace(/^\/+/, ''))), [commands], ); const hasCommand = useCallback( @@ -1517,6 +1514,15 @@ export const ChatEditor = memo( )} + { + const existing = core.getText(); + const sep = existing && !/\s$/.test(existing) ? ' ' : ''; + core.insertText(`${sep}${text} `); + core.focus(); + }} + /> + ); + } else if (isTranscribing) { + control = ( + + + ); + } else { + // idle / connecting / error / notice → icon button + const iconClass = [ + styles.iconBtn, + isError ? styles.error : '', + isConnecting ? styles.connecting : '', + ] + .filter(Boolean) + .join(' '); + control = ( + + ); + } + + const showInterim = (isRecording && interimText) || isError || isNotice; + + return ( + + {control} + {showInterim && ( + + {isError + ? errorMessage || 'Voice error' + : isNotice + ? noticeMessage + : interimText} + + )} + + ); +} diff --git a/packages/web-shell/client/voice/useVoiceCapture.test.tsx b/packages/web-shell/client/voice/useVoiceCapture.test.tsx new file mode 100644 index 0000000000..b3ac96adaa --- /dev/null +++ b/packages/web-shell/client/voice/useVoiceCapture.test.tsx @@ -0,0 +1,342 @@ +/** + * @license + * Copyright 2025 Qwen + * SPDX-License-Identifier: Apache-2.0 + */ + +// @vitest-environment jsdom + +import * as React from 'react'; +import { act } from 'react'; +import { createRoot, type Root } from 'react-dom/client'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { useVoiceCapture, type UseVoiceCaptureReturn } from './useVoiceCapture'; + +( + globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean } +).IS_REACT_ACT_ENVIRONMENT = true; + +class MockWebSocket { + static readonly OPEN = 1; + static latest: MockWebSocket | undefined; + + readonly OPEN = 1; + readyState = MockWebSocket.OPEN; + onopen: (() => void) | null = null; + onmessage: ((event: MessageEvent) => void) | null = null; + onerror: (() => void) | null = null; + onclose: ((event: CloseEvent) => void) | null = null; + readonly sent: unknown[] = []; + + constructor( + readonly url: string, + readonly protocols?: string | string[], + ) { + MockWebSocket.latest = this; + } + + send(data: unknown): void { + this.sent.push(data); + } + + close(): void { + this.readyState = 3; + } +} + +function node() { + return { connect: vi.fn(), disconnect: vi.fn() }; +} + +class MockAudioContext { + state = 'running'; + sampleRate = 16_000; + readonly destination = {}; + createMediaStreamSource = vi.fn(() => node()); + createScriptProcessor = vi.fn(() => ({ ...node(), onaudioprocess: null })); + createGain = vi.fn(() => ({ ...node(), gain: { value: 1 } })); + resume = vi.fn(async () => {}); + close = vi.fn(async () => { + this.state = 'closed'; + }); +} + +let root: Root | null = null; +let container: HTMLDivElement | null = null; +let capture: UseVoiceCaptureReturn | undefined; +const onFinal = vi.fn(); +const onError = vi.fn(); +const track = { stop: vi.fn() }; +let baseUrl = 'http://127.0.0.1:1234'; +let token: string | undefined; + +/** Decode a `qwen-bearer.` subprotocol back to the raw token. */ +function decodeBearerSubprotocol(proto: string): string { + const b64 = proto + .slice('qwen-bearer.'.length) + .replace(/-/g, '+') + .replace(/_/g, '/'); + const binary = atob(b64); + const bytes = Uint8Array.from(binary, (c) => c.charCodeAt(0)); + return new TextDecoder().decode(bytes); +} + +function TestHost() { + capture = useVoiceCapture({ + baseUrl, + token, + onFinal, + onError, + }); + return null; +} + +async function renderHookHost() { + container = document.createElement('div'); + document.body.appendChild(container); + root = createRoot(container); + await act(async () => { + root?.render(React.createElement(TestHost)); + }); + if (!capture) throw new Error('hook did not render'); + return capture; +} + +beforeEach(() => { + capture = undefined; + onFinal.mockReset(); + onError.mockReset(); + track.stop.mockReset(); + baseUrl = 'http://127.0.0.1:1234'; + token = undefined; + MockWebSocket.latest = undefined; + Object.defineProperty(globalThis, 'WebSocket', { + value: MockWebSocket, + configurable: true, + }); + Object.defineProperty(window, 'AudioContext', { + value: MockAudioContext, + configurable: true, + }); + Object.defineProperty(navigator, 'mediaDevices', { + value: { + getUserMedia: vi.fn(async () => ({ + getTracks: () => [track], + })), + }, + configurable: true, + }); +}); + +afterEach(async () => { + if (root) { + await act(async () => { + root?.unmount(); + }); + root = null; + } + container?.remove(); + container = null; + vi.useRealTimers(); +}); + +describe('useVoiceCapture', () => { + it('preserves reverse-proxy base paths in the websocket URL', async () => { + baseUrl = 'https://example.test/qwen/'; + const result = await renderHookHost(); + + await act(async () => { + result.start(); + }); + + expect(MockWebSocket.latest?.url).toBe( + 'wss://example.test/qwen/voice/stream', + ); + }); + + it('carries the bearer token as a Sec-WebSocket-Protocol subprotocol', async () => { + token = 'secret-token-123'; + const result = await renderHookHost(); + + await act(async () => { + result.start(); + }); + + const protocols = MockWebSocket.latest?.protocols; + expect(Array.isArray(protocols)).toBe(true); + const list = protocols as string[]; + // Non-secret marker first (what the daemon selects), then the bearer token. + expect(list).toHaveLength(2); + expect(list[0]).toBe('qwen-ws'); + expect(list[1].startsWith('qwen-bearer.')).toBe(true); + // Round-trips back to the raw token (what the daemon decodes + hashes). + expect(decodeBearerSubprotocol(list[1])).toBe('secret-token-123'); + }); + + it('offers no subprotocol when no token is configured', async () => { + const result = await renderHookHost(); + + await act(async () => { + result.start(); + }); + + expect(MockWebSocket.latest?.protocols).toBeUndefined(); + }); + + it('uses server error frame messages', async () => { + const result = await renderHookHost(); + + await act(async () => { + result.start(); + }); + const ws = MockWebSocket.latest; + if (!ws) throw new Error('WebSocket was not created'); + + await act(async () => { + ws.onopen?.(); + ws.onmessage?.({ + data: JSON.stringify({ + type: 'error', + message: 'No voice model is configured.', + }), + } as MessageEvent); + }); + + expect(onError).toHaveBeenCalledWith('No voice model is configured.'); + expect(capture?.status).toBe('error'); + }); + + it('delivers final transcripts and returns to idle', async () => { + const result = await renderHookHost(); + + await act(async () => { + result.start(); + }); + const ws = MockWebSocket.latest; + if (!ws) throw new Error('WebSocket was not created'); + + await act(async () => { + ws.onopen?.(); + ws.onmessage?.({ + data: JSON.stringify({ + type: 'final', + text: 'hello from voice', + }), + } as MessageEvent); + }); + + expect(onFinal).toHaveBeenCalledWith('hello from voice'); + expect(onError).not.toHaveBeenCalled(); + expect(capture?.status).toBe('idle'); + }); + + it('fails instead of staying transcribing when the socket closes early', async () => { + const result = await renderHookHost(); + + await act(async () => { + result.start(); + }); + const ws = MockWebSocket.latest; + if (!ws) throw new Error('WebSocket was not created'); + + await act(async () => { + ws.onopen?.(); + }); + await act(async () => { + result.stop(); + }); + expect(capture?.status).toBe('transcribing'); + + await act(async () => { + ws.onclose?.({ code: 1006, reason: '' } as CloseEvent); + }); + + expect(onError).toHaveBeenCalledWith( + 'Voice connection closed (code=1006, reason=none).', + ); + expect(capture?.status).toBe('error'); + }); + + it('does not leak a transcription timer when stop is called twice', async () => { + vi.useFakeTimers(); + const result = await renderHookHost(); + + await act(async () => { + result.start(); + }); + const ws = MockWebSocket.latest; + if (!ws) throw new Error('WebSocket was not created'); + + await act(async () => { + ws.onopen?.(); + result.stop(); + }); + expect(vi.getTimerCount()).toBe(1); + + await act(async () => { + result.stop(); + }); + expect(vi.getTimerCount()).toBe(1); + }); + + it('fails when the server sends no response after recording starts', async () => { + vi.useFakeTimers(); + const result = await renderHookHost(); + + await act(async () => { + result.start(); + }); + const ws = MockWebSocket.latest; + if (!ws) throw new Error('WebSocket was not created'); + + await act(async () => { + ws.onopen?.(); + }); + expect(capture?.status).toBe('recording'); + + await act(async () => { + await vi.advanceTimersByTimeAsync(60_000); + }); + + expect(onError).toHaveBeenCalledWith( + 'No response from server. Check that the voice model is running.', + ); + expect(capture?.status).toBe('error'); + }); + + it('ignores stale socket callbacks after a new capture starts', async () => { + const result = await renderHookHost(); + + await act(async () => { + result.start(); + }); + const firstWs = MockWebSocket.latest; + if (!firstWs?.onmessage) throw new Error('first WebSocket was not ready'); + const staleMessage = firstWs.onmessage; + + await act(async () => { + staleMessage({ + data: JSON.stringify({ type: 'error', message: 'first failed' }), + } as MessageEvent); + }); + expect(capture?.status).toBe('error'); + + await act(async () => { + capture?.start(); + }); + const secondWs = MockWebSocket.latest; + if (!secondWs || secondWs === firstWs) { + throw new Error('second WebSocket was not created'); + } + + await act(async () => { + staleMessage({ + data: JSON.stringify({ type: 'final', text: 'stale transcript' }), + } as MessageEvent); + }); + + expect(capture?.status).toBe('connecting'); + expect(secondWs.readyState).toBe(MockWebSocket.OPEN); + expect(onFinal).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/web-shell/client/voice/useVoiceCapture.ts b/packages/web-shell/client/voice/useVoiceCapture.ts new file mode 100644 index 0000000000..38e85419d8 --- /dev/null +++ b/packages/web-shell/client/voice/useVoiceCapture.ts @@ -0,0 +1,485 @@ +/** + * @license + * Copyright 2025 Qwen + * SPDX-License-Identifier: Apache-2.0 + */ + +import { useCallback, useEffect, useRef, useState } from 'react'; + +/** + * Browser-side voice capture for the Web Shell. Captures the microphone via + * `getUserMedia`, downsamples to 16 kHz mono s16le PCM in an AudioWorklet, and + * streams the raw frames to the daemon's `/voice/stream` WebSocket. The daemon + * transcribes server-side (credentials never reach the browser) and returns + * interim/final transcripts. + * + * Note: browsers cannot set an `Authorization` header on a WebSocket. When a + * bearer token is configured it rides in the `Sec-WebSocket-Protocol` + * subprotocol as `qwen-bearer.` (see `bearerSubprotocol`), + * which the daemon's ACP upgrade listener verifies — so this works against both + * no-token loopback and token-required deployments. + */ +export type VoiceCaptureStatus = + | 'idle' + | 'connecting' + | 'recording' + | 'transcribing' + | 'error'; + +export interface UseVoiceCaptureOptions { + baseUrl: string; + token?: string; + /** Called with the final transcript (may be empty). */ + onFinal: (text: string) => void; + onError?: (message: string) => void; +} + +export interface UseVoiceCaptureReturn { + status: VoiceCaptureStatus; + interimText: string; + /** Recent input level, 0..1, for a live meter. */ + audioLevel: number; + errorMessage: string | undefined; + start: () => void; + stop: () => void; + abort: () => void; +} + +const SAMPLE_RATE = 16_000; +const FRAME_SIZE = 4096; +const TRANSCRIPTION_TIMEOUT_MS = 60_000; + +function toWebSocketUrl(baseUrl: string): string { + const base = new URL(baseUrl); + const basePath = base.pathname.replace(/\/?$/, '/'); + const url = new URL('voice/stream', `${base.origin}${basePath}`); + url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:'; + return url.toString(); +} + +/** + * Browsers can't set an `Authorization` header on a WebSocket, so the bearer + * token rides in `Sec-WebSocket-Protocol` as `qwen-bearer.`. + * The daemon's ACP upgrade listener decodes it (serve/acp-http/index.ts) — keep + * this prefix in sync with `WS_BEARER_SUBPROTOCOL_PREFIX` there. + */ +const WS_BEARER_SUBPROTOCOL_PREFIX = 'qwen-bearer.'; +// Non-secret marker offered alongside the bearer subprotocol. The daemon +// completes the handshake by selecting THIS (never echoing the secret), which +// also satisfies WS clients that require the server to pick an offered +// subprotocol when any were requested. Must not start with the bearer prefix. +const WS_AUTH_SUBPROTOCOL = 'qwen-ws'; + +function bearerSubprotocol(token: string): string { + const bytes = new TextEncoder().encode(token); + let binary = ''; + for (let i = 0; i < bytes.length; i++) { + binary += String.fromCharCode(bytes[i]); + } + const b64 = btoa(binary) + .replace(/\+/g, '-') + .replace(/\//g, '_') + .replace(/=+$/, ''); + return `${WS_BEARER_SUBPROTOCOL_PREFIX}${b64}`; +} + +/** Turn a getUserMedia rejection into an actionable, human message. */ +function describeMicError(err: unknown): string { + const name = (err as { name?: string } | undefined)?.name; + switch (name) { + case 'NotAllowedError': + case 'SecurityError': + return 'Microphone blocked. Click the camera/lock icon in the address bar to allow the mic for this site, and enable your browser under System Settings → Privacy → Microphone, then retry.'; + case 'NotFoundError': + case 'DevicesNotFoundError': + case 'OverconstrainedError': + return 'No microphone found. Connect one and retry.'; + case 'NotReadableError': + case 'TrackStartError': + return 'Microphone is in use by another app. Close it and retry.'; + default: + return err instanceof Error ? err.message : String(err); + } +} + +/** Float32 [-1,1] frame → Int16 PCM + RMS level. */ +function floatToPcm16(input: Float32Array): { + pcm: ArrayBuffer; + level: number; +} { + const pcm = new Int16Array(input.length); + let sumSquares = 0; + for (let i = 0; i < input.length; i++) { + let s = input[i]; + if (s > 1) s = 1; + else if (s < -1) s = -1; + pcm[i] = s < 0 ? s * 0x8000 : s * 0x7fff; + sumSquares += s * s; + } + return { + pcm: pcm.buffer, + level: input.length ? Math.sqrt(sumSquares / input.length) : 0, + }; +} + +interface CaptureResources { + ws?: WebSocket; + stream?: MediaStream; + context?: AudioContext; + source?: MediaStreamAudioSourceNode; + // ScriptProcessorNode (not AudioWorklet): the Web Shell CSP `script-src` + // omits `blob:`, which blocks a Blob-URL worklet module. ScriptProcessor + // needs no module load, so it sidesteps CSP entirely. + processor?: ScriptProcessorNode; + sink?: GainNode; + transcribeTimeout?: ReturnType; +} + +export function useVoiceCapture( + options: UseVoiceCaptureOptions, +): UseVoiceCaptureReturn { + const { baseUrl, token, onFinal, onError } = options; + + const [status, setStatus] = useState('idle'); + const [interimText, setInterimText] = useState(''); + const [audioLevel, setAudioLevel] = useState(0); + const [errorMessage, setErrorMessage] = useState( + undefined, + ); + + const resourcesRef = useRef({}); + const mountedRef = useRef(true); + const captureGenerationRef = useRef(0); + // Live status for async WS/worklet callbacks, which would otherwise read a + // stale closure copy of `status`. + const statusRef = useRef('idle'); + // Latest callbacks without re-binding the capture lifecycle. + const onFinalRef = useRef(onFinal); + onFinalRef.current = onFinal; + const onErrorRef = useRef(onError); + onErrorRef.current = onError; + + const applyStatus = useCallback((next: VoiceCaptureStatus) => { + statusRef.current = next; + setStatus(next); + }, []); + + const clearTranscribeTimeout = useCallback(() => { + const res = resourcesRef.current; + if (res.transcribeTimeout) { + clearTimeout(res.transcribeTimeout); + res.transcribeTimeout = undefined; + } + }, []); + + const teardownAudio = useCallback(() => { + const res = resourcesRef.current; + if (res.processor) res.processor.onaudioprocess = null; + for (const node of [res.processor, res.source, res.sink]) { + try { + node?.disconnect(); + } catch { + /* ignore */ + } + } + res.stream?.getTracks().forEach((track) => track.stop()); + if (res.context && res.context.state !== 'closed') { + void res.context.close().catch(() => {}); + } + res.processor = undefined; + res.sink = undefined; + res.source = undefined; + res.stream = undefined; + res.context = undefined; + }, []); + + const cleanup = useCallback(() => { + captureGenerationRef.current++; + teardownAudio(); + const res = resourcesRef.current; + clearTranscribeTimeout(); + if (res.ws) { + try { + res.ws.onmessage = null; + res.ws.onerror = null; + res.ws.onclose = null; + res.ws.close(); + } catch { + /* ignore */ + } + res.ws = undefined; + } + }, [teardownAudio, clearTranscribeTimeout]); + + const fail = useCallback( + (message: string, generation?: number) => { + if ( + !mountedRef.current || + (generation !== undefined && + captureGenerationRef.current !== generation) + ) { + return; + } + cleanup(); + applyStatus('error'); + setInterimText(''); + setAudioLevel(0); + setErrorMessage(message); + onErrorRef.current?.(message); + }, + [cleanup, applyStatus], + ); + + const finishWith = useCallback( + (text: string, generation?: number) => { + if ( + generation !== undefined && + captureGenerationRef.current !== generation + ) { + return; + } + cleanup(); + if (!mountedRef.current) return; + applyStatus('idle'); + setInterimText(''); + setAudioLevel(0); + onFinalRef.current(text); + }, + [cleanup, applyStatus], + ); + + const start = useCallback(() => { + if (statusRef.current !== 'idle' && statusRef.current !== 'error') return; + setErrorMessage(undefined); + setInterimText(''); + applyStatus('connecting'); + const generation = ++captureGenerationRef.current; + const isStale = () => + !mountedRef.current || captureGenerationRef.current !== generation; + + void (async () => { + try { + if (!navigator.mediaDevices?.getUserMedia) { + throw new Error( + window.isSecureContext + ? 'Microphone capture is not supported in this browser.' + : 'Microphone needs a secure context — open the Web Shell via localhost/127.0.0.1 or https.', + ); + } + let stream: MediaStream; + try { + stream = await navigator.mediaDevices.getUserMedia({ + audio: { + channelCount: 1, + echoCancellation: true, + noiseSuppression: true, + }, + }); + } catch (err) { + throw new Error(describeMicError(err)); + } + if (isStale()) { + stream.getTracks().forEach((track) => track.stop()); + return; + } + resourcesRef.current.stream = stream; + + const context = new AudioContext({ sampleRate: SAMPLE_RATE }); + if (context.sampleRate !== SAMPLE_RATE) { + stream.getTracks().forEach((track) => track.stop()); + void context.close().catch(() => {}); + throw new Error( + `Browser audio rate ${context.sampleRate} Hz is not the required ${SAMPLE_RATE} Hz.`, + ); + } + resourcesRef.current.context = context; + // Resume in case the browser created it suspended (pre-gesture). + if (context.state === 'suspended') await context.resume(); + if (isStale()) { + stream.getTracks().forEach((track) => track.stop()); + void context.close().catch(() => {}); + return; + } + + const source = context.createMediaStreamSource(stream); + const processor = context.createScriptProcessor(FRAME_SIZE, 1, 1); + // Silent sink: a ScriptProcessorNode only fires `onaudioprocess` while + // connected to the destination; gain 0 avoids routing mic to speakers. + const sink = context.createGain(); + sink.gain.value = 0; + resourcesRef.current.source = source; + resourcesRef.current.processor = processor; + resourcesRef.current.sink = sink; + + const ws = new WebSocket( + toWebSocketUrl(baseUrl), + token ? [WS_AUTH_SUBPROTOCOL, bearerSubprotocol(token)] : undefined, + ); + ws.binaryType = 'arraybuffer'; + resourcesRef.current.ws = ws; + + processor.onaudioprocess = (event: AudioProcessingEvent) => { + const { pcm, level } = floatToPcm16( + event.inputBuffer.getChannelData(0), + ); + if (mountedRef.current) setAudioLevel(level); + if (ws.readyState === WebSocket.OPEN) ws.send(pcm); + }; + + ws.onopen = () => { + if (isStale()) { + processor.onaudioprocess = null; + for (const node of [processor, source, sink]) { + try { + node.disconnect(); + } catch { + /* ignore */ + } + } + stream.getTracks().forEach((track) => track.stop()); + if (context.state !== 'closed') { + void context.close().catch(() => {}); + } + const res = resourcesRef.current; + if (res.ws === ws) res.ws = undefined; + if (res.processor === processor) res.processor = undefined; + if (res.source === source) res.source = undefined; + if (res.sink === sink) res.sink = undefined; + if (res.stream === stream) res.stream = undefined; + if (res.context === context) res.context = undefined; + try { + ws.close(); + } catch { + /* ignore */ + } + return; + } + ws.send(JSON.stringify({ type: 'start' })); + source.connect(processor); + processor.connect(sink); + sink.connect(context.destination); + clearTranscribeTimeout(); + resourcesRef.current.transcribeTimeout = setTimeout(() => { + if (statusRef.current === 'recording') { + fail( + 'No response from server. Check that the voice model is running.', + generation, + ); + } + }, TRANSCRIPTION_TIMEOUT_MS); + if (mountedRef.current) applyStatus('recording'); + }; + + ws.onmessage = (event: MessageEvent) => { + if ( + statusRef.current === 'connecting' || + statusRef.current === 'recording' + ) { + clearTranscribeTimeout(); + } + let msg: { type?: string; text?: string; message?: string }; + try { + msg = JSON.parse(String(event.data)); + } catch { + return; + } + if (msg.type === 'interim') { + if (mountedRef.current) setInterimText(msg.text ?? ''); + } else if (msg.type === 'final') { + finishWith(msg.text ?? '', generation); + } else if (msg.type === 'error') { + fail( + msg.message ?? msg.text ?? 'Voice transcription failed.', + generation, + ); + } + }; + + ws.onerror = () => { + // The following close event carries the useful code/reason. + }; + ws.onclose = (event) => { + // A close before a final result (and not during normal teardown) + // surfaces as an error so the user isn't left stuck. + if ( + mountedRef.current && + (statusRef.current === 'recording' || + statusRef.current === 'connecting' || + statusRef.current === 'transcribing') + ) { + const code = event.code || 1006; + const reason = event.reason || 'none'; + fail( + `Voice connection closed (code=${code}, reason=${reason}).`, + generation, + ); + } + }; + } catch (error) { + fail( + error instanceof Error ? error.message : String(error), + generation, + ); + } + })(); + }, [baseUrl, token, fail, finishWith, applyStatus, clearTranscribeTimeout]); + + const stop = useCallback(() => { + const ws = resourcesRef.current.ws; + if (!ws || ws.readyState !== WebSocket.OPEN) { + cleanup(); + applyStatus('idle'); + return; + } + // Stop feeding audio, then ask the daemon to finalize. The 'final' frame + // resolves the transcript; teardownAudio releases the mic immediately. + teardownAudio(); + setAudioLevel(0); + applyStatus('transcribing'); + const generation = captureGenerationRef.current; + try { + ws.send(JSON.stringify({ type: 'stop' })); + clearTranscribeTimeout(); + resourcesRef.current.transcribeTimeout = setTimeout(() => { + if (statusRef.current === 'transcribing') { + fail('Transcription timed out.', generation); + } + }, TRANSCRIPTION_TIMEOUT_MS); + } catch { + fail('Failed to finalize voice transcription.', generation); + } + }, [cleanup, teardownAudio, fail, applyStatus, clearTranscribeTimeout]); + + const abort = useCallback(() => { + const ws = resourcesRef.current.ws; + if (ws && ws.readyState === WebSocket.OPEN) { + try { + ws.send(JSON.stringify({ type: 'abort' })); + } catch { + /* ignore */ + } + } + cleanup(); + applyStatus('idle'); + setInterimText(''); + setAudioLevel(0); + }, [cleanup, applyStatus]); + + useEffect(() => { + mountedRef.current = true; + return () => { + mountedRef.current = false; + cleanup(); + }; + }, [cleanup]); + + return { + status, + interimText, + audioLevel, + errorMessage, + start, + stop, + abort, + }; +} diff --git a/packages/web-shell/client/voice/voiceModels.ts b/packages/web-shell/client/voice/voiceModels.ts new file mode 100644 index 0000000000..7d84b2cd28 --- /dev/null +++ b/packages/web-shell/client/voice/voiceModels.ts @@ -0,0 +1,76 @@ +/** + * @license + * Copyright 2025 Qwen + * SPDX-License-Identifier: Apache-2.0 + */ + +/** A voice-transcription model option for the `/model --voice` picker. */ +export interface VoiceModelOption { + /** Raw model id (no auth suffix) — what gets persisted as `voiceModel`. */ + id: string; + label?: string; + authType?: string; + baseUrl?: string; + contextWindow?: number; + modalities?: { audio?: boolean }; +} + +/** + * Mirror of the CLI's `resolveVoiceTransport` id patterns (voice-model.ts): true + * for ids the daemon has an ASR transport for. Kept in sync by hand because the + * Web Shell can't import the CLI's voice modules. + */ +export function isVoiceModelId(id: string): boolean { + const s = id.toLowerCase(); + return ( + /^qwen3-asr-flash-realtime(?:-|$)/.test(s) || + /^qwen3-asr-flash(?:-\d{4}-\d{2}-\d{2})?$/.test(s) || + /^(fun-asr|paraformer).*realtime(?:-|$)/.test(s) + ); +} + +interface ProvidersStatusLike { + providers?: Array<{ + authType?: string; + models?: Array<{ + baseModelId?: string; + modelId?: string; + name?: string; + baseUrl?: string; + contextLimit?: number; + isRuntime?: boolean; + }>; + }>; +} + +/** + * Extract selectable voice models from a `/workspace/providers` status. Voice + * models are hidden from the session's main model list (`voiceOnly`), so the + * picker sources them from the providers surface instead. + */ +export function extractVoiceModels( + status: ProvidersStatusLike | undefined, +): VoiceModelOption[] { + const seen = new Set(); + const out: VoiceModelOption[] = []; + for (const provider of status?.providers ?? []) { + for (const model of provider.models ?? []) { + const id = model.baseModelId; + if (!id || model.isRuntime || !isVoiceModelId(id) || seen.has(id)) { + continue; + } + seen.add(id); + out.push({ + id, + ...(model.name ? { label: model.name } : {}), + ...(provider.authType ? { authType: provider.authType } : {}), + ...(model.baseUrl ? { baseUrl: model.baseUrl } : {}), + ...(typeof model.contextLimit === 'number' + ? { contextWindow: model.contextLimit } + : {}), + modalities: { audio: true }, + }); + } + } + return out; +} diff --git a/packages/web-shell/vite.config.ts b/packages/web-shell/vite.config.ts index 675e4e91aa..47057caef6 100644 --- a/packages/web-shell/vite.config.ts +++ b/packages/web-shell/vite.config.ts @@ -73,6 +73,12 @@ export default defineConfig(({ command }) => ({ '/stat': daemonProxy, '/list': daemonProxy, '/glob': daemonProxy, + // Voice dictation is a WebSocket (`/voice/stream`); `ws: true` makes the + // dev proxy forward the HTTP upgrade to the daemon. Scope it to the exact + // path — a bare `/voice` prefix would shadow the client's own + // `client/voice/*` source modules (e.g. `/voice/voiceModels.ts`), which + // vite must serve, and blanks the page. + '/voice/stream': { ...daemonProxy, ws: true }, }, }, }));