PWA: scope RPC responses to the requesting tab, not every tab

SidecarRelay now rewrites each request's id to a server-assigned one on
the way to the sidecar and back on the way out, tracked against the
originating WebSocket. A result/error only ever reaches the tab that
made the matching request; notify events (repl_data and friends) still
broadcast to every connected tab, since those are board-initiated, not a
reply to anyone. Fixes the cross-wire case #20 reports: two tabs each
starting their own id counter at 1 could otherwise have a response meant
for one land in the other. Closes #20.

Author: bdbarnett

cli/src/mpftp/pwa.py

@@ -5,12 +5,15 @@
 
 Architecture: this process owns one long-lived ``mpftp.sidecar`` subprocess
 (the same JSON-lines protocol the CLI and the VS Code extension speak) and
-relays it to the browser over a hand-rolled WebSocket — every line the
-sidecar writes to stdout is broadcast to connected browser tabs verbatim,
-and every WS text frame from a tab is written to the sidecar's stdin
-verbatim. The server implements no RPC semantics of its own; the sidecar's
+relays it to the browser over a hand-rolled WebSocket. The sidecar's
 existing method surface (``connect``, ``fs_*``, ``repl_start``/``repl_data``,
-...) is unchanged and reused as-is.
+...) is unchanged and reused as-is — the relay only touches request/response
+``id`` fields, rewriting each to a server-assigned id so two browser tabs
+independently starting their own id counter at 1 can never collide on the
+one shared sidecar process (mpftp#20); a ``result``/``error`` is routed back
+only to the tab that made the matching request, while unsolicited
+``notify`` events (``repl_data`` and friends) still broadcast to every
+connected tab, since those are board-initiated, not a reply to anyone.
 
 This HTTP/WS port is a separate, explicitly launched service — distinct from
 the VS Code extension's agent RPC port (ephemeral, closed until a board
@@ -26,6 +29,7 @@
 import base64
 import hashlib
 import http.server
+import json
 import os
 import socket
 import socketserver
@@ -34,7 +38,7 @@
 import threading
 import webbrowser
 from pathlib import Path
-from typing import Optional
+from typing import Any, Optional
 
 DEFAULT_PORT = 8317
 
@@ -156,7 +160,15 @@ def close(self) -> None:
 
 
 class SidecarRelay:
-    """One long-lived ``mpftp.sidecar`` subprocess, broadcast to every connected tab."""
+    """One long-lived ``mpftp.sidecar`` subprocess, shared by every connected tab.
+
+    Each tab's own request ``id`` is rewritten to a server-assigned one on
+    the way to the sidecar (and rewritten back on the way out), so two tabs
+    independently starting their own counter at 1 can never collide on the
+    single shared sidecar (mpftp#20) — a ``result``/``error`` only ever goes
+    back to the tab that made the matching request. ``notify`` events (and
+    anything without a recognized pending id) still broadcast to everyone.
+    """
 
     def __init__(self, python: str) -> None:
         from .cli import _wslenv_forwarded_env
@@ -173,6 +185,8 @@ def __init__(self, python: str) -> None:
         )
         self._lock = threading.Lock()
         self._subscribers: list[WebSocket] = []
+        self._next_id = 1
+        self._pending: dict[int, tuple[WebSocket, Any]] = {}
         self._reader = threading.Thread(target=self._pump, daemon=True)
         self._reader.start()
 
@@ -181,7 +195,25 @@ def _pump(self) -> None:
         for line in self.proc.stdout:
             line = line.rstrip("\n")
             if line:
-                self._broadcast(line)
+                self._route(line)
+
+    def _route(self, line: str) -> None:
+        try:
+            msg = json.loads(line)
+        except json.JSONDecodeError:
+            msg = None
+        if isinstance(msg, dict) and msg.get("type") in ("result", "error") and "id" in msg:
+            with self._lock:
+                entry = self._pending.pop(msg["id"], None)
+            if entry is not None:
+                ws, client_id = entry
+                msg["id"] = client_id
+                try:
+                    ws.send_text(json.dumps(msg))
+                except OSError:
+                    self.unsubscribe(ws)
+                return
+        self._broadcast(line)
 
     def _broadcast(self, line: str) -> None:
         with self._lock:
@@ -200,9 +232,23 @@ def unsubscribe(self, ws: WebSocket) -> None:
         with self._lock:
             if ws in self._subscribers:
                 self._subscribers.remove(ws)
+            stale = [rid for rid, (pending_ws, _) in self._pending.items() if pending_ws is ws]
+            for rid in stale:
+                del self._pending[rid]
 
-    def send(self, line: str) -> None:
+    def send(self, line: str, ws: WebSocket) -> None:
         assert self.proc.stdin
+        try:
+            msg = json.loads(line)
+        except json.JSONDecodeError:
+            msg = None
+        if isinstance(msg, dict) and "id" in msg:
+            with self._lock:
+                server_id = self._next_id
+                self._next_id += 1
+                self._pending[server_id] = (ws, msg["id"])
+            msg["id"] = server_id
+            line = json.dumps(msg)
         self.proc.stdin.write(line + "\n")
         self.proc.stdin.flush()
 
@@ -271,7 +317,7 @@ def _handle_ws_upgrade(self) -> None:
                 text = ws.recv_text()
                 if text is None:
                     break
-                relay.send(text)
+                relay.send(text, ws)
         except OSError:
             pass
         finally:

cli/tests/test_pwa.py

