From a42038b825672507a86576ddcf7f2d37e6540e7d Mon Sep 17 00:00:00 2001
From: Assistant builder <meelstorm@gmail.com>
Date: Fri, 12 Jun 2026 06:59:14 +0000
Subject: [PATCH] Diagnostics: forward upstream events + errors to device as
 'debug' envelopes

Coolify container logs aren't accessible from the dev sandbox, so the only
log surface we can read is the Pi journal. The relay now forwards every
upstream event type to the device as {type:'debug', message:'relay evt=<T>'}
and every mid-session upstream error event as {type:'error',code:'upstream_<C>'}.

The client's Session.dispatch handles 'debug' as an INFO log line. Updates
the wake-flow integration test's receive-loop to skip over the new debug
envelopes via a ReceiveUntil helper.
---
 backend.tests/DeviceHubWakeFlowTests.cs | 16 ++++++++++--
 backend/Realtime/RealtimeSession.cs     | 34 ++++++++++++++++++++++++-
 client/session.py                       |  2 ++
 3 files changed, 49 insertions(+), 3 deletions(-)

diff --git a/backend.tests/DeviceHubWakeFlowTests.cs b/backend.tests/DeviceHubWakeFlowTests.cs
index 8d013fc..497fe88 100644
--- a/backend.tests/DeviceHubWakeFlowTests.cs
+++ b/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}");
+    }
 }
diff --git a/backend/Realtime/RealtimeSession.cs b/backend/Realtime/RealtimeSession.cs
index 8ea67a7..61ee09e 100644
--- a/backend/Realtime/RealtimeSession.cs
+++ b/backend/Realtime/RealtimeSession.cs
@@ -120,15 +120,47 @@ 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)
+            {
+                // Upstream has gone silent. If null persists for >2s, log so
+                // we can spot a dead upstream that's not surfacing as a close.
+                if (++nullPolls % 8 == 0)
+                    _logger.LogDebug("upstream silent ({Polls} x 250ms polls)", nullPolls);
+                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;
diff --git a/client/session.py b/client/session.py
index d52822c..deaf6fc 100644
--- a/client/session.py
+++ b/client/session.py
@@ -125,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)