From 4661bdc47f62934328c30fd9fb9ba2b567ad46ec Mon Sep 17 00:00:00 2001
From: Andre Merzky <andre@merzky.net>
Date: Wed, 19 Aug 2026 11:25:47 +0200
Subject: [PATCH] gateway: the SSE tap renders every event, bytes included

The tap forwards whatever a plugin published, and plugin payloads are
not required to be JSON.  A `bytes` payload -- a raw stream event, for
one -- killed the frame with `TypeError: Object of type bytes is not
JSON serializable`: one dropped notification per event, one logged `tap
callback failed` each, 292 of them in a 45s digital-twin run, and an
SSE consumer which sees nothing while the broker looks busy.

The SSE feed is a monitoring tier, so the invariant is that it renders
every event rather than that every payload round-trips: anything JSON
cannot encode now falls back to `str()`.  A consumer which wants the
payload itself still has the plugin's own channel; the tap only has to
say that the event happened.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 src/radical/orbit/gateway.py    | 13 +++++++++++--
 tests/unittests/test_gateway.py | 19 +++++++++++++++++++
 2 files changed, 30 insertions(+), 2 deletions(-)

diff --git a/src/radical/orbit/gateway.py b/src/radical/orbit/gateway.py
index 2feb42e..1b9ede6 100644
--- a/src/radical/orbit/gateway.py
+++ b/src/radical/orbit/gateway.py
@@ -255,8 +255,17 @@ def detach(self) -> None:
     def _sse_frame(topic: str, data: Any) -> str:
         """Format one SSE frame as HTTP clients expect it:
         ``data: {"topic": <topic>, "data": <data>}\\n\\n`` (no keepalive
-        comments)."""
-        return "data: %s\n\n" % json.dumps({"topic": topic, "data": data})
+        comments).
+
+        The tap forwards whatever a plugin published, and plugin payloads
+        are not required to be JSON -- a ``bytes`` payload (e.g. a raw
+        stream event) used to kill the frame with a ``TypeError``, one
+        dropped notification per event.  The SSE feed is a monitoring
+        tier: it must render every event, so anything JSON cannot encode
+        falls back to ``str()`` rather than raising.
+        """
+        return "data: %s\n\n" % json.dumps({"topic": topic, "data": data},
+                                            default=str)
 
     def _on_event(self, event: Dict[str, Any]) -> None:
         """Raw event tap callback — runs on the **plugin-host loop**.
diff --git a/tests/unittests/test_gateway.py b/tests/unittests/test_gateway.py
index 543cd23..c229850 100644
--- a/tests/unittests/test_gateway.py
+++ b/tests/unittests/test_gateway.py
@@ -455,6 +455,25 @@ def test_sse_topology_on_connect_and_disconnect(harness):
         sse.close()
 
 
+def test_sse_frame_survives_non_json_payloads():
+    # The tap forwards plugin payloads verbatim, and those are not
+    # required to be JSON: a bytes payload (a raw stream event, say) used
+    # to raise TypeError inside json.dumps and silently drop the frame.
+    # The SSE feed is monitoring -- it must render every event.
+    import json as _json
+    from radical.orbit.gateway import Gateway
+
+    frame = Gateway._sse_frame('notification', {'data': b'\x80\x05binary'})
+    assert frame.startswith('data: ') and frame.endswith('\n\n')
+    parsed = _json.loads(frame[len('data: '):])
+    assert parsed['topic'] == 'notification'
+    assert 'binary' in parsed['data']['data']             # str() fallback
+
+    # JSON-clean payloads are untouched by the fallback
+    frame = Gateway._sse_frame('notification', {'data': {'hello': 'world'}})
+    assert _json.loads(frame[len('data: '):])['data'] == {'data': {'hello': 'world'}}
+
+
 def test_sse_queue_drop_oldest_counter():
     # The gateway now uses the shared bounded, drop-oldest queue.
     from radical.orbit.queues import BoundedDropOldestQueue