README: document Plan 3 Pi client surface

README.md

@@ -137,10 +137,77 @@ PY
 Don't send `wake` from the smoke — it would open a real OpenAI session and
 burn budget. Plan 3 exercises that end-to-end.
 
+## Plan 3: Pi client
+
+The Python client lives in `client/`. It is shipped to the Pi inside
+`client.tar.gz` (built into the backend Docker image) and installed by
+`install.sh`. Modules:
+
+- `main.py` — entry point. Sets `PA_ALSA_PLUGHW=1` before importing
+  sounddevice. Wires audio, state, wakeword, session, and playback.
+- `state.py` — `IDLE → WAKE_PENDING → LISTENING ↔ ASSISTANT_SPEAKING →
+  IDLE_PENDING → IDLE`. `wakeword_enabled` is true only in `IDLE`;
+  `uplink_enabled` is true only in `LISTENING`.
+- `audio.py` — USB device discovery, persistent 24 kHz/mono/int16 InputStream
+  and OutputStream, and the 24 → 16 kHz resample for the wakeword feed.
+- `wakeword.py` — openwakeword `alexa` model, threshold + cooldown.
+- `playback.py` — single worker thread that owns the OutputStream and
+  implements `set_volume` as a software gain (the USB Speaker Phone has no
+  hardware playback volume control; only a `PCM Playback Switch`).
+- `session.py` — backend WS client, dispatch, tool round-trip,
+  exponential-backoff reconnect.
+- `config.py` — `~/assistant/state/config.json` (mode 0600).
+- `pair.py` — `python -m client.pair --backend <url> [--code <code>]`.
+
+### Updating the Pi
+
+For a fast iteration loop:
+
+```sh
+sshpass -p 'assistant' rsync -az --delete --exclude __pycache__ --exclude tests \
+  client/ pi@192.168.50.115:assistant/code/client/
+sshpass -p 'assistant' ssh pi@192.168.50.115 'systemctl --user restart assistant'
+sshpass -p 'assistant' ssh pi@192.168.50.115 \
+  'journalctl --user -u assistant -n 50 --no-pager'
+```
+
+For a production update (after a backend deploy):
+
+```sh
+sshpass -p 'assistant' ssh pi@192.168.50.115 \
+  'curl -fsSL https://assistant.volcanic.tes.gd/install.sh | bash'
+```
+
+### Local tests
+
+```sh
+python3 -m venv .venv
+.venv/bin/pip install -r requirements-dev.txt requests websockets numpy scipy
+.venv/bin/python -m pytest client/tests
+```
+
+Hardware-touching modules (`audio.py` streams, `wakeword.py`, `playback.py`,
+`session.run_forever`) are smoke-tested via the live deploy. Deterministic
+modules (`state.py`, `config.py`, `pair.py`, the resample helper, the session
+dispatcher) are covered by the pytest suite.
+
+### End-to-end smoke
+
+With the unit running on the Pi:
+
+```sh
+sshpass -p 'assistant' ssh pi@192.168.50.115 'journalctl --user -u assistant -f'
+```
+
+…then say `alexa` near the Speaker Phone. Expected log sequence: `wakeword
+fired`, `state IDLE -> WAKE_PENDING`, `session_started conversation_id=…`,
+`state WAKE_PENDING -> LISTENING`. Speak a question; the assistant replies
+through the speaker (`state LISTENING -> ASSISTANT_SPEAKING`, then
+`assistant_done`). Say "bye" → `session_ended reason=tool`, then `state
+LISTENING -> IDLE_PENDING -> IDLE`. The conversation row + turns are visible
+via the deployed backend's DB (admin UI in Plan 4).
+
 ## What's next
 
-- **Plan 3** — full Python client on the Pi (wakeword, VAD, audio plumbing,
-  state machine, install + pair CLIs). This is the production caller of
-  `/device` and the integration test for everything above.
 - **Plan 4** — React management UI for the admin dashboard, per-device
   config editor, and conversation history viewer.

client/audio.py

@@ -0,0 +1,139 @@
+"""Audio helpers for the Pi client.
+
+This module is split into two layers:
+
+* Deterministic helpers (no hardware) — `resample_24k_to_16k`. Tested.
+* `AudioStreams` — owns the persistent PortAudio InputStream / OutputStream.
+  Smoke-tested by `main.py` running on the Pi.
+"""
+import queue
+import sys
+from dataclasses import dataclass, field
+from typing import Optional
+
+import numpy as np
+from scipy.signal import resample_poly
+
+from client.log import get_logger
+
+_log = get_logger("audio")
+
+SAMPLE_RATE = 24_000
+FRAME_SAMPLES = 1920            # 80 ms at 24 kHz
+CHANNELS = 1
+DTYPE = "int16"
+BYTES_PER_FRAME = FRAME_SAMPLES * 2  # int16 -> 2 bytes
+
+
+def resample_24k_to_16k(frame: np.ndarray) -> np.ndarray:
+    """Downsample a 24 kHz int16 mono frame to 16 kHz int16 mono (2:3 poly).
+
+    Input length is preserved as `len(frame) * 2 // 3`. For the 1920-sample
+    frames produced by our 80 ms blocksize at 24 kHz, this yields 1280 samples
+    (80 ms at 16 kHz), which is exactly what openwakeword expects.
+    """
+    if frame.dtype != np.int16:
+        raise TypeError(f"expected int16, got {frame.dtype}")
+    if frame.size == 0:
+        raise ValueError("empty input")
+    floats = frame.astype(np.float32)
+    resampled = resample_poly(floats, up=2, down=3)
+    return np.clip(resampled, -32768, 32767).astype(np.int16)
+
+
+def find_usb_device(sd) -> int:
+    """Return the PortAudio device index of the USB Speaker Phone.
+
+    `sd` is the imported `sounddevice` module — passed in so the deferred
+    import of sounddevice (which requires PA_ALSA_PLUGHW already set in env)
+    stays in `main.py`.
+    """
+    devices = sd.query_devices()
+    for idx, dev in enumerate(devices):
+        name = dev["name"].lower()
+        if "usb" in name and dev["max_input_channels"] >= 1:
+            return idx
+    print("USB audio device not found. Devices:", file=sys.stderr)
+    print(devices, file=sys.stderr)
+    raise SystemExit("no USB audio device found")
+
+
+@dataclass
+class AudioStreams:
+    """Wrapper around persistent 24 kHz/mono/int16 InputStream + OutputStream.
+
+    The InputStream is callback-driven; each 1920-sample frame is pushed onto
+    `mic_queue` (a `queue.Queue[bytes]`). The bridge thread drains it. The
+    OutputStream is opened but writes are driven by `PlaybackWorker`.
+    """
+    sd: object                              # sounddevice module
+    device_index: int
+    mic_queue: queue.Queue
+    overflow_counter: int = 0
+    _in_stream: Optional[object] = field(default=None, repr=False)
+    _out_stream: Optional[object] = field(default=None, repr=False)
+
+    @classmethod
+    def open(cls, sd, mic_queue: queue.Queue) -> "AudioStreams":
+        device = find_usb_device(sd)
+        _log.info("opening audio on device %d: %r", device, sd.query_devices(device)["name"])
+
+        streams = cls(sd=sd, device_index=device, mic_queue=mic_queue)
+
+        def _on_input(indata, frames, time_info, status):  # PortAudio thread
+            if status:
+                streams.overflow_counter += 1
+                _log.warning("input status: %s", status)
+            mono = indata[:, 0].tobytes()
+            try:
+                mic_queue.put_nowait(mono)
+            except queue.Full:
+                try:
+                    mic_queue.get_nowait()
+                except queue.Empty:
+                    pass
+                try:
+                    mic_queue.put_nowait(mono)
+                except queue.Full:
+                    pass
+                _log.warning("mic_queue full; dropped oldest frame")
+
+        streams._in_stream = sd.InputStream(
+            samplerate=SAMPLE_RATE,
+            channels=CHANNELS,
+            dtype=DTYPE,
+            device=device,
+            blocksize=FRAME_SAMPLES,
+            callback=_on_input,
+        )
+        streams._out_stream = sd.OutputStream(
+            samplerate=SAMPLE_RATE,
+            channels=CHANNELS,
+            dtype=DTYPE,
+            device=device,
+            blocksize=FRAME_SAMPLES,
+        )
+        return streams
+
+    def start(self) -> None:
+        self._in_stream.start()
+        self._out_stream.start()
+
+    def write(self, pcm16_bytes: bytes) -> None:
+        """Blocking write to the OutputStream (from the playback worker thread)."""
+        if not pcm16_bytes:
+            return
+        buf = np.frombuffer(pcm16_bytes, dtype=np.int16)
+        self._out_stream.write(buf)
+
+    def close(self) -> None:
+        for s in (self._in_stream, self._out_stream):
+            if s is not None:
+                try:
+                    s.stop()
+                except Exception:
+                    pass
+                try:
+                    s.close()
+                except Exception:
+                    pass

