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>

2 files changed, 162 insertions, 195 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()
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/dispatch.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/dispatch.py
index cb2db8f..ee2e3a1 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/dispatch.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/dispatch.py
@@ -2,7 +2,6 @@
 of them."""
 
 import logging
-import time
 
 from meshbay_common.protocol import MNP
 
@@ -13,6 +12,96 @@ log = logging.getLogger("meshbay_node.transport.webrtc_server")
 # until the native client removes remote keypair bundles entirely.
 MAX_PRE_PROOF_FETCHES = 4
 
+# After the handshake: which method answers each message type, and whether it
+# runs as a task of this session (`_spawn`) or before the next message is read.
+# That choice is made per type, deliberately; the golden master in the tests
+# records it for every type.
+SPAWNED, INLINE = True, False
+_HANDLERS = {
+    MNP.INDEX_SYNC: ("_do_index_sync", INLINE),
+    # Spawned rather than answered inline: the reply waits for room
+    # on the channel, and blocking the message loop for that would
+    # stop everything else this peer is doing — including the
+    # uploads whose acks free the very buffer we are waiting on.
+    # Chunks are matched by file and index on the client, so
+    # answering out of order is safe.
+    MNP.FILE_REQUEST: ("_do_file_request", SPAWNED),
+    MNP.CHAT_MESSAGE: ("_do_chat_message", INLINE),
+    MNP.CHAT_HISTORY: ("_do_chat_history", INLINE),
+    MNP.LINK_PREVIEW_REQ: ("_do_link_preview_request", SPAWNED),
+    MNP.PING: ("_do_ping", INLINE),
+    MNP.TRANSFER_OPEN: ("_do_transfer_open", INLINE),
+    MNP.TRANSFER_CLOSE: ("_do_transfer_close", INLINE),
+    MNP.FILE_UPLOAD: ("_do_file_upload", SPAWNED),
+    MNP.DIR_CREATE: ("_do_dir_create", SPAWNED),
+    MNP.DIR_DELETE: ("_do_dir_delete", SPAWNED),
+    MNP.FILE_DELETE: ("_do_file_delete", INLINE),
+    MNP.ADMIN_RESPONSE: ("_do_admin_response", INLINE),
+    MNP.INVITE_CREATE: ("_do_invite_create", INLINE),
+    MNP.INVITE_LINK_CREATE: ("_do_invite_link_create", INLINE),
+    MNP.INVITE_CANCEL: ("_do_invite_cancel", INLINE),
+    MNP.MEMBER_REVOKE: ("_do_member_revoke", INLINE),
+    MNP.DEVICE_REQUEST: ("_do_device_request", SPAWNED),
+    MNP.DEVICE_LOOKUP: ("_do_device_lookup", SPAWNED),
+    MNP.DEVICE_ADD: ("_do_device_add", SPAWNED),
+    MNP.DEVICE_LIST: ("_do_device_list", SPAWNED),
+    MNP.DEVICE_REVOKE: ("_do_device_revoke", SPAWNED),
+    MNP.DEVICE_HELLO: ("_do_device_hello", SPAWNED),
+    MNP.APPS_ENABLED: ("_do_apps_enabled", INLINE),
+    MNP.TRANSFER_LIMITS: ("_do_transfer_limits", INLINE),
+    MNP.SET_SCAN_SETTINGS: ("_do_set_scan_settings", INLINE),
+    MNP.TMDB_CONFIG: ("_do_tmdb_config", INLINE),
+    MNP.TMDB_ENABLED: ("_do_tmdb_enabled", INLINE),
+    MNP.APP_DIRECTORIES: ("_do_app_directories", INLINE),
+    MNP.CHAT_DIRECTORY: ("_do_chat_directory", INLINE),
+    MNP.CHAT_LINK_PREVIEW: ("_do_chat_link_preview", INLINE),
+    MNP.SEARCH_LISTED: ("_do_search_listed", INLINE),
+    MNP.CHAT_EPOCH: ("_do_chat_epoch", INLINE),
+    MNP.CHAT_KEYS_REQ: ("_do_chat_keys_req", SPAWNED),
+    MNP.GROUP_ROSTER_REQ: ("_do_group_roster_req", SPAWNED),
+    MNP.MEDIA_META_REQ: ("_do_media_meta_request", SPAWNED),
+    MNP.SEASON_META_REQ: ("_do_season_meta_request", SPAWNED),
+    MNP.TMDB_SEARCH_REQ: ("_do_tmdb_search_request", SPAWNED),
+    MNP.TMDB_OVERRIDE: ("_do_tmdb_override", INLINE),
+    MNP.TMDB_REMATCH: ("_do_tmdb_rematch", INLINE),
+    MNP.MUSICBRAINZ_ENABLED: ("_do_musicbrainz_enabled", INLINE),
+    MNP.MUSIC_META_REQ: ("_do_music_meta_request", SPAWNED),
+    MNP.AUDIO_TRANSCODE_REQ: ("_do_audio_transcode_request", SPAWNED),
+    MNP.SUBTITLE_REQ: ("_do_subtitle_request", SPAWNED),
+    MNP.MEMBER_UNPIN: ("_do_member_unpin", INLINE),
+    MNP.GEK_ROTATE: ("_do_gek_rotate", INLINE),
+    MNP.NODE_STATUS: ("_do_node_status", SPAWNED),
+    MNP.ROOT_ADD: ("_do_root_add", INLINE),
+    MNP.ROOT_REMOVE: ("_do_root_remove", INLINE),
+    MNP.ROOT_UPDATE: ("_do_root_update", INLINE),
+    MNP.ROOT_EJECT: ("_do_root_eject", INLINE),
+    MNP.ROOT_PLUG: ("_do_root_plug", INLINE),
+    MNP.ROSTER_READ: ("_do_roster_read", SPAWNED),
+    MNP.DENYLIST_READ: ("_do_denylist_read", SPAWNED),
+    MNP.DENYLIST_CLEAR: ("_do_denylist_clear", SPAWNED),
+    MNP.GROUP_ATTACH: ("_do_group_attach", INLINE),
+    MNP.GROUP_DETACH: ("_do_group_detach", INLINE),
+    MNP.NODE_SETTINGS_SET: ("_do_node_settings_set", SPAWNED),
+    MNP.NODE_RELOAD: ("_do_node_reload", SPAWNED),
+    MNP.KEYPAIR_BUNDLE_STORE: ("_do_keypair_bundle_store", SPAWNED),
+    MNP.KEYPAIR_BUNDLE_DELETE: ("_do_keypair_bundle_delete", SPAWNED),
+    MNP.USER_BLOB_STORE: ("_do_user_blob_store", SPAWNED),
+    MNP.USER_BLOB_FETCH: ("_do_user_blob_fetch", SPAWNED),
+    MNP.USER_BLOB_LIST: ("_do_user_blob_list", SPAWNED),
+    MNP.USER_BLOB_DELETE: ("_do_user_blob_delete", SPAWNED),
+    MNP.STREAM_REQUEST: ("_do_stream_request", INLINE),
+    MNP.STREAM_MORE: ("_grant_stream_credit", INLINE),
+    'client_diag': ("_do_client_diag", INLINE),
+    MNP.STREAM_STOP: ("_do_stream_stop", INLINE),
+}
+# The handlers that are called without the message.
+_TAKES_NO_MESSAGE = frozenset({
+    "_do_index_sync",
+    "_do_keypair_bundle_delete",
+    "_do_stream_stop",
+    "_do_user_blob_list",
+})
+
 
 class DispatchMixin:
     def _dispatch_message(self, msg: dict) -> None:
@@ -49,201 +138,21 @@ class DispatchMixin:
                 self._spawn(self._do_join_request(msg))
             elif self._user_id is None:
                 self._send({"type": "error", "detail": "Handshake required"})
-            elif mtype == MNP.INDEX_SYNC:
-                self._do_index_sync()
-            elif mtype == MNP.FILE_REQUEST:
-                # Spawned rather than answered inline: the reply waits for room
-                # on the channel, and blocking the message loop for that would
-                # stop everything else this peer is doing — including the
-                # uploads whose acks free the very buffer we are waiting on.
-                # Chunks are matched by file and index on the client, so
-                # answering out of order is safe.
-                self._spawn(self._do_file_request(msg))
-            elif mtype == MNP.CHAT_MESSAGE:
-                self._do_chat_message(msg)
-            elif mtype == MNP.CHAT_HISTORY:
-                self._do_chat_history(msg)
-            elif mtype == MNP.LINK_PREVIEW_REQ:
-                self._spawn(self._do_link_preview_request(msg))
-            elif mtype == MNP.PING:
-                self._do_ping(msg)
-            elif mtype == MNP.TRANSFER_OPEN:
-                self._do_transfer_open(msg)
-            elif mtype == MNP.TRANSFER_CLOSE:
-                self._do_transfer_close(msg)
-            elif mtype == MNP.FILE_UPLOAD:
-                self._spawn(self._do_file_upload(msg))
-            elif mtype == MNP.DIR_CREATE:
-                self._spawn(self._do_dir_create(msg))
-            elif mtype == MNP.DIR_DELETE:
-                self._spawn(self._do_dir_delete(msg))
-            elif mtype == MNP.FILE_DELETE:
-                self._do_file_delete(msg)
-            elif mtype == MNP.ADMIN_RESPONSE:
-                self._do_admin_response(msg)
-            elif mtype == MNP.INVITE_CREATE:
-                self._do_invite_create(msg)
-            elif mtype == MNP.INVITE_LINK_CREATE:
-                self._do_invite_link_create(msg)
-            elif mtype == MNP.INVITE_CANCEL:
-                self._do_invite_cancel(msg)
-            elif mtype == MNP.MEMBER_REVOKE:
-                self._do_member_revoke(msg)
-            elif mtype == MNP.DEVICE_REQUEST:
-                self._spawn(self._do_device_request(msg))
-            elif mtype == MNP.DEVICE_LOOKUP:
-                self._spawn(self._do_device_lookup(msg))
-            elif mtype == MNP.DEVICE_ADD:
-                self._spawn(self._do_device_add(msg))
-            elif mtype == MNP.DEVICE_LIST:
-                self._spawn(self._do_device_list(msg))
-            elif mtype == MNP.DEVICE_REVOKE:
-                self._spawn(self._do_device_revoke(msg))
-            elif mtype == MNP.DEVICE_HELLO:
-                self._spawn(self._do_device_hello(msg))
-            elif mtype == MNP.APPS_ENABLED:
-                self._do_apps_enabled(msg)
-            elif mtype == MNP.TRANSFER_LIMITS:
-                self._do_transfer_limits(msg)
-            elif mtype == MNP.SET_SCAN_SETTINGS:
-                self._do_set_scan_settings(msg)
-            elif mtype == MNP.TMDB_CONFIG:
-                self._do_tmdb_config(msg)
-            elif mtype == MNP.TMDB_ENABLED:
-                self._do_tmdb_enabled(msg)
-            elif mtype == MNP.APP_DIRECTORIES:
-                self._do_app_directories(msg)
-            elif mtype == MNP.CHAT_DIRECTORY:
-                self._do_chat_directory(msg)
-            elif mtype == MNP.CHAT_LINK_PREVIEW:
-                self._do_chat_link_preview(msg)
-            elif mtype == MNP.SEARCH_LISTED:
-                self._do_search_listed(msg)
-            elif mtype == MNP.CHAT_EPOCH:
-                self._do_chat_epoch(msg)
-            elif mtype == MNP.CHAT_KEYS_REQ:
-                self._spawn(self._do_chat_keys_req(msg))
-            elif mtype == MNP.GROUP_ROSTER_REQ:
-                self._spawn(self._do_group_roster_req(msg))
-            elif mtype == MNP.MEDIA_META_REQ:
-                self._spawn(self._do_media_meta_request(msg))
-            elif mtype == MNP.SEASON_META_REQ:
-                self._spawn(self._do_season_meta_request(msg))
-            elif mtype == MNP.TMDB_SEARCH_REQ:
-                self._spawn(self._do_tmdb_search_request(msg))
-            elif mtype == MNP.TMDB_OVERRIDE:
-                self._do_tmdb_override(msg)
-            elif mtype == MNP.TMDB_REMATCH:
-                self._do_tmdb_rematch(msg)
-            elif mtype == MNP.MUSICBRAINZ_ENABLED:
-                self._do_musicbrainz_enabled(msg)
-            elif mtype == MNP.MUSIC_META_REQ:
-                self._spawn(self._do_music_meta_request(msg))
-            elif mtype == MNP.AUDIO_TRANSCODE_REQ:
-                self._spawn(self._do_audio_transcode_request(msg))
-            elif mtype == MNP.SUBTITLE_REQ:
-                self._spawn(self._do_subtitle_request(msg))
-            elif mtype == MNP.MEMBER_UNPIN:
-                self._do_member_unpin(msg)
-            elif mtype == MNP.GEK_ROTATE:
-                self._do_gek_rotate(msg)
-            elif mtype == MNP.NODE_STATUS:
-                self._spawn(self._do_node_status(msg))
-            elif mtype == MNP.ROOT_ADD:
-                self._do_root_add(msg)
-            elif mtype == MNP.ROOT_REMOVE:
-                self._do_root_remove(msg)
-            elif mtype == MNP.ROOT_UPDATE:
-                self._do_root_update(msg)
-            elif mtype == MNP.ROOT_EJECT:
-                self._do_root_eject(msg)
-            elif mtype == MNP.ROOT_PLUG:
-                self._do_root_plug(msg)
-            elif mtype == MNP.ROSTER_READ:
-                self._spawn(self._do_roster_read(msg))
-            elif mtype == MNP.DENYLIST_READ:
-                self._spawn(self._do_denylist_read(msg))
-            elif mtype == MNP.DENYLIST_CLEAR:
-                self._spawn(self._do_denylist_clear(msg))
-            elif mtype == MNP.GROUP_ATTACH:
-                self._do_group_attach(msg)
-            elif mtype == MNP.GROUP_DETACH:
-                self._do_group_detach(msg)
-            elif mtype == MNP.NODE_SETTINGS_SET:
-                self._spawn(self._do_node_settings_set(msg))
-            elif mtype == MNP.NODE_RELOAD:
-                self._spawn(self._do_node_reload(msg))
-            elif mtype == MNP.KEYPAIR_BUNDLE_STORE:
-                self._spawn(self._do_keypair_bundle_store(msg))
-            elif mtype == MNP.KEYPAIR_BUNDLE_DELETE:
-                self._spawn(self._do_keypair_bundle_delete())
-            elif mtype == MNP.USER_BLOB_STORE:
-                self._spawn(self._do_user_blob_store(msg))
-            elif mtype == MNP.USER_BLOB_FETCH:
-                self._spawn(self._do_user_blob_fetch(msg))
-            elif mtype == MNP.USER_BLOB_LIST:
-                self._spawn(self._do_user_blob_list())
-            elif mtype == MNP.USER_BLOB_DELETE:
-                self._spawn(self._do_user_blob_delete(msg))
-            elif mtype == MNP.STREAM_REQUEST:
-                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))
-            elif mtype == MNP.STREAM_MORE:
-                self._grant_stream_credit(msg)
-            elif mtype == "client_diag":
-                # 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)
-            elif mtype == MNP.STREAM_STOP:
-                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()
             else:
-                log.warning("Unknown MNP message type on DataChannel: %s", mtype)
+                try:
+                    entry = _HANDLERS.get(mtype)
+                except TypeError:
+                    # A list or a map is a legal msgpack value for `type` and not a
+                    # key: it names no handler, as it matched no comparison before.
+                    entry = None
+                if entry is None:
+                    log.warning("Unknown MNP message type on DataChannel: %s", mtype)
+                else:
+                    name, spawned = entry
+                    method = getattr(self, name)
+                    result = method() if name in _TAKES_NO_MESSAGE else method(msg)
+                    if spawned:
+                        self._spawn(result)
         except Exception as e:
             # Log the detail locally; send the peer a generic message. Exception
             # text here carries filesystem paths and internal state (finding L3).