diff --git a/README.md b/README.md index 8023c33..3268b17 100644 --- a/README.md +++ b/README.md @@ -25,7 +25,8 @@ A modular voice pipeline for streaming audio input, speech transcription, agent ```mermaid flowchart TB C[Audio Client] <-->|WebSocket /ws/audio · JSON + binary PCM| G[Audio Gateway · FastAPI] - G --> P[VoicePipeline] + G --> U[VoiceSession · conversation lifecycle] + U --> P[VoicePipeline] P --- S[SessionState · session and audio sequence] P --- T[TurnManager · semantic turn state] G --> STT[NVIDIA ASR · STT per turn] @@ -37,7 +38,7 @@ flowchart TB G -->|Streamed audio| C ``` -The **Audio Gateway** handles transport and protocol messages. **VoicePipeline** coordinates processing and task ownership. **SessionState** tracks session and audio sequence; **TurnManager** enforces turn transitions. A new STT adapter is created for each turn, while Agent and TTS adapters are created per session. +The **Audio Gateway** handles transport, protocol validation, and application composition. **VoiceSession** coordinates STT, finalization, interruption, rollback, and provider cleanup. **VoicePipeline** retains processing, turn-state operations, and execution-task ownership. **SessionState** tracks session and audio sequence; **TurnManager** enforces turn transitions. A new STT adapter is created for each turn, while Agent and TTS adapters are created per session. ## Architecture Backbone & Abstraction Boundaries @@ -57,6 +58,7 @@ The architecture separates five concerns: flowchart TB Client["Web Client / Audio Client"] Gateway["Audio Gateway
FastAPI · WebSocket"] + UseCase["VoiceSession
Conversation Lifecycle"] Pipeline["VoicePipeline
Orchestration & Streaming"] Session["SessionState
Session Lifecycle"] Turns["TurnManager
Semantic Turn Lifecycle"] @@ -83,7 +85,8 @@ flowchart TB end Client <-->|"JSON Events + Binary PCM"| Gateway - Gateway --> Pipeline + Gateway --> UseCase + UseCase --> Pipeline Pipeline --- Session Pipeline --- Turns @@ -144,7 +147,9 @@ The primary configuration components are: | Dynamic provider loading | `providers.py` | | Session lifecycle | `domain/session.py` | | Semantic turn lifecycle | `turn_manager.py` | -| Execution orchestration | `voice_pipeline.py` | +| Conversation lifecycle | `application/voice_session.py` | +| Application output events | `application/events.py` | +| Processing and execution tasks | `voice_pipeline.py` | ### Streaming Execution Backbone @@ -176,7 +181,8 @@ flowchart TB Output["WebSocket PCM Output"] Input --> Gateway - Gateway --> STT + Gateway --> UseCase["VoiceSession"] + UseCase --> STT STT --> Transcript Transcript --> Agent @@ -381,6 +387,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 domain/ Session state and domain rules audio/ Canonical audio models stt/ STT contract, mock and NVIDIA adapter diff --git a/src/voice_assistant/application/__init__.py b/src/voice_assistant/application/__init__.py new file mode 100644 index 0000000..8fdfbf4 --- /dev/null +++ b/src/voice_assistant/application/__init__.py @@ -0,0 +1 @@ +"""Transport-independent voice conversation use cases.""" diff --git a/src/voice_assistant/application/events.py b/src/voice_assistant/application/events.py new file mode 100644 index 0000000..c76347f --- /dev/null +++ b/src/voice_assistant/application/events.py @@ -0,0 +1,59 @@ +"""Application output, independent of transport serialization.""" + +from dataclasses import dataclass + + +@dataclass(frozen=True, slots=True) +class TranscriptPartial: + session_id: str + turn_id: int + text: str + + +@dataclass(frozen=True, slots=True) +class TranscriptFinal: + session_id: str + turn_id: int + text: str + + +@dataclass(frozen=True, slots=True) +class AgentTextDelta: + session_id: str + turn_id: int + text: str + + +@dataclass(frozen=True, slots=True) +class AgentReply: + session_id: str + turn_id: int + text: str + + +@dataclass(frozen=True, slots=True) +class SpeechChunk: + session_id: str + turn_id: int + audio: bytes + + +@dataclass(frozen=True, slots=True) +class TurnCompleted: + session_id: str + turn_id: int + + +@dataclass(frozen=True, slots=True) +class SessionFailure: + session_id: str + turn_id: int | None + code: str + message: str + retryable: bool = False + + +type VoiceEvent = ( + TranscriptPartial | TranscriptFinal | AgentTextDelta | AgentReply + | SpeechChunk | TurnCompleted | SessionFailure +) diff --git a/src/voice_assistant/application/voice_session.py b/src/voice_assistant/application/voice_session.py new file mode 100644 index 0000000..43c0e58 --- /dev/null +++ b/src/voice_assistant/application/voice_session.py @@ -0,0 +1,412 @@ +"""Coordinate a voice conversation independently of its delivery channel.""" + +import asyncio +import logging +from collections.abc import Awaitable, Callable +from contextlib import aclosing + +from voice_assistant.application.events import ( + AgentReply, + AgentTextDelta, + SessionFailure, + SpeechChunk, + TranscriptFinal, + TranscriptPartial, + TurnCompleted, + VoiceEvent, +) +from voice_assistant.stt.adapter import STTAdapter +from voice_assistant.voice_pipeline import ( + PipelineAgentResponse, + PipelineAudioChunk, + PipelineTextDelta, + VoicePipeline, +) + +logger = logging.getLogger(__name__) + +type EventSink = Callable[[VoiceEvent], Awaitable[None]] +type STTFactory = Callable[[], STTAdapter] + + +class VoiceSessionError(Exception): + """An application operation failed; the channel decides how to report it.""" + + def __init__( + self, + code: str, + message: str, + *, + turn_id: int | None = None, + retryable: bool = False, + ) -> None: + super().__init__(message) + self.code = code + self.turn_id = turn_id + self.retryable = retryable + + +class VoiceSession: + """Own STT execution, finalization, interruption, and provider cleanup. + + VoicePipeline retains its existing turn state and Agent/TTS behavior. + An awaited event sink preserves flow control and cancellation without + introducing a second, unbounded output queue. + """ + + def __init__( + self, + *, + pipeline: VoicePipeline, + create_stt: STTFactory, + emit: EventSink, + stt_stop_timeout: float = 10.0, + finalization_cancel_timeout: float = 5.0, + ) -> None: + if stt_stop_timeout <= 0 or finalization_cancel_timeout <= 0: + raise ValueError("Session timeouts must be positive.") + self._pipeline = pipeline + self._create_stt = create_stt + self._emit = emit + self._stt_stop_timeout = stt_stop_timeout + self._finalization_cancel_timeout = finalization_cancel_timeout + self._stt: STTAdapter | None = None + self._stt_events_task: asyncio.Task[None] | None = None + self._finalization_task: asyncio.Task[None] | None = None + self._audio_input_closed = False + self._turn_transition_lock = asyncio.Lock() + self._interrupting_turns: set[int] = set() + self._completed_turns: set[int] = set() + + @property + def session_id(self) -> str: + return self._pipeline.session.session_id + + @property + def active_turn_id(self) -> int | None: + return self._pipeline.session.active_turn_id + + @property + def expected_audio_sequence(self) -> int: + return self._pipeline.session.expected_audio_sequence + + @property + def closed(self) -> bool: + return self._pipeline.session.closed + + def validate_audio_input(self, turn_id: int) -> None: + self._require_open() + if self._audio_input_closed: + raise ValueError("Audio input is closed for this turn.") + if self.active_turn_id != turn_id: + raise ValueError( + f"Unexpected turn_id={turn_id}; " + f"active_turn_id={self.active_turn_id}" + ) + + def _require_open(self) -> None: + if self.closed: + raise VoiceSessionError( + "SESSION_ALREADY_CLOSED", "This session is already closed." + ) + + async def start_turn(self) -> int: + self._require_open() + if self.active_turn_id is not None: + raise VoiceSessionError( + "INVALID_TURN", "A turn is already active.", + turn_id=self.active_turn_id, + ) + turn_id = None + try: + turn_id = self._pipeline.start_turn().turn_id + self._stt = self._create_stt() + await self._stt.start_session() + self._stt_events_task = asyncio.create_task( + self._consume_stt_events(self._stt, turn_id) + ) + self._pipeline.register_turn_task(self._stt_events_task) + except (RuntimeError, ValueError) as exc: + cleanup_complete = await self._rollback_start(turn_id) + raise VoiceSessionError( + "STT_START_FAILED", str(exc), turn_id=turn_id, + retryable=cleanup_complete, + ) from exc + self._audio_input_closed = False + return turn_id + + async def _rollback_start(self, turn_id: int | None) -> bool: + cleanup_complete = True + try: + await self._cancel_stt_task() + except Exception: + cleanup_complete = False + logger.exception( + "Failed to cancel task during audio.start rollback. " + "session_id=%s turn_id=%s", self.session_id, turn_id, + ) + if self._stt is not None: + try: + await self._stt.abort_session() + except Exception: + cleanup_complete = False + logger.exception( + "Failed to abort STT during audio.start rollback. " + "session_id=%s turn_id=%s", self.session_id, turn_id, + ) + if cleanup_complete: + if turn_id is not None: + self._pipeline.interrupt_turn(turn_id) + self._stt = None + return cleanup_complete + + async def send_audio( + self, turn_id: int, sequence_number: int, audio: bytes, + ) -> None: + self.validate_audio_input(turn_id) + self._pipeline.session.accept_audio_chunk( + turn_id, sequence_number, len(audio), + ) + if self._stt is None: + raise VoiceSessionError( + "STT_NOT_STARTED", "audio.start is required before audio.", + turn_id=turn_id, + ) + try: + await self._stt.send_audio(audio) + except (RuntimeError, ValueError) as exc: + raise VoiceSessionError( + "STT_AUDIO_ERROR", str(exc), turn_id=turn_id, retryable=True, + ) from exc + + def stop_turn(self, turn_id: int) -> None: + self._require_open() + if self._stt is None: + raise RuntimeError("STT session has not been started.") + if not self._pipeline.is_active_turn(turn_id): + raise ValueError(f"Unexpected turn_id={turn_id}.") + if self._finalization_task is not None and not self._finalization_task.done(): + raise RuntimeError("Turn finalization is already running.") + self._audio_input_closed = True + self._finalization_task = asyncio.create_task( + self._finalize_turn(turn_id, self._stt) + ) + + async def interrupt(self, turn_id: int) -> None: + self._require_open() + if not self._pipeline.is_active_turn(turn_id): + raise VoiceSessionError( + "INVALID_TURN", f"Turn {turn_id} is no longer active.", + turn_id=turn_id, + ) + try: + async with self._turn_transition_lock: + self._interrupting_turns.add(turn_id) + await self._cancel_finalization_task() + await self._pipeline.cancel_turn_task() + if self._stt is not None: + await self._stt.abort_session() + self._pipeline.interrupt_if_active(turn_id) + self._stt_events_task = None + except Exception as exc: + raise VoiceSessionError( + "TURN_CANCEL_FAILED", str(exc), turn_id=turn_id, + ) from exc + + async def end(self) -> None: + self._require_open() + try: + async with self._turn_transition_lock: + if self.active_turn_id is not None: + self._interrupting_turns.add(self.active_turn_id) + await self._cancel_finalization_task() + await self._pipeline.cancel_turn_task() + if self._stt is not None: + await self._stt.abort_session() + self._pipeline.close_session() + self._stt_events_task = None + self._stt = None + except (RuntimeError, ValueError, TimeoutError) as exc: + raise VoiceSessionError( + "SESSION_END_FAILED", str(exc), turn_id=self.active_turn_id, + retryable=True, + ) from exc + + async def _consume_stt_events(self, stt: STTAdapter, turn_id: int) -> None: + async for event in stt.events(): + if not self._pipeline.is_active_turn(turn_id): + break + if event.type == "partial": + self._pipeline.on_partial(turn_id, event.text) + await self._emit(TranscriptPartial(self.session_id, turn_id, event.text)) + elif event.type == "final": + self._pipeline.on_final(turn_id, event.text) + await self._emit(TranscriptFinal(self.session_id, turn_id, event.text)) + await self._respond(turn_id, event.text) + + async def _respond(self, turn_id: int, text: str) -> None: + if callable(getattr(self._pipeline.agent, "stream", None)): + async with aclosing(self._pipeline.stream_turn_response( + session_id=self.session_id, turn_id=turn_id, text=text, + )) as response_events: + async for event in response_events: + if not self._pipeline.is_active_turn(turn_id): + break + if isinstance(event, PipelineTextDelta): + await self._emit(AgentTextDelta(self.session_id, turn_id, event.text)) + elif isinstance(event, PipelineAgentResponse): + await self._emit(AgentReply(self.session_id, turn_id, event.text)) + elif isinstance(event, PipelineAudioChunk): + await self._emit(SpeechChunk(self.session_id, turn_id, event.audio)) + else: + result = await self._pipeline.process_transcript( + session_id=self.session_id, turn_id=turn_id, text=text, + ) + await self._emit(AgentReply(self.session_id, turn_id, result.response_text)) + async for audio in self._pipeline.synthesize_response(result.response_text): + await self._emit(SpeechChunk(self.session_id, turn_id, audio)) + self._pipeline.on_tts_complete(turn_id) + + async def _finalize_turn(self, turn_id: int, stt: STTAdapter) -> None: + try: + try: + async with asyncio.timeout(self._stt_stop_timeout): + await stt.stop_session() + except TimeoutError: + await self._handle_stt_stop_failure( + turn_id, stt, "STT_STOP_TIMEOUT", + "STT finalization exceeded the time limit.", + ) + return + except Exception: + logger.exception( + "STT finalization failed. session_id=%s turn_id=%s", + self.session_id, turn_id, + ) + await self._handle_stt_stop_failure( + turn_id, stt, "STT_STOP_FAILED", + "STT could not finalize audio input.", + ) + return + await self._pipeline.wait_turn_task() + async with self._turn_transition_lock: + if ( + turn_id in self._interrupting_turns + or turn_id in self._completed_turns + or not self._pipeline.is_active_turn(turn_id) + ): + return + self._pipeline.complete_turn(turn_id) + self._completed_turns.add(turn_id) + await self._emit(TurnCompleted(self.session_id, turn_id)) + except asyncio.CancelledError: + raise + except Exception as exc: + logger.exception( + "Turn finalization failed. session_id=%s turn_id=%s", + self.session_id, turn_id, + ) + if turn_id not in self._interrupting_turns: + try: + await self._emit(SessionFailure( + self.session_id, turn_id, "TURN_EXECUTION_FAILED", str(exc), + )) + except Exception: + logger.exception("Failed to report turn execution failure.") + + async def _handle_stt_stop_failure( + self, turn_id: int, stt: STTAdapter, code: str, message: str, + ) -> None: + if not self._pipeline.is_active_turn(turn_id): + return + cleanup_complete = True + for operation_name, operation in ( + ("turn task cancellation", self._pipeline.cancel_turn_task), + ("STT abort", stt.abort_session), + ): + try: + await operation() + except Exception: + cleanup_complete = False + logger.exception( + "Failed during %s. session_id=%s turn_id=%s", + operation_name, self.session_id, turn_id, + ) + if cleanup_complete: + self._pipeline.interrupt_if_active(turn_id) + await self._emit(SessionFailure( + self.session_id, turn_id, code, message, cleanup_complete, + )) + + async def _cancel_finalization_task(self) -> None: + task = self._finalization_task + if task is None: + return + try: + await self._cancel_task(task, self._finalization_cancel_timeout) + except TimeoutError as exc: + raise TimeoutError("Turn finalization cancellation timed out.") from exc + finally: + if task.done() and self._finalization_task is task: + self._finalization_task = None + + @staticmethod + async def _cancel_task(task: asyncio.Task[None], timeout: float) -> None: + if task is asyncio.current_task(): + raise RuntimeError("A task cannot cancel itself.") + if not task.done(): + task.cancel() + try: + await asyncio.wait_for(asyncio.shield(task), timeout=timeout) + except asyncio.CancelledError: + current = asyncio.current_task() + if current is not None and current.cancelling(): + raise + if not task.cancelled(): + raise + + async def _cancel_stt_task(self) -> None: + task = self._stt_events_task + if task is None: + return + try: + if self._pipeline.turn_task is task: + await self._pipeline.cancel_turn_task() + else: + await self._cancel_task(task, 5.0) + finally: + if task.done(): + self._stt_events_task = None + + async def aclose(self) -> None: + """Attempt every cleanup operation, including after partial startup. + + The caller must shield this awaited cleanup from transport-scope + cancellation. Provider lifecycle contracts are unchanged here. + """ + for description, operation in ( + ("cancel finalization task", self._cancel_finalization_task), + ("cancel voice turn task", self._pipeline.cancel_turn_task), + ("clean up unregistered STT task", self._cancel_stt_task), + ): + try: + await operation() + except Exception: + logger.exception( + f"Failed to {description}. session_id=%s", self.session_id, + ) + if self._stt is not None: + try: + await self._stt.abort_session() + except Exception: + logger.exception("Failed to abort STT session. session_id=%s", self.session_id) + try: + self._pipeline.close_session() + except Exception: + logger.exception("Failed to close voice session. session_id=%s", self.session_id) + close_agent = getattr(self._pipeline.agent, "aclose", None) + if callable(close_agent): + try: + await close_agent() + except Exception: + logger.exception("Failed to close Agent provider. session_id=%s", self.session_id) diff --git a/src/voice_assistant/audio_gateway/app/main.py b/src/voice_assistant/audio_gateway/app/main.py index bcf0f35..4979b95 100644 --- a/src/voice_assistant/audio_gateway/app/main.py +++ b/src/voice_assistant/audio_gateway/app/main.py @@ -1,128 +1,45 @@ -""" -Audio Gateway application and WebSocket transport for the Voice Assistant. - -Purpose: - Expose the realtime audio protocol and connect client sessions - to the VoicePipeline orchestration layer. - -Responsibilities: - - Accept and validate WebSocket protocol messages. - - Receive canonical PCM audio frames. - - Create and configure session-scoped components. - - Forward STT events and TTS audio to connected clients. - - Delegate semantic turn lifecycle to VoicePipeline. - - Coordinate transport cleanup on disconnect. - - Report protocol and cleanup errors. - -Non-responsibilities: - - Implementing STT, Agent, or TTS provider logic. - - Managing semantic turn state directly. - - Loading provider implementations dynamically. - - Implementing provider-specific cancellation. - -Architecture: - WebSocket Client - | - v - Audio Gateway - | - v - VoicePipeline - | | | - STT Agent TTS -""" +"""WebSocket protocol validation, application composition, and event delivery.""" import json import logging +from functools import partial from pathlib import Path -import anyio +import anyio +from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.responses import FileResponse from fastapi.staticfiles import StaticFiles -from fastapi import FastAPI, WebSocket, WebSocketDisconnect from pydantic import BaseModel, ValidationError -from contextlib import aclosing -import asyncio - -from voice_assistant.voice_pipeline import VoicePipeline -from voice_assistant.stt.adapter import STTAdapter -from voice_assistant.tts.adapter import TTSAdapter -from voice_assistant.turn_manager import ( - TurnManager, +from voice_assistant.application.events import ( + AgentReply, AgentTextDelta, SessionFailure, SpeechChunk, + TranscriptFinal, TranscriptPartial, TurnCompleted, VoiceEvent, ) -from voice_assistant.agent.adapter import AgentAdapter - -from voice_assistant.audio.models import ( - CANONICAL_AUDIO_FORMAT, -) - -from voice_assistant.providers import ( - create_agent_adapter, - create_stt_adapter, - create_tts_adapter, -) - - -from voice_assistant.config.loader import ( - load_config, - load_provider_catalog, -) - - +from voice_assistant.application.voice_session import VoiceSession, VoiceSessionError +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, - error_event, - AgentResponse, - Interrupted, - SessionEnded, - SessionReady, - STTFinal, - STTPartial, - TurnEnded, - TurnStarted, - AgentDelta, + PROTOCOL_VERSION, AUDIO_ENCODING, AUDIO_SAMPLE_RATE, AUDIO_CHANNELS, + AudioChunk, AudioStop, Interrupt, SessionEnd, SessionStart, + AgentDelta, AgentResponse, Interrupted, SessionEnded, SessionReady, + STTFinal, STTPartial, TurnEnded, TurnStarted, error_event, ) - -from voice_assistant.voice_pipeline import ( - PipelineTextDelta, - PipelineAgentResponse, - PipelineAudioChunk, -) - - +from voice_assistant.config.loader import load_config, load_provider_catalog from voice_assistant.domain.session import SessionState - -# logger configuration -logger = logging.getLogger(__name__) - -app = FastAPI( - title="Voice Assistant Audio Gateway", - version=PROTOCOL_VERSION, +from voice_assistant.providers import ( + create_agent_adapter, create_stt_adapter, create_tts_adapter, ) +from voice_assistant.turn_manager import TurnManager +from voice_assistant.voice_pipeline import VoicePipeline +logger = logging.getLogger(__name__) +app = FastAPI(title="Voice Assistant Audio Gateway", version=PROTOCOL_VERSION) app_config = load_config() provider_catalog = load_provider_catalog() - -MAX_AUDIO_CHUNK_BYTES = 64 * 1024 # 64 KB +MAX_AUDIO_CHUNK_BYTES = 64 * 1024 STT_STOP_TIMEOUT_SECONDS = 10.0 FINALIZATION_CANCEL_TIMEOUT_SECONDS = 5.0 - WEB_DIR = Path(__file__).resolve().parents[1] / "web" - -app.mount( - "/web", - StaticFiles(directory=WEB_DIR), - name="web", -) +app.mount("/web", StaticFiles(directory=WEB_DIR), name="web") @app.get("/ui", include_in_schema=False) @@ -130,154 +47,6 @@ async def web_client() -> FileResponse: return FileResponse(WEB_DIR / "index.html") -def create_turn_manager() -> TurnManager: - """Create the turn manager used by the audio gateway.""" - return TurnManager() - - -async def send_json( - websocket: WebSocket, - payload: BaseModel | dict, -) -> None: - """ - Serialize and send one protocol event through the WebSocket. - - Typed Pydantic protocol models are preferred. Dictionary support - remains temporarily available for error events until ErrorEvent - is migrated to the typed protocol. - """ - if isinstance(payload, BaseModel): - data = payload.model_dump() - else: - data = payload - - await websocket.send_text( - json.dumps( - data, - separators=(",", ":"), - ) - ) - -async def close_optional_provider(provider: object | None) -> None: - """ - Close providers exposing an optional asynchronous cleanup hook. - - Providers without aclose() require no action. - """ - if provider is None: - return - - close_method = getattr(provider, "aclose", None) - - if callable(close_method): - await close_method() - - -async def forward_stt_events( - websocket: WebSocket, - stt: STTAdapter, - pipeline: VoicePipeline, - session_id: str, - turn_id: int, -) -> None: - async for event in stt.events(): - if not pipeline.is_active_turn(turn_id): - logger.debug( - "Discarding stale STT event. session_id=%s turn_id=%s", - session_id, - turn_id, - ) - break - if event.type == "partial": - pipeline.on_partial( - turn_id, - event.text, - ) - - await send_json( - websocket, - STTPartial( - session_id=session_id, - turn_id=turn_id, - text=event.text, - ), - ) - - elif event.type == "final": - pipeline.on_final( - turn_id, - event.text, - ) - - await send_json( - websocket, - STTFinal( - session_id=session_id, - turn_id=turn_id, - text=event.text, - ), - ) - - if callable(getattr(pipeline.agent, "stream", None)): - async with aclosing( - pipeline.stream_turn_response( - session_id=session_id, - turn_id=turn_id, - text=event.text, - ) - ) as response_events: - async for response_event in response_events: - if not pipeline.is_active_turn(turn_id): - break - - if isinstance(response_event, PipelineTextDelta): - await send_json( - websocket, - AgentDelta( - session_id=session_id, - turn_id=turn_id, - text=response_event.text, - ), - ) - - elif isinstance(response_event, PipelineAgentResponse): - await send_json( - websocket, - AgentResponse( - session_id=session_id, - turn_id=turn_id, - text=response_event.text, - ), - ) - - elif isinstance(response_event, PipelineAudioChunk): - await websocket.send_bytes(response_event.audio) - - else: - result = await pipeline.process_transcript( - session_id=session_id, - turn_id=turn_id, - text=event.text, - ) - - await send_json( - websocket, - AgentResponse( - session_id=session_id, - turn_id=turn_id, - text=result.response_text, - ), - ) - - async for audio_chunk in pipeline.synthesize_response( - result.response_text, - ): - await websocket.send_bytes(audio_chunk) - - pipeline.on_tts_complete(turn_id) - - - @app.get("/health") async def health() -> dict[str, str]: return {"status": "ok"} @@ -288,959 +57,257 @@ async def ready() -> dict[str, str]: return {"status": "ready"} +def create_turn_manager() -> TurnManager: + return TurnManager() + + +async def send_json(websocket: WebSocket, payload: BaseModel | dict) -> None: + data = payload.model_dump() if isinstance(payload, BaseModel) else payload + await websocket.send_text(json.dumps(data, separators=(",", ":"))) + + +async def send_voice_event(websocket: WebSocket, event: VoiceEvent) -> None: + """Translate application output to the existing WebSocket protocol.""" + match event: + case SpeechChunk(audio=audio): + await websocket.send_bytes(audio) + return + case TranscriptPartial(session_id=sid, turn_id=tid, text=text): + payload = STTPartial(session_id=sid, turn_id=tid, text=text) + case TranscriptFinal(session_id=sid, turn_id=tid, text=text): + payload = STTFinal(session_id=sid, turn_id=tid, text=text) + case AgentTextDelta(session_id=sid, turn_id=tid, text=text): + 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 TurnCompleted(session_id=sid, turn_id=tid): + payload = TurnEnded(session_id=sid, turn_id=tid) + case SessionFailure(): + payload = error_event( + event.code, event.message, session_id=event.session_id, + turn_id=event.turn_id, retryable=event.retryable, + ) + case _: + raise TypeError(f"Unsupported application event: {event!r}") + 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.""" + state = SessionState(session_id=session_id) if session_id is not None else SessionState() + agent = create_agent_adapter(config=app_config, catalog=provider_catalog) + try: + tts = create_tts_adapter(config=app_config, catalog=provider_catalog) + pipeline = VoicePipeline( + session=state, turn_manager=create_turn_manager(), agent=agent, tts=tts, + ) + except BaseException: + # A partially composed application has not yet acquired ownership. + with anyio.CancelScope(shield=True): + close_agent = getattr(agent, "aclose", None) + if callable(close_agent): + try: + await close_agent() + except Exception: + logger.exception("Failed to close Agent after session composition failure.") + 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") async def audio_socket(websocket: WebSocket) -> None: await websocket.accept() - - session: SessionState | None = None + session: VoiceSession | None = None pending_audio_chunk: AudioChunk | None = None - stt: STTAdapter | None = None - stt_events_task: asyncio.Task | None = None - - - finalization_task: asyncio.Task[None] | None = None - audio_input_closed = False - turn_transition_lock = asyncio.Lock() - interrupting_turns: set[int] = set() - completed_turns: set[int] = set() - - turn_manager = create_turn_manager() - agent: AgentAdapter | None = None - - tts: TTSAdapter | None = None - pipeline: VoicePipeline | None = None - - async def finalize_turn( - turn_id: int, - turn_stt: STTAdapter, + async def report( + code: str, message: str, *, turn_id: int | None = None, + retryable: bool = False, details: dict | None = None, ) -> None: - """ - Finalize STT and complete the turn without blocking reception. - """ - if pipeline is None or session is None: - return - - try: - try: - async with asyncio.timeout(STT_STOP_TIMEOUT_SECONDS): - await turn_stt.stop_session() - except TimeoutError: - await handle_stt_stop_failure( - turn_id, - turn_stt, - code="STT_STOP_TIMEOUT", - message="STT finalization exceeded the time limit.", - ) - return - except Exception: - logger.exception( - "STT finalization failed. session_id=%s turn_id=%s", - session.session_id, - turn_id, - ) - await handle_stt_stop_failure( - turn_id, - turn_stt, - code="STT_STOP_FAILED", - message="STT could not finalize audio input.", - ) - return - - await pipeline.wait_turn_task() - - async with turn_transition_lock: - if turn_id in interrupting_turns: - return - - if turn_id in completed_turns: - return - - if not pipeline.is_active_turn(turn_id): - return - - pipeline.complete_turn(turn_id) - completed_turns.add(turn_id) - - await send_json( - websocket, - TurnEnded( - session_id=session.session_id, - turn_id=turn_id, - ), - ) - - except asyncio.CancelledError: - raise - - except Exception as exc: - logger.exception( - "Turn finalization failed. session_id=%s turn_id=%s", - session.session_id, - turn_id, - ) - - if turn_id not in interrupting_turns: - try: - await send_json( - websocket, - error_event( - "TURN_EXECUTION_FAILED", - str(exc), - session_id=session.session_id, - turn_id=turn_id, - ), - ) - except Exception: - logger.exception( - "Failed to report turn execution failure." - ) - - async def handle_stt_stop_failure( - turn_id: int, - turn_stt: STTAdapter, - *, - code: str, - message: str, - ) -> None: - if pipeline is None or session is None: - return - - if not pipeline.is_active_turn(turn_id): - return - - cleanup_complete = True - - for operation_name, operation in ( - ("turn task cancellation", pipeline.cancel_turn_task), - ("STT abort", turn_stt.abort_session), - ): - try: - await operation() - except Exception: - cleanup_complete = False - logger.exception( - "Failed during %s. session_id=%s turn_id=%s", - operation_name, - session.session_id, - turn_id, - ) - - if cleanup_complete: - pipeline.interrupt_if_active(turn_id) - - await send_json( - websocket, - error_event( - code, - message, - session_id=session.session_id, - turn_id=turn_id, - retryable=cleanup_complete, - ), - ) - - async def cancel_finalization_task() -> None: - """ - Cancel and await the Gateway-owned finalization task. - - This task is separate from the execution task owned - by VoicePipeline. - """ - nonlocal finalization_task - - task = finalization_task - - if task is None: - return - - if task is asyncio.current_task(): - raise RuntimeError( - "Finalization task cannot cancel itself." - ) - - if not task.done(): - task.cancel() - - try: - await asyncio.wait_for( - asyncio.shield(task), - timeout=FINALIZATION_CANCEL_TIMEOUT_SECONDS, - ) - except asyncio.CancelledError: - current = asyncio.current_task() - - if current is not None and current.cancelling(): - raise - - if not task.cancelled(): - raise - - except TimeoutError as exc: - raise TimeoutError( - "Turn finalization cancellation timed out." - ) from exc - - finally: - if task.done() and finalization_task is task: - finalization_task = None + await send_json(websocket, error_event( + code, message, session_id=session.session_id if session else None, + turn_id=turn_id, retryable=retryable, details=details, + )) try: while True: message = await websocket.receive() - if message.get("type") == "websocket.disconnect": break - - # --------------------------------------------------------- - # Binary WebSocket frame -> PCM - # --------------------------------------------------------- if message.get("bytes") is not None: if session is None: - await send_json( - websocket, - error_event( - "SESSION_NOT_FOUND", - "session.start is required before audio.", - ), - ) + await report("SESSION_NOT_FOUND", "session.start is required before audio.") continue - if pending_audio_chunk is None: - await send_json( - websocket, - error_event( - "INVALID_AUDIO_FRAME", - "Binary PCM frame must be preceded by audio.chunk.", - session_id=session.session_id, - turn_id=session.active_turn_id, - ), + await report( + "INVALID_AUDIO_FRAME", + "Binary PCM frame must be preceded by audio.chunk.", + turn_id=session.active_turn_id, ) continue - - audio_payload = message["bytes"] - - if len(audio_payload) > MAX_AUDIO_CHUNK_BYTES: - await send_json( - websocket, - error_event( - "MESSAGE_TOO_LARGE", - ( - "Audio chunk exceeds maximum size of " - f"{MAX_AUDIO_CHUNK_BYTES} bytes." - ), - session_id=session.session_id, - turn_id=pending_audio_chunk.turn_id, - details={ - "max_bytes": MAX_AUDIO_CHUNK_BYTES, - "received_bytes": len(audio_payload), - }, - ), - ) - pending_audio_chunk = None - continue - - try: - session.accept_audio_chunk( - pending_audio_chunk.turn_id, - pending_audio_chunk.sequence_number, - len(audio_payload), - ) - except (RuntimeError, ValueError) as exc: - await send_json( - websocket, - error_event( - "AUDIO_SEQUENCE_ERROR", - str(exc), - session_id=session.session_id, - turn_id=pending_audio_chunk.turn_id, - details={ - "expected_sequence_number": - session.expected_audio_sequence, - }, - ), - ) - pending_audio_chunk = None - continue - - # echo do PCM para validar o transporte. - if stt is None: - await send_json( - websocket, - error_event( - "STT_NOT_STARTED", - "audio.start is required before audio.", - session_id=session.session_id, - turn_id=session.active_turn_id, - ), - ) - pending_audio_chunk = None - continue - - try: - await stt.send_audio(audio_payload) - except (RuntimeError, ValueError) as exc: - await send_json( - websocket, - error_event( - "STT_AUDIO_ERROR", - str(exc), - session_id=session.session_id, - turn_id=pending_audio_chunk.turn_id, - retryable=True, - ), - ) - pending_audio_chunk = None - continue - + chunk = pending_audio_chunk pending_audio_chunk = None + audio = message["bytes"] + 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)}, + ) + continue + try: + 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}, + ) continue - # --------------------------------------------------------- - # Text WebSocket frame -> JSON control event - # --------------------------------------------------------- text = message.get("text") - if text is None: - await send_json( - websocket, - error_event( - "INVALID_MESSAGE", - "Unsupported WebSocket frame.", - ), - ) + await report("INVALID_MESSAGE", "Unsupported WebSocket frame.") continue - try: payload = json.loads(text) except json.JSONDecodeError: - await send_json( - websocket, - error_event( - "INVALID_MESSAGE", - "Message must contain valid JSON.", - ), - ) + await report("INVALID_MESSAGE", "Message must contain valid JSON.") continue - if not isinstance(payload, dict): - await send_json( - websocket, - error_event( - "INVALID_MESSAGE", - "Message must be a JSON object.", - ), - ) + await report("INVALID_MESSAGE", "Message must be a JSON object.") continue - event_type = payload.get("type") - if not isinstance(event_type, str) or not event_type.strip(): - await send_json( - websocket, - error_event( - "INVALID_MESSAGE", - "Field 'type' must be a non-empty string.", - ), - ) + await report("INVALID_MESSAGE", "Field 'type' must be a non-empty string.") continue - # --------------------------------------------------------- - # session.start - # --------------------------------------------------------- if event_type == "session.start": if session is not None: - await send_json( - websocket, - error_event( - "SESSION_ALREADY_STARTED", - "This WebSocket already has an active session.", - session_id=session.session_id, - ), - ) + await report("SESSION_ALREADY_STARTED", "This WebSocket already has an active session.") continue - try: request = SessionStart.model_validate(payload) except ValidationError as exc: - await send_json( - websocket, - error_event( - "INVALID_MESSAGE", - "Invalid session.start payload.", - details={"errors": exc.errors()}, - ), + await report( + "INVALID_MESSAGE", "Invalid session.start payload.", + details={"errors": exc.errors()}, ) continue - audio = request.audio - if ( audio.encoding != AUDIO_ENCODING or audio.sample_rate != AUDIO_SAMPLE_RATE or audio.channels != AUDIO_CHANNELS ): - await send_json( - websocket, - error_event( - "INVALID_AUDIO_FORMAT", - ( - "Only PCM S16LE, mono, 16 kHz " - "is supported in the MVP." - ), - details={ - "expected": { - "encoding": AUDIO_ENCODING, - "sample_rate": AUDIO_SAMPLE_RATE, - "channels": AUDIO_CHANNELS, - } - }, - ), + await report( + "INVALID_AUDIO_FORMAT", + "Only PCM S16LE, mono, 16 kHz is supported in the MVP.", + details={"expected": { + "encoding": AUDIO_ENCODING, "sample_rate": AUDIO_SAMPLE_RATE, + "channels": AUDIO_CHANNELS, + }}, ) continue - - session = SessionState( - session_id=request.session_id or SessionState().session_id - ) - - agent = create_agent_adapter( - config=app_config, - catalog=provider_catalog, - ) - - tts = create_tts_adapter( - config=app_config, - catalog=provider_catalog, - ) - - pipeline = VoicePipeline( - session=session, - turn_manager=turn_manager, - agent=agent, - tts=tts, - ) - - await send_json( - websocket, - SessionReady( - protocol_version=PROTOCOL_VERSION, - session_id=session.session_id, - audio=CANONICAL_AUDIO_FORMAT, - ), - ) + session = await create_voice_session(websocket, request.session_id or None) + await send_json(websocket, SessionReady( + protocol_version=PROTOCOL_VERSION, session_id=session.session_id, + audio=CANONICAL_AUDIO_FORMAT, + )) continue - - # --------------------------------------------------------- - # Todos os eventos abaixo exigem sessão ativa - # --------------------------------------------------------- if session is None: - await send_json( - websocket, - error_event( - "SESSION_NOT_FOUND", - "session.start is required before this event.", - ), - ) + await report("SESSION_NOT_FOUND", "session.start is required before this event.") continue - if session.closed: - await send_json( - websocket, - error_event( - "SESSION_ALREADY_CLOSED", - "This session is already closed.", - session_id=session.session_id, - ), - ) + await report("SESSION_ALREADY_CLOSED", "This session is already closed.") continue - # --------------------------------------------------------- - # audio.start - # --------------------------------------------------------- if event_type == "audio.start": - if session.active_turn_id is not None: - await send_json( - websocket, - error_event( - "INVALID_TURN", - "A turn is already active.", - session_id=session.session_id, - turn_id=session.active_turn_id, - ), - ) - continue - - turn_id: int | None = None - try: - if pipeline is None: - raise RuntimeError( - "Voice pipeline is not initialized." - ) - - turn = pipeline.start_turn() - turn_id = turn.turn_id - - # Each turn receives a fresh STT provider instance. - stt = create_stt_adapter( - config=app_config, - catalog=provider_catalog, - ) - - await stt.start_session() - - stt_events_task = asyncio.create_task( - forward_stt_events( - websocket=websocket, - stt=stt, - pipeline=pipeline, - session_id=session.session_id, - turn_id=turn_id, - ) - ) - pipeline.register_turn_task(stt_events_task) - - - except (ValidationError, RuntimeError, ValueError) as exc: - # Preserve the original startup failure. - startup_error = exc - cleanup_complete = True - - # Cancel and await any task created during startup. - if stt_events_task is not None: - task = stt_events_task - - try: - if ( - pipeline is not None - and pipeline.turn_task is task - ): - await pipeline.cancel_turn_task() - else: - if not task.done(): - task.cancel() - - try: - await asyncio.wait_for( - asyncio.shield(task), - timeout=5.0, - ) - except asyncio.CancelledError: - if not task.cancelled(): - raise - - except Exception: - cleanup_complete = False - logger.exception( - "Failed to cancel task during audio.start rollback. " - "session_id=%s turn_id=%s", - session.session_id if session else None, - turn_id, - ) - if task.done(): - stt_events_task = None - else: - cleanup_complete = False - - # Abort the partially initialized STT provider. - if stt is not None: - try: - await stt.abort_session() - except Exception: - cleanup_complete = False - logger.exception( - "Failed to abort STT during audio.start rollback. " - "session_id=%s turn_id=%s", - session.session_id if session else None, - turn_id, - ) - - # Only release semantic state after execution - # has stopped and provider cleanup succeeded. - if cleanup_complete: - if turn_id is not None and pipeline is not None: - pipeline.interrupt_turn(turn_id) - - stt = None + turn_id = await session.start_turn() + except VoiceSessionError as exc: + if exc.retryable: pending_audio_chunk = None - - await send_json( - websocket, - error_event( - "STT_START_FAILED", - str(startup_error), - session_id=session.session_id, - turn_id=turn_id, - retryable=cleanup_complete, - ), - ) + await report(exc.code, str(exc), turn_id=exc.turn_id, retryable=exc.retryable) continue - - # Startup completed successfully. - audio_input_closed = False - await send_json( - websocket, - TurnStarted( - session_id=session.session_id, - turn_id=turn_id, - ), - ) + await send_json(websocket, TurnStarted(session_id=session.session_id, turn_id=turn_id)) continue - # --------------------------------------------------------- - # audio.chunk - # --------------------------------------------------------- if event_type == "audio.chunk": try: request = AudioChunk.model_validate(payload) - - if audio_input_closed: - raise ValueError( - "Audio input is closed for this turn." - ) - - if session.active_turn_id != request.turn_id: - raise ValueError( - f"Unexpected turn_id={request.turn_id}; " - f"active_turn_id={session.active_turn_id}" - ) - + session.validate_audio_input(request.turn_id) if pending_audio_chunk is not None: - raise ValueError( - "Previous audio chunk is still pending PCM data." - ) - + raise ValueError("Previous audio chunk is still pending PCM data.") pending_audio_chunk = request - except (ValidationError, ValueError) as exc: - await send_json( - websocket, - error_event( - "AUDIO_SEQUENCE_ERROR", - str(exc), - session_id=session.session_id, - turn_id=payload.get("turn_id"), - details={ - "expected_sequence_number": - session.expected_audio_sequence, - }, - ), + await report( + "AUDIO_SEQUENCE_ERROR", str(exc), turn_id=payload.get("turn_id"), + details={"expected_sequence_number": session.expected_audio_sequence}, ) - continue - continue - # --------------------------------------------------------- - # audio.stop - # --------------------------------------------------------- if event_type == "audio.stop": try: request = AudioStop.model_validate(payload) - - if stt is None: - raise RuntimeError( - "STT session has not been started." - ) - - if pipeline is None: - raise RuntimeError( - "Voice pipeline is not initialized." - ) - - if agent is None or tts is None: - raise RuntimeError( - "Agent or TTS adapter is not initialized." - ) - - if not pipeline.is_active_turn(request.turn_id): - raise ValueError( - f"Unexpected turn_id={request.turn_id}." - ) - if pending_audio_chunk is not None: - raise ValueError( - "audio.stop received while an audio chunk is pending." - ) - - if ( - finalization_task is not None - and not finalization_task.done() - ): - raise RuntimeError( - "Turn finalization is already running." - ) - - audio_input_closed = True - finalization_task = asyncio.create_task( - finalize_turn(request.turn_id, stt) - ) - + raise ValueError("audio.stop received while an audio chunk is pending.") + session.stop_turn(request.turn_id) except (ValidationError, RuntimeError, ValueError) as exc: - await send_json( - websocket, - error_event( - "INVALID_TURN", - str(exc), - session_id=session.session_id, - turn_id=payload.get("turn_id"), - ), - ) - + await report("INVALID_TURN", str(exc), turn_id=payload.get("turn_id")) continue - - # --------------------------------------------------------- - # interrupt - # --------------------------------------------------------- if event_type == "interrupt": try: request = Interrupt.model_validate(payload) - except ValidationError as exc: - await send_json( - websocket, - error_event( - "INVALID_MESSAGE", - "Invalid interrupt payload.", - session_id=session.session_id, - details={"errors": exc.errors()}, - ), + await report( + "INVALID_MESSAGE", "Invalid interrupt payload.", + details={"errors": exc.errors()}, ) continue - - if pipeline is None: - await send_json( - websocket, - error_event( - "INVALID_TURN", - "Voice pipeline is not initialized.", - session_id=session.session_id, - turn_id=request.turn_id, - ), - ) + try: + await session.interrupt(request.turn_id) + except VoiceSessionError as exc: + await report(exc.code, str(exc), turn_id=exc.turn_id, retryable=exc.retryable) continue - - # Only cancel resources belonging to the requested active turn. - if pipeline.is_active_turn(request.turn_id): - try: - async with turn_transition_lock: - interrupting_turns.add(request.turn_id) - - await cancel_finalization_task() - - await pipeline.cancel_turn_task() - - # Abort transcription without requesting a final result. - if stt is not None: - await stt.abort_session() - - # Release semantic turn state after execution stops. - pipeline.interrupt_if_active(request.turn_id) - - # Clear references to the interrupted execution. - stt_events_task = None - pending_audio_chunk = None - - except Exception as exc: - await send_json( - websocket, - error_event( - "TURN_CANCEL_FAILED", - str(exc), - session_id=session.session_id, - turn_id=request.turn_id, - ), - ) - continue - - else: - await send_json( - websocket, - error_event( - "INVALID_TURN", - ( - f"Turn {request.turn_id} " - "is no longer active." - ), - session_id=session.session_id, - turn_id=request.turn_id, - ), - ) - continue - - await send_json( - websocket, - Interrupted( - session_id=session.session_id, - turn_id=request.turn_id, - ), - ) + pending_audio_chunk = None + await send_json(websocket, Interrupted(session_id=session.session_id, turn_id=request.turn_id)) continue - # --------------------------------------------------------- - # session.end - # --------------------------------------------------------- if event_type == "session.end": try: SessionEnd.model_validate(payload) - - if pipeline is None: - raise RuntimeError( - "Voice pipeline is not initialized." - ) - # Prevent normal completion during session shutdown. - async with turn_transition_lock: - if session.active_turn_id is not None: - interrupting_turns.add(session.active_turn_id) - - await cancel_finalization_task() - - # Cancel and await any running STT/Agent/TTS task. - await pipeline.cancel_turn_task() - - # Abort STT without waiting for a final transcript. - if stt is not None: - await stt.abort_session() - - # Close semantic turn and session state. - pipeline.close_session() - - # Release Gateway-owned references. - stt_events_task = None - stt = None - pending_audio_chunk = None - - except ( - ValidationError, - RuntimeError, - ValueError, - TimeoutError, - ) as exc: - await send_json( - websocket, - error_event( - "SESSION_END_FAILED", - str(exc), - session_id=session.session_id, - turn_id=session.active_turn_id, - retryable=True, - ), + except ValidationError as exc: + await report( + "SESSION_END_FAILED", str(exc), turn_id=session.active_turn_id, + retryable=True, ) continue - - await send_json( - websocket, - SessionEnded( - session_id=session.session_id, - ), - ) + try: + await session.end() + except VoiceSessionError as exc: + await report(exc.code, str(exc), turn_id=exc.turn_id, retryable=exc.retryable) + continue + pending_audio_chunk = None + await send_json(websocket, SessionEnded(session_id=session.session_id)) break - # --------------------------------------------------------- - # Evento desconhecido - # --------------------------------------------------------- - await send_json( - websocket, - error_event( - "UNSUPPORTED_EVENT", - f"Unsupported event type: {event_type!r}.", - session_id=session.session_id, - turn_id=session.active_turn_id, - ), + await report( + "UNSUPPORTED_EVENT", f"Unsupported event type: {event_type!r}.", + turn_id=session.active_turn_id, ) - except WebSocketDisconnect: - # The finally block owns connection cleanup. pass - finally: with anyio.CancelScope(shield=True): - # Stop Gateway-owned finalization before provider cleanup. - try: - await cancel_finalization_task() - except Exception: - logger.exception( - "Failed to cancel finalization task. session_id=%s", - session.session_id if session else None, - ) - - # Cancel and await the task owned by VoicePipeline. - if pipeline is not None: - try: - await pipeline.cancel_turn_task() - except Exception: - logger.exception( - "Failed to cancel voice turn task. session_id=%s", - session.session_id if session else None, - ) - - # Handle tasks created before pipeline registration. - if stt_events_task is not None: - task = stt_events_task - - if ( - pipeline is None - or pipeline.turn_task is not task - ): - try: - if not task.done(): - task.cancel() - - try: - await asyncio.wait_for( - asyncio.shield(task), - timeout=5.0, - ) - except asyncio.CancelledError: - if not task.cancelled(): - raise - - except TimeoutError: - logger.error( - "Unregistered STT task cleanup timed out. " - "session_id=%s", - session.session_id if session else None, - ) - except Exception: - logger.exception( - "Failed to clean up unregistered STT task. " - "session_id=%s", - session.session_id if session else None, - ) - - if task.done(): - stt_events_task = None - - # Abort STT without waiting for a final transcript. - if stt is not None: - try: - await stt.abort_session() - except Exception: - logger.exception( - "Failed to abort STT session. session_id=%s", - session.session_id if session else None, - ) - - # Release session and semantic turn state. - try: - if pipeline is not None: - pipeline.close_session() - elif session is not None: - session.close() - except Exception: - logger.exception( - "Failed to close voice session. session_id=%s", - session.session_id if session else None, - ) - - # Release provider-owned HTTP connections. - if agent is not None: - try: - await close_optional_provider(agent) - except Exception: - logger.exception( - "Failed to close Agent provider. session_id=%s", - session.session_id if session else None, - ) - finally: - agent = None \ No newline at end of file + if session is not None: + await session.aclose() diff --git a/src/voice_assistant/voice_pipeline.py b/src/voice_assistant/voice_pipeline.py index 8a4114b..8cc6c69 100644 --- a/src/voice_assistant/voice_pipeline.py +++ b/src/voice_assistant/voice_pipeline.py @@ -20,7 +20,7 @@ Non-responsibilities: - Sending protocol events to clients. Architecture: - Audio Gateway + VoiceSession | v VoicePipeline diff --git a/tests/test_gateway.py b/tests/test_gateway.py index a439345..31c7cc6 100644 --- a/tests/test_gateway.py +++ b/tests/test_gateway.py @@ -28,7 +28,18 @@ from voice_assistant.tts.mock import MockTTSAdapter from voice_assistant.voice_pipeline import VoicePipeline -from voice_assistant.audio_gateway.app.main import forward_stt_events +from voice_assistant.application import voice_session as voice_session_module +from voice_assistant.application.voice_session import VoiceSession + + +async def forward_stt_events(websocket, stt, pipeline, session_id, turn_id): + # Exercise application event handling with the real WebSocket mapper. + async def emit(event): + await main.send_voice_event(websocket, event) + + conversation = VoiceSession(pipeline=pipeline, create_stt=lambda: stt, emit=emit) + await conversation._consume_stt_events(stt, turn_id) + from voice_assistant.stt.models import TranscriptEvent created_stt: list[MockSTTAdapter] = [] @@ -853,7 +864,7 @@ def test_websocket_disconnect_continues_cleanup_after_stt_abort_failure( side_effect=failing_abort, ): with patch.object( - main.logger, + voice_session_module.logger, "exception", ) as log_mock: with client.websocket_connect("/ws/audio") as websocket: @@ -932,7 +943,7 @@ def test_websocket_disconnect_continues_cleanup_after_task_cancel_failure(): side_effect=tracked_abort, ), patch.object( - main.logger, + voice_session_module.logger, "exception", ) as log_mock, ): @@ -1591,7 +1602,7 @@ def test_websocket_audio_start_abort_failure_is_not_retryable(): return_value=failing_stt, ), patch.object( - main.logger, + voice_session_module.logger, "exception", ) as log_mock, ): diff --git a/tests/test_voice_session.py b/tests/test_voice_session.py new file mode 100644 index 0000000..b4b0849 --- /dev/null +++ b/tests/test_voice_session.py @@ -0,0 +1,151 @@ +"""Exercise the conversation use case without a transport or protocol.""" + +import asyncio + +import pytest + +from voice_assistant.agent.mock import MockAgentAdapter +from voice_assistant.application.events import ( + AgentReply, SpeechChunk, TranscriptFinal, TranscriptPartial, TurnCompleted, +) +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.voice_pipeline import VoicePipeline + + +@pytest.fixture +def anyio_backend(): + return "asyncio" + + +def make_session(create_stt, emit): + return VoiceSession( + pipeline=VoicePipeline( + session=SessionState(), turn_manager=TurnManager(), + agent=MockAgentAdapter(), tts=MockTTSAdapter(), + ), + create_stt=create_stt, emit=emit, + ) + + +@pytest.mark.anyio +async def test_conversation_completes_without_websocket(): + events = [] + completed = asyncio.Event() + stt = MockSTTAdapter() + + async def emit(event): + events.append(event) + if isinstance(event, TurnCompleted): + completed.set() + + session = make_session(lambda: stt, emit) + try: + turn_id = await session.start_turn() + await session.send_audio(turn_id, 0, b"\x00\x00" * 160) + await stt.emit_partial("Bom") + await stt.emit_final("Bom dia") + session.stop_turn(turn_id) + await asyncio.wait_for(completed.wait(), timeout=1.0) + + assert isinstance(events[0], TranscriptPartial) + assert isinstance(events[1], TranscriptFinal) + assert any(isinstance(event, AgentReply) for event in events) + assert any(isinstance(event, SpeechChunk) for event in events) + assert isinstance(events[-1], TurnCompleted) + assert all(event.session_id == session.session_id for event in events) + assert all(event.turn_id == turn_id for event in events) + assert session.active_turn_id is None + await session.end() + assert session.closed + finally: + await session.aclose() + + +@pytest.mark.anyio +async def test_interrupt_releases_turn_and_creates_fresh_stt(): + providers = [] + + def create_stt(): + provider = MockSTTAdapter() + providers.append(provider) + return provider + + async def emit(event): + pass + + session = make_session(create_stt, emit) + try: + first = await session.start_turn() + await session.interrupt(first) + assert session.active_turn_id is None + second = await session.start_turn() + assert second > first + assert providers[0] is not providers[1] + assert not providers[0]._started + await session.send_audio(second, 0, b"\x00\x00") + finally: + await session.aclose() + + +@pytest.mark.anyio +async def test_stt_finalization_does_not_block_interrupt(): + started = asyncio.Event() + cancelled = asyncio.Event() + + class WaitingSTT(MockSTTAdapter): + async def stop_session(self): + started.set() + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + cancelled.set() + raise + + async def emit(event): + pass + + stt = WaitingSTT() + session = make_session(lambda: stt, emit) + try: + turn_id = await session.start_turn() + session.stop_turn(turn_id) + await asyncio.wait_for(started.wait(), timeout=1.0) + with pytest.raises(ValueError, match="Audio input is closed"): + await session.send_audio(turn_id, 0, b"\x00\x00") + await asyncio.wait_for(session.interrupt(turn_id), timeout=1.0) + assert cancelled.is_set() + assert session.active_turn_id is None + assert not stt._started + finally: + await session.aclose() + + +@pytest.mark.anyio +@pytest.mark.parametrize("abort_fails", [False, True]) +async def test_startup_rollback_only_releases_safe_turns(abort_fails): + class FailingSTT(MockSTTAdapter): + async def start_session(self): + await super().start_session() + raise RuntimeError("Failed startup") + + async def abort_session(self): + if abort_fails: + raise RuntimeError("Failed cleanup") + await super().abort_session() + + async def emit(event): + pass + + session = make_session(FailingSTT, emit) + try: + with pytest.raises(VoiceSessionError) as failure: + await session.start_turn() + assert failure.value.code == "STT_START_FAILED" + assert failure.value.retryable is (not abort_fails) + assert (session.active_turn_id is not None) is abort_fails + finally: + await session.aclose()