client/config.py

@@ -0,0 +1,49 @@
+"""Read / write the device config file at ~/assistant/state/config.json."""
+import json
+import os
+from dataclasses import asdict, dataclass
+from pathlib import Path
+
+
+class ConfigMissingError(RuntimeError):
+    """Raised when config.json is missing or malformed."""
+
+
+@dataclass(frozen=True)
+class Config:
+    device_id: str
+    device_token: str
+    backend_ws: str
+
+
+def config_path() -> Path:
+    return Path(os.environ["HOME"]) / "assistant" / "state" / "config.json"
+
+
+def load_config() -> Config:
+    path = config_path()
+    if not path.exists():
+        raise ConfigMissingError(f"{path} not found; run `python -m client.pair` first")
+    try:
+        raw = json.loads(path.read_text())
+    except json.JSONDecodeError as exc:
+        raise ConfigMissingError(f"{path} is not valid JSON: {exc}") from exc
+    try:
+        return Config(
+            device_id=str(raw["device_id"]),
+            device_token=str(raw["device_token"]),
+            backend_ws=str(raw["backend_ws"]),
+        )
+    except KeyError as exc:
+        raise ConfigMissingError(f"{path} missing field: {exc}") from exc
+
+
+def save_config(cfg: Config) -> None:
+    path = config_path()
+    path.parent.mkdir(parents=True, exist_ok=True)
+    # Write to a temp file then atomically rename, so a partial write never
+    # leaves a half-baked config on disk.
+    tmp = path.with_suffix(".json.tmp")
+    tmp.write_text(json.dumps(asdict(cfg), indent=2) + "\n")
+    os.chmod(tmp, 0o600)
+    os.replace(tmp, path)

client/log.py

@@ -0,0 +1,28 @@
+"""Structured stdout logging suitable for journald capture."""
+import logging
+import os
+import sys
+
+_CONFIGURED = False
+
+
+def configure(level: str | int | None = None) -> None:
+    """Configure root logging once. Safe to call multiple times."""
+    global _CONFIGURED
+    if _CONFIGURED:
+        return
+    if level is None:
+        level = os.environ.get("ASSISTANT_LOG_LEVEL", "INFO").upper()
+    logging.basicConfig(
+        level=level,
+        format="%(asctime)s %(levelname)-5s %(name)s %(message)s",
+        datefmt="%H:%M:%S",
+        stream=sys.stdout,
+    )
+    _CONFIGURED = True
+
+
+def get_logger(name: str) -> logging.Logger:
+    """Return a logger named `name`; auto-configure on first call."""
+    configure()
+    return logging.getLogger(name)

client/main.py

