diff --git a/loopx/hosts/managed_service_runtime.py b/loopx/hosts/managed_service_runtime.py index 9515c973a5..a44db55ef4 100644 --- a/loopx/hosts/managed_service_runtime.py +++ b/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 diff --git a/tests/test_managed_service_runtime.py b/tests/test_managed_service_runtime.py index cac39a2c04..b88b39d87a 100644 --- a/tests/test_managed_service_runtime.py +++ b/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)