From d401c27edf1918885b5474e51808bb3d3a9d4e88 Mon Sep 17 00:00:00 2001 From: Andrew Date: Fri, 31 Jul 2026 00:48:51 -0700 Subject: [PATCH] fix(wake): address review on client-capture re-arm and feed queue - wake.status reports effective capture from the armed detector (client vs local), plus frame_length/sample_rate; GUI status probes prefer client - Gateway test doubles accept external_audio on start_listening - Desktop PCM feeder uses a bounded ordered queue instead of dropping frames while a wake.feed RPC is in flight - /wake on and status/re-arm paths pass client_capture so remote reattach works --- .../session/hooks/use-prompt-actions/slash.ts | 7 ++- apps/desktop/src/lib/wake-client-capture.ts | 56 ++++++++++++++----- apps/desktop/src/store/wake-word.ts | 25 ++++++++- tests/test_tui_gateway_server.py | 13 ++++- tui_gateway/server.py | 26 ++++++++- 5 files changed, 104 insertions(+), 23 deletions(-) diff --git a/apps/desktop/src/app/session/hooks/use-prompt-actions/slash.ts b/apps/desktop/src/app/session/hooks/use-prompt-actions/slash.ts index 44e70c01e3e94..c9f1836e65e32 100644 --- a/apps/desktop/src/app/session/hooks/use-prompt-actions/slash.ts +++ b/apps/desktop/src/app/session/hooks/use-prompt-actions/slash.ts @@ -659,7 +659,10 @@ export function useSlashCommand(deps: SlashCommandDeps) { } const status = async (): Promise => { - const current = await requestGateway('wake.status', {}) + const current = await requestGateway('wake.status', { + client_capture: true, + surface: 'gui' + }) applyWakeStatus(current) return current @@ -677,7 +680,7 @@ export function useSlashCommand(deps: SlashCommandDeps) { if (action === 'on') { const started = await requestGateway( 'wake.start', - { persist: true, surface: 'gui' }, + { persist: true, surface: 'gui', client_capture: true }, WAKE_START_TIMEOUT_MS ) diff --git a/apps/desktop/src/lib/wake-client-capture.ts b/apps/desktop/src/lib/wake-client-capture.ts index 1bb7867bccab3..92dfd62ac54dd 100644 --- a/apps/desktop/src/lib/wake-client-capture.ts +++ b/apps/desktop/src/lib/wake-client-capture.ts @@ -108,26 +108,55 @@ export async function startClientWakeCapture( let pending = new Float32Array(0) let stopped = false - let inflight = false + // Bounded ordered queue of 16 kHz frames. We never drop the frame that is + // currently being sent; under remote latency we drop the oldest queued + // frames so the detector still sees contiguous recent PCM rather than gaps + // from fire-and-forget discard-while-inflight. + const MAX_QUEUED_FRAMES = 24 // ~1.9s at 80 ms/frame + const queue: Float32Array[] = [] + let draining = false - const pushFrame = async (frame: Float32Array) => { - if (stopped || inflight) { + const drainQueue = async () => { + if (draining) { return } - inflight = true + draining = true try { - const pcm = floatToInt16LE(frame) - await options.request('wake.feed', { - pcm: bytesToBase64(pcm), - sample_rate: TARGET_RATE - }) - } catch (error) { - options.onError?.(error instanceof Error ? error : new Error(String(error))) + while (!stopped && queue.length > 0) { + const frame = queue.shift() + if (!frame) { + break + } + try { + const pcm = floatToInt16LE(frame) + await options.request('wake.feed', { + pcm: bytesToBase64(pcm), + sample_rate: TARGET_RATE + }) + } catch (error) { + options.onError?.(error instanceof Error ? error : new Error(String(error))) + // Keep draining later frames; one failed RPC should not freeze the ear. + } + } } finally { - inflight = false + draining = false + if (!stopped && queue.length > 0) { + void drainQueue() + } } } + const enqueueFrame = (frame: Float32Array) => { + if (stopped) { + return + } + queue.push(frame) + while (queue.length > MAX_QUEUED_FRAMES) { + queue.shift() + } + void drainQueue() + } + processor.onaudioprocess = event => { if (stopped) { return @@ -142,7 +171,7 @@ export async function startClientWakeCapture( while (offset + frameLength <= merged.length) { const frame = merged.subarray(offset, offset + frameLength) offset += frameLength - void pushFrame(new Float32Array(frame)) + enqueueFrame(new Float32Array(frame)) } pending = merged.subarray(offset) } @@ -164,6 +193,7 @@ export async function startClientWakeCapture( return } stopped = true + queue.length = 0 try { processor.disconnect() source.disconnect() diff --git a/apps/desktop/src/store/wake-word.ts b/apps/desktop/src/store/wake-word.ts index 89328ecce7701..2f2fef981ee2b 100644 --- a/apps/desktop/src/store/wake-word.ts +++ b/apps/desktop/src/store/wake-word.ts @@ -89,6 +89,7 @@ export interface WakeStatusResponse { configured_surface?: string /** Config truth (wake_word.enabled) — drives post-voice re-arm. */ enabled?: boolean + frame_length?: number hint?: string input_device?: WakeInputDeviceStatus listening?: boolean @@ -97,6 +98,7 @@ export interface WakeStatusResponse { owner_surface?: string | null phrase?: string provider?: string + sample_rate?: number } export interface WakeStartResponse { @@ -249,10 +251,24 @@ export function applyWakeStopResult(result: WakeStopResponse | null | undefined) */ export async function armWakeWord(request: WakeRequester = gatewayRequester): Promise { try { - const status = await request('wake.status', {}) + const status = await request('wake.status', { + client_capture: true, + surface: 'gui' + }) applyWakeStatus(status) if (!status?.available || status.listening) { + // Armed already (e.g. another surface/restart) — reattach feeder if client. + if (status?.listening) { + const mode = (status.capture || '').toLowerCase() + if (mode === 'client' || mode === 'remote' || mode === 'external') { + void maybeStartClientCapture({ + started: true, + capture: 'client', + frame_length: status.frame_length ?? 1280 + }) + } + } return } @@ -330,7 +346,10 @@ export async function resumeWakeAfterVoice(request: WakeRequester = gatewayReque for (let attempt = 0; attempt < 3; attempt++) { try { - const status = await request('wake.status', {}) + const status = await request('wake.status', { + client_capture: true, + surface: 'gui' + }) applyWakeStatus(status) // Config says off (or the feature can't run) — off is the correct rest @@ -347,7 +366,7 @@ export async function resumeWakeAfterVoice(request: WakeRequester = gatewayReque void maybeStartClientCapture({ started: true, capture: 'client', - frame_length: 1280 + frame_length: status.frame_length ?? 1280 }) } return diff --git a/tests/test_tui_gateway_server.py b/tests/test_tui_gateway_server.py index 18dae86177d8c..e864f185ce9ee 100644 --- a/tests/test_tui_gateway_server.py +++ b/tests/test_tui_gateway_server.py @@ -1473,10 +1473,15 @@ def test_wake_owner_is_sticky_and_routes_detection_to_first_transport(monkeypatc state = {"owner": None, "callback": None, "paused": False} voice_callbacks = {} - def start_listening(callback, *, owner, config): + def start_listening(callback, *, owner, config, external_audio=False): if state["owner"] is not None and state["owner"] is not owner: raise wake_word.WakeWordInUse - state.update(owner=owner, callback=callback, paused=False) + state.update( + owner=owner, + callback=callback, + paused=False, + external_audio=bool(external_audio), + ) def pause_listening(*, owner): if state["owner"] is not owner: @@ -1670,7 +1675,9 @@ def test_wake_toggle_persists_enabled_flag_only_on_explicit_gesture(monkeypatch) listener = {"owner": None} monkeypatch.setattr( wake_word, "start_listening", - lambda callback, *, owner, config: listener.update(owner=owner), + lambda callback, *, owner, config, external_audio=False: listener.update( + owner=owner, external_audio=bool(external_audio) + ), ) monkeypatch.setattr( wake_word, "stop_listening", diff --git a/tui_gateway/server.py b/tui_gateway/server.py index bd8844a15136d..191b2d42e13f8 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -13312,14 +13312,22 @@ def _(rid, params: dict) -> dict: from tools.wake_word import ( audio_is_silent, check_wake_word_requirements, + detector_frame_info, get_input_device_status, is_listening, load_wake_word_config, owns_listener, + resolve_capture_mode, silent_audio_hint, ) cfg = load_wake_word_config() - reqs = check_wake_word_requirements(cfg) + # Prefer client when the GUI asks (desktop remote re-arm / status). + prefer_client = bool(params.get("client_capture")) or str( + params.get("surface") or "" + ).strip().lower() in ("gui", "desktop") + probe_cfg = dict(cfg) + probe_cfg["capture"] = resolve_capture_mode(cfg, prefer_client=prefer_client) + reqs = check_wake_word_requirements(probe_cfg) transport = current_transport() or _stdio_transport owner, owner_surface = _wake_owner_snapshot() owned_by_caller = owns_listener(transport) @@ -13331,7 +13339,19 @@ def _(rid, params: dict) -> dict: hint = f"Wake-word input device could not be resolved: {input_device['error']}" if silent and not hint: hint = silent_audio_hint(input_device) - capture = reqs.get("capture") or str(cfg.get("capture") or "auto") + # Effective capture: prefer the *armed* detector over config/auto. + # With capture:auto the GUI arms client mode, but a bare status probe + # would otherwise report "local" and the desktop would not reattach + # the PCM feeder after wake.detected. + frame = detector_frame_info() + if owned_by_caller and frame.get("external_audio"): + capture = "client" + elif owned_by_caller and listening: + capture = "local" + else: + capture = probe_cfg.get("capture") or reqs.get("capture") or str( + cfg.get("capture") or "auto" + ) return _ok(rid, { "listening": listening, "owned_by_caller": owned_by_caller, @@ -13349,6 +13369,8 @@ def _(rid, params: dict) -> dict: "audio_silent": silent, "capture": capture, "local_input_available": bool(reqs.get("local_input_available")), + "sample_rate": frame.get("sample_rate", 16000), + "frame_length": frame.get("frame_length", 1280), }) except Exception as e: return _err(rid, 5026, str(e))