@@ -1,8 +1,166 @@
-"""Smart Assistant client entry point. Implemented in Plan 3."""
+"""Smart Assistant Pi client entry point.
+
+CRITICAL: set PA_ALSA_PLUGHW BEFORE importing sounddevice (see findings.md §1).
+"""
+import os
+os.environ.setdefault("PA_ALSA_PLUGHW", "1")
+
+import asyncio
+import queue
+import signal
+import sys
+import threading
+from pathlib import Path
+
+import numpy as np
+import sounddevice as sd  # safe to import now
+
+from client.audio import (
+    AudioStreams,
+    FRAME_SAMPLES,
+    resample_24k_to_16k,
+)
+from client.config import ConfigMissingError, load_config
+from client.log import get_logger
+from client.playback import PlaybackWorker
+from client.session import Session
+from client.state import StateMachine
+from client.wakeword import WakewordDetector
+
+_log = get_logger("main")
+
+ASSISTANT_STATE_DIR = Path(os.environ["HOME"]) / "assistant" / "state"
+WAKE_WAV = ASSISTANT_STATE_DIR / "wake.wav"
+SLEEP_WAV = ASSISTANT_STATE_DIR / "sleep.wav"
+
+
+def _build_bridge_thread(
+    mic_queue: queue.Queue,
+    sm: StateMachine,
+    wakeword: WakewordDetector,
+    session: Session,
+    playback: PlaybackWorker,
+    stop_event: threading.Event,
+) -> threading.Thread:
+    def run():
+        while not stop_event.is_set():
+            try:
+                pcm16_bytes = mic_queue.get(timeout=0.25)
+            except queue.Empty:
+                continue
+            if sm.wakeword_enabled:
+                frame = np.frombuffer(pcm16_bytes, dtype=np.int16)
+                if frame.size != FRAME_SAMPLES:
+                    continue
+                feed = resample_24k_to_16k(frame)
+                if wakeword.predict(feed):
+                    if sm.wake_fired():
+                        if WAKE_WAV.exists():
+                            playback.enqueue_wav(WAKE_WAV, "wake")
+                        else:
+                            # Even with no wav, fire the done marker so the
+                            # playback_done(for=wake) envelope is sent.
+                            playback.enqueue_done_marker("wake")
+                        session.call_soon_threadsafe_send_wake()
+            elif sm.uplink_enabled:
+                session.call_soon_threadsafe_send_binary(pcm16_bytes)
+            # else: ASSISTANT_SPEAKING / WAKE_PENDING / IDLE_PENDING -> drop
+    t = threading.Thread(target=run, name="bridge", daemon=True)
+    return t
+
+
+def _make_callbacks(sm: StateMachine, playback: PlaybackWorker):
+    class _CB:
+        def on_session_started(self, conversation_id: str) -> None:
+            _log.info("session_started conversation_id=%s", conversation_id)
+            sm.session_started()
+
+        def on_assistant_audio(self, pcm16: bytes) -> None:
+            sm.assistant_audio_arrived()
+            playback.enqueue_audio(pcm16)
+
+        def on_assistant_done(self) -> None:
+            _log.info("assistant_done")
+            sm.assistant_done()
+
+        def on_session_ended(self, reason: str) -> None:
+            _log.info("session_ended reason=%s", reason)
+            sm.session_ended()
+            if SLEEP_WAV.exists():
+                playback.enqueue_wav(SLEEP_WAV, "sleep")
+            else:
+                playback.enqueue_done_marker("sleep")
+    return _CB()
 
 
 def main() -> None:
-    raise SystemExit("client.main: placeholder; full implementation comes in Plan 3.")
+    try:
+        cfg = load_config()
+    except ConfigMissingError as exc:
+        raise SystemExit(str(exc)) from exc
+
+    _log.info("device_id=%s backend=%s", cfg.device_id, cfg.backend_ws)
+
+    mic_queue: queue.Queue = queue.Queue(maxsize=64)
+    streams = AudioStreams.open(sd, mic_queue=mic_queue)
+    streams.start()
+
+    sm = StateMachine()
+
+    session_ref: list[Session] = []
+
+    def on_playback_done(label: str) -> None:
+        _log.info("playback_done for=%s", label)
+        if not session_ref:
+            return
+        session = session_ref[0]
+        session.call_soon_threadsafe_send_playback_done(label)
+        if label == "sleep":
+            sm.sleep_played()
+
+    playback = PlaybackWorker(streams, on_done=on_playback_done)
+    playback.start()
+
+    wakeword = WakewordDetector()
+
+    callbacks = _make_callbacks(sm, playback)
+    session = Session(
+        backend_ws=cfg.backend_ws,
+        device_token=cfg.device_token,
+        device_id=cfg.device_id,
+        callbacks=callbacks,
+    )
+    session_ref.append(session)
+
+    def set_volume_handler(args: dict) -> dict:
+        level = int(args.get("level", -1))
+        playback.set_volume(level)
+        return {"ok": True, "level": level}
+
+    session.register_tool("set_volume", set_volume_handler)
+
+    stop_event = threading.Event()
+    bridge = _build_bridge_thread(mic_queue, sm, wakeword, session, playback, stop_event)
+    bridge.start()
+
+    def _shutdown(signum, _frame):
+        _log.info("signal %d received; shutting down", signum)
+        stop_event.set()
+        try:
+            playback.stop()
+        finally:
+            streams.close()
+        sys.exit(0)
+
+    signal.signal(signal.SIGTERM, _shutdown)
+    signal.signal(signal.SIGINT, _shutdown)
+
+    try:
+        asyncio.run(session.run_forever())
+    finally:
+        stop_event.set()
+        playback.stop()
+        streams.close()
 
 
 if __name__ == "__main__":

client/pair.py

@@ -1,8 +1,67 @@
-"""Smart Assistant pairing CLI. Implemented in Plan 3."""
+"""Pair this Pi with the backend.
+
+Usage:
+    python -m client.pair --backend https://backend.example.com [--code XYZ12345]
+
+Called from install.sh after it prompts the user for the code.
+"""
+import argparse
+import socket
+
+import requests
+
+from client.config import Config, save_config
+from client.log import get_logger
+
+_log = get_logger("pair")
+
+CLIENT_VERSION = "0.1.0"
+
+
+def pair(*, backend: str, code: str | None, hostname: str) -> None:
+    backend = backend.rstrip("/")
+    if not code:
+        code = input("Pairing code: ").strip()
+    if not code:
+        raise SystemExit("no pairing code supplied")
+
+    url = f"{backend}/api/pair"
+    payload = {"code": code, "hostname": hostname, "client_version": CLIENT_VERSION}
+    _log.info("POST %s code=%s hostname=%s", url, code, hostname)
+
+    try:
+        resp = requests.post(url, json=payload, timeout=30)
+    except requests.RequestException as exc:
+        raise SystemExit(f"pair request failed: {exc}") from exc
+
+    if resp.status_code != 200:
+        try:
+            err = resp.json().get("error", resp.text)
+        except ValueError:
+            err = resp.text
+        raise SystemExit(f"/api/pair returned {resp.status_code}: {err}")
+
+    body = resp.json()
+    cfg = Config(
+        device_id=str(body["device_id"]),
+        device_token=str(body["device_token"]),
+        backend_ws=str(body["backend_ws"]),
+    )
+    save_config(cfg)
+    _log.info("paired device %s; config saved", cfg.device_id)
 
 
 def main() -> None:
-    raise SystemExit("client.pair: placeholder; full implementation comes in Plan 3.")
+    parser = argparse.ArgumentParser(description="Pair this Pi with the backend.")
+    parser.add_argument("--backend", required=True, help="Backend base URL (https://...)")
+    parser.add_argument("--code", default=None, help="Pairing code (prompted if omitted)")
+    parser.add_argument(
+        "--hostname",
+        default=socket.gethostname(),
+        help="Hostname reported to the backend (defaults to socket.gethostname())",
+    )
+    args = parser.parse_args()
+    pair(backend=args.backend, code=args.code, hostname=args.hostname)
 
 
 if __name__ == "__main__":

client/playback.py