@@ -172,23 +172,74 @@ def _relay_with_fake_proc(self, lines):
         relay._reader.join(timeout=2)
         return relay, proc
 
-    def test_broadcasts_every_sidecar_line_to_subscribed_sockets(self):
-        relay, proc = self._relay_with_fake_proc(
-            ['{"type":"notify","method":"repl_data"}\n', '{"type":"result","id":1}\n']
-        )
+    def test_broadcasts_notify_events_to_every_subscribed_socket(self):
+        relay, proc = self._relay_with_fake_proc([])
         ws = mock.Mock()
         relay.subscribe(ws)
-        # The reader thread already drained proc.stdout by the time it joined above,
-        # so re-broadcast is exercised directly for a deterministic assertion.
-        relay._broadcast('{"type":"notify","method":"repl_data"}')
+        relay._route('{"type":"notify","method":"repl_data"}')
         ws.send_text.assert_called_with('{"type":"notify","method":"repl_data"}')
 
-    def test_send_writes_a_newline_terminated_line_to_stdin(self):
+    def test_send_writes_a_newline_terminated_line_with_a_rewritten_id(self):
         relay, proc = self._relay_with_fake_proc([])
-        relay.send('{"method":"ping"}')
-        proc.stdin.write.assert_called_once_with('{"method":"ping"}\n')
+        ws = mock.Mock()
+        relay.send('{"id": 1, "method": "ping"}', ws)
+        written = proc.stdin.write.call_args[0][0]
+        self.assertTrue(written.endswith("\n"))
+        sent = self.mod.json.loads(written)
+        self.assertEqual(sent["method"], "ping")
+        # The id sent to the sidecar is the relay's own counter value, tracked
+        # in _pending against the tab's original id (1) -- not asserting it
+        # merely differs from 1, which the first request could coincide with.
+        self.assertEqual(relay._pending, {sent["id"]: (ws, 1)})
         proc.stdin.flush.assert_called_once()
 
+    def test_a_result_routes_back_only_to_the_requesting_socket_with_its_own_id(self):
+        relay, proc = self._relay_with_fake_proc([])
+        ws_a, ws_b = mock.Mock(), mock.Mock()
+        relay.subscribe(ws_a)
+        relay.subscribe(ws_b)
+        relay.send('{"id": 1, "method": "eval"}', ws_a)
+        server_id = self.mod.json.loads(proc.stdin.write.call_args[0][0])["id"]
+
+        relay._route(self.mod.json.dumps({"type": "result", "id": server_id, "result": {"value": "2"}}))
+
+        ws_a.send_text.assert_called_once()
+        reply = self.mod.json.loads(ws_a.send_text.call_args[0][0])
+        self.assertEqual(reply["id"], 1)  # rewritten back to ws_a's own id
+        ws_b.send_text.assert_not_called()
+
+    def test_two_tabs_reusing_the_same_client_id_do_not_cross_wire(self):
+        """The bug mpftp#20 reports: independent per-tab id counters can collide."""
+        relay, proc = self._relay_with_fake_proc([])
+        ws_a, ws_b = mock.Mock(), mock.Mock()
+        relay.subscribe(ws_a)
+        relay.subscribe(ws_b)
+
+        relay.send('{"id": 1, "method": "connect"}', ws_a)
+        server_id_a = self.mod.json.loads(proc.stdin.write.call_args[0][0])["id"]
+        relay.send('{"id": 1, "method": "list_ports"}', ws_b)
+        server_id_b = self.mod.json.loads(proc.stdin.write.call_args[0][0])["id"]
+        self.assertNotEqual(server_id_a, server_id_b)
+
+        # B's response arrives first; A must not receive it even though both
+        # tabs used client id 1.
+        relay._route(self.mod.json.dumps({"type": "result", "id": server_id_b, "result": []}))
+        ws_a.send_text.assert_not_called()
+        reply_b = self.mod.json.loads(ws_b.send_text.call_args[0][0])
+        self.assertEqual(reply_b["id"], 1)
+
+        relay._route(self.mod.json.dumps({"type": "result", "id": server_id_a, "result": {"ok": True}}))
+        reply_a = self.mod.json.loads(ws_a.send_text.call_args[0][0])
+        self.assertEqual(reply_a["id"], 1)
+        self.assertEqual(ws_b.send_text.call_count, 1)  # unchanged since its own reply
+
+    def test_a_result_with_no_matching_pending_id_falls_back_to_broadcast(self):
+        relay, proc = self._relay_with_fake_proc([])
+        ws = mock.Mock()
+        relay.subscribe(ws)
+        relay._route('{"type":"result","id":999,"result":{}}')
+        ws.send_text.assert_called_once_with('{"type":"result","id":999,"result":{}}')
+
     def test_a_dead_subscriber_is_dropped_after_a_failed_send(self):
         relay, proc = self._relay_with_fake_proc([])
         ws = mock.Mock()
@@ -205,6 +256,14 @@ def test_unsubscribe_stops_further_broadcasts(self):
         relay._broadcast("line")
         ws.send_text.assert_not_called()
 
+    def test_unsubscribe_drops_that_sockets_pending_requests(self):
+        relay, proc = self._relay_with_fake_proc([])
+        ws = mock.Mock()
+        relay.subscribe(ws)
+        relay.send('{"id": 1, "method": "eval"}', ws)
+        relay.unsubscribe(ws)
+        self.assertEqual(relay._pending, {})
+
 
 if __name__ == "__main__":
     unittest.main()