"""Controller-level wiring for the newer behaviours: petctl routing,
barge-in follow-up, quiet hours, screen context, notification forwarding and
the live wake threshold.

Needs a QApplication (signals), so run with QT_QPA_PLATFORM=offscreen.
"""

import json
import sys
from pathlib import Path

import numpy as np
import pytest
from PySide6.QtWidgets import QApplication

from bolt_pet import controller as controller_mod
from bolt_pet.notifications import Notification
from bolt_pet.server_client import Reply
from bolt_pet.state import PetState

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

_app = QApplication.instance() or QApplication(["test"])


@pytest.fixture(autouse=True)
def no_screen_probes(monkeypatch):
    monkeypatch.setattr(controller_mod.screen_context, "is_fullscreen_active", lambda: False)
    monkeypatch.setattr(controller_mod.screen_context, "context_for", lambda text: text)


@pytest.fixture(autouse=True)
def _no_streaming(monkeypatch):
    """Most tests drive the plain request/response path; leaving streaming on
    would have them attempt a real HTTP call, fail, and fall back — passing,
    slowly, for the wrong reason. The streaming path has its own tests."""
    monkeypatch.setattr(controller_mod.config, "STREAMING_REPLIES", False)


@pytest.fixture
def ctrl():
    return controller_mod.PetController()


def _capture(signal):
    events = []
    signal.connect(lambda *a: events.append(a[0] if len(a) == 1 else a))
    return events


# ── petctl routing ──────────────────────────────────────────────────────────

def test_petctl_commands_never_reach_the_shell(monkeypatch, ctrl):
    ran = []
    monkeypatch.setattr(controller_mod.server_client, "run_local_command", lambda cmd: ran.append(cmd))
    actions = _capture(ctrl.action)

    output = ctrl._handle_command("petctl move top-left")

    assert ran == []
    assert actions == [{"action": "move", "anchor": "top-left"}]
    assert "top-left" in output


def test_ordinary_commands_still_run_locally(monkeypatch, ctrl):
    ran = []
    monkeypatch.setattr(controller_mod.server_client, "run_local_command",
                        lambda cmd: ran.append(cmd) or "[exit 0]\n")
    actions = _capture(ctrl.action)

    ctrl._handle_command("df -h /")

    assert ran == ["df -h /"]
    assert actions == []


def test_bad_petctl_syntax_is_reported_back_not_executed(monkeypatch, ctrl):
    ran = []
    monkeypatch.setattr(controller_mod.server_client, "run_local_command", lambda cmd: ran.append(cmd))
    output = ctrl._handle_command("petctl move sideways")
    assert ran == []
    assert "[pet]" in output


def test_petctl_nap_also_flips_the_controller_state(ctrl):
    napping = _capture(ctrl.napping)
    ctrl._handle_command("petctl nap on")
    assert napping == [True]
    assert ctrl._napping is True


# ── filectl routing ──────────────────────────────────────────────────────────
# filectl's wire format is a single-line JSON envelope (see file_ops.py's
# module docstring for why) — build commands with json.dumps so the tests
# don't hardcode escaping by hand.

def _filectl(payload: dict) -> str:
    return "filectl " + json.dumps(payload)


def test_filectl_commands_never_reach_the_shell(monkeypatch, ctrl, tmp_path):
    ran = []
    monkeypatch.setattr(controller_mod.server_client, "run_local_command", lambda cmd: ran.append(cmd))
    target = tmp_path / "a.txt"

    output = ctrl._handle_command(_filectl({"op": "write", "path": str(target), "content": "hello"}))

    assert ran == []
    assert target.read_text() == "hello"
    assert str(target) in output


def test_filectl_edit_round_trips_through_the_relay(monkeypatch, ctrl, tmp_path):
    monkeypatch.setattr(controller_mod.server_client, "run_local_command",
                        lambda cmd: (_ for _ in ()).throw(AssertionError("should not shell out")))
    target = tmp_path / "a.py"
    target.write_text("x = 1\n")

    output = ctrl._handle_command(
        _filectl({"op": "edit", "path": str(target), "old": "x = 1", "new": "x = 2"})
    )

    assert target.read_text() == "x = 2\n"
    assert str(target) in output


def test_bad_filectl_syntax_is_reported_back_not_executed(monkeypatch, ctrl):
    ran = []
    monkeypatch.setattr(controller_mod.server_client, "run_local_command", lambda cmd: ran.append(cmd))
    output = ctrl._handle_command("filectl {not valid json")
    assert ran == []
    assert "[filectl]" in output


def test_filectl_execution_failure_is_reported_back_not_raised(monkeypatch, ctrl):
    output = ctrl._handle_command(_filectl({"op": "read", "path": "/no/such/file.txt"}))
    assert "[filectl]" in output


def test_ordinary_commands_still_run_locally_alongside_filectl(monkeypatch, ctrl):
    ran = []
    monkeypatch.setattr(controller_mod.server_client, "run_local_command",
                        lambda cmd: ran.append(cmd) or "[exit 0]\n")
    ctrl._handle_command("df -h /")
    assert ran == ["df -h /"]


# ── barge-in ────────────────────────────────────────────────────────────────

