From 19a415daaeea10e29a3772813f5d208b9e93274d Mon Sep 17 00:00:00 2001 From: Alex Alves Date: Tue, 13 Oct 2026 19:49:27 +0000 Subject: [PATCH] feat: integrate OCI agent streaming and web voice client --- .gitignore | 1 + README.md | 209 ++++++++++ config/providers.yaml | 6 + src/voice_assistant/agent/oci.py | 8 +- src/voice_assistant/agent/oci_connection.py | 107 +++++ src/voice_assistant/agent/text_chunker.py | 85 ++++ src/voice_assistant/audio_gateway/app/main.py | 93 ++++- .../audio_gateway/app/protocol.py | 14 + src/voice_assistant/audio_gateway/web/app.js | 387 ++++++++++++++++++ .../audio_gateway/web/index.html | 83 ++++ .../audio_gateway/web/pcm-worklet.js | 74 ++++ .../audio_gateway/web/styles.css | 164 ++++++++ src/voice_assistant/voice_pipeline.py | 196 +++++++++ tests/test_gateway.py | 102 +++++ tests/test_oci_connection.py | 132 ++++++ tests/test_protocol.py | 32 +- tests/test_text_chunker.py | 58 +++ tests/test_voice_pipeline.py | 274 ++++++++++++- tests/test_web_client.py | 32 ++ 19 files changed, 2034 insertions(+), 23 deletions(-) create mode 100644 src/voice_assistant/agent/oci_connection.py create mode 100644 src/voice_assistant/agent/text_chunker.py create mode 100644 src/voice_assistant/audio_gateway/web/app.js create mode 100644 src/voice_assistant/audio_gateway/web/index.html create mode 100644 src/voice_assistant/audio_gateway/web/pcm-worklet.js create mode 100644 src/voice_assistant/audio_gateway/web/styles.css create mode 100644 tests/test_oci_connection.py create mode 100644 tests/test_text_chunker.py create mode 100644 tests/test_web_client.py diff --git a/.gitignore b/.gitignore index 8ffc57f..8aa2ae3 100644 --- a/.gitignore +++ b/.gitignore @@ -21,6 +21,7 @@ venv/ .ruff_cache/ .coverage htmlcov/ +config/oci-e2e.yaml # IDE and operating system .vscode/ diff --git a/README.md b/README.md index 8084cad..b1385d6 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,7 @@ A modular voice pipeline for streaming audio input, speech transcription, agent - [Testing](#testing) - [Project structure](#project-structure) - [Known limitations](#known-limitations) +- [Architecture Backbone & Abstraction Boundaries](#architecture-backbone--abstraction-boundaries) ## Architecture @@ -38,6 +39,212 @@ flowchart TB 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. +## Architecture Backbone & Abstraction Boundaries + +AgentVoice follows a **provider-independent architecture** built around explicit contracts, lifecycle ownership, and asynchronous streaming. + +The architecture separates five concerns: + +- **Transport:** WebSocket communication and protocol serialization. +- **Orchestration:** Coordination of STT, Agent, and TTS execution. +- **Lifecycle:** Session state, turn transitions, interruption, and task ownership. +- **Provider contracts:** Abstract interfaces defining component behavior. +- **Provider implementations:** Concrete integrations with NVIDIA, OCI Generative AI, and VoxCPM. + +### Architectural Backbone + +```mermaid +flowchart TB + Client["Web Client / Audio Client"] + Gateway["Audio Gateway
FastAPI · WebSocket"] + Pipeline["VoicePipeline
Orchestration & Streaming"] + Session["SessionState
Session Lifecycle"] + Turns["TurnManager
Semantic Turn Lifecycle"] + + subgraph Contracts["Provider Abstraction Layer"] + direction LR + STTA["STTAdapter
Abstract Contract"] + AgentA["AgentAdapter
Abstract Contract"] + TTSA["TTSAdapter
Abstract Contract"] + end + + subgraph Implementations["Provider Implementation Layer"] + direction LR + Nvidia["NvidiaSTTAdapter"] + OCI["OCIAgentAdapter"] + Mock["MockAgentAdapter"] + Vox["VoxCPMTTSAdapter"] + end + + subgraph External["External AI Services"] + ASR["NVIDIA ASR"] + LLM["OCI Generative AI"] + TTS["VoxCPM"] + end + + Client <-->|"JSON Events + Binary PCM"| Gateway + Gateway --> Pipeline + + Pipeline --- Session + Pipeline --- Turns + + Pipeline --> STTA + Pipeline --> AgentA + Pipeline --> TTSA + + STTA -.->|implements| Nvidia + AgentA -.->|implements| OCI + AgentA -.->|implements| Mock + TTSA -.->|implements| Vox + + Nvidia --> ASR + OCI --> LLM + Vox --> TTS +``` + +The `VoicePipeline` depends on provider contracts rather than concrete implementations. Provider selection is performed through configuration and dynamic factories. + +This allows implementations to evolve without requiring the Audio Gateway to understand provider-specific APIs. + +### Provider Contracts + +| Abstraction | Contract | Implementations | Responsibility | +|---|---|---|---| +| Speech-to-text | `STTAdapter` | `NvidiaSTTAdapter`, `MockSTTAdapter` | Audio ingestion and transcript events | +| Agent | `AgentAdapter` | `OCIAgentAdapter`, `MockAgentAdapter` | Text processing and conversational responses | +| Text-to-speech | `TTSAdapter` | `VoxCPMTTSAdapter`, `MockTTSAdapter` | Speech synthesis and audio streaming | + +The `AgentAdapter` defines the common text-processing interface. OCI-specific incremental generation is currently exposed through an additional `stream()` method, with streaming capability detected by the orchestration layer. + +### Runtime Composition + +Provider implementations are selected through environment configuration and resolved by the provider factory. + +```mermaid +flowchart LR + Config["Environment YAML"] + Catalog["Provider Catalog"] + Factory["Provider Factories"] + Contracts["Adapter Contracts"] + Runtime["VoicePipeline"] + + Config --> Factory + Catalog --> Factory + Factory -->|Validate implementation| Contracts + Factory -->|Instantiate providers| Runtime +``` + +The primary configuration components are: + +| Component | Location | +|---|---| +| Environment selection | `config/loader.py` | +| Typed configuration | `config/models.py` | +| Provider catalog | `config/providers.yaml` | +| Dynamic provider loading | `providers.py` | +| Session lifecycle | `audio_gateway/app/session.py` | +| Semantic turn lifecycle | `turn_manager.py` | +| Execution orchestration | `voice_pipeline.py` | + +### Streaming Execution Backbone + +The streaming pipeline allows OCI Generative AI to continue producing text while VoxCPM synthesizes previously completed text segments. + +```mermaid +flowchart TB + Input["Microphone / PCM Input"] + Gateway["Audio Gateway"] + STT["NVIDIA STT"] + Transcript["Final Transcript"] + + subgraph Pipeline["VoicePipeline · Concurrent Execution"] + direction TB + Agent["OCIAgentAdapter.stream()"] + Bridge["OCI SSE Async Bridge"] + Chunker["TextChunker"] + Queue["Bounded Segment Queue"] + Synth["VoxCPM TTS Streaming"] + + Agent --> Bridge + Bridge --> Chunker + Chunker --> Queue + Queue --> Synth + end + + OCI["OCI Generative AI"] + Vox["VoxCPM Service"] + Output["WebSocket PCM Output"] + + Input --> Gateway + Gateway --> STT + STT --> Transcript + Transcript --> Agent + + Agent <--> OCI + Synth <--> Vox + Synth --> Gateway + Gateway --> Output +``` + +The pipeline uses bounded queues and asynchronous task coordination to support incremental generation, ordered synthesis, and cooperative cancellation. + +### Lifecycle Ownership + +AgentVoice separates semantic state from asynchronous execution. + +```mermaid +flowchart LR + Listening["LISTENING"] + Processing["PROCESSING"] + Speaking["SPEAKING"] + Completed["COMPLETED"] + Interrupted["INTERRUPTED"] + + Listening -->|"STT final"| Processing + Processing -->|"First TTS segment"| Speaking + Speaking -->|"Synthesis complete"| Completed + + Listening --> Interrupted + Processing --> Interrupted + Speaking --> Interrupted +``` + +**Ownership responsibilities:** + +| Component | Owns | +|---|---| +| Audio Gateway | WebSocket connection, protocol messages, provider cleanup coordination | +| SessionState | Session lifecycle and audio sequence validation | +| TurnManager | Semantic turn state transitions | +| VoicePipeline | Turn execution task and Agent/TTS orchestration | +| OCI streaming bridge | SSE reader, event queue, and stream closure | +| Provider adapters | Provider-specific requests, connections, and data formats | + +Cancellation is cooperative. The system requests cancellation and waits for task termination, but termination of blocking external operations depends on the underlying provider and transport behavior. + +### Architectural Extension Points + +The abstraction layer allows additional implementations to be introduced through the provider catalog. + +For example, a future LiteLLM integration could implement the Agent contract without requiring a redesign of the Gateway or the semantic turn lifecycle. + +```mermaid +flowchart TB + Pipeline["VoicePipeline"] + Contract["AgentAdapter"] + + OCI["OCIAgentAdapter
Current"] + Mock["MockAgentAdapter
Testing"] + Lite["LiteLLMAgentAdapter
Future"] + + Pipeline --> Contract + Contract -.-> OCI + Contract -.-> Mock + Contract -.-> Lite +``` + +**Design principle:** transport, orchestration, lifecycle, and provider-specific integrations remain separate concerns. Provider independence is established through contracts and factories, while streaming and cancellation behavior are validated at their respective integration boundaries. + ## Quickstart ### Requirements @@ -197,3 +404,5 @@ tools/ Integration and smoke scripts --- **Status:** V1 development baseline · Python 3.12+ · WebSocket protocol 1.0 + + diff --git a/config/providers.yaml b/config/providers.yaml index c654c92..cea6747 100644 --- a/config/providers.yaml +++ b/config/providers.yaml @@ -27,3 +27,9 @@ providers: realtime: false streaming: false + oci: + name: "OCI Generative AI" + implementation: "voice_assistant.agent.oci:OCIAgentAdapter" + capabilities: + realtime: true + streaming: true \ No newline at end of file diff --git a/src/voice_assistant/agent/oci.py b/src/voice_assistant/agent/oci.py index d586777..b68305c 100644 --- a/src/voice_assistant/agent/oci.py +++ b/src/voice_assistant/agent/oci.py @@ -10,7 +10,6 @@ from __future__ import annotations import json from typing import Any -import asyncio import logging from collections.abc import AsyncIterator from contextlib import aclosing @@ -19,6 +18,7 @@ import oci from voice_assistant.agent.adapter import AgentAdapter from voice_assistant.agent.oci_stream import stream_sse_events +from voice_assistant.agent.oci_connection import open_oci_stream def extract_text_delta(raw_data: str) -> str | None: @@ -205,9 +205,8 @@ class OCIAgentAdapter(AgentAdapter): self.model_id, ) - response = await asyncio.to_thread( - self.client.chat, - details, + response = await open_oci_stream( + lambda: self.client.chat(details) ) if response.status != 200: @@ -256,4 +255,3 @@ class OCIAgentAdapter(AgentAdapter): ) return response_text - diff --git a/src/voice_assistant/agent/oci_connection.py b/src/voice_assistant/agent/oci_connection.py new file mode 100644 index 0000000..35514d7 --- /dev/null +++ b/src/voice_assistant/agent/oci_connection.py @@ -0,0 +1,107 @@ +""" +Lifecycle management for blocking OCI HTTP connection setup. + +Ensures that responses arriving after coroutine cancellation +are closed instead of being abandoned. +""" + +from __future__ import annotations + +import asyncio +import logging +import threading +from typing import Any, Callable + + +logger = logging.getLogger(__name__) + + +def _close_response(response: Any) -> None: + """Close the SSE response returned by the OCI SDK.""" + stream = getattr(response, "data", None) + + if stream is not None: + close = getattr(stream, "close", None) + + if callable(close): + close() + + +async def open_oci_stream( + open_connection: Callable[[], Any], +) -> Any: + """ + Open a blocking OCI connection with explicit response ownership. + + The worker retains ownership until the async caller claims + the response. Cancellation closes any unclaimed response. + + A blocking SDK call may continue after cancellation. + """ + ownership_lock = threading.Lock() + abandoned = False + claimed = False + response = None + + def worker() -> Any: + nonlocal response + + opened = open_connection() + + with ownership_lock: + if abandoned: + should_close = True + else: + response = opened + should_close = False + + if should_close: + try: + _close_response(opened) + except Exception: + logger.exception( + "Failed to close abandoned OCI response." + ) + + return opened + + task = asyncio.create_task( + asyncio.to_thread(worker) + ) + + try: + opened = await asyncio.shield(task) + + with ownership_lock: + if abandoned: + raise RuntimeError( + "OCI connection was abandoned." + ) + + claimed = True + + return opened + + except asyncio.CancelledError: + with ownership_lock: + abandoned = True + + if not claimed: + unclaimed = response + response = None + else: + unclaimed = None + + if unclaimed is not None: + try: + await asyncio.to_thread( + _close_response, + unclaimed, + ) + except Exception: + logger.exception( + "Failed to close unclaimed OCI response." + ) + + raise + diff --git a/src/voice_assistant/agent/text_chunker.py b/src/voice_assistant/agent/text_chunker.py new file mode 100644 index 0000000..26b7d33 --- /dev/null +++ b/src/voice_assistant/agent/text_chunker.py @@ -0,0 +1,85 @@ +""" +Incremental text segmentation for streaming voice synthesis. + +Converts LLM text deltas into speech-ready segments while +preserving word boundaries and punctuation. +""" + +from __future__ import annotations + + +class TextChunker: + """ + Accumulate text deltas and emit segments suitable for TTS. + + Segments are emitted on sentence-ending punctuation or + at a configurable maximum length. + """ + + def __init__( + self, + min_chars: int = 20, + max_chars: int = 160, + ) -> None: + if min_chars < 1: + raise ValueError("min_chars must be positive.") + + if max_chars < min_chars: + raise ValueError( + "max_chars must be greater than or equal to min_chars." + ) + + self.min_chars = min_chars + self.max_chars = max_chars + self._buffer = "" + + def feed(self, delta: str) -> list[str]: + """Accept a text delta and return completed segments.""" + self._buffer += delta + segments: list[str] = [] + + while len(self._buffer) >= self.min_chars: + boundary = self._find_boundary() + + if boundary is None: + break + + segment = self._buffer[:boundary].strip() + self._buffer = self._buffer[boundary:] + + if segment: + segments.append(segment) + + return segments + + def _find_boundary(self) -> int | None: + """Find a sentence boundary or safe length boundary.""" + for index, char in enumerate(self._buffer): + if char in ".!?": + boundary = index + 1 + + if boundary >= self.min_chars: + return boundary + + if len(self._buffer) < self.max_chars: + return None + + # Prefer breaking at a word boundary. + boundary = self._buffer.rfind( + " ", + 0, + self.max_chars + 1, + ) + + if boundary >= self.min_chars: + return boundary + 1 + + return self.max_chars + + def flush(self) -> str | None: + """Return any remaining text after generation completes.""" + remaining = self._buffer.strip() + self._buffer = "" + + return remaining or None + diff --git a/src/voice_assistant/audio_gateway/app/main.py b/src/voice_assistant/audio_gateway/app/main.py index 98439e6..ecfa135 100644 --- a/src/voice_assistant/audio_gateway/app/main.py +++ b/src/voice_assistant/audio_gateway/app/main.py @@ -34,8 +34,13 @@ Architecture: import json import logging +from pathlib import Path + +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 @@ -83,8 +88,16 @@ from voice_assistant.audio_gateway.app.protocol import ( STTPartial, TurnEnded, TurnStarted, + AgentDelta, ) +from voice_assistant.voice_pipeline import ( + PipelineTextDelta, + PipelineAgentResponse, + PipelineAudioChunk, +) + + from voice_assistant.audio_gateway.app.session import SessionState # logger configuration @@ -100,6 +113,18 @@ provider_catalog = load_provider_catalog() MAX_AUDIO_CHUNK_BYTES = 64 * 1024 # 64 KB +WEB_DIR = Path(__file__).resolve().parents[1] / "web" + +app.mount( + "/web", + StaticFiles(directory=WEB_DIR), + name="web", +) + + +@app.get("/ui", include_in_schema=False) +async def web_client() -> FileResponse: + return FileResponse(WEB_DIR / "index.html") def create_turn_manager() -> TurnManager: @@ -176,27 +201,63 @@ async def forward_stt_events( ), ) - result = await pipeline.process_transcript( - 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 - await send_json( - websocket, - AgentResponse( + 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=result.response_text, - ), - ) + text=event.text, + ) - async for audio_chunk in pipeline.synthesize_response( - result.response_text, - ): - await websocket.send_bytes(audio_chunk) + await send_json( + websocket, + AgentResponse( + session_id=session_id, + turn_id=turn_id, + text=result.response_text, + ), + ) - pipeline.on_tts_complete(turn_id) + async for audio_chunk in pipeline.synthesize_response( + result.response_text, + ): + await websocket.send_bytes(audio_chunk) + + pipeline.on_tts_complete(turn_id) diff --git a/src/voice_assistant/audio_gateway/app/protocol.py b/src/voice_assistant/audio_gateway/app/protocol.py index bb3adda..68effc8 100644 --- a/src/voice_assistant/audio_gateway/app/protocol.py +++ b/src/voice_assistant/audio_gateway/app/protocol.py @@ -109,6 +109,20 @@ class STTFinal(BaseModel): text: str +class AgentDelta(BaseModel): + """ + Incremental text generated by the configured Agent. + + Each event belongs to one session and turn. + Text fragments must be concatenated in arrival order. + """ + + type: Literal["agent.delta"] = "agent.delta" + session_id: str + turn_id: int + text: str + + class AgentResponse(BaseModel): """Text response produced by the configured Agent.""" diff --git a/src/voice_assistant/audio_gateway/web/app.js b/src/voice_assistant/audio_gateway/web/app.js new file mode 100644 index 0000000..35c7852 --- /dev/null +++ b/src/voice_assistant/audio_gateway/web/app.js @@ -0,0 +1,387 @@ + +const $ = (id) => document.getElementById(id); + +const status = $("status"); +const talkButton = $("talkButton"); +const talkLabel = $("talkLabel"); +const connectButton = $("connectButton"); +const interruptButton = $("interruptButton"); +const messages = $("messages"); + +let ws = null; +let sessionId = null; +let turnId = null; +let sequence = 0; +let phase = "disconnected"; + +let microphone = null; +let inputContext = null; +let inputSource = null; +let inputNode = null; +let silentGain = null; + +let playbackContext = null; +let playbackTime = 0; +const scheduledAudio = new Set(); + +let userMessage = null; +let agentMessage = null; +let pendingFinal = null; + +function setPhase(next) { + phase = next; + + status.textContent = { + disconnected: "Desconectado", + connecting: "Conectando", + ready: "Conectado", + starting: "Iniciando turno", + recording: "Gravando", + processing: "Processando", + speaking: "Respondendo" + }[next] || next; + + status.classList.toggle( + "connected", + next !== "disconnected" && next !== "connecting" + ); + + connectButton.disabled = next !== "disconnected"; + talkButton.disabled = next !== "ready" && next !== "recording"; + interruptButton.disabled = turnId === null; + + talkButton.classList.toggle("recording", next === "recording"); + talkLabel.textContent = next === "recording" + ? "Clique para terminar de falar" + : next === "ready" + ? "Clique para falar" + : status.textContent; +} + +function addMessage(role, text = "") { + $("emptyState")?.remove(); + + const element = document.createElement("div"); + element.className = `message ${role}`; + + const label = document.createElement("small"); + label.textContent = role === "user" ? "Você" : "AgentVoice"; + + const content = document.createElement("span"); + content.textContent = text; + + element.append(label, content); + messages.appendChild(element); + messages.scrollTop = messages.scrollHeight; + + return content; +} + +function sendJson(payload) { + if (ws?.readyState !== WebSocket.OPEN) { + throw new Error("WebSocket não está conectado."); + } + ws.send(JSON.stringify(payload)); +} + +async function ensurePlayback() { + if (!playbackContext) { + playbackContext = new AudioContext(); + } + await playbackContext.resume(); +} + +async function playPCM(bytes) { + await ensurePlayback(); + + const rate = Number($("outputRate").value); + const sampleCount = Math.floor(bytes.byteLength / 2); + if (!sampleCount) return; + + const view = new DataView(bytes); + const samples = new Float32Array(sampleCount); + + for (let i = 0; i < sampleCount; i++) { + samples[i] = view.getInt16(i * 2, true) / 32768; + } + + const audio = playbackContext.createBuffer( + 1, + sampleCount, + rate + ); + audio.copyToChannel(samples, 0); + + const source = playbackContext.createBufferSource(); + source.buffer = audio; + source.connect(playbackContext.destination); + + const startAt = Math.max( + playbackContext.currentTime + 0.025, + playbackTime + ); + + source.start(startAt); + playbackTime = startAt + audio.duration; + + scheduledAudio.add(source); + source.onended = () => scheduledAudio.delete(source); +} + +function stopPlayback() { + for (const source of scheduledAudio) { + try { source.stop(); } catch {} + } + scheduledAudio.clear(); + playbackTime = 0; +} + +async function stopMicrophone() { + if (inputNode) { + inputNode.port.onmessage = null; + inputNode.disconnect(); + inputNode = null; + } + + inputSource?.disconnect(); + silentGain?.disconnect(); + + inputSource = null; + silentGain = null; + + microphone?.getTracks().forEach((track) => track.stop()); + microphone = null; + + if (inputContext) { + const context = inputContext; + inputContext = null; + await context.close(); + } +} + +async function startMicrophone() { + microphone = await navigator.mediaDevices.getUserMedia({ + audio: { + channelCount: 1, + echoCancellation: true, + noiseSuppression: true, + autoGainControl: true + } + }); + + inputContext = new AudioContext(); + await inputContext.audioWorklet.addModule("/web/pcm-worklet.js"); + + inputSource = inputContext.createMediaStreamSource(microphone); + inputNode = new AudioWorkletNode(inputContext, "pcm-input"); + + inputNode.port.onmessage = ({ data }) => { + if ( + data.type !== "pcm" || + phase !== "recording" || + turnId === null + ) return; + + if (ws?.readyState !== WebSocket.OPEN) return; + + sendJson({ + type: "audio.chunk", + turn_id: turnId, + sequence_number: sequence++ + }); + + ws.send(data.buffer); + }; + + silentGain = inputContext.createGain(); + silentGain.gain.value = 0; + + inputSource.connect(inputNode); + inputNode.connect(silentGain); + silentGain.connect(inputContext.destination); + + await inputContext.resume(); +} + +async function beginTurn() { + if (phase !== "ready") return; + + try { + await ensurePlayback(); + stopPlayback(); + + setPhase("starting"); + sendJson({ type: "audio.start" }); + } catch (error) { + console.error(error); + setPhase("ready"); + } +} + +async function endRecording() { + if (phase !== "recording") return; + + const activeTurn = turnId; + setPhase("processing"); + + await stopMicrophone(); + + if (activeTurn !== null && ws?.readyState === WebSocket.OPEN) { + sendJson({ type: "audio.stop", turn_id: activeTurn }); + } +} + +async function interruptTurn() { + if (turnId === null) return; + + const activeTurn = turnId; + + setPhase("processing"); + await stopMicrophone(); + stopPlayback(); + + if (ws?.readyState === WebSocket.OPEN) { + sendJson({ type: "interrupt", turn_id: activeTurn }); + } +} + +async function handleEvent(event) { + switch (event.type) { + case "session.ready": + sessionId = event.session_id; + setPhase("ready"); + break; + + case "turn.started": + turnId = event.turn_id; + sequence = 0; + userMessage = addMessage("user"); + agentMessage = addMessage("agent"); + pendingFinal = null; + + try { + await startMicrophone(); + setPhase("recording"); + } catch (error) { + console.error(error); + talkLabel.textContent = "Não foi possível acessar o microfone."; + await interruptTurn(); + } + break; + + case "stt.partial": + if (event.turn_id === turnId && userMessage) { + userMessage.textContent = event.text; + } + break; + + case "stt.final": + if (event.turn_id === turnId && userMessage) { + userMessage.textContent = event.text; + } + break; + + case "agent.delta": + if (event.turn_id === turnId && agentMessage) { + agentMessage.textContent += event.text; + } + break; + + case "agent.response": + if (event.turn_id === turnId && agentMessage) { + agentMessage.textContent = event.text; + } + break; + + case "turn.ended": + if (event.turn_id === turnId) { + turnId = null; + await stopMicrophone(); + setPhase("ready"); + } + break; + + case "interrupted": + if (event.turn_id === turnId) { + turnId = null; + await stopMicrophone(); + stopPlayback(); + setPhase("ready"); + } + break; + + case "error": + console.error("Gateway error:", event); + talkLabel.textContent = + `Erro: ${event.code} — ${event.message}`; + + if (phase === "starting") { + setPhase("ready"); + } else if (turnId !== null && phase !== "disconnected") { + await interruptTurn(); + } + break; + + case "session.ended": + ws?.close(); + break; + } +} + +async function connect() { + if (ws && ws.readyState !== WebSocket.CLOSED) return; + + setPhase("connecting"); + + const scheme = location.protocol === "https:" ? "wss" : "ws"; + ws = new WebSocket(`${scheme}://${location.host}/ws/audio`); + ws.binaryType = "arraybuffer"; + + // Serialize event handling to preserve ordering. + let incoming = Promise.resolve(); + + ws.onopen = () => { + sendJson({ + type: "session.start", + audio: { + encoding: "pcm_s16le", + sample_rate: 16000, + channels: 1 + } + }); + }; + + ws.onmessage = ({ data }) => { + incoming = incoming.then(async () => { + if (data instanceof ArrayBuffer) { + if (turnId !== null) { + await playPCM(data); + } + } else { + await handleEvent(JSON.parse(data)); + } + }).catch(console.error); + }; + + ws.onerror = (error) => console.error("WebSocket error", error); + + ws.onclose = async () => { + await stopMicrophone(); + stopPlayback(); + ws = null; + sessionId = null; + turnId = null; + setPhase("disconnected"); + }; +} + +connectButton.addEventListener("click", connect); + +talkButton.addEventListener("click", async () => { + if (phase === "ready") await beginTurn(); + else if (phase === "recording") await endRecording(); +}); + +interruptButton.addEventListener("click", interruptTurn); + +setPhase("disconnected"); diff --git a/src/voice_assistant/audio_gateway/web/index.html b/src/voice_assistant/audio_gateway/web/index.html new file mode 100644 index 0000000..26e6799 --- /dev/null +++ b/src/voice_assistant/audio_gateway/web/index.html @@ -0,0 +1,83 @@ + + + + + + + AgentVoice · Voice Console + + + +
+
+
+
◉
+
+

AgentVoice

+

Realtime Voice Console

+
+
+ Desconectado +
+ +
+ VOICE ASSISTANT · V1 +

Uma conversa, naturalmente.

+

Fale pelo microfone e receba respostas + geradas por inteligência artificial.

+
+ +
+
+
◎
+

Sua conversa aparecerá aqui.

+
+
+ +
+
+ +

Conecte para começar

+
+ +
+ + +
+ +
+ + +
+
+ +
+ NVIDIA STT + OCI Generative AI + VoxCPM TTS +
+
+ + + + diff --git a/src/voice_assistant/audio_gateway/web/pcm-worklet.js b/src/voice_assistant/audio_gateway/web/pcm-worklet.js new file mode 100644 index 0000000..aea2fb2 --- /dev/null +++ b/src/voice_assistant/audio_gateway/web/pcm-worklet.js @@ -0,0 +1,74 @@ + +class PCMInputProcessor extends AudioWorkletProcessor { + constructor() { + super(); + this.targetRate = 16000; + this.step = sampleRate / this.targetRate; + this.buffer = []; + this.position = 0; + this.sequence = 0; + this.chunkSamples = 1600; + this.output = []; + } + + process(inputs) { + const channels = inputs[0]; + + if (!channels || channels.length === 0) { + return true; + } + + const frameLength = channels[0].length; + + for (let i = 0; i < frameLength; i++) { + let sample = 0; + + for (const channel of channels) { + sample += channel[i]; + } + + this.buffer.push(sample / channels.length); + } + + while (this.position + 1 < this.buffer.length) { + const index = Math.floor(this.position); + const fraction = this.position - index; + + const sample = + this.buffer[index] * (1 - fraction) + + this.buffer[index + 1] * fraction; + + const clipped = Math.max(-1, Math.min(1, sample)); + + this.output.push( + clipped < 0 + ? Math.round(clipped * 32768) + : Math.round(clipped * 32767) + ); + + this.position += this.step; + + if (this.output.length >= this.chunkSamples) { + const pcm = new Int16Array(this.output); + this.output = []; + + this.port.postMessage({ + type: "pcm", + sequence: this.sequence++, + buffer: pcm.buffer + }, [pcm.buffer]); + } + } + + const consumed = Math.floor(this.position); + + if (consumed > 0) { + this.buffer.splice(0, consumed); + this.position -= consumed; + } + + return true; + } +} + +registerProcessor("pcm-input", PCMInputProcessor); diff --git a/src/voice_assistant/audio_gateway/web/styles.css b/src/voice_assistant/audio_gateway/web/styles.css new file mode 100644 index 0000000..4adf0b9 --- /dev/null +++ b/src/voice_assistant/audio_gateway/web/styles.css @@ -0,0 +1,164 @@ + +:root { + font-family: Inter, system-ui, -apple-system, sans-serif; + color: #e8edf6; + background: #090e18; + font-synthesis: none; +} + +* { box-sizing: border-box; } + +body { + margin: 0; + min-height: 100vh; + display: grid; + place-items: center; + padding: 24px; + background: + radial-gradient(ellipse at 50% 0%, #17294b, transparent 65%), + #090e18; +} + +.console { + width: min(100%, 820px); + background: #111a2a; + border: 1px solid #26334a; + border-radius: 22px; + overflow: hidden; + box-shadow: 0 30px 90px #0006; +} + +.header, .brand, .actions, footer { + display: flex; + align-items: center; +} + +.header { + justify-content: space-between; + padding: 22px 28px; + border-bottom: 1px solid #26334a; +} + +.brand { gap: 12px; } +.brand-icon { color: #67e8f9; font-size: 34px; } +h1 { font-size: 19px; margin: 0; } +.brand p, .hero p, .empty-state, footer { + color: #93a4bd; +} +.brand p { margin: 3px 0 0; font-size: 12px; } + +.status { + font-size: 12px; + border: 1px solid #42506a; + border-radius: 20px; + padding: 7px 12px; +} + +.status.connected { + color: #86efac; + border-color: #166534; +} + +.hero { text-align: center; padding: 36px 24px 24px; } +.eyebrow { + color: #67e8f9; + font-size: 11px; + letter-spacing: 2px; +} +h2 { font-size: clamp(23px, 4vw, 32px); margin: 14px 0; } +.hero p { font-size: 14px; line-height: 1.6; } + +.messages { + height: 310px; + overflow-y: auto; + padding: 24px; + background: #0c1422; + border-block: 1px solid #26334a; +} + +.empty-state { text-align: center; padding: 70px 0; } +.empty-icon { font-size: 40px; color: #526782; } + +.message { + margin: 12px 0; + padding: 14px 18px; + border-radius: 14px; + max-width: 88%; + line-height: 1.55; + font-size: 14px; + white-space: pre-wrap; + overflow-wrap: anywhere; +} + +.message.user { background: #20385b; margin-left: auto; } +.message.agent { background: #1b2b36; margin-right: auto; } +.message small { + display: block; + color: #91a8c5; + font-size: 11px; + margin-bottom: 5px; +} + +.controls { padding: 24px; text-align: center; } +.mic-container { padding: 8px 0 20px; } + +.talk-button { + width: 88px; + height: 88px; + border-radius: 50%; + border: 1px solid #3a6a9b; + background: linear-gradient(145deg, #23598c, #183755); + color: white; + font-size: 30px; + cursor: pointer; + transition: transform .15s, box-shadow .15s; +} + +.talk-button:hover:not(:disabled) { transform: scale(1.04); } +.talk-button.recording { + background: #9f2442; + border-color: #fb7185; + box-shadow: 0 0 0 12px #fb71851a; +} + +#talkLabel { color: #a4b4ca; font-size: 13px; } +.actions { justify-content: center; gap: 12px; } + +.actions button, select { + padding: 11px 20px; + border-radius: 9px; + border: 1px solid #344862; + color: #e8edf6; + background: #1b2b42; + cursor: pointer; +} + +.actions .primary { background: #2563eb; border-color: #2563eb; } +button:disabled { opacity: .45; cursor: not-allowed; } + +.settings { + display: flex; + justify-content: center; + align-items: center; + flex-wrap: wrap; + gap: 12px; + margin-top: 22px; + font-size: 12px; + color: #93a4bd; +} + +footer { + justify-content: center; + flex-wrap: wrap; + gap: 24px; + border-top: 1px solid #26334a; + padding: 16px; + font-size: 11px; +} + +@media (max-width: 520px) { + body { padding: 0; } + .console { min-height: 100vh; border-radius: 0; } + .header { padding: 18px; } + .messages { height: 36vh; } +} diff --git a/src/voice_assistant/voice_pipeline.py b/src/voice_assistant/voice_pipeline.py index 1041f5b..54d69ae 100644 --- a/src/voice_assistant/voice_pipeline.py +++ b/src/voice_assistant/voice_pipeline.py @@ -41,7 +41,10 @@ import asyncio from collections.abc import AsyncIterator from dataclasses import dataclass +from contextlib import aclosing +from voice_assistant.agent.text_chunker import TextChunker +from voice_assistant.turn_manager import TurnState from voice_assistant.agent.adapter import AgentAdapter from voice_assistant.audio_gateway.app.session import SessionState from voice_assistant.turn_manager import Turn, TurnManager @@ -64,6 +67,21 @@ class PipelineResult: response_text: str +@dataclass(slots=True) +class PipelineTextDelta: + text: str + + +@dataclass(slots=True) +class PipelineAgentResponse: + text: str + + +@dataclass(slots=True) +class PipelineAudioChunk: + audio: bytes + + class VoicePipeline: """ Coordinate session, turn, Agent, and TTS processing. @@ -168,6 +186,184 @@ class VoicePipeline: ): yield audio_chunk + + async def stream_turn_response( + self, + session_id: str, + turn_id: int, + text: str, + ) -> AsyncIterator[ + PipelineTextDelta | PipelineAgentResponse | PipelineAudioChunk + ]: + """ + Stream Agent text and TTS audio concurrently. + + The producer generates text segments while the consumer + synthesizes completed segments in their original order. + """ + turn = self.turn_manager.active_turn + + if turn is None or turn.turn_id != turn_id: + raise ValueError(f"Unexpected turn_id={turn_id}.") + + if turn.state != TurnState.PROCESSING: + raise RuntimeError( + f"Cannot stream response while turn is {turn.state.value}." + ) + + stream_method = getattr(self.agent, "stream", None) + + if not callable(stream_method): + raise RuntimeError( + "Configured Agent does not support streaming." + ) + + events: asyncio.Queue[ + PipelineTextDelta | PipelineAgentResponse | + PipelineAudioChunk | None + ] = asyncio.Queue(maxsize=16) + + segments: asyncio.Queue[str | None] = asyncio.Queue(maxsize=4) + + async def produce() -> None: + chunker = TextChunker() + parts: list[str] = [] + + async with aclosing( + stream_method( + session_id=session_id, + turn_id=turn_id, + text=text, + ) + ) as deltas: + async for delta in deltas: + parts.append(delta) + + await events.put( + PipelineTextDelta(text=delta) + ) + + for segment in chunker.feed(delta): + await segments.put(segment) + + remaining = chunker.flush() + + if remaining: + await segments.put(remaining) + + response_text = "".join(parts).strip() + + if not response_text: + raise RuntimeError( + "Agent returned an empty streaming response." + ) + + await events.put( + PipelineAgentResponse(text=response_text) + ) + + await segments.put(None) + + async def consume() -> None: + first_segment = True + + while True: + segment = await segments.get() + + if segment is None: + break + + if first_segment: + self.turn_manager.on_agent_response(turn_id) + first_segment = False + + async with aclosing( + self.tts.synthesize(segment) + ) as audio_stream: + async for audio_chunk in audio_stream: + await events.put( + PipelineAudioChunk(audio=audio_chunk) + ) + + if first_segment: + raise RuntimeError( + "No text segments were available for TTS." + ) + + self.turn_manager.on_tts_complete(turn_id) + + async def run_pipeline() -> None: + async with asyncio.TaskGroup() as group: + group.create_task(produce()) + group.create_task(consume()) + + execution = asyncio.create_task(run_pipeline()) + + pending_get: asyncio.Task | None = None + + try: + while True: + if execution.done() and events.empty(): + await execution + break + + pending_get = asyncio.create_task(events.get()) + + done, _ = await asyncio.wait( + {pending_get, execution}, + return_when=asyncio.FIRST_COMPLETED, + ) + + if pending_get in done: + event = pending_get.result() + pending_get = None + + if event is not None: + yield event + + continue + + # Supervisor completed before another event arrived. + # Cancel the pending queue read before inspecting errors. + pending_get.cancel() + + await asyncio.gather( + pending_get, + return_exceptions=True, + ) + + pending_get = None + + # Propagate failures, including ExceptionGroup. + await execution + + # Normal completion: drain remaining events. + while not events.empty(): + event = events.get_nowait() + + if event is not None: + yield event + + break + + finally: + if pending_get is not None: + pending_get.cancel() + + await asyncio.gather( + pending_get, + return_exceptions=True, + ) + + if not execution.done(): + execution.cancel() + + await asyncio.gather( + execution, + return_exceptions=True, + ) + + def complete_turn( self, turn_id: int, diff --git a/tests/test_gateway.py b/tests/test_gateway.py index 24004bf..4c75a70 100644 --- a/tests/test_gateway.py +++ b/tests/test_gateway.py @@ -1698,3 +1698,105 @@ def test_websocket_registration_failure_awaits_task_before_abort(): assert len(tasks) == 1 assert tasks[0].done() assert abort_checked + + +def test_websocket_streaming_agent_and_tts(): + """ + Verify streaming Agent text and TTS audio through WebSocket. + + The first audio segment must be delivered before the + Agent's complete response. + """ + created_stt.clear() + + class StreamingTestAgent: + async def stream(self, session_id, turn_id, text): + yield "Olá, como posso ajudar?" + yield " Estou à disposição." + + class StreamingTestTTS: + async def synthesize(self, text): + yield text.encode("utf-8") + + with patch.object( + main, + "create_agent_adapter", + return_value=StreamingTestAgent(), + ): + with patch.object( + main, + "create_tts_adapter", + return_value=StreamingTestTTS(), + ): + with client.websocket_connect("/ws/audio") as websocket: + websocket.send_json( + { + "type": "session.start", + "audio": { + "encoding": AUDIO_ENCODING, + "sample_rate": AUDIO_SAMPLE_RATE, + "channels": AUDIO_CHANNELS, + }, + } + ) + + session_ready = websocket.receive_json() + assert session_ready["type"] == "session.ready" + + websocket.send_json({"type": "audio.start"}) + + turn_started = websocket.receive_json() + assert turn_started["type"] == "turn.started" + + turn_id = turn_started["turn_id"] + stt = created_stt[0] + + stt.emit_final_now("Olá") + + received_text = [] + received_audio = [] + received_complete_response = False + + # The streaming path emits two deltas and + # two TTS audio segments. + for _ in range(20): + message = websocket.receive() + + if message.get("bytes") is not None: + received_audio.append(message["bytes"]) + else: + import json + + event = json.loads(message["text"]) + + if event["type"] == "agent.delta": + received_text.append(event["text"]) + + elif event["type"] == "agent.response": + received_complete_response = True + + elif event["type"] == "error": + raise AssertionError( + f"Gateway returned error: {event}" + ) + + if received_complete_response and len(received_audio) >= 2: + break + + assert "".join(received_text) == ( + "Olá, como posso ajudar?" + " Estou à disposição." + ) + + assert received_audio + assert received_complete_response + + websocket.send_json( + { + "type": "audio.stop", + "turn_id": turn_id, + } + ) + + turn_ended = websocket.receive_json() + assert turn_ended["type"] == "turn.ended" diff --git a/tests/test_oci_connection.py b/tests/test_oci_connection.py new file mode 100644 index 0000000..f697619 --- /dev/null +++ b/tests/test_oci_connection.py @@ -0,0 +1,132 @@ +import asyncio +import threading +from types import SimpleNamespace + +import pytest + +from voice_assistant.agent.oci_connection import open_oci_stream + + +class FakeStream: + def __init__(self): + self.closed = False + + def close(self): + self.closed = True + + +@pytest.mark.anyio +async def test_open_oci_stream_success(): + stream = FakeStream() + + response = SimpleNamespace(data=stream) + + result = await open_oci_stream(lambda: response) + + assert result is response + assert stream.closed is False + + stream.close() + + +@pytest.mark.anyio +async def test_open_oci_stream_closes_late_response(): + started = threading.Event() + release = threading.Event() + finished = threading.Event() + + stream = FakeStream() + response = SimpleNamespace(data=stream) + + def open_connection(): + started.set() + release.wait(timeout=5.0) + + try: + return response + finally: + finished.set() + + task = asyncio.create_task( + open_oci_stream(open_connection) + ) + + try: + assert await asyncio.to_thread( + started.wait, + 1.0, + ) + + task.cancel() + + with pytest.raises(asyncio.CancelledError): + await task + + finally: + release.set() + + assert await asyncio.to_thread( + finished.wait, + 1.0, + ) + + # The worker may still be completing its cleanup. + for _ in range(100): + if stream.closed: + break + await asyncio.sleep(0.001) + + assert stream.closed is True + + +@pytest.mark.anyio +async def test_open_oci_stream_cancellation_during_handoff(): + """ + Verify that a response is not abandoned when cancellation + happens around completion of the opening operation. + """ + stream = FakeStream() + response = SimpleNamespace(data=stream) + + opened = threading.Event() + release = threading.Event() + + def open_connection(): + opened.set() + release.wait(timeout=5.0) + return response + + task = asyncio.create_task( + open_oci_stream(open_connection) + ) + + try: + assert await asyncio.to_thread( + opened.wait, + 1.0, + ) + + release.set() + task.cancel() + + with pytest.raises(asyncio.CancelledError): + await task + + for _ in range(100): + if stream.closed: + break + await asyncio.sleep(0.001) + + assert stream.closed is True + + finally: + release.set() + + if not task.done(): + task.cancel() + + await asyncio.gather( + task, + return_exceptions=True, + ) + diff --git a/tests/test_protocol.py b/tests/test_protocol.py index 011f78b..8d2f778 100644 --- a/tests/test_protocol.py +++ b/tests/test_protocol.py @@ -20,6 +20,7 @@ import pytest from pydantic import ValidationError from voice_assistant.audio.models import CANONICAL_AUDIO_FORMAT from voice_assistant.audio_gateway.app.protocol import ( + AgentDelta, AgentResponse, AudioChunk, AudioStart, @@ -38,6 +39,7 @@ from voice_assistant.audio_gateway.app.protocol import ( ) + @pytest.mark.parametrize( ("model", "expected_type"), [ @@ -101,6 +103,7 @@ def test_session_ready_output(): (TurnStarted, "turn.started"), (STTPartial, "stt.partial"), (STTFinal, "stt.final"), + (AgentDelta, "agent.delta"), (AgentResponse, "agent.response"), (TurnEnded, "turn.ended"), (Interrupted, "interrupted"), @@ -115,7 +118,7 @@ def test_turn_output_event_types( "turn_id": 1, } - if model in (STTPartial, STTFinal, AgentResponse): + if model in (STTPartial, STTFinal, AgentDelta, AgentResponse): data["text"] = "test" event = model(**data) @@ -168,3 +171,30 @@ def test_error_event_defaults(): assert event.retryable is False assert event.details == {} + +def test_agent_delta_serialization(): + event = AgentDelta( + session_id="session-1", + turn_id=7, + text=" mundo", + ) + + assert event.model_dump() == { + "type": "agent.delta", + "session_id": "session-1", + "turn_id": 7, + "text": " mundo", + } + + +def test_agent_delta_rejects_wrong_type(): + with pytest.raises(ValidationError): + AgentDelta.model_validate( + { + "type": "agent.response", + "session_id": "session-1", + "turn_id": 7, + "text": "Olá", + } + ) + diff --git a/tests/test_text_chunker.py b/tests/test_text_chunker.py new file mode 100644 index 0000000..3c9b5a9 --- /dev/null +++ b/tests/test_text_chunker.py @@ -0,0 +1,58 @@ +from voice_assistant.agent.text_chunker import TextChunker + + +def test_chunker_emits_sentence(): + chunker = TextChunker(min_chars=5) + + assert chunker.feed("Olá, tudo bem?") == [ + "Olá, tudo bem?" + ] + + +def test_chunker_handles_incremental_deltas(): + chunker = TextChunker(min_chars=5) + + assert chunker.feed("Olá,") == [] + assert chunker.feed(" tudo") == [] + assert chunker.feed(" bem?") == [ + "Olá, tudo bem?" + ] + + +def test_chunker_preserves_remaining_text(): + chunker = TextChunker(min_chars=5) + + assert chunker.feed("Olá, tudo bem? Posso ajudar") == [ + "Olá, tudo bem?" + ] + + assert chunker.flush() == "Posso ajudar" + + +def test_chunker_breaks_long_text(): + chunker = TextChunker( + min_chars=5, + max_chars=20, + ) + + segments = chunker.feed( + "Esta é uma frase bastante longa sem pontuação" + ) + + assert segments + assert all( + len(segment) <= 20 + for segment in segments + ) + + +def test_chunker_rejects_invalid_limits(): + try: + TextChunker(min_chars=20, max_chars=10) + except ValueError: + pass + else: + raise AssertionError( + "Expected ValueError for invalid limits." + ) + diff --git a/tests/test_voice_pipeline.py b/tests/test_voice_pipeline.py index 1a6f6dd..37f8ae3 100644 --- a/tests/test_voice_pipeline.py +++ b/tests/test_voice_pipeline.py @@ -18,10 +18,17 @@ Non-responsibilities: import pytest import asyncio +from contextlib import aclosing + +from voice_assistant.voice_pipeline import ( + PipelineAudioChunk, + PipelineAgentResponse, + PipelineTextDelta, + VoicePipeline, +) from voice_assistant.audio_gateway.app.session import SessionState from voice_assistant.turn_manager import TurnManager, TurnState -from voice_assistant.voice_pipeline import VoicePipeline from voice_assistant.agent.mock import MockAgentAdapter from voice_assistant.tts.mock import MockTTSAdapter @@ -818,3 +825,268 @@ async def test_pipeline_rejects_new_turn_while_previous_task_runs(): assert second_turn.turn_id == 2 assert session.active_turn_id == 2 +@pytest.mark.anyio +async def test_stream_turn_response_produces_audio_before_agent_finishes(): + """ + Verify that TTS audio can be delivered while the Agent + is still generating its response. + """ + + class StreamingAgent: + def __init__(self): + self.first_delta_sent = asyncio.Event() + self.release_generation = asyncio.Event() + self.finished = False + + async def stream(self, session_id, turn_id, text): + yield "Olá, como posso ajudar?" + + self.first_delta_sent.set() + + await self.release_generation.wait() + + yield " Estou à disposição." + self.finished = True + + class StreamingTTS: + async def synthesize(self, text): + yield text.encode("utf-8") + + agent = StreamingAgent() + + pipeline = VoicePipeline( + session=SessionState(session_id="stream-session"), + turn_manager=TurnManager(), + agent=agent, + tts=StreamingTTS(), + ) + + turn = pipeline.start_turn() + pipeline.on_final(turn.turn_id, "Olá") + + first_audio_received = asyncio.Event() + received_events = [] + + async def consume(): + async with aclosing( + pipeline.stream_turn_response( + session_id="stream-session", + turn_id=turn.turn_id, + text="Olá", + ) + ) as stream: + async for event in stream: + received_events.append(event) + + if isinstance(event, PipelineAudioChunk): + first_audio_received.set() + + task = asyncio.create_task(consume()) + + try: + await asyncio.wait_for( + first_audio_received.wait(), + timeout=2.0, + ) + + assert agent.finished is False + + assert any( + isinstance(event, PipelineTextDelta) + for event in received_events + ) + + assert any( + isinstance(event, PipelineAudioChunk) + for event in received_events + ) + + agent.release_generation.set() + + await asyncio.wait_for(task, timeout=2.0) + + assert agent.finished is True + + assert any( + isinstance(event, PipelineAgentResponse) + for event in received_events + ) + + assert ( + pipeline.turn_manager.active_turn.state + == TurnState.COMPLETED + ) + + finally: + agent.release_generation.set() + + if not task.done(): + task.cancel() + + await asyncio.gather( + task, + return_exceptions=True, + ) + + +@pytest.mark.anyio +async def test_stream_turn_response_propagates_agent_failure(): + class FailingStreamingAgent: + async def stream(self, session_id, turn_id, text): + yield "Olá, como posso ajudar?" + raise RuntimeError("Simulated OCI streaming failure.") + + class StreamingTTS: + async def synthesize(self, text): + yield text.encode("utf-8") + + pipeline = VoicePipeline( + session=SessionState(session_id="agent-failure-session"), + turn_manager=TurnManager(), + agent=FailingStreamingAgent(), + tts=StreamingTTS(), + ) + + turn = pipeline.start_turn() + pipeline.on_final(turn.turn_id, "Olá") + + async def consume(): + async with aclosing( + pipeline.stream_turn_response( + session_id="agent-failure-session", + turn_id=turn.turn_id, + text="Olá", + ) + ) as stream: + return [event async for event in stream] + + with pytest.raises(ExceptionGroup) as exc_info: + await asyncio.wait_for(consume(), timeout=2.0) + + assert any( + isinstance(error, RuntimeError) + and "Simulated OCI streaming failure" in str(error) + for error in exc_info.value.exceptions + ) + +@pytest.mark.anyio +async def test_stream_turn_response_cancels_agent_on_tts_failure(): + class BlockingStreamingAgent: + def __init__(self): + self.cancelled = asyncio.Event() + + async def stream(self, session_id, turn_id, text): + try: + yield "Olá, como posso ajudar?" + await asyncio.Event().wait() + finally: + self.cancelled.set() + + class FailingTTS: + async def synthesize(self, text): + raise RuntimeError("Simulated TTS failure.") + yield b"" # Keep this method an async generator. + + agent = BlockingStreamingAgent() + + pipeline = VoicePipeline( + session=SessionState(session_id="tts-failure-session"), + turn_manager=TurnManager(), + agent=agent, + tts=FailingTTS(), + ) + + turn = pipeline.start_turn() + pipeline.on_final(turn.turn_id, "Olá") + + async def consume(): + async with aclosing( + pipeline.stream_turn_response( + session_id="tts-failure-session", + turn_id=turn.turn_id, + text="Olá", + ) + ) as stream: + async for _ in stream: + pass + + with pytest.raises(ExceptionGroup) as exc_info: + await asyncio.wait_for(consume(), timeout=2.0) + + assert any( + isinstance(error, RuntimeError) + and "Simulated TTS failure" in str(error) + for error in exc_info.value.exceptions + ) + + assert agent.cancelled.is_set() + +@pytest.mark.anyio +async def test_stream_turn_response_external_cancellation(): + class BlockingAgent: + def __init__(self): + self.closed = asyncio.Event() + + async def stream(self, session_id, turn_id, text): + try: + yield "Olá, como posso ajudar?" + await asyncio.Event().wait() + finally: + self.closed.set() + + class BlockingTTS: + def __init__(self): + self.started = asyncio.Event() + self.closed = asyncio.Event() + + async def synthesize(self, text): + try: + self.started.set() + await asyncio.Event().wait() + yield b"audio" + finally: + self.closed.set() + + agent = BlockingAgent() + tts = BlockingTTS() + + pipeline = VoicePipeline( + session=SessionState(session_id="cancel-session"), + turn_manager=TurnManager(), + agent=agent, + tts=tts, + ) + + turn = pipeline.start_turn() + pipeline.on_final(turn.turn_id, "Olá") + + async def consume(): + async with aclosing( + pipeline.stream_turn_response( + session_id="cancel-session", + turn_id=turn.turn_id, + text="Olá", + ) + ) as stream: + async for _ in stream: + pass + + task = asyncio.create_task(consume()) + + try: + await asyncio.wait_for(tts.started.wait(), timeout=2.0) + + task.cancel() + + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, timeout=2.0) + + assert agent.closed.is_set() + assert tts.closed.is_set() + + finally: + if not task.done(): + task.cancel() + + await asyncio.gather(task, return_exceptions=True) + diff --git a/tests/test_web_client.py b/tests/test_web_client.py new file mode 100644 index 0000000..d532247 --- /dev/null +++ b/tests/test_web_client.py @@ -0,0 +1,32 @@ + +from fastapi.testclient import TestClient + +from voice_assistant.audio_gateway.app.main import app + + +client = TestClient(app) + + +def test_web_client_page(): + response = client.get("/ui") + + assert response.status_code == 200 + assert "AgentVoice" in response.text + assert "/web/app.js" in response.text + + +def test_web_client_javascript(): + response = client.get("/web/app.js") + + assert response.status_code == 200 + assert "audio.start" in response.text + assert "audio.stop" in response.text + assert "agent.delta" in response.text + + +def test_web_client_audio_worklet(): + response = client.get("/web/pcm-worklet.js") + + assert response.status_code == 200 + assert "registerProcessor" in response.text + assert "pcm-input" in response.text