@@ -0,0 +1,117 @@
+"""Playback worker thread + wake/sleep cue + software volume.
+
+Items pushed to the queue:
+  ("audio", bytes)     - PCM16 LE 24 kHz mono data to write
+  ("wav", (Path, label)) - WAV file to read and play; triggers on_done(label) after
+  ("done_marker", str) - synthetic marker: after this is popped, call on_done(label)
+  ("stop", None)       - shut the worker down
+
+Volume is applied as a software gain in [0..1.0] on int16 buffers because the
+USB Speaker Phone has no hardware playback volume control (only `PCM Playback
+Switch`, a mute).
+"""
+import queue
+import threading
+import wave
+from pathlib import Path
+from typing import Callable, Tuple
+
+import numpy as np
+
+from client.audio import SAMPLE_RATE
+from client.log import get_logger
+
+_log = get_logger("playback")
+
+
+class PlaybackWorker:
+    def __init__(self, audio_streams, on_done: Callable[[str], None]):
+        """
+        `audio_streams.write(pcm16_bytes)` is the blocking write into the
+        persistent OutputStream. `on_done(label)` is invoked when a tagged
+        cue (e.g. "wake", "sleep") finishes playing.
+        """
+        self._streams = audio_streams
+        self._on_done = on_done
+        self._q: "queue.Queue[Tuple[str, object]]" = queue.Queue()
+        self._volume = 1.0
+        self._thread = threading.Thread(target=self._run, name="playback", daemon=True)
+        self._running = False
+
+    def start(self) -> None:
+        self._running = True
+        self._thread.start()
+
+    def stop(self) -> None:
+        self._running = False
+        self._q.put(("stop", None))
+        self._thread.join(timeout=2.0)
+
+    def set_volume(self, level: int) -> None:
+        """Set software gain. `level` is 0..100."""
+        if not 0 <= level <= 100:
+            raise ValueError("level must be 0..100")
+        self._volume = level / 100.0
+        _log.info("volume set to %d%%", level)
+
+    def enqueue_audio(self, pcm16_bytes: bytes) -> None:
+        self._q.put(("audio", pcm16_bytes))
+
+    def enqueue_wav(self, path: Path, label: str) -> None:
+        """Queue a WAV cue. When fully drained, on_done(label) fires."""
+        self._q.put(("wav", (path, label)))
+
+    def enqueue_done_marker(self, label: str) -> None:
+        """Fire on_done(label) once this marker is processed (after all prior audio)."""
+        self._q.put(("done_marker", label))
+
+    def _apply_gain(self, pcm16: bytes) -> bytes:
+        if self._volume == 1.0:
+            return pcm16
+        arr = np.frombuffer(pcm16, dtype=np.int16).astype(np.int32)
+        arr = (arr * int(self._volume * 1024)) >> 10
+        arr = np.clip(arr, -32768, 32767).astype(np.int16)
+        return arr.tobytes()
+
+    def _play_wav(self, path: Path, label: str) -> None:
+        if not path.exists():
+            _log.info("wav %s not present at %s; skipping", label, path)
+            self._on_done(label)
+            return
+        try:
+            with wave.open(str(path), "rb") as w:
+                if w.getframerate() != SAMPLE_RATE or w.getnchannels() != 1 or w.getsampwidth() != 2:
+                    _log.warning(
+                        "wav %s is %dHz/%dch/%dB; expected %dHz/mono/16-bit",
+                        path, w.getframerate(), w.getnchannels(), w.getsampwidth() * 8, SAMPLE_RATE,
+                    )
+                    self._on_done(label)
+                    return
+                data = w.readframes(w.getnframes())
+        except wave.Error as exc:
+            _log.warning("could not read wav %s: %s", path, exc)
+            self._on_done(label)
+            return
+        chunk = 3840
+        for i in range(0, len(data), chunk):
+            self._streams.write(self._apply_gain(data[i:i + chunk]))
+        self._on_done(label)
+
+    def _run(self) -> None:
+        while self._running:
+            try:
+                kind, payload = self._q.get(timeout=0.5)
+            except queue.Empty:
+                continue
+            try:
+                if kind == "stop":
+                    return
+                if kind == "audio":
+                    self._streams.write(self._apply_gain(payload))
+                elif kind == "wav":
+                    path, label = payload
+                    self._play_wav(path, label)
+                elif kind == "done_marker":
+                    self._on_done(payload)
+            except Exception as exc:  # pragma: no cover - hardware path
+                _log.exception("playback worker error: %s", exc)

client/session.py

