Diagnostics: surface upstream WS close reason + break the null-spin loop

The Pi journal showed 170 000 'upstream silent NxN' messages within one
second — i.e. ReceiveJsonAsync was returning null in zero time, so the
upstream WS to OpenAI was closed. The relay was spinning until idle
watchdog (30 s) fired. Add IRealtimeUpstream.IsClosed + CloseReason,
populated by OpenAIRealtimeUpstream from the close status / WS exception,
and have PumpAsync stop on close and forward
  error code=upstream_closed message=close status=<X> desc=<Y>
to the device. Closes the spin AND gives us the OpenAI close reason.

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.

backend.tests/DeviceHubWakeFlowTests.cs

@@ -52,11 +52,11 @@ public class DeviceHubWakeFlowTests(ApiFactory factory) : IClassFixture<ApiFacto
         });
         FakeRealtimeSessionFactory.Upstream.Push(new JsonObject { ["type"] = "response.done" });
 
-        var done = await ReceiveJson(ws);
+        var done = await ReceiveUntil(ws, "assistant_done");
         done.GetProperty("type").GetString().Should().Be("assistant_done");
 
         await SendJson(ws, new { type = "cancel" });
-        var ended = await ReceiveJson(ws);
+        var ended = await ReceiveUntil(ws, "session_ended");
         ended.GetProperty("type").GetString().Should().Be("session_ended");
     }
 
@@ -114,4 +114,16 @@ public class DeviceHubWakeFlowTests(ApiFactory factory) : IClassFixture<ApiFacto
         }
         return JsonDocument.Parse(stream.ToArray()).RootElement.Clone();
     }
+
+    // The relay forwards an upstream-event "debug" envelope for every event
+    // it sees, so envelopes the test cares about are interleaved with debugs.
+    private static async Task<JsonElement> ReceiveUntil(WebSocket ws, string type)
+    {
+        for (int i = 0; i < 50; i++)
+        {
+            var env = await ReceiveJson(ws);
+            if (env.GetProperty("type").GetString() == type) return env;
+        }
+        throw new TimeoutException($"never saw envelope type={type}");
+    }
 }

backend.tests/ScriptedRealtimeUpstream.cs

@@ -13,6 +13,8 @@ public class ScriptedRealtimeUpstream : IRealtimeUpstream
         Channel.CreateUnbounded<JsonObject>(new UnboundedChannelOptions { SingleWriter = true });
 
     public IList<JsonObject> Sent { get; } = new List<JsonObject>();
+    public bool IsClosed => _toRelay.Reader.Completion.IsCompleted;
+    public string? CloseReason => IsClosed ? "scripted upstream closed" : null;
 
     public Task SendJsonAsync(JsonObject envelope, CancellationToken ct)
     {

backend/DeviceHub/DeviceHubEndpoint.cs

@@ -130,8 +130,14 @@ public static class DeviceHubEndpoint
     {
         var enabledTools = JsonSerializer.Deserialize<string[]>(device.Config?.EnabledToolsJson ?? "[]")
             ?? Array.Empty<string>();
+        // Promote any deprecated Beta-era model name to the GA name. Plan 4's
+        // admin UI will let users set a specific value; until then, anything
+        // that mentions the old preview slug routes to gpt-realtime.
+        var model = device.Config?.Model ?? "gpt-realtime";
+        if (model.StartsWith("gpt-4o-realtime-preview", StringComparison.Ordinal))
+            model = "gpt-realtime";
         var cfg = new RealtimeSettings(
-            Model: device.Config?.Model ?? "gpt-4o-realtime-preview",
+            Model: model,
             Voice: device.Config?.Voice ?? "alloy",
             SystemPrompt: device.Config?.SystemPrompt ?? "",
             IdleTimeoutSeconds: device.Config?.IdleTimeoutSeconds ?? 30,
@@ -171,6 +177,10 @@ public static class DeviceHubEndpoint
         }, ct);
     }
 
