Devices with the Mac view and the MCP adapter: the tunnel follows the headset, a private server runs alongside

The Mac view's tunnel is its own ssh, so it now takes the headset's route (and its
pinned identity, which also checks the USB-C address), and closes when the app
switches headset. The MCP adapter's private server (FRAME_PRIVATE_SSH=1) skips the
one-server lock and can't add, remove or switch headsets.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

tests/test_macview.py

@@ -42,6 +42,23 @@ class Helpers(unittest.TestCase):
         self.assertEqual(h, 1080)
         self.assertAlmostEqual(w / h, 0.5, places=2)
 
+    def test_the_tunnel_follows_the_headset(self):
+        mv = frame_macview.MacView(["ssh"], lambda *a, **k: "", "frame")
+        mv.retarget("frame", ["-o", "ControlPath=/tmp/x-%C", "-o", "HostName=192.0.2.1", "-o", "HostKeyAlias=frame-control-a"])
+        self.assertEqual(mv.host_opts, ["-o", "HostName=192.0.2.1", "-o", "HostKeyAlias=frame-control-a"])
+
+        class Tunnel:
+            ended = False
+            def poll(self): return None
+            def terminate(self): Tunnel.ended = True
+        mv.tunnel, mv.remote_port = Tunnel(), 47999
+        mv.retarget("frame", ["-o", "HostName=192.0.2.9"])  # another address, same headset: keep it
+        self.assertFalse(Tunnel.ended)
+        mv.retarget("frame-2", ["-o", "HostName=192.0.2.2"])  # another headset: never the old one's tunnel
+        self.assertTrue(Tunnel.ended)
+        self.assertIsNone(mv.tunnel)
+        self.assertEqual(mv.frame, "frame-2")
+
     @unittest.skipUnless(shutil.which("bash"), "needs bash")
     @unittest.skipIf(os.name == "nt", "Windows' bash.exe is WSL's launcher, and runners have no distribution")
     def test_launch_script_parses(self):

tests/test_server.py

@@ -260,6 +260,22 @@ class OneServer(unittest.TestCase):
         first.wait(10)
         self.assertIn("Frame Control on", self.start(env).stdout.readline())
 
+    def test_a_private_server_runs_alongside_but_cant_change_headsets(self):
+        """The MCP adapter starts its own server (FRAME_PRIVATE_SSH=1) while the app runs."""
+        data = tempfile.mkdtemp(prefix="frame-one-server-")
+        env = {**os.environ, "FRAME_CONTROL_DATA_DIR": data, "FRAME_ALIAS": "frame-control-test.invalid",
+               "FRAME_CONTROL_SERVER_WAIT": "1"}
+        self.assertIn("Frame Control on", self.start(env).stdout.readline())
+        private = self.start({**env, "FRAME_PRIVATE_SSH": "1"})
+        line = private.stdout.readline()
+        self.assertIn("Frame Control on", line)
+        port = int(line.split("http://127.0.0.1:")[1].split()[0])
+        conn = http.client.HTTPConnection("127.0.0.1", port, timeout=10)
+        conn.request("POST", "/api/devices", body=json.dumps({"action": "use", "id": "x"}),
+                     headers={"Content-Type": "application/json", "X-Frame-UI": "1", "Host": f"127.0.0.1:{port}"})
+        r = conn.getresponse()
+        self.assertEqual(r.status, 403, r.read())
+
 
 @unittest.skipIf(os.name == "nt", "runs on the Frame (Linux); local-bin/ssh is a POSIX shell script")
 class LocalMode(unittest.TestCase):

ui/frame_macview.py

@@ -120,6 +120,7 @@ class MacView:
         self.tunnel_ssh = list(tunnel_ssh)
         self.run = run
         self.frame = frame
+        self.host_opts = []  # the headset in use and how to reach it (see retarget)
         self.track = track or (lambda proc: None)  # the server ends these on exit
         self.lock = threading.Lock()
         self.token = secrets.token_urlsafe(24)
@@ -239,11 +240,32 @@ class MacView:
                 last = self._last_tunnel_error
             raise MacViewError(f"Couldn't open a tunnel from {self.frame} to this Mac: {last or 'no answer through it'}")
 
+    def retarget(self, alias, host_opts):
+        """The server now reaches the headset as `alias` with `host_opts` (another address,
+        or another headset). No lock: this runs while the server routes, which a tunnel
+        being opened may be waiting on; each assignment is atomic."""
+        # host_opts is "-o", "Name=value" pairs; the tunnel keeps its own connection,
+        # not the shared master.
+        opts = [x for flag, value in zip(host_opts[::2], host_opts[1::2])
+                if not value.startswith("ControlPath=") for x in (flag, value)]
+        moved = alias != self.frame
+        self.frame, self.host_opts = alias, opts
+        tunnel = self.tunnel
+        if moved and tunnel is not None:
+            # Another headset: its viewers can't be the old one's. The supervisor
+            # reopens a tunnel to the new one if anything is being shown.
+            self.tunnel, self.remote_port = None, None
+            if tunnel.poll() is None:
+                tunnel.terminate()
+
     def _open_tunnel(self, via, ports):
         """Tries the ports on one route; True once the tunnel answers. With self.lock held."""
         last = ""
         for port in ports:
