Bolt-Pet/tests/test_speech.py

"""The rules about talking — the four kinds of utterance and the mic policy.

These were four near-copies in the controller before, and the copies had
drifted: one didn't arm barge-in, one skipped the follow-up rule. The value of
having one `Speaker` is only real if the differences between the kinds stay
*visible*, so this asserts on the differences rather than on the machinery.
"""

import sys
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from bolt_pet import config, speech
from bolt_pet.speech import Speaker, Utterance
from bolt_pet.state import PetState, PetStateMachine


class FakeTts:
    def __init__(self, completed=True):
        self.completed = completed
        self.calls = []

    def speak(self, text, on_error=None, should_stop=None, voice_id=None, on_level=None):
        self.calls.append({"text": text, "voice_id": voice_id,
                           "interruptible": should_stop is not None})
        if on_level is not None:
            on_level(0.7)
        return self.completed

    def play_pcm(self, pcm, sample_rate, should_stop=None, on_level=None):
        self.calls.append({"pcm": pcm, "rate": sample_rate,
                           "interruptible": should_stop is not None})
        return self.completed


class FakeBargeIn:
    def __init__(self):
        self.resets = 0

    def reset(self):
        self.resets += 1

    def check(self, _frame=None):
        return False


def build(**kwargs):
    """A speaker plus the things worth asserting on."""
    state = PetStateMachine()
    tts = kwargs.pop("tts", None) or FakeTts()
    said, logged, recorded, levels = [], [], [], []
    speaker = Speaker(
        state=state, tts=tts,
        history=recorded.append,
        on_said=said.append,
        on_log=logged.append,
        on_level=levels.append,
        **kwargs,
    )
    return speaker, state, tts, said, logged, recorded, levels


# ── what distinguishes the four kinds ───────────────────────────────────────

def test_a_reply_is_recorded_but_a_holding_line_is_not():
    """Filler must not push the actual answer out of the transcript."""
    speaker, state, _tts, _said, _logged, recorded, _levels = build()
    state.transition(PetState.LISTENING)
    state.transition(PetState.THINKING)

    speaker.say(Utterance.holding("Give me a sec."))
    speaker.say(Utterance.reply("Sixty percent used."))

    assert recorded == ["Sixty percent used."]


def test_a_holding_line_resumes_the_turn_it_interrupted():
    """The turn isn't over — a tool is still running — so it must go back to
    THINKING, not drop to IDLE and end the turn."""
    speaker, state, _tts, *_ = build()
    state.transition(PetState.LISTENING)
    state.transition(PetState.THINKING)

    speaker.say(Utterance.holding("Let me check."))

    assert state.state == PetState.THINKING


def test_a_holding_line_cannot_be_talked_over():
    """Cutting off "give me a sec" strands the tool that's already running."""
    detector = FakeBargeIn()
    speaker, state, tts, *_ = build(barge_in=lambda: detector)
    state.transition(PetState.LISTENING)
    state.transition(PetState.THINKING)

    speaker.say(Utterance.holding("One sec."))
    assert tts.calls[-1]["interruptible"] is False

    speaker.say(Utterance.reply("Done."))
    assert tts.calls[-1]["interruptible"] is True


def test_a_streamed_sentence_stays_talking_between_sentences():
    """Otherwise the sprite flickers idle-talking-idle down a long answer."""
    speaker, state, *_ = build()
    state.transition(PetState.LISTENING)
    state.transition(PetState.THINKING)

    speaker.say(Utterance.stream_chunk("The disk is fine."))

    assert state.state == PetState.TALKING


def test_a_scene_is_recorded_because_the_user_heard_it():
    speaker, state, _tts, _said, _logged, recorded, _levels = build()
    state.transition(PetState.LISTENING)
    state.transition(PetState.THINKING)

    speaker.say_pcm(Utterance.scene("Once upon a time."), pcm=b"\x00\x00", sample_rate=44100)

    assert recorded == ["Once upon a time."]
    assert state.state == PetState.THINKING


# ── barge-in ordering, which is what the copies got wrong ───────────────────

def test_the_detector_is_reset_after_playback_not_only_before():
    """Playback fed the pet's own voice into the wake model's window. If it
    isn't cleared afterwards, the idle listener re-hears the last sentence and
    the pet answers itself."""
    detector = FakeBargeIn()
    speaker, state, *_ = build(barge_in=lambda: detector)
    state.transition(PetState.LISTENING)
    state.transition(PetState.THINKING)

    speaker.say(Utterance.reply("Hello there."))

    assert detector.resets == 2  # armed before, cleared after