@@ -0,0 +1,211 @@
+"""Backend WebSocket client (asyncio)."""
+import asyncio
+import json
+import random
+from typing import Callable, Protocol
+
+from client.log import get_logger
+
+_log = get_logger("session")
+
+CLIENT_VERSION = "0.1.0"
+PING_INTERVAL_S = 15.0
+RECONNECT_MIN_S = 1.0
+RECONNECT_MAX_S = 30.0
+
+
+class Callbacks(Protocol):
+    def on_session_started(self, conversation_id: str) -> None: ...
+    def on_assistant_audio(self, pcm16: bytes) -> None: ...
+    def on_assistant_done(self) -> None: ...
+    def on_session_ended(self, reason: str) -> None: ...
+
+
+class Session:
+    def __init__(
+        self,
+        backend_ws: str,
+        device_token: str,
+        device_id: str,
+        callbacks: Callbacks,
+    ) -> None:
+        self._backend_ws = backend_ws
+        self._device_token = device_token
+        self._device_id = device_id
+        self._cb = callbacks
+        self._tools: dict[str, Callable[[dict], dict]] = {}
+        self._ws = None
+        self._loop: asyncio.AbstractEventLoop | None = None
+
+    def register_tool(self, name: str, handler: Callable[[dict], dict]) -> None:
+        self._tools[name] = handler
+
+    async def run_forever(self) -> None:
+        """Connect, pump, reconnect with exponential backoff. Never returns
+        cleanly except on cancellation."""
+        # Lazy import so the module is importable without `websockets`.
+        import websockets
+
+        self._loop = asyncio.get_running_loop()
+        backoff = RECONNECT_MIN_S
+        while True:
+            try:
+                _log.info("connecting to %s", self._backend_ws)
+                async with websockets.connect(
+                    self._backend_ws,
+                    additional_headers={"Authorization": f"Bearer {self._device_token}"},
+                    max_size=4 * 1024 * 1024,
+                ) as ws:
+                    self._ws = ws
+                    backoff = RECONNECT_MIN_S
+                    ping_task = asyncio.create_task(self._ping_loop(ws))
+                    try:
+                        await self._pump(ws)
+                    finally:
+                        ping_task.cancel()
+                        self._ws = None
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:  # noqa: BLE001
+                _log.warning("session error: %s; reconnecting in %.1fs", exc, backoff)
+            jitter = random.uniform(0, backoff * 0.1)
+            await asyncio.sleep(backoff + jitter)
+            backoff = min(backoff * 2.0, RECONNECT_MAX_S)
+
+    async def _pump(self, ws) -> None:
+        # 1. Send hello.
+        await ws.send(json.dumps({
+            "type": "hello",
+            "device_id": self._device_id,
+            "client_version": CLIENT_VERSION,
+        }))
+        # Track ws so the threadsafe shims work when _pump is exercised directly
+        # (i.e. without going through run_forever, as in unit tests).
+        self._ws = ws
+        try:
+            self._loop = asyncio.get_running_loop()
+        except RuntimeError:
+            pass
+        # 2. Pump.
+        while True:
+            frame = await ws.recv()
+            if isinstance(frame, (bytes, bytearray, memoryview)):
+                self._cb.on_assistant_audio(bytes(frame))
+                continue
+            try:
+                env = json.loads(frame)
+            except json.JSONDecodeError:
+                _log.warning("non-JSON text frame; dropping")
+                continue
+            await self._dispatch(env, ws)
+
+    async def _dispatch(self, env: dict, ws) -> None:
+        t = env.get("type")
+        if t == "hello_ack":
+            _log.info("hello_ack config=%s", env.get("config"))
+        elif t == "pong":
+            pass
+        elif t == "session_started":
+            self._cb.on_session_started(str(env.get("conversation_id", "")))
+        elif t == "assistant_done":
+            self._cb.on_assistant_done()
+        elif t == "session_ended":
+            self._cb.on_session_ended(str(env.get("reason", "unknown")))
+        elif t == "tool_call":
+            await self._handle_tool_call(env, ws)
+        elif t == "error":
+            _log.warning(
+                "backend error code=%s message=%s fatal=%s",
+                env.get("code"), env.get("message"), env.get("fatal"),
+            )
+        elif t == "config_updated":
+            _log.info("config_updated: %s", env.get("config"))
+        else:
+            _log.info("unhandled envelope type=%s", t)
+
+    async def _handle_tool_call(self, env: dict, ws) -> None:
+        call_id = env.get("call_id", "")
+        name = env.get("name", "")
+        args = env.get("arguments") or {}
+        if isinstance(args, str):
+            try:
+                args = json.loads(args)
+            except json.JSONDecodeError:
+                args = {}
+        handler = self._tools.get(name)
+        if handler is None:
+            await ws.send(json.dumps({
+                "type": "tool_result",
+                "call_id": call_id,
+                "ok": False,
+                "error": f"unknown tool: {name}",
+            }))
+            return
+        try:
+            result = handler(args)
+        except Exception as exc:  # noqa: BLE001
+            await ws.send(json.dumps({
+                "type": "tool_result",
+                "call_id": call_id,
+                "ok": False,
+                "error": str(exc),
+            }))
+            return
+        await ws.send(json.dumps({
+            "type": "tool_result",
+            "call_id": call_id,
+            "ok": True,
+            "result": result,
+        }))
+
+    async def _ping_loop(self, ws) -> None:
+        try:
+            while True:
+                await asyncio.sleep(PING_INTERVAL_S)
+                await ws.send(json.dumps({"type": "ping"}))
+        except asyncio.CancelledError:
+            pass
+
+    # ---- Public helpers used by main.py from the asyncio loop ----
+
+    async def send_wake(self) -> None:
+        if self._ws is None:
+            return
+        await self._ws.send(json.dumps({"type": "wake"}))
+
+    async def send_binary(self, pcm16: bytes) -> None:
+        if self._ws is None:
+            return
+        await self._ws.send(pcm16)
+
+    async def send_playback_done(self, label: str) -> None:
+        if self._ws is None:
+            return
+        await self._ws.send(json.dumps({"type": "playback_done", "for": label}))
+
+    async def send_cancel(self) -> None:
+        if self._ws is None:
+            return
+        await self._ws.send(json.dumps({"type": "cancel"}))
+
+    def call_soon_threadsafe_send_binary(self, pcm16: bytes) -> None:
+        """Thread-safe shim used by the bridge thread to queue an uplink send."""
+        if self._loop is None or self._ws is None:
+            return
+        self._loop.call_soon_threadsafe(
+            lambda: asyncio.create_task(self.send_binary(pcm16))
+        )
+
+    def call_soon_threadsafe_send_wake(self) -> None:
+        if self._loop is None or self._ws is None:
+            return
+        self._loop.call_soon_threadsafe(
+            lambda: asyncio.create_task(self.send_wake())
+        )
+
+    def call_soon_threadsafe_send_playback_done(self, label: str) -> None:
+        if self._loop is None or self._ws is None:
+            return
+        self._loop.call_soon_threadsafe(
+            lambda: asyncio.create_task(self.send_playback_done(label))
+        )

client/state.py

@@ -0,0 +1,70 @@
+"""Assistant client state machine.
+
+States flow: IDLE -> WAKE_PENDING -> LISTENING <-> ASSISTANT_SPEAKING
+                                       \\         /
+                                        session_ended
+                                                 \\
+                                                  IDLE_PENDING -> IDLE
+"""
+from enum import Enum, auto
+
+from client.log import get_logger
+
+_log = get_logger("state")
+
+
+class State(Enum):
+    IDLE = auto()
+    WAKE_PENDING = auto()
+    LISTENING = auto()
+    ASSISTANT_SPEAKING = auto()
+    IDLE_PENDING = auto()
+
+
+class StateMachine:
+    def __init__(self) -> None:
+        self._state = State.IDLE
+
+    @property
+    def state(self) -> State:
+        return self._state
+
+    @property
+    def wakeword_enabled(self) -> bool:
+        return self._state is State.IDLE
+
+    @property
+    def uplink_enabled(self) -> bool:
+        return self._state is State.LISTENING
+
+    def _transition(self, target: State) -> None:
+        if target is self._state:
+            return
+        _log.info("state %s -> %s", self._state.name, target.name)
+        self._state = target
+
+    def wake_fired(self) -> bool:
+        if self._state is not State.IDLE:
+            return False
+        self._transition(State.WAKE_PENDING)
+        return True
+
+    def session_started(self) -> None:
+        if self._state in (State.WAKE_PENDING, State.IDLE):
+            self._transition(State.LISTENING)
+
+    def assistant_audio_arrived(self) -> None:
+        if self._state is State.LISTENING:
+            self._transition(State.ASSISTANT_SPEAKING)
+
+    def assistant_done(self) -> None:
+        if self._state is State.ASSISTANT_SPEAKING:
+            self._transition(State.LISTENING)
+
+    def session_ended(self) -> None:
+        if self._state in (State.LISTENING, State.ASSISTANT_SPEAKING, State.WAKE_PENDING):
+            self._transition(State.IDLE_PENDING)
+
+    def sleep_played(self) -> None:
+        if self._state is State.IDLE_PENDING:
+            self._transition(State.IDLE)

