diff --git a/README.md b/README.md index d0648ce..a9e742c 100644 --- a/README.md +++ b/README.md @@ -388,7 +388,7 @@ The script reads `audios/chafik_normalize.wav` (16 kHz, mono, 16-bit PCM), sends config/ Environment YAML and provider catalog src/voice_assistant/ audio_gateway/app/ WebSocket endpoint and protocol - application/ VoiceSession and transport-independent events + application/ VoiceSession, continuous PCM history, and application events domain/ Session state and domain rules vad/ VAD port, speech boundaries, and local Silero adapter audio/ Canonical audio models @@ -418,7 +418,7 @@ tools/ Integration and smoke scripts -## Backend VAD contract (integration pending) +## Backend VAD contract `vad/adapter.py` defines one stateful detector per continuous input stream. Its input is the canonical PCM S16LE, mono, 16 kHz audio format. Arbitrary @@ -432,9 +432,9 @@ boundary. Model padding never extends the input timeline. `reset()` discards detection state and restarts the timeline, and `aclose()` releases resources without emitting new speech events. Calls are serialized. These contracts do not start turns, select providers, or carry transport IDs. -The local Silero implementation is available independently of the running -conversation. Application integration and browser VAD replacement are subsequent -tasks; the active conversation flow still uses its existing browser VAD. +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. ### Local Silero VAD adapter @@ -497,5 +497,49 @@ 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. -This factory makes the configuration usable independently. Connecting it to the -conversation lifecycle remains the next task. +The VAD factory can be supplied to continuous voice sessions. The existing web +client remains on manual turn commands until the browser migration. + + +### Continuous audio and backend turn control + +A `VoiceSession` constructed with a VAD receives continuous canonical PCM through +`await session.receive_audio(pcm)`. The caller supplies audio without a turn ID; +the channel owns transport sequencing. `SpeechStarted` opens an STT turn and +`SpeechEnded` schedules normal STT finalization in the background. New speech +interrupts an active response/finalization before opening a fresh turn. Empty +STT results release the turn through an interruption event rather than leaving +an active turn behind. The VAD is not reset between conversation turns. + +The application's sample-indexed audio history preserves the beginning of speech +when VAD confirmation arrives later. PCM already sent to a turn is not replayed +within that turn; pre-roll never includes the previous utterance. History and +input size are bounded by environment configuration: + +```yaml +voice_input: + pre_roll_ms: 200 + history_ms: 2000 + max_input_chunk_bytes: 65536 +``` + +History is trimmed after processing each input chunk; transient storage is at most +retained history plus one accepted chunk. A speech start older than retained +history is reported as an input error instead of silently dropping speech. +Invalid PCM is rejected before detection or timeline advancement. A processing +failure requires a new session; cleanup still releases VAD and provider resources. + +Use `finish_audio_input()` for normal end of a continuous stream. Session end or +disconnect aborts through `end()`/`aclose()`, without forcing a final speech event. +The caller shields cleanup from transport cancellation, as for manual sessions. + +Backend-controlled sessions emit `TurnOpened` and `TurnInterrupted`; the WebSocket +mapper translates them to the existing `turn.started` and `interrupted` events. +Adapters should discard outdated queued playback when a new turn opens: backend +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. diff --git a/config/development.yaml b/config/development.yaml index 996bd25..aa9732f 100644 --- a/config/development.yaml +++ b/config/development.yaml @@ -24,3 +24,8 @@ vad: min_silence_duration_ms: 300 speech_pad_ms: 30 worker_timeout_seconds: 10.0 + +voice_input: + pre_roll_ms: 200 + history_ms: 2000 + max_input_chunk_bytes: 65536 diff --git a/src/voice_assistant/application/audio_history.py b/src/voice_assistant/application/audio_history.py new file mode 100644 index 0000000..8fdc5a8 --- /dev/null +++ b/src/voice_assistant/application/audio_history.py @@ -0,0 +1,32 @@ +"""A bounded PCM history addressed by absolute input sample positions.""" + +from voice_assistant.audio.models import CANONICAL_AUDIO_FORMAT + + +class AudioHistory: + def __init__(self, history_ms: int) -> None: + if type(history_ms) is not int or history_ms <= 0: + raise ValueError("history_ms must be a positive integer.") + self._retained_samples = history_ms * CANONICAL_AUDIO_FORMAT.sample_rate // 1000 + self._audio = bytearray() + self.start_sample = 0 + self.end_sample = 0 + + def append(self, audio: bytes) -> None: + self._audio.extend(audio) + self.end_sample += len(audio) // CANONICAL_AUDIO_FORMAT.frame_size_bytes + + def read(self, start: int, end: int) -> bytes: + if not self.start_sample <= start <= end <= self.end_sample: + raise ValueError("Requested audio is outside the retained input history.") + width = CANONICAL_AUDIO_FORMAT.frame_size_bytes + return bytes(self._audio[(start - self.start_sample) * width:(end - self.start_sample) * width]) + + def trim(self) -> None: + start = max(self.start_sample, self.end_sample - self._retained_samples) + del self._audio[:(start - self.start_sample) * CANONICAL_AUDIO_FORMAT.frame_size_bytes] + self.start_sample = start + + def clear(self) -> None: + self._audio.clear() + self.start_sample = self.end_sample diff --git a/src/voice_assistant/application/events.py b/src/voice_assistant/application/events.py index 145594a..6a1c717 100644 --- a/src/voice_assistant/application/events.py +++ b/src/voice_assistant/application/events.py @@ -41,6 +41,18 @@ class SpeechChunk: audio_format: AudioFormat +@dataclass(frozen=True, slots=True) +class TurnOpened: + session_id: str + turn_id: int + + +@dataclass(frozen=True, slots=True) +class TurnInterrupted: + session_id: str + turn_id: int + + @dataclass(frozen=True, slots=True) class TurnCompleted: session_id: str @@ -58,5 +70,5 @@ class SessionFailure: type VoiceEvent = ( TranscriptPartial | TranscriptFinal | AgentTextDelta | AgentReply - | SpeechChunk | TurnCompleted | SessionFailure + | SpeechChunk | TurnOpened | TurnInterrupted | TurnCompleted | SessionFailure ) diff --git a/src/voice_assistant/application/voice_session.py b/src/voice_assistant/application/voice_session.py index 19ab639..f79b08a 100644 --- a/src/voice_assistant/application/voice_session.py +++ b/src/voice_assistant/application/voice_session.py @@ -6,7 +6,12 @@ from collections.abc import Awaitable, Callable from contextlib import aclosing from voice_assistant.agent.adapter import StreamingAgentAdapter +from voice_assistant.application.audio_history import AudioHistory +from voice_assistant.vad.adapter import VADAdapter +from voice_assistant.vad.models import SpeechStarted, SpeechEnded, VADEvent from voice_assistant.application.events import ( + TurnOpened, + TurnInterrupted, AgentReply, AgentTextDelta, SessionFailure, @@ -18,6 +23,7 @@ from voice_assistant.application.events import ( ) from voice_assistant.audio.models import AudioFormat, CANONICAL_AUDIO_FORMAT from voice_assistant.stt.adapter import STTAdapter +from voice_assistant.turn_manager import TurnState from voice_assistant.voice_pipeline import ( PipelineAgentResponse, PipelineAudioChunk, @@ -64,9 +70,30 @@ class VoiceSession: emit: EventSink, stt_stop_timeout: float = 10.0, finalization_cancel_timeout: float = 5.0, + vad: VADAdapter | None = None, + pre_roll_ms: int = 200, + history_ms: int = 2000, + max_input_chunk_bytes: int = 65536, ) -> None: if stt_stop_timeout <= 0 or finalization_cancel_timeout <= 0: raise ValueError("Session timeouts must be positive.") + if type(pre_roll_ms) is not int or pre_roll_ms < 0: + raise ValueError("pre_roll_ms must be a nonnegative integer.") + if type(history_ms) is not int or history_ms <= 0 or pre_roll_ms > history_ms: + raise ValueError("history_ms must be positive and cover pre_roll_ms.") + if type(max_input_chunk_bytes) is not int or max_input_chunk_bytes <= 0: + raise ValueError("max_input_chunk_bytes must be a positive integer.") + self._vad = vad + self._input_lock = asyncio.Lock() + self._history = AudioHistory(history_ms) + self._pre_roll_samples = pre_roll_ms * CANONICAL_AUDIO_FORMAT.sample_rate // 1000 + self._max_input_chunk_bytes = max_input_chunk_bytes + self._input_failed = False + self._input_finished = False + self._closing = False + self._capture_turn_id: int | None = None + self._sent_until = 0 + self._last_voice_end = 0 self._pipeline = pipeline self._create_stt = create_stt self._emit = emit @@ -116,7 +143,17 @@ class VoiceSession: "SESSION_ALREADY_CLOSED", "This session is already closed." ) + def _require_manual_input(self) -> None: + if self._vad is not None: + raise VoiceSessionError( + "TURN_CONTROLLED_BY_VAD", "This session uses server-side turn detection.", + ) + async def start_turn(self) -> int: + self._require_manual_input() + return await self._start_turn() + + async def _start_turn(self) -> int: self._require_open() if self.active_turn_id is not None: raise VoiceSessionError( @@ -168,6 +205,12 @@ class VoiceSession: async def send_audio( self, turn_id: int, sequence_number: int, audio: bytes, + ) -> None: + self._require_manual_input() + await self._send_audio(turn_id, sequence_number, audio) + + async def _send_audio( + self, turn_id: int, sequence_number: int, audio: bytes, ) -> None: self.validate_audio_input(turn_id) CANONICAL_AUDIO_FORMAT.validate_chunk(audio) @@ -187,6 +230,10 @@ class VoiceSession: ) from exc def stop_turn(self, turn_id: int) -> None: + self._require_manual_input() + self._stop_turn(turn_id) + + def _stop_turn(self, turn_id: int) -> None: self._require_open() if self._stt is None: raise RuntimeError("STT session has not been started.") @@ -200,6 +247,12 @@ class VoiceSession: ) async def interrupt(self, turn_id: int) -> None: + async with self._input_lock: + await self._interrupt(turn_id) + if self._capture_turn_id == turn_id: + self._capture_turn_id = None + + async def _interrupt(self, turn_id: int) -> None: self._require_open() if not self._pipeline.is_active_turn(turn_id): raise VoiceSessionError( @@ -221,6 +274,11 @@ class VoiceSession: ) from exc async def end(self) -> None: + self._closing = True + async with self._input_lock: + await self._end() + + async def _end(self) -> None: self._require_open() try: async with self._turn_transition_lock: @@ -306,6 +364,12 @@ class VoiceSession: or not self._pipeline.is_active_turn(turn_id) ): return + turn = self._pipeline.turn_manager.active_turn + if self._vad is not None and turn is not None and turn.state == TurnState.LISTENING: + # A VAD trigger may produce no recognizable transcript. + self._pipeline.interrupt_if_active(turn_id) + await self._emit(TurnInterrupted(self.session_id, turn_id)) + return self._pipeline.complete_turn(turn_id) self._completed_turns.add(turn_id) await self._emit(TurnCompleted(self.session_id, turn_id)) @@ -389,6 +453,11 @@ class VoiceSession: self._stt_events_task = None async def aclose(self) -> None: + self._closing = True + async with self._input_lock: + await self._close_resources() + + async def _close_resources(self) -> None: """Attempt every cleanup operation, including after partial startup. The caller must shield this awaited cleanup from transport-scope @@ -425,3 +494,113 @@ class VoiceSession: "Failed to close %s provider. session_id=%s", name, self.session_id, ) + + if self._vad is not None: + try: + await self._vad.aclose() + except Exception: + logger.exception("Failed to close VAD. session_id=%s", self.session_id) + self._history.clear() + self._capture_turn_id = None + + def _require_continuous_input(self) -> VADAdapter: + self._require_open() + if self._vad is None: + raise VoiceSessionError("INVALID_INPUT_MODE", "Continuous input requires VAD.") + if self._closing or self._input_finished: + raise VoiceSessionError("INPUT_CLOSED", "Continuous audio input is closed.") + if self._input_failed: + raise VoiceSessionError("VOICE_INPUT_FAILED", "Create a new session after input failure.") + return self._vad + + async def receive_audio(self, audio: bytes) -> None: + """Receive continuous PCM independently of client-managed turn IDs. + + The channel serializes and validates its own packet sequence. This + operation validates PCM before advancing the application's timeline. + """ + CANONICAL_AUDIO_FORMAT.validate_chunk(audio) + if len(audio) > self._max_input_chunk_bytes: + raise ValueError("Continuous audio chunk exceeds its configured size limit.") + async with self._input_lock: + vad = self._require_continuous_input() + self._history.append(audio) + try: + events = await vad.process(audio) + if self._closing: + raise VoiceSessionError("INPUT_CLOSED", "Session shutdown is in progress.") + await self._handle_voice_activity(events) + await self._forward_input_until(self._history.end_sample) + except asyncio.CancelledError: + self._input_failed = True + raise + except Exception as exc: + self._input_failed = True + if isinstance(exc, VoiceSessionError): + raise + raise VoiceSessionError( + "VOICE_INPUT_FAILED", str(exc), turn_id=self.active_turn_id, + ) from exc + finally: + self._history.trim() + + async def finish_audio_input(self) -> None: + """Finish a normal input stream; session end/disconnect aborts instead.""" + async with self._input_lock: + vad = self._require_continuous_input() + try: + events = await vad.flush() + if self._closing: + raise VoiceSessionError("INPUT_CLOSED", "Session shutdown is in progress.") + await self._handle_voice_activity(events) + self._input_finished = True + except asyncio.CancelledError: + self._input_failed = True + raise + except Exception as exc: + self._input_failed = True + if isinstance(exc, VoiceSessionError): + raise + raise VoiceSessionError( + "VOICE_INPUT_FAILED", str(exc), turn_id=self.active_turn_id, + ) from exc + finally: + self._history.trim() + + async def _handle_voice_activity(self, events: tuple[VADEvent, ...]) -> None: + for event in events: + if isinstance(event, SpeechStarted): + if self._capture_turn_id is not None: + raise ValueError("VAD emitted a duplicate speech start.") + if not self._history.start_sample <= event.sample_offset < self._history.end_sample: + raise ValueError("Speech start is outside retained audio; increase history_ms.") + previous = self.active_turn_id + if previous is not None: + await self._interrupt(previous) + await self._emit(TurnInterrupted(self.session_id, previous)) + else: + await self._cancel_finalization_task() + turn_id = await self._start_turn() + self._capture_turn_id = turn_id + self._sent_until = max( + self._history.start_sample, self._last_voice_end, + event.sample_offset - self._pre_roll_samples, + ) + await self._emit(TurnOpened(self.session_id, turn_id)) + elif isinstance(event, SpeechEnded): + if not 0 <= event.sample_offset <= self._history.end_sample: + raise ValueError("Speech end is outside input audio.") + self._last_voice_end = event.sample_offset + if self._capture_turn_id is not None: + await self._forward_input_until(event.sample_offset) + turn_id = self._capture_turn_id + self._capture_turn_id = None + self._stop_turn(turn_id) + + async def _forward_input_until(self, end_sample: int) -> None: + turn_id = self._capture_turn_id + if turn_id is None or end_sample <= self._sent_until: + return + audio = self._history.read(self._sent_until, end_sample) + await self._send_audio(turn_id, self.expected_audio_sequence, audio) + self._sent_until = end_sample diff --git a/src/voice_assistant/audio_gateway/app/main.py b/src/voice_assistant/audio_gateway/app/main.py index baf6c80..d486427 100644 --- a/src/voice_assistant/audio_gateway/app/main.py +++ b/src/voice_assistant/audio_gateway/app/main.py @@ -13,7 +13,7 @@ from pydantic import BaseModel, ValidationError from voice_assistant.application.events import ( AgentReply, AgentTextDelta, SessionFailure, SpeechChunk, - TranscriptFinal, TranscriptPartial, TurnCompleted, VoiceEvent, + TranscriptFinal, TranscriptPartial, TurnCompleted, TurnOpened, TurnInterrupted, VoiceEvent, ) from voice_assistant.application.voice_session import VoiceSession, VoiceSessionError from voice_assistant.audio.models import CANONICAL_AUDIO_FORMAT @@ -26,9 +26,8 @@ from voice_assistant.audio_gateway.app.protocol import ( from voice_assistant.config.loader import load_config, load_provider_catalog from voice_assistant.domain.session import SessionState from voice_assistant.providers import ( - create_agent_adapter, create_stt_adapter, create_tts_adapter, + create_agent_adapter, create_stt_adapter, create_tts_adapter, create_vad_adapter, ) -from voice_assistant.tts.adapter import TTSAdapter from voice_assistant.turn_manager import TurnManager from voice_assistant.voice_pipeline import VoicePipeline @@ -81,6 +80,10 @@ async def send_voice_event(websocket: WebSocket, event: VoiceEvent) -> None: payload = AgentDelta(session_id=sid, turn_id=tid, text=text) case AgentReply(session_id=sid, turn_id=tid, text=text): payload = AgentResponse(session_id=sid, turn_id=tid, text=text) + case TurnOpened(session_id=sid, turn_id=tid): + payload = TurnStarted(session_id=sid, turn_id=tid) + case TurnInterrupted(session_id=sid, turn_id=tid): + payload = Interrupted(session_id=sid, turn_id=tid) case TurnCompleted(session_id=sid, turn_id=tid): payload = TurnEnded(session_id=sid, turn_id=tid) case SessionFailure(): @@ -93,20 +96,34 @@ async def send_voice_event(websocket: WebSocket, event: VoiceEvent) -> None: await send_json(websocket, payload) -async def create_voice_session(websocket: WebSocket, session_id: str | None) -> VoiceSession: - """Compose providers at the channel boundary, injecting application ports.""" +async def create_voice_session( + websocket: WebSocket, session_id: str | None, *, server_vad: bool = False, +) -> VoiceSession: + """Compose manual input or continuous input controlled by backend VAD.""" state = SessionState(session_id=session_id) if session_id is not None else SessionState() - agent = create_agent_adapter(config=app_config, catalog=provider_catalog) - tts: TTSAdapter | None = None + agent = None + tts = None + vad = None try: + agent = create_agent_adapter(config=app_config, catalog=provider_catalog) tts = create_tts_adapter(config=app_config, catalog=provider_catalog) + if server_vad: + vad = create_vad_adapter(config=app_config, catalog=provider_catalog) pipeline = VoicePipeline( session=state, turn_manager=create_turn_manager(), agent=agent, tts=tts, ) + return VoiceSession( + pipeline=pipeline, + create_stt=partial(create_stt_adapter, config=app_config, catalog=provider_catalog), + emit=partial(send_voice_event, websocket), + stt_stop_timeout=STT_STOP_TIMEOUT_SECONDS, + finalization_cancel_timeout=FINALIZATION_CANCEL_TIMEOUT_SECONDS, + vad=vad, + **app_config.voice_input, + ) except BaseException: - # A partially composed application has not yet acquired ownership. with anyio.CancelScope(shield=True): - for name, provider in (("Agent", agent), ("TTS", tts)): + for name, provider in (("Agent", agent), ("TTS", tts), ("VAD", vad)): if provider is None: continue try: @@ -116,13 +133,6 @@ async def create_voice_session(websocket: WebSocket, session_id: str | None) -> "Failed to close %s after session composition failure.", name, ) raise - return VoiceSession( - pipeline=pipeline, - create_stt=partial(create_stt_adapter, config=app_config, catalog=provider_catalog), - emit=partial(send_voice_event, websocket), - stt_stop_timeout=STT_STOP_TIMEOUT_SECONDS, - finalization_cancel_timeout=FINALIZATION_CANCEL_TIMEOUT_SECONDS, - ) @app.websocket("/ws/audio") diff --git a/src/voice_assistant/config/models.py b/src/voice_assistant/config/models.py index c5ff8aa..d741672 100644 --- a/src/voice_assistant/config/models.py +++ b/src/voice_assistant/config/models.py @@ -72,3 +72,4 @@ class AppConfig(BaseModel): tts: ComponentConfig agent: ComponentConfig vad: ComponentConfig | None = None + voice_input: dict[str, Any] = Field(default_factory=dict) diff --git a/tests/test_vad_session_composition.py b/tests/test_vad_session_composition.py new file mode 100644 index 0000000..d5500ce --- /dev/null +++ b/tests/test_vad_session_composition.py @@ -0,0 +1,93 @@ +"""Composition selects VAD input and routes application events to the channel.""" + +import json +from pathlib import Path +from unittest.mock import AsyncMock + +import pytest + +from voice_assistant.agent.mock import MockAgentAdapter +from voice_assistant.application.events import TurnOpened, TurnInterrupted +from voice_assistant.audio_gateway.app import main +from voice_assistant.config.loader import load_config +from voice_assistant.tts.mock import MockTTSAdapter +from voice_assistant.vad.adapter import VADAdapter + + +@pytest.fixture +def anyio_backend(): + return 'asyncio' + + +class VAD(VADAdapter): + async def process(self, audio): + return () + + async def flush(self): + return () + + async def reset(self): + pass + + async def aclose(self): + pass + + +@pytest.mark.anyio +@pytest.mark.parametrize('server_vad', [False, True]) +async def test_composition_selects_vad_and_accepts_environment_policy(monkeypatch, server_vad): + config = load_config('development', config_dir=Path(__file__).resolve().parents[1] / 'config') + assert config.voice_input == { + 'pre_roll_ms': 200, 'history_ms': 2000, 'max_input_chunk_bytes': 65536, + } + monkeypatch.setattr(main, 'app_config', config) + monkeypatch.setattr(main, 'create_agent_adapter', lambda **kwargs: MockAgentAdapter()) + monkeypatch.setattr(main, 'create_tts_adapter', lambda **kwargs: MockTTSAdapter()) + detector = VAD() + calls = [] + + def factory(**kwargs): + calls.append(kwargs) + return detector + + monkeypatch.setattr(main, 'create_vad_adapter', factory) + session = await main.create_voice_session(AsyncMock(), None, server_vad=server_vad) + try: + if server_vad: + assert len(calls) == 1 + assert calls[0]['config'] is config + await session.receive_audio(bytes(512)) + else: + assert calls == [] + with pytest.raises(Exception, match='requires VAD'): + await session.receive_audio(bytes(512)) + finally: + await session.aclose() + + +@pytest.mark.anyio +async def test_bad_policy_releases_all_created_providers(monkeypatch): + config = main.app_config.model_copy(update={'voice_input': {'history_ms': 0}}) + agent, tts, vad = MockAgentAdapter(), MockTTSAdapter(), VAD() + for provider in [agent, tts, vad]: + provider.aclose = AsyncMock() + monkeypatch.setattr(main, 'app_config', config) + monkeypatch.setattr(main, 'create_agent_adapter', lambda **kwargs: agent) + monkeypatch.setattr(main, 'create_tts_adapter', lambda **kwargs: tts) + monkeypatch.setattr(main, 'create_vad_adapter', lambda **kwargs: vad) + with pytest.raises(ValueError, match='history_ms'): + await main.create_voice_session(AsyncMock(), None, server_vad=True) + for provider in [agent, tts, vad]: + provider.aclose.assert_awaited_once() + + +@pytest.mark.anyio +async def test_automatic_turn_events_map_to_existing_protocol(): + socket = AsyncMock() + await main.send_voice_event(socket, TurnOpened('session', 2)) + await main.send_voice_event(socket, TurnInterrupted('session', 1)) + events = [json.loads(call.args[0]) for call in socket.send_text.await_args_list] + assert events == [ + {'type': 'turn.started', 'session_id': 'session', 'turn_id': 2}, + {'type': 'interrupted', 'session_id': 'session', 'turn_id': 1}, + ] diff --git a/tests/test_voice_session_vad.py b/tests/test_voice_session_vad.py new file mode 100644 index 0000000..88e0df5 --- /dev/null +++ b/tests/test_voice_session_vad.py @@ -0,0 +1,370 @@ +"""Continuous, channel-independent PCM controls conversation turns.""" + +import asyncio +import struct + +import pytest + +from voice_assistant.agent.mock import MockAgentAdapter +from voice_assistant.application.events import TurnCompleted, TurnOpened, TurnInterrupted, SpeechChunk +from voice_assistant.application.voice_session import VoiceSession, VoiceSessionError +from voice_assistant.domain.session import SessionState +from voice_assistant.stt.mock import MockSTTAdapter +from voice_assistant.tts.mock import MockTTSAdapter +from voice_assistant.turn_manager import TurnManager +from voice_assistant.vad.adapter import VADAdapter +from voice_assistant.vad.models import SpeechStarted, SpeechEnded +from voice_assistant.voice_pipeline import VoicePipeline + + +@pytest.fixture +def anyio_backend(): + return "asyncio" + + +class ScriptedVAD(VADAdapter): + def __init__(self, script=(), tail=()): + self.script = list(script) + self.tail = 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 RecordingSTT(MockSTTAdapter): + def __init__(self): + super().__init__() + self.received = [] + self.aborted = False + + async def send_audio(self, audio): + await super().send_audio(audio) + self.received.append(audio) + + async def stop_session(self): + await self.emit_final("Olá") + await super().stop_session() + + async def abort_session(self): + self.aborted = True + await super().abort_session() + + +def build(vad, factory=RecordingSTT, event_sink=None, **options): + providers = [] + events = [] + completed = asyncio.Event() + + def create_stt(): + provider = factory() + providers.append(provider) + return provider + + async def emit(event): + events.append(event) + if event_sink is not None: + await event_sink(event) + if isinstance(event, TurnCompleted): + completed.set() + + session = VoiceSession( + pipeline=VoicePipeline( + session=SessionState(), turn_manager=TurnManager(), + agent=MockAgentAdapter(), tts=MockTTSAdapter(), + ), + create_stt=create_stt, emit=emit, vad=vad, **options, + ) + return session, providers, events, completed + + +def pcm(value, samples=256): + return struct.pack(f'<{samples}h', *([value] * samples)) + + +@pytest.mark.anyio +async def test_delayed_start_replays_pre_roll_without_duplicating_audio(): + vad = ScriptedVAD([(), (), (SpeechStarted(400),), ()]) + session, providers, events, _ = build(vad, pre_roll_ms=10) + chunks = [pcm(i) for i in [1, 2, 3, 4]] + try: + await session.receive_audio(chunks[0]) + await session.receive_audio(chunks[1]) + assert providers == [] + await session.receive_audio(chunks[2]) + await session.receive_audio(chunks[3]) + assert len(providers) == 1 + assert b''.join(providers[0].received) == b''.join(chunks)[240 * 2:] + assert isinstance(events[0], TurnOpened) + finally: + await session.aclose() + + +@pytest.mark.anyio +async def test_speech_end_finalizes_in_background_while_silence_continues(): + vad = ScriptedVAD([(SpeechStarted(0),), (SpeechEnded(400, 'silence'),), ()]) + session, providers, events, completed = build(vad, pre_roll_ms=0) + try: + await session.receive_audio(pcm(1)) + await session.receive_audio(pcm(2)) + await session.receive_audio(pcm(3)) + await asyncio.wait_for(completed.wait(), timeout=1.0) + assert b''.join(providers[0].received) == pcm(1) + pcm(2)[:144 * 2] + assert isinstance(events[-1], TurnCompleted) + assert session.active_turn_id is None + finally: + await session.aclose() + + +@pytest.mark.anyio +async def test_multiple_speech_intervals_in_one_chunk_keep_audio_separate(): + vad = ScriptedVAD([( + SpeechStarted(0), SpeechEnded(160, 'silence'), + SpeechStarted(240), SpeechEnded(400, 'silence'), + )]) + session, providers, events, completed = build(vad, pre_roll_ms=0) + audio = struct.pack('<512h', *range(512)) + try: + await session.receive_audio(audio) + await asyncio.wait_for(completed.wait(), timeout=1.0) + assert len(providers) == 2 + assert b''.join(providers[0].received) == audio[:320] + assert b''.join(providers[1].received) == audio[480:800] + assert any(isinstance(event, TurnInterrupted) and event.turn_id == 1 for event in events) + assert [event.turn_id for event in events if isinstance(event, TurnOpened)] == [1, 2] + finally: + await session.aclose() + + +@pytest.mark.anyio +async def test_new_speech_cancels_stt_finalization_and_opens_fresh_turn(): + started = asyncio.Event() + cancelled = asyncio.Event() + + class WaitingSTT(RecordingSTT): + async def stop_session(self): + started.set() + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + cancelled.set() + raise + + vad = ScriptedVAD([ + (SpeechStarted(0),), (SpeechEnded(400, 'silence'),), (SpeechStarted(600),), + ]) + session, providers, events, _ = build(vad, factory=WaitingSTT, pre_roll_ms=0) + try: + await session.receive_audio(pcm(1)) + await session.receive_audio(pcm(2)) + await asyncio.wait_for(started.wait(), timeout=1.0) + await session.receive_audio(pcm(3)) + assert cancelled.is_set() + assert providers[0].aborted + assert len(providers) == 2 + assert session.active_turn_id == 2 + assert b''.join(providers[1].received) == pcm(3)[88 * 2:] + assert any(isinstance(event, TurnInterrupted) for event in events) + finally: + await session.aclose() + + +@pytest.mark.anyio +@pytest.mark.parametrize('audio', [b'', b'\x00', bytes(65538)]) +async def test_invalid_input_does_not_reach_vad_or_advance_history(audio): + vad = ScriptedVAD([(SpeechStarted(0),)]) + session, providers, _, _ = build(vad, pre_roll_ms=0) + try: + with pytest.raises(ValueError): + await session.receive_audio(audio) + assert vad.received == [] + await session.receive_audio(pcm(1)) + assert b''.join(providers[0].received) == pcm(1) + finally: + await session.aclose() + + +@pytest.mark.anyio +async def test_manual_commands_cannot_compete_with_vad_turn_control(): + session, _, _, _ = build(ScriptedVAD()) + try: + for operation in [session.start_turn, lambda: session.send_audio(1, 0, pcm(1))]: + with pytest.raises(VoiceSessionError) as error: + await operation() + assert error.value.code == 'TURN_CONTROLLED_BY_VAD' + with pytest.raises(VoiceSessionError): + session.stop_turn(1) + finally: + await session.aclose() + + +@pytest.mark.anyio +async def test_input_error_prevents_unsafe_retry_and_cleanup_closes_vad(): + class FailingVAD(ScriptedVAD): + async def process(self, audio): + raise RuntimeError('Inference failed') + + vad = FailingVAD() + session, _, _, _ = build(vad) + try: + with pytest.raises(VoiceSessionError, match='Inference failed'): + await session.receive_audio(pcm(1)) + with pytest.raises(VoiceSessionError, match='new session'): + await session.receive_audio(pcm(1)) + finally: + await session.aclose() + assert vad.closed + + +@pytest.mark.anyio +async def test_speech_older_than_history_is_reported_instead_of_silently_truncated(): + vad = ScriptedVAD([(), (SpeechStarted(0),)]) + session, providers, _, _ = build(vad, pre_roll_ms=0, history_ms=1) + try: + await session.receive_audio(pcm(1)) + with pytest.raises(VoiceSessionError, match='increase history_ms'): + await session.receive_audio(pcm(2)) + assert providers == [] + finally: + await session.aclose() + + +@pytest.mark.anyio +async def test_normal_end_of_input_flushes_active_speech(): + vad = ScriptedVAD([(SpeechStarted(0),)], tail=(SpeechEnded(256, 'end_of_input'),)) + session, _, _, completed = build(vad, pre_roll_ms=0) + try: + await session.receive_audio(pcm(1)) + await session.finish_audio_input() + await asyncio.wait_for(completed.wait(), timeout=1.0) + with pytest.raises(VoiceSessionError, match='closed'): + await session.receive_audio(pcm(2)) + finally: + await session.aclose() + + +@pytest.mark.anyio +async def test_session_end_during_vad_processing_does_not_open_late_turn(): + started = asyncio.Event() + release = asyncio.Event() + + class WaitingVAD(ScriptedVAD): + async def process(self, audio): + started.set() + await release.wait() + return (SpeechStarted(0),) + + vad = WaitingVAD() + session, providers, events, _ = build(vad) + receiving = asyncio.create_task(session.receive_audio(pcm(1))) + ending = None + try: + await asyncio.wait_for(started.wait(), timeout=1.0) + ending = asyncio.create_task(session.end()) + await asyncio.sleep(0) + release.set() + with pytest.raises(VoiceSessionError, match='shutdown'): + await receiving + await asyncio.wait_for(ending, timeout=1.0) + assert session.closed + assert providers == [] + assert events == [] + finally: + release.set() + await asyncio.gather(receiving, return_exceptions=True) + if ending: + await asyncio.gather(ending, return_exceptions=True) + await session.aclose() + + +@pytest.mark.parametrize('options', [ + {'pre_roll_ms': -1}, {'pre_roll_ms': True}, {'history_ms': 0}, + {'history_ms': 100, 'pre_roll_ms': 200}, {'max_input_chunk_bytes': 0}, +]) +def test_invalid_audio_history_policy_is_rejected(options): + with pytest.raises(ValueError): + build(ScriptedVAD(), **options) + + +@pytest.mark.anyio +async def test_empty_stt_result_releases_turn_without_stopping_continuous_input(): + interrupted = asyncio.Event() + + async def sink(event): + if isinstance(event, TurnInterrupted): + interrupted.set() + + vad = ScriptedVAD([ + (SpeechStarted(0),), (SpeechEnded(400, 'silence'),), (SpeechStarted(600),), + ]) + session, providers, _, _ = build(vad, factory=MockSTTAdapter, event_sink=sink) + try: + await session.receive_audio(pcm(1)) + await session.receive_audio(pcm(2)) + await asyncio.wait_for(interrupted.wait(), timeout=1.0) + assert session.active_turn_id is None + await session.receive_audio(pcm(3)) + assert session.active_turn_id == 2 + assert len(providers) == 2 + finally: + await session.aclose() + + +@pytest.mark.anyio +async def test_new_speech_cancels_response_delivery_before_opening_next_turn(): + delivering = asyncio.Event() + cancelled = asyncio.Event() + + async def sink(event): + if isinstance(event, SpeechChunk) and event.turn_id == 1: + delivering.set() + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + cancelled.set() + raise + + vad = ScriptedVAD([ + (SpeechStarted(0),), (SpeechEnded(400, 'silence'),), (SpeechStarted(600),), + ]) + session, providers, events, _ = build(vad, event_sink=sink, pre_roll_ms=0) + try: + await session.receive_audio(pcm(1)) + await session.receive_audio(pcm(2)) + await asyncio.wait_for(delivering.wait(), timeout=1.0) + await session.receive_audio(pcm(3)) + assert cancelled.is_set() + assert providers[0].aborted + assert session.active_turn_id == 2 + opened = next(i for i, event in enumerate(events) if isinstance(event, TurnOpened) and event.turn_id == 2) + assert not any(isinstance(event, SpeechChunk) and event.turn_id == 1 for event in events[opened:]) + finally: + await session.aclose() + + +@pytest.mark.anyio +async def test_pre_roll_never_replays_previous_utterance_into_next_stt(): + vad = ScriptedVAD([( + SpeechStarted(0), SpeechEnded(160, 'silence'), + SpeechStarted(240), SpeechEnded(400, 'silence'), + )]) + session, providers, _, completed = build(vad, pre_roll_ms=200) + audio = struct.pack('<512h', *range(512)) + try: + await session.receive_audio(audio) + await asyncio.wait_for(completed.wait(), timeout=1.0) + assert b''.join(providers[0].received) == audio[:320] + assert b''.join(providers[1].received) == audio[320:800] + finally: + await session.aclose()