fix(runtime): wait for launchd job removal before service handoff

loopx/hosts/managed_service_runtime.py

@@ -96,6 +96,21 @@ def _launchctl(*args: str) -> None:
         raise RuntimeError("service_runtime_alignment_failed")
 
 
+def _wait_unloaded(label: str) -> None:
+    # bootout can return before launchd removes the job. Loading its replacement
+    # immediately can fail in both the forward handoff and rollback. Keep the
+    # ingress fence until the original manager proves the label is gone.
+    deadline = time.monotonic() + 5
+    while True:
+        result = subprocess.run(["launchctl", "print", label],
+            capture_output=True, timeout=5)
+        if result.returncode == 113:  # launchd: service not found
+            return
+        if result.returncode or time.monotonic() >= deadline:
+            raise RuntimeError("service_runtime_alignment_failed")
+        time.sleep(0.05)
+
+
 def has_runtime_binding() -> bool:
     for kind in ("status", "chat"):
         path = Path.home() / "Library/LaunchAgents" / f"com.loopx.{kind}.plist"
@@ -240,9 +255,11 @@ def _recover_alignment(journal: Path) -> None:
         try:
             for path in loaded:
                 _launchctl("bootout", f"{domain}/{path.stem}")
+                _wait_unloaded(f"{domain}/{path.stem}")
             for path, content in originals.items():
                 write_bytes_atomic(path, content, preserve_mode=True, fsync_directory=True)
             for path in originals:
+                _wait_unloaded(f"{domain}/{path.stem}")
                 _launchctl("bootstrap", domain, str(path))
         except (OSError, RuntimeError, subprocess.SubprocessError) as exc:
             raise RuntimeError("service_runtime_rollback_failed") from exc
@@ -312,6 +329,7 @@ def _align_managed_runtime(executable: str, journal: Path) -> dict:
             for plist in updates:
                 _launchctl("bootout", f"{domain}/{plist.stem}")
                 stopped.append(plist)
+                _wait_unloaded(f"{domain}/{plist.stem}")
             for plist, content in updates.items():
                 if plist.read_bytes() != originals[plist]:
                     raise RuntimeError("service_runtime_route_unverified")
@@ -327,11 +345,13 @@ def _align_managed_runtime(executable: str, journal: Path) -> dict:
             try:
                 for plist in restarted:
                     _launchctl("bootout", f"{domain}/{plist.stem}")
+                    _wait_unloaded(f"{domain}/{plist.stem}")
                 for plist in written:
                     if plist.read_bytes() != updates[plist]:
                         raise RuntimeError("service_runtime_route_unverified")
                     write_bytes_atomic(plist, originals[plist], preserve_mode=True, fsync_directory=True)
                 for plist in stopped:
+                    _wait_unloaded(f"{domain}/{plist.stem}")
                     _launchctl("bootstrap", domain, str(plist))
             except (OSError, RuntimeError, subprocess.SubprocessError) as exc:
                 raise RuntimeError("service_runtime_rollback_failed") from exc

tests/test_managed_service_runtime.py

@@ -60,7 +60,7 @@ def launchctl(args, **kwargs):
         label = Path(args[-1]).stem if verb == "bootstrap" else args[-1].split("/")[-1]
         if verb == "print":
             text = "arguments = {\n" + "\n".join(loaded[label]["ProgramArguments"]) + "\n}\npid = 900\n" if label in loaded else ""
-            return subprocess.CompletedProcess(args, 0 if label in loaded else 1, text, "")
+            return subprocess.CompletedProcess(args, 0 if label in loaded else 113, text, "")
         calls.append((verb, label))
         if fail and (verb, label) == fail[0]:
             fail.pop(0)
@@ -95,6 +95,62 @@ def test_pair_adopts_target_without_changing_routes_or_cli_and_is_idempotent(man
     assert calls == []
 
 
+@pytest.mark.parametrize("rollback", [False, True])
+def test_async_bootout_finishes_before_replacement_or_rollback(managed, monkeypatch, rollback):
+    _, agents, target, loaded, _, _ = managed
+    before = {p: p.read_bytes() for p in agents.glob("*.plist")}
+    run = subprocess.run
+    unloading = {}
+    observations = []
+
+    def launchctl(args, **kwargs):
+        if args[0] != "launchctl":
+            return run(args, **kwargs)
+        verb = args[1]
+        label = Path(args[-1]).stem if verb == "bootstrap" else args[-1].split("/")[-1]
+        if verb == "bootout":
+            payload = loaded[label]
+            result = run(args, **kwargs)
+            loaded[label] = payload
+            unloading[label] = 2
+            return result
+        if verb == "print" and label in unloading:
+            observations.append(label)
+            if unloading[label]:
+                unloading[label] -= 1
+            else:
+                unloading.pop(label)
+                loaded.pop(label)
+        if verb == "bootstrap" and label in loaded:
+            return subprocess.CompletedProcess(args, 5, b"", b"job still unloading")
+        return run(args, **kwargs)
+
+    monkeypatch.setattr(activation.subprocess, "run", launchctl)
+    if rollback:
+        def unavailable(*_):
+            raise RuntimeError("service_runtime_alignment_failed")
+        monkeypatch.setattr(managed_runtime, "_verify_services", unavailable)
+        with pytest.raises(RuntimeError, match="service_runtime_alignment_failed"):
+            align_managed_runtime(str(target))
+        assert {p: p.read_bytes() for p in before} == before
+        assert loaded == {p.stem: plistlib.loads(content) for p, content in before.items()}
+    else:
+        assert align_managed_runtime(str(target))["status"] == "aligned"
+    assert len(observations) == (12 if rollback else 6)
+    assert not unloading
+    assert not (agents / ".loopx-runtime-activation.json").exists()
+
+
+@pytest.mark.parametrize("returncode", [1, 0])
+def test_unload_requires_proved_absence_and_has_a_deadline(monkeypatch, returncode):
+    monkeypatch.setattr(activation.subprocess, "run",
+        lambda *a, **kw: subprocess.CompletedProcess(a, returncode, b"", b"unproved"))
+    clock = iter([0, 10])
+    monkeypatch.setattr(managed_runtime.time, "monotonic", lambda: next(clock))
+    with pytest.raises(RuntimeError, match="service_runtime_alignment_failed"):
+        managed_runtime._wait_unloaded("gui/1/com.loopx.chat")
+
+
 def test_pending_chat_prevents_both_service_changes(managed):
     root, agents, target, loaded, calls, _ = managed
     store = ChatSessionStore(root)