Devices: fixes from review round 7

- Removing every headset leaves none in use (commands fail at once) instead of
  falling back to the `frame` alias.
- ssh goes to the IP that answered for IPv6 too, with a link-local address's
  interface (verified: frame.local over fe80::…%en9 on the real Frame).
- Find results only show in the panel of the headset they were for.

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

tests/test_devices.py

@@ -292,6 +292,11 @@ class Registry(Base):
         self.assertEqual(self.reg.get(b["id"])["user"], "deck")  # a rejected edit changes nothing
         self.reg.remove_device(b["id"])
         self.assertEqual(self.reg.active(), a["id"])
+        self.assertFalse(self.reg.emptied())
+        self.reg.remove_device(a["id"])
+        self.assertTrue(self.reg.emptied())  # the connector then uses no headset at all
+        self.reg.add_device("frame-4")
+        self.assertFalse(self.reg.emptied())
         with self.assertRaises(fd.DeviceError):
             self.reg.get(b["id"])
 

tests/test_link.py

@@ -150,12 +150,12 @@ class Connecting(unittest.TestCase):
     def test_falls_through_to_the_address_that_is_really_the_headset(self):
         # Tried in this order: a name that doesn't resolve, a different device, the headset.
         self.listen6()
-        d = self.device("nothing.invalid", "::1", "localhost")
-        self.hosts({"::1": "wrong", "localhost": "ok"})
+        d = self.device("nothing.invalid", "::1", "127.0.0.1")
+        self.hosts({"::1": "wrong", "127.0.0.1": "ok"})
         self.link.connect(["start"])
         s = self.link.snapshot()
         self.assertEqual(s["phase"], "connected", s["error"])
-        self.assertEqual(s["via"]["host"], "localhost")
+        self.assertEqual(s["via"]["host"], "127.0.0.1")
         self.assertEqual([st["state"] for st in s["stages"]], ["done"] * 5)
         rows = {p["host"]: p for p in s["probes"]}
         self.assertEqual(rows["nothing.invalid"]["state"], "unresolved")
@@ -164,16 +164,15 @@ class Connecting(unittest.TestCase):
         # Every ssh command was pointed at the winner, with the host key pinned per device.
         alias, opts = self.routes[-1]
         self.assertEqual(alias, "frame-t")
-        ip = rows["localhost"]["ip"]  # an IPv4 answer is used as is; an IPv6 one keeps the name
-        self.assertIn("HostName=" + (ip if "." in ip else "localhost"), opts)
+        self.assertIn("HostName=127.0.0.1", opts)
         self.assertIn(f"HostKeyAlias=frame-control-{d['id']}", opts)
         self.assertIn(f"Port={self.port}", opts)
         master = [c for c in self.calls() if "ControlMaster=yes" in c][-1]
         self.assertIn("StrictHostKeyChecking=accept-new", master)  # first connection: nothing pinned yet
         self.assertTrue(fd.known_hosts(d["id"]).parent.is_dir())  # where ssh saves the key it accepts
-        # It learned: localhost works on this network.
+        # It learned: 127.0.0.1 works on this network.
         learned = {a["host"]: a for a in self.reg.get(d["id"])["addresses"]}
-        self.assertEqual(learned["localhost"]["networks"], ["n-test"])
+        self.assertEqual(learned["127.0.0.1"]["networks"], ["n-test"])
         self.assertEqual(learned["::1"]["networks"], [])
         self.assertTrue(self.link.alive())
         self.link.close_master()
@@ -233,6 +232,15 @@ class Connecting(unittest.TestCase):
         self.assertEqual((s["phase"], s["via"]["host"], s["via"]["ip"]), ("connected", "localhost", "127.0.0.1"))
         self.assertIn("HostName=127.0.0.1", self.routes[-1][1])
 