def test_the_interrupt_log_reports_what_fired_not_the_reset_counters(monkeypatch, ctrl):
    """_speak resets the detector after playback so the pet's own voice
    doesn't linger in the wake model's window. Reading the stats after that
    reset reports 0.000 at frame 0 for every interruption, which is worse
    than no instrumentation — it looks like hard evidence and isn't."""
    from bolt_pet.audio import barge_in as barge_in_mod

    class Detector(barge_in_mod.WakeWordBargeIn):
        def __init__(self):
            self._frames, self._peak, self._last, self._last_threshold = 0, 0.0, 0.0, 0.5
            self._frame_len = 1280
            self.reset_calls = 0

        def reset(self):
            self.reset_calls += 1
            self._frames, self._peak, self._last = 0, 0.0, 0.0

    detector = Detector()
    ctrl._barge_in = detector
    logs = _capture(ctrl.log)

    def interrupted_playback(text, on_error=None, should_stop=None, voice_id=None, on_level=None):
        # What really happens: frames get scored during playback, then one
        # clears the threshold and playback aborts.
        detector._frames, detector._peak, detector._last = 7, 0.81, 0.81
        return False

    monkeypatch.setattr(controller_mod.tts, "speak", interrupted_playback)

    ctrl._speak("a very long explanation")

    interrupted = next(m for m in logs if "Interrupted" in m)
    assert "0.810" in interrupted and "frame 7" in interrupted
    # Once before playback (clear the window) and once after (drop the pet's
    # own voice) — the point is that the *read* happens between them.
    assert detector.reset_calls == 2


def test_interrupted_playback_queues_an_immediate_next_turn(monkeypatch, ctrl):
    logs = _capture(ctrl.log)
    monkeypatch.setattr(controller_mod.tts, "speak",
                        lambda text, on_error=None, should_stop=None, voice_id=None, on_level=None: False)  # interrupted

    ctrl._speak("a very long explanation")

    assert ctrl._talk_now.is_set()  # loop picks the next turn up without a wake word
    assert any("Interrupted" in message for message in logs)


def test_uninterrupted_playback_does_not_queue_a_turn(monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.tts, "speak",
                        lambda text, on_error=None, should_stop=None, voice_id=None, on_level=None: True)
    ctrl._speak("short answer")
    assert not ctrl._talk_now.is_set()


# ── follow-up listening ─────────────────────────────────────────────────────

@pytest.fixture
def spoke(monkeypatch):
    """Playback that always completes, so only the follow-up rule decides
    whether another turn is queued."""
    monkeypatch.setattr(controller_mod.tts, "speak",
                        lambda text, on_error=None, should_stop=None, voice_id=None, on_level=None: True)


def test_a_reply_ending_in_a_question_keeps_listening(spoke, ctrl):
    logs = _capture(ctrl.log)

    ctrl._speak("You're still in ~/Documents/bolt-pet. Ready to run a command?")

    assert ctrl._talk_now.is_set()      # no wake word needed for the answer
    assert ctrl._pending_follow_up      # and the next turn knows it's an answer
    assert any("listening for your answer" in message for message in logs)


def test_a_statement_does_not_keep_listening(spoke, ctrl):
    ctrl._speak("It's 7:15 AM on July 23, 2026.")
    assert not ctrl._talk_now.is_set()
    assert not ctrl._pending_follow_up


def test_a_question_in_passing_does_not_count(spoke, ctrl):
    """Only a reply that *ends* on a question is waiting for an answer."""
    ctrl._speak("What time is it? It's 7:15 AM.")
    assert not ctrl._talk_now.is_set()


def test_follow_ups_stop_at_the_cap(spoke, monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.config, "FOLLOW_UP_MAX_TURNS", 2)
    logs = _capture(ctrl.log)

    for _ in range(2):
        ctrl._speak("Want me to keep going?")
        ctrl._talk_now.clear()
    assert ctrl._follow_ups == 2

    ctrl._speak("Want me to keep going?")

    assert not ctrl._talk_now.is_set()  # chain broken until you re-trigger it
    assert any("Follow-up limit reached" in message for message in logs)


def test_the_cap_is_not_announced_on_ordinary_replies(spoke, monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.config, "FOLLOW_UP_MAX_TURNS", 1)
    ctrl._follow_ups = 1
    logs = _capture(ctrl.log)

    ctrl._speak("Done — the file is saved.")

    assert not any("Follow-up limit" in message for message in logs)


def test_starting_a_turn_yourself_resets_the_chain(spoke, monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.mic, "record_utterance",
                        lambda *a, **kw: None)  # you said nothing
    ctrl._follow_ups = 3

    ctrl._handle_conversation_turn()

    assert ctrl._follow_ups == 0


def test_an_answered_question_gets_a_longer_grace_period(spoke, monkeypatch, ctrl):
    """You were just asked something — you get longer to think than when you
    deliberately said the wake word."""
    grace = []
    monkeypatch.setattr(controller_mod.mic, "record_utterance",
                        lambda *a, **kw: grace.append(kw.get("grace_s")) or None)

    ctrl._handle_conversation_turn()                 # you started this one
    ctrl._pending_follow_up = True
    ctrl._handle_conversation_turn()                 # this one answers a question

    assert grace == [None, controller_mod.config.FOLLOW_UP_GRACE_SECONDS]


def test_muting_stops_follow_ups(spoke, ctrl):
    ctrl._muted = True
    ctrl._speak("Shall I continue?")
    assert not ctrl._talk_now.is_set()  # mute means don't listen, question or not


def test_follow_up_can_be_turned_off(spoke, monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.config, "FOLLOW_UP_LISTEN", False)
    ctrl._speak("Shall I continue?")
    assert not ctrl._talk_now.is_set()


def test_being_interrupted_restarts_the_chain(monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.tts, "speak",
                        lambda text, on_error=None, should_stop=None, voice_id=None, on_level=None: False)
    ctrl._follow_ups = 3

    ctrl._speak("a very long explanation")

    assert ctrl._follow_ups == 0  # you're clearly engaged
    assert ctrl._talk_now.is_set()


def test_speech_is_recorded_in_the_history(monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.tts, "speak",
                        lambda text, on_error=None, should_stop=None, voice_id=None, on_level=None: True)
    ctrl._speak("**bold** reply")
    assert ctrl.history.last().text == "**bold** reply"  # raw, for copy/paste


