voiceagent/tests/test_session_turn_lifecycle.py

"""
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.domain.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