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.

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" },
+                ["audio"] = new JsonObject
+                {
+                    ["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/main.py

@@ -43,11 +43,28 @@ def _build_bridge_thread(
     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:
@@ -90,6 +107,14 @@ def _make_callbacks(sm: StateMachine, playback: PlaybackWorker):
                 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()
 
 

client/session.py

@@ -19,6 +19,9 @@ class Callbacks(Protocol):
     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:
@@ -64,10 +67,12 @@ class Session:
                     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)
@@ -120,6 +125,8 @@ class Session:
             )
         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)
 
@@ -158,6 +165,17 @@ class Session:
             "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:
@@ -167,26 +185,32 @@ class Session:
             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:
-        if self._ws is None:
-            return
-        await self._ws.send(json.dumps({"type": "wake"}))
+        await self._safe_send(json.dumps({"type": "wake"}))
 
     async def send_binary(self, pcm16: bytes) -> None:
-        if self._ws is None:
-            return
-        await self._ws.send(pcm16)
+        await self._safe_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}))
+        await self._safe_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"}))
+        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."""

client/state.py

@@ -68,3 +68,10 @@ class StateMachine:
     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/test_state.py

@@ -100,3 +100,28 @@ 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

@@ -39,9 +39,14 @@ class WakewordDetector:
     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", score)
+            _log.info("wakeword fired score=%.3f rms=%.0f", score, rms)
             return True
         return False