+    def test_no_headset_after_removing_them_all(self):
+        d = self.device("localhost")
+        self.reg.remove_device(d["id"])
+        self.link.connect(["switch"])
+        s = self.link.snapshot()
+        self.assertEqual((s["phase"], s["device"]["id"], s["retry_at"]), ("failed", "none", None))
+        self.assertIn("No headset", s["error"]["message"])
+        self.assertEqual(self.routes[-1], ("frame-control-no-headset", ["-o", "HostName=no-headset.invalid"]))
+
     def test_a_bare_alias_lets_ssh_config_decide(self):
         self.link.override = "frame-bare"
         self.hosts({"frame-bare": "ok"})  # the stand-in ssh has no config: the alias is the host
@@ -354,6 +362,9 @@ class Connecting(unittest.TestCase):
         for body in bad:
             with self.assertRaises(fd.DeviceError, msg=body):
                 fl.devices_action(self.link, body, open_setup=lambda *a: self.fail("setup ran"))
+        # Removing every headset leaves none in use, rather than falling back to the `frame` alias.
+        spare = self.reg.add_device("frame-spare")
+        fl.devices_action(self.link, {"action": "remove", "id": spare["id"]}, None)
         # The only headset, whose ssh alias would stay: not without removing that too.
         (self.dir / "ssh" / "config").write_text("# >>> steam-frame (frame-t) >>>\nHost frame-t\n  HostName localhost\n"
                                                 "Host *\n# <<< steam-frame (frame-t) <<<\n")

ui/frame_devices.py

@@ -486,6 +486,10 @@ class Registry:
         with self.lock:
             return self.data.get("active")
 
+    def emptied(self):
+        with self.lock:
+            return bool(self.data.get("emptied")) and not self.data["devices"]
+
     def set_active(self, device_id):
         with self.lock:
             self._find(device_id)
@@ -510,6 +514,7 @@ class Registry:
                 if host and not any(a["host"] == host for a in d["addresses"]):
                     d["addresses"].append(new_address(host))
             self.data["devices"].append(d)
+            self.data.pop("emptied", None)
             if not self.data.get("active"):
                 self.data["active"] = device_id
             self.save()
@@ -534,6 +539,8 @@ class Registry:
             d = self._find(device_id)
             self.data["devices"].remove(d)
             self.data.setdefault("dismissed", {})[d["alias"]] = d.get("config_host") or ""
+            if not self.data["devices"]:
+                self.data["emptied"] = True  # removed on purpose: don't fall back to the `frame` alias
             if self.data.get("active") == device_id:
                 self.data["active"] = self.data["devices"][0]["id"] if self.data["devices"] else None
             self.save()

ui/frame_link.py

