feat: establish voice assistant V1 baseline

tests/test_session_turn_lifecycle.py

@@ -0,0 +1,281 @@
+"""
+Tests for SessionState and TurnManager lifecycle ownership.
+
+Purpose:
+    Define and verify the separation between session lifecycle state
+    and semantic turn state.
+
+Responsibilities:
+    - Verify SessionState owns session and audio sequencing state.
+    - Verify TurnManager owns semantic turn transitions.
+    - Verify the expected lifecycle of a complete turn.
+    - Verify interruption and invalid transitions.
+
+Non-responsibilities:
+    - Testing STT, Agent, or TTS providers.
+    - Testing WebSocket transport.
+    - Testing provider configuration.
+"""
+
+import pytest
+
+from voice_assistant.audio_gateway.app.session import SessionState
+from voice_assistant.turn_manager import TurnManager, TurnState
+
+
+def test_session_creates_sequential_turn_ids():
+    session = SessionState()
+
+    first_turn = session.start_turn()
+
+    assert first_turn == 1
+    assert session.active_turn_id == 1
+    assert session.expected_audio_sequence == 0
+
+    session.end_turn(first_turn)
+
+    second_turn = session.start_turn()
+
+    assert second_turn == 2
+    assert session.active_turn_id == 2
+    assert session.expected_audio_sequence == 0
+
+
+def test_session_owns_audio_sequence():
+    session = SessionState()
+
+    turn_id = session.start_turn()
+
+    session.accept_audio_chunk(
+        turn_id=turn_id,
+        sequence_number=0,
+        payload_size=320,
+    )
+
+    assert session.expected_audio_sequence == 1
+
+    session.accept_audio_chunk(
+        turn_id=turn_id,
+        sequence_number=1,
+        payload_size=320,
+    )
+
+    assert session.expected_audio_sequence == 2
+
+
+def test_session_rejects_out_of_order_audio():
+    session = SessionState()
+
+    turn_id = session.start_turn()
+
+    with pytest.raises(
+        ValueError,
+        match="Expected sequence_number=0",
+    ):
+        session.accept_audio_chunk(
+            turn_id=turn_id,
+            sequence_number=1,
+            payload_size=320,
+        )
+
+
+def test_turn_manager_owns_semantic_turn_state():
+    manager = TurnManager()
+
+    turn = manager.start(turn_id=1)
+
+    assert turn.turn_id == 1
+    assert turn.state == TurnState.LISTENING
+
+    manager.on_partial(
+        turn_id=1,
+        text="Olá",
+    )
+
+    assert turn.state == TurnState.LISTENING
+
+    manager.on_final(
+        turn_id=1,
+        text="Olá!",
+    )
+
+    assert turn.state == TurnState.PROCESSING
+
+    manager.on_agent_response(turn_id=1)
+
+    assert turn.state == TurnState.SPEAKING
+
+    manager.on_tts_complete(turn_id=1)
+
+    assert turn.state == TurnState.COMPLETED
+
+
+def test_turn_manager_rejects_invalid_transition():
+    manager = TurnManager()
+
+    manager.start(turn_id=1)
+
+    with pytest.raises(
+        RuntimeError,
+        match="Cannot start speaking",
+    ):
+        manager.on_agent_response(turn_id=1)
+
+
+def test_turn_manager_interrupts_active_turn():
+    manager = TurnManager()
+
+    manager.start(turn_id=1)
+
+    turn = manager.interrupt(turn_id=1)
+
+    assert turn.state == TurnState.INTERRUPTED
+    assert manager.active_turn is turn
+
+
+def test_turn_manager_stop_requires_completed_state():
+    manager = TurnManager()
+
+    manager.start(turn_id=1)
+
+    manager.on_final(
+        turn_id=1,
+        text="Olá!",
+    )
+
+    with pytest.raises(
+        RuntimeError,
+        match="Cannot stop turn while turn is processing",
+    ):
+        manager.stop(turn_id=1)
+
+
+def test_turn_manager_stop_allows_completed_turn():
+    manager = TurnManager()
+
+    manager.start(turn_id=1)
+
+    manager.on_final(
+        turn_id=1,
+        text="Olá!",
+    )
+
+    manager.on_agent_response(turn_id=1)
+
+    manager.on_tts_complete(turn_id=1)
+
+    turn = manager.stop(turn_id=1)
+
+    assert turn.state == TurnState.COMPLETED
+    assert turn.final_text == "Olá!"
+
+def test_session_and_turn_manager_complete_same_turn():
+    session = SessionState()
+    manager = TurnManager()
+
+    turn_id = session.start_turn()
+    turn = manager.start(turn_id)
+
+    assert session.active_turn_id == turn_id
+    assert turn.turn_id == turn_id
+    assert turn.state == TurnState.LISTENING
+
+    manager.on_final(
+        turn_id,
+        "Olá!",
+    )
+
+    manager.on_agent_response(turn_id)
+    manager.on_tts_complete(turn_id)
+
+    completed_turn = manager.stop(turn_id)
+
+    assert completed_turn.state == TurnState.COMPLETED
+
+    manager.clear(turn_id)
+    session.end_turn(turn_id)
+
+    assert manager.active_turn is None
+    assert session.active_turn_id is None
+
+def test_session_and_turn_manager_must_use_same_turn_id():
+    session = SessionState()
+    manager = TurnManager()
+
+    session_turn_id = session.start_turn()
+    manager.start(1)
+
+    with pytest.raises(
+        ValueError,
+        match="Unexpected turn_id=2",
+    ):
+        session.accept_audio_chunk(
+            turn_id=2,
+            sequence_number=0,
+            payload_size=320,
+        )
+
+    assert session.active_turn_id == session_turn_id
+    assert manager.active_turn is not None
+    assert manager.active_turn.turn_id == 1
+
+
+def test_session_ends_turn_after_turn_manager_completion():
+    session = SessionState()
+    manager = TurnManager()
+
+    turn_id = session.start_turn()
+    manager.start(turn_id)
+
+    manager.on_final(
+        turn_id,
+        "Olá!",
+    )
+
+    manager.on_agent_response(turn_id)
+    manager.on_tts_complete(turn_id)
+
+    manager.stop(turn_id)
+    manager.clear(turn_id)
+    session.end_turn(turn_id)
+
+    assert session.active_turn_id is None
+
+
+def test_session_and_turn_manager_interrupt_same_turn():
+    session = SessionState()
+    manager = TurnManager()
+
+    turn_id = session.start_turn()
+    manager.start(turn_id)
+
+    session.interrupt(turn_id)
+    interrupted_turn = manager.interrupt(turn_id)
+
+    assert session.active_turn_id is None
+    assert manager.active_turn is interrupted_turn
+    assert interrupted_turn.turn_id == turn_id
+    assert interrupted_turn.state == TurnState.INTERRUPTED
+
+
+def test_session_can_start_next_turn_after_interrupt():
+    session = SessionState()
+    manager = TurnManager()
+
+    first_turn_id = session.start_turn()
+    manager.start(first_turn_id)
+
+    session.interrupt(first_turn_id)
+    manager.interrupt(first_turn_id)
+    manager.clear(first_turn_id)
+
+    second_turn_id = session.start_turn()
+    manager.start(second_turn_id)
+
+    assert second_turn_id == 2
+    assert session.active_turn_id == 2
+    assert manager.active_turn is not None
+    assert manager.active_turn.turn_id == 2
+    assert manager.active_turn.state == TurnState.LISTENING
+
+