Never reboot a HyperDeck to fix a connection that is switched off; turn connections back on instead.

On 2026-10-08 a hold on HD1, whose Companion connection had been disabled,
sent the deck a reboot. Now a disabled connection on any button is fixed by
enabling it, before any other fix is considered. The HyperDeck fix asks the
deck itself (transport info on port 9993): if it answers - recording or not -
only Companion's connection is restarted; only a deck that accepts a
connection and says nothing at all is rebooted; an unreachable deck is left
alone and the button shows FIX FAILED. Tested against a fake deck.

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

tests/test_status_panel.py

@@ -469,11 +469,19 @@ class TestFixes(unittest.TestCase):
         for i in (2, 3, 5):
             self.snap.conns["HyperDeck_%d" % i]["category"] = "error"
         self.snap.vars[("HyperDeck_3", "status")] = "Record"          # last known state: recording
-        self.assertEqual(self.plan("rec"), [("fix", "hd_reboot"), ("fix", "hd_reboot")])
+        self.assertEqual(self.plan("rec"), [("fix", "hyperdeck"), ("fix", "hyperdeck")])
         decks = [chk["conn"] for _, _, _, chk in self.engine.fix_plan("rec", self.snap)]
         self.assertEqual(decks, ["HyperDeck_2", "HyperDeck_5"])
         self.assertEqual(self.plan("hd3"), [])
 
+    def test_disabled_connection_is_switched_on_never_rebooted(self):
+        # 2026-10-08: HyperDeck_1 disabled in Companion, a hold sent the deck a reboot.
+        self.snap.conns["HyperDeck_1"]["enabled"] = False
+        self.snap.conns["x32"]["enabled"] = False
+        self.assertEqual(self.plan("hd1"), [("enable", "HyperDeck_1")])
+        self.assertEqual(self.plan("rec"), [("enable", "HyperDeck_1")])
+        self.assertEqual(self.plan("x32"), [("enable", "x32")])
+
     def test_not_recording_is_not_fixed_by_a_reboot(self):
         for i in (1, 3):
             self.snap.vars[("HyperDeck_%d" % i, "status")] = "Record"
@@ -559,7 +567,79 @@ class TestFixes(unittest.TestCase):
 
     def test_every_fix_is_valid(self):
         for f in CONFIG["fixes"].values():
-            self.assertIn(f.get("kind", "ssh"), ("ssh", "local", "hyperdeck_reboot"))
+            self.assertIn(f.get("kind", "ssh"), ("ssh", "local", "hyperdeck"))
+
+
+class FakeDeck:
+    """A pretend HyperDeck on a local port: answers like a real one, stays silent, or greets only."""
+
+    def __init__(self, mode):
+        import socket as _s
+        self.mode, self.received = mode, b""
+        self.srv = _s.socket()
+        self.srv.bind(("127.0.0.1", 0))
+        self.srv.listen(5)
+        self.port = self.srv.getsockname()[1]
+        threading.Thread(target=self.serve, daemon=True).start()
+
+    def serve(self):
+        while True:
+            try:
+                conn, _ = self.srv.accept()
+            except OSError:
+                return
+            with conn:
+                conn.settimeout(4)
+                if self.mode != "silent":
+                    conn.sendall(b"500 connection info:\r\nprotocol version: 1.11\r\nmodel: HyperDeck Studio 4K Pro\r\n\r\n")
+                try:
+                    data = conn.recv(1024)
+                    self.received += data
+                    if self.mode in ("record", "preview") and b"transport info" in data:
+                        conn.sendall(b"208 transport info:\r\nstatus: %s\r\nspeed: 100\r\n\r\n" % self.mode.encode())
+                    more = conn.recv(1024)
+                    self.received += more
+                except OSError:
+                    pass
+
+    def close(self):
+        self.srv.close()
+
+
+class TestHyperDeckProbe(unittest.TestCase):
+    def test_probe_answers(self):
+        for mode, expected in (("record", "record"), ("preview", "preview"), ("greet", "answered"), ("silent", "hung")):
+            deck = FakeDeck(mode)
+            try:
+                self.assertEqual(sp.hyperdeck_probe("127.0.0.1", timeout=1, port=deck.port), expected, mode)
+            finally:
+                deck.close()
+        deck = FakeDeck("preview")
+        port = deck.port
+        deck.close()
+        self.assertEqual(sp.hyperdeck_probe("127.0.0.1", timeout=1, port=port), "unreachable")
+
+    def test_only_a_silent_deck_is_rebooted(self):
+        deck = FakeDeck("silent")
+        try:
+            chk = {"conn": "HyperDeck_1", "ip": "127.0.0.1", "port": deck.port}
+            self.assertTrue(sp.execute_fix(CONFIG, "hyperdeck", chk))
+            time.sleep(0.3)
+            self.assertIn(b"reboot", deck.received)
+        finally:
+            deck.close()
+
+    def test_a_recording_deck_is_never_rebooted(self):
+        deck = FakeDeck("record")
+        try:
+            cfg = json.loads(json.dumps(CONFIG))
+            cfg["companion"]["port"] = 1                      # no Companion here: the reconnect just fails
+            chk = {"conn": "HyperDeck_1", "ip": "127.0.0.1", "port": deck.port}
+            sp.execute_fix(cfg, "hyperdeck", chk)
+            time.sleep(0.3)
+            self.assertNotIn(b"reboot", deck.received)
+        finally:
+            deck.close()
 
 
 class FakeCompanion(BaseHTTPRequestHandler):