Bolt-Pet/bolt_pet/controller.py

"""Orchestrates the pet's mic -> wake word -> STT -> server -> TTS pipeline.

Runs on a background QThread (see ui/app.py) so the Qt event loop / window
painting is never blocked by audio I/O or network calls. Talks to the UI
only through Qt signals (state_changed / said / log / action / napping),
which Qt marshals safely across threads — this class never touches a QWidget
directly.

Beyond the core loop it owns the side channels that let the pet act on its
own: the heartbeat (proactive announcements), the desktop notification
bridge, quiet hours, barge-in, and the live wake-word threshold.
"""

from __future__ import annotations

import threading
import time
from typing import Optional

from PySide6.QtCore import QObject, Signal

from . import (
    config, dialogue as dialogue_mod, speech, file_delivery, file_ops,
    history as history_mod, monitors as monitors_mod, notifications,
    pet_actions, quiet, screen_context, screen_text, self_restart,
    server_client, speech_text, updater,
)
from . import __version__
from .audio import barge_in, mic, stt, stt_stream, tts, wake_word
from .state import PetState, PetStateMachine

# A burst while the pet is busy is batched into one turn, but the queue still
# needs a ceiling — a notification storm must not become an unbounded backlog
# that gets read out minutes later.
_MAX_PENDING_NOTIFICATIONS = 12

# How often to re-check whether the pet should be napping. The fullscreen
# probe shells out to xprop, so this deliberately isn't every heartbeat tick.
_NAP_CHECK_INTERVAL_SECONDS = 10.0


class PetController(QObject):
    state_changed = Signal(str)  # PetState.value
    said = Signal(str)  # text now showing in the speech bubble
    log = Signal(str)
    action = Signal(dict)  # parsed petctl action for the UI to perform
    napping = Signal(bool)  # quiet hours / fullscreen do-not-disturb
    voice_changed = Signal(str)  # name of the server-picked voice ("" = default)
    mouth = Signal(float)  # 0..1 speech loudness, for lip-sync while talking
    restart_requested = Signal(str)  # version we just updated to
    finished = Signal()

    def __init__(self):
        super().__init__()
        self._running = True
        self._muted = False
        self._talk_now = threading.Event()
        self._stream = None
        self._last_heartbeat = 0.0
        self._state = PetStateMachine(on_change=self._handle_state_change)

        # Conversation scrollback, shared read-only with the UI's History
        # window. Append-only from this thread; the UI only ever snapshots it.
        self.history = history_mod.ConversationHistory(limit=config.HISTORY_LIMIT)

        # The screen layout, as published by the UI (see set_monitors). Held
        # here rather than probed, so "monitor 2" means the same thing to the
        # controller and to the window that has to jump there — see
        # monitors.py for why that matters.
        self._monitors: list[monitors_mod.Monitor] = []
        self._pet_monitor: Optional[int] = None

        # Wake-word sensitivity is live-tunable (tray tuner), so it's read
        # through a callable on every frame rather than captured per listen.
        self._wake_threshold = config.WAKE_WORD_THRESHOLD
        self._near_misses = wake_word.NearMissLog()

        # The voice the server last picked for us with `speak_as` ("" = the
        # configured default). Held here rather than passed straight through
        # to one tts.speak() call because it's sticky by default — see
        # _apply_voice for why.
        self._voice_id = ""
        self._voice_name = ""

        self._barge_in: Optional[barge_in.BargeInDetector] = None
        # Everything the pet says goes through here; see speech.py for why the
        # four hand-rolled copies of this became one.
        self._speaker = speech.Speaker(
            state=self._state, tts=tts,
            history=lambda text: self.history.add(history_mod.PET, text, time.time()),
            on_said=self.said.emit, on_log=self.log.emit,
            barge_in=lambda: self._barge_in,
            detail_of=self._barge_in_detail,
            voice_id=lambda: self._voice_id,
            on_level=self.mouth.emit,
        )
        self._napping = False
        self._nap_forced: Optional[bool] = None  # petctl nap on/off overrides the schedule
        self._last_nap_check = 0.0

        # When a reply ends on a question the pet keeps listening for the
        # answer. _follow_ups counts how many have chained without you
        # re-triggering, so a server that ends every reply with "?" can't
        # loop forever off mic noise.
        self._pending_follow_up = False
        self._follow_ups = 0

        self._last_update_check = 0.0
        self._update_pending = False  # applied on disk, waiting for the restart
        # Armed by `petctl self_restart`, fired after the turn it was asked in
        # (see _arm_self_restart for why it can't happen inline).
        self._restart_context = None

        self._notification_watcher: Optional[notifications.NotificationWatcher] = None
        self._notification_gate = notifications.NotificationGate(
            config.NOTIFICATION_FILTER, config.NOTIFICATION_MIN_INTERVAL_SECONDS
        )
        self._pending_notifications: list[notifications.Notification] = []
        self._notification_lock = threading.Lock()

    # ── external controls (safe to call from the Qt/UI thread) ─────────

    def request_talk_now(self) -> None:
        self._talk_now.set()

    def toggle_mute(self) -> bool:
        self._muted = not self._muted
        self.log.emit("Muted." if self._muted else "Unmuted.")
        return self._muted

    def set_napping(self, napping: Optional[bool]) -> None:
        """Force the nap state on/off, or pass None to hand control back to
        the quiet-hours schedule."""
        self._nap_forced = napping
        if napping is not None:
            self._apply_nap_state(napping)

    def wake_threshold(self) -> float:
        return self._wake_threshold

    def set_wake_threshold(self, value: float) -> None:
        self._wake_threshold = min(max(float(value), 0.01), 0.99)

    def wake_stats(self) -> dict:
        return {"peak": self._near_misses.peak, "near_misses": self._near_misses.entries()}

    def reset_wake_stats(self) -> None:
        self._near_misses.clear()

    def current_voice(self) -> str:
        """Name (or id) of the server-picked voice in use, "" for the default."""
        return self._voice_name or self._voice_id

    def reset_voice(self) -> None:
        """Drop a server-picked voice and go back to Bolt's own. The tray's
        way out of a voice you didn't want to keep — the server has no way to
        ask for the default back, since it never learns what it is."""
        if not self._voice_id:
            return
        self._voice_id = self._voice_name = ""
        self.log.emit("Voice: back to the default.")
        self.voice_changed.emit("")

    def stop(self) -> None:
        self._running = False
        self._talk_now.set()  # wake up anything blocked waiting on it
        if self._notification_watcher is not None:
            self._notification_watcher.stop()

    # ── internal ─────────────────────────────────────────────────────────

    def _handle_state_change(self, _old: PetState, new: PetState) -> None:
        self.state_changed.emit(new.value)

    def _should_continue(self) -> bool:
        return self._running

    def run(self) -> None:
        """Thread entry point (connected to QThread.started)."""
        missing = config.missing_config()
        if missing:
            self.log.emit(f"Missing config: {', '.join(missing)} — set them in .env and restart.")
            self.finished.emit()
            return
        try:
            self._stream = mic.open_input_stream()
        except Exception as exc:
            self.log.emit(f"Could not open microphone: {exc}")
            self.finished.emit()
            return

        if config.BARGE_IN:
            # In wake mode the detector shares the idle listener's model and
            # its live threshold, so the tray tuner's slider applies to
            # interrupting as well as waking (unless BARGE_IN_WAKE_THRESHOLD
            # pins it to a fixed, stricter number).
            self._barge_in = barge_in.make_detector(
                self._stream,
                wake_threshold=(
                    config.BARGE_IN_WAKE_THRESHOLD
                    if config.BARGE_IN_WAKE_THRESHOLD is not None
                    else self.wake_threshold
                ),
            )
            self.log.emit(f"Barge-in: {config.BARGE_IN_MODE} mode.")

        with self._stream:
            try:
                health = server_client.check_health()
                self.log.emit(f"Connected to server: {health}")
            except Exception as exc:
                self.log.emit(f"Server not reachable yet ({exc}) — will keep trying per-request.")
            self._start_notification_bridge()
            self._report_self_restart()
            self._loop()
        if self._notification_watcher is not None:
            self._notification_watcher.stop()
        self.finished.emit()

    def _loop(self) -> None:
        while self._running:
            if self._muted:
                triggered = self._talk_now.wait(timeout=0.5)
                if not self._running:
                    return
                if not triggered:
                    continue
                self._talk_now.clear()
            else:
                if not self._wait_for_wake_or_click():
                    if not self._running:
                        return
                    continue
            self._handle_conversation_turn()

    def _wait_for_wake_or_click(self) -> bool:
        """True once either the wake phrase was heard or a click-to-talk
        request came in; False on a spurious wakeup (loop again)."""
        def should_continue() -> bool:
            return self._running and not self._talk_now.is_set()

        detected = wake_word.listen_for_wake_word(
            self._stream,
            should_continue=should_continue,
            threshold=self.wake_threshold,  # callable: the tuner slider is live
            on_tick=self._maybe_heartbeat,
            on_score=self._observe_wake_score,
        )
        if not self._running:
            return False
        if detected:
            self._talk_now.clear()  # in case both fired around the same time
            return True
        if self._talk_now.is_set():
            self._talk_now.clear()
            return True
        return False

    def _observe_wake_score(self, score: float, threshold: float) -> None:
        self._near_misses.observe(score, threshold, time.time())

    def _handle_conversation_turn(self) -> None:
        # A turn you started yourself ends any follow-up chain in progress.
        following_up, self._pending_follow_up = self._pending_follow_up, False
        if not following_up:
            self._follow_ups = 0

        self._state.transition(PetState.LISTENING)
        # Transcribe while they talk rather than after: frames go to Deepgram
        # as they are captured, so the text is ready the moment the VAD says
        # they stopped. None here just means the one-shot path will do it.
        streamed = stt_stream.StreamingTranscriber.open()
        pcm = mic.record_utterance(
            self._stream,
            should_continue=self._should_continue,
            # Answering a question deserves longer than saying the wake word
            # on purpose does — you were just asked something.
            grace_s=config.FOLLOW_UP_GRACE_SECONDS if following_up else None,
            on_frame=streamed.feed if streamed is not None else None,
        )
        if pcm is None:
            if streamed is not None:
                streamed.finish()
            self._follow_ups = 0  # silence ends the chain
            self._state.transition(PetState.IDLE)
            return

        self._state.transition(PetState.THINKING)
        try:
            text = self._transcribe(pcm, streamed)
        except stt.SttError as exc:
            self.log.emit(f"STT failed: {exc}")
            self._state.transition(PetState.ERROR)
            self._state.transition(PetState.IDLE)
            return
        if not text:
            self._state.transition(PetState.IDLE)
            return
        self.log.emit(f"You: {text}")
        self.history.add(history_mod.USER, text, time.time())

        try:
            # What's focused right now rides along, so "what's this error?"
            # has a referent without you having to describe the window.
            reply = self._ask_server(self._with_context(text))
        except server_client.ServerError as exc:
            self.log.emit(f"Server error: {exc}")
            self._state.transition(PetState.ERROR)
            self._state.transition(PetState.IDLE)
            return

        self._check_deliveries()
        self._apply_voice(reply)
        if reply.spoken:
            # Streamed: every sentence was spoken and logged as it arrived.
            # The text still matters — a reply ending on a question should keep
            # the mic open — but saying it again would repeat the whole answer.
            self._after_speaking(reply.text, completed=True)
        else:
            self._speak(reply.text)
        self._state.transition(PetState.IDLE)
        self._maybe_self_restart()

    def _transcribe(self, pcm, streamed) -> str:
        """The transcript, from the live stream if it produced one.

        The fallback is not a rare path to be tolerated — it is the safety net
        that lets streaming be switched on at all. Whatever happened to the
        socket, the full audio is still buffered here, so a failed stream costs
        one ordinary upload and nothing else."""
        if streamed is not None:
            try:
                text = streamed.finish()
            except Exception:
                self.log.emit("Streaming STT failed — falling back.")
                text = ""
            if text:
                self.log.emit("(transcribed while you spoke)")
                return text
        return stt.transcribe(pcm)

    def _with_context(self, text: str) -> str:
        """Everything the server gets alongside what you actually said: the
        focused window title, and a one-line note about the screen layout so
        Bolt knows how many monitors there are and where he's standing
        without having to ask. Only the *layout* rides along for free — the
        text on those screens costs an OCR pass, so it stays behind
        `petctl read`."""
        text = screen_context.context_for(text)
        if config.MONITOR_CONTEXT:
            text = monitors_mod.annotate(text, self._monitors, self._pet_monitor)
        return text

    # ── screen layout, published by the UI ───────────────────────────────

    def set_monitors(self, monitors: list) -> None:
        """Slot: the window telling us what screens exist (queued signal)."""
        self._monitors = list(monitors)
        self.log.emit(
            "Screens: " + (monitors_mod.summary(self._monitors) or "none reported")
        )

    def set_pet_monitor(self, index: int) -> None:
        """Slot: the window telling us which screen the pet is standing on."""
        self._pet_monitor = int(index)

    def _handle_command(self, command: str) -> str:
        """Server-relayed command. `petctl ...` drives the pet's body and
        `filectl ...` does local file read/write/edit — neither ever reaches
        a shell; everything else is a real command, exactly as before (see
        the security notes in the README)."""
        try:
            action = pet_actions.parse(command)
        except pet_actions.ActionError as exc:
            self.log.emit(f"petctl: {exc}")
            return f"[pet] {exc}"
        if action is not None:
            self.log.emit(f"Pet action: {action}")

            # Queries answer from here rather than from pet_actions.describe():
            # their output *is* the useful part, and it's what the server reads
            # back off the tool-result relay.
            kind = action["action"]
            if kind == "monitors":
                return monitors_mod.describe(self._monitors, self._pet_monitor)
            if kind == "self_restart":
                return self._arm_self_restart(action.get("reason") or "")
            if kind == "voice":
                # Answered here, not by describe(): the UI has no part in it,
                # and the server needs to hear whether there was anything to
                # drop — it can't see which voice we're using.
                previous = self.current_voice()
                self.reset_voice()
                return (
                    f"[pet] back to your own voice (was {previous})" if previous
                    else "[pet] already using your own voice"
                )
            if kind == "read":
                return self._read_screen(action["target"])
            if kind == "jump":
                try:
                    target = monitors_mod.resolve(
                        self._monitors, action["target"], self._pet_monitor
                    )
                except ValueError as exc:
                    self.log.emit(f"petctl jump: {exc}")
                    return f"[pet] {exc}"
                # Hand the window a resolved index, so it can't re-resolve the
                # spec against a different screen ordering.
                self.action.emit({"action": "jump", "monitor": target.index})
                return f"[pet] jumped to monitor {target.label}"

            if kind == "nap":
                self.set_napping(bool(action["enabled"]))
            self.action.emit(action)
            return pet_actions.describe(action)

        try:
            scene = dialogue_mod.parse(command)
        except dialogue_mod.DialogueError as exc:
            self.log.emit(f"dialoguectl: {exc}")
            return f"[dialogue] {exc}"
        if scene is not None:
            return self._play_dialogue(scene)

        try:
            file_action = file_ops.parse(command)
        except file_ops.FileOpError as exc:
            self.log.emit(f"filectl: {exc}")
            return f"[filectl] {exc}"
        if file_action is not None:
            self.log.emit(file_ops.describe(file_action))
            try:
                return file_ops.execute(file_action)
            except file_ops.FileOpError as exc:
                self.log.emit(f"filectl: {exc}")
                return f"[filectl] {exc}"

        return server_client.run_local_command(command)

    def _read_screen(self, target: str) -> str:
        """`petctl read` — OCR a screen and hand the text back to the server."""
        if not config.SCREEN_TEXT:
            return "[pet] screen reading is disabled (set SCREEN_TEXT=true in .env)"
        if not self._monitors:
            return "[pet] no monitor information available"
        limit = config.SCREEN_TEXT_MAX_CHARS
        if target in ("all", "everything", "*"):
            self.log.emit(f"Reading all {len(self._monitors)} screens…")
            return screen_text.read_monitors(self._monitors, limit)
        if target in ("here", "", "this", "current"):
            index = self._pet_monitor if self._pet_monitor is not None else 0
            monitor = self._monitors[min(index, len(self._monitors) - 1)]
        else:
            try:
                monitor = monitors_mod.resolve(
                    self._monitors, target, self._pet_monitor
                )
            except ValueError as exc:
                return f"[pet] {exc}"
        self.log.emit(f"Reading monitor {monitor.number} ({monitor.name})…")
        return screen_text.read_monitor(monitor, limit)

    def _arm_self_restart(self, reason: str) -> str:
        """`petctl self_restart` — check the code, then arm a restart.

        Nothing restarts here. The tool result has to get back up the relay
        before this process can die (otherwise the server waits out its
        timeout on a turn that will never finish), so the restart is armed and
        `_maybe_self_restart` fires it once the turn has been spoken. The
        preflight import runs *now*, in this turn, so a syntax error Bolt just
        introduced comes back as something he can read and fix rather than as
        a pet that never comes back."""
        if not config.SELF_RESTART:
            return "[pet] self-restart is disabled on this device (SELF_RESTART=false)"
        if self._restart_context is not None:
            return "[pet] a restart is already armed for the end of this turn"
        try:
            self_restart.check_loop_guard(self_restart.load())
            self.log.emit("Self-restart requested — checking the code imports first…")
            self_restart.preflight()
        except self_restart.RestartError as exc:
            self.log.emit(f"Self-restart refused: {exc}")
            return f"[pet] restart refused — {exc}"

        recent = [entry.text[:120] for entry in self.history.entries()[-4:]]
        self._restart_context = self_restart.arm(
            reason or "no reason given",
            verify=reason,
            version=__version__,
            session=config.SESSION_ID,
            recent=recent,
        )
        self.log.emit("Self-restart armed; it happens after this turn.")
        return (
            "[pet] code imports cleanly; restarting as soon as this turn finishes. "
            "I'll come back and tell you what version I'm on and what I found — "
            "wrap up your reply now, the next thing you hear from me is the report."
        )

    def _maybe_self_restart(self) -> bool:
        """Fire an armed restart, once the turn is over and the reply spoken.

        Returns True if a restart was requested, so the caller can stop
        driving the pipeline — the process is on its way out."""
        if self._restart_context is None:
            return False
        self._update_pending = True  # same latch the updater uses: no double restart
        self.log.emit("Restarting now.")
        self._state.force(PetState.IDLE)
        self.restart_requested.emit(f"self-restart: {self._restart_context.reason[:60]}")
        return True

    def _report_self_restart(self) -> None:
        """On the way up: tell the server we're back, and why we left.

        Runs once, before the listen loop starts, and only when a context file
        was left behind. The report goes through the ordinary conversation
        path, so Bolt's answer is spoken out loud like any other turn — which
        is what makes "restart and check the sprites load" finish as a
        sentence instead of a silence."""
        context = self_restart.load()
        if context is None:
            return
        self_restart.clear()
        message = self_restart.report(context, version=__version__)
        self.log.emit(f"Back from a self-restart ({context.reason[:80]}).")
        self.history.add(history_mod.SYSTEM, message, time.time())
        try:
            reply = server_client.converse(message, on_command=self._handle_command)
        except server_client.ServerError as exc:
            # The restart still worked; only the report failed. Say so locally
            # rather than pretending nothing happened.
            self.log.emit(f"Couldn't report the restart to the server: {exc}")
            return
        self._apply_voice(reply)
        if reply.text.strip() and not self._napping:
            self._speak(reply.text)
        self._state.force(PetState.IDLE)

    def _ask_server(self, text: str):
        """One turn with the server, streamed when possible.

        Streaming speaks each sentence as it is generated, so the wait is
        time-to-first-sentence rather than the whole model call. It falls back
        to the ordinary request/response path when the server has no streaming
        endpoint, or when a stream dies *before* anything was spoken — after
        that, retrying would say the first half twice."""
        if config.STREAMING_REPLIES:
            try:
                return server_client.converse_stream(
                    text, on_say=self._speak_stream_chunk,
                    on_command=self._handle_command,
                )
            except server_client.ServerError as exc:
                self.log.emit(f"Streaming unavailable ({exc}) — using the plain path.")
        return server_client.converse(
            text, on_command=self._handle_command, on_say=self._speak_holding,
        )

    def _play_dialogue(self, scene: dict) -> str:
        """Play a `dialoguectl` scene and report back up the relay.

        This runs *mid-turn* (the server is still waiting on the tool result),
        so the pet has to look like it's talking and then go back to waiting —
        hence the TALKING → THINKING leg rather than the usual return to IDLE.
        Everything a normal reply gets, a scene gets too: the bubble, the
        transcript, and barge-in, so a long scene can be talked over exactly
        like a long answer."""
        if not config.DIALOGUE:
            return "[dialogue] disabled on this device (DIALOGUE=false)"
        try:
            inputs = dialogue_mod.resolve(
                scene,
                voices=dialogue_mod.parse_voice_map(config.DIALOGUE_VOICES),
                self_voice=self._voice_id or config.ELEVENLABS_VOICE_ID,
            )
        except dialogue_mod.DialogueError as exc:
            self.log.emit(f"dialoguectl: {exc}")
            return f"[dialogue] {exc}"

        text = dialogue_mod.spoken_text(scene)
        self.log.emit(f"Dialogue ({len(inputs)} lines): {text[:120]}")
        try:
            pcm, sample_rate = tts.synthesize_dialogue(
                inputs, model_id=scene.get("model"), stability=scene.get("stability")
            )
        except tts.TtsError as exc:
            self.log.emit(f"Dialogue failed: {exc}")
            # Reported, not raised: the server can read this, shorten the
            # scene or fix the voice, and try again inside the same turn.
            return f"[dialogue] couldn't synthesize it: {exc}"

        completed = self._speaker.say_pcm(
            speech.Utterance.scene(text), pcm=pcm, sample_rate=sample_rate)

        if not completed:
            return dialogue_mod.describe(scene) + " (interrupted — they talked over it)"
        return dialogue_mod.describe(scene)

    def _apply_voice(self, reply) -> None:
        """Adopt (or drop) the voice the server tagged this reply with.

        The server's `speak_as` marker names an ElevenLabs voice it just
        picked — and it adds a Voice Library pick to the account first, so by
        the time the id gets here it's usable for TTS. It tags *one* reply,
        but the voice sticks by default: the server strips the marker before
        storing the turn, so it can't recall the id later, and "keep talking
        like that" would otherwise send it searching for a voice all over
        again. `VOICE_STICKY=false` makes each pick last exactly one reply.

        Untagged replies never *change* the voice — with stickiness on they
        just keep whatever's in use, which is what makes the rest of the
        conversation stay in the requested voice."""
        voice_id = getattr(reply, "voice_id", "")
        if voice_id:
            if voice_id != self._voice_id:
                self._voice_id = voice_id
                self._voice_name = getattr(reply, "voice_name", "") or ""
                self.log.emit(f"Voice: {self.current_voice()}")
                self.voice_changed.emit(self.current_voice())
        elif not config.VOICE_STICKY:
            self.reset_voice()

    def _speak(self, text: str) -> None:
        """The answer: transcript, bubble, and the follow-up rule."""
        completed = self._speaker.say(speech.Utterance.reply(text))
        self._after_speaking(text, completed=completed,
                             detail=self._speaker.last_detail)

    def _speak_holding(self, text: str) -> None:
        """"Give me a sec" while a tool runs — filler, so no transcript, and it
        resumes the state it interrupted because the turn isn't over."""
        self._speaker.say(speech.Utterance.holding(text))

    def _speak_stream_chunk(self, text: str) -> None:
        """One sentence of a streamed answer, spoken the moment it arrives."""
        completed = self._speaker.say(speech.Utterance.stream_chunk(text))
        if not completed:
            self.log.emit("Interrupted — listening.")
            self._follow_ups = 0
            self._talk_now.set()

    def _after_speaking(self, text: str, *, completed: bool, detail: str = "") -> None:
        """What happens once an answer has been said, however it was said.

        Shared by the plain and streamed paths: a streamed reply is spoken
        sentence by sentence, but it still has to obey the same rules about
        keeping the mic open when it ended on a question."""
        if not completed:
            # You talked over it — take that as the start of the next turn
            # rather than making you say the wake word again.
            self.log.emit(f"Interrupted — listening. {detail}")
            self._follow_ups = 0  # you're clearly engaged; start the count over
            self._talk_now.set()
        elif self._should_follow_up(text):
            self._follow_ups += 1
            cap = config.FOLLOW_UP_MAX_TURNS
            self.log.emit(
                f"Asked a question — listening for your answer "
                f"({self._follow_ups}{'/' + str(cap) if cap > 0 else ''})."
            )
            self._pending_follow_up = True
            self._talk_now.set()

    def _should_follow_up(self, text: str) -> bool:
        """Whether *text* leaves the pet waiting on an answer.

        The rule itself — question, cap, off switch — is
        `speech.follow_up_decision`, because it is a rule with an off-by-one in
        it and deserves a test that needs no audio. What stays here is the part
        that is genuinely the controller's: mute, and the log line. Muted is
        excluded because mute means "don't listen to me" and an automatic turn
        would walk straight past it. Napping isn't: quiet hours suppress the
        pet *starting* something, and a question is only ever asked in reply to
        you."""
        if self._muted:
            return False
        keep, why = speech.follow_up_decision(text, completed=True, follow_ups=self._follow_ups)
        if not keep and "cap" in why:
            # Only worth mentioning the cap on a reply that would otherwise have
            # kept listening, or it fires on every statement the pet makes.
            self.log.emit("Follow-up limit reached — say the wake word to keep going.")
        return keep

    def _barge_in_detail(self) -> str:
        """Why the interruption fired, for the log. How far into playback it
        happened is the tell: frame 1 means the detector was still holding
        audio from before this reply started, whereas a hit several seconds
        in is something the mic actually heard."""
        detector = self._barge_in
        if isinstance(detector, barge_in.WakeWordBargeIn):
            return (
                f"(wake score {detector.last_score:.3f} >= {detector.last_threshold:.2f}, "
                f"peak {detector.peak_score:.3f}, at frame {detector.frames_checked} / "
                f"{detector.seconds_checked:.1f}s into playback)"
            )
        if isinstance(detector, barge_in.BargeInDetector):
            return f"(loud frames {detector.loud_frames}, threshold {config.BARGE_IN_RMS_THRESHOLD})"
        return ""

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

    def _apply_nap_state(self, napping: bool) -> None:
        if napping == self._napping:
            return
        self._napping = napping
        self.log.emit("Napping — no proactive noise." if napping else "Awake.")
        self.napping.emit(napping)

    def _refresh_nap_state(self) -> None:
        if self._nap_forced is not None:
            self._apply_nap_state(self._nap_forced)
            return
        now = time.monotonic()
        if now - self._last_nap_check < _NAP_CHECK_INTERVAL_SECONDS:
            return
        self._last_nap_check = now
        napping = quiet.is_quiet(
            config.QUIET_HOURS,
            on_error=lambda exc: self.log.emit(f"QUIET_HOURS is malformed ({exc}) — ignoring it."),
        )
        if not napping and config.DND_ON_FULLSCREEN:
            napping = screen_context.is_fullscreen_active()
        self._apply_nap_state(napping)

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

    def _start_notification_bridge(self) -> None:
        if not config.NOTIFICATION_BRIDGE:
            return
        watcher = notifications.NotificationWatcher(self._queue_notification)
        problem = watcher.start()
        if problem:
            self.log.emit(problem)
            return
        self._notification_watcher = watcher
        self.log.emit("Notification bridge on.")

    def _queue_notification(self, notification: notifications.Notification) -> None:
        """Called on the watcher thread — just queue it; forwarding happens on
        the pipeline thread where it can't collide with a live conversation.

        Only the *filter* applies here. The rate limit used to as well, which
        meant a second message arriving inside the window was silently thrown
        away — with NOTIFICATION_MIN_INTERVAL_SECONDS=60, two texts a minute
        apart and you only ever heard about one of them. Losing a message from
        a person to save a round trip is the wrong trade; they are batched at
        the far end instead, which costs the same one round trip and keeps
        them all."""
        if not self._notification_gate.matches(notification):
            return
        with self._notification_lock:
            if len(self._pending_notifications) >= _MAX_PENDING_NOTIFICATIONS:
                self._pending_notifications.pop(0)  # bound it; oldest goes first
            self._pending_notifications.append(notification)

    def _drain_notifications(self) -> None:
        """Forward everything waiting as ONE turn.

        Batching is what makes it safe to keep every notification: five that
        arrived while the pet was mid-conversation become one message and one
        round trip, instead of five separate interruptions queued up to fire
        back to back."""
        with self._notification_lock:
            pending, self._pending_notifications = self._pending_notifications, []
        if not pending or not self._running or self._napping:
            return
        for notification in pending:
            self.log.emit(f"Notification: {notification.as_text()}")
            self.history.add(history_mod.SYSTEM, notification.as_text(), time.time())

        if len(pending) == 1:
            message = f"[desktop notification] {pending[0].as_text()}"
        else:
            lines = "\n".join(f"- {n.as_text()}" for n in pending)
            message = f"[{len(pending)} desktop notifications]\n{lines}"
        try:
            reply = server_client.converse(
                message, on_command=self._handle_command, on_say=self._speak_holding,
            )
        except server_client.ServerError as exc:
            self.log.emit(f"Couldn't forward notification: {exc}")
            return
        self._check_deliveries()
        self._apply_voice(reply)
        if reply.text.strip() and not reply.spoken:
            self._speak(reply.text)
        self._state.transition(PetState.IDLE)

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

    def _check_deliveries(self) -> None:
        """Download anything the server has queued via deliver_files —
        called right after a conversation/notification turn (the common
        case: "send me that file") and once per heartbeat for anything
        queued out-of-band. Best-effort: a failure here is logged, not
        raised, so it can't sour a turn that already got its spoken reply."""
        if not config.RECEIVE_FILES:
            return
        try:
            queued = server_client.list_outbox_files()
        except server_client.ServerError as exc:
            self.log.emit(f"Couldn't check for delivered files: {exc}")
            return
        for entry in queued:
            file_id = entry.get("id")
            name = entry.get("name") or file_id
            if not file_id:
                continue
            try:
                data = server_client.download_outbox_file(file_id)
            except server_client.ServerError as exc:
                self.log.emit(f"Couldn't download {name}: {exc}")
                continue
            path = file_delivery.save(config.DELIVERED_FILES_DIR, name, data)
            self.log.emit(f"Received file: {path}")

    # ── auto-update ──────────────────────────────────────────────────────

    def _maybe_update(self) -> None:
        """Poll the Gitea releases page and, if there's a newer tag, apply it
        and ask the UI to restart.

        Only ever runs from the wake-listener's tick, so the pet is IDLE and
        between turns by construction — an update can't land mid-sentence.
        Failures are logged and the interval resets, so a server that's down
        (or a release that rolls back) costs one log line an hour, not a
        retry storm."""
        if not config.AUTO_UPDATE or self._update_pending:
            return
        now = time.monotonic()
        if now - self._last_update_check < config.UPDATE_CHECK_INTERVAL_SECONDS:
            return
        self._last_update_check = now
        try:
            release = updater.check_for_update()
        except updater.UpdateError as exc:
            self.log.emit(f"Update check failed: {exc}")
            return
        if release is None:
            return

        self.log.emit(f"Update available: {release.tag} — applying.")
        try:
            previous = updater.apply_update(release.tag, on_log=self.log.emit)
        except updater.UpdateError as exc:
            self.log.emit(f"Update to {release.tag} failed: {exc}")
            return

        self._update_pending = True
        self.log.emit(f"Updated {previous} -> {release.tag}; restarting.")
        if not self._napping:
            # Napping means no proactive noise, so a silent restart it is.
            self._speak(f"Updating to {release.tag}. Back in a second.")
            self._state.transition(PetState.IDLE)
        self.restart_requested.emit(release.tag)

    # ── heartbeat ────────────────────────────────────────────────────────

    def _maybe_heartbeat(self) -> None:
        """Called from the wake listener's tick (~every WAKE_CHECK_INTERVAL_SECONDS).

        Two different cadences live here, and conflating them was costing
        minutes. A *notification* is an event that already happened — it should
        go out as soon as the pet is free, which is the next tick. The
        *heartbeat* is a poll, and polling the server every 1.2s would be
        absurd, so it stays on its own interval."""
        self._refresh_nap_state()
        self._maybe_update()
        if self._update_pending:
            return  # on the way out — don't start a conversation now

        # Notifications: every tick, not every heartbeat.
        if self._state.state == PetState.IDLE and not self._napping:
            self._drain_notifications()

        now = time.monotonic()
        if now - self._last_heartbeat < config.HEARTBEAT_INTERVAL_SECONDS:
            return
        if self._state.state != PetState.IDLE:
            # Mid-conversation. Do NOT stamp the clock — an earlier version
            # did, so a heartbeat that landed while the pet was talking burned
            # its slot and waited another full interval. With follow-up
            # listening that could repeat for several cycles, which is why a
            # notification could sit unspoken for five or ten minutes.
            return
        if self._napping:
            return  # quiet hours: still answers when spoken to, just doesn't start
        self._last_heartbeat = now
        self._check_deliveries()
        if self._state.state != PetState.IDLE:
            return
        try:
            announcement = server_client.report_status()
        except server_client.ServerError as exc:
            self.log.emit(f"Heartbeat failed: {exc}")
            return
        if announcement:
            self._speak(announcement)
            self._state.transition(PetState.IDLE)