# ── screen context ──────────────────────────────────────────────────────────

def test_the_active_window_rides_along_with_the_utterance(monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.mic, "record_utterance",
                        lambda *a, **k: np.zeros(10, dtype=np.int16))
    monkeypatch.setattr(controller_mod.stt, "transcribe", lambda pcm: "what's this error?")
    monkeypatch.setattr(controller_mod.screen_context, "context_for",
                        lambda text: f"{text}\n\n[on screen right now: app.py]")
    monkeypatch.setattr(controller_mod.tts, "speak",
                        lambda text, on_error=None, should_stop=None, voice_id=None, on_level=None: True)
    sent = []
    monkeypatch.setattr(controller_mod.server_client, "converse",
                        lambda text, on_command=None, on_say=None: sent.append(text) or Reply("that's a KeyError"))

    ctrl._handle_conversation_turn()

    assert "[on screen right now: app.py]" in sent[0]
    # ...but the *history* keeps what you actually said, not the annotation.
    assert ctrl.history.entries()[0].text == "what's this error?"


# ── quiet hours / do-not-disturb ────────────────────────────────────────────

def test_quiet_hours_suppress_the_heartbeat(monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.config, "HEARTBEAT_INTERVAL_SECONDS", 0)
    monkeypatch.setattr(controller_mod.config, "QUIET_HOURS", "00:00-23:59")
    called = {"n": 0}
    monkeypatch.setattr(controller_mod.server_client, "report_status",
                        lambda: called.__setitem__("n", called["n"] + 1))

    ctrl._maybe_heartbeat()

    assert called["n"] == 0
    assert ctrl._napping is True


def test_fullscreen_triggers_do_not_disturb(monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.config, "QUIET_HOURS", "")
    monkeypatch.setattr(controller_mod.config, "DND_ON_FULLSCREEN", True)
    monkeypatch.setattr(controller_mod.screen_context, "is_fullscreen_active", lambda: True)
    napping = _capture(ctrl.napping)

    ctrl._refresh_nap_state()

    assert napping == [True]


def test_a_manual_nap_overrides_the_schedule(monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.config, "QUIET_HOURS", "")
    ctrl.set_napping(True)
    ctrl._last_nap_check = 0.0
    ctrl._refresh_nap_state()
    assert ctrl._napping is True  # the schedule doesn't wake it back up

    ctrl.set_napping(None)  # back on schedule
    ctrl._last_nap_check = 0.0
    ctrl._refresh_nap_state()
    assert ctrl._napping is False


def test_napping_still_answers_when_spoken_to(monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.mic, "record_utterance",
                        lambda *a, **k: np.zeros(10, dtype=np.int16))
    monkeypatch.setattr(controller_mod.stt, "transcribe", lambda pcm: "you awake?")
    monkeypatch.setattr(controller_mod.server_client, "converse",
                        lambda text, on_command=None, on_say=None: Reply("always"))
    spoken = []
    monkeypatch.setattr(controller_mod.tts, "speak",
                        lambda text, on_error=None, should_stop=None, voice_id=None, on_level=None: spoken.append(text) or True)
    ctrl.set_napping(True)

    ctrl._handle_conversation_turn()

    assert spoken == ["always"]


# ── notification bridge ─────────────────────────────────────────────────────

def test_notifications_are_forwarded_and_spoken(monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.config, "NOTIFICATION_MIN_INTERVAL_SECONDS", 0)
    ctrl._notification_gate = controller_mod.notifications.NotificationGate("", 0)
    sent, spoken = [], []
    monkeypatch.setattr(controller_mod.server_client, "converse",
                        lambda text, on_command=None, on_say=None: sent.append(text) or Reply("your build is green"))
    monkeypatch.setattr(controller_mod.tts, "speak",
                        lambda text, on_error=None, should_stop=None, voice_id=None, on_level=None: spoken.append(text) or True)

    ctrl._queue_notification(Notification(app="CI", summary="Build finished", body=""))
    ctrl._drain_notifications()

    assert "CI: Build finished" in sent[0]
    assert spoken == ["your build is green"]
    assert ctrl._state.state == PetState.IDLE


def test_filtered_out_notifications_are_never_queued(ctrl):
    ctrl._notification_gate = controller_mod.notifications.NotificationGate("deploy", 0)
    ctrl._queue_notification(Notification(app="Chat", summary="lunch?", body=""))
    assert ctrl._pending_notifications == []


def test_notifications_are_not_forwarded_while_napping(monkeypatch, ctrl):
    ctrl._notification_gate = controller_mod.notifications.NotificationGate("", 0)
    called = {"n": 0}
    monkeypatch.setattr(controller_mod.server_client, "converse",
                        lambda text, on_command=None, on_say=None: called.__setitem__("n", called["n"] + 1))
    ctrl.set_napping(True)

    ctrl._queue_notification(Notification(app="CI", summary="Build finished", body=""))
    ctrl._drain_notifications()

    assert called["n"] == 0


# ── wake threshold ──────────────────────────────────────────────────────────

def test_threshold_is_live_and_clamped(ctrl):
    ctrl.set_wake_threshold(0.72)
    assert ctrl.wake_threshold() == pytest.approx(0.72)
    ctrl.set_wake_threshold(5)
    assert ctrl.wake_threshold() == pytest.approx(0.99)
    ctrl.set_wake_threshold(-1)
    assert ctrl.wake_threshold() == pytest.approx(0.01)


def test_near_misses_are_recorded_for_the_tuner(ctrl):
    ctrl.set_wake_threshold(0.5)
    ctrl._observe_wake_score(0.42, 0.5)   # near miss
    ctrl._observe_wake_score(0.01, 0.5)   # background noise, not interesting
    stats = ctrl.wake_stats()
    assert len(stats["near_misses"]) == 1
    assert stats["peak"] == pytest.approx(0.42)

    ctrl.reset_wake_stats()
    assert ctrl.wake_stats()["near_misses"] == []


