From e9d92b0ba2d5a471e385a85d7d7ed530c7260498 Mon Sep 17 00:00:00 2001 From: themajesticmagician Date: Thu, 23 Jul 2026 07:55:43 -0600 Subject: [PATCH] ... --- .claude/settings.local.json | 3 +- .env.example | 24 +++++ CLAUDE.md | 33 ++++++- bolt_pet/audio/mic.py | 9 +- bolt_pet/config.py | 28 ++++++ bolt_pet/controller.py | 54 ++++++++++- bolt_pet/server_client.py | 19 +++- bolt_pet/speech_text.py | 16 ++++ bolt_pet/sudo_askpass.py | 145 ++++++++++++++++++++++++++++++ bolt_pet/updater.py | 24 +++++ tests/test_controller_features.py | 142 +++++++++++++++++++++++++++++ tests/test_speech_text.py | 22 ++++- tests/test_sudo_askpass.py | 122 +++++++++++++++++++++++++ tests/test_updater.py | 27 +++++- 14 files changed, 657 insertions(+), 11 deletions(-) create mode 100644 bolt_pet/sudo_askpass.py create mode 100644 tests/test_sudo_askpass.py diff --git a/.claude/settings.local.json b/.claude/settings.local.json index b6c36e9..7ab6472 100644 --- a/.claude/settings.local.json +++ b/.claude/settings.local.json @@ -29,7 +29,8 @@ "Bash(QT_QPA_PLATFORM=offscreen .venv/bin/python -c ' *)", "Bash(git push *)", "Bash(git remote *)", - "Bash(grep -v '^$')" + "Bash(grep -v '^$')", + "Bash(.venv/bin/pip install *)" ] } } diff --git a/.env.example b/.env.example index ea57cc7..1cfd165 100644 --- a/.env.example +++ b/.env.example @@ -40,6 +40,17 @@ ELEVENLABS_VOICE_ID= #VAD_SILENCE_END_SEC=1.2 #VAD_MAX_UTTERANCE_SECONDS=15 #VAD_MIN_UTTERANCE_SECONDS=0.4 +#VAD_GRACE_SECONDS=4 # how long to wait for you to start talking + +# ── Follow-up listening (optional) ────────────────────────────────────────── +# When a reply ends on a question, the pet keeps listening for your answer +# instead of dropping back to idle and making you say the wake word again. +# FOLLOW_UP_MAX_TURNS caps how many question-and-answer rounds can chain +# without you re-triggering it (0 = no cap) — a stop on runaway loops if the +# server ends every reply with "?" and the mic keeps feeding it noise. +#FOLLOW_UP_LISTEN=true +#FOLLOW_UP_MAX_TURNS=3 +#FOLLOW_UP_GRACE_SECONDS=7 # longer than VAD_GRACE_SECONDS: you were asked something # ── Pet window (optional) ─────────────────────────────────────────────────── #PET_SIZE=160 @@ -128,6 +139,19 @@ ELEVENLABS_VOICE_ID= #WAKE_NEAR_MISS_MARGIN=0.2 #WAKE_NEAR_MISS_LIMIT=40 +# ── sudo password prompts (optional) ──────────────────────────────────────── +# The pet has no terminal, so a server-relayed `sudo` would block forever on +# a tty nobody is watching. With this on, bare `sudo` becomes `sudo -A` and +# the password is collected in a desktop dialog you have to answer — a real +# askpass binary if one is installed, otherwise a generated zenity/kdialog +# wrapper in ~/.cache/bolt-pet/askpass.sh. +# Set it to false if you'd rather Bolt never be able to ask for root: sudo +# commands then just fail. Read the dialogs — that box is the only thing +# between "Bolt decided to run sudo" and it running. +#SUDO_ASKPASS_PROMPT=true +#SUDO_ASKPASS_HELPER= # blank = auto-detect +#SUDO_COMMAND_TIMEOUT_SECONDS=180 # long enough for a human to answer + # ── Misc (optional) ────────────────────────────────────────────────────────── #COMMAND_TIMEOUT_SECONDS=30 #HEARTBEAT_INTERVAL_SECONDS=60 diff --git a/CLAUDE.md b/CLAUDE.md index 830423b..6d1e394 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -108,6 +108,14 @@ logs a missing-config message and exits its thread instead of starting. `dbus-monitor`, parses Notify calls (pure `iter_notifications()`), filters and rate-limits them (`NotificationGate`), and the controller forwards survivors through `converse()`. Off by default — each one is a round trip. +- **`sudo_askpass.py`** — makes server-relayed `sudo` usable from a process + with no terminal, by pointing sudo's `SUDO_ASKPASS` at a GUI helper and + rewriting bare `sudo` to `sudo -A` (`add_askpass_flag`, a conservative regex + that skips anything already carrying a flag and anything inside quotes). + Prefers a real askpass binary and falls back to generating a + zenity/kdialog wrapper in `~/.cache/bolt-pet/askpass.sh`. Resolution order + is injectable (`is_executable`/`which`) so it's testable on a machine with a + different set installed. See the security notes — the dialog is the boundary. - **`updater.py`** — self-update from the Gitea releases API. Polls `/releases/latest` for a tag newer than `bolt_pet.__version__` and moves the checkout to it with @@ -137,7 +145,13 @@ logs a missing-config message and exits its thread instead of starting. covered) strips markdown, emoji, URLs and stray symbols the voice would read literally ("asterisk asterisk"), turns bullet lists into full sentences, and words a few symbols (`&` → "and"). `for_display()` is the looser version for - the speech bubble — markdown syntax gone, emoji kept. Pure string logic, no + the speech bubble — markdown syntax gone, emoji kept. `is_question()` decides + whether a reply leaves the pet waiting on an answer: it tests the *spoken* + form (so a '?' inside a stripped code block or URL doesn't count) and only a + trailing one counts, since a question asked in passing isn't awaiting a + reply. `controller._should_follow_up` uses it to keep listening without the + wake word, capped by `FOLLOW_UP_MAX_TURNS` so a server that ends every reply + with a question can't loop forever off mic noise. Pure string logic, no Qt/audio imports. - **`ui/`** — `app.py` wires `QApplication` + `PetWindow` + `PetTray` + the history/tuner windows + the push-to-talk hotkey + the controller thread @@ -231,6 +245,23 @@ matches the trust model of the server repo's other desk clients. Keep `DESK_API_KEY` private and don't expose the desk API port to the open internet. +`sudo_askpass.py` widens that further, by design: with `SUDO_ASKPASS_PROMPT` +on (the default), a relayed bare `sudo` is rewritten to `sudo -A` and the +password is collected in a desktop dialog, so commands can escalate to root +instead of hanging on a tty the pet doesn't have. The dialog is the security +boundary — it's the only thing between the server deciding to run `sudo` and +it running, so the prompt is deliberately not suppressible per-command and +those commands get their own longer timeout (`SUDO_COMMAND_TIMEOUT_SECONDS`) +rather than being made non-interactive. Set `SUDO_ASKPASS_PROMPT=false` to +take the capability away entirely; sudo commands then fail. Note that +`sudo -n` / `sudo -A` / `sudo -u …` in a relayed command are never rewritten, +so an explicit non-interactive sudo stays non-interactive. + +**Don't run the pet as root.** It needs no privileges of its own, PortAudio +can't reach the user's PipeWire socket from a root session (raw ALSA devices +reject the 16 kHz capture rate — `paInvalidSampleRate`), and every relayed +command would run unconstrained. + Two newer features widen what leaves this machine, both switchable in `.env`: `SCREEN_CONTEXT` appends the focused window's *title* to each utterance (titles often contain file paths, document names, or subject lines), and diff --git a/bolt_pet/audio/mic.py b/bolt_pet/audio/mic.py index ba0e933..52beaaa 100644 --- a/bolt_pet/audio/mic.py +++ b/bolt_pet/audio/mic.py @@ -44,6 +44,7 @@ def record_utterance( silence_end_sec: float = None, max_utterance_s: float = None, min_utterance_s: float = None, + grace_s: float = None, frame_len: int = config.FRAME_LEN, sample_rate: int = config.SAMPLE_RATE, ) -> Optional[np.ndarray]: @@ -53,18 +54,24 @@ def record_utterance( *should_continue* is polled each frame so a caller can cancel recording (e.g. the pet window was closed) without needing threading primitives baked into this function. + + *grace_s* is how long to wait for speech to *begin* before giving up. + The controller stretches it for follow-up questions, where you're being + asked something and need a moment to think rather than having just said + the wake word on purpose. """ rms_threshold = config.RMS_THRESHOLD if rms_threshold is None else rms_threshold silence_end_sec = config.SILENCE_END_SEC if silence_end_sec is None else silence_end_sec max_utterance_s = config.MAX_UTTERANCE_S if max_utterance_s is None else max_utterance_s min_utterance_s = config.MIN_UTTERANCE_S if min_utterance_s is None else min_utterance_s + grace_s = config.GRACE_SECONDS if grace_s is None else grace_s frames: list[np.ndarray] = [] started = False silence_frames = 0 silence_limit = int(silence_end_sec * sample_rate / frame_len) max_frames = int(max_utterance_s * sample_rate / frame_len) - grace_frames = int(4.0 * sample_rate / frame_len) # wait up to 4s for speech to begin + grace_frames = int(grace_s * sample_rate / frame_len) # how long to wait for speech to begin waited = 0 while should_continue(): diff --git a/bolt_pet/config.py b/bolt_pet/config.py index fd9ae1a..5aaa026 100644 --- a/bolt_pet/config.py +++ b/bolt_pet/config.py @@ -78,8 +78,36 @@ RMS_THRESHOLD = int(os.environ.get("VAD_RMS_THRESHOLD", "300")) SILENCE_END_SEC = float(os.environ.get("VAD_SILENCE_END_SEC", "1.2")) MAX_UTTERANCE_S = float(os.environ.get("VAD_MAX_UTTERANCE_SECONDS", "15")) MIN_UTTERANCE_S = float(os.environ.get("VAD_MIN_UTTERANCE_SECONDS", "0.4")) +# How long to wait for you to *start* talking before giving up on a turn. +GRACE_SECONDS = float(os.environ.get("VAD_GRACE_SECONDS", "4")) + +# ── follow-up listening ───────────────────────────────────────────────────── +# When a reply ends on a question, the pet keeps listening for the answer +# instead of dropping back to idle and making you say the wake word again. +# The grace period is longer than a normal turn's because you were asked +# something and may need a beat to think. FOLLOW_UP_MAX_TURNS caps how many +# question-and-answer rounds can chain without you re-triggering it — a stop +# on runaway loops if the server ends every reply with a question and the mic +# keeps feeding it noise. 0 means no cap. + +FOLLOW_UP_LISTEN = os.environ.get("FOLLOW_UP_LISTEN", "true").lower() in ("1", "true", "yes", "on") +FOLLOW_UP_MAX_TURNS = int(os.environ.get("FOLLOW_UP_MAX_TURNS", "3")) +FOLLOW_UP_GRACE_SECONDS = float(os.environ.get("FOLLOW_UP_GRACE_SECONDS", "7")) COMMAND_TIMEOUT_SECONDS = int(os.environ.get("COMMAND_TIMEOUT_SECONDS", "30")) + +# ── sudo password prompts ─────────────────────────────────────────────────── +# The pet has no terminal, so a relayed `sudo` would block on a tty nobody is +# watching. With this on, bare `sudo` is rewritten to `sudo -A` and the +# password is collected in a desktop dialog (a real askpass binary if one is +# installed, otherwise a generated zenity/kdialog wrapper). Turn it off and +# sudo commands simply fail, which is the safer default if you'd rather Bolt +# never be able to ask for root at all. +SUDO_ASKPASS_PROMPT = os.environ.get("SUDO_ASKPASS_PROMPT", "true").lower() in ("1", "true", "yes", "on") +SUDO_ASKPASS_HELPER = os.environ.get("SUDO_ASKPASS_HELPER", "") # blank = auto-detect +# Longer than COMMAND_TIMEOUT_SECONDS because a person has to notice the +# dialog, read it, and type — 30s is nowhere near enough for that. +SUDO_COMMAND_TIMEOUT_SECONDS = int(os.environ.get("SUDO_COMMAND_TIMEOUT_SECONDS", "180")) HEARTBEAT_INTERVAL_SECONDS = float(os.environ.get("HEARTBEAT_INTERVAL_SECONDS", "60")) # ── barge-in (interrupt playback while the pet is talking) ────────────────── diff --git a/bolt_pet/controller.py b/bolt_pet/controller.py index d7b4748..2927ff5 100644 --- a/bolt_pet/controller.py +++ b/bolt_pet/controller.py @@ -63,6 +63,13 @@ class PetController(QObject): 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 @@ -200,9 +207,21 @@ class PetController(QObject): 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) - pcm = mic.record_utterance(self._stream, should_continue=self._should_continue) + 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, + ) if pcm is None: + self._follow_ups = 0 # silence ends the chain self._state.transition(PetState.IDLE) return @@ -270,6 +289,9 @@ class PetController(QObject): on_error=lambda exc: self.log.emit(f"TTS failed: {exc}"), should_stop=should_stop, ) + # Read the scoring history *before* resetting, or the log reports the + # blank counters instead of what actually fired. + detail = self._barge_in_detail() if self._barge_in is not None: # Playback fed the pet's own voice into the wake model's rolling # window. Clear it before the idle listener starts scoring again, @@ -278,8 +300,36 @@ class PetController(QObject): 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. {self._barge_in_detail()}") + 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. + + Muted is excluded because mute means "don't listen to me" — 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 not config.FOLLOW_UP_LISTEN or self._muted: + return False + if not speech_text.is_question(text): + return False + # Only worth mentioning the cap on a reply that would otherwise have + # kept listening, or it fires on every statement the pet makes. + if config.FOLLOW_UP_MAX_TURNS > 0 and self._follow_ups >= config.FOLLOW_UP_MAX_TURNS: + self.log.emit("Follow-up limit reached — say the wake word to keep going.") + return False + return True def _barge_in_detail(self) -> str: """Why the interruption fired, for the log. How far into playback it diff --git a/bolt_pet/server_client.py b/bolt_pet/server_client.py index 524cc47..f0f4e77 100644 --- a/bolt_pet/server_client.py +++ b/bolt_pet/server_client.py @@ -20,7 +20,7 @@ from typing import Callable, Optional import requests -from . import config +from . import config, sudo_askpass _MAX_RELAY_HOPS = 16 @@ -42,12 +42,25 @@ def check_health(timeout: float = 10.0) -> dict: def run_local_command(command: str, timeout: int = None) -> str: """Execute a command relayed by the server, exactly as bolt_desk.py does — "full desktop control" for things like "open firefox" or "how full is my - disk". Runs as the current desktop user. See README security notes.""" + disk". Runs as the current desktop user. See README security notes. + + `sudo` gets special handling: the pet has no terminal, so sudo would sit + waiting on a tty that nobody is looking at. With SUDO_ASKPASS_PROMPT on, + bare `sudo` becomes `sudo -A` and the password is collected in a desktop + dialog you have to answer — which also gives those commands a longer + timeout, since a human has to notice the window and type.""" + env = None + if config.SUDO_ASKPASS_PROMPT and sudo_askpass.needs_password_prompt(command): + helper = sudo_askpass.find_helper() + if helper: + env = sudo_askpass.environment(helper) + command = sudo_askpass.add_askpass_flag(command) + timeout = timeout or config.SUDO_COMMAND_TIMEOUT_SECONDS timeout = timeout or config.COMMAND_TIMEOUT_SECONDS try: completed = subprocess.run( command, shell=True, capture_output=True, text=True, - timeout=timeout, cwd=str(Path.home()), + timeout=timeout, cwd=str(Path.home()), env=env, ) output = (completed.stdout or "") + (completed.stderr or "") return f"[exit {completed.returncode}]\n{output}"[:6000] diff --git a/bolt_pet/speech_text.py b/bolt_pet/speech_text.py index e022e70..5aa2b13 100644 --- a/bolt_pet/speech_text.py +++ b/bolt_pet/speech_text.py @@ -118,6 +118,22 @@ def for_speech(text: str) -> str: return text.strip() +def is_question(text: str) -> bool: + """True if the reply *ends* by asking the user something — the cue for + the pet to keep listening instead of making you say the wake word again. + + Deliberately only looks at the end. A reply that asks something in + passing ("What time is it? It's 7:15.") isn't waiting on an answer, + whereas one that finishes on a question mark is. The test runs on the + spoken form, so a '?' that only exists inside a stripped code block or a + URL doesn't count, and trailing decoration (emoji, quotes, brackets) is + peeled off first so "Ready to go? 🚀" still reads as a question.""" + spoken = for_speech(text) + while spoken and not (spoken[-1].isalnum() or spoken[-1] == "?"): + spoken = spoken[:-1] + return spoken.endswith("?") + + def for_display(text: str) -> str: """What the speech bubble shows: markdown syntax removed (the bubble can't render it) but emoji and layout-ish punctuation left alone.""" diff --git a/bolt_pet/sudo_askpass.py b/bolt_pet/sudo_askpass.py new file mode 100644 index 0000000..791e30a --- /dev/null +++ b/bolt_pet/sudo_askpass.py @@ -0,0 +1,145 @@ +"""Graphical password prompts for server-relayed `sudo` commands. + +The pet has no terminal. When the server relays something like +`sudo apt update`, sudo tries to read a password from a tty, finds none (or +finds the terminal the pet was launched from, which you're not looking at), +and the command fails with no way to answer it. + +sudo's own answer to this is SUDO_ASKPASS: with `-A`, it runs a helper +program and reads the password from the helper's stdout instead of a tty. +Any GUI prompt that prints what was typed works, so this module finds a real +askpass binary if one is installed and otherwise generates a one-line wrapper +around zenity/kdialog, which every desktop has one of. + +Worth being clear about what this changes: the prompt is a *feature*, not +just plumbing. Server-relayed commands already run as your desktop user (see +server_client.run_local_command); this lets them ask to run as root, and the +dialog is the only thing standing between "Bolt decided to run sudo" and it +happening. Leave SUDO_ASKPASS_PROMPT on, and read the dialogs. + +The parts that decide *what* to run are pure functions so they're tested +without a display, a password, or a working sudo. +""" + +from __future__ import annotations + +import os +import re +import shutil +import stat +from pathlib import Path +from typing import Callable, Optional + +from . import config + +# Real askpass binaries, in preference order. These are purpose-built for +# this (they grab the keyboard, hide the input, and don't leave the password +# in a process argument), so they win over a generated wrapper. +_KNOWN_HELPERS = ( + "/usr/bin/ssh-askpass", + "/usr/lib/ssh/ssh-askpass", + "/usr/lib/openssh/gnome-ssh-askpass3", + "/usr/lib/openssh/gnome-ssh-askpass", + "/usr/libexec/openssh/ssh-askpass", + "/usr/bin/ksshaskpass", + "/usr/bin/lxqt-openssh-askpass", +) + +# Dialog tools we can wrap when no askpass binary exists. Each must print the +# typed password to stdout and nothing else. +_WRAPPABLE = { + "zenity": '{tool} --password --title="Bolt" --text="$1" 2>/dev/null', + "kdialog": '{tool} --password "$1" --title "Bolt" 2>/dev/null', +} + +# `sudo` at the start of the command or right after a shell separator, not +# already carrying a flag. Deliberately conservative: a `sudo` inside a +# quoted string or a heredoc is left alone, because rewriting it could change +# what the command means. +_SUDO = re.compile(r"(^|[;&|]\s*|\n\s*)(sudo)(\s+)(?!-)") + + +def add_askpass_flag(command: str) -> str: + """Insert `-A` after each bare `sudo`, so it prompts through the helper + instead of a tty. Commands that already pass a flag (`sudo -n`, `sudo -A`, + `sudo -u bob`) are left exactly as they are — the caller was explicit.""" + return _SUDO.sub(r"\1\2 -A\3", command or "") + + +def needs_password_prompt(command: str) -> bool: + """True if *command* has a `sudo` that might sit waiting on a dialog. + Used to give those commands a longer timeout — 30 seconds is fine for a + shell command and nowhere near enough for a human to notice a window, + read it, and type a password.""" + return bool(_SUDO.search(command or "")) + + +def helper_script(tool_path: str) -> str: + """The wrapper script for a dialog *tool_path*. sudo passes its prompt + ("[sudo] password for maji:") as $1, which is worth showing — it names + the user the password is for.""" + name = Path(tool_path).name + body = _WRAPPABLE[name].format(tool=tool_path) + return f"#!/bin/sh\n# Generated by bolt-pet. Prints the typed password on stdout for sudo -A.\n{body}\n" + + +def _default_cache_dir() -> Path: + base = os.environ.get("XDG_CACHE_HOME") or (Path.home() / ".cache") + return Path(base) / "bolt-pet" + + +def find_helper( + configured: str = None, + is_executable: Callable[[str], bool] = None, + which: Callable[[str], Optional[str]] = None, + cache_dir: Path = None, + write: bool = True, +) -> Optional[str]: + """Path to an askpass helper, or None if the desktop has nothing we can + use. Order: whatever SUDO_ASKPASS_HELPER names, then a real askpass + binary, then a generated wrapper around zenity/kdialog. + + The lookups are injectable so the resolution order is testable on a box + with a different set of these installed than yours.""" + configured = config.SUDO_ASKPASS_HELPER if configured is None else configured + is_executable = is_executable or (lambda path: os.path.isfile(path) and os.access(path, os.X_OK)) + which = which or shutil.which + + if configured: + return configured if is_executable(configured) else None + + for candidate in _KNOWN_HELPERS: + if is_executable(candidate): + return candidate + + for tool in _WRAPPABLE: + tool_path = which(tool) + if not tool_path: + continue + if not write: + return tool_path + return _write_wrapper(tool_path, cache_dir or _default_cache_dir()) + return None + + +def _write_wrapper(tool_path: str, cache_dir: Path) -> Optional[str]: + """Drop the wrapper script somewhere sudo can execute it. Mode 0700: it + isn't secret, but it's a thing that pops up a password box, so nobody + else on the machine gets to edit it.""" + try: + cache_dir.mkdir(parents=True, exist_ok=True) + script = cache_dir / "askpass.sh" + source = helper_script(tool_path) + if not script.exists() or script.read_text(encoding="utf-8") != source: + script.write_text(source, encoding="utf-8") + script.chmod(stat.S_IRWXU) + return str(script) + except Exception: + return None # no prompt is better than a crashed command relay + + +def environment(helper: str, base: dict = None) -> dict: + """The subprocess environment with SUDO_ASKPASS pointed at *helper*.""" + env = dict(os.environ if base is None else base) + env["SUDO_ASKPASS"] = helper + return env diff --git a/bolt_pet/updater.py b/bolt_pet/updater.py index da77bbc..b41fec7 100644 --- a/bolt_pet/updater.py +++ b/bolt_pet/updater.py @@ -183,6 +183,22 @@ def current_ref(run: GitRunner) -> str: return output.strip() +def already_at_tag(run: GitRunner, tag: str) -> bool: + """True if HEAD is already the commit *tag* points at. + + Guards the loop you get when a release is cut without bumping + __version__ in the tagged commit: the checkout succeeds (it's a no-op), + the pet restarts, reads the same old __version__, sees the same "newer" + tag, and does it again — a restart every check, forever.""" + code, head = run(["rev-parse", "HEAD"]) + if code != 0 or not head.strip(): + return False + code, target = run(["rev-parse", f"tags/{tag}^{{commit}}"]) + if code != 0 or not target.strip(): + return False # tag isn't known locally yet, so we're certainly not on it + return head.strip() == target.strip() + + def _requirements_changed(run: GitRunner, before: str, after: str) -> bool: code, output = run(["diff", "--name-only", before, after, "--", "requirements.txt"]) return code == 0 and bool(output.strip()) @@ -236,6 +252,14 @@ def apply_update( if code != 0: raise UpdateError(f"git fetch failed: {output}") + if already_at_tag(run, tag): + from . import __version__ + + raise UpdateError( + f"already checked out {tag}, but __version__ still reads {__version__} — " + f"bump it in the tagged commit, or every check re-applies the same release" + ) + code, output = run(["checkout", "--force", f"tags/{tag}"]) if code != 0: raise UpdateError(f"git checkout {tag} failed: {output}") diff --git a/tests/test_controller_features.py b/tests/test_controller_features.py index c710acb..9eff467 100644 --- a/tests/test_controller_features.py +++ b/tests/test_controller_features.py @@ -81,6 +81,44 @@ def test_petctl_nap_also_flips_the_controller_state(ctrl): # ── 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): + # 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", @@ -99,6 +137,110 @@ def test_uninterrupted_playback_does_not_queue_a_turn(monkeypatch, ctrl): 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: 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: 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: True) diff --git a/tests/test_speech_text.py b/tests/test_speech_text.py index dee7d75..53c9d5f 100644 --- a/tests/test_speech_text.py +++ b/tests/test_speech_text.py @@ -6,7 +6,7 @@ from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) -from bolt_pet.speech_text import for_display, for_speech +from bolt_pet.speech_text import for_display, for_speech, is_question def test_bold_markers_are_not_spoken(): @@ -57,3 +57,23 @@ def test_blank_and_symbol_only_input(): def test_display_keeps_emoji_but_drops_markdown(): assert for_display("**Done** ✅") == "Done ✅" assert for_display("* one\n* two") == "• one • two" + + +def test_is_question_only_fires_on_a_trailing_question(): + assert is_question("Ready to run a command or start a project?") + assert is_question("It's 7:15 AM. Want me to set a timer?") + assert not is_question("It's 7:15 AM on July 23, 2026.") + assert not is_question("What time is it? It's 7:15 AM.") # asked in passing + + +def test_is_question_ignores_trailing_decoration(): + assert is_question("Ready to go? 🚀") + assert is_question('Shall I continue?"') + assert is_question("Want me to fix it? **") + + +def test_is_question_ignores_question_marks_that_are_not_spoken(): + # The '?' here is inside a URL query string, which for_speech strips. + assert not is_question("Docs are at https://example.com/x?y=1") + assert not is_question("") + assert not is_question(None) diff --git a/tests/test_sudo_askpass.py b/tests/test_sudo_askpass.py new file mode 100644 index 0000000..cc0ed0e --- /dev/null +++ b/tests/test_sudo_askpass.py @@ -0,0 +1,122 @@ +"""Graphical sudo prompts: command rewriting and helper resolution. +Pure logic — no display, no sudo, no password.""" + +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from bolt_pet import sudo_askpass + + +# ── rewriting ──────────────────────────────────────────────────────────────── + + +@pytest.mark.parametrize( + "command,expected", + [ + ("sudo apt update", "sudo -A apt update"), + ("sudo apt update", "sudo -A apt update"), # spacing preserved + ("apt update && sudo apt upgrade", "apt update && sudo -A apt upgrade"), + ("ls; sudo reboot", "ls; sudo -A reboot"), + ("echo hi | sudo tee /etc/motd", "echo hi | sudo -A tee /etc/motd"), + ("sudo systemctl restart x\nsudo systemctl status x", + "sudo -A systemctl restart x\nsudo -A systemctl status x"), + ], +) +def test_bare_sudo_gets_the_askpass_flag(command, expected): + assert sudo_askpass.add_askpass_flag(command) == expected + + +@pytest.mark.parametrize( + "command", + [ + "sudo -n apt update", # explicitly non-interactive + "sudo -A apt update", # already asking + "sudo -u bob whoami", # the caller was explicit + "ls -la", # no sudo at all + "echo 'run sudo later'", # inside a quoted string + "pseudo --version", # not the word sudo + ], +) +def test_commands_that_must_not_be_rewritten(command): + assert sudo_askpass.add_askpass_flag(command) == command + + +def test_blank_input(): + assert sudo_askpass.add_askpass_flag("") == "" + assert sudo_askpass.add_askpass_flag(None) == "" + + +@pytest.mark.parametrize( + "command,expected", + [ + ("sudo apt update", True), + ("ls && sudo reboot", True), + ("ls -la", False), + ("echo 'sudo'", False), + ("", False), + ], +) +def test_which_commands_get_the_longer_timeout(command, expected): + assert sudo_askpass.needs_password_prompt(command) is expected + + +# ── helper resolution ──────────────────────────────────────────────────────── + + +def test_a_configured_helper_wins(): + found = sudo_askpass.find_helper( + configured="/opt/my-askpass", is_executable=lambda p: p == "/opt/my-askpass", + ) + assert found == "/opt/my-askpass" + + +def test_a_configured_helper_that_is_not_executable_is_not_silently_replaced(): + """Better to have sudo fail than to quietly prompt with something the + user didn't choose.""" + assert sudo_askpass.find_helper( + configured="/opt/typo", is_executable=lambda p: False, which=lambda t: "/usr/bin/zenity", + ) is None + + +def test_a_real_askpass_binary_beats_a_generated_wrapper(): + found = sudo_askpass.find_helper( + configured="", is_executable=lambda p: p == "/usr/bin/ksshaskpass", + which=lambda tool: "/usr/bin/zenity", + ) + assert found == "/usr/bin/ksshaskpass" + + +def test_falls_back_to_wrapping_a_dialog_tool(tmp_path): + found = sudo_askpass.find_helper( + configured="", is_executable=lambda p: False, + which=lambda tool: "/usr/bin/zenity" if tool == "zenity" else None, + cache_dir=tmp_path, + ) + script = tmp_path / "askpass.sh" + assert found == str(script) + assert "zenity --password" in script.read_text() + assert script.stat().st_mode & 0o777 == 0o700 # nobody else edits the password box + + +def test_no_helper_available_at_all(): + assert sudo_askpass.find_helper( + configured="", is_executable=lambda p: False, which=lambda tool: None, + ) is None + + +def test_the_wrapper_passes_sudos_prompt_through(): + """sudo hands the helper its prompt as $1 — it names the account the + password is for, which is worth showing in the dialog.""" + script = sudo_askpass.helper_script("/usr/bin/zenity") + assert script.startswith("#!/bin/sh") + assert '"$1"' in script + + +def test_environment_points_sudo_at_the_helper(): + env = sudo_askpass.environment("/tmp/askpass.sh", base={"PATH": "/usr/bin"}) + assert env["SUDO_ASKPASS"] == "/tmp/askpass.sh" + assert env["PATH"] == "/usr/bin" # the rest of the environment survives diff --git a/tests/test_updater.py b/tests/test_updater.py index b7ac47a..d006622 100644 --- a/tests/test_updater.py +++ b/tests/test_updater.py @@ -67,12 +67,15 @@ class FakeGit: *failures* maps a leading-args tuple to the (code, output) it should return, so a test can make exactly one command fail.""" - def __init__(self, ref="main", dirty=False, failures=None, requirements_changed=False): + def __init__(self, ref="main", dirty=False, failures=None, requirements_changed=False, + head="abc1234", tag_commit="def5678"): self.calls = [] self._ref = ref self._dirty = dirty self._failures = failures or {} self._requirements_changed = requirements_changed + self._head = head + self._tag_commit = tag_commit # equal to head = "already on this tag" def __call__(self, args): self.calls.append(list(args)) @@ -87,7 +90,7 @@ class FakeGit: if head == "symbolic-ref": return (0, self._ref) if self._ref else (1, "") if head == "rev-parse": - return 0, "abc1234" + return 0, self._tag_commit if args[1].startswith("tags/") else self._head if head == "diff": return 0, "requirements.txt" if self._requirements_changed else "" return 0, "" @@ -160,6 +163,26 @@ def test_a_verify_that_raises_something_unexpected_still_rolls_back(tmp_path): assert ["checkout", "--force", "main"] in git.calls +def test_a_release_tagged_without_bumping_the_version_does_not_loop(tmp_path): + """Cut a release but forget to bump __version__ in the tagged commit and + every check would see the same "newer" tag: check out (a no-op), restart, + read the old version, repeat — a restart loop every check interval.""" + git = FakeGit(head="same1234", tag_commit="same1234") + + with pytest.raises(updater.UpdateError, match="bump it in the tagged commit"): + updater.apply_update("v0.2.1", run=git, repo=tmp_path, install_deps=False, + verify=lambda repo: None) + + assert "checkout" not in git.commands() # nothing moved, so nothing to restart into + + +def test_an_unknown_tag_is_not_mistaken_for_being_already_on_it(tmp_path): + git = FakeGit(failures={("rev-parse", "tags/"): (128, "unknown revision")}) + # The failure key matches by prefix, so make it explicit that a tag we + # can't resolve means "not there yet" rather than "already applied". + assert updater.already_at_tag(git, "v9.9.9") is False + + def test_a_failed_fetch_never_moves_the_checkout(tmp_path): git = FakeGit(failures={("fetch",): (1, "could not resolve host")}) with pytest.raises(updater.UpdateError, match="git fetch failed"):