MIDI hold re-adds a peer stuck half-connected

After the network test SuperRack sat in NA Ingest's session as
"initializing"; the hold saw it listed and did nothing ("already").
connect_peer now drops a listed-but-not-live peer and adds it back,
building the second write from the first so a slow re-read cannot leave
it dropped. Tested live today: DVS x4, APPS and INGEST holds, NET clear,
CAPTIONS and BRIDGE restart keys.

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

coremidi.py

@@ -160,9 +160,35 @@ def _edit_session(session_name, edit):
     raise LookupError("no network MIDI session named %r on this Mac" % session_name)
 
 
-def connect_peer(session_name, peer_name, address):
+LIVE_STATES = ("running", "connected")    # "connected" for ~6 s while a peer joins, then "running"
+
+
+def stuck_peer(sess, peer_name):
+    """True if the peer is listed but not live (e.g. stuck "initializing" after a network drop)."""
+    return any(p.get("name") == peer_name and p.get("state") not in LIVE_STATES for p in sess.get("peers", []))
+
+
+def connect_peer(session_name, peer_name, address, settle=time.sleep):
     """Ask macOS to connect `peer_name` at `address` ("10.1.10.206:5004") to the named session,
-    like Audio MIDI Setup's Connect. Returns "already" (it is in the session) or "asked"."""
+    like Audio MIDI Setup's Connect. Returns "already" (it is in the session and live), "asked",
+    or "kicked" (it was listed but stuck: dropped, then connected again)."""
+    with _lock:
+        found = [(e, s) for e, s in _session_entities()
+                 if (s.get("dns-sd-name") or _string_prop(e, "name")) == session_name]
+    if found and stuck_peer(found[0][1], peer_name):
+        ent, sess = found[0]
+        dropped = without_peer(sess, peer_name)
+        with _lock:
+            err = _set_dict_prop(ent, "apple.midirtp.session", dropped)
+        if err:
+            raise OSError("MIDIObjectSetDictionaryProperty failed: %d" % err)
+        settle(1.0)
+        # Built from what was just written, not a fresh read that may still list the old entry.
+        with _lock:
+            err = _set_dict_prop(ent, "apple.midirtp.session", with_peer(dropped, peer_name, address))
+        if err:
+            raise OSError("MIDIObjectSetDictionaryProperty failed: %d" % err)
+        return "kicked"
     result = _edit_session(session_name, lambda sess: with_peer(sess, peer_name, address))
     return "already" if result == "unchanged" else result
 

status_panel.py

@@ -1548,7 +1548,7 @@ def execute_fix(cfg, fix_id, chk):
                 result = coremidi.connect_peer(*args)
             log.info("FIX %s -> %s at %s, session %s%s: %s", label, chk["peer"], chk["address"], chk["session"],
                      (" on " + chk["host"]) if chk.get("host") else "", result)
-            return result in ("asked", "already")
+            return result in ("asked", "already", "kicked")
         if kind == "netgear_poe":
             # Netgear AV switch on the Dante network: power-cycle one PoE port (netgear.py).
             import netgear

tests/test_status_panel.py

@@ -789,6 +789,24 @@ class TestFixes(unittest.TestCase):
         self.assertEqual(back, sess)
         self.assertIsNone(coremidi.without_peer(sess, "Lighting PC"))                    # not there
 
+    def test_a_peer_stuck_initializing_is_kicked(self):
+        import coremidi
+        from unittest import mock
+        sess = {"dns-sd-name": "NA Ingest", "port": 5009, "peers": [
+            {"name": "Lighting PC", "address": "10.1.10.206:5004", "state": "running"},
+            {"name": "SuperRack-LiveBox", "address": "10.1.13.140:5004", "state": "initializing"}]}
+        self.assertTrue(coremidi.stuck_peer(sess, "SuperRack-LiveBox"))
+        self.assertFalse(coremidi.stuck_peer(sess, "Lighting PC"))
+        writes = []
+        with mock.patch.object(coremidi, "_session_entities", return_value=[(1, sess)]), \
+                mock.patch.object(coremidi, "_string_prop", return_value=None), \
+                mock.patch.object(coremidi, "_set_dict_prop", side_effect=lambda e, k, v: writes.append(v) or 0):
+            self.assertEqual(coremidi.connect_peer("NA Ingest", "SuperRack-LiveBox", "10.1.13.140:5004",
+                                                   settle=lambda s: None), "kicked")
+            self.assertEqual(coremidi.connect_peer("NA Ingest", "Lighting PC", "10.1.10.206:5004"), "already")
+        self.assertEqual([[p["name"] for p in w["peers"]] for w in writes],
+                         [["Lighting PC"], ["Lighting PC", "SuperRack-LiveBox"]])   # dropped, then added back
+
     def test_an_off_pc_is_woken(self):
         from unittest import mock
         woke = []