# ── file delivery ────────────────────────────────────────────────────────────

def test_check_deliveries_downloads_and_saves_queued_files(monkeypatch, ctrl, tmp_path):
    monkeypatch.setattr(controller_mod.config, "DELIVERED_FILES_DIR", tmp_path)
    monkeypatch.setattr(controller_mod.server_client, "list_outbox_files",
                        lambda: [{"id": "abc", "name": "report.pdf", "size": 5}])
    monkeypatch.setattr(controller_mod.server_client, "download_outbox_file",
                        lambda file_id: b"hello" if file_id == "abc" else b"")

    ctrl._check_deliveries()

    assert (tmp_path / "report.pdf").read_bytes() == b"hello"


def test_check_deliveries_is_a_noop_when_nothing_queued(monkeypatch, ctrl, tmp_path):
    monkeypatch.setattr(controller_mod.config, "DELIVERED_FILES_DIR", tmp_path)
    monkeypatch.setattr(controller_mod.server_client, "list_outbox_files", lambda: [])
    downloaded = []
    monkeypatch.setattr(controller_mod.server_client, "download_outbox_file",
                        lambda file_id: downloaded.append(file_id))

    ctrl._check_deliveries()

    assert downloaded == []


def test_check_deliveries_can_be_disabled(monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.config, "RECEIVE_FILES", False)
    called = {"n": 0}
    monkeypatch.setattr(controller_mod.server_client, "list_outbox_files",
                        lambda: called.__setitem__("n", called["n"] + 1))

    ctrl._check_deliveries()

    assert called["n"] == 0


def test_check_deliveries_logs_and_continues_on_download_failure(monkeypatch, ctrl, tmp_path):
    monkeypatch.setattr(controller_mod.config, "DELIVERED_FILES_DIR", tmp_path)
    monkeypatch.setattr(controller_mod.server_client, "list_outbox_files", lambda: [
        {"id": "bad", "name": "a.txt", "size": 1},
        {"id": "good", "name": "b.txt", "size": 1},
    ])

    def fake_download(file_id):
        if file_id == "bad":
            raise controller_mod.server_client.ServerError("gone")
        return b"ok"

    monkeypatch.setattr(controller_mod.server_client, "download_outbox_file", fake_download)
    logs = _capture(ctrl.log)

    ctrl._check_deliveries()

    assert (tmp_path / "b.txt").read_bytes() == b"ok"
    assert not (tmp_path / "a.txt").exists()
    assert any("bad" in msg or "a.txt" in msg for msg in logs)


def test_check_deliveries_runs_after_a_conversation_turn(monkeypatch, ctrl, tmp_path):
    monkeypatch.setattr(controller_mod.mic, "record_utterance",
                        lambda *a, **k: np.zeros(10, dtype=np.int16))
    monkeypatch.setattr(controller_mod.stt, "transcribe", lambda pcm: "send me that file")
    monkeypatch.setattr(controller_mod.server_client, "converse",
                        lambda text, on_command=None, on_say=None: Reply("it's on the way"))
    monkeypatch.setattr(controller_mod.tts, "speak",
                        lambda text, on_error=None, should_stop=None, voice_id=None, on_level=None: True)
    monkeypatch.setattr(controller_mod.config, "DELIVERED_FILES_DIR", tmp_path)
    monkeypatch.setattr(controller_mod.server_client, "list_outbox_files",
                        lambda: [{"id": "abc", "name": "notes.txt", "size": 2}])
    monkeypatch.setattr(controller_mod.server_client, "download_outbox_file",
                        lambda file_id: b"hi")

    ctrl._handle_conversation_turn()

    assert (tmp_path / "notes.txt").read_bytes() == b"hi"


# ── server-picked voice (the desk API's speak_as marker) ────────────────────

def _voice_turn(monkeypatch, ctrl, reply, said="talk like a pirate"):
    """Run one full conversation turn whose reply is *reply*, returning the
    voice_id each tts.speak() call was given."""
    voices = []
    monkeypatch.setattr(controller_mod.mic, "record_utterance",
                        lambda *a, **k: np.zeros(10, dtype=np.int16))
    monkeypatch.setattr(controller_mod.stt, "transcribe", lambda pcm: said)
    monkeypatch.setattr(controller_mod.server_client, "converse",
                        lambda text, on_command=None, on_say=None: reply)
    monkeypatch.setattr(
        controller_mod.tts, "speak",
        lambda text, on_error=None, should_stop=None, voice_id=None, on_level=None:
            voices.append(voice_id) or True,
    )
    ctrl._handle_conversation_turn()
    return voices


def test_a_speak_as_reply_is_spoken_in_that_voice(monkeypatch, ctrl):
    voices = _voice_turn(monkeypatch, ctrl, Reply("Ahoy.", "VOICE1", "Terence"))
    assert voices == ["VOICE1"]
    assert ctrl.current_voice() == "Terence"


def test_the_picked_voice_sticks_for_later_replies(monkeypatch, ctrl):
    """The server tags one reply and doesn't keep the id in its history, so
    it can't re-request the voice when you say "keep talking like that"."""
    monkeypatch.setattr(controller_mod.config, "VOICE_STICKY", True)
    _voice_turn(monkeypatch, ctrl, Reply("Ahoy.", "VOICE1", "Terence"))
    voices = _voice_turn(monkeypatch, ctrl, Reply("Still me."), said="and now?")
    assert voices == ["VOICE1"]


def test_voice_stickiness_can_be_turned_off(monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.config, "VOICE_STICKY", False)
    _voice_turn(monkeypatch, ctrl, Reply("Ahoy.", "VOICE1", "Terence"))
    voices = _voice_turn(monkeypatch, ctrl, Reply("Back to normal."), said="and now?")
    assert voices == [None]
    assert ctrl.current_voice() == ""


def test_a_new_pick_replaces_the_old_one(monkeypatch, ctrl):
    changes = _capture(ctrl.voice_changed)
    _voice_turn(monkeypatch, ctrl, Reply("Ahoy.", "VOICE1", "Terence"))
    voices = _voice_turn(monkeypatch, ctrl, Reply("こんにちは。", "VOICE2", "Asahi"),
                         said="say that in Japanese")
    assert voices == ["VOICE2"]
    assert changes == ["Terence", "Asahi"]


def test_resetting_the_voice_goes_back_to_the_default(monkeypatch, ctrl):
    _voice_turn(monkeypatch, ctrl, Reply("Ahoy.", "VOICE1", "Terence"))
    changes = _capture(ctrl.voice_changed)

    ctrl.reset_voice()

    assert ctrl.current_voice() == ""
    assert changes == [""]  # the tray's menu entry follows this signal
    voices = _voice_turn(monkeypatch, ctrl, Reply("Normal again."), said="hi")
    assert voices == [None]


def test_an_unnamed_voice_still_reports_something_resettable(monkeypatch, ctrl):
    """voice_name is optional server-side — falling back to the id keeps the
    tray entry from reading "now: " with nothing after it."""
    _voice_turn(monkeypatch, ctrl, Reply("Ahoy.", "VOICE1"))
    assert ctrl.current_voice() == "VOICE1"


def test_petctl_voice_reset_returns_bolt_to_his_own_voice(monkeypatch, ctrl):
    """The server can pick a voice but can't ask for the default back — it
    was never told what Bolt's own voice id is. This is how it asks."""
    ran = []
    monkeypatch.setattr(controller_mod.server_client, "run_local_command", lambda cmd: ran.append(cmd))
    _voice_turn(monkeypatch, ctrl, Reply("Ahoy.", "VOICE1", "Terence"))

    output = ctrl._handle_command("petctl voice reset")

    assert ran == []  # never reaches a shell, like every other petctl verb
    assert "Terence" in output  # the server can't see the voice; tell it what changed
    assert ctrl.current_voice() == ""
    assert _voice_turn(monkeypatch, ctrl, Reply("Normal again."), said="hi") == [None]


def test_petctl_voice_reset_says_so_when_there_was_nothing_to_reset(ctrl):
    assert "already" in ctrl._handle_command("petctl voice reset")


# ── dialoguectl (multi-voice scenes) ────────────────────────────────────────

def _dialogue_command(*lines):
    import json
    return "dialoguectl " + json.dumps({"lines": list(lines)})


def test_dialoguectl_never_reaches_the_shell(monkeypatch, ctrl):
    ran = []
    monkeypatch.setattr(controller_mod.server_client, "run_local_command", lambda cmd: ran.append(cmd))
    monkeypatch.setattr(controller_mod.config, "ELEVENLABS_VOICE_ID", "aaorr6ZHIL88gEexu7dC")
    monkeypatch.setattr(controller_mod.tts, "synthesize_dialogue",
                        lambda inputs, model_id=None, stability=None: (np.zeros(4, dtype=np.int16), 24000))
    monkeypatch.setattr(controller_mod.tts, "play_pcm",
                        lambda pcm, rate, should_stop=None, on_level=None: True)

    output = ctrl._handle_command(_dialogue_command({"voice": "self", "text": "[cheerfully] hi"}))

    assert ran == []
    assert "[dialogue] played 1 line" in output


def test_a_scene_shows_in_the_bubble_with_the_delivery_tags_stripped(monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.config, "ELEVENLABS_VOICE_ID", "aaorr6ZHIL88gEexu7dC")
    monkeypatch.setattr(controller_mod.config, "DIALOGUE_VOICES", "narrator:9BWtsMINqrJLrRacOk9x")
    monkeypatch.setattr(controller_mod.tts, "synthesize_dialogue",
                        lambda inputs, model_id=None, stability=None: (np.zeros(4, dtype=np.int16), 24000))
    monkeypatch.setattr(controller_mod.tts, "play_pcm", lambda pcm, rate, should_stop=None, on_level=None: True)
    said = _capture(ctrl.said)

    ctrl._handle_command(_dialogue_command(
        {"voice": "self", "text": "[cheerfully] Hello there!"},
        {"voice": "narrator", "text": "[whispering] He is lying."},
    ))

    assert said == ["Hello there! He is lying."]
    assert ctrl.history.last().text == "Hello there! He is lying."


def test_a_mid_turn_scene_returns_to_thinking_not_idle(monkeypatch, ctrl):
    """The server is still waiting on the tool result, so the pet talks and
    goes back to waiting — dropping to IDLE would look like the turn ended."""
    monkeypatch.setattr(controller_mod.config, "ELEVENLABS_VOICE_ID", "aaorr6ZHIL88gEexu7dC")
    monkeypatch.setattr(controller_mod.tts, "synthesize_dialogue",
                        lambda inputs, model_id=None, stability=None: (np.zeros(4, dtype=np.int16), 24000))
    monkeypatch.setattr(controller_mod.tts, "play_pcm", lambda pcm, rate, should_stop=None, on_level=None: True)
    ctrl._state.transition(PetState.LISTENING)
    ctrl._state.transition(PetState.THINKING)
    states = _capture(ctrl.state_changed)

    ctrl._handle_command(_dialogue_command({"voice": "self", "text": "hi"}))

    assert states == ["talking", "thinking"]
    assert ctrl._state.state == PetState.THINKING


def test_the_scene_uses_a_voice_the_server_picked_with_speak_as(monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.config, "ELEVENLABS_VOICE_ID", "aaorr6ZHIL88gEexu7dC")
    seen = {}

    def capture(inputs, model_id=None, stability=None):
        seen["inputs"] = inputs
        return np.zeros(4, dtype=np.int16), 24000

    monkeypatch.setattr(controller_mod.tts, "synthesize_dialogue", capture)
    monkeypatch.setattr(controller_mod.tts, "play_pcm", lambda pcm, rate, should_stop=None, on_level=None: True)
    ctrl._apply_voice(controller_mod.server_client.Reply("ok", "PICKEDvoice123456789", "Terence"))

    ctrl._handle_command(_dialogue_command({"voice": "self", "text": "hi"}))

    assert seen["inputs"][0]["voice_id"] == "PICKEDvoice123456789"


def test_a_synthesis_failure_is_reported_back_for_bolt_to_retry(monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.config, "ELEVENLABS_VOICE_ID", "aaorr6ZHIL88gEexu7dC")

    def boom(inputs, model_id=None, stability=None):
        raise controller_mod.tts.TtsError("voice_id not found")

    monkeypatch.setattr(controller_mod.tts, "synthesize_dialogue", boom)

    output = ctrl._handle_command(_dialogue_command({"voice": "self", "text": "hi"}))

    assert "couldn't synthesize" in output and "voice_id not found" in output
    assert ctrl._state.state == PetState.IDLE  # nothing left half-transitioned


def test_a_bad_voice_name_comes_back_as_advice_not_an_exception(monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.config, "DIALOGUE_VOICES", "narrator:9BWtsMINqrJLrRacOk9x")
    output = ctrl._handle_command(_dialogue_command({"voice": "wizard", "text": "hi"}))
    assert "unknown voice" in output and "narrator" in output


def test_dialogue_can_be_switched_off_on_this_device(monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.config, "DIALOGUE", False)
    called = []
    monkeypatch.setattr(controller_mod.tts, "synthesize_dialogue",
                        lambda *a, **k: called.append(1))
    output = ctrl._handle_command(_dialogue_command({"voice": "self", "text": "hi"}))
    assert "disabled" in output and called == []


def test_talking_over_a_scene_is_reported_up_the_relay(monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.config, "ELEVENLABS_VOICE_ID", "aaorr6ZHIL88gEexu7dC")
    monkeypatch.setattr(controller_mod.tts, "synthesize_dialogue",
                        lambda inputs, model_id=None, stability=None: (np.zeros(4, dtype=np.int16), 24000))
    monkeypatch.setattr(controller_mod.tts, "play_pcm",
                        lambda pcm, rate, should_stop=None, on_level=None: False)  # barge-in

    output = ctrl._handle_command(_dialogue_command({"voice": "self", "text": "hi"}))

    assert "interrupted" in output


# ── petctl self_restart ─────────────────────────────────────────────────────

def test_self_restart_arms_after_the_turn_rather_than_dying_mid_relay(monkeypatch, ctrl, tmp_path):
    """Restarting inline would kill the HTTP tool relay before the result was
    posted, and the server would wait out its timeout on a turn that can never
    finish. So the command returns, the turn completes, *then* the pet dies."""
    monkeypatch.setattr(controller_mod.self_restart, "DEFAULT_STATE_PATH", tmp_path / "ctx.json")
    monkeypatch.setattr(controller_mod.self_restart, "preflight", lambda *a, **k: None)
    restarts = _capture(ctrl.restart_requested)

    output = ctrl._handle_command("petctl self_restart check the new dialogue code")

    assert "restarting as soon as this turn finishes" in output
    assert restarts == []  # nothing has happened yet

    assert ctrl._maybe_self_restart() is True
    assert restarts and "check the new dialogue code" in restarts[0]


def test_a_broken_edit_is_reported_instead_of_leaving_nothing_running(monkeypatch, ctrl, tmp_path):
    monkeypatch.setattr(controller_mod.self_restart, "DEFAULT_STATE_PATH", tmp_path / "ctx.json")

    def boom(*args, **kwargs):
        raise controller_mod.self_restart.RestartError(
            "the current code does not import, so restarting would leave you with "
            "nothing running. Fix this first:\nSyntaxError: invalid syntax"
        )

    monkeypatch.setattr(controller_mod.self_restart, "preflight", boom)
    restarts = _capture(ctrl.restart_requested)

    output = ctrl._handle_command("petctl self_restart try the new code")

    assert "SyntaxError" in output and "refused" in output
    assert ctrl._maybe_self_restart() is False
    assert restarts == []


def test_a_second_restart_request_in_one_turn_is_a_no_op(monkeypatch, ctrl, tmp_path):
    monkeypatch.setattr(controller_mod.self_restart, "DEFAULT_STATE_PATH", tmp_path / "ctx.json")
    monkeypatch.setattr(controller_mod.self_restart, "preflight", lambda *a, **k: None)
    ctrl._handle_command("petctl self_restart first")
    assert "already armed" in ctrl._handle_command("petctl self_restart second")


def test_self_restart_can_be_switched_off_on_this_device(monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.config, "SELF_RESTART", False)
    checked = []
    monkeypatch.setattr(controller_mod.self_restart, "preflight",
                        lambda *a, **k: checked.append(1))
    assert "disabled" in ctrl._handle_command("petctl self_restart go")
    assert checked == []


def test_coming_back_up_reports_to_the_server_and_speaks_the_reply(monkeypatch, ctrl, tmp_path):
    """The half that makes it a loop: the new process tells Bolt it's back and
    why, and his answer is spoken like any other turn."""
    state = tmp_path / "ctx.json"
    monkeypatch.setattr(controller_mod.self_restart, "DEFAULT_STATE_PATH", state)
    controller_mod.self_restart.arm("check the walk cycle", version="0.2.3",
                                    path=state, now=1000.0)
    sent, spoken = [], []
    monkeypatch.setattr(controller_mod.server_client, "converse",
                        lambda text, on_command=None, on_say=None: sent.append(text) or Reply("Good, it's up."))
    monkeypatch.setattr(controller_mod.tts, "speak",
                        lambda text, on_error=None, should_stop=None, voice_id=None, on_level=None:
                            spoken.append(text) or True)

    ctrl._report_self_restart()

    assert "[pet self-restart]" in sent[0] and "check the walk cycle" in sent[0]
    assert spoken == ["Good, it's up."]
    # Consumed, so the next start doesn't announce the same restart again.
    assert controller_mod.self_restart.load(state) is None


def test_an_ordinary_start_reports_nothing(monkeypatch, ctrl, tmp_path):
    monkeypatch.setattr(controller_mod.self_restart, "DEFAULT_STATE_PATH", tmp_path / "none.json")
    called = []
    monkeypatch.setattr(controller_mod.server_client, "converse",
                        lambda text, on_command=None, on_say=None: called.append(text))

    ctrl._report_self_restart()

    assert called == []


# ── holding line: "give me a sec" while a tool runs ────────────────────────
# The server used to discard whatever the model wrote alongside a tool call,
# so the whole round trip was silence — and the model, with no evidence its
# sentence landed, said it again in the final reply.

def test_a_holding_line_is_spoken_before_the_tool_runs(monkeypatch, ctrl):
    order = []
    monkeypatch.setattr(controller_mod.tts, "speak",
                        lambda text, on_error=None, should_stop=None, voice_id=None, on_level=None:
                            order.append(("spoke", text)) or True)
    monkeypatch.setattr(controller_mod.server_client, "run_local_command",
                        lambda cmd: order.append(("ran", cmd)) or "[exit 0]")

    ctrl._speak_holding("Give me a sec.")
    ctrl._handle_command("df -h /")

    assert order == [("spoke", "Give me a sec."), ("ran", "df -h /")]


def test_the_holding_line_shows_in_the_bubble_but_not_the_transcript(monkeypatch, ctrl):
    """It's filler. The transcript should keep the actual answer."""
    monkeypatch.setattr(controller_mod.tts, "speak",
                        lambda text, on_error=None, should_stop=None, voice_id=None, on_level=None: True)
    said = _capture(ctrl.said)

    ctrl._speak_holding("I'll check on that — one moment.")

    assert said == ["I'll check on that — one moment."]
    assert ctrl.history.entries() == []


def test_a_holding_line_returns_to_waiting_not_idle(monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.tts, "speak",
                        lambda text, on_error=None, should_stop=None, voice_id=None, on_level=None: True)
    ctrl._state.transition(PetState.LISTENING)
    ctrl._state.transition(PetState.THINKING)
    states = _capture(ctrl.state_changed)

    ctrl._speak_holding("one sec")

    assert states == ["talking", "thinking"]


def test_the_relay_speaks_whatever_the_server_attaches(monkeypatch, ctrl):
    spoken = []
    monkeypatch.setattr(controller_mod.tts, "speak",
                        lambda text, on_error=None, should_stop=None, voice_id=None, on_level=None:
                            spoken.append(text) or True)
    monkeypatch.setattr(controller_mod.mic, "record_utterance",
                        lambda *a, **k: np.zeros(10, dtype=np.int16))
    monkeypatch.setattr(controller_mod.stt, "transcribe", lambda pcm: "what's in that folder?")

    def fake_converse(text, on_command=None, on_say=None):
        on_say("Let me check that for you.")      # what the server now sends
        on_command("filectl {\"op\": \"list\", \"path\": \"/tmp\"}")
        return Reply("It's got three files in it.")

    monkeypatch.setattr(controller_mod.server_client, "converse", fake_converse)

    ctrl._handle_conversation_turn()

    assert spoken == ["Let me check that for you.", "It's got three files in it."]


# ── streamed replies ───────────────────────────────────────────────────────
# Without streaming the pet waits out the entire model call before saying a
# word. With it, the wait is time-to-first-sentence.

def test_each_sentence_is_spoken_as_it_arrives(monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.config, "STREAMING_REPLIES", True)
    spoken = []
    monkeypatch.setattr(controller_mod.tts, "speak",
                        lambda text, on_error=None, should_stop=None, voice_id=None, on_level=None:
                            spoken.append(text) or True)

    def fake_stream(text, on_say, on_command=None, timeout=180.0):
        on_say("The disk is fine.")
        on_say("About sixty percent used.")
        return controller_mod.server_client.Reply(
            "The disk is fine. About sixty percent used.", spoken=True)

    monkeypatch.setattr(controller_mod.server_client, "converse_stream", fake_stream)
    monkeypatch.setattr(controller_mod.mic, "record_utterance",
                        lambda *a, **k: np.zeros(10, dtype=np.int16))
    monkeypatch.setattr(controller_mod.stt, "transcribe", lambda pcm: "how's the disk?")

    ctrl._handle_conversation_turn()

    assert spoken == ["The disk is fine.", "About sixty percent used."]
    # ...and the empty final reply is not spoken as an extra blank utterance.
    assert ctrl.history.entries()[-1].text == "About sixty percent used."


def test_a_stream_that_fails_before_speaking_falls_back_silently(monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.config, "STREAMING_REPLIES", True)
    spoken = []
    monkeypatch.setattr(controller_mod.tts, "speak",
                        lambda text, on_error=None, should_stop=None, voice_id=None, on_level=None:
                            spoken.append(text) or True)

    def broken_stream(text, on_say, on_command=None, timeout=180.0):
        raise controller_mod.server_client.ServerError("no streaming endpoint")

    monkeypatch.setattr(controller_mod.server_client, "converse_stream", broken_stream)
    monkeypatch.setattr(controller_mod.server_client, "converse",
                        lambda text, on_command=None, on_say=None: Reply("Fell back fine."))

    assert ctrl._ask_server("hello").text == "Fell back fine."
    assert spoken == []   # nothing was said twice


def test_streaming_can_be_switched_off(monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.config, "STREAMING_REPLIES", False)
    called = []
    monkeypatch.setattr(controller_mod.server_client, "converse_stream",
                        lambda *a, **k: called.append(1))
    monkeypatch.setattr(controller_mod.server_client, "converse",
                        lambda text, on_command=None, on_say=None: Reply("plain path"))

    assert ctrl._ask_server("hi").text == "plain path"
    assert called == []


def test_talking_over_a_streamed_reply_still_interrupts(monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.config, "STREAMING_REPLIES", True)
    monkeypatch.setattr(controller_mod.tts, "speak",
                        lambda text, on_error=None, should_stop=None, voice_id=None, on_level=None: False)

    ctrl._speak_stream_chunk("A long explanation you cut short.")

    assert ctrl._talk_now.is_set()


# ── notification latency (reported 2026-08-02: 5-10 minutes) ───────────────
# Draining was wired to the heartbeat's 60s interval, and a heartbeat that
# landed mid-conversation stamped its clock before noticing — so it burned the
# slot and waited another full interval. Several of those in a row is minutes.

def test_a_notification_goes_out_on_the_next_tick_not_the_next_heartbeat(monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.config, "NOTIFICATION_MIN_INTERVAL_SECONDS", 0)
    ctrl._notification_gate = controller_mod.notifications.NotificationGate("", 0)
    sent = []
    monkeypatch.setattr(controller_mod.server_client, "converse",
                        lambda text, on_command=None, on_say=None:
                            sent.append(text) or Reply("noted"))
    monkeypatch.setattr(controller_mod.tts, "speak",
                        lambda text, on_error=None, should_stop=None, voice_id=None, on_level=None: True)
    monkeypatch.setattr(controller_mod.server_client, "report_status", lambda: None)
    # A heartbeat has just run, so the next one is a full interval away.
    ctrl._last_heartbeat = controller_mod.time.monotonic()

    ctrl._queue_notification(Notification(app="Signal", summary="Harry", body="you there?"))
    ctrl._maybe_heartbeat()

    assert sent and "Harry" in sent[0]


def test_a_heartbeat_skipped_mid_conversation_does_not_burn_its_slot(monkeypatch, ctrl):
    monkeypatch.setattr(controller_mod.server_client, "report_status", lambda: None)
    ctrl._last_heartbeat = 0.0
    ctrl._state.transition(PetState.LISTENING)
    ctrl._state.transition(PetState.THINKING)

    ctrl._maybe_heartbeat()          # due, but the pet is busy
    assert ctrl._last_heartbeat == 0.0, "the clock must not advance on a skipped tick"

    ctrl._state.transition(PetState.IDLE)
    ctrl._maybe_heartbeat()          # free now — runs immediately
    assert ctrl._last_heartbeat > 0.0


def test_notifications_wait_while_the_pet_is_mid_turn(monkeypatch, ctrl):
    """Speaking over the answer they're already getting would be worse than
    waiting a couple of seconds."""
    ctrl._notification_gate = controller_mod.notifications.NotificationGate("", 0)
    sent = []
    monkeypatch.setattr(controller_mod.server_client, "converse",
                        lambda text, on_command=None, on_say=None: sent.append(text))
    ctrl._state.transition(PetState.LISTENING)

    ctrl._queue_notification(Notification(app="CI", summary="Build finished", body=""))
    ctrl._maybe_heartbeat()

    assert sent == []
    assert len(ctrl._pending_notifications) == 1   # kept, not dropped


def test_a_second_message_inside_the_rate_limit_is_no_longer_lost(monkeypatch, ctrl):
    """The gate used to drop it. With NOTIFICATION_MIN_INTERVAL_SECONDS=60,
    two texts a minute apart meant you heard about one of them."""
    ctrl._notification_gate = controller_mod.notifications.NotificationGate("", 60)
    sent = []
    monkeypatch.setattr(controller_mod.server_client, "converse",
                        lambda text, on_command=None, on_say=None:
                            sent.append(text) or Reply("ok"))
    monkeypatch.setattr(controller_mod.tts, "speak",
                        lambda text, on_error=None, should_stop=None, voice_id=None, on_level=None: True)

    ctrl._queue_notification(Notification(app="Signal", summary="Harry", body="you there?"))
    ctrl._queue_notification(Notification(app="Signal", summary="Harry", body="it's urgent"))
    ctrl._drain_notifications()

    assert len(sent) == 1                      # one round trip...
    assert "you there?" in sent[0] and "it's urgent" in sent[0]   # ...both messages
    assert "2 desktop notifications" in sent[0]


def test_the_filter_still_applies(ctrl):
    ctrl._notification_gate = controller_mod.notifications.NotificationGate("deploy", 0)
    ctrl._queue_notification(Notification(app="Chat", summary="lunch?", body=""))
    assert ctrl._pending_notifications == []


def test_a_notification_storm_cannot_become_an_unbounded_backlog(ctrl):
    ctrl._notification_gate = controller_mod.notifications.NotificationGate("", 0)
    for index in range(40):
        ctrl._queue_notification(Notification(app="Spam", summary=f"#{index}", body=""))

    assert len(ctrl._pending_notifications) == controller_mod._MAX_PENDING_NOTIFICATIONS
    assert "#39" in ctrl._pending_notifications[-1].as_text()   # newest kept