-            proc = subprocess.Popen([*self.tunnel_ssh, *via, "-o", "ExitOnForwardFailure=yes",
+            # `via` first: ssh keeps the first value of an option, so USB-C's HostName wins
+            # while the headset's pinned identity (in host_opts) still checks it.
+            proc = subprocess.Popen([*self.tunnel_ssh, "-o", "ControlPath=none", *via, *self.host_opts,
+                                     "-o", "ExitOnForwardFailure=yes",
                                      "-o", "ServerAliveInterval=5", "-o", "ServerAliveCountMax=3", "-N",
                                      "-R", f"127.0.0.1:{port}:127.0.0.1:{self.port}", self.frame],
                                     stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
@@ -294,6 +316,8 @@ class MacView:
             socket.create_connection((ip, 22), timeout=1).close()
         except OSError:
             return []  # not plugged into this Mac
+        if any(o.startswith("HostKeyAlias=") for o in self.host_opts):
+            return ["-o", f"HostName={ip}"]  # checked against the headset's own pinned key
         alias = self.frame
         try:
             cfg = subprocess.run(["ssh", "-G", self.frame], capture_output=True, text=True, timeout=5).stdout

ui/server.py

@@ -71,7 +71,10 @@ if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*", FRAME):
     sys.exit(f"FRAME_ALIAS must be a plain host alias, not {FRAME!r}")
 # Reuse one SSH connection for the frequent status/screenshot calls, where ssh
 # supports it (not on Windows: there every command connects on its own).
-CONTROL = None if LOCAL else frame_host.control_path(private=os.environ.get("FRAME_PRIVATE_SSH") == "1")
+# A private server (the MCP adapter starts one per session) keeps its own SSH masters, and
+# uses the headsets without editing them.
+PRIVATE = os.environ.get("FRAME_PRIVATE_SSH") == "1"
+CONTROL = None if LOCAL else frame_host.control_path(private=PRIVATE)
 # The ControlPath itself is per headset: the connector puts it in HOST_OPTS.
 MUX = ["ssh", "-o", "BatchMode=yes"]
 MUX_BASE = list(MUX)
@@ -101,6 +104,9 @@ def route(alias, host_opts):
         MUX[:] = [*MUX_BASE, *HOST_OPTS]
         SSH[:] = [*MUX, *SSH_TAIL]
         frame_android.SSH_OPTS = SSH[1:]
+    mv = globals().get("macview")
+    if mv:
+        mv.retarget(alias, host_opts)  # its tunnel is its own ssh: it must follow the headset too
 
 
 LINK = None  # the connector (frame_link.Link); None on the Frame itself
@@ -1436,6 +1442,8 @@ def open_setup(alias, host=None):
 def devices_post(body):
     if not LINK:
         raise Failure("Headsets are managed from the computer app", 400)
+    if PRIVATE:
+        raise Failure("Headsets are managed in the Frame Control app", 403)
     try:
         # Under the work lock: nothing can start on the old headset while it switches.
         with _work_lock:
@@ -1846,7 +1854,8 @@ def one_server():
     """Only one Frame Control server per user: two would each connect, reconnect and
     edit the headsets on their own, and could move each other's installs to another
     headset. Held until this process exits. (FRAME_CONTROL_DATA_DIR gives a second,
-    separate one, as the tests do.)"""
+    separate one, as the tests do. A private server, the MCP adapter's, runs alongside:
+    it can't add, remove or switch headsets.)"""
     lock = frame_devices.file_lock(frame_host.data_dir("server.lock"), timeout=float(os.environ.get("FRAME_CONTROL_SERVER_WAIT") or 20))  # while the app restarts it
     try:
         lock.__enter__()
@@ -1868,7 +1877,8 @@ def main():
     frame_telemetry.start()
     global LINK, _ONE_SERVER
     if not LOCAL:
-        _ONE_SERVER = one_server()
+        if not PRIVATE:  # a private server only uses the headsets (see one_server)
+            _ONE_SERVER = one_server()
         LINK = frame_link.Link(frame_devices.Registry(), env_alias=FRAME if FRAME_FROM_ENV else None,
                                mux_base=MUX_BASE, control=CONTROL, apply=route, explain=unreachable)
         LINK.work_lock, LINK.work = _work_lock, lambda: _work[0]