diff --git a/README.md b/README.md index fc7c218..4c02097 100644 --- a/README.md +++ b/README.md @@ -264,10 +264,29 @@ Dentro do container, monte o arquivo OCI e as chaves/tokens necessários somente para leitura, com caminhos acessíveis ao usuário do processo, e ajuste `auth_file_location`; `~` refere-se ao usuário do container, não ao do Mac. Não há integração MCP/RAG automática nesse bloco; a implementação usa `tools=[]`. +A entrada `deepagents` do catálogo seleciona `OCIDeepAgentsAdapter`, que compõe +modelo OCI, grafo e memória. `DeepAgentsAdapter` recebe um grafo pronto e não +importa o SDK OCI. A configuração existente continua válida; chamadas Python +diretas com opções OCI devem usar a nova classe de composição. + +`initialize()` prepara a sessão antes de `session.ready`. A criação/autenticação +do modelo ocorre em worker fora do event loop. `initialization_timeout_seconds` +(padrão 5 s) limita essa espera; `cleanup_timeout_seconds` (padrão 1 s) limita a +espera pelo worker durante o fechamento. Um timeout não mata a thread: a tarefa +retém a posse do cliente e o fecha quando o worker terminar. Se ainda houver +trabalho pendente, o fechamento reporta timeout, em vez de confirmar liberação. +A limpeza fecha tanto o HTTP assíncrono do modelo quanto a sessão HTTP do SDK. +Prazos maiores devem considerar o limite atual de 10 s do handshake da UI. + +`tools=[]` fornece zero ferramentas adicionais; as ferramentas nativas do +DeepAgents continuam habilitadas. Essa entrega preserva essa política. O backend +de arquivos padrão do framework usa estado virtual; não implica acesso automático +aos arquivos do host. Ferramentas futuras devem ser selecionadas na composição. + `agent.options.timeout_seconds` limita a espera por cada próximo evento do DeepAgents (inclusive eventos internos), sem incluir o tempo em que o consumidor -está processando um fragmento já entregue. Não é prazo total da resposta nem -limite de inicialização/autenticação. A expiração fecha o stream e propaga erro. +está processando um fragmento já entregue. Não é prazo total da resposta; +a inicialização tem um limite separado. A expiração fecha o stream e propaga erro. ## Quickstart diff --git a/config/providers.yaml b/config/providers.yaml index d5aacee..e52d72b 100644 --- a/config/providers.yaml +++ b/config/providers.yaml @@ -36,7 +36,7 @@ providers: deepagents: name: "LangChain Deep Agents" - implementation: "voice_assistant.agent.deep_agent:DeepAgentsAdapter" + implementation: "voice_assistant.agent.oci_deep_agent:OCIDeepAgentsAdapter" capabilities: realtime: true streaming: true diff --git a/src/voice_assistant/agent/adapter.py b/src/voice_assistant/agent/adapter.py index b415ac7..9c910b1 100644 --- a/src/voice_assistant/agent/adapter.py +++ b/src/voice_assistant/agent/adapter.py @@ -6,6 +6,14 @@ from contextlib import aclosing class AgentAdapter(ABC): + async def initialize(self) -> None: + """Prepare session-owned resources without blocking the event loop. + + Resource-free adapters inherit this no-op. Implementations retain + ownership of work still running after cancellation or timeout, and + aclose() must clean partially initialized resources as well. + """ + @abstractmethod async def process( self, diff --git a/src/voice_assistant/agent/deep_agent.py b/src/voice_assistant/agent/deep_agent.py index b57c757..2358369 100644 --- a/src/voice_assistant/agent/deep_agent.py +++ b/src/voice_assistant/agent/deep_agent.py @@ -2,10 +2,11 @@ Deep Agents implementation of the AgentVoice Agent port. Responsibilities: -- OCI Generative AI inference through LangChain. +- Translate a prepared DeepAgents graph to the Agent streaming contract. - Contextual memory per session. - Text streaming compatible with VoicePipeline. -- Asynchronous resource cleanup. +- Explicit closure of consumed event streams. +- Model/client ownership belongs to the composition that supplies the graph. Does not manage: - WebSocket connections. @@ -21,27 +22,22 @@ from collections.abc import AsyncGenerator, Iterator from typing import Any from contextlib import aclosing -from deepagents import create_deep_agent from langchain_core.messages import HumanMessage from langchain_core.runnables import RunnableConfig -from langchain_oci import ChatOCIGenAI -from langgraph.checkpoint.memory import MemorySaver +from langgraph.graph.state import CompiledStateGraph from voice_assistant.agent.adapter import StreamingAgentAdapter -DEFAULT_SYSTEM_PROMPT = """ -Você é o AgentVoice, um assistente pessoal por voz. Que responde pelo nome de Roberta. -Responda em português brasileiro, de forma natural, -objetiva e adequada para reprodução por voz. - -Mantenha o contexto das mensagens anteriores. -Interprete correções e complementos do usuário -considerando o histórico da conversa. - -Não invente informações atuais ou resultados -de ferramentas que não foram executadas. -""" +def positive_timeout(value: float, name: str) -> float: + if ( + isinstance(value, bool) + or not isinstance(value, (int, float)) + or not math.isfinite(value) + or value <= 0 + ): + raise ValueError(f"{name} must be finite and positive.") + return float(value) def _text_fragments(content: str | list[str | dict[str, Any]]) -> Iterator[str]: @@ -62,64 +58,11 @@ def _text_fragments(content: str | list[str | dict[str, Any]]) -> Iterator[str]: class DeepAgentsAdapter(StreamingAgentAdapter): - def __init__( - self, - region: str, - compartment_id: str, - model_id: str, - temperature: float, - max_completion_tokens: int, - timeout_seconds: float, - system_prompt: str = DEFAULT_SYSTEM_PROMPT, - *, - auth_type: str = "INSTANCE_PRINCIPAL", - auth_profile: str = "DEFAULT", - auth_file_location: str = "~/.oci/config", - ) -> None: - if not region: - raise ValueError("OCI region is required.") - - if not compartment_id: - raise ValueError("OCI compartment_id is required.") - - if not model_id: - raise ValueError("OCI model_id is required.") - - if ( - isinstance(timeout_seconds, bool) - or not isinstance(timeout_seconds, (int, float)) - or not math.isfinite(timeout_seconds) - or timeout_seconds <= 0 - ): - raise ValueError("timeout_seconds must be finite and positive.") - - self._stream_event_timeout = float(timeout_seconds) - - self._model = ChatOCIGenAI( - model_id=model_id, - compartment_id=compartment_id, - service_endpoint=( - f"https://inference.generativeai.{region}.oci.oraclecloud.com" - ), - auth_type=auth_type, - auth_profile=auth_profile, - auth_file_location=auth_file_location, - is_stream=True, - model_kwargs={ - "temperature": temperature, - "max_tokens": max_completion_tokens, - }, + def __init__(self, graph: CompiledStateGraph, timeout_seconds: float) -> None: + self._stream_event_timeout = positive_timeout( + timeout_seconds, "timeout_seconds" ) - - self._checkpointer = MemorySaver() - - self._graph = create_deep_agent( - model=self._model, - tools=[], - system_prompt=system_prompt, - checkpointer=self._checkpointer, - ) - + self._graph = graph self._closed = False @staticmethod @@ -176,5 +119,4 @@ class DeepAgentsAdapter(StreamingAgentAdapter): if self._closed: return - await self._model.aclose() self._closed = True diff --git a/src/voice_assistant/agent/oci_deep_agent.py b/src/voice_assistant/agent/oci_deep_agent.py new file mode 100644 index 0000000..ac51921 --- /dev/null +++ b/src/voice_assistant/agent/oci_deep_agent.py @@ -0,0 +1,178 @@ +"""Compose OCI authentication, model and graph behind the Agent contract.""" + +from __future__ import annotations + +import asyncio +import logging +from collections.abc import AsyncGenerator +from contextlib import aclosing +from functools import partial + +from deepagents import create_deep_agent +from langchain_oci import ChatOCIGenAI +from langgraph.checkpoint.memory import MemorySaver + +from voice_assistant.agent.adapter import StreamingAgentAdapter +from voice_assistant.agent.deep_agent import DeepAgentsAdapter, positive_timeout + +logger = logging.getLogger(__name__) +DEFAULT_SYSTEM_PROMPT = """ +Você é o AgentVoice, um assistente pessoal por voz. Que responde pelo nome de Roberta. + +Responda em português brasileiro, de forma natural, +objetiva e adequada para reprodução por voz. + +Mantenha o contexto das mensagens anteriores. +Interprete correções e complementos do usuário +considerando o histórico da conversa. + +Não invente informações atuais ou resultados +de ferramentas que não foram executadas. +""" + + +class OCIDeepAgentsAdapter(StreamingAgentAdapter): + """Session-scoped composition; construction itself performs no OCI I/O.""" + + def __init__( + self, + region: str, + compartment_id: str, + model_id: str, + temperature: float, + max_completion_tokens: int, + timeout_seconds: float, + system_prompt: str = DEFAULT_SYSTEM_PROMPT, + *, + auth_type: str = "INSTANCE_PRINCIPAL", + auth_profile: str = "DEFAULT", + auth_file_location: str = "~/.oci/config", + initialization_timeout_seconds: float = 5.0, + cleanup_timeout_seconds: float = 1.0, + ) -> None: + for name, value in ( + ("region", region), + ("compartment_id", compartment_id), + ("model_id", model_id), + ): + if not value: + raise ValueError(f"OCI {name} is required.") + self._event_timeout = positive_timeout(timeout_seconds, "timeout_seconds") + self._initialization_timeout = positive_timeout( + initialization_timeout_seconds, "initialization_timeout_seconds" + ) + self._cleanup_timeout = positive_timeout( + cleanup_timeout_seconds, "cleanup_timeout_seconds" + ) + self._create_model = partial( + ChatOCIGenAI, + model_id=model_id, + compartment_id=compartment_id, + service_endpoint=f"https://inference.generativeai.{region}.oci.oraclecloud.com", + auth_type=auth_type, + auth_profile=auth_profile, + auth_file_location=auth_file_location, + is_stream=True, + model_kwargs={ + "temperature": temperature, + "max_tokens": max_completion_tokens, + }, + ) + self._system_prompt = system_prompt + self._model: ChatOCIGenAI | None = None + self._delegate: DeepAgentsAdapter | None = None + self._initialization: asyncio.Task[None] | None = None + self._closing = False + self._closed = False + + async def initialize(self) -> None: + if self._closing: + raise RuntimeError("OCI Deep Agent is closing or closed.") + if self._delegate is not None: + return + if self._initialization is None: + self._initialization = asyncio.create_task(self._prepare()) + self._initialization.add_done_callback(self._observe_initialization) + try: + await asyncio.wait_for( + asyncio.shield(self._initialization), + self._initialization_timeout, + ) + except TimeoutError as exc: + self._closing = True + raise TimeoutError( + f"OCI Agent initialization exceeded {self._initialization_timeout:g} seconds." + ) from exc + except BaseException: + self._closing = True + raise + if self._closing: + raise RuntimeError("OCI Deep Agent closed during initialization.") + + async def _prepare(self) -> None: + # Shielded ownership: an abandoned caller never cancels this task. + self._model = await asyncio.to_thread(self._create_model) + try: + if self._closing: + return + graph = create_deep_agent( + model=self._model, + tools=[], # Additional tools only; DeepAgents native tools remain enabled. + system_prompt=self._system_prompt, + checkpointer=MemorySaver(), + ) + self._delegate = DeepAgentsAdapter( + graph, timeout_seconds=self._event_timeout + ) + finally: + if self._closing or self._delegate is None: + await self._dispose_model() + + def _observe_initialization(self, task: asyncio.Task[None]) -> None: + if task.cancelled(): + return + error = task.exception() + if error is not None and self._closing: + logger.error( + "Abandoned OCI Agent initialization failed.", + exc_info=(type(error), error, error.__traceback__), + ) + + async def _dispose_model(self) -> None: + if self._model is None: + return + model = self._model + try: + await model.aclose() + finally: + # ChatOCIGenAI.aclose closes only its asynchronous HTTP client. + model.client.base_client.session.close() + self._model = None + + async def stream( + self, session_id: str, turn_id: int, text: str + ) -> AsyncGenerator[str, None]: + await self.initialize() + if self._delegate is None: + raise RuntimeError("OCI Deep Agent is not initialized.") + async with aclosing(self._delegate.stream(session_id, turn_id, text)) as deltas: + async for delta in deltas: + yield delta + + async def aclose(self) -> None: + if self._closed: + return + self._closing = True + if self._initialization is not None: + done, _ = await asyncio.wait( + {self._initialization}, timeout=self._cleanup_timeout + ) + if not done: + raise TimeoutError( + "OCI initialization is still running; cleanup ownership retained." + ) + if self._delegate is not None: + await self._delegate.aclose() + self._delegate = None + await self._dispose_model() + self._closed = True diff --git a/src/voice_assistant/audio_gateway/app/main.py b/src/voice_assistant/audio_gateway/app/main.py index a042c05..f2b9d54 100644 --- a/src/voice_assistant/audio_gateway/app/main.py +++ b/src/voice_assistant/audio_gateway/app/main.py @@ -106,6 +106,7 @@ async def create_voice_session( vad = None try: agent = create_agent_adapter(config=app_config, catalog=provider_catalog) + await agent.initialize() tts = create_tts_adapter(config=app_config, catalog=provider_catalog) if server_vad: vad = create_vad_adapter(config=app_config, catalog=provider_catalog) diff --git a/tests/test_deep_agent.py b/tests/test_deep_agent.py index 824b42a..bcf2389 100644 --- a/tests/test_deep_agent.py +++ b/tests/test_deep_agent.py @@ -1,6 +1,6 @@ import pytest -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import MagicMock from voice_assistant.agent.adapter import AgentAdapter @@ -37,24 +37,17 @@ def test_deep_agent_session_config(): def test_deep_agent_cleanup_is_idempotent(): import asyncio - from voice_assistant.agent.deep_agent import DeepAgentsAdapter - adapter = object.__new__(DeepAgentsAdapter) - adapter._model = MagicMock() - adapter._model.aclose = AsyncMock() - adapter._closed = False - async def scenario(): + adapter = DeepAgentsAdapter(MagicMock(), timeout_seconds=1.0) await adapter.aclose() await adapter.aclose() + with pytest.raises(RuntimeError, match="closed"): + await anext(adapter.stream("session", 1, "Olá")) asyncio.run(scenario()) - adapter._model.aclose.assert_awaited_once() - assert adapter._closed is True - - def test_deep_agent_stream_can_be_cancelled(): import asyncio @@ -341,54 +334,3 @@ def test_deep_agent_checkpoint_survives_cancellation(): started = None asyncio.run(scenario()) - -@pytest.mark.parametrize( - ("options", "expected_auth"), - [ - ({}, ("INSTANCE_PRINCIPAL", "DEFAULT", "~/.oci/config")), - ( - {"auth_type": "API_KEY", "auth_profile": "LOCAL", "auth_file_location": "/config/oci"}, - ("API_KEY", "LOCAL", "/config/oci"), - ), - ( - {"auth_type": "SECURITY_TOKEN", "auth_profile": "SESSION"}, - ("SECURITY_TOKEN", "SESSION", "~/.oci/config"), - ), - ({"auth_type": "RESOURCE_PRINCIPAL"}, ("RESOURCE_PRINCIPAL", "DEFAULT", "~/.oci/config")), - ], -) -def test_deep_agent_factory_configures_authentication(monkeypatch, options, expected_auth): - import voice_assistant.agent.deep_agent as module - from voice_assistant.config.loader import load_provider_catalog - from voice_assistant.config.models import AppConfig - from voice_assistant.providers import create_agent_adapter - - model_factory = MagicMock() - agent_factory = MagicMock() - monkeypatch.setattr(module, "ChatOCIGenAI", model_factory) - monkeypatch.setattr(module, "create_deep_agent", agent_factory) - config = AppConfig.model_validate({ - "stt": {"provider": "nvidia"}, - "tts": {"provider": "voxcpm"}, - "agent": { - "provider": "deepagents", - "options": { - "region": "sa-vinhedo-1", - "compartment_id": "test-compartment", - "model_id": "test-model", - "temperature": 0.3, - "max_completion_tokens": 256, - "timeout_seconds": 60.0, - **options, - }, - }, - }) - - adapter = create_agent_adapter(config=config, catalog=load_provider_catalog()) - - model_factory.assert_called_once() - kwargs = model_factory.call_args.kwargs - assert (kwargs["auth_type"], kwargs["auth_profile"], kwargs["auth_file_location"]) == expected_auth - assert adapter._model is model_factory.return_value - assert agent_factory.call_args.kwargs["model"] is adapter._model - diff --git a/tests/test_deep_agent_stream.py b/tests/test_deep_agent_stream.py index 0c9367b..9c0b354 100644 --- a/tests/test_deep_agent_stream.py +++ b/tests/test_deep_agent_stream.py @@ -111,18 +111,9 @@ def test_timeout_is_renewed_for_each_next_event(): @pytest.mark.parametrize("timeout", [0, -1, float("inf"), float("nan"), True, "60", None]) -def test_invalid_timeout_fails_before_model_construction(monkeypatch, timeout): - import voice_assistant.agent.deep_agent as module - - def unexpected_model(**kwargs): - pytest.fail("Invalid configuration must not create a model") - - monkeypatch.setattr(module, "ChatOCIGenAI", unexpected_model) +def test_invalid_stream_timeout_is_rejected(timeout): with pytest.raises(ValueError, match="finite and positive"): - DeepAgentsAdapter( - region="sa-vinhedo-1", compartment_id="test", model_id="test", - temperature=0.3, max_completion_tokens=256, timeout_seconds=timeout, - ) + DeepAgentsAdapter(graph=SimpleNamespace(), timeout_seconds=timeout) def test_adapter_streams_real_deepagents_graph_with_fake_model(): diff --git a/tests/test_oci_deep_agent.py b/tests/test_oci_deep_agent.py new file mode 100644 index 0000000..e34453a --- /dev/null +++ b/tests/test_oci_deep_agent.py @@ -0,0 +1,273 @@ +import asyncio +import threading +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from voice_assistant.agent.oci_deep_agent import OCIDeepAgentsAdapter +from voice_assistant.config.loader import load_provider_catalog +from voice_assistant.config.models import AppConfig +from voice_assistant.providers import create_agent_adapter + + +def options(**overrides): + return { + "region": "sa-vinhedo-1", + "compartment_id": "test", + "model_id": "test-model", + "temperature": 0.3, + "max_completion_tokens": 256, + "timeout_seconds": 1.0, + **overrides, + } + + +def fake_model(): + model = MagicMock() + model.aclose = AsyncMock() + return model + + +def patch_runtime(monkeypatch, model_factory=None, graph_factory=None): + import voice_assistant.agent.oci_deep_agent as module + + model_factory = model_factory or MagicMock(return_value=fake_model()) + graph_factory = graph_factory or MagicMock() + monkeypatch.setattr(module, "ChatOCIGenAI", model_factory) + monkeypatch.setattr(module, "create_deep_agent", graph_factory) + return model_factory, graph_factory + + +@pytest.mark.parametrize( + ("auth", "expected"), + [ + ({}, ("INSTANCE_PRINCIPAL", "DEFAULT", "~/.oci/config")), + ( + { + "auth_type": "API_KEY", + "auth_profile": "LOCAL", + "auth_file_location": "/config/oci", + }, + ("API_KEY", "LOCAL", "/config/oci"), + ), + ( + {"auth_type": "SECURITY_TOKEN", "auth_profile": "SESSION"}, + ("SECURITY_TOKEN", "SESSION", "~/.oci/config"), + ), + ( + {"auth_type": "RESOURCE_PRINCIPAL"}, + ("RESOURCE_PRINCIPAL", "DEFAULT", "~/.oci/config"), + ), + ], +) +def test_catalog_selects_oci_composition_and_initializes_once( + monkeypatch, auth, expected +): + model_factory, graph_factory = patch_runtime(monkeypatch) + config = AppConfig.model_validate( + { + "stt": {"provider": "nvidia"}, + "tts": {"provider": "voxcpm"}, + "agent": {"provider": "deepagents", "options": options(**auth)}, + } + ) + + async def scenario(): + adapter = create_agent_adapter(config=config, catalog=load_provider_catalog()) + assert isinstance(adapter, OCIDeepAgentsAdapter) + model_factory.assert_not_called() + await asyncio.gather(adapter.initialize(), adapter.initialize()) + model_factory.assert_called_once() + kwargs = model_factory.call_args.kwargs + assert ( + kwargs["auth_type"], + kwargs["auth_profile"], + kwargs["auth_file_location"], + ) == expected + assert graph_factory.call_args.kwargs["model"] is model_factory.return_value + await adapter.aclose() + await adapter.aclose() + model_factory.return_value.aclose.assert_awaited_once() + model_factory.return_value.client.base_client.session.close.assert_called_once() + + asyncio.run(scenario()) + + +def test_model_authentication_does_not_block_event_loop(monkeypatch): + entered, release = threading.Event(), threading.Event() + model = fake_model() + + def factory(**kwargs): + entered.set() + assert release.wait(2) + return model + + patch_runtime(monkeypatch, factory) + + async def scenario(): + adapter = OCIDeepAgentsAdapter(**options()) + task = asyncio.create_task(adapter.initialize()) + try: + async with asyncio.timeout(1): + while not entered.is_set(): + await asyncio.sleep(0.001) + await asyncio.sleep(0.01) + assert not task.done() + finally: + release.set() + await task + await adapter.aclose() + + asyncio.run(scenario()) + + +@pytest.mark.parametrize("abandon", ["timeout", "cancel", "close"]) +def test_abandoned_initialization_retains_ownership_and_closes_late_model( + monkeypatch, abandon +): + entered, release = threading.Event(), threading.Event() + model = fake_model() + + def factory(**kwargs): + entered.set() + assert release.wait(2) + return model + + _, graph_factory = patch_runtime(monkeypatch, factory) + + async def scenario(): + adapter = OCIDeepAgentsAdapter( + **options(initialization_timeout_seconds=0.05, cleanup_timeout_seconds=0.01) + ) + task = asyncio.create_task(adapter.initialize()) + try: + async with asyncio.timeout(1): + while not entered.is_set(): + await asyncio.sleep(0.001) + if abandon == "cancel": + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + elif abandon == "timeout": + with pytest.raises(TimeoutError): + await task + else: + with pytest.raises(TimeoutError, match="ownership retained"): + await adapter.aclose() + with pytest.raises(TimeoutError, match="ownership retained"): + await adapter.aclose() + assert not adapter._initialization.done() + finally: + release.set() + await asyncio.wait_for(adapter._initialization, 1) + await asyncio.gather(task, return_exceptions=True) + await adapter.aclose() + model.aclose.assert_awaited_once() + model.client.base_client.session.close.assert_called_once() + graph_factory.assert_not_called() + + asyncio.run(scenario()) + + +def test_graph_failure_closes_partially_initialized_model(monkeypatch): + model = fake_model() + patch_runtime( + monkeypatch, + MagicMock(return_value=model), + MagicMock(side_effect=ValueError("graph failed")), + ) + + async def scenario(): + adapter = OCIDeepAgentsAdapter(**options()) + with pytest.raises(ValueError, match="graph failed"): + await adapter.initialize() + await adapter.aclose() + model.aclose.assert_awaited_once() + model.client.base_client.session.close.assert_called_once() + + asyncio.run(scenario()) + + +def test_composition_closes_graph_stream_on_early_exit(monkeypatch): + closed = asyncio.Event() + + async def source(*args, **kwargs): + try: + yield ( + SimpleNamespace(content="Olá", tool_call_chunks=[]), + {"langgraph_node": "model"}, + ) + await asyncio.Event().wait() + finally: + closed.set() + + graph = SimpleNamespace(astream=source) + patch_runtime(monkeypatch, graph_factory=MagicMock(return_value=graph)) + + async def scenario(): + adapter = OCIDeepAgentsAdapter(**options()) + stream = adapter.stream("session", 1, "Olá") + assert await anext(stream) == "Olá" + await stream.aclose() + assert closed.is_set() + await adapter.aclose() + + asyncio.run(scenario()) + + +@pytest.mark.parametrize( + "field", + ["timeout_seconds", "initialization_timeout_seconds", "cleanup_timeout_seconds"], +) +@pytest.mark.parametrize("value", [0, -1, float("inf"), float("nan"), True]) +def test_invalid_deadline_fails_before_creating_model(monkeypatch, field, value): + model_factory, _ = patch_runtime(monkeypatch) + with pytest.raises(ValueError, match="finite and positive"): + OCIDeepAgentsAdapter(**options(**{field: value})) + model_factory.assert_not_called() + + +def test_gateway_awaits_initialization_before_creating_other_providers(monkeypatch): + from voice_assistant.agent.mock import MockAgentAdapter + from voice_assistant.audio_gateway.app import main + from voice_assistant.tts.mock import MockTTSAdapter + + order = [] + + class Agent(MockAgentAdapter): + async def initialize(self): + order.append("initialized") + + def tts_factory(**kwargs): + assert order == ["initialized"] + return MockTTSAdapter() + + monkeypatch.setattr(main, "create_agent_adapter", lambda **kwargs: Agent()) + monkeypatch.setattr(main, "create_tts_adapter", tts_factory) + + async def scenario(): + session = await main.create_voice_session(AsyncMock(), None) + await session.aclose() + + asyncio.run(scenario()) + + +def test_gateway_closes_agent_when_initialization_fails(monkeypatch): + from voice_assistant.agent.mock import MockAgentAdapter + from voice_assistant.audio_gateway.app import main + + agent = MockAgentAdapter() + agent.initialize = AsyncMock(side_effect=TimeoutError("initialization deadline")) + agent.aclose = AsyncMock() + tts_factory = MagicMock() + monkeypatch.setattr(main, "create_agent_adapter", lambda **kwargs: agent) + monkeypatch.setattr(main, "create_tts_adapter", tts_factory) + + async def scenario(): + with pytest.raises(TimeoutError, match="initialization deadline"): + await main.create_voice_session(AsyncMock(), None) + agent.aclose.assert_awaited_once() + tts_factory.assert_not_called() + + asyncio.run(scenario()) diff --git a/tools/deepagents_adapter_smoke_test.py b/tools/deepagents_adapter_smoke_test.py index 80614ad..3f8acd4 100644 --- a/tools/deepagents_adapter_smoke_test.py +++ b/tools/deepagents_adapter_smoke_test.py @@ -7,7 +7,7 @@ import asyncio import os from contextlib import aclosing -from voice_assistant.agent.deep_agent import DeepAgentsAdapter +from voice_assistant.agent.oci_deep_agent import OCIDeepAgentsAdapter async def main(): @@ -16,12 +16,13 @@ async def main(): if not compartment_id: raise RuntimeError("OCI_COMPARTMENT_ID is required.") - adapter = DeepAgentsAdapter( + adapter = OCIDeepAgentsAdapter( region="sa-vinhedo-1", compartment_id=compartment_id, model_id="openai.gpt-4o-mini", temperature=0.3, max_completion_tokens=256, + timeout_seconds=60.0, ) try: