"""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)