refactor(node): dispatch MNP messages through a table

The pre-authentication guards stay explicit code, in the same order and
text. After the handshake, a table maps each type to its handler and to
whether it runs as a task, the choice each branch made; the three inline
blocks become StreamingMixin methods, unchanged. The dispatch golden is
identical, including types that are not strings.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

1 files changed, 58 insertions, 0 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py
index 05ab9e7..a425c65 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py
@@ -641,3 +641,61 @@ class StreamingMixin:
                  reason, index, time.monotonic() - self._stream_started_at)
         log.info("Streamed %s: %d segments", entry.name, index)
         self._audit("stream_video", entry.name)
+
+    def _do_stream_request(self, msg: dict) -> None:
+        sem = self._ctx.get("_transcode_sem")
+        log.info("stream: req file=%s credits=%s slots_free=%s prev=%s",
+                 str(msg.get("file_id"))[:12], msg.get("credits"),
+                 getattr(sem, "_value", "?"),
+                 "alive" if (self._stream_task and
+                             not self._stream_task.done()) else "none")
+        self._spawn(self._replace_stream(msg))
+
+    def _do_client_diag(self, msg: dict) -> None:
+        # Diagnostics only. The node acts on none of it — it writes it
+        # next to its own view of the same stream, which is the only
+        # place the two halves can be compared when the client is a
+        # phone with no console.
+        # Every field is peer-controlled, so each is stringified and
+        # cut short: this is a log line, not a channel for writing
+        # whatever one likes into the operator's file.
+        def _f(key: str, n: int = 24) -> str:
+            return str(msg.get(key))[:n].replace("\n", " ")
+        if msg.get("event"):
+            # Once per stream or per seek, not once per five seconds —
+            # and a seek nobody asked for looks exactly like a viewer
+            # dragging the scrubber from this side, so it has to be
+            # visible without turning DEBUG on.
+            log.info(
+                "stream: client %s target=%s t=%ss offset=%s ready=%s "
+                "duration=%s ranges=[%s]",
+                _f("event", 16), _f("target"), _f("t"), _f("offset"),
+                _f("ready"), _f("duration"), _f("ranges", 120))
+        # Debug: one line every five seconds per viewer. Run the daemon
+        # with --log-level debug to see inside a player that is
+        # misbehaving — it is the only view of the browser there is
+        # when the browser is a phone.
+        else:
+            # `ahead` on its own cannot say whether a short buffer is
+            # the player's own gate holding or the network failing to
+            # keep up, and those two want opposite answers. `limit` is
+            # what the gate is set to for this film and `budget` the
+            # byte budget it was derived from, so the three read as one
+            # sentence.
+            log.debug(
+                "stream: client t=%ss ahead=%ss/%ss budget=%sMB "
+                "ready=%s paused=%s "
+                "stalled=%s q=%s inflight=%s appending=%s updating=%s "
+                "quota=%s ms=%s err=%s ranges=[%s] (sent=%d)",
+                _f("t"), _f("ahead"), _f("limit"), _f("budgetMB"),
+                _f("ready"), _f("paused"),
+                _f("stalled"), _f("q"), _f("inflight"), _f("appending"),
+                _f("updating"), _f("quota"), _f("ms"), _f("err", 80),
+                _f("ranges", 120), self._stream_segments)
+
+    def _do_stream_stop(self) -> None:
+        age = (time.monotonic() - self._stream_started_at
+               if self._stream_started_at else -1)
+        log.info("stream: stop received %.1fs after start, %d segments sent",
+                 age, self._stream_segments)
+        self._stop_stream()