diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..59d5171 --- /dev/null +++ b/.env.example @@ -0,0 +1,68 @@ +# Bolt Assistant settings. +# +# Copy this file to .env and fill in the three required values: +# +# cp .env.example .env (bash) copy .env.example .env (cmd/PowerShell) +# +# Lines that carry a value below are working defaults; remove the leading # +# only to change one. Blank means "not set". + +# ── server ─────────────────────────────────────────────────────────────────── +# Where your Bolt server's desk API listens (port 5002 in the default setup). +BOLT_SERVER_URL= +# A desk API key for this device (ask Bolt to "create an api key for Bolt +# Assistant" — it arrives as a file through the heartbeat). +DESK_API_KEY= +# Its own conversation on the server. Defaults to pet-assistant-; +# the "pet-" prefix gives it voice-client behaviour (short, speakable replies). +# DESK_SESSION_ID= +# The ElevenLabs voice the server's /desk/tts speaks in (a voice id). +ELEVENLABS_VOICE_ID= +# TTS_SAMPLE_RATE=16000 +# STREAMING_REPLIES=true # speak sentences as they form; false waits for the whole reply + +# ── audio devices ──────────────────────────────────────────────────────────── +# Blank = the system default; a number = the device index; anything else is a +# name substring (sounddevice's rules). `run.bat --devices` lists them. +# MIC_DEVICE= +# SPEAKER_DEVICE= + +# ── wake phrases ───────────────────────────────────────────────────────────── +# With REQUIRE_WAKE_WORD off (the default) the open mic itself is the +# activation: anything that sounds like speech is sent and answered. Set it to +# true to need a phrase first ("Hey Bolt", "Thunderbolt", "Hey Assistant"...). +# REQUIRE_WAKE_WORD=false +# 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 +# These count anywhere in what was said ("what's the weather, assistant?"). +# ANYWHERE_WAKE_WORDS=assistant + +# ── listening (local voice-activity detection) ─────────────────────────────── +# Nothing leaves the machine until the detector hears speech. Raise +# VAD_RMS_THRESHOLD if background noise keeps getting transcribed; lower it if +# you have to shout. +# VAD_RMS_THRESHOLD=500 +# VAD_START_FRAMES=3 # loud 80 ms frames in a row before speech starts +# VAD_SILENCE_END_SEC=0.8 # quiet seconds that end an utterance +# MIN_UTTERANCE_SEC=0.35 # shorter bursts are never sent +# MAX_UTTERANCE_SEC=12 # the longest utterance sent +# COMMAND_WAIT_SEC=8 # after a bare "Hey Bolt", how long for the request +# FOLLOW_UP_SEC=8 # when the reply ends on a question, keep listening +# FOLLOW_UP_MAX_TURNS=5 +# CHIMES=true + +# ── what Bolt may do on this machine ───────────────────────────────────────── +# There's no screen to approve things on, so ASSISTANT_COMMANDS decides: +# off nothing runs +# read (default) filectl list/read only +# all everything, shell commands and file writes included, like Bolt-Pet +ASSISTANT_COMMANDS=read +# COMMAND_TIMEOUT_SECONDS=60 + +# ── heartbeat: reminders and delivered files ───────────────────────────────── +# Every HEARTBEAT_SECONDS the assistant reports in to the server — the channel +# Bolt's reminders and any files he delivers to this client arrive on. +# Announcements are spoken when the mic is idle, and files land in +# BOLT_DOWNLOAD_DIR (default ~/Downloads/Bolt) without overwriting what is +# already there. Set to 0 to turn the heartbeat off. +HEARTBEAT_SECONDS=60 +# BOLT_DOWNLOAD_DIR= diff --git a/.gitignore b/.gitignore index f6664e3..50ec8e6 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,4 @@ __pycache__/ *.pyc .env .pytest_cache/ +.claude/ diff --git a/README.md b/README.md index c190498..048101d 100644 --- a/README.md +++ b/README.md @@ -45,6 +45,11 @@ other than the system default. `run.bat --type` lets you type instead of speak, 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. +5. **Reminders and delivered files ride a heartbeat.** Every `HEARTBEAT_SECONDS` (default 60) the + assistant reports in to `/desk/report_status`, which is how Bolt's reminders and any files he + sends to this client arrive — the same channel Bolt-Pet uses. Announcements are spoken when the + mic is idle, and files download to `BOLT_DOWNLOAD_DIR` (default `~/Downloads/Bolt`), never + overwriting what's already there. Set `HEARTBEAT_SECONDS=0` to turn it off. ## Cost and privacy @@ -85,6 +90,9 @@ See `.env.example`. The ones you're most likely to change: (default `assistant`: "what's the weather, assistant?"). Blank to turn off. - `COMMAND_WAIT_SEC`, `FOLLOW_UP_SEC`: how long it waits after "Hey Bolt", or after asking you something. +- `HEARTBEAT_SECONDS`: how often it checks in for reminders and delivered files (default 60; + `0` turns the heartbeat off). +- `BOLT_DOWNLOAD_DIR`: where delivered files land (default `~/Downloads/Bolt`). It has its own conversation on the server (`pet-assistant-`), which shows up in Bolt Console's Chats list. diff --git a/bolt_assistant/__main__.py b/bolt_assistant/__main__.py index 41290c6..ba77254 100644 --- a/bolt_assistant/__main__.py +++ b/bolt_assistant/__main__.py @@ -99,6 +99,25 @@ def main(argv=None) -> int: return 0 stop = threading.Event() + from . import heartbeat + + heartbeat.fetch_server_health() + + def _on_heartbeat(result: dict) -> None: + # What the server hands out is popped the moment it is handed over, so + # everything goes into the spoken queue — announced when the mic is idle. + if result.get("announcement"): + assistant.announce(result["announcement"]) + saved = result.get("saved") or [] + if saved: + names = ", ".join(path.name for path in saved) + folder = str(saved[0].parent) + assistant.announce( + f"Saved {names} to {folder}." + if len(saved) <= 3 else f"Saved {len(saved)} files to {folder}.") + + threading.Thread(target=heartbeat.heartbeat_loop, args=(stop, _on_heartbeat), + name="bolt-heartbeat", daemon=True).start() try: mic = audio.open_mic(assistant.on_mic_frame) except Exception as exc: diff --git a/bolt_assistant/assistant.py b/bolt_assistant/assistant.py index 09f0a05..49f8293 100644 --- a/bolt_assistant/assistant.py +++ b/bolt_assistant/assistant.py @@ -20,6 +20,10 @@ background thread, so the mic keeps being segmented meanwhile and whatever is said next becomes its own utterance and is sent off too. Finished transcripts are delivered oldest-first between mic frames. +Announcements (Bolt's reminders, delivered-file notices) ride the heartbeat +thread and wait in a queue; the run loop speaks them one at a time whenever +it is idle, then flushes the echo of Bolt's own voice like after a reply. + 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. """ @@ -115,6 +119,7 @@ class Assistant: self._mute_until = 0.0 self._voice = "" # a voice the server picked (speak_as) sticks for later replies self._tts_failed = False + self._announcements: "queue.Queue[str]" = queue.Queue() # heartbeat notices, spoken when idle # ── audio in ───────────────────────────────────────────────────────── @@ -135,6 +140,7 @@ class Assistant: else: self.process(frame) self.handle_transcripts() + self._drain_announcements() def listening_banner(self) -> str: if not self.wake_required: @@ -208,6 +214,29 @@ class Assistant: self.log(" (stopped waiting)") self.state = "idle" + # ── heartbeat announcements ────────────────────────────────────────── + + def announce(self, text: str) -> None: + """Queue something to say once the room is free — a reminder Bolt + fired, a delivered-file notice. Never interrupts: it waits for idle.""" + text = (text or "").strip() + if text: + self._announcements.put(text) + + def _drain_announcements(self) -> None: + """Say what the heartbeat brought, one at a time, while idle — never + mid-sentence, while a transcript is still in flight, or during the + echo guard after Bolt's own voice.""" + if (self.state != "idle" or self._pending or self.segmenter.in_speech + or self.clock() < self._mute_until): + return + try: + text = self._announcements.get_nowait() + except queue.Empty: + return + self.speak(text) + self._back_to_listening("idle") # flush the mic's echo of Bolt's own voice + # ── what was said ──────────────────────────────────────────────────── def heard(self, text: str) -> None: diff --git a/bolt_assistant/config.py b/bolt_assistant/config.py index 6eba79c..90938b6 100644 --- a/bolt_assistant/config.py +++ b/bolt_assistant/config.py @@ -50,6 +50,17 @@ 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") +# ── heartbeat: reminders and delivered files ───────────────────────────────── +# Every HEARTBEAT_SECONDS the assistant reports in to /desk/report_status — +# the channel Bolt's reminders and any files he delivers to this client +# arrive on. Announcements are spoken when the mic is idle (see +# assistant.announce), and files download to DOWNLOAD_DIR without +# overwriting what is already there. Set HEARTBEAT_SECONDS to 0 to turn the +# heartbeat off entirely. +HEARTBEAT_SECONDS = max(0, int(os.environ.get("HEARTBEAT_SECONDS", "60"))) +DOWNLOAD_DIR = Path(os.environ.get("BOLT_DOWNLOAD_DIR", "").strip() + or str(Path.home() / "Downloads" / "Bolt")) + # ── audio devices ─────────────────────────────────────────────────────────── MIC_DEVICE = _device(os.environ.get("MIC_DEVICE", "")) diff --git a/bolt_assistant/heartbeat.py b/bolt_assistant/heartbeat.py new file mode 100644 index 0000000..fea6356 --- /dev/null +++ b/bolt_assistant/heartbeat.py @@ -0,0 +1,193 @@ +"""The heartbeat that brings Bolt's reminders and delivered files in. + +Same protocol as Bolt-Pet's heartbeat (desk_client/bolt_desk.py): + + POST /desk/report_status {"session_id", "download_dir"} + -> {"ok": true} or {"reply": ""} (plus + "suggestions" when there are any) + GET /desk/files?session_id=... -> {"files": [{"id", "name", "size"}]} + GET /desk/files/?session_id=...-> the bytes; the server *dequeues* + on this GET, so write to a .part first and rename only once the + whole file is here, and never overwrite an existing file. + +A heartbeat thread is the only thing standing between a delivered file and +this machine, so every failure is logged and swallowed: one bad beat must +never stop the loop, and a refused key gets a loud warning once at startup +instead of silent nothing-ever-arrives. + +Announcements returned by report_status are popped server-side the moment +they are handed out, so the caller (__main__) feeds them to +assistant.announce() rather than dropping them. Everything network-shaped +is injectable for the tests. +""" + +from __future__ import annotations + +import os +import re +import time +from pathlib import Path +from typing import Callable, List, Optional + +import requests + +from . import config +from .server import _headers + +log: Callable[[str], None] = print + + +def _url(path: str) -> str: + return f"{config.SERVER_URL}{path}" + + +def _safe_filename(name: str) -> str: + """A server-sent name is a display hint, not a trusted path: strip path + separators, control characters and leading dots before it touches disk.""" + cleaned = re.sub(r'[\\/:*?"<>|\x00-\x1f]', "_", str(name or "")).strip().strip(".") + return cleaned or "download" + + +def _free_path(folder: Path, name: str) -> Path: + """A name that doesn't exist yet: the server never overwrites, and neither + do we. "note (2).txt" after "note.txt" and "note (1).txt", like a browser.""" + target = folder / name + if not target.exists(): + return target + stem, suffix = target.stem, target.suffix + for n in range(1, 1000): + candidate = folder / f"{stem} ({n}){suffix}" + if not candidate.exists(): + return candidate + return folder / f"{stem} ({int(time.time())}){suffix}" + + +def fetch_server_health(timeout: float = 10.0, get: Callable = requests.get) -> None: + """One keyed probe at startup. A refusal gets a loud warning — a dead + DESK_API_KEY means total silence (no answers, no files, no reminders) + with no other symptom.""" + try: + response = get(_url("/desk/health"), headers=_headers(), timeout=timeout) + info = response.json() + log(f"server: {info}") + if info.get("key_ok") is False: + log("!! the server refused this device's DESK_API_KEY — no files, no reminders,") + log("!! and no answers will reach this assistant until it is updated") + log("!! set DESK_API_KEY in .env — it is re-read on the next request,") + log("!! so no restart is needed once the file is corrected") + except Exception as exc: + log(f"warning: server not reachable yet ({exc}) — will keep trying per-request") + + +def _download(file_id: str, name: str, folder: Path, + get: Callable = requests.get) -> Optional[Path]: + """Fetch one queued file and save it. The server dequeues on this GET, so + write to a .part first and rename only once the bytes are all here.""" + try: + response = get( + _url(f"/desk/files/{file_id}"), + params={"session_id": config.SESSION_ID}, headers=_headers(), + timeout=(10, 120), stream=True, + ) + if response.status_code != 200: + log(f"file download refused ({response.status_code}) for {name}") + return None + folder.mkdir(parents=True, exist_ok=True) + target = _free_path(folder, _safe_filename(name)) + part = target.with_name(target.name + ".part") + try: + with open(part, "wb") as handle: + for chunk in response.iter_content(65536): + if chunk: + handle.write(chunk) + os.replace(part, target) + except Exception: + try: + part.unlink() + except OSError: + pass + raise + return target + except Exception as exc: + log(f"file download failed for {name}: {exc}") + return None + + +def _fetch_outbox(folder: Path, get: Callable = requests.get) -> List[Path]: + try: + response = get( + _url("/desk/files"), + params={"session_id": config.SESSION_ID}, headers=_headers(), + timeout=(5, 30), + ) + except Exception as exc: + log(f"couldn't list queued files: {exc}") + return [] + if response.status_code != 200: + return [] + try: + body = response.json() + except ValueError: + return [] + entries = body.get("files") if isinstance(body, dict) else None + saved: List[Path] = [] + for entry in entries or []: + if not isinstance(entry, dict) or not entry.get("id"): + continue + target = _download(str(entry["id"]), str(entry.get("name") or "download"), folder, get) + if target is not None: + saved.append(target) + return saved + + +def heartbeat_once(download_dir=None, post: Callable = requests.post, + get: Callable = requests.get) -> dict: + """One beat: report in (reminders, files and presence all flow), then + collect whatever the server is holding for this device. Returns + {"ok", "announcement", "saved"} — *announcement* is something Bolt wants + spoken, *saved* the Paths just downloaded.""" + folder = Path(download_dir) if download_dir is not None else config.DOWNLOAD_DIR + result: dict = {"ok": False, "announcement": "", "saved": []} + try: + folder.mkdir(parents=True, exist_ok=True) + except OSError as exc: + log(f"can't create the download folder {folder}: {exc}") + payload = {"session_id": config.SESSION_ID, "download_dir": str(folder)} + try: + response = post(_url("/desk/report_status"), json=payload, + headers=_headers(), timeout=20) + except Exception as exc: + log(f"heartbeat failed: {exc}") + return result + if response.status_code != 200: + return result + try: + body = response.json() + except ValueError: + body = {} + announcement = body.get("reply") if isinstance(body, dict) else None + if isinstance(announcement, str) and announcement.strip(): + result["announcement"] = " ".join(announcement.split()) + result["saved"] = _fetch_outbox(folder, get) + if result["saved"]: + names = ", ".join(path.name for path in result["saved"]) + log(f"saved {len(result['saved'])} file(s) to {folder}: {names}") + result["ok"] = True + return result + + +def heartbeat_loop(stop, on_result: Optional[Callable[[dict], None]] = None) -> None: + """Beat every HEARTBEAT_SECONDS until *stop* is set, handing each result + to *on_result*. A bad iteration is logged, never fatal; a typo in the + interval must not become a request flood (10 s floor).""" + if config.HEARTBEAT_SECONDS <= 0: + return + interval = max(10, config.HEARTBEAT_SECONDS) + while not stop.is_set(): + try: + result = heartbeat_once() + if on_result is not None: + on_result(result) + except Exception as exc: + log(f"heartbeat iteration failed: {exc}") + stop.wait(interval) diff --git a/tests/test_assistant.py b/tests/test_assistant.py index e99c9d4..dc3f660 100644 --- a/tests/test_assistant.py +++ b/tests/test_assistant.py @@ -411,3 +411,55 @@ def test_transcription_falls_back_to_one_upload(): 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 + + +# ── heartbeat announcements ────────────────────────────────────────────────── + +def test_announcements_are_spoken_when_idle(): + a, asked, played, logs, _ = make([]) + a.announce("You have a reminder.") + a._drain_announcements() + assert "You have a reminder." in spoken(played) + assert a.state == "idle" + + +def test_announcements_wait_for_idle(): + a, asked, played, logs, _ = make(["It's three o'clock."]) + a.state = "busy" + a.announce("You have a reminder.") + a._drain_announcements() + assert spoken(played) == [] # held back while Bolt is talking + a.state = "idle" + a._drain_announcements() + assert "You have a reminder." in spoken(played) + + +def test_announcements_hold_while_a_transcript_is_in_flight(): + a, asked, played, logs, _ = make([]) + a._pending.append(object()) # an utterance is still at the server + a.announce("You have a reminder.") + a._drain_announcements() + assert spoken(played) == [] + a._pending.clear() + a._drain_announcements() + assert "You have a reminder." in spoken(played) + + +def test_announcements_hold_while_someone_is_speaking(): + a, asked, played, logs, _ = make([]) + FakeTranscription.script.append("") # the start event still mints a transcriber + for f in [frame(LOUD)] * 3: + a.process(f) # the segmenter is mid-utterance + a.announce("You have a reminder.") + a._drain_announcements() + assert spoken(played) == [] + a.segmenter.reset() + a._drain_announcements() + assert "You have a reminder." in spoken(played) + + +def test_blank_announcements_are_dropped(): + a, asked, played, logs, _ = make([]) + a.announce(" ") + a._drain_announcements() + assert spoken(played) == [] diff --git a/tests/test_heartbeat.py b/tests/test_heartbeat.py new file mode 100644 index 0000000..6eb6779 --- /dev/null +++ b/tests/test_heartbeat.py @@ -0,0 +1,191 @@ +"""The heartbeat that brings reminders and delivered files in — with fakes +for the network. Covers the report_status POST, the outbox list, the +download rules (.part first, never overwrite, clean up on failure) and the +loop's stop/disable behaviour.""" + +import sys +import threading +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import pytest + +from bolt_assistant import config, heartbeat + + +class FakeResponse: + def __init__(self, payload=None, data=b"", status_code=200): + self._payload = payload + self._data = data + self.status_code = status_code + + def json(self): + return self._payload + + def iter_content(self, chunk_size=65536): + yield self._data + + +@pytest.fixture(autouse=True) +def _quiet(monkeypatch): + monkeypatch.setattr(heartbeat, "log", lambda msg: None) + + +def _files_get(data=b"hello", files=None, status_code=200): + def get(url, params=None, headers=None, timeout=None, stream=None): + if url.endswith("/desk/files"): + return FakeResponse(payload={"files": files or []}) + return FakeResponse(data=data, status_code=status_code) + + return get + + +# ── one beat ───────────────────────────────────────────────────────────────── + +def test_heartbeat_once_reports_in_and_returns_ok(tmp_path): + seen = {} + + def post(url, json=None, headers=None, timeout=None): + seen["url"], seen["json"], seen["headers"] = url, json, headers + return FakeResponse(payload={"ok": True}) + + result = heartbeat.heartbeat_once(tmp_path, post=post, get=_files_get()) + assert result == {"ok": True, "announcement": "", "saved": []} + assert seen["url"].endswith("/desk/report_status") + assert seen["json"] == {"session_id": config.SESSION_ID, "download_dir": str(tmp_path)} + assert seen["headers"] == {"X-Desk-Api-Key": config.API_KEY} + + +def test_heartbeat_once_carries_the_announcement(tmp_path): + def post(url, json=None, headers=None, timeout=None): + return FakeResponse(payload={"reply": " You have a reminder. "}) + + result = heartbeat.heartbeat_once(tmp_path, post=post, get=_files_get()) + assert result["announcement"] == "You have a reminder." + + +def test_heartbeat_once_downloads_the_outbox(tmp_path): + seen = {} + + def get(url, params=None, headers=None, timeout=None, stream=None): + if url.endswith("/desk/files"): + seen["list"] = params + return FakeResponse(payload={"files": [{"id": "abc", "name": "note.txt", "size": 5}]}) + seen["download"] = params + return FakeResponse(data=b"hello") + + def post(url, json=None, headers=None, timeout=None): + return FakeResponse(payload={"ok": True}) + + result = heartbeat.heartbeat_once(tmp_path, post=post, get=get) + assert seen["list"] == seen["download"] == {"session_id": config.SESSION_ID} + assert len(result["saved"]) == 1 + target = result["saved"][0] + assert target.parent == tmp_path and target.name == "note.txt" + assert target.read_bytes() == b"hello" + assert not list(tmp_path.glob("*.part")) + + +def test_heartbeat_once_network_failure_is_not_fatal(tmp_path): + def post(url, json=None, headers=None, timeout=None): + raise OSError("no network") + + result = heartbeat.heartbeat_once(tmp_path, post=post, get=_files_get()) + assert result == {"ok": False, "announcement": "", "saved": []} + + +def test_heartbeat_once_refused_does_not_touch_the_outbox(tmp_path): + got = [] + + def get(url, params=None, headers=None, timeout=None, stream=None): + got.append(url) + return FakeResponse(payload={"files": []}) + + def post(url, json=None, headers=None, timeout=None): + return FakeResponse(payload={"error": "unauthorized"}, status_code=401) + + result = heartbeat.heartbeat_once(tmp_path, post=post, get=get) + assert result == {"ok": False, "announcement": "", "saved": []} + assert got == [] + + +# ── download rules ─────────────────────────────────────────────────────────── + +def test_download_never_overwrites(tmp_path): + (tmp_path / "note.txt").write_text("old") + target = heartbeat._download("abc", "note.txt", tmp_path, get=_files_get(data=b"new")) + assert target.name == "note (1).txt" + assert (tmp_path / "note.txt").read_text() == "old" + assert target.read_bytes() == b"new" + + +def test_download_refused(tmp_path): + assert heartbeat._download("abc", "note.txt", tmp_path, + get=_files_get(status_code=500)) is None + assert list(tmp_path.iterdir()) == [] + + +def test_download_cleans_up_the_part_file_on_failure(tmp_path): + class Exploding(FakeResponse): + def iter_content(self, chunk_size=65536): + yield b"part of" + raise OSError("connection reset") + + assert heartbeat._download("abc", "note.txt", tmp_path, + get=lambda **kw: Exploding()) is None + assert list(tmp_path.iterdir()) == [] # no .part left behind + + +def test_safe_filename_strips_paths_and_control_characters(): + assert heartbeat._safe_filename("a/b\\c:d.txt") == "a_b_c_d.txt" + assert heartbeat._safe_filename("..") == "download" + assert heartbeat._safe_filename("") == "download" + assert heartbeat._safe_filename(" fine name.txt ") == "fine name.txt" + + +# ── health probe and the loop ──────────────────────────────────────────────── + +def test_fetch_server_health_warns_loudly_on_a_refused_key(monkeypatch): + lines = [] + monkeypatch.setattr(heartbeat, "log", lines.append) + + def get(url, headers=None, timeout=None): + return FakeResponse(payload={"key_ok": False}) + + heartbeat.fetch_server_health(get=get) + assert any("DESK_API_KEY" in line for line in lines) + + +def test_fetch_server_health_logs_a_healthy_server(monkeypatch): + lines = [] + monkeypatch.setattr(heartbeat, "log", lines.append) + + def get(url, headers=None, timeout=None): + return FakeResponse(payload={"key_ok": True, "version": "1"}) + + heartbeat.fetch_server_health(get=get) + assert lines and lines[0].startswith("server:") + + +def test_heartbeat_loop_reports_each_beat(monkeypatch): + monkeypatch.setattr(config, "HEARTBEAT_SECONDS", 60) + stop = threading.Event() + results = [] + monkeypatch.setattr(heartbeat, "heartbeat_once", + lambda: {"ok": True, "announcement": "hi", "saved": []}) + + def on_result(result): + results.append(result) + stop.set() + + heartbeat.heartbeat_loop(stop, on_result) + assert results == [{"ok": True, "announcement": "hi", "saved": []}] + + +def test_heartbeat_loop_disabled_when_heartbeat_seconds_is_zero(monkeypatch): + monkeypatch.setattr(config, "HEARTBEAT_SECONDS", 0) + called = [] + monkeypatch.setattr(heartbeat, "heartbeat_once", lambda: called.append(1)) + heartbeat.heartbeat_loop(threading.Event(), None) + assert called == []