From 7cbe86ed39fca1e34ae325c07a987c2152e54a9e Mon Sep 17 00:00:00 2001 From: alex alves Date: Tue, 3 Nov 2026 18:34:45 -0300 Subject: [PATCH] =?UTF-8?q?feat:=20migra=20cliente=20web=20para=20=C3=A1ud?= =?UTF-8?q?io=20cont=C3=ADnuo=20com=20VAD=20no=20backend?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 54 +- .../application/voice_session.py | 4 + src/voice_assistant/audio_gateway/app/main.py | 82 +- .../audio_gateway/app/protocol.py | 16 + .../audio_gateway/web/index.html | 4 +- .../audio_gateway/web/pcm-worklet.js | 19 +- src/voice_assistant/audio_gateway/web/vad.js | 92 -- .../audio_gateway/web/voice-session.js | 833 ++++------------ tests/js/pcm-worklet.test.mjs | 48 + tests/js/vad.test.mjs | 83 -- tests/js/voice-session.test.mjs | 938 +++++------------- tests/test_continuous_gateway.py | 223 +++++ tests/test_protocol.py | 2 + tests/test_web_client.py | 10 + 14 files changed, 834 insertions(+), 1574 deletions(-) delete mode 100644 src/voice_assistant/audio_gateway/web/vad.js create mode 100644 tests/js/pcm-worklet.test.mjs delete mode 100644 tests/js/vad.test.mjs create mode 100644 tests/test_continuous_gateway.py diff --git a/README.md b/README.md index a9e742c..80ab69a 100644 --- a/README.md +++ b/README.md @@ -321,17 +321,16 @@ sequenceDiagram participant C as Client participant G as Audio Gateway participant P as VoicePipeline / Providers - C->>G: session.start + C->>G: session.start (turn_detection=server) G-->>C: session.ready - C->>G: audio.start - G-->>C: turn.started - loop Audio input - C->>G: audio.chunk (turn_id, sequence_number) + loop Continuous audio input, including silence and playback + C->>G: audio.input (sequence_number) C->>G: Binary PCM frame + G-->>C: turn.started (when backend VAD detects speech) P-->>G: STT partial events G-->>C: stt.partial end - C->>G: audio.stop (turn_id) + Note over G,P: Backend VAD ends input after silence P-->>G: STT final transcript G-->>C: stt.final P-->>G: Agent response @@ -341,7 +340,7 @@ sequenceDiagram G-->>C: turn.ended ``` -STT events may arrive while input audio is still streaming. Binary audio input is preceded by an `audio.chunk` control message containing `turn_id` and `sequence_number`. The Gateway limits each binary chunk to 64 KiB. Input frames must contain complete PCM samples; rejected frames do not advance the audio sequence. `session.ready.audio` describes input and `session.ready.output_audio` describes output. The browser configures playback from the declared output format. VoxCPM reassembles HTTP fragments into complete PCM frames and rejects a truncated final sample. Its `sample_rate` option declares the expected native output rate; it does not request resampling from the external service. +STT events may arrive while input audio is still streaming. In server detection mode, each binary PCM frame is preceded by `audio.input` with a session-wide `sequence_number`, starting at zero and continuing across turns. `session.ready.max_audio_chunk_bytes` declares the accepted limit, capped at 64 KiB. The browser sizes worklet chunks to respect it. Manual mode uses `audio.chunk` with `turn_id` and a per-turn sequence. Input frames must contain complete PCM samples; rejected frames do not advance the audio sequence. `session.ready.audio` describes input and `session.ready.output_audio` describes output. The browser configures playback from the declared output format. VoxCPM reassembles HTTP fragments into complete PCM frames and rejects a truncated final sample. Its `sample_rate` option declares the expected native output rate; it does not request resampling from the external service. ### Interruption and session closure @@ -351,7 +350,7 @@ STT events may arrive while input audio is still streaming. Binary audio input i ## Lifecycle and interruption -The system separates **semantic state** from **asynchronous execution**. `TurnManager` controls listening, processing, speaking, and completion; `VoicePipeline` owns the registered turn task. The Gateway coordinates provider shutdown and WebSocket delivery. +The system separates **semantic state** from **asynchronous execution**. `TurnManager` controls listening, processing, speaking, and completion; `VoicePipeline` owns the registered turn task. VoiceSession coordinates provider shutdown; the Gateway owns WebSocket delivery and shields cleanup from transport cancellation. Cancellation is cooperative: `task.cancel()` requests termination, and the Gateway awaits task completion before confirming a successful interruption. Startup rollback attempts to cancel any created task and abort a partially initialized STT provider. Failed cleanup is reported rather than represented as a successful retry. @@ -433,8 +432,8 @@ boundary. Model padding never extends the input timeline. releases resources without emitting new speech events. Calls are serialized. These contracts do not start turns, select providers, or carry transport IDs. The Silero adapter can control turns through the continuous VoiceSession API. -Browser/WebSocket migration is the next task; the existing web client continues -to use its current turn commands until that migration. +The main web client requests server detection and streams continuous PCM through +the WebSocket route. Speech detection and turn decisions run in the backend. ### Local Silero VAD adapter @@ -497,8 +496,8 @@ remains compatible with existing configurations; explicitly creating VAD in such an environment raises a configuration error. Changing YAML values affects new adapter instances after configuration is reloaded; there is no live reload. -The VAD factory can be supplied to continuous voice sessions. The existing web -client remains on manual turn commands until the browser migration. +The main web client uses the VAD factory through continuous voice sessions. +Its detection settings come from the active environment YAML. ### Continuous audio and backend turn control @@ -541,5 +540,32 @@ turn completion describes produced output, not physical speaker completion. Manual `start_turn`/`send_audio`/`stop_turn` commands remain available on sessions without VAD. Mixing those commands with a VAD-controlled session is rejected. `create_voice_session(..., server_vad=True)` composes the continuous backend API; -the current `/ws/audio` receive loop still uses the manual mode. The next task -connects continuous input to that route and simplifies the browser accordingly. +`/ws/audio` selects this mode when `session.start.turn_detection` is `server`. +Omitting that option selects the existing manual input mode for compatible clients. + + +### Browser continuous capture + +The main `/ui` client requests `turn_detection: server` and waits for the matching +confirmation in `session.ready` before starting microphone capture. The worklet +then stays active throughout listening, processing, response playback, and +interruption. Input sequence numbers do not reset between turns. Microphone +permission, echo cancellation, and audio device ownership remain in the page. + +The browser's custom `vad.js` and its detection tests have been removed. The voice +controller now handles connection, continuous PCM transport, protocol events, +and playback. Existing visual amplitude analysis is used only for animation. +It sends no automatic `audio.start`, `audio.stop`, or `interrupt` commands. + +A new `turn.started` stops previous playback and discards obsolete queued audio. +`turn.ended` waits for scheduled speaker playback before updating the display to +listening; microphone streaming continues during that wait. Stale turn events +and completion callbacks cannot overwrite a newer turn. Outbound WebSocket +buffering and handshake waiting are bounded; failures stop local capture and +close the session. + +`audio.input.end` finishes a finite continuous stream normally and requires no +pending binary frame. It is intended for file/terminal clients; the live browser +uses `session.end` or disconnect to abort. Manual input remains available with +`turn_detection: client`, including the standalone manual `app.js` and existing +smoke scripts. The main page uses `voice-session.js`. diff --git a/src/voice_assistant/application/voice_session.py b/src/voice_assistant/application/voice_session.py index f79b08a..a726934 100644 --- a/src/voice_assistant/application/voice_session.py +++ b/src/voice_assistant/application/voice_session.py @@ -127,6 +127,10 @@ class VoiceSession: def output_audio_format(self) -> AudioFormat: return self._pipeline.tts.output_format + @property + def max_input_chunk_bytes(self) -> int: + return self._max_input_chunk_bytes + def validate_audio_input(self, turn_id: int) -> None: self._require_open() if self._audio_input_closed: diff --git a/src/voice_assistant/audio_gateway/app/main.py b/src/voice_assistant/audio_gateway/app/main.py index d486427..a042c05 100644 --- a/src/voice_assistant/audio_gateway/app/main.py +++ b/src/voice_assistant/audio_gateway/app/main.py @@ -19,7 +19,7 @@ from voice_assistant.application.voice_session import VoiceSession, VoiceSession from voice_assistant.audio.models import CANONICAL_AUDIO_FORMAT from voice_assistant.audio_gateway.app.protocol import ( PROTOCOL_VERSION, AUDIO_ENCODING, AUDIO_SAMPLE_RATE, AUDIO_CHANNELS, - AudioChunk, AudioStop, Interrupt, SessionEnd, SessionStart, + AudioChunk, InputAudioChunk, InputAudioEnd, AudioStop, Interrupt, SessionEnd, SessionStart, AgentDelta, AgentResponse, Interrupted, SessionEnded, SessionReady, STTFinal, STTPartial, TurnEnded, TurnStarted, error_event, ) @@ -139,7 +139,10 @@ async def create_voice_session( async def audio_socket(websocket: WebSocket) -> None: await websocket.accept() session: VoiceSession | None = None - pending_audio_chunk: AudioChunk | None = None + pending_audio_chunk: AudioChunk | InputAudioChunk | None = None + server_vad = False + expected_input_sequence = 0 + max_audio_chunk_bytes = MAX_AUDIO_CHUNK_BYTES async def report( code: str, message: str, *, turn_id: int | None = None, @@ -169,22 +172,27 @@ async def audio_socket(websocket: WebSocket) -> None: chunk = pending_audio_chunk pending_audio_chunk = None audio = message["bytes"] - if len(audio) > MAX_AUDIO_CHUNK_BYTES: + turn_id = chunk.turn_id if isinstance(chunk, AudioChunk) else session.active_turn_id + if len(audio) > max_audio_chunk_bytes: await report( "MESSAGE_TOO_LARGE", - f"Audio chunk exceeds maximum size of {MAX_AUDIO_CHUNK_BYTES} bytes.", - turn_id=chunk.turn_id, - details={"max_bytes": MAX_AUDIO_CHUNK_BYTES, "received_bytes": len(audio)}, + f"Audio chunk exceeds maximum size of {max_audio_chunk_bytes} bytes.", + turn_id=turn_id, + details={"max_bytes": max_audio_chunk_bytes, "received_bytes": len(audio)}, ) continue try: - await session.send_audio(chunk.turn_id, chunk.sequence_number, audio) + if isinstance(chunk, InputAudioChunk): + await session.receive_audio(audio) + expected_input_sequence += 1 + else: + await session.send_audio(chunk.turn_id, chunk.sequence_number, audio) except VoiceSessionError as exc: await report(exc.code, str(exc), turn_id=exc.turn_id, retryable=exc.retryable) except (RuntimeError, ValueError) as exc: await report( - "AUDIO_SEQUENCE_ERROR", str(exc), turn_id=chunk.turn_id, - details={"expected_sequence_number": session.expected_audio_sequence}, + "AUDIO_SEQUENCE_ERROR", str(exc), turn_id=turn_id, + details={"expected_sequence_number": expected_input_sequence if server_vad else session.expected_audio_sequence}, ) continue @@ -232,11 +240,28 @@ async def audio_socket(websocket: WebSocket) -> None: }}, ) continue - session = await create_voice_session(websocket, request.session_id or None) + try: + session = await create_voice_session( + websocket, request.session_id or None, + server_vad=request.turn_detection == "server", + ) + except Exception: + logger.exception("Failed to initialize voice session.") + await report("SESSION_START_FAILED", "Voice session could not be initialized.") + continue + server_vad = request.turn_detection == "server" + max_audio_chunk_bytes = min( + MAX_AUDIO_CHUNK_BYTES, session.max_input_chunk_bytes, + ) if server_vad else MAX_AUDIO_CHUNK_BYTES + if max_audio_chunk_bytes < 2: + await report("INVALID_CONFIGURATION", "Audio input limit must allow a complete PCM frame.") + break await send_json(websocket, SessionReady( protocol_version=PROTOCOL_VERSION, session_id=session.session_id, audio=CANONICAL_AUDIO_FORMAT, output_audio=session.output_audio_format, + turn_detection=request.turn_detection, + max_audio_chunk_bytes=max_audio_chunk_bytes, )) continue if session is None: @@ -246,6 +271,43 @@ async def audio_socket(websocket: WebSocket) -> None: await report("SESSION_ALREADY_CLOSED", "This session is already closed.") continue + if server_vad and event_type in {"audio.start", "audio.chunk", "audio.stop"}: + await report("TURN_CONTROLLED_BY_VAD", "This session uses server-side turn detection.") + continue + + if event_type == "audio.input": + if not server_vad: + await report("INVALID_INPUT_MODE", "Continuous audio requires server-side turn detection.") + continue + try: + request = InputAudioChunk.model_validate(payload) + if pending_audio_chunk is not None: + raise ValueError("Previous audio chunk is still pending PCM data.") + if request.sequence_number != expected_input_sequence: + raise ValueError(f"Expected sequence_number={expected_input_sequence}.") + pending_audio_chunk = request + except (ValidationError, ValueError) as exc: + await report( + "AUDIO_SEQUENCE_ERROR", str(exc), + details={"expected_sequence_number": expected_input_sequence}, + ) + continue + + if event_type == "audio.input.end": + if not server_vad: + await report("INVALID_INPUT_MODE", "Continuous audio requires server-side turn detection.") + continue + try: + InputAudioEnd.model_validate(payload) + if pending_audio_chunk is not None: + raise ValueError("An audio chunk is still pending PCM data.") + await session.finish_audio_input() + except VoiceSessionError as exc: + await report(exc.code, str(exc), turn_id=exc.turn_id, retryable=exc.retryable) + except (ValidationError, ValueError) as exc: + await report("INVALID_MESSAGE", str(exc)) + continue + if event_type == "audio.start": try: turn_id = await session.start_turn() diff --git a/src/voice_assistant/audio_gateway/app/protocol.py b/src/voice_assistant/audio_gateway/app/protocol.py index f537fab..21a951c 100644 --- a/src/voice_assistant/audio_gateway/app/protocol.py +++ b/src/voice_assistant/audio_gateway/app/protocol.py @@ -20,6 +20,7 @@ class SessionStart(BaseModel): type: Literal["session.start"] = "session.start" session_id: str | None = None + turn_detection: Literal["client", "server"] = "client" audio: AudioFormat = Field( default_factory=lambda: CANONICAL_AUDIO_FORMAT.model_copy() @@ -54,6 +55,19 @@ class AudioChunk(BaseModel): sequence_number: int +class InputAudioChunk(BaseModel): + """Metadata for continuous PCM, sequenced across the entire session.""" + + type: Literal["audio.input"] = "audio.input" + sequence_number: int = Field(ge=0, strict=True) + + +class InputAudioEnd(BaseModel): + """Normal end of a finite continuous input stream.""" + + type: Literal["audio.input.end"] = "audio.input.end" + + class AudioStop(BaseModel): """Request that finishes audio input for the active turn.""" @@ -82,6 +96,8 @@ class SessionReady(BaseModel): session_id: str audio: AudioFormat output_audio: AudioFormat + turn_detection: Literal["client", "server"] = "client" + max_audio_chunk_bytes: int = Field(default=65536, ge=2) class TurnStarted(BaseModel): diff --git a/src/voice_assistant/audio_gateway/web/index.html b/src/voice_assistant/audio_gateway/web/index.html index 32b9918..d484c94 100644 --- a/src/voice_assistant/audio_gateway/web/index.html +++ b/src/voice_assistant/audio_gateway/web/index.html @@ -72,7 +72,7 @@ const audioBands=new Float32Array(3); let voiceController=null; const activeStates=new Set([ - 'connecting','listening','starting','recording','flushing', + 'connecting','listening','recording', 'processing','speaking','waiting-playback' ]); @@ -107,9 +107,7 @@ const assistantIntegration={ const messages={ connecting:'Conectando ao AgentVoice...', listening:'Estou ouvindo. Pode falar.', - starting:'Iniciando reconhecimento...', recording:'Estou ouvindo sua pergunta...', - flushing:'Finalizando captura de áudio...', processing:'Preparando sua resposta...', speaking:'Reproduzindo resposta...', 'waiting-playback':'Aguardando fim da reprodução.' diff --git a/src/voice_assistant/audio_gateway/web/pcm-worklet.js b/src/voice_assistant/audio_gateway/web/pcm-worklet.js index ff83ef5..8b0f6fd 100644 --- a/src/voice_assistant/audio_gateway/web/pcm-worklet.js +++ b/src/voice_assistant/audio_gateway/web/pcm-worklet.js @@ -16,10 +16,19 @@ class PCMInputProcessor extends AudioWorkletProcessor { this.port.onmessage = ({ data }) => { switch (data?.type) { - case "start": + case "start": { + const chunkSamples = data.chunkSamples ?? 1600; + if (!Number.isInteger(chunkSamples) || chunkSamples < 1 || chunkSamples > 1600) { + this.capturing = false; + this.reset(); + this.port.postMessage({ type: "error", message: "Tamanho de captura inválido." }); + return; + } + this.chunkSamples = chunkSamples; this.reset(); this.capturing = true; break; + } case "flush": this.capturing = false; @@ -44,14 +53,16 @@ class PCMInputProcessor extends AudioWorkletProcessor { emitPCM() { if (!this.output.length) return; - const pcm = new Int16Array(this.output); + const pcm = new ArrayBuffer(this.output.length * 2); + const view = new DataView(pcm); + this.output.forEach((sample, index) => view.setInt16(index * 2, sample, true)); this.output = []; this.port.postMessage({ type: "pcm", sequence: this.sequence++, - buffer: pcm.buffer - }, [pcm.buffer]); + buffer: pcm + }, [pcm]); } flush() { diff --git a/src/voice_assistant/audio_gateway/web/vad.js b/src/voice_assistant/audio_gateway/web/vad.js deleted file mode 100644 index 05a82ff..0000000 --- a/src/voice_assistant/audio_gateway/web/vad.js +++ /dev/null @@ -1,92 +0,0 @@ - -export class VoiceActivityDetector { - constructor({ - onSpeechStart, - onSpeechEnd, - minSpeechMs = 250, - silenceMs = 800, - threshold = 0.012, - noiseMultiplier = 3, - } = {}) { - this.onSpeechStart = onSpeechStart; - this.onSpeechEnd = onSpeechEnd; - - this.minSpeechMs = minSpeechMs; - this.silenceMs = silenceMs; - this.threshold = threshold; - this.noiseMultiplier = noiseMultiplier; - - this.noiseFloor = 0.003; - this.speaking = false; - this.speechStartedAt = null; - this.lastVoiceAt = null; - this.active = false; - } - - start() { - this.active = true; - this.reset(); - } - - stop() { - this.active = false; - this.reset(); - } - - reset() { - this.speaking = false; - this.speechStartedAt = null; - this.lastVoiceAt = null; - } - - process(rms, now = performance.now()) { - if (!this.active) return; - - if (!Number.isFinite(rms) || rms < 0) return; - - const threshold = Math.max( - this.threshold, - this.noiseFloor * this.noiseMultiplier - ); - - const voiced = rms >= threshold; - - if (!this.speaking) { - // Update noise estimation only outside detected speech. - if (!voiced) { - this.noiseFloor = - this.noiseFloor * 0.98 + rms * 0.02; - } - - if (voiced) { - if (this.speechStartedAt === null) { - this.speechStartedAt = now; - } - - if (now - this.speechStartedAt >= this.minSpeechMs) { - this.speaking = true; - this.lastVoiceAt = now; - this.onSpeechStart?.(); - } - } else { - this.speechStartedAt = null; - } - - return; - } - - if (voiced) { - this.lastVoiceAt = now; - return; - } - - if ( - this.lastVoiceAt !== null && - now - this.lastVoiceAt >= this.silenceMs - ) { - this.speaking = false; - this.onSpeechEnd?.(); - this.reset(); - } - } -} diff --git a/src/voice_assistant/audio_gateway/web/voice-session.js b/src/voice_assistant/audio_gateway/web/voice-session.js index a08bd03..85d7095 100644 --- a/src/voice_assistant/audio_gateway/web/voice-session.js +++ b/src/voice_assistant/audio_gateway/web/voice-session.js @@ -1,5 +1,3 @@ - -import { VoiceActivityDetector } from "./vad.js"; import { PCMPlayer } from "./pcm-player.js"; export class VoiceSessionController { @@ -8,78 +6,45 @@ export class VoiceSessionController { onTranscript = () => {}, onAgentText = () => {}, onError = () => {}, - silenceMs = 800, - prebufferChunks = 4, - maxPendingChunks = 100, + maxBufferedBytes = 256 * 1024, + handshakeTimeoutMs = 10000, } = {}) { + if (!Number.isSafeInteger(maxBufferedBytes) || maxBufferedBytes < 2) { + throw new Error("Limite de envio de áudio inválido."); + } + if (!Number.isFinite(handshakeTimeoutMs) || handshakeTimeoutMs <= 0) { + throw new Error("Prazo de conexão inválido."); + } this.onStateChange = onStateChange; this.onTranscript = onTranscript; this.onAgentText = onAgentText; this.onError = onError; - - this.prebufferChunks = prebufferChunks; - this.maxPendingChunks = maxPendingChunks; - + this.maxBufferedBytes = maxBufferedBytes; + this.handshakeTimeoutMs = handshakeTimeoutMs; this.state = "closed"; this.ws = null; - this.stream = null; this.context = null; - this.analyser = null; this.worklet = null; this.source = null; this.silentGain = null; - - this.turnId = null; - this.sequence = 0; - this.prebuffer = []; - this.pending = []; - this.stopRequested = false; - this.frameId = 0; - this.rmsSamples = null; - - this.vad = new VoiceActivityDetector({ - silenceMs, - onSpeechStart: () => this.beginTurn(), - onSpeechEnd: () => this.finishSpeech(), - }); - this.player = null; - this.playbackPending = Promise.resolve(); + this.ready = false; + this.sequence = 0; + this.turnId = null; + this.latestTurnId = 0; this.gatewayTurnEnded = false; this.playbackGeneration = 0; - this.bargeInPending = false; - this.bargeInBuffer = []; - this.maxBargeInChunks = 100; - this.bargeInSpeechEnded = false; - this.interruptedTurnId = null; - this.bargeInCaptureStarted = false; - - this.bargeInVad = new VoiceActivityDetector({ - minSpeechMs: 350, - silenceMs: 800, - threshold: 0.025, - noiseMultiplier: 4, - - onSpeechStart: () => this.requestBargeIn(), - onSpeechEnd: () => this.finishBargeInSpeech(), - }); + this.connectionGeneration = 0; + this.playbackPending = Promise.resolve(); + this.handshakeTimer = null; + this.maxAudioChunkBytes = 0; } getPlaybackAnalyser() { - return this.player?.getAnalyser() ?? null; + return this.player?.getAnalyser() ?? null; } setState(state) { - console.debug( - `[AgentVoice][STATE] ${this.state} -> ${state}`, - { - turnId: this.turnId, - playerFinished: this.player?.finished, - bargeInPending: this.bargeInPending, - bargeInRecording: this.bargeInRecording, - } - ); - this.state = state; this.onStateChange(state); } @@ -91,688 +56,234 @@ export class VoiceSessionController { this.ws.send(JSON.stringify(event)); } - async start({ stream, audioContext, analyser }) { - if (this.state !== "closed") { - throw new Error("Sessão já iniciada."); - } - - this.stream = stream; + async start({ stream, audioContext }) { + if (this.state !== "closed") throw new Error("Sessão já iniciada."); + const generation = ++this.connectionGeneration; this.context = audioContext; - this.analyser = analyser; - + this.ready = false; + this.sequence = 0; + this.turnId = null; + this.latestTurnId = 0; this.player = new PCMPlayer({ audioContext, - - onPlaybackStart: () => { - if ( - this.state === "processing" || - this.state === "speaking" - ) { - this.setState("speaking"); - this.bargeInVad.start(); - this.monitorVoice(); - } - }, + onPlaybackStart: () => { + if (this.ready && this.turnId !== null) this.setState("speaking"); + }, }); - this.rmsSamples = new Float32Array(analyser.fftSize); - this.setState("connecting"); - try { - await this.context.audioWorklet.addModule( - "/web/pcm-worklet.js" - ); - - this.source = this.context.createMediaStreamSource(stream); - this.worklet = new AudioWorkletNode( - this.context, - "pcm-input" - ); - - this.silentGain = this.context.createGain(); + await audioContext.audioWorklet.addModule("/web/pcm-worklet.js"); + if (generation !== this.connectionGeneration) return; + this.source = audioContext.createMediaStreamSource(stream); + this.worklet = new AudioWorkletNode(audioContext, "pcm-input"); + this.silentGain = audioContext.createGain(); this.silentGain.gain.value = 0; - this.source.connect(this.worklet); this.worklet.connect(this.silentGain); - this.silentGain.connect(this.context.destination); - + this.silentGain.connect(audioContext.destination); this.worklet.port.onmessage = ({ data }) => { - this.handleWorkletMessage(data); + if (generation !== this.connectionGeneration) return; + if (data?.type === "pcm") this.handlePCM(data.buffer); + if (data?.type === "error") this.fail(new Error(data.message)); }; - const scheme = location.protocol === "https:" ? "wss" : "ws"; - const socket = new WebSocket( - `${scheme}://${location.host}/ws/audio` - ); - + const socket = new WebSocket(`${scheme}://${location.host}/ws/audio`); this.ws = socket; socket.binaryType = "arraybuffer"; - socket.onopen = () => { if (this.ws !== socket) return; - - this.send({ - type: "session.start", - audio: { - encoding: "pcm_s16le", - sample_rate: 16000, - channels: 1, - }, - }); + try { + this.send({ + type: "session.start", + turn_detection: "server", + audio: { encoding: "pcm_s16le", sample_rate: 16000, channels: 1 }, + }); + } catch (error) { + this.fail(error); + } }; - socket.onmessage = ({ data }) => { if (this.ws !== socket) return; - - if (typeof data === "string") { - try { - this.handleGatewayEvent(JSON.parse(data)); - } catch (error) { - this.fail(error); - } - } else if (data instanceof ArrayBuffer) { - this.handleAudioChunk(data); + try { + if (typeof data === "string") this.handleGatewayEvent(JSON.parse(data)); + else if (data instanceof ArrayBuffer) this.handleAudioChunk(data); + else throw new Error("Resposta de áudio inválida."); + } catch (error) { + this.fail(error); } - }; - + }; socket.onerror = () => { - if (this.ws === socket) { - this.fail(new Error("Falha na conexão WebSocket.")); - } + if (this.ws === socket) this.fail(new Error("Falha na conexão WebSocket.")); }; - socket.onclose = () => { - if (this.ws === socket && this.state !== "closed") { - this.fail(new Error("Conexão WebSocket encerrada.")); - } + if (this.ws === socket) this.fail(new Error("Conexão WebSocket encerrada.")); }; + this.handshakeTimer = setTimeout(() => { + if (this.ws === socket && !this.ready) { + this.fail(new Error("O servidor não confirmou o início da conversa.")); + } + }, this.handshakeTimeoutMs); } catch (error) { + if (generation !== this.connectionGeneration) return; this.close(); throw error; } } - - startListening() { - this.playbackGeneration++; - - this.gatewayTurnEnded = false; - this.player?.reset(); - this.playbackPending = Promise.resolve(); - - this.bargeInVad.stop(); - this.bargeInPending = false; - this.bargeInSpeechEnded = false; - this.interruptedTurnId = null; - this.bargeInBuffer = []; - - this.turnId = null; - this.sequence = 0; - this.prebuffer = []; - this.pending = []; - this.stopRequested = false; - - this.worklet.port.postMessage({ type: "start" }); - - this.vad.start(); - this.setState("listening"); - this.monitorVoice(); - } - - - monitorVoice() { - cancelAnimationFrame(this.frameId); - - const tick = (now) => { - const normalListening = [ - "listening", - "starting", - "recording" - ].includes(this.state); - - const bargeInListening = [ - "processing", - "speaking", - "waiting-playback", - "interrupting" - ].includes(this.state) || this.bargeInRecording; - - if (!normalListening && !bargeInListening) { - return; - } - - this.analyser.getFloatTimeDomainData(this.rmsSamples); - - let power = 0; - - for (const sample of this.rmsSamples) { - power += sample * sample; - } - - const rms = Math.sqrt( - power / this.rmsSamples.length - ); - - // Normal speech detection. - if ( - normalListening && - !this.bargeInRecording && - !this.stopRequested - ) { - this.vad.process(rms, now); - } - - // Continue tracking speech initiated during TTS. - if ( - bargeInListening && - this.bargeInVad.active - ) { - this.bargeInVad.process(rms, now); - } - - this.frameId = requestAnimationFrame(tick); - }; - - this.frameId = requestAnimationFrame(tick); - } - - beginTurn() { - if (this.state !== "listening") return; - - this.pending = [...this.prebuffer]; - this.setState("starting"); - - try { - this.send({ type: "audio.start" }); - } catch (error) { - this.fail(error); - } - } - - handleWorkletMessage(data) { - if (data?.type === "pcm") { - this.handlePCM(data.buffer); - } else if (data?.type === "flushed") { - this.handleFlushed(); - } - } - handlePCM(buffer) { - if (this.state === "listening") { - this.prebuffer.push(buffer); - - if (this.prebuffer.length > this.prebufferChunks) { - this.prebuffer.shift(); - } - return; - } - - if ( - this.state === "processing" || - this.state === "speaking" || - this.state === "waiting-playback" || - this.state === "interrupting" - ) { - this.bargeInBuffer.push(buffer); - - if (this.bargeInBuffer.length > this.maxBargeInChunks) { - this.bargeInBuffer.shift(); - } - - return; - } - - if (this.state === "starting") { - this.pending.push(buffer); - - if (this.pending.length > this.maxPendingChunks) { - this.fail(new Error("Buffer de início de turno excedido.")); - } - return; - } - - if (this.state === "recording" || - this.state === "flushing") { - this.sendPCM(buffer); - } - } - - sendPCM(buffer) { - if (this.turnId === null) return; - - if (this.ws?.readyState !== WebSocket.OPEN) return; - - this.send({ - type: "audio.chunk", - turn_id: this.turnId, - sequence_number: this.sequence++, - }); - - this.ws.send(buffer); - } - - finishSpeech() { - if (this.stopRequested) return; - - if (this.state !== "starting" && - this.state !== "recording") return; - - this.stopRequested = true; - this.vad.stop(); - - if (this.state === "recording") { - this.flushTurn(); - } - // If still starting, flush after turn.started. - } - - flushTurn() { - if (this.state !== "recording") return; - - this.setState("flushing"); - this.worklet.port.postMessage({ type: "flush" }); - } - - handleFlushed() { - if (this.state !== "flushing") return; - - this.send({ - type: "audio.stop", - turn_id: this.turnId, - }); - - this.setState("processing"); - - // Resume local PCM capture for a possible interruption. - // Audio will remain buffered until a new turn is accepted. - this.worklet.port.postMessage({ type: "start" }); - - // The microphone remains available for barge-in. - // Detection starts when TTS playback begins. - this.monitorVoice(); - } - - - handleAudioChunk(buffer) { - if ( - this.state === "closed" || - this.state === "interrupting" || - this.bargeInPending || - this.turnId === null || - this.gatewayTurnEnded - ) { - return; - } - - console.debug("[AgentVoice][PCM recebido]", { - bytes: buffer.byteLength, - turnId: this.turnId, - state: this.state, - generation: this.playbackGeneration, - finished: this.player?.finished, - gatewayTurnEnded: this.gatewayTurnEnded, - }); - - const generation = this.playbackGeneration; - - this.playbackPending = this.playbackPending - .then(async () => { - if (generation !== this.playbackGeneration) { - return; - } - - if ( - this.state === "interrupting" || - this.bargeInPending - ) { - return; - } - - console.debug("[AgentVoice][PCM enqueue]", { - turnId: this.turnId, - state: this.state, - generation, - currentGeneration: this.playbackGeneration, - finished: this.player?.finished, - }); - - - await this.player.enqueue(buffer); - }) - .catch((error) => { - // An obsolete playback must never terminate the session. - if (generation !== this.playbackGeneration) { - return; - } - - if ( - this.state === "interrupting" || - this.bargeInPending - ) { - return; - } - - this.fail(error); - }); - } - - - async finishPlaybackAndResume() { - const generation = this.playbackGeneration; - + if (!this.ready || this.state === "closed") return; try { - await this.playbackPending; - - if (generation !== this.playbackGeneration) return; - if (this.state === "closed") return; - - this.player.finish(); - await this.player.waitForCompletion(); - - if (generation !== this.playbackGeneration) return; - if (this.state === "closed") return; - if (!this.gatewayTurnEnded) return; - - this.startListening(); - } catch (error) { - if (generation === this.playbackGeneration) { - this.fail(error); - } - } - } - - - requestBargeIn() { - if (this.bargeInPending) return; - - if ( - this.state !== "speaking" && - this.state !== "waiting-playback" - ) { - return; - } - - if (this.turnId === null) return; - - const interruptedTurnId = this.turnId; - - this.bargeInPending = true; - this.bargeInSpeechEnded = false; - this.bargeInCaptureStarted = true; - this.interruptedTurnId = interruptedTurnId; - - // Stop local playback immediately. - this.playbackGeneration++; - this.player?.stop(); - this.playbackPending = Promise.resolve(); - - // Prevent normal playback completion from resuming listening. - this.gatewayTurnEnded = false; - - // The normal VAD must not start a competing turn. - this.vad.stop(); - - this.setState("interrupting"); - - try { - this.send({ - type: "interrupt", - turn_id: interruptedTurnId, - }); + if (!(buffer instanceof ArrayBuffer) || !buffer.byteLength || buffer.byteLength % 2) { + throw new Error("Áudio capturado inválido."); + } + if (buffer.byteLength > this.maxAudioChunkBytes) { + throw new Error("O bloco de áudio excede o limite do servidor."); + } + if ((this.ws?.bufferedAmount ?? 0) + buffer.byteLength > this.maxBufferedBytes) { + throw new Error("A conexão não está acompanhando o envio de áudio."); + } + this.send({ type: "audio.input", sequence_number: this.sequence }); + this.ws.send(buffer); + this.sequence++; } catch (error) { this.fail(error); } } - finishBargeInSpeech() { - if (!this.bargeInPending && !this.bargeInRecording) { - return; - } + invalidatePlayback() { + this.playbackGeneration++; + this.player?.stop(); + this.playbackPending = Promise.resolve(); + this.gatewayTurnEnded = false; + } - this.bargeInSpeechEnded = true; + handleAudioChunk(buffer) { + if (!this.ready || this.turnId === null || this.gatewayTurnEnded) return; + const generation = this.playbackGeneration; + this.playbackPending = this.playbackPending.then(async () => { + if (generation !== this.playbackGeneration || !this.ready) return; + await this.player.enqueue(buffer); + }).catch((error) => { + if (generation === this.playbackGeneration && this.ready) this.fail(error); + }); + } - if ( - this.bargeInRecording && - this.state === "recording" - ) { - this.bargeInRecording = false; - this.bargeInVad.stop(); - - this.stopRequested = true; - this.flushTurn(); + async finishPlayback(generation, turnId) { + try { + await this.playbackPending; + if (generation !== this.playbackGeneration || !this.ready) return; + this.player.finish(); + await this.player.waitForCompletion(); + if (generation !== this.playbackGeneration || !this.ready || this.turnId !== turnId) return; + this.turnId = null; + this.setState("listening"); + } catch (error) { + if (generation === this.playbackGeneration && this.ready) this.fail(error); } } handleGatewayEvent(event) { - switch (event.type) { - case "session.ready": - this.player.configure(event.output_audio); - this.startListening(); - break; - - // Debug logging for all events received from the Gateway. - console.debug("[AgentVoice][Gateway]", { - type: event.type, - turnId: event.turn_id, - currentTurnId: this.turnId, - state: this.state, - playbackGeneration: this.playbackGeneration, - playerFinished: this.player?.finished, - playerPlaying: this.player?.playing, - bargeInPending: this.bargeInPending, - bargeInRecording: this.bargeInRecording, - text: event.type === "stt.final" - ? event.text - : event.type === "agent.response" - ? event.text - : undefined, - error: event.type === "error" - ? { code: event.code, message: event.message } - : undefined, - }); - // Debug logging for all events received from the Gateway. - - - case "turn.started": - if (this.state !== "starting") return; - - // Prepare playback for the newly accepted turn. - // The previous TTS may have been stopped by barge-in. - this.playbackGeneration++; - this.player?.reset(); - this.playbackPending = Promise.resolve(); - this.gatewayTurnEnded = false; - - this.turnId = event.turn_id; - this.sequence = 0; - - this.setState("recording"); - - for (const buffer of this.pending) { - this.sendPCM(buffer); - } - - this.pending = []; - - if (this.bargeInPending) { - this.bargeInPending = false; - this.bargeInCaptureStarted = false; - - if (this.bargeInSpeechEnded) { - // User finished speaking before turn.started. - this.bargeInRecording = false; - this.bargeInVad.stop(); - this.stopRequested = true; - } else { - // User is still speaking. - // Preserve the active barge-in VAD state. - this.bargeInRecording = true; - this.stopRequested = false; - } - - this.bargeInSpeechEnded = false; + if (this.state === "closed") return; + if (event.type === "session.ready") { + if (this.ready) throw new Error("Confirmação de sessão duplicada."); + if ( + event.turn_detection !== "server" || event.audio?.encoding !== "pcm_s16le" || + event.audio.sample_rate !== 16000 || event.audio.channels !== 1 || + !Number.isInteger(event.max_audio_chunk_bytes) || event.max_audio_chunk_bytes < 2 + ) { + throw new Error("O servidor não oferece suporte a esta versão do cliente."); } - - if (this.stopRequested) { - this.flushTurn(); - } - break; - + this.player.configure(event.output_audio); + this.maxAudioChunkBytes = event.max_audio_chunk_bytes; + this.ready = true; + clearTimeout(this.handshakeTimer); + this.handshakeTimer = null; + this.worklet.port.postMessage({ + type: "start", chunkSamples: Math.min(1600, Math.floor(this.maxAudioChunkBytes / 2)), + }); + this.setState("listening"); + return; + } + if (event.type === "error") { + this.fail(new Error(`${event.code}: ${event.message}`)); + return; + } + if (event.type === "session.ended") { + this.close(); + return; + } + if (!this.ready) return; + if (event.type === "turn.started") { + if (!Number.isSafeInteger(event.turn_id) || event.turn_id <= this.latestTurnId) return; + this.invalidatePlayback(); + this.player.reset(); + this.latestTurnId = event.turn_id; + this.turnId = event.turn_id; + this.setState("recording"); + return; + } + if (event.turn_id !== this.turnId) return; + switch (event.type) { case "stt.partial": case "stt.final": - if (event.turn_id === this.turnId) { - this.onTranscript(event.text, event.type === "stt.final"); - } + this.onTranscript(event.text, event.type === "stt.final"); + if (event.type === "stt.final") this.setState("processing"); break; - case "agent.delta": - if (event.turn_id === this.turnId) { - this.onAgentText(event.text, false); - } - break; - case "agent.response": - if (event.turn_id === this.turnId) { - this.onAgentText(event.text, true); - } + this.onAgentText(event.text, event.type === "agent.response"); break; - - case "turn.ended": - if ( - event.turn_id === this.turnId && - !this.bargeInPending - ) { - this.gatewayTurnEnded = true; - this.setState("waiting-playback"); - - void this.finishPlaybackAndResume(); - } - break; - case "interrupted": - if ( - !this.bargeInPending || - event.turn_id !== this.interruptedTurnId - ) { - return; - } - - this.interruptedTurnId = null; + this.invalidatePlayback(); this.turnId = null; - this.sequence = 0; - this.pending = [...this.bargeInBuffer]; - this.bargeInBuffer = []; - - this.stopRequested = this.bargeInSpeechEnded; - - this.setState("starting"); - - try { - this.send({ type: "audio.start" }); - } catch (error) { - this.fail(error); - } + this.setState("listening"); break; - - case "error": - if ( - event.code === "INVALID_TURN" && - this.bargeInPending && - event.turn_id === this.interruptedTurnId - ) { - // The old turn completed before interrupt was processed. - // Its semantic state is already released by the Gateway. - this.interruptedTurnId = null; - this.turnId = null; - this.sequence = 0; - - this.pending = [...this.bargeInBuffer]; - this.bargeInBuffer = []; - - this.stopRequested = this.bargeInSpeechEnded; - this.gatewayTurnEnded = false; - - this.setState("starting"); - - try { - this.send({ type: "audio.start" }); - } catch (error) { - this.fail(error); - } - - break; - } - - this.fail( - new Error(`${event.code}: ${event.message}`) - ); - break; - - case "session.ended": - this.close(); + case "turn.ended": + if (this.gatewayTurnEnded) return; + this.gatewayTurnEnded = true; + this.setState("waiting-playback"); + void this.finishPlayback(this.playbackGeneration, this.turnId); break; } } fail(error) { - this.onError(error); - this.close(); + if (this.state === "closed") return; + try { + this.onError(error); + } finally { + this.close(); + } } close() { if (this.state === "closed") return; - - this.playbackGeneration++; - this.player?.stop(); - - this.vad.stop(); - this.bargeInVad.stop(); - - this.bargeInPending = false; - this.bargeInSpeechEnded = false; - this.bargeInCaptureStarted = false; - this.interruptedTurnId = null; - this.bargeInBuffer = []; - - cancelAnimationFrame(this.frameId); - + this.connectionGeneration++; + this.ready = false; + clearTimeout(this.handshakeTimer); + this.handshakeTimer = null; + this.invalidatePlayback(); if (this.worklet) { this.worklet.port.postMessage({ type: "stop" }); this.worklet.port.onmessage = null; this.worklet.disconnect(); } - this.source?.disconnect(); this.silentGain?.disconnect(); - const socket = this.ws; this.ws = null; - if (socket) { - socket.onclose = null; - socket.onerror = null; - socket.onmessage = null; - - if (socket.readyState === WebSocket.OPEN) { - if (this.turnId === null) { - socket.send(JSON.stringify({ type: "session.end" })); - } else { - socket.send(JSON.stringify({ - type: "interrupt", - turn_id: this.turnId, - })); - } + socket.onopen = socket.onmessage = socket.onerror = socket.onclose = null; + try { + if (socket.readyState === WebSocket.OPEN) socket.send(JSON.stringify({ type: "session.end" })); + } catch { + // Local cleanup must complete even when the transport has already failed. + } finally { + socket.close(); } - - socket.close(); } - - this.worklet = null; - this.source = null; - this.silentGain = null; + this.worklet = this.source = this.silentGain = null; this.turnId = null; - this.prebuffer = []; - this.pending = []; this.setState("closed"); } } diff --git a/tests/js/pcm-worklet.test.mjs b/tests/js/pcm-worklet.test.mjs new file mode 100644 index 0000000..ff0f2da --- /dev/null +++ b/tests/js/pcm-worklet.test.mjs @@ -0,0 +1,48 @@ +import assert from "node:assert/strict"; +import { readFileSync } from "node:fs"; +import { test } from "node:test"; +import vm from "node:vm"; + +function processor() { + let Processor; + class Base { + constructor() { + this.port = { sent: [], postMessage(data) { this.sent.push(data); } }; + } + } + vm.runInNewContext(readFileSync(new URL("../../src/voice_assistant/audio_gateway/web/pcm-worklet.js", import.meta.url), "utf8"), { + AudioWorkletProcessor: Base, sampleRate: 16000, + registerProcessor(name, constructor) { assert.equal(name, "pcm-input"); Processor = constructor; }, + }); + return new Processor(); +} + +test("negotiated capture emits complete little-endian PCM16 frames", () => { + const input = processor(); + input.port.onmessage({ data: { type: "start", chunkSamples: 2 } }); + input.process([[new Float32Array([-1, 0, 1])]]); + const chunk = input.port.sent[0]; + assert.equal(chunk.type, "pcm"); + assert.equal(chunk.buffer.byteLength, 4); + const samples = new DataView(chunk.buffer); + assert.equal(samples.getInt16(0, true), -32768); + assert.equal(samples.getInt16(2, true), 0); +}); + +test("capture continues across process callbacks until stop", () => { + const input = processor(); + input.port.onmessage({ data: { type: "start", chunkSamples: 2 } }); + input.process([[new Float32Array([0, 0, 0])]]); + input.process([[new Float32Array([0, 0])]]); + assert.equal(input.port.sent.filter((event) => event.type === "pcm").length, 2); + input.port.onmessage({ data: { type: "stop" } }); + input.process([[new Float32Array([0, 0, 0])]]); + assert.equal(input.port.sent.length, 2); +}); + +test("invalid capture window reports an error", () => { + const input = processor(); + input.port.onmessage({ data: { type: "start", chunkSamples: 0 } }); + assert.equal(input.port.sent[0].type, "error"); + assert.equal(input.capturing, false); +}); diff --git a/tests/js/vad.test.mjs b/tests/js/vad.test.mjs deleted file mode 100644 index c3f269e..0000000 --- a/tests/js/vad.test.mjs +++ /dev/null @@ -1,83 +0,0 @@ - -import assert from "node:assert/strict"; -import { test } from "node:test"; -import { VoiceActivityDetector } from "../../src/voice_assistant/audio_gateway/web/vad.js"; - -test("detects speech start after minimum duration", () => { - let starts = 0; - - const vad = new VoiceActivityDetector({ - minSpeechMs: 250, - onSpeechStart: () => starts++, - }); - - vad.start(); - - vad.process(0.04, 0); - vad.process(0.04, 100); - - assert.equal(starts, 0); - - vad.process(0.04, 260); - - assert.equal(starts, 1); -}); - -test("detects speech end after silence", () => { - let ends = 0; - - const vad = new VoiceActivityDetector({ - minSpeechMs: 100, - silenceMs: 800, - onSpeechEnd: () => ends++, - }); - - vad.start(); - - vad.process(0.04, 0); - vad.process(0.04, 120); - vad.process(0.04, 300); - - vad.process(0.001, 600); - vad.process(0.001, 1000); - - assert.equal(ends, 0); - - vad.process(0.001, 1101); - - assert.equal(ends, 1); -}); - -test("ignores short noise spikes", () => { - let starts = 0; - - const vad = new VoiceActivityDetector({ - minSpeechMs: 250, - onSpeechStart: () => starts++, - }); - - vad.start(); - - vad.process(0.05, 0); - vad.process(0.001, 80); - vad.process(0.001, 300); - - assert.equal(starts, 0); -}); - -test("stop disables detection", () => { - let starts = 0; - - const vad = new VoiceActivityDetector({ - minSpeechMs: 100, - onSpeechStart: () => starts++, - }); - - vad.start(); - vad.stop(); - - vad.process(0.05, 0); - vad.process(0.05, 200); - - assert.equal(starts, 0); -}); diff --git a/tests/js/voice-session.test.mjs b/tests/js/voice-session.test.mjs index fe1e250..f97c8ac 100644 --- a/tests/js/voice-session.test.mjs +++ b/tests/js/voice-session.test.mjs @@ -1,746 +1,270 @@ - import assert from "node:assert/strict"; import { afterEach, test } from "node:test"; +import { VoiceSessionController } from "../../src/voice_assistant/audio_gateway/web/voice-session.js"; -import { VoiceSessionController } from - "../../src/voice_assistant/audio_gateway/web/voice-session.js"; - -const originalWebSocket = globalThis.WebSocket; - -class FakeWebSocket { +const controllers = []; +class Node { + constructor() { this.disconnected = false; } + connect() {} + disconnect() { this.disconnected = true; } +} +class Context { + constructor() { + this.destination = {}; + this.currentTime = 0; + this.sources = []; + this.audioWorklet = { addModule: async () => {} }; + this.nextResume = null; + } + createAnalyser() { return new Node(); } + createGain() { return Object.assign(new Node(), { gain: { value: 1 } }); } + createMediaStreamSource() { this.input = new Node(); return this.input; } + async resume() { const pending = this.nextResume; this.nextResume = null; if (pending) await pending; } + createBuffer(channels, length, sampleRate) { + return { duration: length / sampleRate, copyToChannel(samples) { this.samples = samples; } }; + } + createBufferSource() { + const source = Object.assign(new Node(), { + stopped: false, onended: null, + start: () => this.sources.push(source), + stop() { this.stopped = true; }, + end() { this.onended?.(); }, + }); + return source; + } +} +class Worklet extends Node { + constructor() { + super(); + this.port = { messages: [], postMessage(data) { this.messages.push(data); }, onmessage: null }; + } +} +class Socket { static OPEN = 1; static CLOSED = 3; - - constructor() { - this.readyState = FakeWebSocket.OPEN; - this.sent = []; - this.closed = false; - } - - send(data) { - this.sent.push(data); - } - - close() { - this.closed = true; - this.readyState = FakeWebSocket.CLOSED; - } + constructor(url) { this.url = url; this.readyState = 0; this.sent = []; this.bufferedAmount = 0; } + send(data) { if (this.sendFailure) throw new Error("Transport failed"); this.sent.push(data); } + open() { this.readyState = 1; this.onopen?.(); } + close() { this.readyState = 3; this.onclose?.(); } + receive(event) { this.onmessage?.({ data: JSON.stringify(event) }); } } - -globalThis.WebSocket = FakeWebSocket; -globalThis.cancelAnimationFrame = () => {}; -globalThis.requestAnimationFrame = () => 1; - - -afterEach(() => { - // Restore only state modified by individual tests. +globalThis.WebSocket = Socket; +globalThis.AudioWorkletNode = Worklet; +globalThis.location = { protocol: "http:", host: "localhost:8001" }; +afterEach(() => { for (const controller of controllers.splice(0)) controller.close(); }); +const tick = () => new Promise((resolve) => setImmediate(resolve)); +const ready = (overrides = {}) => ({ + type: "session.ready", turn_detection: "server", + audio: { encoding: "pcm_s16le", sample_rate: 16000, channels: 1 }, + output_audio: { encoding: "pcm_s16le", sample_rate: 48000, channels: 1 }, + max_audio_chunk_bytes: 65536, ...overrides, }); - -function makeController() { - const states = []; - const errors = []; - const workletMessages = []; - +function messages(socket) { return socket.sent.filter((data) => typeof data === "string").map(JSON.parse); } +async function setup(options = {}, confirm = true) { + const errors = [], states = [], transcripts = [], replies = []; const controller = new VoiceSessionController({ - onStateChange: (state) => states.push(state), - onError: (error) => errors.push(error), + onError: (error) => errors.push(error), onStateChange: (state) => states.push(state), + onTranscript: (...args) => transcripts.push(args), onAgentText: (...args) => replies.push(args), + ...options, }); - - controller.ws = new FakeWebSocket(); - controller.worklet = { - port: { - postMessage: (data) => workletMessages.push(data), - onmessage: null, - }, - disconnect() {}, - }; - - controller.source = { disconnect() {} }; - controller.silentGain = { disconnect() {} }; - - return { - controller, - states, - errors, - workletMessages, - }; + controllers.push(controller); + const context = new Context(); + await controller.start({ stream: {}, audioContext: context }); + const socket = controller.ws; + socket.open(); + if (confirm) socket.receive(ready()); + return { controller, context, socket, errors, states, transcripts, replies }; } -test("speech detection requests a new turn", () => { - const { controller } = makeController(); - - controller.setState("listening"); - controller.beginTurn(); - - assert.equal(controller.state, "starting"); - - assert.deepEqual( - JSON.parse(controller.ws.sent[0]), - { type: "audio.start" } - ); -}); - -test("PCM is buffered until turn.started", () => { - const { controller } = makeController(); - - controller.setState("listening"); - +test("requests server VAD and starts capture only after session.ready", async () => { + const { controller, socket } = await setup({}, false); + assert.equal(messages(socket)[0].turn_detection, "server"); + assert.deepEqual(controller.worklet.port.messages, []); controller.handlePCM(new ArrayBuffer(4)); + assert.equal(socket.sent.length, 1); + socket.receive(ready()); + assert.deepEqual(controller.worklet.port.messages, [{ type: "start", chunkSamples: 1600 }]); + assert.equal(controller.state, "listening"); +}); - assert.equal(controller.prebuffer.length, 1); - assert.equal(controller.ws.sent.length, 0); - - controller.beginTurn(); - controller.handlePCM(new ArrayBuffer(6)); - - assert.equal(controller.pending.length, 2); - assert.equal(controller.ws.sent.length, 1); - - controller.handleGatewayEvent({ - type: "turn.started", - turn_id: 1, +for (const state of ["listening", "recording", "processing", "speaking", "waiting-playback"]) { + test(`sends continuous PCM while ${state}, without turn metadata`, async () => { + const { controller, socket } = await setup(); + controller.setState(state); + const audio = new ArrayBuffer(4); + controller.handlePCM(audio); + assert.deepEqual(messages(socket).at(-1), { type: "audio.input", sequence_number: 0 }); + assert.equal(socket.sent.at(-1), audio); }); +} +test("global input sequence never resets when backend opens another turn", async () => { + const { controller, socket } = await setup(); + controller.handlePCM(new ArrayBuffer(4)); + socket.receive({ type: "turn.started", turn_id: 1 }); + controller.handlePCM(new ArrayBuffer(4)); + socket.receive({ type: "interrupted", turn_id: 1 }); + socket.receive({ type: "turn.started", turn_id: 2 }); + controller.handlePCM(new ArrayBuffer(4)); + assert.deepEqual(messages(socket).filter((m) => m.type === "audio.input").map((m) => m.sequence_number), [0, 1, 2]); + assert.equal(messages(socket).some((m) => ["audio.start", "audio.stop", "interrupt"].includes(m.type)), false); + assert.equal(controller.worklet.port.messages.length, 1); +}); + +test("uses the negotiated chunk size for microphone capture", async () => { + const { controller, socket } = await setup({}, false); + socket.receive(ready({ max_audio_chunk_bytes: 1000 })); + assert.deepEqual(controller.worklet.port.messages, [{ type: "start", chunkSamples: 500 }]); +}); + +test("turn.ended waits for speakers without pausing microphone input", async () => { + const { controller, context, socket } = await setup(); + socket.receive({ type: "turn.started", turn_id: 1 }); + controller.handleAudioChunk(new ArrayBuffer(8)); + await controller.playbackPending; + assert.equal(controller.state, "speaking"); + socket.receive({ type: "turn.ended", turn_id: 1 }); + await tick(); + assert.equal(controller.state, "waiting-playback"); + controller.handlePCM(new ArrayBuffer(4)); + assert.equal(messages(socket).at(-1).type, "audio.input"); + context.sources[0].end(); + await tick(); + assert.equal(controller.state, "listening"); + assert.equal(controller.worklet.port.messages.length, 1); +}); + +test("a new backend turn stops old playback and ignores its pending completion", async () => { + const { controller, context, socket } = await setup(); + socket.receive({ type: "turn.started", turn_id: 1 }); + controller.handleAudioChunk(new ArrayBuffer(8)); + await controller.playbackPending; + socket.receive({ type: "turn.ended", turn_id: 1 }); + await tick(); + socket.receive({ type: "turn.started", turn_id: 2 }); + await tick(); + assert.equal(context.sources[0].stopped, true); + assert.equal(controller.turnId, 2); assert.equal(controller.state, "recording"); - assert.equal(controller.pending.length, 0); - assert.equal(controller.sequence, 2); - assert.equal(controller.ws.sent.length, 5); }); -test("PCM metadata precedes each binary frame", () => { - const { controller } = makeController(); - - controller.turnId = 7; - controller.setState("recording"); - - const pcm = new ArrayBuffer(8); - - controller.sendPCM(pcm); - - assert.deepEqual( - JSON.parse(controller.ws.sent[0]), - { - type: "audio.chunk", - turn_id: 7, - sequence_number: 0, - } - ); - - assert.equal(controller.ws.sent[1], pcm); - assert.equal(controller.sequence, 1); +test("old queued PCM is discarded after playback generation changes", async () => { + const { controller, context, socket } = await setup(); + let release; + context.nextResume = new Promise((resolve) => { release = resolve; }); + socket.receive({ type: "turn.started", turn_id: 1 }); + controller.handleAudioChunk(new ArrayBuffer(8)); + const obsolete = controller.playbackPending; + await tick(); + socket.receive({ type: "turn.started", turn_id: 2 }); + release(); + await obsolete; + assert.equal(context.sources.length, 0); + controller.handleAudioChunk(new ArrayBuffer(4)); + await controller.playbackPending; + assert.equal(context.sources.length, 1); }); -test("speech end requests worklet flush", () => { - const { controller, workletMessages } = makeController(); - - controller.turnId = 1; - controller.setState("recording"); - - controller.finishSpeech(); - - assert.equal(controller.state, "flushing"); - assert.deepEqual( - workletMessages.at(-1), - { type: "flush" } - ); - - assert.equal(controller.ws.sent.length, 0); +test("a failure in obsolete playback does not close the new turn", async () => { + const { controller, context, socket, errors } = await setup(); + let reject; + context.nextResume = new Promise((resolve, fail) => { reject = fail; }); + socket.receive({ type: "turn.started", turn_id: 1 }); + controller.handleAudioChunk(new ArrayBuffer(8)); + const obsolete = controller.playbackPending; + await tick(); + socket.receive({ type: "turn.started", turn_id: 2 }); + reject(new Error("Old playback failed")); + await obsolete; + assert.equal(errors.length, 0); + assert.equal(controller.turnId, 2); + assert.equal(socket.readyState, Socket.OPEN); }); - -test("audio.stop is sent after flushed", () => { - const { controller } = makeController(); - - controller.turnId = 1; - controller.setState("flushing"); - - controller.handleFlushed(); - - const messages = controller.ws.sent - .filter((message) => typeof message === "string") - .map((message) => JSON.parse(message)); - - assert.ok( - messages.some( - (message) => - message.type === "audio.stop" && - message.turn_id === 1 - ) - ); - - assert.equal(controller.state, "processing"); +test("stale turn events cannot change the current response", async () => { + const { controller, socket, transcripts, replies } = await setup(); + socket.receive({ type: "turn.started", turn_id: 2 }); + socket.receive({ type: "stt.final", turn_id: 1, text: "old" }); + socket.receive({ type: "agent.response", turn_id: 1, text: "old" }); + socket.receive({ type: "interrupted", turn_id: 1 }); + socket.receive({ type: "turn.ended", turn_id: 1 }); + socket.receive({ type: "turn.started", turn_id: 1 }); + assert.equal(controller.turnId, 2); + assert.deepEqual(transcripts, []); + assert.deepEqual(replies, []); + socket.receive({ type: "stt.final", turn_id: 2, text: "new" }); + socket.receive({ type: "agent.response", turn_id: 2, text: "reply" }); + assert.deepEqual(transcripts, [["new", true]]); + assert.deepEqual(replies, [["reply", true]]); }); +test("interrupted returns to listening without stopping capture or starting a client turn", async () => { + const { controller, socket } = await setup(); + socket.receive({ type: "turn.started", turn_id: 1 }); + socket.receive({ type: "interrupted", turn_id: 1 }); + assert.equal(controller.state, "listening"); + assert.equal(controller.turnId, null); + assert.deepEqual(messages(socket).map((m) => m.type), ["session.start"]); + assert.equal(controller.worklet.port.messages.length, 1); +}); -test("speech ending before turn.started is handled", () => { - const { controller, workletMessages } = makeController(); - - controller.setState("listening"); - controller.beginTurn(); - controller.finishSpeech(); - - assert.equal(controller.stopRequested, true); - assert.equal(controller.state, "starting"); - - controller.handleGatewayEvent({ - type: "turn.started", - turn_id: 3, - }); - - assert.equal(controller.state, "flushing"); - assert.deepEqual( - workletMessages.at(-1), - { type: "flush" } - ); -}); - -test("closing a session releases resources", () => { - const { controller, workletMessages } = makeController(); - const socket = controller.ws; - - controller.setState("listening"); - controller.close(); - - assert.equal(controller.state, "closed"); - assert.equal(socket.closed, true); - assert.equal(controller.ws, null); - assert.deepEqual( - workletMessages.at(-1), - { type: "stop" } - ); -}); - -test("gateway errors close the session", () => { - const { controller, errors } = makeController(); - - controller.setState("processing"); - - controller.handleGatewayEvent({ - type: "error", - code: "STT_START_FAILED", - message: "Provider unavailable", +for (const metadata of [{ turn_detection: "client" }, { turn_detection: undefined }, { max_audio_chunk_bytes: 1 }]) { + test(`rejects incompatible session metadata: ${JSON.stringify(metadata)}`, async () => { + const { controller, socket, errors } = await setup({}, false); + socket.receive(ready(metadata)); + assert.equal(controller.state, "closed"); + assert.equal(errors.length, 1); }); +} +test("backpressure closes the session before queuing more PCM", async () => { + const { controller, socket, errors } = await setup({ maxBufferedBytes: 8 }); + socket.bufferedAmount = 8; + controller.handlePCM(new ArrayBuffer(4)); assert.equal(controller.state, "closed"); assert.equal(errors.length, 1); - assert.match( - errors[0].message, - /STT_START_FAILED/ - ); + assert.equal(messages(socket).some((m) => m.type === "audio.input"), false); }); - -test("turn.ended waits for playback before listening", async () => { - const { controller } = makeController(); - - let resolvePlayback; - - controller.player = { - finish() {}, - reset() {}, - stop() {}, - waitForCompletion() { - return new Promise((resolve) => { - resolvePlayback = resolve; - }); - }, - }; - - controller.turnId = 1; - controller.setState("processing"); - - let resumed = false; - - controller.startListening = () => { - resumed = true; - }; - - controller.handleGatewayEvent({ - type: "turn.ended", - turn_id: 1, - }); - - await Promise.resolve(); - await Promise.resolve(); - - assert.equal(resumed, false); - - resolvePlayback(); - - await Promise.resolve(); - await Promise.resolve(); - - assert.equal(resumed, true); +test("invalid PCM closes the session without emitting metadata", async () => { + const { controller, socket, errors } = await setup(); + controller.handlePCM(new ArrayBuffer(3)); + assert.equal(errors.length, 1); + assert.equal(messages(socket).some((m) => m.type === "audio.input"), false); }); -test("closing during playback prevents listening resume", async () => { - const { controller } = makeController(); - - let resolvePlayback; - - controller.player = { - finish() {}, - reset() {}, - stop() {}, - waitForCompletion() { - return new Promise((resolve) => { - resolvePlayback = resolve; - }); - }, - }; - - controller.turnId = 1; - controller.setState("processing"); - - let resumed = false; - - controller.startListening = () => { - resumed = true; - }; - - controller.handleGatewayEvent({ - type: "turn.ended", - turn_id: 1, - }); - - await Promise.resolve(); - await Promise.resolve(); - - controller.close(); - - resolvePlayback(); - - await Promise.resolve(); - await Promise.resolve(); - - assert.equal(resumed, false); - assert.equal(controller.state, "closed"); -}); - -test("binary audio is forwarded to PCMPlayer", async () => { - const { controller } = makeController(); - - const received = []; - - controller.player = { - async enqueue(buffer) { - received.push(buffer); - }, - finish() {}, - reset() {}, - stop() {}, - waitForCompletion: async () => {}, - }; - - controller.turnId = 1; - controller.setState("processing"); - - const pcm = new ArrayBuffer(8); - - controller.handleAudioChunk(pcm); - - await controller.playbackPending; - - assert.deepEqual(received, [pcm]); -}); - - - -test("barge-in stops playback and starts a new turn", () => { - const { controller } = makeController(); - - let playbackStopped = false; - - controller.player = { - stop() { - playbackStopped = true; - }, - reset() {}, - }; - - controller.turnId = 1; - controller.setState("speaking"); - - // Simulate user speech detected during TTS. - controller.requestBargeIn(); - - assert.equal(playbackStopped, true); - assert.equal(controller.state, "interrupting"); - assert.equal(controller.bargeInPending, true); - - assert.deepEqual( - JSON.parse(controller.ws.sent[0]), - { - type: "interrupt", - turn_id: 1, - } - ); - - // Gateway confirms the previous turn was cancelled. - controller.handleGatewayEvent({ - type: "interrupted", - turn_id: 1, - }); - - assert.equal(controller.state, "starting"); - assert.equal(controller.ws.closed, false); - - assert.deepEqual( - JSON.parse(controller.ws.sent[1]), - { - type: "audio.start", - } - ); - - // Gateway accepts the new turn. - controller.handleGatewayEvent({ - type: "turn.started", - turn_id: 2, - }); - - assert.equal(controller.turnId, 2); - assert.equal(controller.state, "recording"); - assert.equal(controller.bargeInPending, false); -}); - - - -test("barge-in preserves PCM until new turn starts", () => { - const { controller } = makeController(); - - controller.player = { - stop() {}, - reset() {}, - }; - - controller.turnId = 1; - controller.setState("speaking"); - - const pcm1 = new ArrayBuffer(4); - const pcm2 = new ArrayBuffer(6); - const pcm3 = new ArrayBuffer(8); - - // Audio captured before speech confirmation. - controller.handlePCM(pcm1); - - controller.requestBargeIn(); - - // Audio captured while interruption is pending. - controller.handlePCM(pcm2); - - assert.equal(controller.ws.sent.length, 1); - assert.equal(controller.bargeInBuffer.length, 2); - - controller.handleGatewayEvent({ - type: "interrupted", - turn_id: 1, - }); - - // Still no PCM sent to the old turn. - assert.equal(controller.state, "starting"); - assert.equal(controller.ws.sent.length, 2); - - // Additional speech arrives before turn.started. - controller.handlePCM(pcm3); - - controller.handleGatewayEvent({ - type: "turn.started", - turn_id: 2, - }); - - const sent = controller.ws.sent; - - assert.deepEqual( - JSON.parse(sent[0]), - { type: "interrupt", turn_id: 1 } - ); - - assert.deepEqual( - JSON.parse(sent[1]), - { type: "audio.start" } - ); - - assert.deepEqual( - JSON.parse(sent[2]), - { - type: "audio.chunk", - turn_id: 2, - sequence_number: 0, - } - ); - - assert.equal(sent[3], pcm1); - - assert.deepEqual( - JSON.parse(sent[4]), - { - type: "audio.chunk", - turn_id: 2, - sequence_number: 1, - } - ); - - assert.equal(sent[5], pcm2); - - assert.deepEqual( - JSON.parse(sent[6]), - { - type: "audio.chunk", - turn_id: 2, - sequence_number: 2, - } - ); - - assert.equal(sent[7], pcm3); - - assert.equal(controller.sequence, 3); - assert.equal(controller.state, "recording"); -}); - - -test("barge-in finishes speech before new turn starts", () => { - const { controller, workletMessages } = makeController(); - - controller.player = { - stop() {}, - reset() {}, - }; - - controller.turnId = 1; - controller.setState("speaking"); - - const pcm = new ArrayBuffer(8); - - controller.handlePCM(pcm); - controller.requestBargeIn(); - - // User finishes speaking before the Gateway responds. - controller.finishBargeInSpeech(); - - assert.equal(controller.bargeInSpeechEnded, true); - - controller.handleGatewayEvent({ - type: "interrupted", - turn_id: 1, - }); - - assert.equal(controller.state, "starting"); - - controller.handleGatewayEvent({ - type: "turn.started", - turn_id: 2, - }); - - // Buffered audio must be transmitted before flushing. - assert.equal(controller.sequence, 1); - - assert.deepEqual( - JSON.parse(controller.ws.sent[2]), - { - type: "audio.chunk", - turn_id: 2, - sequence_number: 0, - } - ); - - assert.equal(controller.ws.sent[3], pcm); - assert.equal(controller.state, "flushing"); - - assert.deepEqual( - workletMessages.at(-1), - { type: "flush" } - ); -}); - - - - -test("barge-in continues recording until speech ends", () => { - const { controller, workletMessages } = makeController(); - - controller.player = { - stop() {}, - reset() {}, - }; - - controller.turnId = 1; - controller.setState("speaking"); - - controller.bargeInVad.start(); - controller.requestBargeIn(); - - controller.handleGatewayEvent({ - type: "interrupted", - turn_id: 1, - }); - - controller.handleGatewayEvent({ - type: "turn.started", - turn_id: 2, - }); - - assert.equal(controller.state, "recording"); - assert.equal(controller.bargeInRecording, true); - assert.equal(controller.bargeInVad.active, true); - - // Speech continues after the new turn starts. - const pcm = new ArrayBuffer(8); - controller.handlePCM(pcm); - - assert.equal(controller.sequence, 1); - - // Barge-in VAD detects the end of speech. - controller.finishBargeInSpeech(); - - assert.equal(controller.bargeInRecording, false); - assert.equal(controller.state, "flushing"); - - assert.deepEqual( - workletMessages.at(-1), - { type: "flush" } - ); - - // Worklet confirms that remaining PCM was emitted. - controller.handleFlushed(); - - assert.equal(controller.state, "processing"); - - const messages = controller.ws.sent - .filter((message) => typeof message === "string") - .map((message) => JSON.parse(message)); - - assert.ok( - messages.some( - (message) => - message.type === "audio.stop" && - message.turn_id === 2 - ) - ); -}); - - -test("late INVALID_TURN recovers barge-in session", () => { - const { controller } = makeController(); - - controller.player = { - stop() {}, - reset() {}, - }; - - controller.turnId = 1; - controller.setState("speaking"); - - controller.requestBargeIn(); - - assert.equal(controller.state, "interrupting"); - - controller.handleGatewayEvent({ - type: "error", - code: "INVALID_TURN", - message: "Turn 1 is no longer active.", - turn_id: 1, - }); - - assert.equal(controller.state, "starting"); - assert.equal(controller.ws.closed, false); - - const messages = controller.ws.sent - .filter((message) => typeof message === "string") - .map((message) => JSON.parse(message)); - - assert.deepEqual( - messages.map((message) => message.type), - ["interrupt", "audio.start"] - ); - - controller.handleGatewayEvent({ - type: "turn.started", - turn_id: 2, - }); - - assert.equal(controller.state, "recording"); - assert.equal(controller.turnId, 2); -}); - - -test("late TTS chunk during barge-in does not close session", () => { - const { controller } = makeController(); - - let stopped = false; - - controller.player = { - stop() { - stopped = true; - }, - reset() {}, - }; - - controller.turnId = 1; - controller.setState("speaking"); - - controller.requestBargeIn(); - - assert.equal(stopped, true); - assert.equal(controller.state, "interrupting"); - - // A late PCM chunk from the interrupted TTS. +test("close stops capture and playback even if session.end cannot be sent", async () => { + const { controller, socket, context } = await setup(); + const worklet = controller.worklet; + socket.receive({ type: "turn.started", turn_id: 1 }); controller.handleAudioChunk(new ArrayBuffer(8)); - - // The WebSocket session must remain alive. - assert.equal(controller.ws.closed, false); - assert.equal(controller.state, "interrupting"); - assert.equal(controller.bargeInPending, true); + await controller.playbackPending; + socket.sendFailure = true; + controller.close(); + assert.equal(controller.state, "closed"); + assert.equal(worklet.disconnected, true); + assert.deepEqual(worklet.port.messages.at(-1), { type: "stop" }); + assert.equal(context.sources[0].stopped, true); + assert.equal(socket.readyState, Socket.CLOSED); }); - -test("barge-in resets player before new TTS response", () => { - const { controller } = makeController(); - - let finished = false; - let resetCount = 0; - - controller.player = { - stop() { - finished = true; - }, - - reset() { - finished = false; - resetCount++; - }, - }; - - controller.turnId = 1; - controller.setState("speaking"); - - controller.requestBargeIn(); - - assert.equal(finished, true); - - controller.handleGatewayEvent({ - type: "interrupted", - turn_id: 1, - }); - - assert.equal(controller.state, "starting"); - - controller.handleGatewayEvent({ - type: "turn.started", - turn_id: 2, - }); - - assert.equal(controller.state, "recording"); - assert.equal(finished, false); - assert.equal(resetCount, 1); +test("closing during worklet loading prevents orphan microphone graphs and sockets", async () => { + let release; + const context = new Context(); + context.audioWorklet.addModule = () => new Promise((resolve) => { release = resolve; }); + const controller = new VoiceSessionController(); + controllers.push(controller); + const starting = controller.start({ stream: {}, audioContext: context }); + controller.close(); + release(); + await starting; + assert.equal(controller.ws, null); + assert.equal(context.input, undefined); }); - -test("session.ready configures output before starting microphone capture", () => { - const { controller } = makeController(); - const operations = []; - const format = { encoding: "pcm_s16le", sample_rate: 24000, channels: 1 }; - controller.player = { configure: (received) => operations.push(received) }; - controller.startListening = () => operations.push("listen"); - controller.handleGatewayEvent({ type: "session.ready", output_audio: format }); - assert.deepEqual(operations, [format, "listen"]); +test("missing session.ready times out and closes the connection", async () => { + const { controller, errors } = await setup({ handshakeTimeoutMs: 5 }, false); + await new Promise((resolve) => setTimeout(resolve, 20)); + assert.equal(controller.state, "closed"); + assert.equal(errors.length, 1); }); diff --git a/tests/test_continuous_gateway.py b/tests/test_continuous_gateway.py new file mode 100644 index 0000000..87a5c0e --- /dev/null +++ b/tests/test_continuous_gateway.py @@ -0,0 +1,223 @@ +"""Continuous PCM WebSocket mode delegates turn decisions to backend VAD.""" + +from contextlib import contextmanager + +import pytest +from fastapi.testclient import TestClient + +from voice_assistant.agent.mock import MockAgentAdapter +from voice_assistant.audio_gateway.app import main +from voice_assistant.stt.mock import MockSTTAdapter +from voice_assistant.tts.mock import MockTTSAdapter +from voice_assistant.vad.adapter import VADAdapter +from voice_assistant.vad.models import SpeechStarted, SpeechEnded + + +class VAD(VADAdapter): + def __init__(self): + self.script = [] + self.tail = () + self.received = [] + self.closed = False + + async def process(self, audio): + self.received.append(audio) + return self.script.pop(0) if self.script else () + + async def flush(self): + return self.tail + + async def reset(self): + pass + + async def aclose(self): + self.closed = True + + +class STT(MockSTTAdapter): + def __init__(self): + super().__init__() + self.received = [] + + async def send_audio(self, audio): + await super().send_audio(audio) + self.received.append(audio) + + async def stop_session(self): + await self.emit_final('Pergunta de teste') + await super().stop_session() + + +@pytest.fixture +def components(monkeypatch): + detector = VAD() + providers = [] + + def create_stt(**kwargs): + stt = STT() + providers.append(stt) + return stt + + monkeypatch.setattr(main, 'create_vad_adapter', lambda **kwargs: detector) + monkeypatch.setattr(main, 'create_stt_adapter', create_stt) + monkeypatch.setattr(main, 'create_agent_adapter', lambda **kwargs: MockAgentAdapter()) + monkeypatch.setattr(main, 'create_tts_adapter', lambda **kwargs: MockTTSAdapter()) + return detector, providers + + +@contextmanager +def connect(mode='server'): + with TestClient(main.app).websocket_connect('/ws/audio') as socket: + socket.send_json({'type': 'session.start', 'turn_detection': mode}) + ready = socket.receive_json() + assert ready['type'] == 'session.ready' + yield socket, ready + + +def send_audio(socket, sequence, audio): + socket.send_json({'type': 'audio.input', 'sequence_number': sequence}) + socket.send_bytes(audio) + + +def read_turn(socket): + events, audio = [], [] + for _ in range(20): + message = socket.receive() + if message.get('bytes') is not None: + audio.append(message['bytes']) + else: + import json + event = json.loads(message['text']) + assert event['type'] != 'error', event + events.append(event) + if event['type'] == 'turn.ended': + return events, audio + raise AssertionError('Turn did not complete within expected events') + + +def test_continuous_mode_receives_audio_before_first_turn(components): + detector, providers = components + with connect() as (socket, ready): + assert ready['turn_detection'] == 'server' + assert ready['max_audio_chunk_bytes'] == 65536 + send_audio(socket, 0, bytes(512)) + socket.send_json({'type': 'session.end'}) + assert socket.receive_json()['type'] == 'session.ended' + assert detector.received == [bytes(512)] + assert providers == [] + assert detector.closed + + +def test_vad_opens_and_closes_turn_and_input_sequence_survives_next_turn(components): + detector, providers = components + detector.script = [ + (SpeechStarted(0),), (SpeechEnded(400, 'silence'),), + (), (SpeechStarted(900),), + ] + detector.tail = (SpeechEnded(1024, 'end_of_input'),) + with connect() as (socket, ready): + send_audio(socket, 0, bytes(512)) + assert socket.receive_json()['type'] == 'turn.started' + send_audio(socket, 1, bytes(512)) + events, audio = read_turn(socket) + assert events[-1]['turn_id'] == 1 + assert any(event['type'] == 'stt.final' for event in events) + assert b''.join(audio) == b'\x01\x02\x03\x04\x05\x06\x07\x08' + send_audio(socket, 2, bytes(512)) + send_audio(socket, 3, bytes(512)) + assert socket.receive_json()['turn_id'] == 2 + socket.send_json({'type': 'audio.input.end'}) + events, audio = read_turn(socket) + assert events[-1]['turn_id'] == 2 + assert audio + assert len(providers) == 2 + assert len(detector.received) == 4 + + +@pytest.mark.parametrize('sequence', [-1, True, 0.5, '0', 1, 2]) +def test_bad_sequence_is_rejected_without_consuming_input(components, sequence): + detector, _ = components + with connect() as (socket, ready): + socket.send_json({'type': 'audio.input', 'sequence_number': sequence}) + error = socket.receive_json() + assert error['code'] == 'AUDIO_SEQUENCE_ERROR' + assert error['details']['expected_sequence_number'] == 0 + send_audio(socket, 0, bytes(512)) + socket.send_json({'type': 'session.end'}) + assert socket.receive_json()['type'] == 'session.ended' + assert len(detector.received) == 1 + + +@pytest.mark.parametrize(('audio', 'code'), [ + (b'', 'AUDIO_SEQUENCE_ERROR'), (bytes(3), 'AUDIO_SEQUENCE_ERROR'), + (bytes(6), 'MESSAGE_TOO_LARGE'), +]) +def test_bad_pcm_does_not_advance_global_sequence(components, monkeypatch, audio, code): + detector, _ = components + config = main.app_config.model_copy(update={'voice_input': {'max_input_chunk_bytes': 4}}) + monkeypatch.setattr(main, 'app_config', config) + with connect() as (socket, ready): + assert ready['max_audio_chunk_bytes'] == 4 + send_audio(socket, 0, audio) + assert socket.receive_json()['code'] == code + send_audio(socket, 0, bytes(4)) + socket.send_json({'type': 'session.end'}) + assert socket.receive_json()['type'] == 'session.ended' + assert detector.received == [bytes(4)] + + +def test_duplicate_metadata_does_not_overwrite_pending_pcm(components): + detector, _ = components + with connect() as (socket, ready): + socket.send_json({'type': 'audio.input', 'sequence_number': 0}) + socket.send_json({'type': 'audio.input', 'sequence_number': 1}) + assert socket.receive_json()['code'] == 'AUDIO_SEQUENCE_ERROR' + socket.send_bytes(bytes(512)) + socket.send_json({'type': 'session.end'}) + assert socket.receive_json()['type'] == 'session.ended' + assert len(detector.received) == 1 + + +def test_binary_frame_requires_metadata(components): + with connect() as (socket, ready): + socket.send_bytes(bytes(512)) + assert socket.receive_json()['code'] == 'INVALID_AUDIO_FRAME' + + +@pytest.mark.parametrize('event', [ + {'type': 'audio.start'}, {'type': 'audio.stop', 'turn_id': 1}, + {'type': 'audio.chunk', 'turn_id': 1, 'sequence_number': 0}, +]) +def test_client_turn_commands_are_rejected_in_server_mode(components, event): + with connect() as (socket, ready): + socket.send_json(event) + assert socket.receive_json()['code'] == 'TURN_CONTROLLED_BY_VAD' + + +@pytest.mark.parametrize('event', [ + {'type': 'audio.input', 'sequence_number': 0}, {'type': 'audio.input.end'}, +]) +def test_manual_mode_rejects_continuous_input(components, event): + with connect('client') as (socket, ready): + assert ready['turn_detection'] == 'client' + socket.send_json(event) + assert socket.receive_json()['code'] == 'INVALID_INPUT_MODE' + + +def test_input_end_cannot_overtake_pending_binary_frame(components): + with connect() as (socket, ready): + socket.send_json({'type': 'audio.input', 'sequence_number': 0}) + socket.send_json({'type': 'audio.input.end'}) + assert socket.receive_json()['code'] == 'INVALID_MESSAGE' + + +def test_initialization_failure_is_reported_and_can_be_followed_by_manual_session(components, monkeypatch): + def unavailable(**kwargs): + raise ValueError('VAD unavailable') + + monkeypatch.setattr(main, 'create_vad_adapter', unavailable) + with TestClient(main.app).websocket_connect('/ws/audio') as socket: + socket.send_json({'type': 'session.start', 'turn_detection': 'server'}) + assert socket.receive_json()['code'] == 'SESSION_START_FAILED' + socket.send_json({'type': 'session.start'}) + assert socket.receive_json()['type'] == 'session.ready' diff --git a/tests/test_protocol.py b/tests/test_protocol.py index a7c0eb9..37df946 100644 --- a/tests/test_protocol.py +++ b/tests/test_protocol.py @@ -95,6 +95,8 @@ def test_session_ready_output(): "sample_rate": 16000, "channels": 1, }, + "turn_detection": "client", + "max_audio_chunk_bytes": 65536, "output_audio": { "encoding": "pcm_s16le", "sample_rate": 48000, diff --git a/tests/test_web_client.py b/tests/test_web_client.py index f86c63a..87988ae 100644 --- a/tests/test_web_client.py +++ b/tests/test_web_client.py @@ -32,3 +32,13 @@ def test_web_client_audio_worklet(): assert response.status_code == 200 assert "registerProcessor" in response.text assert "pcm-input" in response.text + + + +def test_active_web_client_uses_continuous_server_detection(): + response = client.get("/web/voice-session.js") + assert response.status_code == 200 + assert 'turn_detection: "server"' in response.text + assert 'type: "audio.input"' in response.text + assert "VoiceActivityDetector" not in response.text + assert client.get("/web/vad.js").status_code == 404