import asyncio import threading import pytest from unittest.mock import patch, AsyncMock, Mock from collections.abc import AsyncIterator from fastapi.testclient import TestClient from voice_assistant.stt.mock import MockSTTAdapter from voice_assistant.turn_manager import ( TurnManager, TurnState, ) from voice_assistant.audio_gateway.app import main from voice_assistant.audio_gateway.app.protocol import ( AUDIO_CHANNELS, AUDIO_ENCODING, AUDIO_SAMPLE_RATE, ) # Teste Agent from voice_assistant.agent.mock import MockAgentAdapter # Teste TTS from voice_assistant.tts.mock import MockTTSAdapter from voice_assistant.voice_pipeline import VoicePipeline from voice_assistant.audio_gateway.app.main import forward_stt_events from voice_assistant.stt.models import TranscriptEvent created_stt: list[MockSTTAdapter] = [] created_turn_managers: list[TurnManager] = [] created_agents: list[MockAgentAdapter] = [] created_tts: list[MockTTSAdapter] = [] def create_test_agent( config=None, catalog=None, ) -> MockAgentAdapter: agent = MockAgentAdapter() created_agents.append(agent) return agent def create_mock_stt( config=None, catalog=None, ) -> MockSTTAdapter: stt = MockSTTAdapter() created_stt.append(stt) return stt def create_test_turn_manager() -> TurnManager: manager = TurnManager() created_turn_managers.append(manager) return manager def create_test_tts( config=None, catalog=None, ) -> MockTTSAdapter: tts = MockTTSAdapter() created_tts.append(tts) return tts app = main.app client = TestClient(app) main.create_stt_adapter = create_mock_stt main.create_turn_manager = create_test_turn_manager main.create_agent_adapter = create_test_agent main.create_tts_adapter = create_test_tts def test_health(): response = client.get("/health") assert response.status_code == 200 assert response.json() == {"status": "ok"} def test_ready(): response = client.get("/ready") assert response.status_code == 200 assert response.json() == {"status": "ready"} def test_websocket_session_start(): created_agents.clear() created_tts.clear() 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, }, } ) response = websocket.receive_json() assert response["type"] == "session.ready" assert response["protocol_version"] == "1.0" assert response["session_id"] assert len(created_agents) == 1 assert len(created_tts) == 1 def test_websocket_audio_turn(): created_stt.clear() created_agents.clear() created_tts.clear() 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" assert turn_started["turn_id"] == 1 turn_id = turn_started["turn_id"] assert len(created_stt) == 1 stt = created_stt[0] audio = b"\x00\x00" * 160 websocket.send_json( { "type": "audio.chunk", "turn_id": turn_id, "sequence_number": 0, } ) websocket.send_bytes(audio) stt.emit_partial_now("Olá") partial = websocket.receive_json() assert partial == { "type": "stt.partial", "session_id": session_ready["session_id"], "turn_id": turn_id, "text": "Olá", } stt.emit_final_now("Olá") final = websocket.receive_json() assert final == { "type": "stt.final", "session_id": session_ready["session_id"], "turn_id": turn_id, "text": "Olá", } agent_response = websocket.receive_json() assert agent_response == { "type": "agent.response", "session_id": session_ready["session_id"], "turn_id": turn_id, "text": "Resposta simulada para: Olá", } audio_chunk_1 = websocket.receive_bytes() audio_chunk_2 = websocket.receive_bytes() assert audio_chunk_1 == b"\x01\x02\x03\x04" assert audio_chunk_2 == b"\x05\x06\x07\x08" websocket.send_json( { "type": "audio.stop", "turn_id": turn_id, } ) turn_ended = websocket.receive_json() assert turn_ended["type"] == "turn.ended" assert turn_ended["turn_id"] == turn_id def test_websocket_sequence_error(): created_stt.clear() 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, }, } ) websocket.receive_json() websocket.send_json( { "type": "audio.start", } ) turn_started = websocket.receive_json() turn_id = turn_started["turn_id"] websocket.send_json( { "type": "audio.chunk", "turn_id": turn_id, "sequence_number": 1, } ) websocket.send_bytes(b"\x00\x00" * 160) error = websocket.receive_json() assert error["type"] == "error" assert error["code"] == "AUDIO_SEQUENCE_ERROR" def test_websocket_interrupt(): created_stt.clear() 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" assert turn_started["turn_id"] == 1 turn_id = turn_started["turn_id"] assert len(created_stt) == 1 websocket.send_json( { "type": "interrupt", "turn_id": turn_id, } ) interrupted = websocket.receive_json() assert interrupted["type"] == "interrupted" assert interrupted["session_id"] == session_ready["session_id"] assert interrupted["turn_id"] == turn_id websocket.send_json( { "type": "audio.start", } ) next_turn = websocket.receive_json() assert next_turn["type"] == "turn.started" assert next_turn["turn_id"] == 2 def test_websocket_session_end(): created_stt.clear() 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" session_id = session_ready["session_id"] websocket.send_json( { "type": "session.end", } ) session_ended = websocket.receive_json() assert session_ended["type"] == "session.ended" assert session_ended["session_id"] == session_id def test_websocket_session_end_closes_agent_provider(): """ Explicit session.end must close the Agent provider. """ agent = MockAgentAdapter() agent.aclose = AsyncMock() with patch.object( main, "create_agent_adapter", return_value=agent, ): 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, }, }) ready = websocket.receive_json() assert ready["type"] == "session.ready" websocket.send_json({ "type": "session.end", }) ended = websocket.receive_json() assert ended["type"] == "session.ended" agent.aclose.assert_awaited_once() def test_websocket_disconnect_closes_agent_provider(): """ Unexpected WebSocket disconnect must close the Agent provider. """ agent = MockAgentAdapter() agent.aclose = AsyncMock() with patch.object( main, "create_agent_adapter", return_value=agent, ): 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, }, }) ready = websocket.receive_json() assert ready["type"] == "session.ready" # Leave without sending session.end. # TestClient closes the WebSocket context. agent.aclose.assert_awaited_once() def test_websocket_interrupt_preserves_agent_provider(): """ Interrupt must cancel the turn without closing the Agent provider. The provider is closed only when the WebSocket session ends. """ created_stt.clear() agent = MockAgentAdapter() agent.aclose = AsyncMock() with patch.object( main, "create_agent_adapter", return_value=agent, ): 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, }, }) ready = websocket.receive_json() assert ready["type"] == "session.ready" websocket.send_json({ "type": "audio.start", }) started = websocket.receive_json() assert started["type"] == "turn.started" websocket.send_json({ "type": "interrupt", "turn_id": started["turn_id"], }) interrupted = websocket.receive_json() assert interrupted["type"] == "interrupted" # Interrupt must not close the provider. agent.aclose.assert_not_awaited() # The same WebSocket must accept a new turn. websocket.send_json({ "type": "audio.start", }) next_turn = websocket.receive_json() assert next_turn["type"] == "turn.started" assert next_turn["turn_id"] != started["turn_id"] # Provider must remain open. agent.aclose.assert_not_awaited() websocket.send_json({ "type": "session.end", }) ended = websocket.receive_json() assert ended["type"] == "session.ended" agent.aclose.assert_awaited_once() def test_websocket_forwards_stt_events(): created_stt.clear() created_turn_managers.clear() created_agents.clear() created_tts.clear() 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"] assert len(created_turn_managers) == 1 turn_manager = created_turn_managers[0] assert len(created_agents) == 1 agent = created_agents[0] assert len(created_stt) == 1 stt = created_stt[0] assert len(created_tts) == 1 tts = created_tts[0] stt.emit_partial_now( "Ô Alex tudo bem" ) partial = websocket.receive_json() assert partial == { "type": "stt.partial", "session_id": session_ready["session_id"], "turn_id": turn_id, "text": "Ô Alex tudo bem", } assert turn_manager.active_turn is not None assert turn_manager.active_turn.turn_id == turn_id assert turn_manager.active_turn.partial_text == ( "Ô Alex tudo bem" ) assert turn_manager.active_turn.state == TurnState.LISTENING stt.emit_final_now( "Ô Alex tudo bem?" ) final = websocket.receive_json() assert final == { "type": "stt.final", "session_id": session_ready["session_id"], "turn_id": turn_id, "text": "Ô Alex tudo bem?", } assert turn_manager.active_turn is not None assert turn_manager.active_turn.turn_id == turn_id assert turn_manager.active_turn.final_text == ( "Ô Alex tudo bem?" ) assert turn_manager.active_turn.partial_text == ( "Ô Alex tudo bem?" ) assert agent.calls == [ ( session_ready["session_id"], turn_id, "Ô Alex tudo bem?", ) ] agent_response = websocket.receive_json() assert agent_response == { "type": "agent.response", "session_id": session_ready["session_id"], "turn_id": turn_id, "text": ( "Resposta simulada para: " "Ô Alex tudo bem?" ), } assert tts.calls == [ "Resposta simulada para: Ô Alex tudo bem?" ] audio_chunk_1 = websocket.receive_bytes() audio_chunk_2 = websocket.receive_bytes() assert audio_chunk_1 == b"\x01\x02\x03\x04" assert audio_chunk_2 == b"\x05\x06\x07\x08" assert turn_manager.active_turn is not None assert turn_manager.active_turn.state == TurnState.COMPLETED websocket.send_json( { "type": "audio.stop", "turn_id": turn_id, } ) turn_ended = websocket.receive_json() assert turn_ended == { "type": "turn.ended", "session_id": session_ready["session_id"], "turn_id": turn_id, } def test_websocket_disconnect_cleans_active_turn(): """ Verify cleanup when a WebSocket disconnects during an active turn. Responsibilities: - Confirm the STT session is aborted. - Confirm the semantic turn is released. - Confirm the Gateway handles disconnect without an error. """ created_stt.clear() created_turn_managers.clear() 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" assert turn_started["turn_id"] == 1 assert len(created_stt) == 1 assert len(created_turn_managers) == 1 stt = created_stt[0] turn_manager = created_turn_managers[0] assert stt._started is True assert turn_manager.active_turn is not None assert turn_manager.active_turn.state == TurnState.LISTENING # Leaving the context closes the client WebSocket. # The Gateway's finally block should perform cleanup. assert stt._started is False assert turn_manager.active_turn is None def test_websocket_disconnect_executes_cleanup(): """ Verify that WebSocket disconnection executes real cleanup. Purpose: Confirm cancellation and STT abort are invoked while preserving their original implementations. """ created_stt.clear() created_turn_managers.clear() real_cancel = VoicePipeline.cancel_turn_task real_abort = MockSTTAdapter.abort_session async def tracked_cancel(pipeline): return await real_cancel(pipeline) async def tracked_abort(stt): """Record STT abort while preserving the real implementation.""" return await real_abort(stt) with ( patch.object( VoicePipeline, "cancel_turn_task", autospec=True, side_effect=tracked_cancel, ) as cancel_mock, patch.object( MockSTTAdapter, "abort_session", autospec=True, side_effect=tracked_abort, ) as abort_mock, ): 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" cancel_mock.assert_awaited() abort_mock.assert_awaited() assert created_stt[0]._started is False assert created_turn_managers[0].active_turn is None def test_websocket_disconnect_awaits_active_turn_task(): """ Verify that disconnect cleanup awaits the active turn task. Purpose: Ensure asynchronous turn execution terminates before the Gateway releases the STT provider and semantic state. Responsibilities: - Capture the actual turn task created by the Gateway. - Verify task cancellation on disconnect. - Verify task termination precedes STT abort. - Verify semantic turn state is released. """ created_stt.clear() created_turn_managers.clear() registered_tasks = [] cleanup_order = [] real_register = VoicePipeline.register_turn_task real_abort = MockSTTAdapter.abort_session def tracked_register(pipeline, task): registered_tasks.append(task) return real_register(pipeline, task) async def tracked_abort(stt): cleanup_order.append("stt.abort") return await real_abort(stt) with ( patch.object( VoicePipeline, "register_turn_task", autospec=True, side_effect=tracked_register, ), patch.object( MockSTTAdapter, "abort_session", autospec=True, side_effect=tracked_abort, ), ): 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" assert len(registered_tasks) == 1 # The context manager closes the WebSocket and waits # for the Gateway's disconnect cleanup. task = registered_tasks[0] assert task.done() assert task.cancelled() assert cleanup_order == ["stt.abort"] assert created_stt[0]._started is False assert created_turn_managers[0].active_turn is None def test_websocket_disconnect_continues_cleanup_after_stt_abort_failure( ): """ Verify that STT abort failures do not prevent session cleanup. Purpose: Ensure the Gateway continues releasing semantic state when the STT provider fails during disconnect cleanup. Responsibilities: - Simulate an STT abort failure. - Verify the failure is logged. - Verify the semantic turn is released. """ created_stt.clear() created_turn_managers.clear() real_abort = MockSTTAdapter.abort_session async def failing_abort(stt): raise RuntimeError("Simulated STT abort failure.") with patch.object( MockSTTAdapter, "abort_session", autospec=True, side_effect=failing_abort, ): with patch.object( main.logger, "exception", ) as log_mock: 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" assert log_mock.called assert created_turn_managers[0].active_turn is None def test_websocket_disconnect_continues_cleanup_after_task_cancel_failure(): """ Verify that task cancellation failures do not stop Gateway cleanup. Purpose: Ensure STT abort and semantic state cleanup still execute when VoicePipeline.cancel_turn_task() raises an exception. Responsibilities: - Simulate a cancellation failure. - Verify the failure is logged. - Verify STT abort is attempted. - Verify the active semantic turn is released. """ created_stt.clear() created_turn_managers.clear() real_cancel = VoicePipeline.cancel_turn_task real_abort = MockSTTAdapter.abort_session cancel_calls = 0 abort_calls = 0 async def failing_cancel(pipeline): nonlocal cancel_calls cancel_calls += 1 # Simulate failure while still stopping the real task, # so the test does not leak asynchronous work. await real_cancel(pipeline) raise RuntimeError("Simulated turn task cancellation failure.") async def tracked_abort(stt): nonlocal abort_calls abort_calls += 1 await real_abort(stt) with ( patch.object( VoicePipeline, "cancel_turn_task", autospec=True, side_effect=failing_cancel, ), patch.object( MockSTTAdapter, "abort_session", autospec=True, side_effect=tracked_abort, ), patch.object( main.logger, "exception", ) as log_mock, ): 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" assert cancel_calls >= 1 assert abort_calls >= 1 assert any( "Failed to cancel voice turn task" in str(call) for call in log_mock.call_args_list ) assert created_stt[0]._started is False assert created_turn_managers[0].active_turn is None def test_websocket_interrupt_timeout_preserves_active_turn(): """ Verify Gateway behavior when turn cancellation times out. Purpose: Ensure a cancellation timeout does not falsely confirm interruption or release an active semantic turn. Responsibilities: - Simulate cancellation timeout. - Verify TURN_CANCEL_FAILED response. - Preserve the active turn. - Reject overlapping turn creation. """ created_stt.clear() created_turn_managers.clear() real_cancel = VoicePipeline.cancel_turn_task async def simulated_timeout(pipeline): raise TimeoutError( "Turn task cancellation timed out." ) with patch.object( VoicePipeline, "cancel_turn_task", autospec=True, side_effect=simulated_timeout, ): 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"] websocket.send_json( { "type": "interrupt", "turn_id": turn_id, } ) response = websocket.receive_json() assert response["type"] == "error" assert response["code"] == "TURN_CANCEL_FAILED" assert response["turn_id"] == turn_id assert "timed out" in response["message"] assert created_turn_managers[0].active_turn is not None assert ( created_turn_managers[0].active_turn.turn_id == turn_id ) websocket.send_json({"type": "audio.start"}) next_response = websocket.receive_json() assert next_response["type"] == "error" assert next_response["code"] == "INVALID_TURN" def test_websocket_interrupt_during_agent(): """Interrupt an active Agent without producing a late response.""" created_stt.clear() created_turn_managers.clear() agent = BlockingAgentAdapter() with patch.object( main, "create_agent_adapter", return_value=agent, ): 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á!") final_event = websocket.receive_json() assert final_event["type"] == "stt.final" assert agent.started.wait(timeout=2.0) websocket.send_json( { "type": "interrupt", "turn_id": turn_id, } ) interrupted = websocket.receive_json() assert interrupted["type"] == "interrupted" assert interrupted["turn_id"] == turn_id assert agent.cancelled.is_set() assert created_turn_managers[0].active_turn is None # A new turn must remain possible after cancellation. websocket.send_json({"type": "audio.start"}) next_turn = websocket.receive_json() assert next_turn["type"] == "turn.started" assert next_turn["turn_id"] == 2 def test_websocket_interrupt_during_tts(): """Interrupt TTS while synthesis is blocked.""" created_stt.clear() created_turn_managers.clear() tts = BlockingTTSAdapter() with patch.object( main, "create_tts_adapter", return_value=tts, ): 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á!") final_event = websocket.receive_json() assert final_event["type"] == "stt.final" agent_event = websocket.receive_json() assert agent_event["type"] == "agent.response" assert tts.started.wait(timeout=2.0) websocket.send_json( { "type": "interrupt", "turn_id": turn_id, } ) interrupted = websocket.receive_json() assert interrupted["type"] == "interrupted" assert interrupted["turn_id"] == turn_id assert tts.cancelled.is_set() assert created_turn_managers[0].active_turn is None class BlockingAgentAdapter(MockAgentAdapter): """Agent that waits until cancelled.""" def __init__(self) -> None: super().__init__() self.started = threading.Event() self.cancelled = threading.Event() async def process( self, session_id: str, turn_id: int, text: str, ) -> str: self.started.set() try: await asyncio.Event().wait() except asyncio.CancelledError: self.cancelled.set() raise class BlockingTTSAdapter(MockTTSAdapter): """TTS that waits until cancelled before producing audio.""" def __init__(self) -> None: super().__init__() self.started = threading.Event() self.cancelled = threading.Event() async def synthesize( self, text: str, ) -> AsyncIterator[bytes]: self.started.set() try: await asyncio.Event().wait() yield b"\x01\x02\x03\x04" except asyncio.CancelledError: self.cancelled.set() raise class StreamingBlockingTTSAdapter(MockTTSAdapter): """ Mock TTS that emits one audio chunk and then blocks. Purpose: Test interruption during an active audio stream. Responsibilities: - Emit the first audio chunk. - Wait indefinitely before producing the second. - Record cancellation. """ FIRST_CHUNK = b"\x01\x02\x03\x04" SECOND_CHUNK = b"\x05\x06\x07\x08" def __init__(self) -> None: super().__init__() self.first_chunk_sent = threading.Event() self.cancelled = threading.Event() async def synthesize( self, text: str, ) -> AsyncIterator[bytes]: self.calls.append(text) yield self.FIRST_CHUNK self.first_chunk_sent.set() try: await asyncio.Event().wait() yield self.SECOND_CHUNK except asyncio.CancelledError: self.cancelled.set() raise def test_websocket_interrupt_between_tts_chunks(): """ Verify interruption between streamed TTS audio chunks. Purpose: Ensure that an interrupted turn cannot continue transmitting audio after interruption confirmation. """ created_stt.clear() created_turn_managers.clear() tts = StreamingBlockingTTSAdapter() with patch.object( main, "create_tts_adapter", return_value=tts, ): 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á!") final_event = websocket.receive_json() assert final_event["type"] == "stt.final" agent_event = websocket.receive_json() assert agent_event["type"] == "agent.response" # First PCM chunk is delivered normally. first_chunk = websocket.receive_bytes() assert first_chunk == tts.FIRST_CHUNK assert tts.first_chunk_sent.wait(timeout=2.0) # Interrupt while TTS is waiting to emit another chunk. websocket.send_json( { "type": "interrupt", "turn_id": turn_id, } ) interrupted = websocket.receive_json() assert interrupted["type"] == "interrupted" assert interrupted["turn_id"] == turn_id assert tts.cancelled.is_set() assert created_turn_managers[0].active_turn is None # A new turn must be possible after cancellation. websocket.send_json({"type": "audio.start"}) next_turn = websocket.receive_json() assert next_turn["type"] == "turn.started" assert next_turn["turn_id"] == 2 def test_websocket_uses_distinct_stt_per_turn(): """Each audio turn must receive a fresh STT adapter.""" created_stt.clear() 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, }, } ) assert websocket.receive_json()["type"] == "session.ready" websocket.send_json({"type": "audio.start"}) first_turn = websocket.receive_json() assert first_turn["type"] == "turn.started" assert len(created_stt) == 1 first_stt = created_stt[0] websocket.send_json( { "type": "interrupt", "turn_id": first_turn["turn_id"], } ) assert websocket.receive_json()["type"] == "interrupted" websocket.send_json({"type": "audio.start"}) second_turn = websocket.receive_json() assert second_turn["type"] == "turn.started" assert len(created_stt) == 2 second_stt = created_stt[1] assert first_stt is not second_stt assert first_stt._started is False assert second_stt._started is True def test_websocket_audio_start_rolls_back_failed_stt(): """ Verify rollback when STT initialization fails. The Gateway must abort the partially initialized provider, release semantic turn state, and allow a new attempt. """ created_stt.clear() created_turn_managers.clear() class FailingStartSTTAdapter(MockSTTAdapter): def __init__(self): super().__init__() self.abort_called = False async def start_session(self): await super().start_session() raise RuntimeError( "Simulated STT initialization failure." ) async def abort_session(self): self.abort_called = True await super().abort_session() failing_stt = FailingStartSTTAdapter() successful_stt = MockSTTAdapter() adapters = iter([failing_stt, successful_stt]) def create_test_stt(config=None, catalog=None): return next(adapters) with patch.object( main, "create_stt_adapter", side_effect=create_test_stt, ): 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, }, } ) assert websocket.receive_json()["type"] == "session.ready" websocket.send_json({"type": "audio.start"}) error = websocket.receive_json() assert error["type"] == "error" assert error["code"] == "STT_START_FAILED" assert "Simulated STT initialization failure" in error["message"] assert created_turn_managers[0].active_turn is None assert failing_stt.abort_called is True assert failing_stt._started is False websocket.send_json({"type": "audio.start"}) next_turn = websocket.receive_json() assert next_turn["type"] == "turn.started", ( f"Second audio.start failed: {next_turn!r}" ) assert next_turn["turn_id"] == 2 assert successful_stt._started is True def test_websocket_audio_start_rolls_back_task_registration_failure(): """ Verify rollback when turn task registration fails. Responsibilities: - Simulate failure after the STT task is created. - Verify the unregistered task is terminated. - Verify STT resources are released. - Verify semantic turn state is released. - Verify the next turn can start. """ created_stt.clear() created_turn_managers.clear() real_register = VoicePipeline.register_turn_task registered_tasks = [] registration_attempts = 0 def failing_register(pipeline, task): nonlocal registration_attempts registration_attempts += 1 registered_tasks.append(task) if registration_attempts == 1: raise RuntimeError( "Simulated turn task registration failure." ) return real_register(pipeline, task) with patch.object( VoicePipeline, "register_turn_task", autospec=True, side_effect=failing_register, ): 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"}) error = websocket.receive_json() assert error["type"] == "error" assert error["code"] == "STT_START_FAILED" assert "Simulated turn task registration failure" in ( error["message"] ) # The task created before registration must be terminated. assert len(registered_tasks) == 1 first_task = registered_tasks[0] assert first_task.done() assert first_task.cancelled() # The failed STT session must be released. assert len(created_stt) == 1 assert created_stt[0]._started is False # The semantic turn must not remain active. assert created_turn_managers[0].active_turn is None # A subsequent turn must be able to start normally. websocket.send_json({"type": "audio.start"}) next_turn = websocket.receive_json() assert next_turn["type"] == "turn.started", ( f"Second audio.start failed: {next_turn!r}" ) assert next_turn["turn_id"] == 2 assert registration_attempts == 2 assert len(created_stt) == 2 assert created_stt[1]._started is True def test_websocket_audio_start_abort_failure_is_not_retryable(): """ Verify that failed STT cleanup prevents unsafe retry. Responsibilities: - Simulate partial STT initialization failure. - Simulate abort_session() failure. - Verify STT_START_FAILED is not retryable. - Verify the semantic turn remains active. - Verify another audio.start is rejected. """ created_turn_managers.clear() class FailingCleanupSTTAdapter(MockSTTAdapter): async def start_session(self): await super().start_session() raise RuntimeError("Simulated STT startup failure.") async def abort_session(self): raise RuntimeError("Simulated STT abort failure.") failing_stt = FailingCleanupSTTAdapter() with ( patch.object( main, "create_stt_adapter", return_value=failing_stt, ), patch.object( main.logger, "exception", ) as log_mock, ): 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, }, } ) assert websocket.receive_json()["type"] == "session.ready" websocket.send_json({"type": "audio.start"}) error = websocket.receive_json() assert error["type"] == "error" assert error["code"] == "STT_START_FAILED" assert error["retryable"] is False assert created_turn_managers[0].active_turn is not None assert created_turn_managers[0].active_turn.turn_id == 1 assert log_mock.called websocket.send_json({"type": "audio.start"}) next_response = websocket.receive_json() assert next_response["type"] == "error" assert next_response["code"] == "INVALID_TURN" @pytest.mark.anyio async def test_forward_stt_events_discards_stale_turn(): """ Events belonging to an interrupted turn must not be forwarded or processed after another turn becomes active. """ websocket = AsyncMock() pipeline = Mock(spec=VoicePipeline) pipeline.is_active_turn.return_value = False pipeline.process_transcript = AsyncMock() class LateSTT: async def events(self): yield TranscriptEvent( type="final", text="Transcrição atrasada", ) stt = LateSTT() await forward_stt_events( websocket=websocket, stt=stt, pipeline=pipeline, session_id="test-session", turn_id=1, ) pipeline.is_active_turn.assert_called_once_with(1) websocket.send_text.assert_not_awaited() websocket.send_bytes.assert_not_awaited() pipeline.on_final.assert_not_called() pipeline.process_transcript.assert_not_awaited() pipeline.synthesize_response.assert_not_called() await forward_stt_events( websocket=websocket, stt=stt, pipeline=pipeline, session_id="test-session", turn_id=1, ) websocket.send_text.assert_not_awaited() websocket.send_bytes.assert_not_awaited() pipeline.on_final.assert_not_called() pipeline.process_transcript.assert_not_awaited() pipeline.synthesize_response.assert_not_called() @pytest.mark.anyio async def test_stt_event_send_cancellation_prevents_late_delivery(): """ Verify that cancellation stops a blocked STT event send. The old turn must not deliver its event after cancellation. """ send_started = asyncio.Event() send_cancelled = asyncio.Event() class BlockingWebSocket: def __init__(self): self.delivered = [] async def send_text(self, data): send_started.set() try: await asyncio.Event().wait() except asyncio.CancelledError: send_cancelled.set() raise self.delivered.append(data) class SingleEventSTT: async def events(self): yield TranscriptEvent( type="partial", text="Evento antigo", ) websocket = BlockingWebSocket() stt = SingleEventSTT() session = main.SessionState() turn_manager = TurnManager() pipeline = VoicePipeline( session=session, turn_manager=turn_manager, agent=MockAgentAdapter(), tts=MockTTSAdapter(), ) turn = pipeline.start_turn() task = asyncio.create_task( forward_stt_events( websocket=websocket, stt=stt, pipeline=pipeline, session_id=session.session_id, turn_id=turn.turn_id, ) ) pipeline.register_turn_task(task) try: await asyncio.wait_for( send_started.wait(), timeout=1.0, ) await pipeline.cancel_turn_task(timeout=1.0) assert send_cancelled.is_set() assert task.done() assert task.cancelled() assert pipeline.turn_task is None assert websocket.delivered == [] pipeline.interrupt_turn(turn.turn_id) assert turn_manager.active_turn is None assert session.active_turn_id is None finally: if not task.done(): task.cancel() await asyncio.gather( task, return_exceptions=True, ) def test_websocket_registration_failure_awaits_task_before_abort(): """ Verify that registration rollback terminates an unregistered task before aborting its STT provider. """ created_stt.clear() real_register = VoicePipeline.register_turn_task real_abort = MockSTTAdapter.abort_session tasks = [] abort_checked = [] def failing_register(pipeline, task): tasks.append(task) raise RuntimeError("Simulated registration failure.") async def tracked_abort(stt): assert tasks assert tasks[0].done() abort_checked.append(True) await real_abort(stt) with ( patch.object( VoicePipeline, "register_turn_task", autospec=True, side_effect=failing_register, ), patch.object( MockSTTAdapter, "abort_session", autospec=True, side_effect=tracked_abort, ), ): 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, }, } ) assert websocket.receive_json()["type"] == "session.ready" websocket.send_json({"type": "audio.start"}) response = websocket.receive_json() assert response["type"] == "error" assert response["code"] == "STT_START_FAILED" 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" def test_websocket_interrupt_during_streaming_tts(): """ Verify that interrupt cancels an active streaming TTS turn while normal turn finalization is pending. """ created_stt.clear() tts_started = threading.Event() tts_cancelled = threading.Event() class SlowStreamingAgent: async def stream(self, session_id, turn_id, text): yield "Olá, esta é uma resposta longa." class SlowStreamingTTS: async def synthesize(self, text): try: tts_started.set() yield b"\x01\x02\x03\x04" # Keep synthesis active until cancellation. await asyncio.Event().wait() except asyncio.CancelledError: tts_cancelled.set() raise with patch.object( main, "create_agent_adapter", return_value=SlowStreamingAgent(), ): with patch.object( main, "create_tts_adapter", return_value=SlowStreamingTTS(), ): 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, }, }) ready = websocket.receive_json() assert ready["type"] == "session.ready" websocket.send_json({"type": "audio.start"}) started = websocket.receive_json() assert started["type"] == "turn.started" turn_id = started["turn_id"] stt = created_stt[-1] stt.emit_final_now("Olá") # Wait until the Gateway delivers the first TTS audio chunk. first_audio_received = False for _ in range(20): message = websocket.receive() if message.get("bytes") is not None: assert message["bytes"] == b"\x01\x02\x03\x04" first_audio_received = True break import json event = json.loads(message["text"]) if event["type"] == "error": raise AssertionError( f"Gateway returned error: {event}" ) assert first_audio_received assert tts_started.is_set() websocket.send_json({ "type": "audio.stop", "turn_id": turn_id, }) websocket.send_json({ "type": "interrupt", "turn_id": turn_id, }) # Streaming events may precede interruption. for _ in range(20): message = websocket.receive() if message.get("bytes") is not None: continue import json event = json.loads(message["text"]) if event["type"] == "interrupted": assert event["turn_id"] == turn_id break if event["type"] == "error": raise AssertionError( f"Gateway returned error: {event}" ) else: raise AssertionError( "Gateway did not confirm interruption." ) assert tts_cancelled.wait(timeout=2.0) # A new turn must be accepted in the same session. websocket.send_json({"type": "audio.start"}) next_turn = websocket.receive_json() assert next_turn["type"] == "turn.started" assert next_turn["turn_id"] != turn_id websocket.send_json({ "type": "interrupt", "turn_id": next_turn["turn_id"], }) interrupted = websocket.receive_json() assert interrupted["type"] == "interrupted" def test_websocket_session_end_during_streaming_tts(): """ Verify session.end cancels an active streaming TTS turn. The Gateway must cancel provider execution before acknowledging session closure. """ import json created_stt.clear() tts_cancelled = threading.Event() class SlowStreamingAgent: async def stream(self, session_id, turn_id, text): yield "Esta é uma resposta longa." class SlowStreamingTTS: async def synthesize(self, text): try: yield b"\x01\x02\x03\x04" # Keep synthesis running until cancelled. await asyncio.Event().wait() except asyncio.CancelledError: tts_cancelled.set() raise with patch.object( main, "create_agent_adapter", return_value=SlowStreamingAgent(), ): with patch.object( main, "create_tts_adapter", return_value=SlowStreamingTTS(), ): 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, }, }) ready = websocket.receive_json() assert ready["type"] == "session.ready" websocket.send_json({"type": "audio.start"}) started = websocket.receive_json() assert started["type"] == "turn.started" turn_id = started["turn_id"] stt = created_stt[-1] stt.emit_final_now("Olá") # Wait until TTS audio reaches the client. first_audio_received = False for _ in range(20): message = websocket.receive() if message.get("bytes") is not None: assert message["bytes"] == b"\x01\x02\x03\x04" first_audio_received = True break event = json.loads(message["text"]) if event["type"] == "error": raise AssertionError( f"Gateway returned error: {event}" ) assert first_audio_received websocket.send_json({ "type": "audio.stop", "turn_id": turn_id, }) websocket.send_json({ "type": "session.end", }) # Streaming events may already be queued. for _ in range(20): message = websocket.receive() if message.get("bytes") is not None: continue event = json.loads(message["text"]) if event["type"] == "session.ended": assert event["session_id"] == ready["session_id"] break if event["type"] == "turn.ended": raise AssertionError( "Turn completed normally during session shutdown." ) if event["type"] == "error": raise AssertionError( f"Gateway returned error: {event}" ) else: raise AssertionError( "Gateway did not confirm session closure." ) assert tts_cancelled.wait(timeout=2.0) def test_websocket_interrupt_after_turn_completed(): """ Verify behavior when interrupt arrives after turn.ended. A completed turn must not be cancelled again. """ created_stt.clear() 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, }, }) ready = websocket.receive_json() assert ready["type"] == "session.ready" websocket.send_json({"type": "audio.start"}) started = websocket.receive_json() assert started["type"] == "turn.started" turn_id = started["turn_id"] stt = created_stt[-1] stt.emit_final_now("Olá") # Consume events until the first turn finishes. for _ in range(30): message = websocket.receive() if message.get("bytes") is not None: continue import json event = json.loads(message["text"]) if event["type"] == "error": raise AssertionError( f"Gateway returned error: {event}" ) if event["type"] == "stt.final": websocket.send_json({ "type": "audio.stop", "turn_id": turn_id, }) if event["type"] == "turn.ended": break else: raise AssertionError( "Turn did not complete normally." ) # Interrupt the already completed turn. websocket.send_json({ "type": "interrupt", "turn_id": turn_id, }) response = websocket.receive_json() # Existing Gateway behavior may acknowledge stale IDs. # Capture the current contract before changing it. assert response["type"] == "error" assert response["code"] == "INVALID_TURN" assert response["turn_id"] == turn_id def test_websocket_interrupt_wins_pending_tts_completion(): """ Verify interruption wins while TTS completion is pending. The interrupted turn must not emit turn.ended, and a new turn must be accepted. """ import json created_stt.clear() tts_cancelled = threading.Event() class ControlledAgent: async def stream(self, session_id, turn_id, text): yield "Esta resposta será interrompida." class ControlledTTS: async def synthesize(self, text): try: yield b"\x01\x02\x03\x04" # Completion remains pending until cancellation. await asyncio.Event().wait() except asyncio.CancelledError: tts_cancelled.set() raise with patch.object( main, "create_agent_adapter", return_value=ControlledAgent(), ): with patch.object( main, "create_tts_adapter", return_value=ControlledTTS(), ): 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, }, }) ready = websocket.receive_json() assert ready["type"] == "session.ready" websocket.send_json({ "type": "audio.start", }) started = websocket.receive_json() assert started["type"] == "turn.started" turn_id = started["turn_id"] stt = created_stt[-1] stt.emit_final_now("Olá") # Observe the first PCM chunk. for _ in range(20): message = websocket.receive() if message.get("bytes") is not None: assert message["bytes"] == b"\x01\x02\x03\x04" break event = json.loads(message["text"]) if event["type"] == "error": raise AssertionError(event) else: raise AssertionError( "TTS did not produce its first chunk." ) websocket.send_json({ "type": "audio.stop", "turn_id": turn_id, }) websocket.send_json({ "type": "interrupt", "turn_id": turn_id, }) terminal_events = [] for _ in range(20): message = websocket.receive() if message.get("bytes") is not None: continue event = json.loads(message["text"]) if event["type"] in { "turn.ended", "interrupted", }: terminal_events.append(event["type"]) if event["type"] == "error": raise AssertionError(event) if event["type"] == "interrupted": break else: raise AssertionError( "Interruption was not acknowledged." ) assert terminal_events == ["interrupted"] assert tts_cancelled.wait(timeout=2.0) # Confirm that the session accepts another turn. websocket.send_json({ "type": "audio.start", }) next_turn = websocket.receive_json() assert next_turn["type"] == "turn.started" assert next_turn["turn_id"] != turn_id websocket.send_json({ "type": "interrupt", "turn_id": next_turn["turn_id"], }) response = websocket.receive_json() assert response["type"] == "interrupted" assert response["turn_id"] == next_turn["turn_id"] @pytest.mark.parametrize("session_started", [False, True]) @pytest.mark.parametrize( "payload", [ None, [], "text", 42, True, {}, {"type": None}, {"type": 42}, {"type": ""}, {"type": " "}, ], ) def test_websocket_rejects_invalid_message_and_remains_usable( payload, session_started, ): with client.websocket_connect("/ws/audio") as websocket: if session_started: websocket.send_json({"type": "session.start"}) assert websocket.receive_json()["type"] == "session.ready" websocket.send_json(payload) error = websocket.receive_json() assert error["type"] == "error" assert error["code"] == "INVALID_MESSAGE" if session_started: websocket.send_json({"type": "session.end"}) assert websocket.receive_json()["type"] == "session.ended" else: websocket.send_json({"type": "session.start"}) assert websocket.receive_json()["type"] == "session.ready" @pytest.mark.parametrize( "event_type", [ "audio.start", "audio.chunk", "audio.stop", "interrupt", "session.end", ], ) def test_websocket_requires_session_before_control_events(event_type): with client.websocket_connect("/ws/audio") as websocket: websocket.send_json( { "type": event_type, "turn_id": 1, "sequence_number": 0, } ) error = websocket.receive_json() assert error["type"] == "error" assert error["code"] == "SESSION_NOT_FOUND" websocket.send_json({"type": "session.start"}) assert websocket.receive_json()["type"] == "session.ready" @pytest.mark.parametrize( "text", ["{", "", '{type: "session.start"}'], ) def test_websocket_rejects_malformed_json_and_remains_usable(text): with client.websocket_connect("/ws/audio") as websocket: websocket.send_text(text) error = websocket.receive_json() assert error["type"] == "error" assert error["code"] == "INVALID_MESSAGE" websocket.send_json({"type": "session.start"}) assert websocket.receive_json()["type"] == "session.ready" def test_websocket_duplicate_audio_start_preserves_active_turn(): created_stt.clear() created_turn_managers.clear() with client.websocket_connect("/ws/audio") as websocket: websocket.send_json({"type": "session.start"}) assert websocket.receive_json()["type"] == "session.ready" websocket.send_json({"type": "audio.start"}) started = websocket.receive_json() assert started["type"] == "turn.started" turn_id = started["turn_id"] stt = created_stt[0] turn = created_turn_managers[0].active_turn websocket.send_json({"type": "audio.start"}) error = websocket.receive_json() assert error["type"] == "error" assert error["code"] == "INVALID_TURN" assert error["turn_id"] == turn_id assert len(created_stt) == 1 assert stt._started is True assert created_turn_managers[0].active_turn is turn assert turn.state == TurnState.LISTENING websocket.send_json( {"type": "interrupt", "turn_id": turn_id} ) assert websocket.receive_json()["type"] == "interrupted" assert stt._started is False assert created_turn_managers[0].active_turn is None