diff --git a/src/voice_assistant/audio_gateway/app/main.py b/src/voice_assistant/audio_gateway/app/main.py index 22325af..61c4c7b 100644 --- a/src/voice_assistant/audio_gateway/app/main.py +++ b/src/voice_assistant/audio_gateway/app/main.py @@ -554,14 +554,24 @@ async def audio_socket(websocket: WebSocket) -> None: ) continue - event_type = payload.get("type") - - if not isinstance(event_type, str): + if not isinstance(payload, dict): await send_json( websocket, error_event( "INVALID_MESSAGE", - "Field 'type' is required.", + "Message must be a JSON object.", + ), + ) + continue + + event_type = payload.get("type") + + if not isinstance(event_type, str) or not event_type.strip(): + await send_json( + websocket, + error_event( + "INVALID_MESSAGE", + "Field 'type' must be a non-empty string.", ), ) continue @@ -654,6 +664,16 @@ async def audio_socket(websocket: WebSocket) -> None: # --------------------------------------------------------- # Todos os eventos abaixo exigem sessão ativa # --------------------------------------------------------- + if session is None: + await send_json( + websocket, + error_event( + "SESSION_NOT_FOUND", + "session.start is required before this event.", + ), + ) + continue + if session.closed: await send_json( websocket, @@ -669,6 +689,18 @@ async def audio_socket(websocket: WebSocket) -> None: # audio.start # --------------------------------------------------------- if event_type == "audio.start": + if session.active_turn_id is not None: + await send_json( + websocket, + error_event( + "INVALID_TURN", + "A turn is already active.", + session_id=session.session_id, + turn_id=session.active_turn_id, + ), + ) + continue + turn_id: int | None = None try: diff --git a/tests/test_gateway.py b/tests/test_gateway.py index 95e20f4..93a4c5a 100644 --- a/tests/test_gateway.py +++ b/tests/test_gateway.py @@ -1044,7 +1044,7 @@ def test_websocket_interrupt_timeout_preserves_active_turn(): next_response = websocket.receive_json() assert next_response["type"] == "error" - assert next_response["code"] == "STT_START_FAILED" + assert next_response["code"] == "INVALID_TURN" def test_websocket_interrupt_during_agent(): @@ -1625,7 +1625,7 @@ def test_websocket_audio_start_abort_failure_is_not_retryable(): next_response = websocket.receive_json() assert next_response["type"] == "error" - assert next_response["code"] == "STT_START_FAILED" + assert next_response["code"] == "INVALID_TURN" @@ -2410,3 +2410,121 @@ def test_websocket_interrupt_wins_pending_tts_completion(): 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