@@ -26,6 +26,7 @@ when the page asks.
 Python stdlib only. Runs on this computer, never on the Frame.
 """
 import copy
+import ipaddress
 import queue
 import re
 import socket
@@ -101,6 +102,11 @@ def probe(host, port, timeout=PROBE_TIMEOUT, update=None):
     infos = infos[:4]
     for n, (family, kind, proto, _, addr) in enumerate(infos):
         ip = addr[0]
+        if family == socket.AF_INET6 and len(addr) > 3 and addr[3] and "%" not in ip:
+            try:  # a link-local IPv6 address only works with its interface
+                ip = f"{ip}%{socket.if_indextoname(addr[3])}"
+            except (OSError, AttributeError):
+                pass
         left = deadline - now()
         if left <= 0:
             break
@@ -126,8 +132,13 @@ def probe(host, port, timeout=PROBE_TIMEOUT, update=None):
 
 
 def ssh_target(host, ip):
-    """Where ssh should go for an address whose probe answered from `ip`."""
-    return ip if ip and re.fullmatch(r"\d{1,3}(\.\d{1,3}){3}", ip) else host
+    """Where ssh should go for an address whose probe answered from `ip`: that IP, so ssh
+    doesn't look the name up again and try an address that didn't answer."""
+    try:
+        ipaddress.ip_address((ip or "").split("%")[0])
+        return ip
+    except ValueError:
+        return host
 
 
 def probe_raw(host, port, result):
@@ -308,7 +319,14 @@ class Link:
             except frame_devices.DeviceError:
                 pass
         devices = self.reg.devices()
-        return devices[0] if devices else self.bare("frame")
+        if devices:
+            return devices[0]
+        if self.reg.emptied():
+            return self.NONE  # every headset was removed: reach nothing until one is added
+        return self.bare("frame")  # never set up here, or set up before the registry existed
+
+    NONE = {"id": "none", "name": "No headset", "alias": "frame-control-no-headset", "user": None, "port": None,
+            "addresses": [], "transient": True, "none": True, "identity_files": []}
 
     @staticmethod
     def bare(alias):
@@ -318,6 +336,8 @@ class Link:
 
     def host_opts(self, device, host):
         """What every ssh command adds to reach DEVICE at HOST."""
+        if device.get("none"):
+            return ["-o", "HostName=no-headset.invalid"]  # fails at once, with ssh's own "can't resolve"
         if device.get("transient") or not host:
             return []
         return ["-o", f"HostName={frame_devices.ssh_host(host)}",
@@ -434,8 +454,8 @@ class Link:
                     self.state.update(phase="connected", retry_at=None, error=None)
                 else:
                     self.fails += 1
-                    self.state.update(phase="failed",
-                                      retry_at=now() + RETRY[min(self.fails, len(RETRY)) - 1])
+                    self.state.update(phase="failed", retry_at=None if device.get("none") else
+                                      now() + RETRY[min(self.fails, len(RETRY)) - 1])
                     if not self.state["error"]:
                         self.state["error"] = {"stage": "find", "message": "Couldn't connect", "raw": ""}
                 self.version += 1
@@ -462,6 +482,9 @@ class Link:
             self.state["error"] = {"stage": sid, "message": message, "raw": raw}
 
     def attempt(self, device):
+        if device.get("none"):
+            self.fail("find", "No headset is set up. Add one on the Devices tab.")
+            return False
         # 1. this computer's network
         self.stage("network", "active")
         net = frame_network.current_network(self.last_fp)
@@ -654,9 +677,8 @@ class Link:
     def handshake(self, device, a, found, user):
         """SSH to one address, following ssh -v through stages 3-5.
         -> "ok", "next" (try another address) or "stop"."""
-        # An IPv4 address that answered is used as is, so ssh doesn't look the name up
-        # again and try an address that didn't answer (a dead IPv6 route, say). IPv6
-        # answers keep the name: a link-local one needs its zone, which ssh adds itself.
+        # The IP that answered (with a link-local IPv6 address's zone), so ssh doesn't
+        # look the name up again and stall on an address that didn't answer.
         opts = self.host_opts(device, ssh_target(a["host"], found.get("ip")))
         alias = device["alias"]
         with self.route_lock:
@@ -862,7 +884,7 @@ def devices_view(link):
     active = link.active_device()
     names = {nid: link.reg.network_name(dict(n, id=nid)) for nid, n in snap["networks"].items()}
     devices = []
-    if active.get("transient"):
+    if active.get("transient") and not active.get("none"):
         devices.append(dict(link.public_device(active), active=True, addresses=[], managed=False, pinned=False))
     for d in snap["devices"]:
         view = {k: v for k, v in d.items() if k not in ("config_host", "addresses")}

ui/index.html

@@ -2812,7 +2812,7 @@ async function findOn(where, d, btn) {
     dv.found = { where, dev: d.id, data: await api(`/api/devices/${where}?id=${encodeURIComponent(d.id)}`) };
   } catch (e) { dv.found = { where, dev: d.id, error: e.message }; }
   finally { btn.disabled = false; }
-  renderFound(d);
+  if (dv.sel === d.id) renderFound(d);  // another headset may be showing by now
 }
 function renderFound(d) {
   const f = dv.found, el = $("dvFound");