client/tests/conftest.py

@@ -0,0 +1,8 @@
+"""Shared pytest fixtures for the client test suite."""
+import sys
+from pathlib import Path
+
+# Make the repo root importable so `import client.<module>` works without install.
+REPO_ROOT = Path(__file__).resolve().parents[2]
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))

client/tests/pytest.ini

@@ -0,0 +1,2 @@
+[pytest]
+asyncio_mode = auto

client/tests/test_audio_resample.py

@@ -0,0 +1,47 @@
+import numpy as np
+import pytest
+
+scipy_signal = pytest.importorskip("scipy.signal")
+
+from client.audio import resample_24k_to_16k
+
+
+def test_length_and_dtype():
+    frame = np.zeros(1920, dtype=np.int16)
+    out = resample_24k_to_16k(frame)
+    assert out.shape == (1280,)
+    assert out.dtype == np.int16
+
+
+def test_constant_signal_stays_constant():
+    frame = np.full(1920, 1000, dtype=np.int16)
+    out = resample_24k_to_16k(frame)
+    # poly resample of a constant is the same constant up to filter ringing
+    assert np.abs(out.astype(int).mean() - 1000) < 5
+
+
+def test_silence_in_silence_out():
+    frame = np.zeros(1920, dtype=np.int16)
+    out = resample_24k_to_16k(frame)
+    assert int(np.max(np.abs(out))) == 0
+
+
+def test_sine_peak_frequency_preserved():
+    # 1 kHz sine at 24 kHz: 80 ms = 80 cycles.
+    t = np.arange(1920) / 24_000.0
+    sig = (8000 * np.sin(2 * np.pi * 1000 * t)).astype(np.int16)
+    out = resample_24k_to_16k(sig)
+    # 1 kHz in a 1280-sample 16 kHz window is bin (1000 / 16000) * 1280 = 80.
+    spec = np.abs(np.fft.rfft(out.astype(np.float32)))
+    peak = int(np.argmax(spec))
+    assert abs(peak - 80) <= 1
+
+
+def test_zero_length_input_raises():
+    with pytest.raises(ValueError):
+        resample_24k_to_16k(np.zeros(0, dtype=np.int16))
+
+
+def test_wrong_dtype_rejected():
+    with pytest.raises(TypeError):
+        resample_24k_to_16k(np.zeros(1920, dtype=np.float32))

client/tests/test_config.py

@@ -0,0 +1,59 @@
+import json
+import os
+import stat
+
+import pytest
+
+from client.config import (
+    Config,
+    ConfigMissingError,
+    config_path,
+    load_config,
+    save_config,
+)
+
+
+def test_config_path_under_assistant_state(tmp_path, monkeypatch):
+    monkeypatch.setenv("HOME", str(tmp_path))
+    assert config_path() == tmp_path / "assistant" / "state" / "config.json"
+
+
+def test_load_missing_raises(tmp_path, monkeypatch):
+    monkeypatch.setenv("HOME", str(tmp_path))
+    with pytest.raises(ConfigMissingError):
+        load_config()
+
+
+def test_save_then_load_round_trip(tmp_path, monkeypatch):
+    monkeypatch.setenv("HOME", str(tmp_path))
+    cfg = Config(
+        device_id="11111111-1111-1111-1111-111111111111",
+        device_token="t0k3n",
+        backend_ws="wss://example/device",
+    )
+    save_config(cfg)
+    loaded = load_config()
+    assert loaded == cfg
+
+
+def test_save_creates_parent_dirs(tmp_path, monkeypatch):
+    monkeypatch.setenv("HOME", str(tmp_path))
+    cfg = Config(device_id="x", device_token="y", backend_ws="wss://z")
+    save_config(cfg)
+    assert (tmp_path / "assistant" / "state").is_dir()
+
+
+def test_save_sets_mode_0600(tmp_path, monkeypatch):
+    monkeypatch.setenv("HOME", str(tmp_path))
+    cfg = Config(device_id="x", device_token="y", backend_ws="wss://z")
+    save_config(cfg)
+    mode = stat.S_IMODE(os.stat(config_path()).st_mode)
+    assert mode == 0o600
+
+
+def test_load_rejects_missing_fields(tmp_path, monkeypatch):
+    monkeypatch.setenv("HOME", str(tmp_path))
+    config_path().parent.mkdir(parents=True, exist_ok=True)
+    config_path().write_text(json.dumps({"device_id": "x"}))
+    with pytest.raises(ConfigMissingError):
+        load_config()

client/tests/test_pair.py

@@ -0,0 +1,86 @@
+import io
+
+import pytest
+import responses
+
+from client.config import load_config
+from client.pair import pair
+
+
+@responses.activate
+def test_pair_posts_to_api_pair_and_writes_config(tmp_path, monkeypatch):
+    monkeypatch.setenv("HOME", str(tmp_path))
+    responses.post(
+        "https://backend.test/api/pair",
+        json={
+            "device_id": "00000000-0000-0000-0000-000000000001",
+            "device_token": "T0K3N",
+            "backend_ws": "wss://backend.test/device",
+        },
+        status=200,
+    )
+
+    pair(backend="https://backend.test", code="ABC12345", hostname="testhost")
+
+    cfg = load_config()
+    assert cfg.device_id == "00000000-0000-0000-0000-000000000001"
+    assert cfg.device_token == "T0K3N"
+    assert cfg.backend_ws == "wss://backend.test/device"
+
+    body = responses.calls[0].request.body
+    assert b"ABC12345" in body
+    assert b"testhost" in body
+
+
+@responses.activate
+def test_pair_strips_trailing_slash_from_backend(tmp_path, monkeypatch):
+    monkeypatch.setenv("HOME", str(tmp_path))
+    responses.post(
+        "https://backend.test/api/pair",
+        json={"device_id": "x", "device_token": "y", "backend_ws": "wss://z"},
+        status=200,
+    )
+    pair(backend="https://backend.test/", code="ABC12345", hostname="h")
+    assert responses.calls[0].request.url == "https://backend.test/api/pair"
+
+
+@responses.activate
+def test_pair_raises_on_404(tmp_path, monkeypatch):
+    monkeypatch.setenv("HOME", str(tmp_path))
+    responses.post(
+        "https://backend.test/api/pair",
+        json={"error": "code invalid or expired"},
+        status=404,
+    )
+    with pytest.raises(SystemExit) as exc:
+        pair(backend="https://backend.test", code="WRONGCOD", hostname="h")
+    assert "code invalid or expired" in str(exc.value)
+
+
+def test_prompt_used_when_code_not_provided(tmp_path, monkeypatch):
+    monkeypatch.setenv("HOME", str(tmp_path))
+    monkeypatch.setattr("sys.stdin", io.StringIO("FROMSTDIN\n"))
+
+    called_with = {}
+
+    def fake_post(url, json, timeout):
+        called_with["json"] = json
+
+        class R:
+            status_code = 200
+
+            def json(self_):
+                return {
+                    "device_id": "x",
+                    "device_token": "y",
+                    "backend_ws": "wss://z",
+                }
+
+            def raise_for_status(self_):
+                pass
+
+        return R()
+
+    monkeypatch.setattr("client.pair.requests.post", fake_post)
+    pair(backend="https://backend.test", code=None, hostname="h")
+    assert called_with["json"]["code"] == "FROMSTDIN"

