diff --git a/pyproject.toml b/pyproject.toml index 0940a88..169f31c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,6 +5,7 @@ description = "Realtime voice assistant with WebSocket audio gateway, streaming readme = "README.md" requires-python = ">=3.12" dependencies = [ + "anyio>=4.15.1,<5", "deepagents>=0.7.26", "fastapi>=0.141.1", "httpx>=0.28,<1", diff --git a/src/voice_assistant/audio_gateway/app/main.py b/src/voice_assistant/audio_gateway/app/main.py index af2717d..22325af 100644 --- a/src/voice_assistant/audio_gateway/app/main.py +++ b/src/voice_assistant/audio_gateway/app/main.py @@ -35,6 +35,7 @@ Architecture: import json import logging from pathlib import Path +import anyio from fastapi.responses import FileResponse from fastapi.staticfiles import StaticFiles @@ -1044,93 +1045,93 @@ async def audio_socket(websocket: WebSocket) -> None: pass finally: - # 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: + with anyio.CancelScope(shield=True): + # Stop Gateway-owned finalization before provider cleanup. try: - await pipeline.cancel_turn_task() + await cancel_finalization_task() except Exception: logger.exception( - "Failed to cancel voice turn task. session_id=%s", + "Failed to cancel finalization task. session_id=%s", session.session_id if session else None, ) - # A task may have been created before registration failed. - # Handle it independently if the pipeline does not own it. - if stt_events_task is not None: - task = stt_events_task - - if ( - pipeline is None - or pipeline.turn_task is not task - ): + # Cancel and await the task owned by VoicePipeline. + if pipeline is not None: 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, - ) + await pipeline.cancel_turn_task() except Exception: logger.exception( - "Failed to clean up unregistered STT task. " - "session_id=%s", + "Failed to cancel voice turn 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: + # 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: - await stt.abort_session() + if pipeline is not None: + pipeline.close_session() + elif session is not None: + session.close() except Exception: logger.exception( - "Failed to abort STT session. session_id=%s", + "Failed to close voice 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 - + # 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 diff --git a/uv.lock b/uv.lock index 173ce6b..ed41543 100644 --- a/uv.lock +++ b/uv.lock @@ -2924,6 +2924,7 @@ name = "voice-assistant" version = "0.1.0" source = { editable = "." } dependencies = [ + { name = "anyio" }, { name = "deepagents" }, { name = "fastapi" }, { name = "httpx" }, @@ -2946,6 +2947,7 @@ dev = [ [package.metadata] requires-dist = [ + { name = "anyio", specifier = ">=4.15.1,<5" }, { name = "deepagents", specifier = ">=0.7.26" }, { name = "fastapi", specifier = ">=0.141.1" }, { name = "httpx", specifier = ">=0.28,<1" },