def test_the_barge_in_detail_is_captured_before_the_reset():
    """Read it after the reset and every interruption reports zeroed counters —
    which reads like hard evidence and is nothing of the sort."""
    detector = FakeBargeIn()
    details = ["score 0.81 at frame 12", "score 0.000 at frame 0"]
    speaker, state, *_ = build(barge_in=lambda: detector,
                               detail_of=lambda: details.pop(0))
    state.transition(PetState.LISTENING)
    state.transition(PetState.THINKING)

    speaker.say(Utterance.reply("Hello there."))

    assert speaker.last_detail == "score 0.81 at frame 12"


def test_a_detector_that_appears_late_is_still_used():
    """The detector is built after the speaker — it needs the mic stream — so
    it's read through a callable. Holding a copy is how the two drift apart."""
    detector = None
    speaker, state, tts, *_ = build(barge_in=lambda: detector)
    state.transition(PetState.LISTENING)
    state.transition(PetState.THINKING)

    speaker.say(Utterance.reply("Before."))
    assert tts.calls[-1]["interruptible"] is False

    detector = FakeBargeIn()
    speaker.say(Utterance.reply("After."))
    assert tts.calls[-1]["interruptible"] is True


# ── the mouth ───────────────────────────────────────────────────────────────

def test_the_mouth_is_closed_when_the_line_ends():
    """A pet left mid-vowel after the audio stops looks broken."""
    speaker, state, _tts, _said, _logged, _recorded, levels = build()
    state.transition(PetState.LISTENING)
    state.transition(PetState.THINKING)

    speaker.say(Utterance.reply("Talking."))

    assert levels[0] > 0
    assert levels[-1] == 0.0


# ── text handling ───────────────────────────────────────────────────────────

def test_an_empty_line_is_not_spoken_at_all():
    """A reply that is nothing but an audio tag reduces to '' — and a silent
    bubble with no audio is better than the pet announcing "laughing"."""
    speaker, state, tts, said, *_ = build()

    assert speaker.say(Utterance.reply("[laughing]")) is True
    assert tts.calls == []
    assert said == []
    assert state.state == PetState.IDLE


def test_the_bubble_gets_display_text_and_tts_gets_the_original():
    """The bubble keeps emoji and drops markdown; TTS does its own stripping
    (inside speak(), so every path is covered) and needs the real text."""
    speaker, state, tts, said, *_ = build()
    state.transition(PetState.LISTENING)
    state.transition(PetState.THINKING)

    speaker.say(Utterance.reply("**Sixty** percent 🎉"))

    assert said == ["Sixty percent 🎉"]
    assert tts.calls[-1]["text"] == "**Sixty** percent 🎉"


def test_a_voice_override_reaches_tts():
    speaker, state, tts, *_ = build(voice_id=lambda: "voice-abc")
    state.transition(PetState.LISTENING)
    state.transition(PetState.THINKING)

    speaker.say(Utterance.reply("In character."))

    assert tts.calls[-1]["voice_id"] == "voice-abc"


# ── the follow-up rule ──────────────────────────────────────────────────────

@pytest.fixture
def follow_ups_on(monkeypatch):
    monkeypatch.setattr(config, "FOLLOW_UP_LISTEN", True)
    monkeypatch.setattr(config, "FOLLOW_UP_MAX_TURNS", 3)


def test_an_interruption_always_reopens_the_mic(follow_ups_on):
    """You talked over it — you are mid-sentence, so it has to listen."""
    keep, why = speech.follow_up_decision("Anything else?", completed=False, follow_ups=99)
    assert keep is True
    assert why == "interrupted"


def test_a_question_keeps_the_mic_open_without_the_wake_word(follow_ups_on):
    keep, why = speech.follow_up_decision("Want me to check?", completed=True, follow_ups=0)
    assert (keep, why) == (True, "question")


def test_a_statement_ends_the_turn(follow_ups_on):
    keep, _why = speech.follow_up_decision("Sixty percent used.", completed=True, follow_ups=0)
    assert keep is False


def test_the_cap_stops_a_server_that_ends_every_reply_with_a_question(follow_ups_on):
    """Otherwise mic noise loops it forever."""
    assert speech.follow_up_decision("Ok?", completed=True, follow_ups=2)[0] is True
    keep, why = speech.follow_up_decision("Ok?", completed=True, follow_ups=3)
    assert keep is False
    assert "cap" in why


def test_the_rule_can_be_switched_off(monkeypatch):
    monkeypatch.setattr(config, "FOLLOW_UP_LISTEN", False)
    assert speech.follow_up_decision("Ok?", completed=True, follow_ups=0)[0] is False