+    // Bump on every backend change so we can verify a Coolify deploy landed
+    // by looking at the hello_ack config.server_version field in the Pi journal.
+    public const string ServerVersion = "plan3-debug-2026-06-12-09-closeprobe";
+
     private static async Task SendHelloAckAsync(ActiveDevice active, Device device, CancellationToken ct)
     {
         var enabled = device.Config?.EnabledToolsJson ?? "[]";
@@ -180,7 +190,8 @@ public static class DeviceHubEndpoint
           "model": "{{Escape(device.Config?.Model)}}",
           "system_prompt": "{{Escape(device.Config?.SystemPrompt)}}",
           "idle_timeout_seconds": {{device.Config?.IdleTimeoutSeconds ?? 30}},
-          "enabled_tools": {{enabled}}
+          "enabled_tools": {{enabled}},
+          "server_version": "{{ServerVersion}}"
         }
         """)!.AsObject();
 

backend/Realtime/IRealtimeUpstream.cs

@@ -6,4 +6,8 @@ public interface IRealtimeUpstream : IAsyncDisposable
 {
     Task SendJsonAsync(JsonObject envelope, CancellationToken ct);
     Task<JsonObject?> ReceiveJsonAsync(CancellationToken ct);
+    // Once a receive returns null because the WebSocket transitioned out of
+    // the Open state, these surface why. Both stay null until that happens.
+    bool IsClosed { get; }
+    string? CloseReason { get; }
 }

backend/Realtime/OpenAIRealtimeUpstream.cs

@@ -7,11 +7,16 @@ namespace backend.Realtime;
 public class OpenAIRealtimeUpstream : IRealtimeUpstream
 {
     private readonly ClientWebSocket _ws = new();
+    private string? _closeReason;
+
+    public bool IsClosed => _ws.State != WebSocketState.Open && _ws.State != WebSocketState.Connecting;
+    public string? CloseReason => _closeReason;
 
     public async Task ConnectAsync(string apiKey, string model, CancellationToken ct)
     {
         _ws.Options.SetRequestHeader("Authorization", $"Bearer {apiKey}");
-        _ws.Options.SetRequestHeader("OpenAI-Beta", "realtime=v1");
+        // The GA Realtime API does NOT take the OpenAI-Beta: realtime=v1 header.
+        // Sending it returns upstream error code=beta_api_shape_disabled.
         var uri = new Uri($"wss://api.openai.com/v1/realtime?model={Uri.EscapeDataString(model)}");
         await _ws.ConnectAsync(uri, ct);
     }
@@ -30,11 +35,27 @@ public class OpenAIRealtimeUpstream : IRealtimeUpstream
         while (true)
         {
             WebSocketReceiveResult result;
-            try { result = await _ws.ReceiveAsync(buffer, ct); }
-            catch (WebSocketException) { return null; }
-            catch (OperationCanceledException) { return null; }
+            try
+            {
+                result = await _ws.ReceiveAsync(buffer, ct);
+            }
+            catch (WebSocketException exc)
+            {
+                _closeReason ??= $"WebSocketException:{exc.WebSocketErrorCode}:{exc.Message}";
+                return null;
+            }
+            catch (OperationCanceledException)
+            {
+                return null;
+            }
 
-            if (result.MessageType == WebSocketMessageType.Close) return null;
+            if (result.MessageType == WebSocketMessageType.Close)
+            {
+                var status = _ws.CloseStatus?.ToString() ?? "<none>";
+                var desc = _ws.CloseStatusDescription ?? "<no description>";
+                _closeReason ??= $"close status={status} desc={desc}";
+                return null;
+            }
             stream.Write(buffer, 0, result.Count);
             if (result.EndOfMessage)
             {

backend/Realtime/RealtimeEvents.cs

@@ -6,6 +6,12 @@ namespace backend.Realtime;
 
 public static class RealtimeEvents
 {
+    // GA Realtime API shape (the Beta API shape was retired in 2026 with
+    // upstream error "beta_api_shape_disabled"). Key differences:
+    //   - session.type = "realtime" is required.
+    //   - voice / audio formats / transcription / turn_detection now nested
+    //     under session.audio.{input,output}.
+    //   - No OpenAI-Beta header on the upgrade (see OpenAIRealtimeUpstream).
     public static JsonObject SessionUpdate(RealtimeSettings cfg, IEnumerable<ITool> enabledTools)
     {
         var tools = new JsonArray();
@@ -20,23 +26,38 @@ public static class RealtimeEvents
             });
         }
 
+        // Minimal GA session.update. We drop explicit audio.input/output.format
+        // objects (the GA shape accepts them but OpenAI defaults to 24 kHz PCM16
+        // mono on both directions, which is what we use). Removing the explicit
+        // format objects matches the documented minimal session config and
+        // eliminates a known source of OpenAI silently ignoring the audio.
         return new JsonObject
         {
             ["type"] = "session.update",
             ["session"] = new JsonObject
             {
-                ["voice"] = cfg.Voice,
+                ["type"] = "realtime",
+                ["model"] = cfg.Model,
                 ["instructions"] = cfg.SystemPrompt,
-                ["modalities"] = new JsonArray("text", "audio"),
-                ["input_audio_format"] = "pcm16",
-                ["output_audio_format"] = "pcm16",
-                ["input_audio_transcription"] = new JsonObject { ["model"] = "whisper-1" },
-                ["turn_detection"] = new JsonObject
+                ["audio"] = new JsonObject
                 {
-                    ["type"] = "server_vad",
-                    ["threshold"] = 0.5,
-                    ["prefix_padding_ms"] = 300,
-                    ["silence_duration_ms"] = 500,
+                    ["input"] = new JsonObject
+                    {
+                        ["transcription"] = new JsonObject { ["model"] = "whisper-1" },
+                        ["turn_detection"] = new JsonObject
+                        {
+                            ["type"] = "server_vad",
+                            ["threshold"] = 0.5,
+                            ["prefix_padding_ms"] = 300,
+                            ["silence_duration_ms"] = 500,
+                            ["create_response"] = true,
+                            ["interrupt_response"] = true,
+                        },
+                    },
+                    ["output"] = new JsonObject
+                    {
+                        ["voice"] = cfg.Voice,
+                    },
                 },
                 ["tools"] = tools,
                 ["tool_choice"] = "auto",

backend/Realtime/RealtimeSession.cs

@@ -65,11 +65,15 @@ public class RealtimeSession
         try
         {
             var enabled = _registry.EnabledFor(_cfg.EnabledTools).ToList();
+            _logger.LogInformation(
+                "relay opening session: model={Model} voice={Voice} tools=[{Tools}]",
+                _cfg.Model, _cfg.Voice, string.Join(",", enabled.Select(t => t.Name)));
             await _upstream.SendJsonAsync(RealtimeEvents.SessionUpdate(_cfg, enabled), ct);
 
             var ack = await WaitForUpstreamTypeAsync("session.updated", ct);
             if (ack is null)
             {
+                _logger.LogWarning("upstream closed before session.updated");
                 _endReason = EndReason.Error;
                 return;
             }
@@ -116,15 +120,71 @@ public class RealtimeSession
 
     private async Task PumpAsync(CancellationToken ct)
     {
+        int nullPolls = 0;
         while (!ct.IsCancellationRequested && !_endRequested)
         {
             var evt = await ReceiveOrIdleAsync(ct);
             if (evt is null && _endRequested) return;
             if (evt is null && ct.IsCancellationRequested) return;
-            if (evt is null) continue;
+            if (evt is null)
+            {
+                // If the upstream WS is closed, stop spinning — surface the
+                // close reason to the device and bail with EndReason=Error.
+                if (_upstream.IsClosed)
+                {
+                    var reason = _upstream.CloseReason ?? "upstream closed";
+                    _logger.LogWarning("upstream closed mid-session: {Reason}", reason);
+                    await _output.WriteEnvelopeAsync(new JsonObject
+                    {
+                        ["type"] = "error",
+                        ["code"] = "upstream_closed",
+                        ["message"] = reason,
+                        ["fatal"] = false,
+                    }, ct);
+                    _endReason = EndReason.Error;
+                    return;
+                }
+                // Surface "alive but silent" every ~2s so we can tell that
+                // from "the upstream WS was closed". Coolify logs aren't
+                // reachable from the dev sandbox.
+                if (++nullPolls % 8 == 0)
+                {
+                    _logger.LogInformation("upstream silent ({Polls} x 250ms polls)", nullPolls);
+                    await _output.WriteEnvelopeAsync(new JsonObject
+                    {
+                        ["type"] = "debug",
+                        ["message"] = "upstream silent " + nullPolls + "x250ms",
+                    }, ct);
+                }
+                continue;
+            }
+            nullPolls = 0;
             var type = (string?)evt["type"];
+            _logger.LogInformation("relay evt={EvtType}", type);
+            // Forward the event type to the device so it appears in the Pi
+            // journal — Coolify logs aren't accessible from the dev sandbox.
+            await _output.WriteEnvelopeAsync(new JsonObject
+            {
+                ["type"] = "debug",
+                ["message"] = "relay evt=" + (type ?? "<null>"),
+            }, ct);
             switch (type)
             {
+                case "error":
+                {
+                    var err = evt["error"] as JsonObject;
+                    var code = (string?)err?["code"] ?? "unknown";
+                    var msg = (string?)err?["message"] ?? evt.ToJsonString();
+                    _logger.LogWarning("upstream mid-session error code={Code} message={Message}", code, msg);
+                    await _output.WriteEnvelopeAsync(new JsonObject
+                    {
+                        ["type"] = "error",
+                        ["code"] = "upstream_" + code,
+                        ["message"] = msg,
+                        ["fatal"] = false,
+                    }, ct);
+                    break;
+                }
                 case "input_audio_buffer.speech_started":
                     _lastSpeechAt = DateTime.UtcNow;
                     break;
@@ -274,7 +334,25 @@ public class RealtimeSession
         {
             var evt = await _upstream.ReceiveJsonAsync(ct);
             if (evt is null) return null;
-            if ((string?)evt["type"] == type) return evt;
+            var evtType = (string?)evt["type"];
+            _logger.LogInformation("upstream evt={EvtType}", evtType);
+            if (evtType == "error")
+            {
+                var err = evt["error"] as JsonObject;
+                var code = (string?)err?["code"] ?? "unknown";
+                var msg = (string?)err?["message"] ?? evt.ToJsonString();
+                _logger.LogWarning("upstream error code={Code} message={Message}", code, msg);
+                await _output.WriteEnvelopeAsync(new JsonObject
+                {
+                    ["type"] = "error",
+                    ["code"] = "upstream_" + code,
+                    ["message"] = msg,
+                    ["fatal"] = false,
+                }, ct);
+                _endReason = EndReason.Error;
+                return null;
+            }
+            if (evtType == type) return evt;
         }
     }
 }

backend/Settings/SystemSettings.cs

@@ -6,6 +6,6 @@ public class SystemSettings
     public string DefaultSystemPrompt { get; set; } =
         "You are a helpful voice assistant. Keep responses concise.";
     public string DefaultVoice { get; set; } = "alloy";
-    public string DefaultModel { get; set; } = "gpt-4o-realtime-preview";
+    public string DefaultModel { get; set; } = "gpt-realtime";
     public int DefaultIdleTimeoutSeconds { get; set; } = 30;
 }

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,191 @@
-"""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():
+        frame_counter = 0
+        recent_rms_max = 0.0
+        while not stop_event.is_set():
+            try:
+                pcm16_bytes = mic_queue.get(timeout=0.25)
+            except queue.Empty:
+                continue
+            frame_counter += 1
+            raw = np.frombuffer(pcm16_bytes, dtype=np.int16)
+            rms = float(np.sqrt(np.mean(raw.astype(np.float32) ** 2)))
+            if rms > recent_rms_max:
+                recent_rms_max = rms
+            # Every ~5 s emit a heartbeat so we can see if the InputStream is
+            # actually producing frames and what state we're routing to.
+            if frame_counter % 62 == 0:
+                _log.info(
+                    "bridge frames=%d state=%s wake_en=%s up_en=%s qsize=%d rms_max_5s=%.0f",
+                    frame_counter, sm.state.name,
+                    sm.wakeword_enabled, sm.uplink_enabled, mic_queue.qsize(),
+                    recent_rms_max,
+                )
+                recent_rms_max = 0.0
+            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")
+
+        def on_disconnected(self) -> None:
+            # The conversation is gone with the WS — reset to IDLE so the
+            # wakeword detector can re-arm. Any pending assistant audio left
+            # in the playback queue still drains, but we drop a "sleep" done
+            # marker only if we were mid-session.
+            _log.info("on_disconnected: forcing state to IDLE")
+            sm.force_idle()
+    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,235 @@
+"""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: ...
+    # Called once per backend WS lifecycle when the connection drops, so the
+    # owner can reset state (the conversation is dead).
+    def on_disconnected(self) -> 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
+                        self._notify_disconnected()
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:  # noqa: BLE001
+                _log.warning("session error: %s; reconnecting in %.1fs", exc, backoff)
+                self._notify_disconnected()
+            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"))
+        elif t == "debug":
+            _log.info("backend %s", env.get("message"))
+        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,
+        }))
+
+    def _notify_disconnected(self) -> None:
+        # Tolerant: Callbacks Protocol marks on_disconnected as part of the
+        # surface, but real callers (and tests) may omit it.
+        fn = getattr(self._cb, "on_disconnected", None)
+        if fn is None:
+            return
+        try:
+            fn()
+        except Exception as exc:  # noqa: BLE001
+            _log.warning("on_disconnected raised: %s", exc)
+
+    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 ----
+    #
+    # All of these are called fire-and-forget via call_soon_threadsafe, so they
+    # must NEVER raise — the asyncio loop has nowhere to surface the exception
+    # except as an unhandled-task-exception traceback in the journal.
+
+    async def _safe_send(self, payload) -> None:
+        if self._ws is None:
+            return
+        try:
+            await self._ws.send(payload)
+        except Exception as exc:  # noqa: BLE001
+            # ConnectionClosed / ConnectionClosedOK / ConnectionClosedError all
+            # land here. The reconnect loop already handles WS lifecycle.
+            _log.debug("send dropped: %s", exc)
+
+    async def send_wake(self) -> None:
+        await self._safe_send(json.dumps({"type": "wake"}))
+
+    async def send_binary(self, pcm16: bytes) -> None:
+        await self._safe_send(pcm16)
+
+    async def send_playback_done(self, label: str) -> None:
+        await self._safe_send(json.dumps({"type": "playback_done", "for": label}))
+
+    async def send_cancel(self) -> None:
+        await self._safe_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,77 @@
+"""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)
+
+    def force_idle(self) -> None:
+        """Unconditional reset to IDLE — used when the backend WS drops and
+        the conversation is dead regardless of what state we thought we were in.
+        Re-enables the wakeword detector."""
+        if self._state is not State.IDLE:
+            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,127 @@
+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
+
+
+def test_force_idle_from_listening_resets_state_and_reenables_wakeword():
+    sm = StateMachine()
+    sm.wake_fired()
+    sm.session_started()
+    assert sm.state is State.LISTENING
+    sm.force_idle()
+    assert sm.state is State.IDLE
+    assert sm.wakeword_enabled
+
+
+def test_force_idle_from_assistant_speaking_resets():
+    sm = StateMachine()
+    sm.wake_fired()
+    sm.session_started()
+    sm.assistant_audio_arrived()
+    sm.force_idle()
+    assert sm.state is State.IDLE
+
+
+def test_force_idle_on_idle_is_noop():
+    sm = StateMachine()
+    sm.force_idle()
+    assert sm.state is State.IDLE

client/wakeword.py

@@ -0,0 +1,52 @@
+"""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))
+        rms = float(np.sqrt(np.mean(frame_16k_int16.astype(np.float32) ** 2)))
+        now = time.monotonic()
+        # Diagnostic: log any non-trivial score so we can see what's getting
+        # close to firing.
+        if score >= 0.1:
+            _log.info("score=%.3f rms=%.0f", score, rms)
+        if score >= self._threshold and (now - self._last_fired) >= self._cooldown_s:
+            self._last_fired = now
+            _log.info("wakeword fired score=%.3f rms=%.0f", score, rms)
+            return True
+        return False

requirements-dev.txt

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