From 4d7ff46a36962a283ca4ad3fce946157bac83915 Mon Sep 17 00:00:00 2001 From: alex alves Date: Tue, 3 Nov 2026 16:15:14 -0300 Subject: [PATCH] refactor: formaliza contratos de streaming e ciclo de vida dos provedores --- README.md | 2 +- src/voice_assistant/agent/__init__.py | 3 +- src/voice_assistant/agent/adapter.py | 45 +++- src/voice_assistant/agent/deep_agent.py | 10 +- src/voice_assistant/agent/oci.py | 21 +- .../application/voice_session.py | 23 ++- src/voice_assistant/audio_gateway/app/main.py | 13 +- src/voice_assistant/tts/adapter.py | 24 ++- src/voice_assistant/tts/mock.py | 4 +- src/voice_assistant/tts/voxcpm.py | 4 +- src/voice_assistant/voice_pipeline.py | 15 +- tests/test_gateway.py | 18 +- tests/test_provider_contracts.py | 194 ++++++++++++++++++ tests/test_voice_pipeline.py | 18 +- 14 files changed, 337 insertions(+), 57 deletions(-) create mode 100644 tests/test_provider_contracts.py diff --git a/README.md b/README.md index 3268b17..20186ad 100644 --- a/README.md +++ b/README.md @@ -117,7 +117,7 @@ This allows implementations to evolve without requiring the Audio Gateway to und | 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. +The `AgentAdapter` defines complete-response processing and session cleanup. `StreamingAgentAdapter` explicitly adds incremental text generation; OCI and DeepAgents implement this contract. Streaming returns an asynchronous generator that the consumer closes on early exit. `TTSAdapter` defines the same ownership rule for synthesized audio. Both contracts expose `aclose()`; providers retaining session resources override its default no-op implementation. ### Runtime Composition diff --git a/src/voice_assistant/agent/__init__.py b/src/voice_assistant/agent/__init__.py index d3ecd81..8341629 100644 --- a/src/voice_assistant/agent/__init__.py +++ b/src/voice_assistant/agent/__init__.py @@ -1,7 +1,8 @@ -from .adapter import AgentAdapter +from .adapter import AgentAdapter, StreamingAgentAdapter from .mock import MockAgentAdapter __all__ = [ "AgentAdapter", + "StreamingAgentAdapter", "MockAgentAdapter", ] \ No newline at end of file diff --git a/src/voice_assistant/agent/adapter.py b/src/voice_assistant/agent/adapter.py index 338c07d..b415ac7 100644 --- a/src/voice_assistant/agent/adapter.py +++ b/src/voice_assistant/agent/adapter.py @@ -1,4 +1,8 @@ +"""Agent processing, optional streaming capability, and session cleanup.""" + from abc import ABC, abstractmethod +from collections.abc import AsyncGenerator +from contextlib import aclosing class AgentAdapter(ABC): @@ -9,6 +13,45 @@ class AgentAdapter(ABC): turn_id: int, text: str, ) -> str: - """Process the final text of a turn and return the response.""" + """Return a complete response, propagating cancellation and errors.""" raise NotImplementedError + async def aclose(self) -> None: + """Release session-owned resources after execution has stopped. + + Override when the adapter retains connections or other resources. + Repeated calls must be safe; cleanup errors must propagate so the + application can report them and continue cleaning other providers. + Resource-free adapters inherit this no-op implementation. + """ + + +class StreamingAgentAdapter(AgentAdapter): + @abstractmethod + def stream( + self, + session_id: str, + turn_id: int, + text: str, + ) -> AsyncGenerator[str, None]: + """Yield ordered text deltas, preserving spaces between fragments. + + The consumer owns the generator and closes it on early exit. + Implementations must release stream-owned resources in finally + blocks and propagate cancellation and provider errors. + """ + raise NotImplementedError + + async def process( + self, + session_id: str, + turn_id: int, + text: str, + ) -> str: + """Offer complete-response processing for streaming implementations.""" + async with aclosing(self.stream(session_id, turn_id, text)) as deltas: + parts = [delta async for delta in deltas] + response = "".join(parts).strip() + if not response: + raise RuntimeError("Agent returned an empty streaming response.") + return response diff --git a/src/voice_assistant/agent/deep_agent.py b/src/voice_assistant/agent/deep_agent.py index 2d59316..ecdb672 100644 --- a/src/voice_assistant/agent/deep_agent.py +++ b/src/voice_assistant/agent/deep_agent.py @@ -16,7 +16,7 @@ Does not manage: from __future__ import annotations -from collections.abc import AsyncIterator +from collections.abc import AsyncGenerator from contextlib import aclosing from deepagents import create_deep_agent @@ -24,7 +24,7 @@ from langchain_core.messages import HumanMessage from langchain_oci import ChatOCIGenAI from langgraph.checkpoint.memory import MemorySaver -from voice_assistant.agent.adapter import AgentAdapter +from voice_assistant.agent.adapter import StreamingAgentAdapter DEFAULT_SYSTEM_PROMPT = """ Você é o AgentVoice, um assistente pessoal por voz. Que responde pelo nome de Roberta. @@ -41,7 +41,7 @@ de ferramentas que não foram executadas. """ -class DeepAgentsAdapter(AgentAdapter): +class DeepAgentsAdapter(StreamingAgentAdapter): def __init__( self, region: str, @@ -133,7 +133,7 @@ class DeepAgentsAdapter(AgentAdapter): session_id: str, turn_id: int, text: str, - ) -> AsyncIterator[str]: + ) -> AsyncGenerator[str, None]: if self._closed: raise RuntimeError("Deep Agent is closed.") @@ -169,5 +169,5 @@ class DeepAgentsAdapter(AgentAdapter): if self._closed: return - self._closed = True await self.model.aclose() + self._closed = True diff --git a/src/voice_assistant/agent/oci.py b/src/voice_assistant/agent/oci.py index b68305c..0d8b5f4 100644 --- a/src/voice_assistant/agent/oci.py +++ b/src/voice_assistant/agent/oci.py @@ -7,16 +7,17 @@ chat inference using the GenericChatRequest format. from __future__ import annotations +import asyncio import json from typing import Any import logging -from collections.abc import AsyncIterator +from collections.abc import AsyncGenerator from contextlib import aclosing import oci -from voice_assistant.agent.adapter import AgentAdapter +from voice_assistant.agent.adapter import StreamingAgentAdapter from voice_assistant.agent.oci_stream import stream_sse_events from voice_assistant.agent.oci_connection import open_oci_stream @@ -91,7 +92,7 @@ def extract_text_delta(raw_data: str) -> str | None: logger = logging.getLogger(__name__) -class OCIAgentAdapter(AgentAdapter): +class OCIAgentAdapter(StreamingAgentAdapter): """ OCI Generative AI streaming Agent. @@ -135,6 +136,7 @@ class OCIAgentAdapter(AgentAdapter): self.temperature = temperature self.max_completion_tokens = max_completion_tokens self.timeout_seconds = timeout_seconds + self._closed = False signer = ( oci.auth.signers @@ -184,7 +186,7 @@ class OCIAgentAdapter(AgentAdapter): session_id: str, turn_id: int, text: str, - ) -> AsyncIterator[str]: + ) -> AsyncGenerator[str, None]: """ Stream OCI Generative AI text deltas. @@ -192,6 +194,9 @@ class OCIAgentAdapter(AgentAdapter): event loop. SSE events are consumed through the asynchronous bridge. """ + if self._closed: + raise RuntimeError("OCI Agent is closed.") + if not text.strip(): raise ValueError("Agent input text cannot be empty.") @@ -255,3 +260,11 @@ class OCIAgentAdapter(AgentAdapter): ) return response_text + + + async def aclose(self) -> None: + """Release the OCI HTTP session after turn execution has stopped.""" + if self._closed: + return + await asyncio.to_thread(self.client.base_client.session.close) + self._closed = True diff --git a/src/voice_assistant/application/voice_session.py b/src/voice_assistant/application/voice_session.py index 43c0e58..5b17a6d 100644 --- a/src/voice_assistant/application/voice_session.py +++ b/src/voice_assistant/application/voice_session.py @@ -5,6 +5,7 @@ import logging from collections.abc import Awaitable, Callable from contextlib import aclosing +from voice_assistant.agent.adapter import StreamingAgentAdapter from voice_assistant.application.events import ( AgentReply, AgentTextDelta, @@ -245,7 +246,7 @@ class VoiceSession: 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)): + if isinstance(self._pipeline.agent, StreamingAgentAdapter): async with aclosing(self._pipeline.stream_turn_response( session_id=self.session_id, turn_id=turn_id, text=text, )) as response_events: @@ -263,8 +264,11 @@ class VoiceSession: 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)) + async with aclosing( + self._pipeline.synthesize_response(result.response_text) + ) as audio_stream: + async for audio in audio_stream: + 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: @@ -404,9 +408,14 @@ class VoiceSession: 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): + for name, provider in ( + ("Agent", self._pipeline.agent), + ("TTS", self._pipeline.tts), + ): try: - await close_agent() + await provider.aclose() except Exception: - logger.exception("Failed to close Agent provider. session_id=%s", self.session_id) + logger.exception( + "Failed to close %s provider. session_id=%s", + name, self.session_id, + ) diff --git a/src/voice_assistant/audio_gateway/app/main.py b/src/voice_assistant/audio_gateway/app/main.py index 4979b95..998757e 100644 --- a/src/voice_assistant/audio_gateway/app/main.py +++ b/src/voice_assistant/audio_gateway/app/main.py @@ -28,6 +28,7 @@ from voice_assistant.domain.session import SessionState from voice_assistant.providers import ( create_agent_adapter, create_stt_adapter, create_tts_adapter, ) +from voice_assistant.tts.adapter import TTSAdapter from voice_assistant.turn_manager import TurnManager from voice_assistant.voice_pipeline import VoicePipeline @@ -96,6 +97,7 @@ async def create_voice_session(websocket: WebSocket, session_id: str | None) -> """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) + tts: TTSAdapter | None = None try: tts = create_tts_adapter(config=app_config, catalog=provider_catalog) pipeline = VoicePipeline( @@ -104,12 +106,15 @@ async def create_voice_session(websocket: WebSocket, session_id: str | None) -> 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): + for name, provider in (("Agent", agent), ("TTS", tts)): + if provider is None: + continue try: - await close_agent() + await provider.aclose() except Exception: - logger.exception("Failed to close Agent after session composition failure.") + logger.exception( + "Failed to close %s after session composition failure.", name, + ) raise return VoiceSession( pipeline=pipeline, diff --git a/src/voice_assistant/tts/adapter.py b/src/voice_assistant/tts/adapter.py index e394d1f..ae81075 100644 --- a/src/voice_assistant/tts/adapter.py +++ b/src/voice_assistant/tts/adapter.py @@ -1,12 +1,24 @@ +"""Speech synthesis streaming and session cleanup contract.""" + from abc import ABC, abstractmethod -from collections.abc import AsyncIterator +from collections.abc import AsyncGenerator class TTSAdapter(ABC): @abstractmethod - async def synthesize( - self, - text: str, - ) -> AsyncIterator[bytes]: - """Convert text into audio chunks.""" + def synthesize(self, text: str) -> AsyncGenerator[bytes, None]: + """Return an audio generator consumed directly with async for. + + The caller closes it on early exit. Implementations must release + stream-owned resources in finally blocks and propagate cancellation + and provider errors. The caller does not await synthesize itself. + """ raise NotImplementedError + + async def aclose(self) -> None: + """Release retained resources after synthesis has stopped. + + Override when resources are retained between synthesis calls. + Repeated calls must be safe and cleanup errors must propagate. + Adapters owning resources only inside synthesize inherit this no-op. + """ diff --git a/src/voice_assistant/tts/mock.py b/src/voice_assistant/tts/mock.py index cd183ab..94f4e86 100644 --- a/src/voice_assistant/tts/mock.py +++ b/src/voice_assistant/tts/mock.py @@ -1,4 +1,4 @@ -from collections.abc import AsyncIterator +from collections.abc import AsyncGenerator from .adapter import TTSAdapter @@ -10,7 +10,7 @@ class MockTTSAdapter(TTSAdapter): async def synthesize( self, text: str, - ) -> AsyncIterator[bytes]: + ) -> AsyncGenerator[bytes, None]: self.calls.append(text) yield b"\x01\x02\x03\x04" diff --git a/src/voice_assistant/tts/voxcpm.py b/src/voice_assistant/tts/voxcpm.py index 90ac56f..6d107e5 100644 --- a/src/voice_assistant/tts/voxcpm.py +++ b/src/voice_assistant/tts/voxcpm.py @@ -1,4 +1,4 @@ -from collections.abc import AsyncIterator +from collections.abc import AsyncGenerator import httpx @@ -27,7 +27,7 @@ class VoxCPMTTSAdapter(TTSAdapter): async def synthesize( self, text: str, - ) -> AsyncIterator[bytes]: + ) -> AsyncGenerator[bytes, None]: if not text: raise ValueError("Text cannot be empty.") diff --git a/src/voice_assistant/voice_pipeline.py b/src/voice_assistant/voice_pipeline.py index 8cc6c69..42cbb5b 100644 --- a/src/voice_assistant/voice_pipeline.py +++ b/src/voice_assistant/voice_pipeline.py @@ -45,7 +45,7 @@ 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.agent.adapter import AgentAdapter, StreamingAgentAdapter from voice_assistant.domain.session import SessionState from voice_assistant.turn_manager import Turn, TurnManager from voice_assistant.tts.adapter import TTSAdapter @@ -181,10 +181,9 @@ class VoicePipeline: The TTS adapter remains responsible for provider-specific synthesis and audio streaming behavior. """ - async for audio_chunk in self.tts.synthesize( - response_text, - ): - yield audio_chunk + async with aclosing(self.tts.synthesize(response_text)) as audio_stream: + async for audio_chunk in audio_stream: + yield audio_chunk async def stream_turn_response( @@ -211,9 +210,9 @@ class VoicePipeline: f"Cannot stream response while turn is {turn.state.value}." ) - stream_method = getattr(self.agent, "stream", None) + agent = self.agent - if not callable(stream_method): + if not isinstance(agent, StreamingAgentAdapter): raise RuntimeError( "Configured Agent does not support streaming." ) @@ -230,7 +229,7 @@ class VoicePipeline: parts: list[str] = [] async with aclosing( - stream_method( + agent.stream( session_id=session_id, turn_id=turn_id, text=text, diff --git a/tests/test_gateway.py b/tests/test_gateway.py index 31c7cc6..2a392a7 100644 --- a/tests/test_gateway.py +++ b/tests/test_gateway.py @@ -1,3 +1,5 @@ +from voice_assistant.agent.adapter import StreamingAgentAdapter +from voice_assistant.tts.adapter import TTSAdapter import asyncio from concurrent.futures import ThreadPoolExecutor from contextlib import contextmanager @@ -1858,12 +1860,12 @@ def test_websocket_streaming_agent_and_tts(): """ created_stt.clear() - class StreamingTestAgent: + class StreamingTestAgent(StreamingAgentAdapter): async def stream(self, session_id, turn_id, text): yield "Olá, como posso ajudar?" yield " Estou à disposição." - class StreamingTestTTS: + class StreamingTestTTS(TTSAdapter): async def synthesize(self, text): yield text.encode("utf-8") @@ -1961,11 +1963,11 @@ def test_websocket_interrupt_during_streaming_tts(): tts_started = threading.Event() tts_cancelled = threading.Event() - class SlowStreamingAgent: + class SlowStreamingAgent(StreamingAgentAdapter): async def stream(self, session_id, turn_id, text): yield "Olá, esta é uma resposta longa." - class SlowStreamingTTS: + class SlowStreamingTTS(TTSAdapter): async def synthesize(self, text): try: tts_started.set() @@ -2100,11 +2102,11 @@ def test_websocket_session_end_during_streaming_tts(): tts_cancelled = threading.Event() - class SlowStreamingAgent: + class SlowStreamingAgent(StreamingAgentAdapter): async def stream(self, session_id, turn_id, text): yield "Esta é uma resposta longa." - class SlowStreamingTTS: + class SlowStreamingTTS(TTSAdapter): async def synthesize(self, text): try: yield b"\x01\x02\x03\x04" @@ -2297,11 +2299,11 @@ def test_websocket_interrupt_wins_pending_tts_completion(): tts_cancelled = threading.Event() - class ControlledAgent: + class ControlledAgent(StreamingAgentAdapter): async def stream(self, session_id, turn_id, text): yield "Esta resposta será interrompida." - class ControlledTTS: + class ControlledTTS(TTSAdapter): async def synthesize(self, text): try: yield b"\x01\x02\x03\x04" diff --git a/tests/test_provider_contracts.py b/tests/test_provider_contracts.py new file mode 100644 index 0000000..3296188 --- /dev/null +++ b/tests/test_provider_contracts.py @@ -0,0 +1,194 @@ +"""Behavioral guarantees for streaming and provider cleanup.""" + +import asyncio +from contextlib import aclosing +from unittest.mock import AsyncMock + +import pytest + +from voice_assistant.agent.adapter import StreamingAgentAdapter +from voice_assistant.agent.mock import MockAgentAdapter +from voice_assistant.application.events import SpeechChunk +from voice_assistant.application.voice_session import VoiceSession +from voice_assistant.domain.session import SessionState +from voice_assistant.stt.mock import MockSTTAdapter +from voice_assistant.tts.adapter import TTSAdapter +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_pipeline(agent, tts): + return VoicePipeline( + session=SessionState(), turn_manager=TurnManager(), agent=agent, tts=tts, + ) + + +@pytest.mark.anyio +async def test_streaming_agent_collects_fragments_preserving_spaces(): + closed = asyncio.Event() + + class Agent(StreamingAgentAdapter): + async def stream(self, session_id, turn_id, text): + try: + yield " Olá" + yield ", mundo! " + finally: + closed.set() + + assert await Agent().process("session", 1, "Olá") == "Olá, mundo!" + assert closed.is_set() + + +@pytest.mark.anyio +async def test_streaming_failure_propagates_and_releases_stream(): + closed = asyncio.Event() + + class Agent(StreamingAgentAdapter): + async def stream(self, session_id, turn_id, text): + try: + yield "Resposta parcial" + raise RuntimeError("Provider failed") + finally: + closed.set() + + with pytest.raises(RuntimeError, match="Provider failed"): + await Agent().process("session", 1, "Olá") + assert closed.is_set() + + +@pytest.mark.anyio +async def test_streaming_cancellation_releases_stream(): + started = asyncio.Event() + closed = asyncio.Event() + + class Agent(StreamingAgentAdapter): + async def stream(self, session_id, turn_id, text): + try: + yield "Resposta parcial" + started.set() + await asyncio.Event().wait() + finally: + closed.set() + + task = asyncio.create_task(Agent().process("session", 1, "Olá")) + try: + await asyncio.wait_for(started.wait(), timeout=1.0) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert closed.is_set() + finally: + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + +@pytest.mark.anyio +async def test_stopping_synthesis_early_closes_provider_generator(): + closed = asyncio.Event() + + class TTS(TTSAdapter): + async def synthesize(self, text): + try: + yield b"audio" + await asyncio.Event().wait() + finally: + closed.set() + + pipeline = make_pipeline(MockAgentAdapter(), TTS()) + async with aclosing(pipeline.synthesize_response("Olá")) as stream: + assert await anext(stream) == b"audio" + assert closed.is_set() + + +@pytest.mark.anyio +async def test_interrupt_during_delivery_closes_nonstreaming_tts(): + delivery_started = asyncio.Event() + synthesis_closed = asyncio.Event() + stt = MockSTTAdapter() + + class TTS(TTSAdapter): + async def synthesize(self, text): + try: + yield b"audio" + await asyncio.Event().wait() + finally: + synthesis_closed.set() + + async def emit(event): + if isinstance(event, SpeechChunk): + delivery_started.set() + await asyncio.Event().wait() + + session = VoiceSession( + pipeline=make_pipeline(MockAgentAdapter(), TTS()), + create_stt=lambda: stt, emit=emit, + ) + try: + turn_id = await session.start_turn() + await stt.emit_final("Olá") + await asyncio.wait_for(delivery_started.wait(), timeout=1.0) + await asyncio.wait_for(session.interrupt(turn_id), timeout=1.0) + assert synthesis_closed.is_set() + assert session.active_turn_id is None + finally: + await session.aclose() + + +@pytest.mark.anyio +async def test_agent_cleanup_failure_does_not_prevent_tts_cleanup(): + agent = MockAgentAdapter() + tts = MockTTSAdapter() + agent.aclose = AsyncMock(side_effect=RuntimeError("Agent close failed")) + tts.aclose = AsyncMock() + + async def emit(event): + pass + + session = VoiceSession( + pipeline=make_pipeline(agent, tts), create_stt=MockSTTAdapter, emit=emit, + ) + await session.aclose() + agent.aclose.assert_awaited_once() + tts.aclose.assert_awaited_once() + assert session.closed + + +@pytest.mark.anyio +@pytest.mark.parametrize("phase", ["tts", "turn_manager"]) +@pytest.mark.parametrize("agent_close_fails", [False, True]) +async def test_partial_composition_closes_created_providers( + monkeypatch, phase, agent_close_fails, +): + from voice_assistant.audio_gateway.app import main + + agent = MockAgentAdapter() + tts = MockTTSAdapter() + agent.aclose = AsyncMock( + side_effect=RuntimeError("Agent close failed") if agent_close_fails else None, + ) + tts.aclose = AsyncMock() + + def fail(*args, **kwargs): + raise RuntimeError("Composition failed") + + monkeypatch.setattr(main, "create_agent_adapter", lambda **kwargs: agent) + monkeypatch.setattr( + main, "create_tts_adapter", fail if phase == "tts" else lambda **kwargs: tts, + ) + monkeypatch.setattr(main, "create_turn_manager", fail) + + with pytest.raises(RuntimeError, match="Composition failed"): + await main.create_voice_session(AsyncMock(), None) + + agent.aclose.assert_awaited_once() + if phase == "tts": + tts.aclose.assert_not_awaited() + else: + tts.aclose.assert_awaited_once() diff --git a/tests/test_voice_pipeline.py b/tests/test_voice_pipeline.py index 8577b28..a50a9bc 100644 --- a/tests/test_voice_pipeline.py +++ b/tests/test_voice_pipeline.py @@ -1,3 +1,5 @@ +from voice_assistant.agent.adapter import StreamingAgentAdapter +from voice_assistant.tts.adapter import TTSAdapter """ Tests for the Voice Pipeline lifecycle orchestration. @@ -832,7 +834,7 @@ async def test_stream_turn_response_produces_audio_before_agent_finishes(): is still generating its response. """ - class StreamingAgent: + class StreamingAgent(StreamingAgentAdapter): def __init__(self): self.first_delta_sent = asyncio.Event() self.release_generation = asyncio.Event() @@ -848,7 +850,7 @@ async def test_stream_turn_response_produces_audio_before_agent_finishes(): yield " Estou à disposição." self.finished = True - class StreamingTTS: + class StreamingTTS(TTSAdapter): async def synthesize(self, text): yield text.encode("utf-8") @@ -931,12 +933,12 @@ async def test_stream_turn_response_produces_audio_before_agent_finishes(): @pytest.mark.anyio async def test_stream_turn_response_propagates_agent_failure(): - class FailingStreamingAgent: + class FailingStreamingAgent(StreamingAgentAdapter): async def stream(self, session_id, turn_id, text): yield "Olá, como posso ajudar?" raise RuntimeError("Simulated OCI streaming failure.") - class StreamingTTS: + class StreamingTTS(TTSAdapter): async def synthesize(self, text): yield text.encode("utf-8") @@ -971,7 +973,7 @@ async def test_stream_turn_response_propagates_agent_failure(): @pytest.mark.anyio async def test_stream_turn_response_cancels_agent_on_tts_failure(): - class BlockingStreamingAgent: + class BlockingStreamingAgent(StreamingAgentAdapter): def __init__(self): self.cancelled = asyncio.Event() @@ -982,7 +984,7 @@ async def test_stream_turn_response_cancels_agent_on_tts_failure(): finally: self.cancelled.set() - class FailingTTS: + class FailingTTS(TTSAdapter): async def synthesize(self, text): raise RuntimeError("Simulated TTS failure.") yield b"" # Keep this method an async generator. @@ -1023,7 +1025,7 @@ async def test_stream_turn_response_cancels_agent_on_tts_failure(): @pytest.mark.anyio async def test_stream_turn_response_external_cancellation(): - class BlockingAgent: + class BlockingAgent(StreamingAgentAdapter): def __init__(self): self.closed = asyncio.Event() @@ -1034,7 +1036,7 @@ async def test_stream_turn_response_external_cancellation(): finally: self.closed.set() - class BlockingTTS: + class BlockingTTS(TTSAdapter): def __init__(self): self.started = asyncio.Event() self.closed = asyncio.Event()