client/tests/test_session.py

@@ -0,0 +1,186 @@
+import asyncio
+import json
+
+import pytest
+
+from client.session import Session
+
+
+class FakeWebSocket:
+    """Minimal stand-in for a `websockets` client connection."""
+    def __init__(self, inbox: list):
+        # inbox is a list of either str (text frame) or bytes (binary frame).
+        self._inbox = list(inbox)
+        self.sent: list = []
+
+    async def send(self, frame):
+        self.sent.append(frame)
+
+    async def recv(self):
+        if not self._inbox:
+            await asyncio.sleep(3600)  # block forever; tests cancel
+        return self._inbox.pop(0)
+
+    async def close(self):
+        pass
+
+
+class _FakeCallbacks:
+    """Implements client.session.Callbacks via duck-typing."""
+    def __init__(self):
+        self.started: list = []
+        self.audio: list = []
+        self.assistant_done = 0
+        self.session_ended: list = []
+
+    def on_session_started(self, conversation_id: str) -> None:
+        self.started.append(conversation_id)
+
+    def on_assistant_audio(self, pcm16: bytes) -> None:
+        self.audio.append(pcm16)
+
+    def on_assistant_done(self) -> None:
+        self.assistant_done += 1
+
+    def on_session_ended(self, reason: str) -> None:
+        self.session_ended.append(reason)
+
+
+async def _run_pump_briefly(session, ws, settle: float = 0.05):
+    task = asyncio.create_task(session._pump(ws))
+    await asyncio.sleep(settle)
+    task.cancel()
+    try:
+        await task
+    except (asyncio.CancelledError, BaseException):
+        pass
+
+
+async def test_hello_is_sent_on_connect_and_hello_ack_logged():
+    ws = FakeWebSocket(inbox=[
+        json.dumps({"type": "hello_ack", "config": {"voice": "alloy"}}),
+    ])
+    session = Session(
+        backend_ws="wss://x/device",
+        device_token="T",
+        device_id="00000000-0000-0000-0000-000000000001",
+        callbacks=_FakeCallbacks(),
+    )
+    await _run_pump_briefly(session, ws)
+
+    first = json.loads(ws.sent[0])
+    assert first["type"] == "hello"
+    assert first["device_id"] == "00000000-0000-0000-0000-000000000001"
+    assert first["client_version"]
+
+
+async def test_session_started_calls_callback_with_conversation_id():
+    cb = _FakeCallbacks()
+    ws = FakeWebSocket(inbox=[
+        json.dumps({"type": "hello_ack", "config": {}}),
+        json.dumps({
+            "type": "session_started",
+            "conversation_id": "cccccccc-cccc-cccc-cccc-cccccccccccc",
+        }),
+    ])
+    session = Session("wss://x/device", "T", "did", cb)
+    await _run_pump_briefly(session, ws)
+    assert cb.started == ["cccccccc-cccc-cccc-cccc-cccccccccccc"]
+
+
+async def test_binary_frame_goes_to_audio_callback():
+    cb = _FakeCallbacks()
+    payload = bytes(range(256)) * 15  # 3840 bytes
+    ws = FakeWebSocket(inbox=[
+        json.dumps({"type": "hello_ack", "config": {}}),
+        payload,
+    ])
+    session = Session("wss://x/device", "T", "did", cb)
+    await _run_pump_briefly(session, ws)
+    assert cb.audio == [payload]
+
+
+async def test_assistant_done_and_session_ended_callbacks():
+    cb = _FakeCallbacks()
+    ws = FakeWebSocket(inbox=[
+        json.dumps({"type": "hello_ack", "config": {}}),
+        json.dumps({"type": "assistant_done"}),
+        json.dumps({"type": "session_ended", "reason": "idle"}),
+    ])
+    session = Session("wss://x/device", "T", "did", cb)
+    await _run_pump_briefly(session, ws)
+    assert cb.assistant_done == 1
+    assert cb.session_ended == ["idle"]
+
+
+async def test_tool_call_runs_handler_and_sends_tool_result():
+    cb = _FakeCallbacks()
+    ws = FakeWebSocket(inbox=[
+        json.dumps({"type": "hello_ack", "config": {}}),
+        json.dumps({
+            "type": "tool_call",
+            "call_id": "c1",
+            "name": "set_volume",
+            "arguments": {"level": 42},
+        }),
+    ])
+    session = Session("wss://x/device", "T", "did", cb)
+    captured = {}
+
+    def handler(args):
+        captured["args"] = args
+        return {"ok": True, "level": args["level"]}
+
+    session.register_tool("set_volume", handler)
+    await _run_pump_briefly(session, ws)
+
+    assert captured["args"] == {"level": 42}
+    reply = json.loads(ws.sent[-1])
+    assert reply["type"] == "tool_result"
+    assert reply["call_id"] == "c1"
+    assert reply["ok"] is True
+    assert reply["result"] == {"ok": True, "level": 42}
+
+
+async def test_tool_call_with_exception_sends_ok_false():
+    cb = _FakeCallbacks()
+    ws = FakeWebSocket(inbox=[
+        json.dumps({"type": "hello_ack", "config": {}}),
+        json.dumps({
+            "type": "tool_call",
+            "call_id": "c2",
+            "name": "broken",
+            "arguments": {},
+        }),
+    ])
+    session = Session("wss://x/device", "T", "did", cb)
+
+    def handler(args):
+        raise RuntimeError("kaboom")
+
+    session.register_tool("broken", handler)
+    await _run_pump_briefly(session, ws)
+
+    reply = json.loads(ws.sent[-1])
+    assert reply["type"] == "tool_result"
+    assert reply["call_id"] == "c2"
+    assert reply["ok"] is False
+    assert "kaboom" in reply["error"]
+
+
+async def test_unknown_tool_sends_ok_false():
+    cb = _FakeCallbacks()
+    ws = FakeWebSocket(inbox=[
+        json.dumps({"type": "hello_ack", "config": {}}),
+        json.dumps({
+            "type": "tool_call",
+            "call_id": "c3",
+            "name": "ghost",
+            "arguments": {},
+        }),
+    ])
+    session = Session("wss://x/device", "T", "did", cb)
+    await _run_pump_briefly(session, ws)
+    reply = json.loads(ws.sent[-1])
+    assert reply["ok"] is False
+    assert "ghost" in reply["error"]

