From 8bcd539b3107fb090b4ad7cedc07e4fa6c220a4f Mon Sep 17 00:00:00 2001 From: TheMajesticNetwork Date: Sat, 26 Sep 2026 22:59:33 -0600 Subject: [PATCH] Bolt Assistant: wake-phrase voice assistant for Bolt --- .env.example | 28 +++ .gitignore | 5 + README.md | 88 +++++++++ bolt_assistant/__init__.py | 3 + bolt_assistant/__main__.py | 117 +++++++++++ bolt_assistant/assistant.py | 259 ++++++++++++++++++++++++ bolt_assistant/audio.py | 61 ++++++ bolt_assistant/config.py | 110 +++++++++++ bolt_assistant/file_ops.py | 287 +++++++++++++++++++++++++++ bolt_assistant/listener.py | 116 +++++++++++ bolt_assistant/relay_json.py | 157 +++++++++++++++ bolt_assistant/server.py | 186 ++++++++++++++++++ bolt_assistant/speech_text.py | 173 ++++++++++++++++ bolt_assistant/stt.py | 152 ++++++++++++++ bolt_assistant/tools.py | 93 +++++++++ bolt_assistant/wake.py | 50 +++++ requirements.txt | 4 + run.bat | 13 ++ run.sh | 11 ++ tests/__init__.py | 0 tests/test_assistant.py | 359 ++++++++++++++++++++++++++++++++++ tests/test_file_ops.py | 343 ++++++++++++++++++++++++++++++++ tests/test_speech_text.py | 111 +++++++++++ 23 files changed, 2726 insertions(+) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 README.md create mode 100644 bolt_assistant/__init__.py create mode 100644 bolt_assistant/__main__.py create mode 100644 bolt_assistant/assistant.py create mode 100644 bolt_assistant/audio.py create mode 100644 bolt_assistant/config.py create mode 100644 bolt_assistant/file_ops.py create mode 100644 bolt_assistant/listener.py create mode 100644 bolt_assistant/relay_json.py create mode 100644 bolt_assistant/server.py create mode 100644 bolt_assistant/speech_text.py create mode 100644 bolt_assistant/stt.py create mode 100644 bolt_assistant/tools.py create mode 100644 bolt_assistant/wake.py create mode 100644 requirements.txt create mode 100644 run.bat create mode 100644 run.sh create mode 100644 tests/__init__.py create mode 100644 tests/test_assistant.py create mode 100644 tests/test_file_ops.py create mode 100644 tests/test_speech_text.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..575d47f --- /dev/null +++ b/.env.example @@ -0,0 +1,28 @@ +# Bolt Assistant. Copy to .env and fill in the first three. +BOLT_SERVER_URL=https://your-bolt-server.example.com +DESK_API_KEY= +# The voice replies are spoken in (an ElevenLabs voice id the server can use) +ELEVENLABS_VOICE_ID= + +# Microphone / speaker: blank = system default, or an index / name from --devices +#MIC_DEVICE= +#SPEAKER_DEVICE= + +# Wake phrases, comma-separated (matched on the transcript, at the start of what's said) +#WAKE_WORDS=hey bolt,bolt,thunderbolt,hey thunderbolt,ok bolt,okay bolt,hey assistant + +# What Bolt may do on this machine when he relays a command: +# off | read (default: list and read files) | all (shell commands and file changes too) +#ASSISTANT_COMMANDS=read + +# Local speech detection: nothing is sent to the server until someone seems to be talking. +# Raise the threshold if background noise keeps getting transcribed. +#VAD_RMS_THRESHOLD=500 +#VAD_SILENCE_END_SEC=0.8 +#MIN_UTTERANCE_SEC=0.35 +#MAX_UTTERANCE_SEC=12 +#COMMAND_WAIT_SEC=8 +#FOLLOW_UP_SEC=8 +#CHIMES=true +#STREAMING_REPLIES=true +#DESK_SESSION_ID= diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..f6664e3 --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +.venv/ +__pycache__/ +*.pyc +.env +.pytest_cache/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..6a9941d --- /dev/null +++ b/README.md @@ -0,0 +1,88 @@ +# Bolt Assistant + +A voice assistant for Bolt. Say **"Hey Bolt"** (or "Bolt", "Thunderbolt", "Hey Assistant" and a few +others), ask something, and Bolt answers through your speakers. It uses the same server, memory and +persona as Bolt-Pet, Bolt Console and the phone app. + +``` +Listening for "Hey Bolt" (or "Thunderbolt", "Hey Assistant"). Ctrl+C to quit. +> what time is it +< It's quarter past three. +* Yes? +> what's in my downloads folder + file list: C:\Users\aaron\Downloads +< You've got three installers and a PDF called taxes. +``` + +## Setup + +```bash +cp .env.example .env # BOLT_SERVER_URL, DESK_API_KEY, ELEVENLABS_VOICE_ID +run.bat # Windows (./run.sh on Linux/macOS); creates .venv on first run +``` + +`run.bat --check` tests the settings, server, microphone, speaker and speech-to-text. +`run.bat --devices` lists audio devices. Set `MIC_DEVICE` / `SPEAKER_DEVICE` to use something +other than the system default. `run.bat --type` lets you type instead of speak, and +`run.bat --say "hello"` speaks one line. + +## How it works + +1. **Listening happens on your machine first.** A voice-activity detector watches the mic's loudness. + Silence, hum and clicks are never sent anywhere. When it hears something that sounds like + talking, that stretch of audio is sent to your server's `/desk/stt` relay (Deepgram, as in + Bolt-Pet) and transcribed as you speak. +2. **Only transcripts that start with a wake phrase count.** "Hey Bolt, what time is it" asks "what + time is it". "I need to tighten the bolt" is ignored, because the phrase has to open what you say, + with at most a filler word such as "um" or "okay" before it. Saying just "Hey Bolt" plays a chime, + and whatever you say in the next 8 seconds is the request. +3. **The answer is spoken** through the server's `/desk/tts`, one sentence at a time as Bolt + produces them. If the answer ends in a question, the assistant keeps listening for your reply + without needing the wake phrase again. +4. **While Bolt speaks, the mic is ignored**, so his own voice through the speakers (which often + says "Bolt") can't wake him. + +## Cost and privacy + +The server's speech-to-text is what recognises the wake phrase, so **everything said near the mic +that sounds like speech is sent to your server and transcribed**, not just things said to Bolt. The +server counts this against your credits like a chat turn. The local detector keeps that to actual +speech: silence is never sent, and bursts shorter than `MIN_UTTERANCE_SEC` are dropped. In a room +with a TV or a lot of conversation, it will still transcribe a lot. + +If that becomes a problem, the alternative is on-device wake-word detection, which is what Bolt-Pet +does with openWakeWord and its `thunderbolt.onnx` model. Nothing leaves the machine until the wake +word is heard, but it only recognises the phrases it was trained on. + +## What Bolt may do on this machine + +There's no screen to approve things on, so `ASSISTANT_COMMANDS` decides: + +| Setting | What Bolt can do | +|---|---| +| `off` | Nothing on this computer | +| `read` (default) | List folders and read files ("what's in my Downloads?") | +| `all` | Everything, including shell commands and file changes, like Bolt-Pet. Anyone within earshot can ask for anything. | + +Anything that isn't allowed is refused, and Bolt is told to say he can't do it here. + +## Settings + +See `.env.example`. The ones you're most likely to change: + +- `VAD_RMS_THRESHOLD`: raise it if background noise keeps getting transcribed, lower it if you + have to raise your voice. +- `WAKE_WORDS`: the phrases, comma-separated. +- `COMMAND_WAIT_SEC`, `FOLLOW_UP_SEC`: how long it waits after "Hey Bolt", or after asking you + something. + +It has its own conversation on the server (`pet-assistant-`), which shows up in Bolt +Console's Chats list. + +## Tests + +```bash +.venv\Scripts\python -m pytest tests # Windows; .venv/bin/python on Linux +``` + +No mic, speaker or server needed. The tests use synthetic audio and fakes for the network. diff --git a/bolt_assistant/__init__.py b/bolt_assistant/__init__.py new file mode 100644 index 0000000..205e5f8 --- /dev/null +++ b/bolt_assistant/__init__.py @@ -0,0 +1,3 @@ +"""Bolt Assistant: a voice assistant for Bolt (wake phrase, server speech-to-text, spoken replies).""" + +__version__ = "0.1.0" diff --git a/bolt_assistant/__main__.py b/bolt_assistant/__main__.py new file mode 100644 index 0000000..38e6120 --- /dev/null +++ b/bolt_assistant/__main__.py @@ -0,0 +1,117 @@ +"""Bolt Assistant: say "Hey Bolt", ask, hear the answer. + + python -m bolt_assistant listen on the default mic + python -m bolt_assistant --verbose also show what was heard but ignored + python -m bolt_assistant --type type requests instead of speaking (testing) + python -m bolt_assistant --say "hello" speak a line through the speaker and exit + python -m bolt_assistant --devices list microphones and speakers + python -m bolt_assistant --check check settings, server, mic and voice +""" + +from __future__ import annotations + +import sys +import threading + +from . import __version__, audio, config, server + + +def _devices() -> int: + import sounddevice as sd + + print(sd.query_devices()) + print("\nDefault input/output:", sd.default.device, + "\nSet MIC_DEVICE / SPEAKER_DEVICE in .env to an index or a name to choose another.") + return 0 + + +def _check() -> int: + from .stt import connect + + ok = True + + def row(name, good, detail): + nonlocal ok + ok = ok and good + print(f"[{' ok ' if good else 'FAIL'}] {name:12} {detail}") + + missing = config.missing_config() + row("settings", not missing, "all set" if not missing else "missing " + ", ".join(missing)) + try: + server.check_health() + row("server", True, config.SERVER_URL) + except Exception as exc: + row("server", False, f"{config.SERVER_URL}: {exc}") + try: + import sounddevice as sd + + mic = sd.query_devices(config.MIC_DEVICE, "input")["name"] + out = sd.query_devices(config.SPEAKER_DEVICE, "output")["name"] + row("microphone", True, mic) + row("speaker", True, out) + except Exception as exc: + row("audio", False, f"{exc} (pip install sounddevice; Linux: libportaudio2)") + try: + connect().close() + row("speech-to-text", True, "/desk/stt reachable") + except Exception as exc: + row("speech-to-text", False, f"{exc} (pip install websocket-client)") + row("commands", True, f"ASSISTANT_COMMANDS={config.COMMANDS}") + return 0 if ok else 1 + + +def main(argv=None) -> int: + argv = list(sys.argv[1:] if argv is None else argv) + for stream in (sys.stdout, sys.stderr): + try: # replies carry dashes and curly quotes; a cp1252 console must not crash on them + stream.reconfigure(errors="replace") + except Exception: + pass + if "-h" in argv or "--help" in argv: + print(__doc__.strip()) + return 0 + if "--devices" in argv: + return _devices() + if "--check" in argv: + return _check() + missing = config.missing_config() + if missing: + print(f"Missing {', '.join(missing)}. Copy .env.example to .env and fill them in.") + return 2 + + from .assistant import Assistant + + assistant = Assistant(verbose="--verbose" in argv) + print(f"Bolt Assistant {__version__} | {config.SERVER_URL} | session {config.SESSION_ID}") + + if "--say" in argv: + assistant.speak(" ".join(argv[argv.index("--say") + 1:]) or "Hello, I'm Bolt.") + return 0 + + if "--type" in argv: # the whole pipeline minus the microphone + print("Type what you'd say (including the wake phrase). Ctrl+C to quit.") + try: + while True: + assistant.heard(input("say> ")) + except (EOFError, KeyboardInterrupt): + return 0 + + stop = threading.Event() + try: + mic = audio.open_mic(assistant.on_mic_frame) + except Exception as exc: + print(f"Couldn't open the microphone: {exc}\nTry --devices, and MIC_DEVICE in .env.") + return 1 + try: + assistant.run(stop) + except KeyboardInterrupt: + pass + finally: + stop.set() + mic.stop() + mic.close() + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/bolt_assistant/assistant.py b/bolt_assistant/assistant.py new file mode 100644 index 0000000..954aaf5 --- /dev/null +++ b/bolt_assistant/assistant.py @@ -0,0 +1,259 @@ +"""The assistant: hear the wake phrase, take the request, answer out loud. + + idle every utterance the local detector passes on is transcribed by + the server; only ones that open with a wake phrase count + "Hey Bolt, what's the time" -> ask "what's the time" + "Hey Bolt" -> chime, go to `command` + command the next thing said (within COMMAND_WAIT_SEC) is the request + follow_up the reply ended on a question: the next thing said (within + FOLLOW_UP_SEC) is the answer, no wake phrase needed + +Speaking is half-duplex: the mic is ignored while Bolt talks and for a +moment after, so his own voice through the speakers can't wake him (his +replies say "Bolt" often enough). + +Everything that touches hardware or the network is injected, so the flow +is tested with fakes; __main__ wires in the real mic, server and speaker. +""" + +from __future__ import annotations + +import platform +import queue +import threading +import time +from typing import Callable, Optional + +from . import audio, config, server, speech_text +from .listener import Segmenter +from .stt import SttError, Transcription +from .tools import Tools +from .wake import find_wake + +ECHO_GUARD_SEC = 0.4 # ignore the mic this long after speaking + + +def os_label() -> str: + try: + return f"{platform.system()} {platform.release()}".strip() or "a computer" + except Exception: + return "a computer" + + +def annotate(text: str, commands: str) -> str: + """The note on every request: it's voice, and what Bolt may do here.""" + can = { + "off": "You can't act on this computer from here.", + "read": ('You can list and read files with your command tool, e.g. ' + '`command: filectl {"op": "list", "path": "..."}` (ops: list, read), and the real ' + "output comes back. You can't run shell commands or change files from here."), + "all": ('You can act on this computer with your command tool: `command: filectl {"op": ...}` ' + "for files (list, read, write, edit) or `command: `; the real " + "output comes back."), + }[commands] + return (f"{text}\n\n[Bolt Assistant: a voice-only assistant on {os_label()}. The request was " + "spoken and transcribed, so it may contain recognition mistakes. Your reply is read " + "aloud: keep it short and conversational, with no markdown, lists, links or code. " + f"{can} Never make up a command's output.]") + + +class Assistant: + def __init__( + self, + *, + transcription: Callable = Transcription, + converse: Callable = server.converse, + converse_stream: Optional[Callable] = server.converse_stream, + tts: Callable = server.tts_stream, + play: Callable = audio.play, + tools: Optional[Tools] = None, + log: Callable[[str], None] = print, + clock: Callable[[], float] = time.monotonic, + wake_words=None, + verbose: bool = False, + ): + self._transcription = transcription + self._converse = converse + self._stream = converse_stream if config.STREAMING_REPLIES else None + self._tts, self._play = tts, play + self.log, self.clock, self.verbose = log, clock, verbose + self.tools = tools or Tools(log=log) + self.wake_words = wake_words or config.WAKE_WORDS + self.segmenter = Segmenter(config.VAD_RMS_THRESHOLD, config.VAD_START_FRAMES, + config.VAD_SILENCE_END_SEC, config.MIN_UTTERANCE_SEC, + config.MAX_UTTERANCE_SEC) + self.frames: "queue.Queue[bytes]" = queue.Queue(maxsize=2000) # ~160 s of audio + self.state = "idle" + self._deadline = 0.0 + self._follow_ups = 0 + self._stt = None + self._mute_until = 0.0 + self._voice = "" # a voice the server picked (speak_as) sticks for later replies + self._tts_failed = False + + # ── audio in ───────────────────────────────────────────────────────── + + def on_mic_frame(self, frame: bytes) -> None: + """Called from the audio thread: queue it, never block.""" + try: + self.frames.put_nowait(frame) + except queue.Full: + pass # we've fallen far behind; dropping audio beats unbounded memory + + def run(self, stop: threading.Event) -> None: + self.log(self.listening_banner()) + while not stop.is_set(): + try: + frame = self.frames.get(timeout=0.2) + except queue.Empty: + self._check_deadline() + continue + self.process(frame) + + def listening_banner(self) -> str: + return 'Listening for "Hey Bolt" (or "Thunderbolt", "Hey Assistant"). Ctrl+C to quit.' + + def process(self, frame: bytes) -> None: + if self.clock() < self._mute_until: + return + self._check_deadline() + for event in self.segmenter.feed(frame): + if event.kind == "start": + self._stt = self._transcription() + self._stt.add(event.frames) + elif event.kind == "frame" and self._stt is not None: + self._stt.add(event.frames) + elif event.kind == "dropped": + if self._stt is not None: + self._stt.cancel() + self._stt = None + elif event.kind == "end": + stt, self._stt = self._stt, None + try: + text = stt.finish(event.audio) if stt is not None else "" + except SttError as exc: + self.log(f"! transcription failed: {exc}") + if self.state != "idle": + self._chime(audio.ERROR_CHIME) + continue + self.heard(text) + + def _check_deadline(self) -> None: + if self.state in ("command", "follow_up") and self.clock() > self._deadline \ + and not self.segmenter.in_speech: + if self.verbose: + self.log(" (stopped waiting)") + self.state = "idle" + + # ── what was said ──────────────────────────────────────────────────── + + def heard(self, text: str) -> None: + text = (text or "").strip() + if not text: + return + self._check_deadline() + woke, rest = find_wake(text, self.wake_words) + if self.state == "idle": + if not woke: + if self.verbose: + self.log(f" (not for me) {text}") + return + if not rest: + self.log("* Yes?") + self._chime(audio.WAKE_CHIME) + self._wait("command", config.COMMAND_WAIT_SEC) + return + self._follow_ups = 0 + self.ask(rest) + return + # command / follow_up: whatever was said is the request. A repeated + # "Hey Bolt" in front of it is dropped; on its own it just re-arms. + if woke and not rest: + self._chime(audio.WAKE_CHIME) + self._wait(self.state, config.COMMAND_WAIT_SEC) + return + if self.state == "command": + self._follow_ups = 0 + self.ask(rest if woke else text) + + def _wait(self, state: str, seconds: float) -> None: + self.state = state + self._deadline = self.clock() + seconds + + # ── asking and answering ───────────────────────────────────────────── + + def ask(self, request: str) -> None: + self.state = "busy" + self.log(f"> {request}") + spoken = [] + + def say(sentence: str) -> None: + spoken.append(sentence) + self.speak(sentence) + + message = annotate(request, self.tools.mode) + try: + reply = None + if self._stream is not None: + try: + reply = self._stream(message, on_say=say, on_command=self.tools) + except server.ServerError: + reply = None # nothing was said yet: the plain path is invisible + if reply is None: + reply = self._converse(message, on_command=self.tools, on_say=say) + except server.ServerError as exc: + self.log(f"! {exc}") + self._chime(audio.ERROR_CHIME) + self._back_to_listening("idle") + return + if reply.voice_id: + self._voice = reply.voice_id + if not reply.streamed and reply.text.strip(): + spoken.append(reply.text) + self.speak(reply.text) + full = " ".join(spoken) + if speech_text.is_question(full) and self._follow_ups < config.FOLLOW_UP_MAX_TURNS: + self._follow_ups += 1 + self._back_to_listening("follow_up", config.FOLLOW_UP_SEC) + else: + self._follow_ups = 0 + self._back_to_listening("idle") + + def speak(self, text: str) -> None: + words = speech_text.for_speech(text) + if not words: + return + self.log(f"< {speech_text.for_display(text)}") + try: + self._play(self._tts(words, self._voice)) + self._tts_failed = False + except server.TtsError as exc: + if not self._tts_failed: # say it once, not once per sentence + self.log(f"! can't speak: {exc}") + self._tts_failed = True + except Exception as exc: + self.log(f"! playback failed: {exc}") + + def _chime(self, sound: bytes) -> None: + if not config.CHIMES: + return + try: + self._play([sound]) + except Exception: + pass + self._mute_until = self.clock() + ECHO_GUARD_SEC + + def _back_to_listening(self, state: str, seconds: float = 0.0) -> None: + """Drop what the mic picked up while Bolt was talking (his own voice), + start detection fresh, and give the room a moment to go quiet.""" + while True: + try: + self.frames.get_nowait() + except queue.Empty: + break + self.segmenter.reset() + if self._stt is not None: + self._stt.cancel() + self._stt = None + self._mute_until = self.clock() + ECHO_GUARD_SEC + self._wait(state, seconds + ECHO_GUARD_SEC) diff --git a/bolt_assistant/audio.py b/bolt_assistant/audio.py new file mode 100644 index 0000000..ea9b813 --- /dev/null +++ b/bolt_assistant/audio.py @@ -0,0 +1,61 @@ +"""The default microphone in, the default speaker out, and two short chimes. + +sounddevice (PortAudio) for both directions; MIC_DEVICE / SPEAKER_DEVICE in +.env pick something other than the system default. The chimes are made +here from sine waves — no sound files to ship. +""" + +from __future__ import annotations + +import array +import math +from typing import Callable, Iterable + +from . import config + + +def open_mic(on_frame: Callable[[bytes], None]): + """Start the microphone: 16 kHz mono int16, one 80 ms frame per call to + *on_frame*, from PortAudio's thread. Returns the stream (stop()/close()).""" + import sounddevice as sd + + stream = sd.RawInputStream( + samplerate=config.SAMPLE_RATE, channels=1, dtype="int16", + blocksize=config.FRAME_SAMPLES, device=config.MIC_DEVICE, + callback=lambda data, _frames, _time, _status: on_frame(bytes(data)), + ) + stream.start() + return stream + + +def play(chunks: Iterable[bytes], sample_rate: int = config.TTS_SAMPLE_RATE, + should_stop: Callable[[], bool] = lambda: False) -> bool: + """Play PCM16 mono chunks as they arrive. False if stopped early.""" + import sounddevice as sd + + with sd.RawOutputStream(samplerate=sample_rate, channels=1, dtype="int16", + device=config.SPEAKER_DEVICE) as out: + for chunk in chunks: + if should_stop(): + out.abort() + return False + out.write(chunk) + return True + + +def tone(notes, sample_rate: int = config.TTS_SAMPLE_RATE, volume: float = 0.18) -> bytes: + """(frequency Hz, seconds) notes as PCM16, each with a quick fade in and + out so it doesn't click.""" + samples = array.array("h") + for freq, seconds in notes: + count = int(sample_rate * seconds) + fade = max(1, int(sample_rate * 0.012)) + for i in range(count): + envelope = min(1.0, i / fade, (count - i) / fade) + samples.append(int(32767 * volume * envelope * math.sin(2 * math.pi * freq * i / sample_rate))) + return samples.tobytes() + + +# Rising two notes: "I'm listening". Falling: "didn't work". +WAKE_CHIME = tone([(880, 0.07), (1320, 0.09)]) +ERROR_CHIME = tone([(520, 0.10), (390, 0.14)]) diff --git a/bolt_assistant/config.py b/bolt_assistant/config.py new file mode 100644 index 0000000..ec588d3 --- /dev/null +++ b/bolt_assistant/config.py @@ -0,0 +1,110 @@ +"""Settings for Bolt Assistant, from a `.env` next to the project (or the +real environment, which wins). Everything else reads config from here.""" + +from __future__ import annotations + +import os +import platform +from pathlib import Path + +HERE = Path(__file__).resolve().parent.parent # project root + + +def _load_env() -> None: + env_path = HERE / ".env" + if not env_path.exists(): + return + for line in env_path.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + key, _, value = line.partition("=") + os.environ.setdefault(key.strip(), value.strip()) + + +_load_env() + + +def _flag(name: str, default: str) -> bool: + return os.environ.get(name, default).strip().lower() in ("1", "true", "yes", "on") + + +def _device(raw: str): + """An input/output device: blank = system default, digits = an index, + anything else = a name substring (sounddevice's rules).""" + raw = (raw or "").strip() + if not raw: + return None + return int(raw) if raw.lstrip("-").isdigit() else raw + + +# ── server ────────────────────────────────────────────────────────────────── + +SERVER_URL = os.environ.get("BOLT_SERVER_URL", "").rstrip("/") +API_KEY = os.environ.get("DESK_API_KEY", "") +# Its own conversation, separate from the pet's and the console's. The "pet-" +# prefix gets the desk API's voice-client behaviour (short, speakable replies). +SESSION_ID = os.environ.get("DESK_SESSION_ID") or "pet-assistant-" + (platform.node() or "desktop") +# The voice the server's /desk/tts speaks in (an ElevenLabs voice id). +VOICE_ID = os.environ.get("ELEVENLABS_VOICE_ID", "") +TTS_SAMPLE_RATE = int(os.environ.get("TTS_SAMPLE_RATE", "16000")) +STREAMING_REPLIES = _flag("STREAMING_REPLIES", "true") + +# ── audio devices ─────────────────────────────────────────────────────────── + +MIC_DEVICE = _device(os.environ.get("MIC_DEVICE", "")) +SPEAKER_DEVICE = _device(os.environ.get("SPEAKER_DEVICE", "")) + +# ── wake phrases ──────────────────────────────────────────────────────────── +# Heard through the server's speech-to-text, so these are matched against +# *text*. A phrase counts only near the start of what was said ("hey bolt, +# what's the time"), not buried in a sentence ("tighten that bolt"). Speech +# recognisers often write "Bolt" as "bolts"/"volt"/"bold"; those spellings +# only count after "hey"/"ok", where they can't be ordinary words. + +WAKE_WORDS = [w.strip().lower() for w in os.environ.get( + "WAKE_WORDS", + "hey bolt,bolt,thunderbolt,hey thunderbolt,ok bolt,okay bolt,yo bolt,hi bolt," + "hey assistant,ok assistant,okay assistant," + "hey bolts,hey volt,hey bold,hey boat,okay volt,hey thunder bolt,thunder bolt", +).split(",") if w.strip()] + +# ── listening (local voice-activity detection, before anything is sent) ──── +# Nothing leaves the machine until this says someone is talking: silence and +# background hum are never sent (or billed). Raise VAD_RMS_THRESHOLD if +# background noise keeps getting transcribed; lower it if you have to shout. + +VAD_RMS_THRESHOLD = int(os.environ.get("VAD_RMS_THRESHOLD", "500")) +VAD_START_FRAMES = int(os.environ.get("VAD_START_FRAMES", "3")) # 80 ms each +VAD_SILENCE_END_SEC = float(os.environ.get("VAD_SILENCE_END_SEC", "0.8")) +MIN_UTTERANCE_SEC = float(os.environ.get("MIN_UTTERANCE_SEC", "0.35")) # shorter: never sent +MAX_UTTERANCE_SEC = float(os.environ.get("MAX_UTTERANCE_SEC", "12")) +# After a bare "Hey Bolt", how long to wait for the actual request. +COMMAND_WAIT_SEC = float(os.environ.get("COMMAND_WAIT_SEC", "8")) +# When a reply ends on a question, keep listening for the answer without the +# wake word, for this long, at most this many times in a row. +FOLLOW_UP_SEC = float(os.environ.get("FOLLOW_UP_SEC", "8")) +FOLLOW_UP_MAX_TURNS = int(os.environ.get("FOLLOW_UP_MAX_TURNS", "5")) +CHIMES = _flag("CHIMES", "true") + +# ── what Bolt may do on this machine ──────────────────────────────────────── +# The server can relay commands back. With no screen to approve them on: +# off nothing runs; Bolt is told it can't here +# read (default) listing and reading files only, via filectl +# all everything, including shell commands and file writes — like +# Bolt-Pet. Only if you trust every request that reaches your mic. + +COMMANDS = os.environ.get("ASSISTANT_COMMANDS", "read").strip().lower() +COMMAND_TIMEOUT_SECONDS = int(os.environ.get("COMMAND_TIMEOUT_SECONDS", "60")) + +SAMPLE_RATE = 16000 +FRAME_SAMPLES = 1280 # 80 ms + + +def is_configured() -> bool: + return bool(SERVER_URL and API_KEY) + + +def missing_config() -> list: + return [name for name, value in (("BOLT_SERVER_URL", SERVER_URL), ("DESK_API_KEY", API_KEY), + ("ELEVENLABS_VOICE_ID", VOICE_ID)) if not value] diff --git a/bolt_assistant/file_ops.py b/bolt_assistant/file_ops.py new file mode 100644 index 0000000..830a30c --- /dev/null +++ b/bolt_assistant/file_ops.py @@ -0,0 +1,287 @@ +"""Local file read/edit/write, intercepted from the server-relayed command +channel before it reaches a shell (see tools.Tools.handle). Whatever text the server's +"command" tool sends is just a string this repo is free to interpret before +it ever reaches subprocess — a `filectl` pseudo-command is one such +interpretation, giving the model a way to read/write/edit files on this +machine without constructing a raw shell heredoc, where quoting, `$`, +backticks, and embedded quotes make anything beyond a one-liner failure-prone. + +Executing arbitrary commands already works today (that's exactly what +run_local_command/subprocess.run does) — filectl doesn't add that ability, +it only makes the read/write/edit slice of it reliable. It also doesn't +expand what the server can already do to this machine: a relayed shell +command could already overwrite any file the desktop user can write (see the +security notes in the README) — filectl is a safer *path* to the same +capability, not a new capability. + +Wire format: `filectl `, where is a single-line, compact JSON +object — critically, ONE LINE. The server's "command" tool marker only +captures the argument up to the next newline (see TOOL_SPECS/ +_extract_all_tool_calls in the main repo's ai/agents/default.py — "command" +is not declared multiline), so a marker-delimited multi-line payload (this +module's first design) silently got truncated at the first line no matter +how the prompt worded it. JSON sidesteps that for free: json.dumps() already +encodes embedded newlines as the two characters "\n", not an actual line +break, so arbitrarily multi-line file content still fits on the one physical +line the extractor captures. + + filectl {"op": "list", "path": "", "pattern": "", "recursive": } + filectl {"op": "read", "path": "", "start": , "end": } + filectl {"op": "write", "path": "", "content": ""} + filectl {"op": "edit", "path": "", "old": "", "new": ""} + +"pattern"/"recursive" (list) and "start"/"end" (read) are optional. `list` +exists even though a real `ls`/`dir` shell command already works, because +this repo is cross-platform (Windows/macOS/Linux) and the model shouldn't +have to guess which listing command applies on this machine — one glob-based +op covers all three. Must be invoked as the argument to the +ordinary `command` tool marker (e.g. `command: filectl {"op": "write", ...}`) +— see the pet-only paragraph in the main repo's ai/desk_api.py +_system_context for the exact instruction the model is given, including the +"keep it one line" requirement. + +Pure parsing (parse) is separated from the filesystem I/O (execute) so the +syntax is unit-testable without touching disk. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Optional + +from . import relay_json + +_PREFIXES = ("filectl", "file") + +# Keeps a runaway read/write/list from blowing up the tool_result relay (and, +# for read/list, from flooding the model's context with a giant response). +_MAX_READ_BYTES = 200_000 +_MAX_WRITE_BYTES = 2_000_000 +_MAX_LIST_ENTRIES = 500 + +HELP = ( + 'filectl {"op": "list", "path": "", "pattern": "", "recursive": }\n' + 'filectl {"op": "read", "path": "", "start": , "end": }\n' + 'filectl {"op": "write", "path": "", "content": ""}\n' + 'filectl {"op": "edit", "path": "", "old": "", "new": ""}\n' + "Must be all on one line — this rides the single-line \"command\" marker." +) + + +class FileOpError(Exception): + """Bad filectl syntax or a filesystem error — reported back to the + server as command output.""" + + +def is_file_command(command: str) -> bool: + stripped = (command or "").strip() + if not stripped: + return False + first_word = stripped.split(None, 1)[0] + return first_word.lower() in _PREFIXES + + +def parse(command: str) -> Optional[dict]: + """Parse a `filectl ` string into an action dict, or None if this + isn't a filectl command at all (caller should try the next handler, or + fall back to a real shell command). Raises FileOpError on a filectl + command that doesn't make sense.""" + if not is_file_command(command): + return None + stripped = command.strip() + _, _, rest = stripped.partition(" ") + rest = rest.strip() + if not rest or rest.lower() in ("help", "-h", "--help"): + return {"action": "help"} + + # Lenient on purpose — see relay_json. A stray quote in machine-written + # JSON should not cost a turn, but the repair is reported back so the model + # is told it sent something broken while it can still learn from it. + try: + payload, repairs = relay_json.loads(rest) + except relay_json.RelayJsonError as exc: + raise FileOpError(f"couldn't parse filectl JSON — {exc}\nusage:\n{HELP}") from exc + if not isinstance(payload, dict): + raise FileOpError(f"filectl payload must be a JSON object; usage:\n{HELP}") + + op = str(payload.get("op") or "help").lower() + # Carried on the action so execute() can tell the model what it got wrong; + # a silent repair would fix today's call and guarantee tomorrow's. + tag = {"_repairs": repairs} if repairs else {} + + if op == "help": + return {"action": "help", **tag} + + if op == "list": + path = _required_str(payload, "path") + pattern = payload.get("pattern", "*") + if not isinstance(pattern, str) or not pattern: + raise FileOpError('"pattern" must be a non-empty string') + return { + "action": "list", "path": path, + "pattern": pattern, "recursive": bool(payload.get("recursive")), **tag, + } + + if op == "read": + path = _required_str(payload, "path") + return { + "action": "read", "path": path, + "start": _line_number(payload.get("start"), "start"), + "end": _line_number(payload.get("end"), "end"), **tag, + } + + if op == "write": + path = _required_str(payload, "path") + if payload.get("content") is None: + raise FileOpError('write needs "content"') + return {"action": "write", "path": path, "content": str(payload["content"]), **tag} + + if op == "edit": + path = _required_str(payload, "path") + if payload.get("old") is None or payload.get("new") is None: + raise FileOpError('edit needs "old" and "new"') + old, new = str(payload["old"]), str(payload["new"]) + if old == new: + raise FileOpError("old and new text are identical — nothing to edit") + return {"action": "edit", "path": path, "old": old, "new": new, **tag} + + raise FileOpError(f"unknown filectl op {op!r}; usage:\n{HELP}") + + +def _required_str(payload: dict, key: str) -> str: + value = payload.get(key) + if not isinstance(value, str) or not value.strip(): + raise FileOpError(f'filectl needs a non-empty "{key}"') + return value + + +def _line_number(value, label: str) -> Optional[int]: + if value is None: + return None + if isinstance(value, bool) or not isinstance(value, int): + raise FileOpError(f"{label} must be an integer line number, got {value!r}") + return value + + +def describe(action: dict) -> str: + """A one-line summary of the action, for logs.""" + kind = action.get("action") + if kind == "help": + return "[filectl] help" + return f"[filectl] {kind} {action.get('path', '')}" + + +def execute(action: dict) -> str: + """Actually perform a parsed filectl action, returning the text to send + back to the server as the command's output. Raises FileOpError on any + filesystem problem, same as a bad-syntax parse error.""" + kind = action.get("action") + if kind == "help": + return HELP + if kind == "list": + output = _do_list(action) + elif kind == "read": + output = _do_read(action) + elif kind == "write": + output = _do_write(action) + elif kind == "edit": + output = _do_edit(action) + else: + output = "[filectl] ok" + return output + relay_json.repair_note(action.get("_repairs") or []) + + +def _resolve(path_str: str) -> Path: + return Path(path_str).expanduser() + + +def _do_list(action: dict) -> str: + path = _resolve(action["path"]) + if not path.is_dir(): + raise FileOpError(f"no such directory: {path}") + pattern = action["pattern"] + glob = path.rglob if action["recursive"] else path.glob + try: + entries = sorted(glob(pattern), key=lambda p: str(p).lower()) + except OSError as exc: + raise FileOpError(f"couldn't list {path}: {exc}") from exc + if not entries: + return f"[no entries matching {pattern!r} in {path}]" + truncated = len(entries) > _MAX_LIST_ENTRIES + lines = [] + for entry in entries[:_MAX_LIST_ENTRIES]: + rel = entry.relative_to(path) + if entry.is_dir(): + lines.append(f"{rel}/") + continue + try: + size = entry.stat().st_size + except OSError: + size = -1 + lines.append(f"{rel}\t{size}B") + if truncated: + lines.append(f"... truncated at {_MAX_LIST_ENTRIES} entries (of {len(entries)}) — narrow \"pattern\"") + return "\n".join(lines) + + +def _do_read(action: dict) -> str: + path = _resolve(action["path"]) + if not path.is_file(): + raise FileOpError(f"no such file: {path}") + try: + data = path.read_bytes() + except OSError as exc: + raise FileOpError(f"couldn't read {path}: {exc}") from exc + if len(data) > _MAX_READ_BYTES: + raise FileOpError( + f"{path} is {len(data)} bytes, over the {_MAX_READ_BYTES}-byte filectl read limit — " + 'pass "start"/"end" to read a slice instead' + ) + try: + text = data.decode("utf-8") + except UnicodeDecodeError as exc: + raise FileOpError(f"{path} isn't valid UTF-8 text: {exc}") from exc + lines = text.splitlines() + start = max(1, action.get("start") or 1) + end = min(len(lines), action.get("end") or len(lines)) + if not lines: + return "[empty file]" + return "\n".join(f"{i:>6}\t{lines[i - 1]}" for i in range(start, end + 1)) + + +def _do_write(action: dict) -> str: + path = _resolve(action["path"]) + content = action["content"] + if len(content.encode("utf-8")) > _MAX_WRITE_BYTES: + raise FileOpError(f"content is over the {_MAX_WRITE_BYTES}-byte filectl write limit") + try: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content, encoding="utf-8") + except OSError as exc: + raise FileOpError(f"couldn't write {path}: {exc}") from exc + return f"[filectl] wrote {len(content)} chars to {path}" + + +def _do_edit(action: dict) -> str: + path = _resolve(action["path"]) + old, new = action["old"], action["new"] + if not path.is_file(): + raise FileOpError(f"no such file: {path}") + try: + text = path.read_text(encoding="utf-8") + except OSError as exc: + raise FileOpError(f"couldn't read {path}: {exc}") from exc + count = text.count(old) + if count == 0: + raise FileOpError(f"didn't find that text in {path} — nothing changed") + if count > 1: + raise FileOpError( + f"that text appears {count} times in {path} — filectl edit needs a unique match; " + "include more surrounding context" + ) + try: + path.write_text(text.replace(old, new, 1), encoding="utf-8") + except OSError as exc: + raise FileOpError(f"couldn't write {path}: {exc}") from exc + return f"[filectl] edited {path}" diff --git a/bolt_assistant/listener.py b/bolt_assistant/listener.py new file mode 100644 index 0000000..9da9afb --- /dev/null +++ b/bolt_assistant/listener.py @@ -0,0 +1,116 @@ +"""Local voice-activity detection: deciding what's worth sending to the server. + +The server's speech-to-text is what recognises "Hey Bolt", which means audio +has to reach it — but not all of it. `Segmenter` watches loudness frame by +frame and only opens an utterance when someone seems to be talking: + + VAD_START_FRAMES loud frames in a row -> speech started + VAD_SILENCE_END_SEC of quiet -> speech ended + MAX_UTTERANCE_SEC -> cut off regardless + shorter than MIN_UTTERANCE_SEC -> dropped, never sent + +Silence, hum and single clicks never leave the machine (or cost anything). +A short pre-roll is kept so the first syllable — the "H" of "Hey" — isn't +clipped by the time the detector is sure. + +Pure: frames in, events out. No audio devices, no threads, no network. +""" + +from __future__ import annotations + +import array +import math +from collections import deque +from typing import List, Optional + +PRE_ROLL_FRAMES = 4 # 320 ms kept from before speech was detected + + +def rms(frame: bytes) -> float: + samples = array.array("h") + samples.frombytes(frame[: len(frame) - len(frame) % 2]) + return math.sqrt(sum(s * s for s in samples) / len(samples)) if samples else 0.0 + + +class Event: + """What feed() reports. kind: "start" (frames = pre-roll + the loud + frames that triggered it), "frame" (one more frame of an utterance in + progress), "end" (audio = the whole utterance), "dropped" (too short + to be a wake phrase; nothing was sent).""" + + __slots__ = ("kind", "frames", "audio", "seconds") + + def __init__(self, kind: str, frames: Optional[List[bytes]] = None, + audio: bytes = b"", seconds: float = 0.0): + self.kind, self.frames, self.audio, self.seconds = kind, frames or [], audio, seconds + + def __repr__(self) -> str: + return f"Event({self.kind!r}, {self.seconds:.2f}s)" + + +class Segmenter: + def __init__(self, threshold: float, start_frames: int, silence_end_sec: float, + min_sec: float, max_sec: float, frame_sec: float = 0.08): + self.threshold = threshold + self.start_frames = max(1, start_frames) + self.silence_frames = max(1, round(silence_end_sec / frame_sec)) + self.min_frames = round(min_sec / frame_sec) + self.max_frames = max(1, round(max_sec / frame_sec)) + self.frame_sec = frame_sec + self.reset() + + def reset(self) -> None: + """Forget everything — e.g. after the assistant spoke, so its own + voice through the speakers isn't taken for the start of a request.""" + self._pre = deque(maxlen=PRE_ROLL_FRAMES) + self._loud_run: List[bytes] = [] + self._utterance: Optional[List[bytes]] = None + self._quiet = 0 + self._voiced = 0 + + @property + def in_speech(self) -> bool: + return self._utterance is not None + + def feed(self, frame: bytes) -> List[Event]: + loud = rms(frame) >= self.threshold + if self._utterance is None: + return self._idle(frame, loud) + return self._speaking(frame, loud) + + def _idle(self, frame: bytes, loud: bool) -> List[Event]: + if not loud: + if self._loud_run: # a click or a cough, not speech + self._pre.extend(self._loud_run) + self._loud_run = [] + self._pre.append(frame) + return [] + self._loud_run.append(frame) + if len(self._loud_run) < self.start_frames: + return [] + frames = list(self._pre) + self._loud_run + self._utterance = list(frames) + self._voiced, self._quiet = len(self._loud_run), 0 + self._pre.clear() + self._loud_run = [] + return [Event("start", frames=frames)] + + def _speaking(self, frame: bytes, loud: bool) -> List[Event]: + self._utterance.append(frame) + if loud: + self._voiced += 1 + self._quiet = 0 + else: + self._quiet += 1 + events = [Event("frame", frames=[frame])] + if self._quiet >= self.silence_frames or len(self._utterance) >= self.max_frames: + events.append(self._close()) + return events + + def _close(self) -> Event: + utterance, voiced = self._utterance, self._voiced + self._utterance = None + seconds = len(utterance) * self.frame_sec + if voiced < self.min_frames: + return Event("dropped", seconds=seconds) + return Event("end", audio=b"".join(utterance), seconds=seconds) diff --git a/bolt_assistant/relay_json.py b/bolt_assistant/relay_json.py new file mode 100644 index 0000000..31c6ce4 --- /dev/null +++ b/bolt_assistant/relay_json.py @@ -0,0 +1,157 @@ +"""Lenient JSON for the relayed command channel — and honest about it. + +`filectl` and `dialoguectl` both take a single line of compact JSON, hand-typed +by a language model into a tool marker. Models get that *nearly* right and then +get it wrong in a small, boringly repeatable set of ways: + + {"op":"list","path":"/home/x","recursive":false"} ← stray quote after a literal + {"op": "read", "path": "/tmp/a.txt",} ← trailing comma + {'op': 'read', 'path': '/tmp/a.txt'} ← single quotes + {“op”: “read”, “path”: “/tmp/a.txt”} ← smart quotes + {"op": "list", "recursive": False} ← Python literals + ```json {"op": "list"} ``` ← fenced + +Observed live (2026-07-30): a stray quote after `false` cost an entire desk +turn — the call was rejected, the model re-sent the *identical* line, was +rejected again, and then gave up and told the user "I'll check now" without +ever calling anything. The user got a promise instead of an answer because of +one character. + +Strict parsing is the wrong trade here. Nothing about a misplaced quote is +ambiguous, the payload is machine-written and machine-read, and the cost of +refusing is a wasted round trip that the model has already demonstrated it +won't recover from. So: try strict first, then apply narrow repairs, and +accept a repair **only if the result parses**. + +Two rules keep this from becoming "guess what they meant": + +1. **Repairs are conservative and named.** Each one fixes a known malformation, + is applied in isolation, and is reported by name. +2. **Repairs are never silent.** The caller appends the repair note to the tool + result, so the model is told it sent broken JSON *while it still has the + turn* — the fix works today and teaches within the conversation. Hiding it + would trade a visible failure for an invisible one. + +When nothing parses, the error points at the exact character with a caret, +because "Expecting ',' delimiter: char 74" is not something a model can act on +and a pointed-at fragment is. +""" + +from __future__ import annotations + +import json +import re +from typing import Any, Callable + +# Ordered, cheapest and safest first. Each entry is (name, transform); after +# each one the payload is re-parsed, so the first repair that works wins and +# nothing more aggressive gets applied than the input actually needed. +_REPAIRS: tuple[tuple[str, Callable[[str], str]], ...] = ( + ( + "stripped a markdown code fence", + lambda text: re.sub(r"^\s*```(?:json)?\s*|\s*```\s*$", "", text), + ), + ( + "replaced smart quotes with straight ones", + lambda text: text.translate(str.maketrans({"“": '"', "”": '"', + "‘": "'", "’": "'"})), + ), + ( + "doubled backslashes that weren't valid JSON escapes", + # {"path": "C:\Users\x"} -> {"path": "C:\\Users\\x"} — a Windows path + # typed raw. Only *invalid* escapes are touched, and `\\` pairs are + # consumed whole, so nothing that already parsed can change meaning. + lambda text: re.sub( + r'\\([\\/"bfnrtu])|\\', + lambda m: m.group(0) if m.group(1) else "\\\\", text), + ), + ( + "removed a stray quote after a bare value", + # {"recursive":false"} -> {"recursive":false} + lambda text: re.sub( + r"(:\s*(?:true|false|null|-?\d+(?:\.\d+)?))\s*\"(\s*[,}\]])", r"\1\2", text), + ), + ( + "removed a trailing comma", + lambda text: re.sub(r",(\s*[}\]])", r"\1", text), + ), + ( + "converted Python literals (True/False/None) to JSON", + lambda text: re.sub(r"(:\s*)(True|False|None)\b", + lambda m: m.group(1) + {"True": "true", "False": "false", + "None": "null"}[m.group(2)], text), + ), + ( + "converted single-quoted strings to double-quoted", + lambda text: re.sub(r"'([^'\"]*)'", r'"\1"', text), + ), +) + + +class RelayJsonError(ValueError): + """Unparseable even after repairs — carries a pointed-at fragment.""" + + +def loads(payload: str) -> tuple[Any, list[str]]: + """Parse *payload*, repairing common model mistakes. + + Returns (data, repairs) where *repairs* names what had to be fixed — empty + when the input was already valid. Raises RelayJsonError with a caret at the + offending character when nothing works.""" + text = str(payload or "").strip() + if not text: + raise RelayJsonError("empty payload") + try: + return json.loads(text), [] + except json.JSONDecodeError as exc: + # Bound to a plain name: Python deletes the `as` target at the end of + # the except block, so referring to it further down would raise + # UnboundLocalError instead of reporting the parse failure. + first_error = exc + + applied: list[str] = [] + candidate = text + for name, repair in _REPAIRS: + repaired = repair(candidate) + if repaired == candidate: + continue + candidate = repaired + applied.append(name) + try: + return json.loads(candidate), applied + except json.JSONDecodeError: + continue # keep going: a payload can be broken in more than one way + + raise RelayJsonError(point_at(text, first_error)) + + +def point_at(text: str, error: json.JSONDecodeError, width: int = 28) -> str: + """Show the failure where it happened. + + A model can act on "you wrote `false\"}` here"; it cannot act on + "Expecting ',' delimiter: line 1 column 75".""" + position = max(0, min(len(text), getattr(error, "pos", 0))) + start = max(0, position - width) + end = min(len(text), position + width) + fragment = text[start:end] + caret = " " * (position - start) + "^" + lead = "…" if start > 0 else "" + tail = "…" if end < len(text) else "" + return ( + f"{error.msg} at character {position}:\n" + f" {lead}{fragment}{tail}\n" + f" {' ' * len(lead)}{caret}" + ) + + +def repair_note(repairs: list[str]) -> str: + """The line appended to a tool result when repairs were needed. + + Phrased as feedback rather than an apology: the model is the author of the + broken JSON and is the one who can stop sending it.""" + if not repairs: + return "" + return ( + " (note: your JSON was malformed — I " + "; ".join(repairs) + + " and ran it anyway. Send valid single-line JSON next time.)" + ) diff --git a/bolt_assistant/server.py b/bolt_assistant/server.py new file mode 100644 index 0000000..bd3a861 --- /dev/null +++ b/bolt_assistant/server.py @@ -0,0 +1,186 @@ +"""The desk API: conversation (with the command relay) and text-to-speech. + + text -> POST /desk/converse_stream (NDJSON, sentence by sentence) + or /desk/converse + [server may relay a `command`; the answer goes to /desk/tool_result] + reply <- final text + text -> POST /desk/tts -> raw 16 kHz mono PCM16, streamed + +Same protocol and conventions as Bolt Console's server_client and Bolt-Pet's +tts, which this is adapted from. +""" + +from __future__ import annotations + +import json +from typing import Callable, Iterable, Iterator, NamedTuple, Optional + +import requests + +from . import config + +_MAX_RELAY_HOPS = 16 + + +class ServerError(Exception): + pass + + +class Reply(NamedTuple): + text: str + streamed: bool = False # already handed over sentence by sentence + voice_id: str = "" # the server's speak_as pick, if any + + +def _headers() -> dict: + return {"X-Desk-Api-Key": config.API_KEY} + + +def check_health(timeout: float = 8.0) -> dict: + response = requests.get(f"{config.SERVER_URL}/desk/health", headers=_headers(), timeout=timeout) + response.raise_for_status() + return response.json() + + +def _reply_from(payload: dict, streamed: bool = False) -> Reply: + if payload.get("type") == "reply": + return Reply(str(payload.get("text") or ""), streamed, str(payload.get("voice_id") or "")) + raise ServerError(str(payload.get("error") or "unknown server response")) + + +def _relay(payload: dict, on_command, on_say) -> dict: + for _ in range(_MAX_RELAY_HOPS): + if payload.get("type") != "command": + break + holding = str(payload.get("say") or "").strip() + if holding and on_say is not None: + try: + on_say(holding) + except Exception: + pass + output = on_command(str(payload.get("command") or "")) + try: + payload = requests.post( + f"{config.SERVER_URL}/desk/tool_result", + json={"session_id": config.SESSION_ID, "token": payload.get("token"), "output": output}, + headers=_headers(), timeout=180, + ).json() + except Exception as exc: + raise ServerError(f"couldn't reach the server during tool relay: {exc}") from exc + return payload + + +def converse(text: str, on_command: Callable[[str], str], + on_say: Optional[Callable[[str], None]] = None, timeout: float = 120.0) -> Reply: + try: + payload = requests.post( + f"{config.SERVER_URL}/desk/converse", + json={"session_id": config.SESSION_ID, "text": text}, + headers=_headers(), timeout=timeout, + ).json() + except Exception as exc: + raise ServerError(f"couldn't reach the server: {exc}") from exc + return _reply_from(_relay(payload, on_command, on_say)) + + +def converse_stream(text: str, on_say: Callable[[str], None], on_command: Callable[[str], str], + timeout: float = 180.0) -> Reply: + """Each sentence to *on_say* as it's generated, so speaking starts after + the first sentence, not the whole answer. Raises ServerError only if + nothing was said yet (the caller then uses converse()).""" + said = [] + try: + response = requests.post( + f"{config.SERVER_URL}/desk/converse_stream", + json={"session_id": config.SESSION_ID, "text": text}, + headers=_headers(), timeout=timeout, stream=True, + ) + response.raise_for_status() + for raw in response.iter_lines(decode_unicode=True): + if not raw: + continue + try: + event = json.loads(raw) + except (TypeError, ValueError): + continue + kind = str(event.get("type") or "") + if kind == "say": + line = str(event.get("text") or "").strip() + if line: + said.append(line) + on_say(line) + elif kind == "command": + response.close() + return _reply_from(_relay(event, on_command, on_say)) + elif kind == "reply": + if event.get("already_spoken"): + return _reply_from(event, streamed=True) + return unsaid(_reply_from(event), said) + elif kind == "error": + raise ServerError(str(event.get("error") or "stream failed")) + except ServerError: + raise + except Exception as exc: + if said: + return Reply("", streamed=True) # never repeat half an answer out loud + raise ServerError(f"streaming failed: {exc}") from exc + raise ServerError("stream ended without a reply") + + +def unsaid(reply: Reply, said: list) -> Reply: + """The part of *reply* not already spoken as streamed sentences — the + server can stream the sentences and then send the whole text again + without `already_spoken` (compared ignoring whitespace).""" + if not said: + return reply + target = len("".join("".join(said).split())) + seen, cut = 0, None + for index, char in enumerate(reply.text): + if not char.isspace(): + seen += 1 + if seen == target: + cut = index + 1 + break + if cut is None or "".join(reply.text[:cut].split()) != "".join("".join(said).split()): + return reply + rest = reply.text[cut:].strip() + return Reply(rest, streamed=not rest, voice_id=reply.voice_id) + + +# ── text to speech ────────────────────────────────────────────────────────── + +class TtsError(Exception): + pass + + +def tts_stream(text: str, voice_id: str = "", chunk_bytes: int = 4096) -> Iterator[bytes]: + """PCM16 bytes as the server synthesises them (whole samples only). + Raises TtsError before yielding anything if the request fails.""" + voice = (voice_id or "").strip() or config.VOICE_ID + if not voice: + raise TtsError("ELEVENLABS_VOICE_ID isn't set, so the server doesn't know which voice to use") + try: + response = requests.post( + f"{config.SERVER_URL}/desk/tts", headers=_headers(), + json={"session_id": config.SESSION_ID, "text": text, "voice_id": voice}, + timeout=60, stream=True, + ) + response.raise_for_status() + except Exception as exc: + raise TtsError(f"text-to-speech request failed: {exc}") from exc + return whole_samples(response.iter_content(chunk_size=chunk_bytes)) + + +def whole_samples(chunks: Iterable[bytes]) -> Iterator[bytes]: + """HTTP chunks split wherever they like, including inside a 16-bit + sample. Carry the odd byte forward, or everything after the first split + plays as static (a Bolt-Pet lesson).""" + carry = b"" + for chunk in chunks: + if not chunk: + continue + data = carry + chunk + usable = len(data) - len(data) % 2 + carry = data[usable:] + if usable: + yield data[:usable] diff --git a/bolt_assistant/speech_text.py b/bolt_assistant/speech_text.py new file mode 100644 index 0000000..fd82242 --- /dev/null +++ b/bolt_assistant/speech_text.py @@ -0,0 +1,173 @@ +"""Turn a server reply into something worth *hearing*. + +The server's persona writes for a chat window: markdown emphasis, bullet +lists, emoji, bare URLs. A TTS voice reads those literally ("asterisk +asterisk OS colon", "https colon slash slash..."), so everything spoken goes +through for_speech() first. Pure string logic, no Qt/audio imports — cheap to +unit test (see tests/test_speech_text.py). + +for_display() is the lighter sibling used for the speech bubble: it drops the +markdown *syntax* but keeps emoji and punctuation, since those render fine. +""" + +from __future__ import annotations + +import re + +# Pictographs, symbols, flags, dingbats, arrows, box drawing, variation +# selectors, ZWJ — anything a voice would either skip or read as a name +# ("black right-pointing triangle"). +_EMOJI = re.compile( + "[" + "\U0001F000-\U0001FAFF" # emoji / pictographs / symbols blocks + "\U00002190-\U000021FF" # arrows + "\U00002300-\U000023FF" # misc technical (⌘ ⏱ …) + "\U000025A0-\U000027BF" # geometric shapes, misc symbols, dingbats + "\U00002B00-\U00002BFF" # extra arrows / shapes + "\U0000FE00-\U0000FE0F" # variation selectors + "\U0001F1E6-\U0001F1FF" # regional indicators (flags) + "\U0000200D" # zero-width joiner + "]+", + flags=re.UNICODE, +) + +# Characters that are markup or decoration rather than speech. Kept out of +# the spoken text entirely; ordinary punctuation (. , ! ? ; : ' " ( ) -) is +# preserved because it shapes prosody. +_UNSPEAKABLE = re.compile(r"[*_#`~^|<>\\{}\[\]/=+@©®™•·–—]+") + +_FENCED_CODE = re.compile(r"```.*?```", re.DOTALL) +_INLINE_CODE = re.compile(r"`([^`]*)`") +_MD_IMAGE = re.compile(r"!\[([^\]]*)\]\([^)]*\)") +_MD_LINK = re.compile(r"\[([^\]]+)\]\([^)]*\)") +_URL = re.compile(r"\b(?:https?://|www\.)\S+") +_HEADING = re.compile(r"^\s{0,3}#{1,6}\s*", re.MULTILINE) +_BLOCKQUOTE = re.compile(r"^\s{0,3}>\s?", re.MULTILINE) +_RULE = re.compile(r"^\s*([-*_=])(?:\s*\1){2,}\s*$", re.MULTILINE) +_BULLET = re.compile(r"^\s*(?:[-*+•·]|\d+[.)])\s+", re.MULTILINE) +_EMPHASIS = re.compile(r"(\*{1,3}|_{1,3})(\S(?:.*?\S)?)\1", re.DOTALL) +_TABLE_PIPE = re.compile(r"[ \t]*\|[ \t]*") + +# Symbols worth saying out loud rather than dropping — a bare "&" read as +# nothing turns "R&D" into "RD". Split in two because the non-ASCII ones sit +# inside the arrow/symbol blocks _EMOJI strips, so they have to be worded +# before that pass; the ASCII ones must wait until *after* markdown parsing +# (an "=" turned into " equals " would stop a "====" rule matching _RULE). +_PRE_SPOKEN_SYMBOLS = { + "→": " to ", + "×": " by ", + "°": " degrees ", + "≈": " about ", +} +_SPOKEN_SYMBOLS = { + "&": " and ", + "%": " percent ", + "@": " at ", + "+": " plus ", + "=": " equals ", +} + +_MULTI_SPACE = re.compile(r"[ \t]+") +_MULTI_PUNCT = re.compile(r"(?:\s*\.){2,}") + + +def _strip_markdown(text: str, *, keep_emoji: bool) -> str: + text = _FENCED_CODE.sub(" (code) ", text) + text = _INLINE_CODE.sub(r"\1", text) + text = _MD_IMAGE.sub(r"\1", text) + text = _MD_LINK.sub(r"\1", text) + text = _RULE.sub("", text) + text = _HEADING.sub("", text) + text = _BLOCKQUOTE.sub("", text) + text = _EMPHASIS.sub(r"\2", text) + if not keep_emoji: + text = _EMOJI.sub(" ", text) + return text + + +def _bullets_to_sentences(text: str) -> str: + """A read-aloud list needs pauses where the eye would see line breaks, + otherwise "OS Linux Uptime 1 day CPU load moderate" runs together.""" + lines = [_BULLET.sub("", line).strip() for line in text.splitlines()] + lines = [line for line in lines if line] + if len(lines) < 2: + return lines[0] if lines else "" + # A bullet like "OS: Linux" ends mid-thought — give the voice a full stop + # so consecutive items don't slur into one run-on sentence. + return " ".join(line if line[-1] in ".!?:,;" else line + "." for line in lines) + + +# ElevenLabs v3 delivery tags — "[laughing]", "[whispering]", "[sighs]". They +# are instructions to the *dialogue* model (see dialogue.py), and only inside a +# dialoguectl scene. Left in an ordinary reply they reach eleven_flash_v2, +# which has no idea what they mean and simply reads the word: observed live +# 2026-07-31, the pet announcing "laughing That one came through clean". +# +# Matched narrowly — a bracketed adverb/gerund, or one of the noise words that +# aren't either — so real bracketed text ("[1]", "[see the docs]") survives. +_AUDIO_TAG_RE = re.compile( + r"\[\s*(?:" + r"[a-z]+(?:ly|ing)" + r"|laughs?|chuckles?|sighs?|exhales?|inhales?|gulps?|pause|beat" + r"|clears throat|shouts?|whispers?|cries|sings?|gasps?|snorts?" + r"|sarcastic|excited|curious|nervous|angry|sad|happy|deadpan|flat" + r")\s*\]", + re.IGNORECASE, +) + + +def strip_audio_tags(text: str) -> str: + """Remove v3 delivery tags from text destined for the ordinary voice.""" + return _AUDIO_TAG_RE.sub(" ", str(text or "")) + + +def for_speech(text: str) -> str: + """Plain prose for the TTS engine: no markdown, no emoji, no bare URLs, + no stray symbols that would be read out character by character.""" + text = strip_audio_tags((text or "").strip()) + if not text: + return "" + for symbol, spoken in _PRE_SPOKEN_SYMBOLS.items(): + text = text.replace(symbol, spoken) + text = _strip_markdown(text, keep_emoji=False) + text = _URL.sub(" link ", text) + text = _TABLE_PIPE.sub(", ", text) + text = _bullets_to_sentences(text) + for symbol, spoken in _SPOKEN_SYMBOLS.items(): + text = text.replace(symbol, spoken) + text = _UNSPEAKABLE.sub(" ", text) + text = _MULTI_PUNCT.sub(".", text) + text = _MULTI_SPACE.sub(" ", text) + text = re.sub(r"\s+([.,!?;:])", r"\1", text) + 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. + + Delivery tags go too — the voice no longer says them, so showing them + would caption a laugh nobody heard.""" + text = strip_audio_tags((text or "").strip()) + if not text: + return "" + text = _strip_markdown(text, keep_emoji=True) + lines = [_BULLET.sub("• ", line).strip() for line in text.splitlines()] + text = " ".join(line for line in lines if line) + return _MULTI_SPACE.sub(" ", text).strip() diff --git a/bolt_assistant/stt.py b/bolt_assistant/stt.py new file mode 100644 index 0000000..f0b4147 --- /dev/null +++ b/bolt_assistant/stt.py @@ -0,0 +1,152 @@ +"""Speech-to-text for one utterance, through the Bolt server's /desk/stt +websocket relay (to Deepgram) — the route Bolt-Pet and the console's mic use, +authenticated with the desk API key; no Deepgram account here. + + t = Transcription(connect) + t.add(frames) # as the utterance is spoken (streamed live) + text = t.finish(audio) # when it ends; `audio` is the whole utterance + t.cancel() # it was too short to matter + +Live when possible, so the text is ready the moment you stop talking. If the +socket never opened or died, finish() opens a fresh one and sends `audio` +in one go, from an empty transcript (so a half-live session can't return +words twice). A healthy socket is never re-sent — that would pay twice. +""" + +from __future__ import annotations + +import json +import threading +import urllib.parse +from typing import Callable, List, Optional + +from . import config + + +class SttError(Exception): + pass + + +def connect(sample_rate: int = config.SAMPLE_RATE): + import websocket # websocket-client + + query = urllib.parse.urlencode({"session_id": config.SESSION_ID, "encoding": "linear16", + "sample_rate": sample_rate}) + return websocket.create_connection( + config.SERVER_URL.replace("http", "ws", 1) + "/desk/stt?" + query, + header=[f"X-Desk-Api-Key: {config.API_KEY}"], timeout=10) + + +def parse_result(message) -> tuple: + """(text, is_final) from a Deepgram results frame; ("", False) otherwise.""" + try: + if isinstance(message, bytes): + message = message.decode("utf-8", "ignore") + data = json.loads(message) + except (TypeError, ValueError): + return "", False + if not isinstance(data, dict) or data.get("type") not in (None, "Results"): + return "", False + alternatives = (data.get("channel") or {}).get("alternatives") or [] + if not alternatives: + return "", False + text = str((alternatives[0] or {}).get("transcript") or "").strip() + return text, bool(data.get("is_final") or data.get("speech_final")) + + +class Transcription: + def __init__(self, connect_fn: Callable = connect): + self._connect = connect_fn + self._finals: List[str] = [] + self._interim = "" + self._lock = threading.Lock() + self._socket = None + self._ok = False + self._reader: Optional[threading.Thread] = None + self._open() + + def _open(self) -> None: + try: + self._socket = self._connect(config.SAMPLE_RATE) + self._ok = True + except Exception: + self._socket, self._ok = None, False + return + self._reader = threading.Thread(target=self._read, name="stt-reader", daemon=True) + self._reader.start() + + def add(self, frames: List[bytes]) -> None: + for frame in frames: + self._send(frame) + + def _send(self, data: bytes) -> None: + if self._ok: + try: + self._socket.send_binary(data) + except Exception: + self._ok = False + + def finish(self, audio: bytes) -> str: + """The transcript. Raises SttError only if even the fallback couldn't + reach the server.""" + if self._ok: + self._flush() + return self._text() + self._close() + if self._reader is not None: + self._reader.join(timeout=1) + with self._lock: + self._finals, self._interim = [], "" + self._open() + if not self._ok: + raise SttError("couldn't reach the transcription server") + step = config.FRAME_SAMPLES * 2 * 8 + for offset in range(0, len(audio), step): + self._send(audio[offset:offset + step]) + self._flush() + return self._text() + + def cancel(self) -> None: + self._ok = False + self._close() + + def _flush(self) -> None: + """Deepgram sends its last results after CloseStream; wait for them, + bounded, so a hung socket can't hold up the reply.""" + try: + self._socket.send(json.dumps({"type": "CloseStream"})) + except Exception: + pass + if self._reader is not None: + self._reader.join(timeout=4) + self._close() + + def _close(self) -> None: + socket, self._socket = self._socket, None + if socket is not None: + try: + socket.close() + except Exception: + pass + + def _read(self) -> None: + socket = self._socket + while socket is not None: + try: + message = socket.recv() + except Exception: + break + if not message: + break + text, is_final = parse_result(message) + with self._lock: + if is_final: + if text: + self._finals.append(text) + self._interim = "" + elif text: + self._interim = text + + def _text(self) -> str: + with self._lock: + return " ".join(self._finals + ([self._interim] if self._interim else [])).strip() diff --git a/bolt_assistant/tools.py b/bolt_assistant/tools.py new file mode 100644 index 0000000..f97925d --- /dev/null +++ b/bolt_assistant/tools.py @@ -0,0 +1,93 @@ +"""What happens to a command the server relays back, when there's no screen +to approve it on. + +ASSISTANT_COMMANDS decides: + + off nothing runs + read (default) filectl list/read only: "what's in my Downloads?" works, + nothing on the machine can change + all everything, shell commands and file writes included, like + Bolt-Pet. Anyone within earshot of the mic can ask for anything. + +A refusal goes back to the model as the command's output, phrased so it +tells you rather than retrying. Pet-only pseudo-commands (petctl, +dialoguectl) are explained, never run. +""" + +from __future__ import annotations + +import re +import subprocess +from pathlib import Path +from typing import Callable + +from . import config, file_ops + +_PET = ("petctl", "bolt-pet", "dialoguectl") +_LABEL = re.compile(r"^(?:command|cmd)\s*:\s*", re.IGNORECASE) + + +def run_shell(command: str, timeout: int = None) -> str: + timeout = timeout or config.COMMAND_TIMEOUT_SECONDS + try: + done = subprocess.run( + command, shell=True, capture_output=True, text=True, errors="replace", + stdin=subprocess.DEVNULL, timeout=timeout, cwd=str(Path.home()), + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + return f"[exit {done.returncode}]\n{(done.stdout or '') + (done.stderr or '')}"[:6000] + except subprocess.TimeoutExpired: + return f"[command timed out after {timeout}s]" + except Exception as exc: + return f"[command failed: {exc}]" + + +def refusal(what: str, mode: str) -> str: + return (f"[assistant] Not run: this is Bolt Assistant, a voice-only client, and it's set to " + f"ASSISTANT_COMMANDS={mode}, which doesn't allow {what}. Tell the user in a sentence " + "that you can't do that from here (Bolt Console can); don't retry.") + + +class Tools: + def __init__(self, mode: str = None, log: Callable[[str], None] = print, + shell: Callable[[str], str] = run_shell): + mode = mode or config.COMMANDS + self.mode = mode if mode in ("off", "read", "all") else "read" # unknown: the safe default + self.log = log + self.shell = shell + self.calls = 0 + + def __call__(self, command: str) -> str: + self.calls += 1 + command = _LABEL.sub("", (command or "").strip(), count=1).strip().strip("`").strip() + if not command: + return "[assistant] empty command" + first = command.split(None, 1)[0].lower() + if first in _PET: + self.log(f" (declined {first}: no pet body here)") + return (f"[assistant] {first} isn't available: this is Bolt Assistant, voice only, " + "with no pet body or screen.") + try: + action = file_ops.parse(command) + except file_ops.FileOpError as exc: + return f"[filectl] {exc}" + if action is not None: + return self._file(action) + if self.mode != "all": + self.log(f" (declined shell command: {command[:80]})") + return refusal("shell commands", self.mode) + self.log(f" $ {command[:120]}") + return self.shell(command) + + def _file(self, action: dict) -> str: + kind = action["action"] + allowed = {"off": set(), "read": {"help", "list", "read"}, + "all": {"help", "list", "read", "write", "edit"}}[self.mode] + if kind not in allowed: + self.log(f" (declined file {kind}: {action.get('path', '')})") + return refusal(f"file {kind}s" if kind != "help" else "file access", self.mode) + self.log(f" file {kind}: {action.get('path', '')}") + try: + return file_ops.execute(action) + except file_ops.FileOpError as exc: + return f"[filectl] {exc}" diff --git a/bolt_assistant/wake.py b/bolt_assistant/wake.py new file mode 100644 index 0000000..da4ab31 --- /dev/null +++ b/bolt_assistant/wake.py @@ -0,0 +1,50 @@ +"""Finding the wake phrase in a transcript, and what was said after it. + + >>> find_wake("Hey Bolt, what's the weather?", WAKE) + (True, "what's the weather?") + >>> find_wake("tighten that bolt", WAKE) + (False, '') + +Rules, all aimed at not waking up on ordinary conversation: + +- The phrase has to **open** what was said. Only filler may come before it + ("um, hey Bolt", "okay so Bolt"), at most two words of it, so "tighten + that bolt" and "the bolt on the door" never count. +- Whole words only: "boltage" is not "bolt". +- Longest phrase wins, so "hey thunderbolt" isn't read as "hey" + garbage. +""" + +from __future__ import annotations + +import re +from typing import Iterable, Tuple + +_WORD = re.compile(r"[a-z0-9']+") +_LEAD_SLACK = 2 # filler words allowed before the wake phrase +_FILLER = {"um", "uh", "erm", "er", "so", "oh", "ok", "okay", "well", "and", "right", + "alright", "yeah", "hmm", "hm", "ah", "hi", "hey", "yo"} + + +def _words(text: str) -> list: + """Lower-case words with their character spans in the original text.""" + return [(m.group(0), m.start(), m.end()) for m in _WORD.finditer(text.lower())] + + +def find_wake(transcript: str, phrases: Iterable[str]) -> Tuple[bool, str]: + """(heard, remainder): whether a wake phrase opens *transcript*, and the + request that followed it in the same breath ("" if it was just the + phrase).""" + words = _words(transcript or "") + if not words: + return False, "" + candidates = sorted({tuple(_WORD.findall(p.lower())) for p in phrases if p.strip()}, + key=len, reverse=True) + for start in range(min(_LEAD_SLACK + 1, len(words))): + if any(word not in _FILLER for word, _, _ in words[:start]): + break # a real word came first: this isn't addressed to Bolt + for phrase in candidates: + span = words[start:start + len(phrase)] + if len(span) == len(phrase) and all(w == p for (w, _, _), p in zip(span, phrase)): + rest = transcript[span[-1][2]:] + return True, rest.lstrip(" \t,.;:!?-—").strip() + return False, "" diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..48bab21 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +requests>=2.31 # the desk API +sounddevice>=0.4.6 # microphone and speaker (Linux also: sudo apt install libportaudio2) +websocket-client>=1.7 # the server's /desk/stt speech-to-text relay +pytest>=8.0 # tests only diff --git a/run.bat b/run.bat new file mode 100644 index 0000000..f078b2d --- /dev/null +++ b/run.bat @@ -0,0 +1,13 @@ +@echo off +REM Bolt Assistant: listens for "Hey Bolt" on the default mic. Keep this window +REM open (it shows what was heard and said); Ctrl+C or close it to stop. +setlocal +set "ROOT=%~dp0" +if not exist "%ROOT%.venv" ( + echo First run: creating .venv and installing requirements... + python -m venv "%ROOT%.venv" + "%ROOT%.venv\Scripts\python" -m pip install --upgrade pip + "%ROOT%.venv\Scripts\python" -m pip install -r "%ROOT%requirements.txt" +) +set "PYTHONPATH=%ROOT%;%PYTHONPATH%" +"%ROOT%.venv\Scripts\python" -m bolt_assistant %* diff --git a/run.sh b/run.sh new file mode 100644 index 0000000..c90d386 --- /dev/null +++ b/run.sh @@ -0,0 +1,11 @@ +#!/usr/bin/env bash +# Bolt Assistant: listens for "Hey Bolt" on the default mic. Ctrl+C to stop. +set -euo pipefail +ROOT="$(cd "$(dirname "$0")" && pwd)" +if [ ! -d "$ROOT/.venv" ]; then + python3 -m venv "$ROOT/.venv" + "$ROOT/.venv/bin/pip" install --upgrade pip + "$ROOT/.venv/bin/pip" install -r "$ROOT/requirements.txt" +fi +export PYTHONPATH="$ROOT${PYTHONPATH:+:$PYTHONPATH}" +exec "$ROOT/.venv/bin/python" -m bolt_assistant "$@" diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_assistant.py b/tests/test_assistant.py new file mode 100644 index 0000000..71d0498 --- /dev/null +++ b/tests/test_assistant.py @@ -0,0 +1,359 @@ +"""Bolt Assistant without a mic, speaker or server: synthetic audio frames, +a fake transcriber, fake converse/tts/playback. Covers wake matching, the +voice-activity detector, the listen -> ask -> speak flow, and the command +rules.""" + +import array +import json +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import pytest + +from bolt_assistant import config, server +from bolt_assistant.assistant import Assistant, annotate +from bolt_assistant.listener import Segmenter +from bolt_assistant.stt import Transcription, parse_result +from bolt_assistant.tools import Tools +from bolt_assistant.wake import find_wake + +LOUD, QUIET = 3000, 20 + + +def frame(level, samples=1280): + return array.array("h", [level, -level] * (samples // 2)).tobytes() + + +# ── wake phrases ──────────────────────────────────────────────────────────── + +W = config.WAKE_WORDS + + +@pytest.mark.parametrize("said, expected", [ + ("Hey Bolt, what's the weather?", (True, "what's the weather?")), + ("Bolt.", (True, "")), + ("Thunderbolt turn on the lights", (True, "turn on the lights")), + ("hey thunderbolt!", (True, "")), + ("um, hey Bolt what time is it", (True, "what time is it")), + ("Hey assistant. Set a timer", (True, "Set a timer")), + ("Hey volt, play music", (True, "play music")), # a common mishearing, after "hey" + ("okay so bolt, hi", (True, "hi")), + ("tighten that bolt", (False, "")), + ("the bolt on the door", (False, "")), + ("I was bold", (False, "")), # the mishearing alone isn't a wake word + ("assistant set a timer", (False, "")), # "assistant" needs the "hey" + ("boltage is high", (False, "")), + ("", (False, "")), +]) +def test_find_wake(said, expected): + assert find_wake(said, W) == expected + + +# ── voice-activity detection ──────────────────────────────────────────────── + +def seg(**kw): + args = dict(threshold=500, start_frames=3, silence_end_sec=0.4, min_sec=0.3, max_sec=2.0) + args.update(kw) + return Segmenter(**args) + + +def feed(s, frames): + events = [] + for f in frames: + events += s.feed(f) + return events + + +def test_silence_never_starts_an_utterance(): + assert feed(seg(), [frame(QUIET)] * 50) == [] + + +def test_speech_starts_with_pre_roll_and_ends_after_silence(): + s = seg() + events = feed(s, [frame(QUIET)] * 6 + [frame(LOUD)] * 6 + [frame(QUIET)] * 5) + kinds = [e.kind for e in events] + assert kinds[0] == "start" and kinds[-1] == "end" + start = events[0] + assert len(start.frames) == 4 + 3 # pre-roll kept the first syllable, plus the trigger + end = events[-1] + assert len(end.audio) == (4 + 6 + 5) * 2560 # pre-roll + speech + trailing silence + + +def test_a_click_is_not_speech_and_a_short_burst_is_dropped(): + s = seg(min_sec=0.5) # needs 6+ loud frames + assert feed(s, [frame(LOUD), frame(QUIET)] * 5) == [] # isolated clicks + events = feed(s, [frame(LOUD)] * 3 + [frame(QUIET)] * 5) + assert [e.kind for e in events if e.kind != "frame"] == ["start", "dropped"] + + +def test_long_speech_is_cut_off_at_the_cap(): + events = feed(seg(max_sec=1.0), [frame(LOUD)] * 30) + ends = [e for e in events if e.kind == "end"] + assert ends and ends[0].seconds <= 1.0 + 1e-6 + + +# ── the flow ──────────────────────────────────────────────────────────────── + +class FakeTranscription: + """Each utterance returns the next scripted transcript.""" + + script = [] + + def __init__(self): + self.frames = 0 + self.cancelled = False + + def add(self, frames): + self.frames += len(frames) + + def finish(self, audio): + return FakeTranscription.script.pop(0) + + def cancel(self): + self.cancelled = True + + +class Clock: + def __init__(self): + self.t = 1000.0 + + def __call__(self): + return self.t + + +def utterance(): + return [frame(LOUD)] * 8 + [frame(QUIET)] * 12 + + +def make(replies, commands="read", streaming=False, **kw): + asked, played, logs = [], [], [] + + def converse(message, on_command, on_say=None): + asked.append(message) + reply = replies.pop(0) + if callable(reply): + return reply(on_command) + return server.Reply(reply) + + clock = Clock() + a = Assistant(transcription=FakeTranscription, converse=converse, + converse_stream=None if not streaming else kw.pop("stream"), + tts=lambda text, voice: [text.encode()], play=lambda chunks: played.extend(chunks), + tools=Tools(mode=commands, log=logs.append), log=logs.append, clock=clock, **kw) + return a, asked, played, logs, clock + + +def say(a, transcript): + FakeTranscription.script.append(transcript) + for f in utterance(): + a.process(f) + + +def spoken(played): + return [p.decode() for p in played if not p.startswith(b"\x00") and len(p) < 2000] + + +@pytest.fixture(autouse=True) +def _config(monkeypatch): + monkeypatch.setattr(config, "STREAMING_REPLIES", False) + monkeypatch.setattr(config, "VAD_RMS_THRESHOLD", 500) + FakeTranscription.script = [] + + +def test_ignores_speech_without_the_wake_phrase(): + a, asked, played, logs, _ = make([]) + say(a, "so anyway I was telling him about the bolt") + assert asked == [] and played == [] + + +def test_wake_phrase_and_request_in_one_breath(): + a, asked, played, logs, _ = make(["It's three o'clock."]) + say(a, "Hey Bolt, what time is it?") + assert asked[0].startswith("what time is it?\n\n[Bolt Assistant: a voice-only assistant") + assert "It's three o'clock." in spoken(played) + assert a.state == "idle" + + +def test_bare_wake_phrase_chimes_then_takes_the_next_utterance(): + a, asked, played, logs, clock = make(["Sure."]) + say(a, "Hey Bolt") + assert a.state == "command" and asked == [] + assert played # the chime + clock.t += 1 + say(a, "turn it up") # no wake phrase needed now + assert asked[0].startswith("turn it up") + + +def test_the_command_window_closes(): + a, asked, played, logs, clock = make([]) + say(a, "Thunderbolt") + clock.t += config.COMMAND_WAIT_SEC + 5 + say(a, "so then we left") + assert asked == [] and a.state == "idle" + + +def test_a_question_keeps_listening_for_the_answer(): + a, asked, played, logs, clock = make(["Which room, the kitchen or the bedroom?", "Done."]) + say(a, "Hey Bolt, turn on the lights") + assert a.state == "follow_up" + clock.t += 2 + say(a, "the kitchen") + assert asked[1].startswith("the kitchen") and a.state == "idle" + + +def test_mic_audio_during_speech_is_discarded(): + a, asked, played, logs, clock = make(["Hello there."]) + a.on_mic_frame(frame(LOUD)) # queued while... (simulating backlog) + say(a, "hey bolt hello") + assert a.frames.empty() # what the mic heard while Bolt talked is gone + assert not a.segmenter.in_speech + + +def test_streamed_sentences_are_spoken_as_they_arrive(): + def stream(message, on_say, on_command): + on_say("First.") + on_say("Second.") + return server.Reply("First. Second.", streamed=True) + + a, asked, played, logs, _ = make([], streaming=True, stream=stream) + a._stream = stream + say(a, "hey bolt tell me two things") + assert spoken(played)[-2:] == ["First.", "Second."] + + +def test_server_failure_chimes_and_goes_back_to_listening(): + def boom(on_command): + raise server.ServerError("down") + + a, asked, played, logs, _ = make([boom]) + say(a, "hey bolt anything") + assert any("down" in line for line in logs) and a.state == "idle" + + +def test_markdown_is_not_read_aloud(): + a, asked, played, logs, _ = make(["**Sure** — see https://example.com for `details`."]) + say(a, "hey bolt help") + text = spoken(played)[-1] + assert "**" not in text and "https" not in text and "`" not in text + + +def test_the_note_describes_what_bolt_may_do(): + assert "can't run shell commands" in annotate("x", "read") + assert "`command: `" in annotate("x", "all") + assert "can't act on this computer" in annotate("x", "off") + + +# ── commands relayed by the server ────────────────────────────────────────── + +def filectl(**payload): + return "filectl " + json.dumps(payload) + + +def test_read_mode_lists_files_but_refuses_shell_and_writes(tmp_path): + (tmp_path / "a.txt").write_text("hi") + ran = [] + tools = Tools(mode="read", log=lambda _m: None, shell=lambda c: ran.append(c) or "[exit 0]\n") + assert "a.txt" in tools(filectl(op="list", path=str(tmp_path))) + assert "hi" in tools(filectl(op="read", path=str(tmp_path / "a.txt"))) + assert "Not run" in tools("rm -rf /") + assert "Not run" in tools(filectl(op="write", path=str(tmp_path / "b.txt"), content="x")) + assert ran == [] and not (tmp_path / "b.txt").exists() + + +def test_all_mode_runs_commands_and_off_runs_nothing(tmp_path): + ran = [] + shell = lambda c: ran.append(c) or "[exit 0]\n" # noqa: E731 + assert Tools(mode="all", log=lambda _m: None, shell=shell)("command: echo hi") == "[exit 0]\n" + assert ran == ["echo hi"] # the stray "command:" label was stripped + assert "Not run" in Tools(mode="off", log=lambda _m: None, shell=shell)(filectl(op="list", path=".")) + assert "isn't available" in Tools(mode="all", log=lambda _m: None)("petctl emote wave") + + +def test_unknown_mode_falls_back_to_read(): + assert Tools(mode="everything", log=lambda _m: None).mode == "read" + + +def test_a_relayed_command_goes_through_the_rules(tmp_path): + (tmp_path / "notes.txt").write_text("buy milk") + + def reply_with_tool(on_command): + output = on_command(filectl(op="read", path=str(tmp_path / "notes.txt"))) + assert "buy milk" in output + return server.Reply("Your note says buy milk.") + + a, asked, played, logs, _ = make([reply_with_tool]) + say(a, "hey bolt read my notes") + assert "Your note says buy milk." in spoken(played) + + +# ── server pieces ─────────────────────────────────────────────────────────── + +def test_whole_samples_carries_an_odd_byte_across_chunks(): + out = b"".join(server.whole_samples([b"\x01", b"\x02\x03", b"\x04\x05"])) + assert out == b"\x01\x02\x03\x04" # the lone trailing byte waits for its partner + + +def test_unsaid_drops_what_was_already_spoken(): + reply = server.unsaid(server.Reply("One. Two.\n\nThree."), ["One.", "Two."]) + assert reply == server.Reply("Three.", streamed=False) + + +def test_tts_needs_a_voice(monkeypatch): + monkeypatch.setattr(config, "VOICE_ID", "") + with pytest.raises(server.TtsError, match="ELEVENLABS_VOICE_ID"): + server.tts_stream("hi") + + +def test_tts_request_shape(monkeypatch): + monkeypatch.setattr(config, "SERVER_URL", "http://s") + monkeypatch.setattr(config, "API_KEY", "k") + monkeypatch.setattr(config, "VOICE_ID", "v1") + response = MagicMock() + response.iter_content.return_value = [b"\x00\x01"] + with patch.object(server.requests, "post", return_value=response) as post: + assert list(server.tts_stream("hello", voice_id="v2")) == [b"\x00\x01"] + assert post.call_args.args[0] == "http://s/desk/tts" + assert post.call_args.kwargs["json"]["voice_id"] == "v2" # a speak_as pick wins + assert post.call_args.kwargs["headers"] == {"X-Desk-Api-Key": "k"} + + +def test_parse_result(): + msg = json.dumps({"type": "Results", "is_final": True, "channel": {"alternatives": [{"transcript": "hey bolt"}]}}) + assert parse_result(msg) == ("hey bolt", True) + assert parse_result("junk") == ("", False) + + +def test_transcription_falls_back_to_one_upload(): + class Sock: + def __init__(self): + self.audio, self.replies = [], [json.dumps({"is_final": True, "channel": { + "alternatives": [{"transcript": "hey bolt"}]}}), ""] + + def send_binary(self, data): + self.audio.append(data) + + def send(self, _text): + pass + + def recv(self): + return self.replies.pop(0) + + def close(self): + pass + + late = Sock() + calls = iter([ConnectionError("down"), late]) + + def connect(_rate): + item = next(calls) + if isinstance(item, Exception): + raise item + return item + + t = Transcription(connect) + t.add([frame(LOUD)]) # nowhere to go: the live socket never opened + assert t.finish(frame(LOUD) * 2) == "hey bolt" + assert b"".join(late.audio) == frame(LOUD) * 2 diff --git a/tests/test_file_ops.py b/tests/test_file_ops.py new file mode 100644 index 0000000..56a80f9 --- /dev/null +++ b/tests/test_file_ops.py @@ -0,0 +1,343 @@ +"""filectl parsing + execution — the pseudo-commands the server can relay to +read/write/edit local files instead of a raw shell heredoc. + +filectl {"op": ...} is a single-line JSON envelope (not a multi-line +marker block) because it rides the "command" tool marker, which the main +repo's tool-call extractor only captures up to the next newline — see the +module docstring in bolt_assistant/file_ops.py.""" + +import json +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from bolt_assistant import file_ops + + +def _cmd(payload: dict) -> str: + return "filectl " + json.dumps(payload) + + +# ── parsing ────────────────────────────────────────────────────────────────── + +def test_non_file_commands_are_left_alone(): + assert file_ops.parse("ls -la") is None + assert file_ops.parse("systemctl restart nginx") is None + assert file_ops.parse("") is None + # "filed" must not be mistaken for the "file" prefix + assert file_ops.parse("filed --list") is None + + +def test_list_parses_defaults(): + assert file_ops.parse(_cmd({"op": "list", "path": "/tmp"})) == { + "action": "list", "path": "/tmp", "pattern": "*", "recursive": False, + } + + +def test_list_parses_pattern_and_recursive(): + assert file_ops.parse(_cmd({"op": "list", "path": "/tmp", "pattern": "*.py", "recursive": True})) == { + "action": "list", "path": "/tmp", "pattern": "*.py", "recursive": True, + } + + +def test_list_requires_a_path(): + with pytest.raises(file_ops.FileOpError): + file_ops.parse(_cmd({"op": "list"})) + + +def test_list_rejects_empty_pattern(): + with pytest.raises(file_ops.FileOpError): + file_ops.parse(_cmd({"op": "list", "path": "/tmp", "pattern": ""})) + + +def test_read_parses_path_and_optional_line_range(): + assert file_ops.parse(_cmd({"op": "read", "path": "/tmp/a.txt"})) == { + "action": "read", "path": "/tmp/a.txt", "start": None, "end": None, + } + assert file_ops.parse(_cmd({"op": "read", "path": "/tmp/a.txt", "start": 10, "end": 40})) == { + "action": "read", "path": "/tmp/a.txt", "start": 10, "end": 40, + } + + +def test_read_requires_a_path(): + with pytest.raises(file_ops.FileOpError): + file_ops.parse(_cmd({"op": "read"})) + + +def test_read_rejects_non_numeric_line_args(): + with pytest.raises(file_ops.FileOpError): + file_ops.parse('filectl {"op": "read", "path": "/tmp/a.txt", "start": "start"}') + + +def test_write_parses_path_and_content(): + assert file_ops.parse(_cmd({"op": "write", "path": "/tmp/a.txt", "content": "hello\nworld"})) == { + "action": "write", "path": "/tmp/a.txt", "content": "hello\nworld", + } + + +def test_write_allows_empty_content(): + assert file_ops.parse(_cmd({"op": "write", "path": "/tmp/a.txt", "content": ""})) == { + "action": "write", "path": "/tmp/a.txt", "content": "", + } + + +def test_write_without_content_raises(): + with pytest.raises(file_ops.FileOpError): + file_ops.parse(_cmd({"op": "write", "path": "/tmp/a.txt"})) + + +def test_edit_parses_old_and_new(): + assert file_ops.parse(_cmd({"op": "edit", "path": "/tmp/a.txt", "old": "foo\nbar", "new": "baz"})) == { + "action": "edit", "path": "/tmp/a.txt", "old": "foo\nbar", "new": "baz", + } + + +def test_edit_rejects_identical_old_and_new(): + with pytest.raises(file_ops.FileOpError): + file_ops.parse(_cmd({"op": "edit", "path": "/tmp/a.txt", "old": "same", "new": "same"})) + + +def test_edit_missing_fields_raises(): + with pytest.raises(file_ops.FileOpError): + file_ops.parse(_cmd({"op": "edit", "path": "/tmp/a.txt"})) + + +def test_content_with_embedded_quotes_and_shell_metacharacters_survives(): + payload = 'echo "hi $USER" `whoami` && rm -rf /' + command = _cmd({"op": "write", "path": "/tmp/a.txt", "content": payload}) + assert "\n" not in command # stays on one line, as the command marker requires + assert file_ops.parse(command) == {"action": "write", "path": "/tmp/a.txt", "content": payload} + + +def test_multiline_content_stays_on_one_physical_line(): + content = "line one\nline two\nline three with \"quotes\" and \\backslashes\\" + command = _cmd({"op": "write", "path": "/tmp/a.txt", "content": content}) + assert "\n" not in command + assert file_ops.parse(command)["content"] == content + + +def test_help(): + assert file_ops.parse("filectl help") == {"action": "help"} + assert file_ops.parse("filectl") == {"action": "help"} + assert file_ops.parse(_cmd({"op": "help"})) == {"action": "help"} + + +def test_invalid_json_raises(): + with pytest.raises(file_ops.FileOpError): + file_ops.parse("filectl {not valid json") + + +def test_non_object_json_raises(): + with pytest.raises(file_ops.FileOpError): + file_ops.parse("filectl [1, 2, 3]") + + +def test_unknown_op_raises(): + with pytest.raises(file_ops.FileOpError): + file_ops.parse(_cmd({"op": "frobnicate", "path": "/tmp/a.txt"})) + + +# ── execution ──────────────────────────────────────────────────────────────── + +def test_list_shows_files_and_subdirectories(tmp_path): + (tmp_path / "a.txt").write_text("hi") + (tmp_path / "sub").mkdir() + (tmp_path / "sub" / "b.txt").write_text("nested") + + action = file_ops.parse(_cmd({"op": "list", "path": str(tmp_path)})) + output = file_ops.execute(action) + + assert "a.txt\t2B" in output + assert "sub/" in output + assert "b.txt" not in output # non-recursive: nested file not shown + + +def test_list_recursive_finds_nested_files(tmp_path): + (tmp_path / "sub").mkdir() + (tmp_path / "sub" / "b.txt").write_text("nested") + + action = file_ops.parse(_cmd({"op": "list", "path": str(tmp_path), "recursive": True})) + output = file_ops.execute(action) + + assert "sub/b.txt" in output or "sub\\b.txt" in output # os-dependent separator + + +def test_list_pattern_filters_entries(tmp_path): + (tmp_path / "a.py").write_text("x") + (tmp_path / "b.txt").write_text("y") + + action = file_ops.parse(_cmd({"op": "list", "path": str(tmp_path), "pattern": "*.py"})) + output = file_ops.execute(action) + + assert "a.py" in output + assert "b.txt" not in output + + +def test_list_empty_directory_says_so(tmp_path): + action = file_ops.parse(_cmd({"op": "list", "path": str(tmp_path)})) + output = file_ops.execute(action) + assert "no entries" in output + + +def test_list_missing_directory_raises(): + with pytest.raises(file_ops.FileOpError): + file_ops.execute({"action": "list", "path": "/no/such/dir", "pattern": "*", "recursive": False}) + + +def test_list_rejects_a_file_path(tmp_path): + target = tmp_path / "a.txt" + target.write_text("hi") + with pytest.raises(file_ops.FileOpError): + file_ops.execute({"action": "list", "path": str(target), "pattern": "*", "recursive": False}) + + +def test_list_truncates_past_the_entry_cap(tmp_path, monkeypatch): + monkeypatch.setattr(file_ops, "_MAX_LIST_ENTRIES", 3) + for i in range(5): + (tmp_path / f"f{i}.txt").write_text("x") + + action = file_ops.parse(_cmd({"op": "list", "path": str(tmp_path)})) + output = file_ops.execute(action) + + assert "truncated" in output + assert output.count(".txt") == 3 + + +def test_write_then_read_round_trips(tmp_path): + target = tmp_path / "notes.txt" + write_action = file_ops.parse(_cmd({"op": "write", "path": str(target), "content": "line one\nline two"})) + result = file_ops.execute(write_action) + assert target.read_text() == "line one\nline two" + assert str(target) in result + + read_action = file_ops.parse(_cmd({"op": "read", "path": str(target)})) + output = file_ops.execute(read_action) + assert "line one" in output + assert "line two" in output + + +def test_read_missing_file_raises(): + with pytest.raises(file_ops.FileOpError): + file_ops.execute({"action": "read", "path": "/no/such/file.txt", "start": None, "end": None}) + + +def test_read_respects_line_range(tmp_path): + target = tmp_path / "a.txt" + target.write_text("\n".join(f"line{i}" for i in range(1, 11))) + action = file_ops.parse(_cmd({"op": "read", "path": str(target), "start": 3, "end": 5})) + output = file_ops.execute(action) + assert "line3" in output and "line5" in output + assert "line1" not in output and "line6" not in output + + +def test_edit_replaces_a_unique_match(tmp_path): + target = tmp_path / "a.py" + target.write_text("def foo():\n return 1\n") + action = file_ops.parse(_cmd({"op": "edit", "path": str(target), "old": "return 1", "new": "return 2"})) + file_ops.execute(action) + assert target.read_text() == "def foo():\n return 2\n" + + +def test_edit_fails_when_text_not_found(tmp_path): + target = tmp_path / "a.py" + target.write_text("def foo():\n return 1\n") + action = file_ops.parse(_cmd({"op": "edit", "path": str(target), "old": "nope", "new": "x"})) + with pytest.raises(file_ops.FileOpError): + file_ops.execute(action) + assert target.read_text() == "def foo():\n return 1\n" # untouched + + +def test_edit_fails_when_text_is_ambiguous(tmp_path): + target = tmp_path / "a.py" + target.write_text("x = 1\nx = 1\n") + action = file_ops.parse(_cmd({"op": "edit", "path": str(target), "old": "x = 1", "new": "x = 2"})) + with pytest.raises(file_ops.FileOpError): + file_ops.execute(action) + assert target.read_text() == "x = 1\nx = 1\n" # untouched + + +def test_write_creates_parent_directories(tmp_path): + target = tmp_path / "nested" / "dir" / "a.txt" + action = file_ops.parse(_cmd({"op": "write", "path": str(target), "content": "hi"})) + file_ops.execute(action) + assert target.read_text() == "hi" + + +# ── lenient JSON (see relay_json) ─────────────────────────────────────────── +# Machine-written JSON fails in a small, repeatable set of ways. Observed live +# 2026-07-30: a stray quote after `false` cost a whole desk turn — rejected, +# re-sent identically, rejected again, then abandoned with a promise to the +# user that nothing fulfilled. + +def test_the_stray_quote_that_cost_a_live_turn_now_parses(): + action = file_ops.parse( + 'filectl {"op":"list","path":"/home/maji/Documents","pattern":"*","recursive":false"}' + ) + assert action["action"] == "list" + assert action["path"] == "/home/maji/Documents" + assert action["recursive"] is False + assert action["_repairs"] == ["removed a stray quote after a bare value"] + + +@pytest.mark.parametrize("payload,expected", [ + ('{"op": "read", "path": "/tmp/a.txt",}', "removed a trailing comma"), + ("{'op': 'read', 'path': '/tmp/a.txt'}", "converted single-quoted strings to double-quoted"), + ('{“op”: “read”, “path”: “/tmp/a.txt”}', "replaced smart quotes with straight ones"), + ('```json {"op": "read", "path": "/tmp/a.txt"} ```', "stripped a markdown code fence"), +]) +def test_common_model_json_mistakes_are_repaired(payload, expected): + action = file_ops.parse("filectl " + payload) + assert action["action"] == "read" + assert action["path"] == "/tmp/a.txt" + assert expected in action["_repairs"] + + +def test_python_literals_are_converted(): + action = file_ops.parse('filectl {"op": "list", "path": "/tmp", "recursive": True}') + assert action["recursive"] is True + + +def test_a_repair_is_reported_in_the_output_never_hidden(tmp_path): + """Silently fixing it would work today and guarantee the same broken call + tomorrow — the model has to be told while it still has the turn.""" + (tmp_path / "a.txt").write_text("hello", encoding="utf-8") + action = file_ops.parse( + 'filectl {"op":"list","path":%s,"recursive":false"}' % json.dumps(str(tmp_path)) + ) + output = file_ops.execute(action) + assert "a.txt" in output + assert "your JSON was malformed" in output + assert "stray quote" in output + + +def test_valid_json_gets_no_repair_note(tmp_path): + (tmp_path / "a.txt").write_text("hello", encoding="utf-8") + action = file_ops.parse('filectl {"op": "list", "path": %s}' % json.dumps(str(tmp_path))) + assert "_repairs" not in action + assert "malformed" not in file_ops.execute(action) + + +def test_genuinely_unparseable_json_points_at_the_character(): + """"Expecting ',' delimiter: char 74" is not something a model can act on.""" + with pytest.raises(file_ops.FileOpError) as excinfo: + file_ops.parse('filectl {"op": "read", "path": "/tmp/a.txt" "extra": 1}') + message = str(excinfo.value) + assert "^" in message # caret under the offending character + assert '"extra"' in message # ...and the fragment around it + + +def test_a_raw_windows_path_is_repaired_and_reported(): + r"""Models type Windows paths with single backslashes. `\U` isn't a JSON + escape, so strict parsing would reject the whole call.""" + action = file_ops.parse(r'filectl {"op": "read", "path": "C:\Users\sam\data.txt"}') + assert action["path"] == r"C:\Users\sam\data.txt" + assert any("backslash" in repair for repair in action["_repairs"]) + + +def test_the_backslash_repair_leaves_valid_escapes_alone(): + # The escaped pair and the valid \n survive; only the lone \U is doubled. + action = file_ops.parse(r'filectl {"op": "read", "path": "C:\\ok\Users\n.txt"}') + assert action["path"] == "C:\\ok\\Users\n.txt" diff --git a/tests/test_speech_text.py b/tests/test_speech_text.py new file mode 100644 index 0000000..665c36d --- /dev/null +++ b/tests/test_speech_text.py @@ -0,0 +1,111 @@ +"""Sanitizing chat-formatted replies into speakable prose. Pure string logic +— no audio hardware, no Qt (see the testing conventions in CLAUDE.md).""" + +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from bolt_assistant.speech_text import for_display, for_speech, is_question + + +def test_bold_markers_are_not_spoken(): + assert "*" not in for_speech("Here's the **maji-desktop** snapshot") + assert for_speech("Here's the **maji-desktop** snapshot") == "Here's the maji-desktop snapshot" + + +def test_italics_and_underscores_dropped(): + assert for_speech("that is _really_ odd") == "that is really odd" + assert for_speech("***everything*** is fine") == "everything is fine" + + +def test_bullet_list_becomes_sentences(): + spoken = for_speech( + "System status:\n" + "* **OS:** Linux 7.0.0\n" + "* **Uptime:** 1 day\n" + ) + assert "*" not in spoken + assert spoken == "System status: OS: Linux 7.0.0. Uptime: 1 day." + + +def test_emoji_and_symbols_removed(): + assert for_speech("✅ Done 🚀 — all good") == "Done all good" + assert for_speech("→ next step") == "to next step" + + +def test_urls_and_code_are_not_read_out_character_by_character(): + assert for_speech("see https://example.com/x?y=1 for docs") == "see link for docs" + assert for_speech("run `sudo reboot` now") == "run sudo reboot now" + assert for_speech("here:\n```\nls -la\n```\n") == "here: (code)." + + +def test_headings_quotes_and_rules_stripped(): + assert for_speech("## Summary\n---\n> quoted bit") == "Summary. quoted bit." + + +def test_ampersand_and_percent_are_spoken_as_words(): + assert for_speech("R&D at 50% capacity") == "R and D at 50 percent capacity" + + +def test_blank_and_symbol_only_input(): + assert for_speech("") == "" + assert for_speech(None) == "" + assert for_speech("***") == "" + + +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) + + +# ── ElevenLabs v3 delivery tags (observed live 2026-07-31) ────────────────── +# "[laughing] That one came through clean" was spoken as "laughing That one +# came through clean": the tags mean something to the dialogue model inside a +# dialoguectl scene, and nothing at all to the ordinary reply voice. + +def test_delivery_tags_are_never_spoken_aloud(): + spoken = for_speech("[laughing] That one came through clean.") + assert "laughing" not in spoken + assert spoken.startswith("That one came through clean") + + +def test_the_bubble_does_not_caption_a_laugh_nobody_heard(): + assert "[laughing]" not in for_display("[laughing] All good.") + + +@pytest.mark.parametrize("tag", [ + "[whispering]", "[cheerfully]", "[sighs]", "[laughs]", "[clears throat]", + "[nervously]", "[excited]", "[pause]", "[shouting]", +]) +def test_the_common_tags_are_all_covered(tag): + assert "" == for_speech(tag).strip() + + +def test_ordinary_bracketed_text_survives(): + """The tags are matched narrowly on purpose — real bracketed content is + part of what the user asked to hear.""" + assert "1" in for_speech("See reference [1] for details.") + assert "docs" in for_speech("It's in [the docs] somewhere.")