client/tests/test_state.py

@@ -0,0 +1,102 @@
+from client.state import State, StateMachine
+
+
+def test_starts_idle_with_wakeword_enabled_and_uplink_disabled():
+    sm = StateMachine()
+    assert sm.state is State.IDLE
+    assert sm.wakeword_enabled
+    assert not sm.uplink_enabled
+
+
+def test_wake_fired_goes_idle_to_wake_pending_and_disables_wakeword():
+    sm = StateMachine()
+    assert sm.wake_fired() is True
+    assert sm.state is State.WAKE_PENDING
+    assert not sm.wakeword_enabled
+    assert not sm.uplink_enabled
+
+
+def test_wake_fired_is_idempotent_when_not_idle():
+    sm = StateMachine()
+    sm.wake_fired()
+    assert sm.wake_fired() is False
+    assert sm.state is State.WAKE_PENDING
+
+
+def test_session_started_goes_wake_pending_to_listening_and_unmutes_uplink():
+    sm = StateMachine()
+    sm.wake_fired()
+    sm.session_started()
+    assert sm.state is State.LISTENING
+    assert sm.uplink_enabled
+    assert not sm.wakeword_enabled
+
+
+def test_assistant_audio_arrived_mutes_uplink():
+    sm = StateMachine()
+    sm.wake_fired()
+    sm.session_started()
+    sm.assistant_audio_arrived()
+    assert sm.state is State.ASSISTANT_SPEAKING
+    assert not sm.uplink_enabled
+
+
+def test_assistant_audio_arrived_is_noop_outside_listening():
+    sm = StateMachine()
+    sm.assistant_audio_arrived()
+    assert sm.state is State.IDLE  # ignored from IDLE
+    sm.wake_fired()
+    sm.assistant_audio_arrived()
+    assert sm.state is State.WAKE_PENDING  # ignored from WAKE_PENDING
+
+
+def test_assistant_done_returns_to_listening_from_speaking():
+    sm = StateMachine()
+    sm.wake_fired()
+    sm.session_started()
+    sm.assistant_audio_arrived()
+    sm.assistant_done()
+    assert sm.state is State.LISTENING
+    assert sm.uplink_enabled
+
+
+def test_assistant_done_is_noop_when_not_speaking():
+    sm = StateMachine()
+    sm.wake_fired()
+    sm.session_started()  # LISTENING; never went to ASSISTANT_SPEAKING
+    sm.assistant_done()
+    assert sm.state is State.LISTENING
+
+
+def test_session_ended_routes_through_idle_pending_then_idle():
+    sm = StateMachine()
+    sm.wake_fired()
+    sm.session_started()
+    sm.session_ended()
+    assert sm.state is State.IDLE_PENDING
+    assert not sm.wakeword_enabled
+    assert not sm.uplink_enabled
+    sm.sleep_played()
+    assert sm.state is State.IDLE
+    assert sm.wakeword_enabled
+
+
+def test_session_ended_works_from_assistant_speaking():
+    sm = StateMachine()
+    sm.wake_fired()
+    sm.session_started()
+    sm.assistant_audio_arrived()
+    sm.session_ended()
+    assert sm.state is State.IDLE_PENDING
+
+
+def test_session_ended_from_idle_is_noop():
+    sm = StateMachine()
+    sm.session_ended()
+    assert sm.state is State.IDLE
+
+
+def test_sleep_played_outside_idle_pending_is_noop():
+    sm = StateMachine()
+    sm.sleep_played()
+    assert sm.state is State.IDLE

client/wakeword.py

@@ -0,0 +1,47 @@
+"""openwakeword wrapper for the Pi client.
+
+Loads the stock `alexa` model. Each call to `predict()` takes a 1280-sample
+16 kHz mono int16 frame (produced by `client.audio.resample_24k_to_16k`).
+Returns True at most once per cooldown window.
+"""
+import time
+
+import numpy as np
+
+from client.log import get_logger
+
+_log = get_logger("wakeword")
+
+WAKEWORD = "alexa"
+DEFAULT_THRESHOLD = 0.5
+DEFAULT_COOLDOWN_S = 1.5
+
+
+class WakewordDetector:
+    def __init__(self, *, threshold: float = DEFAULT_THRESHOLD,
+                 cooldown_s: float = DEFAULT_COOLDOWN_S) -> None:
+        # Lazy-import openwakeword to keep the test suite importable without it.
+        import openwakeword.utils
+        from openwakeword.model import Model
+
+        _log.info("downloading openwakeword models (idempotent)")
+        # No-arg form dodges the None-handling crash documented in findings.md §4.
+        openwakeword.utils.download_models()
+        _log.info("loading openwakeword model %r", WAKEWORD)
+        t0 = time.monotonic()
+        self._model = Model(wakeword_models=[WAKEWORD], inference_framework="onnx")
+        _log.info("openwakeword model loaded in %.2fs", time.monotonic() - t0)
+
+        self._threshold = threshold
+        self._cooldown_s = cooldown_s
+        self._last_fired = 0.0
+
+    def predict(self, frame_16k_int16: np.ndarray) -> bool:
+        scores = self._model.predict(frame_16k_int16)
+        score = float(scores.get(WAKEWORD, 0.0))
+        now = time.monotonic()
+        if score >= self._threshold and (now - self._last_fired) >= self._cooldown_s:
+            self._last_fired = now
+            _log.info("wakeword fired score=%.3f", score)
+            return True
+        return False

requirements-dev.txt

@@ -0,0 +1,3 @@
+pytest>=8.0
+pytest-asyncio>=0.23
+responses>=0.25