diff --git a/.github/workflows/package-smoke.yml b/.github/workflows/package-smoke.yml index 042e596550..7e2d9399ef 100644 --- a/.github/workflows/package-smoke.yml +++ b/.github/workflows/package-smoke.yml @@ -43,6 +43,7 @@ jobs: run: | python -m compileall -q packages/loopx-codex-provider-routing/src python packages/loopx-codex-provider-routing/smoke/native_host_smoke.py + python packages/loopx-codex-provider-routing/smoke/native_host_clients_smoke.py python packages/loopx-codex-provider-routing/smoke/codex_provider_routing_smoke.py python packages/loopx-codex-provider-routing/smoke/recovery_contracts_smoke.py python packages/loopx-codex-provider-routing/smoke/operator_smoke.py diff --git a/packages/loopx-codex-provider-routing/NATIVE.md b/packages/loopx-codex-provider-routing/NATIVE.md index 0719b64927..e5ffd5d5a3 100644 --- a/packages/loopx-codex-provider-routing/NATIVE.md +++ b/packages/loopx-codex-provider-routing/NATIVE.md @@ -61,6 +61,19 @@ Launch Services uses the inspected PID and executable instead. Missing fields, invalid types and non-finite timestamps cannot authorize force quit. A successful normal quit never escalates. Other CLI/file owners remain holds. Force quit can interrupt work in that instance; it does not authorize replay of submitted jobs. +Version 0.12.4 also records the inspected App's descendants before quitting. +Only those helpers, with unchanged kernel executable and microsecond start +identity, can be retired after the App exits. An older orphan with no observed +parentage, a standalone CLI using the same bundled binary, or a later App launch +remains a hold. Bundle paths alone never authorize termination. +With client restart enabled, the known LoopX Chat launch agent can also be +paused when its observed process tree actually holds this home. Its restoration +intent is saved privately before unloading, and restoration is checked even +after a partial failure. A failed restore is reported as `chat_resume_failed`; +the next native request reconciles that intent without repeating the account +transaction. A changed service definition is preserved and requires local +repair rather than bootstrapping unchecked bytes. Installation does not enable +client restart or start a new Chat service. An operator-owned service independent of the App can run the same observer: ```sh diff --git a/packages/loopx-codex-provider-routing/extension.toml b/packages/loopx-codex-provider-routing/extension.toml index 929efcb1a8..3737dac713 100644 --- a/packages/loopx-codex-provider-routing/extension.toml +++ b/packages/loopx-codex-provider-routing/extension.toml @@ -1,6 +1,6 @@ schema_version = "loopx_extension_manifest_v0" id = "loopx-codex-provider-routing" -version = "0.12.3" +version = "0.12.4" requires_loopx_api = ">=1,<2" permissions = [] diff --git a/packages/loopx-codex-provider-routing/pyproject.toml b/packages/loopx-codex-provider-routing/pyproject.toml index 94cf4c67a2..e2972f4bbf 100644 --- a/packages/loopx-codex-provider-routing/pyproject.toml +++ b/packages/loopx-codex-provider-routing/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "loopx-codex-provider-routing" -version = "0.12.3" +version = "0.12.4" description = "Native Codex account slots and exhausted-account handoff for LoopX" readme = "README.md" requires-python = ">=3.11" diff --git a/packages/loopx-codex-provider-routing/smoke/native_host_clients_smoke.py b/packages/loopx-codex-provider-routing/smoke/native_host_clients_smoke.py new file mode 100644 index 0000000000..84656cad41 --- /dev/null +++ b/packages/loopx-codex-provider-routing/smoke/native_host_clients_smoke.py @@ -0,0 +1,297 @@ +"""Offline production-switch counterexamples for scoped client coordination.""" +from __future__ import annotations + +import importlib +import json +import os +from pathlib import Path +import plistlib +from types import SimpleNamespace +import subprocess +import sys +import tempfile +from unittest.mock import patch + +PACKAGE_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(PACKAGE_ROOT / "src")) +host_module = importlib.import_module("loopx_codex_provider_routing.native_host") +from loopx_codex_provider_routing.native import private_read + +BUNDLE = "/Applications/Example.app" +EXE = BUNDLE + "/Contents/MacOS/Example" +HELPER = BUNDLE + "/Contents/Resources/codex" +LABEL = "com.loopx.personal-bots.chat" + + +class Fixture: + def __init__(self, home): + self.home, self.app_pid, self.active = home, 100, "a" + self.revision, self.switches = "before", 0 + self.holders, self.killed, self.calls = [101, 200], [], [] + self.loaded, self.pause_error, self.resume_error = True, None, None + self.records = {100: (1, EXE, 1000, 1), 101: (100, HELPER, 1001, 2), + 200: (1, HELPER, 999, 3), 300: (1, "/test/chat", 900, 4), + 301: (300, "/test/app-server", 901, 5)} + self.records[os.getpid()] = (1, "/test/python", 800, 8) + self.plist = home / "Library/LaunchAgents" / (LABEL + ".plist") + self.plist.parent.mkdir(parents=True) + self.plist.write_bytes(plistlib.dumps({"Label": LABEL, "ProgramArguments": ["/test/chat"]})) + self.plist.chmod(0o600) + self.root = home / "native-accounts" + self.root.mkdir(mode=0o700) + self.host = host_module.NativeHost.__new__(host_module.NativeHost) + self.host.home = home + self.host.spec = {"home": home} + self.host.not_after, self.host.expected = None, "before" + self.host.accounts = SimpleNamespace(private_dir=lambda p: p.mkdir(mode=0o700, exist_ok=True)) + self.host.apps = SimpleNamespace(app_bundle=lambda *_: Path(BUNDLE), executable=lambda *_: EXE, + running=lambda *_: self.app_pid, open_app=self.open_app) + self.host.revision = lambda: self.revision + self.host.status = lambda: {"selection_revision": self.revision, "recovery_required": False, + "current_login_matches": True, "active_slot": self.active, + "client_running": self.app_pid is not None} + self.host.manager = SimpleNamespace(root=self.root, idle=self.idle, switch=self.switch) + self.host.close_client = self.close + + def close(self, _expected, **kwargs): + assert kwargs["inspected"] == (100, self.records[100]) + self.app_pid = None + self.records[101] = (1, *self.records[101][1:]) + return True + + def open_app(self, *_): + self.app_pid = 400 + + def idle(self): + if self.holders: + raise host_module.HostError("waiting_for_idle") + + def switch(self, _slot): + self.idle() + self.switches += 1 + self.active, self.revision = "b", "after" + return {"history_untouched": True} + + def run(self, command, **_kwargs): + self.calls.append(command) + if command[0] == "/bin/ps": + return SimpleNamespace(returncode=0, stdout="\n".join(map(str, self.records))) + if command[0] == "/usr/sbin/lsof": + return SimpleNamespace(returncode=0, stdout="".join(f"p{p}\n" for p in self.holders)) + verb = command[1] + if verb == "print": + return SimpleNamespace(returncode=0 if self.loaded else 113, stdout="pid = 300\n" if self.loaded else "") + if verb == "bootout": + self.loaded = False + self.holders = [p for p in self.holders if p != 301] + if self.pause_error: + raise self.pause_error + if verb == "bootstrap": + if self.resume_error == "nonzero": + return SimpleNamespace(returncode=5, stdout="") + if self.resume_error == "timeout_no_effect": + raise subprocess.TimeoutExpired(command, 20) + self.loaded = True + if self.resume_error == "timeout_after_effect": + raise subprocess.TimeoutExpired(command, 20) + return SimpleNamespace(returncode=0, stdout="") + + def kill(self, pid, _signal): + if _signal == 0: + if pid not in self.records: + raise ProcessLookupError() + return + self.killed.append(pid) + self.holders.remove(pid) + + def context(self): + from contextlib import ExitStack + stack = ExitStack() + stack.enter_context(patch.object(host_module.sys, "platform", "darwin")) + stack.enter_context(patch.object(host_module, "process_identity", side_effect=lambda p: self.records.get(p))) + stack.enter_context(patch.object(host_module.subprocess, "run", side_effect=self.run)) + stack.enter_context(patch.object(host_module.os, "kill", side_effect=self.kill)) + return stack + + def handoff(self, **kwargs): + return self.host.switch("b", "before", reopen=True, restart_clients=True, **kwargs) + + +def error(code, operation): + try: + operation() + except host_module.HostError as failure: + assert failure.code == code, failure.code + else: + raise AssertionError("expected " + code) + + +with tempfile.TemporaryDirectory() as temporary: + counter = 0 + def fixture(): + global counter + counter += 1 + home = Path(temporary).resolve() / str(counter) + home.mkdir(mode=0o700) + return Fixture(home) + + # The same bundled binary does not establish launch ownership. Only the + # captured descendant is eligible after it becomes orphaned. + f = fixture() + with f.context(): + inspected, helpers = f.host.inspect_client_helpers() + assert set(helpers) == {101} + f.close("before", inspected=inspected) + f.host.retire_app_helpers(helpers) + assert f.killed == [101] and f.holders == [200] + + f = fixture() + with f.context(): + error("client_changed", lambda: host_module.NativeHost.close_client( + f.host, "before", inspected=(100, (1, EXE, 99, 0)))) + + # A remaining independent CLI blocks the full switch and leaves its work + # untouched; the source App is reopened without changing the account. + f = fixture() + clock = iter(range(0, 1000, 20)) + with f.context(), patch.object(host_module.time, "monotonic", side_effect=lambda: next(clock)): + error("other_clients_busy", f.handoff) + assert f.killed == [101] and f.holders == [200] and f.switches == 0 + assert f.active == "a" and f.app_pid is not None + + # PID reuse, a prior orphan, a new launch and unprovable identity stay holds. + for mutation in ("pid_reused", "unreadable", "new_launch", "selection_changed", "quota_stale"): + f = fixture() + with f.context(): + inspected, helpers = f.host.inspect_client_helpers() + f.close("before", inspected=inspected) + if mutation == "pid_reused": + f.records[101] = (1, HELPER, 1002, 7) + elif mutation == "unreadable": + del f.records[101] + elif mutation == "new_launch": + f.app_pid = 400 + elif mutation == "selection_changed": + f.revision = "manual" + else: + f.host.not_after = 0 + if mutation in ("new_launch", "selection_changed", "quota_stale"): + error({"new_launch": "client_changed", "selection_changed": "selection_changed", + "quota_stale": "quota_stale"}[mutation], lambda: f.host.retire_app_helpers(helpers)) + else: + f.host.retire_app_helpers(helpers) + assert not f.killed + + # Exercise the production switch: a valid descendant and same-home Chat + # recover, with one credential transaction and the original App reopened. + f = fixture() + f.holders = [101, 301] + with f.context(): + result = f.handoff() + assert f.killed == [101] and f.loaded and f.switches == 1 + assert result["active_slot"] == "b" and not result["launch_failed"] + assert private_read(f.root / "chat-recovery.json")["agents"] == {} + + # Disabled restart retains the baseline: no signals or service effects. + f = fixture() + f.holders = [] + with f.context(): + f.host.switch("b", "before", reopen=False, restart_clients=False) + assert not f.killed and not f.calls + assert not (f.root / "chat-recovery.json").exists() + + # A product Chat label attached to another home is not paused. + f = fixture() + with f.context(): + f.host.pause_chat_agents() + assert not any(c[1] == "bootout" for c in f.calls) + assert not (f.root / "chat-recovery.json").exists() + + # bootout has already taken effect when its reply times out. finally must + # still restore Chat and must not attempt the credential transaction. + f = fixture() + f.holders = [101, 301] + f.pause_error = subprocess.TimeoutExpired("bootout", 20) + with f.context(): + error("chat_pause_failed", f.handoff) + assert f.loaded and f.switches == 0 and f.app_pid is not None + assert private_read(f.root / "chat-recovery.json")["agents"] == {} + + for failure in ("nonzero", "timeout_no_effect"): + f = fixture() + f.holders = [101, 301] + f.resume_error = failure + with f.context(): + error("chat_resume_failed", f.handoff) + assert not f.loaded and f.switches == 1 + assert f.app_pid is not None + assert private_read(f.root / "chat-recovery.json")["agents"] + # The next native request reconciles the existing intent. It does not + # switch credentials again, and handles an already-restored service. + f.resume_error = None + with f.context(): + f.host.resume_chat_agents() + f.host.resume_chat_agents() + assert f.loaded and f.switches == 1 + assert private_read(f.root / "chat-recovery.json")["agents"] == {} + + f = fixture() + f.holders = [101, 301] + f.resume_error = "timeout_after_effect" + with f.context(): + assert f.handoff()["active_slot"] == "b" + assert f.loaded and f.switches == 1 + + # Recovery cannot bootstrap a user-edited plist or follow an injected label. + f = fixture() + f.holders = [301] + with f.context(): + f.host.pause_chat_agents() + f.plist.write_text("changed by user") + error("chat_resume_failed", f.host.resume_chat_agents) + assert not f.loaded + assert private_read(f.root / "chat-recovery.json")["agents"] + + # A separate status host cannot restore a live worker's paused services. + # A dead/reused worker PID does permit bounded restoration of that intent. + f = fixture() + f.holders = [301] + with f.context(): + f.host.pause_chat_agents() + journal = f.root / "chat-recovery.json" + recovery = private_read(journal) + recovery["owner"]["pid"] = 500 + f.records[500] = (1, *recovery["owner"]["identity"]) + journal.write_text(json.dumps(recovery)) + count = len(f.calls) + f.host.resume_chat_agents() + assert not f.loaded and len(f.calls) == count + error("chat_recovery_pending", f.handoff) + assert f.switches == 0 + f.records[500] = None + f.host.resume_chat_agents() + assert not f.loaded # Unknown identity of a live PID is not exit. + f.records[500] = (1, "/test/python", 1200, 10) + f.host.resume_chat_agents() + assert f.loaded and private_read(journal)["agents"] == {} + + f = fixture() + (f.root / "chat-recovery.json").symlink_to(f.plist) + with f.context(): + error("state_untrusted", f.host.resume_chat_agents) + assert not f.calls + + f = fixture() + (f.root / "chat-recovery.json").write_text(json.dumps({ + "schema_version": "codex_native_accounts_v1", "agents": {"unrelated.service": "digest"}})) + (f.root / "chat-recovery.json").chmod(0o600) + with f.context(): + error("state_untrusted", f.host.resume_chat_agents) + assert not f.calls + +# Exercise the real Darwin kernel probe without inspecting another user's args. +if sys.platform == "darwin": + identity = host_module.process_identity(os.getpid()) + assert identity and identity[2] > 0 and Path(identity[1]).is_absolute(), identity + +print("native host client handoff smoke passed (identity, scope, switch and recovery)") diff --git a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native_host.py b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native_host.py index 68b12dd6f5..5c0d7f967d 100644 --- a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native_host.py +++ b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native_host.py @@ -7,13 +7,17 @@ from __future__ import annotations import hashlib +import ctypes import importlib import json import math import os from pathlib import Path +import plistlib +import re import selectors import shutil +import signal import subprocess import sys import tempfile @@ -29,6 +33,33 @@ def __init__(self, code: str): super().__init__(code) +# LoopX chat surfaces keep a Codex app-server attached to the GPT home while +# idle. They are product-owned clients of that home, so an unattended handoff +# pauses them instead of treating them as an unrelated CLI hold. +CHAT_AGENTS = ("com.loopx.personal-bots.chat",) + + +def process_identity(pid: int) -> tuple[int, str, int, int] | None: + """Kernel executable/start identity; command text is not process ownership.""" + class BSDInfo(ctypes.Structure): + _fields_ = ([(name, ctypes.c_uint32) for name in ( + "flags", "status", "xstatus", "pid", "ppid", "uid", "gid", "ruid", + "rgid", "svuid", "svgid", "reserved")] + + [("comm", ctypes.c_char * 16), ("name", ctypes.c_char * 32)] + + [("tail", ctypes.c_uint32 * 6), ("seconds", ctypes.c_uint64), + ("microseconds", ctypes.c_uint64)]) + try: + lib = ctypes.CDLL("/usr/lib/libproc.dylib") + info, path = BSDInfo(), ctypes.create_string_buffer(4096) + if (lib.proc_pidinfo(pid, 3, 0, ctypes.byref(info), ctypes.sizeof(info)) != ctypes.sizeof(info) + or info.pid != pid or info.uid != os.getuid() + or lib.proc_pidpath(pid, path, len(path)) <= 0): + return None + return info.ppid, os.fsdecode(path.value), info.seconds, info.microseconds + except (OSError, ValueError): + return None + + # Address one inspected process, not an App bundle id shared by other routes. # terminate() requests the normal AppKit quit lifecycle and can be refused. QUIT = '''ObjC.import("AppKit"); @@ -113,6 +144,8 @@ def idle(): self.home, self.spec, idle, lambda: str(self.codex), lambda selected: self.apps.clean_env(selected, self.home), ) + if sys.platform == "darwin" and (self.manager.root / "chat-recovery.json").exists(): + self.resume_chat_agents() def revision(self) -> str: # last.json also changes for a manual re-selection of the same slot. @@ -131,9 +164,13 @@ def status(self) -> dict: result["client_running"] = self.apps.running(exe, self.spec) is not None return result - def close_client(self, expected: str, *, timeout: float = 25, force: bool = False) -> bool: + def close_client(self, expected: str, *, timeout: float = 25, force: bool = False, + inspected: tuple | None = None) -> bool: exe = self.apps.executable(self.apps.app_bundle(self.spec, "gpt", self.home)) pid = self.apps.running(exe, self.spec) + if inspected is not None and (pid != inspected[0] or + (pid is not None and process_identity(pid) != inspected[1])): + raise HostError("client_changed") if self.revision() != expected: raise HostError("selection_changed") if pid is None: @@ -238,10 +275,211 @@ def reopen(self, expected: str) -> dict: self.apps.open_app(self.home, "gpt", []) return self.status() + def _home_owners(self) -> list[int]: + result = subprocess.run(["/usr/sbin/lsof", "-Fp", "+d", str(self.spec["home"])], + capture_output=True, text=True, timeout=10) + if result.returncode not in (0, 1): + raise HostError("ownership_unverified") + return [int(line[1:]) for line in result.stdout.splitlines() + if line.startswith("p") and line[1:].isdigit()] + + def inspect_client_helpers(self) -> tuple[tuple, dict]: + """Freeze descendants of the inspected launch BEFORE requesting quit.""" + bundle = str(self.apps.app_bundle(self.spec, "gpt", self.home)) + exe = self.apps.executable(Path(bundle)) + pid = self.apps.running(exe, self.spec) + if pid is None: + return (None, None), {} + root = process_identity(pid) + if root is None or root[1] != str(exe): + raise HostError("ownership_unverified") + result = subprocess.run(["/bin/ps", "-axo", "pid="], + capture_output=True, text=True, check=True, timeout=5) + processes = {int(p): process_identity(int(p)) for p in result.stdout.split() if p.isdigit()} + descendants, helpers = {pid}, {} + while True: + added = {p for p, identity in processes.items() + if identity is not None and identity[0] in descendants and p not in descendants} + if not added: + break + descendants.update(added) + for child in descendants - {pid}: + identity = processes[child] + if identity[1].startswith(bundle + "/"): + helpers[child] = identity[1:] # Reparenting after quit is expected. + if process_identity(pid) != root or self.apps.running(exe, self.spec) != pid: + raise HostError("client_changed") + return (pid, root), helpers + + def retire_app_helpers(self, helpers: dict) -> None: + """Retire only still-identical, previously observed launch descendants.""" + exe = self.apps.executable(self.apps.app_bundle(self.spec, "gpt", self.home)) + if self.apps.running(exe, self.spec) is not None: + raise HostError("client_changed") + for pid in self._home_owners(): + if pid not in helpers: + continue + identity = process_identity(pid) + if identity is None or identity[1:] != helpers[pid]: + continue + if self.revision() != self.expected: + raise HostError("selection_changed") + if self.not_after is not None and time.time() >= self.not_after: + raise HostError("quota_stale") + try: + os.kill(pid, signal.SIGTERM) + except OSError: + pass + + def _chat_recovery(self) -> tuple[Path, dict]: + from .native import SCHEMA, private_read + path = self.manager.root / "chat-recovery.json" + value = private_read(path) if path.exists() else {"schema_version": SCHEMA, "agents": {}} + if (not isinstance(value.get("agents"), dict) or + any(label not in CHAT_AGENTS or not isinstance(digest, str) + for label, digest in value["agents"].items())): + raise HostError("state_untrusted") + if value["agents"]: + owner = value.get("owner", {}) + identity = owner.get("identity") if isinstance(owner, dict) else None + if (not isinstance(owner, dict) or type(owner.get("pid")) is not int or owner["pid"] <= 1 or + not isinstance(identity, list) or len(identity) != 3 or + not isinstance(identity[0], str) or + any(type(v) is not int or v < 0 for v in identity[1:])): + raise HostError("state_untrusted") + return path, value + + def _chat_definition(self, plist: Path, label: str) -> str: + raw = trusted_file(plist).read_bytes() + try: + value = plistlib.loads(raw) + if not isinstance(value, dict) or value.get("Label") != label: + raise ValueError() + except (ValueError, plistlib.InvalidFileException): + raise HostError("state_untrusted") from None + return hashlib.sha256(raw).hexdigest() + + def pause_chat_agents(self) -> None: + """Unload known chat agents that keep this home attached while idle.""" + from .native import atomic + self.accounts.private_dir(self.manager.root) + path, recovery = self._chat_recovery() + if recovery["agents"]: + raise HostError("chat_recovery_pending") + for label in CHAT_AGENTS: + plist = self.home / "Library/LaunchAgents" / f"{label}.plist" + if not plist.is_file(): + continue + target = f"gui/{os.getuid()}/{label}" + loaded = subprocess.run(["/bin/launchctl", "print", target], + capture_output=True, text=True, timeout=10) + if loaded.returncode != 0: + continue + match = re.search(r"^\s*pid = (\d+)\s*$", loaded.stdout, re.MULTILINE) + if match is None: + continue + service_pid = int(match[1]) + service = process_identity(service_pid) + if service is None: + continue + # A product label alone is not enough: this loaded instance must + # actually own the inspected home, possibly through app-server. + owns_home = False + for owner in self._home_owners(): + seen = set() + while owner > 1 and owner not in seen: + seen.add(owner) + identity = process_identity(owner) + if identity is None: + break + if owner == service_pid: + owns_home |= identity == service + break + owner = identity[0] + if not owns_home: + continue + digest = self._chat_definition(plist, label) + current = subprocess.run(["/bin/launchctl", "print", target], + capture_output=True, text=True, timeout=10) + if (current.returncode or process_identity(service_pid) != service or + re.search(r"^\s*pid = " + str(service_pid) + r"\s*$", current.stdout, re.MULTILINE) is None): + raise HostError("client_changed") + if self.revision() != self.expected: + raise HostError("selection_changed") + if self.not_after is not None and time.time() >= self.not_after: + raise HostError("quota_stale") + # Retain recovery BEFORE an operation which can take effect and + # then time out. The detached worker's finally or the next native + # request reconciles this same intent, never another account swap. + recovery["agents"][label] = digest + owner = process_identity(os.getpid()) + if owner is None: + raise HostError("ownership_unverified") + recovery["owner"] = {"pid": os.getpid(), "identity": list(owner[1:])} + atomic(path, recovery) + try: + result = subprocess.run(["/bin/launchctl", "bootout", target], + capture_output=True, text=True, timeout=20) + if result.returncode or self._chat_loaded(target): + raise HostError("chat_pause_failed") + except (OSError, subprocess.SubprocessError): + raise HostError("chat_pause_failed") from None + + def _chat_loaded(self, target: str) -> bool: + return subprocess.run(["/bin/launchctl", "print", target], + capture_output=True, text=True, timeout=10).returncode == 0 + + def resume_chat_agents(self) -> None: + from .native import atomic + path, recovery = self._chat_recovery() + if recovery["agents"]: + owner = recovery["owner"] + current = process_identity(owner["pid"]) + if owner["pid"] != os.getpid(): + if current is not None and list(current[1:]) == owner["identity"]: + return # A concurrent status read must not undo a live handoff. + if current is None: + try: + os.kill(owner["pid"], 0) + except ProcessLookupError: + pass + except OSError: + return + else: + return # Unreadable identity does not prove worker exit. + failed = False + for label, digest in list(recovery["agents"].items()): + plist = self.home / "Library/LaunchAgents" / f"{label}.plist" + target = f"gui/{os.getuid()}/{label}" + try: + if self._chat_definition(plist, label) != digest: + raise HostError("chat_resume_failed") + if not self._chat_loaded(target): + try: + subprocess.run(["/bin/launchctl", "bootstrap", f"gui/{os.getuid()}", str(plist)], + capture_output=True, text=True, timeout=20) + except (OSError, subprocess.SubprocessError): + if not self._chat_loaded(target): + raise HostError("chat_resume_failed") from None + # A timeout/nonzero reply is ambiguous. Read the actual + # service state before accepting or retrying restoration. + if not self._chat_loaded(target): + raise HostError("chat_resume_failed") + except (HostError, OSError, subprocess.SubprocessError): + failed = True + continue # Attempt every outstanding restoration. + del recovery["agents"][label] + atomic(path, recovery) + if failed: + raise HostError("chat_resume_failed") + def switch(self, slot: str, expected: str, *, reopen: bool, restart_clients: bool = False, force_quit_clients: bool = False, not_after: float | None = None) -> dict: if sys.platform != "darwin": raise HostError("platform_unavailable") + self.resume_chat_agents() + if self._chat_recovery()[1]["agents"]: + raise HostError("chat_recovery_pending") self.expected = expected self.not_after = not_after before = self.status() @@ -250,36 +488,56 @@ def switch(self, slot: str, expected: str, *, reopen: bool, restart_clients: boo # idle() is rechecked by the native owner under launch/account locks, # and immediately before its first mutation. This is the CAS fence, # including against manual selections outside LoopX. + result, restore_error = None, None try: - if restart_clients: - if not_after is not None and time.time() >= not_after: - raise HostError("quota_stale") - self.close_client(expected, **({"force": True} if force_quit_clients else {})) - # Children may flush and exit just after their App root. An - # unrelated CLI/file owner remains a hold, never a kill target. - deadline = time.monotonic() + 10 - while True: + try: + if restart_clients: + if not_after is not None and time.time() >= not_after: + raise HostError("quota_stale") + inspected, helpers = self.inspect_client_helpers() + self.close_client(expected, inspected=inspected, + **({"force": True} if force_quit_clients else {})) + # Orphaned App helpers and idle chat transports are + # product-owned clients of this home. Retire or pause them; + # an unrelated CLI/file owner remains a hold, never a kill + # target. + self.retire_app_helpers(helpers) + self.pause_chat_agents() + deadline = time.monotonic() + 15 + while True: + try: + self.manager.idle() + break + except HostError as error: + if error.code != "waiting_for_idle": + raise + if time.monotonic() >= deadline: + raise HostError("other_clients_busy") from None + self.retire_app_helpers(helpers) + time.sleep(0.2) + result = self.manager.switch(slot) + except Exception: + # A refused/failed transaction must not unnecessarily leave a + # normally closed source App down. Native preflight blocks + # opening an incomplete credential journal; a later manual + # choice wins. + if restart_clients and before.get("client_running"): try: - self.manager.idle() - break - except HostError as error: - if error.code != "waiting_for_idle": - raise - if time.monotonic() >= deadline: - raise HostError("other_clients_busy") from None - time.sleep(0.2) - result = self.manager.switch(slot) - except Exception: - # A refused/failed transaction must not unnecessarily leave a - # normally closed source App down. Native preflight blocks opening - # an incomplete credential journal; a later manual choice wins. - if restart_clients and before.get("client_running"): - try: - if self.status().get("client_running") is False: - self.reopen(expected) - except (HostError, OSError, ValueError, subprocess.SubprocessError): - pass - raise + if self.status().get("client_running") is False: + self.reopen(expected) + except (HostError, OSError, ValueError, subprocess.SubprocessError): + pass + raise + finally: + # Restore product-owned transports even when the handoff fails. + try: + self.resume_chat_agents() + except HostError as error: + if result is None: + raise + # A committed account still needs its App reopened. Report the + # retained service recovery after that original continuation. + restore_error = error value = self.status() if value["active_slot"] != slot or value["current_login_matches"] is not True: raise HostError("readback_failed") @@ -293,6 +551,8 @@ def switch(self, slot: str, expected: str, *, reopen: bool, restart_clients: boo if after["selection_revision"] != value["selection_revision"]: raise HostError("selection_changed") value = after + if restore_error is not None: + raise restore_error return value | {"history_untouched": result["history_untouched"], "launch_failed": launch_failed} def recover(self, rollback: bool, expected: str) -> dict: diff --git a/tests/presentation/test_codex_accounts.py b/tests/presentation/test_codex_accounts.py index d1c5182d28..8080ff62ed 100644 --- a/tests/presentation/test_codex_accounts.py +++ b/tests/presentation/test_codex_accounts.py @@ -648,13 +648,20 @@ def disable(): assert provider.tick()["action"] == "disabled" and len(transport.calls) == before -def client_host(native, monkeypatch): +def client_host(native, monkeypatch, tmp_path): from loopx_codex_provider_routing import native_host as module monkeypatch.setattr(module.sys, "platform", "darwin") adapter = module.NativeHost.__new__(module.NativeHost) state = {"pid": 11, "active": "a", "revision": "initial", "quit": [], "switch": 0, "open": 0} - adapter.home, adapter.spec = Path("/synthetic"), {} + adapter.home, adapter.spec = tmp_path, {"home": tmp_path} + executable = Path("/synthetic/GPT.app/Contents/MacOS/GPT") + identity = (1, str(executable), 1000, 1) + # This fixture owns one synthetic App launch, with no helpers or installed + # Chat agents. The package client smoke covers discovery and service effects. + monkeypatch.setattr(module, "process_identity", lambda pid: identity if pid == 11 else None) + adapter.inspect_client_helpers = lambda: ((state["pid"], module.process_identity(state["pid"])), {}) + adapter._home_owners = lambda: [] adapter.revision = lambda: state["revision"] adapter.status = lambda: {"active_slot": state["active"], "selection_revision": state["revision"], "current_login_matches": True, "recovery_required": False, @@ -680,9 +687,11 @@ def quit(command, **parameters): assert command[-2:] == ["11", "/synthetic/GPT.app/Contents/MacOS/GPT"] state["pid"] = None - adapter.manager = SimpleNamespace(switch=switch, idle=idle) - adapter.apps = SimpleNamespace(executable=lambda _: Path("/synthetic/GPT.app/Contents/MacOS/GPT"), - app_bundle=lambda *_: None, running=lambda *_: state["pid"], open_app=reopen) + adapter.manager = SimpleNamespace(root=tmp_path / "native-accounts", switch=switch, idle=idle) + adapter.accounts = SimpleNamespace(private_dir=lambda path: path.mkdir(mode=0o700, exist_ok=True)) + adapter.apps = SimpleNamespace(executable=lambda _: executable, + app_bundle=lambda *_: executable.parents[2], + running=lambda *_: state["pid"], open_app=reopen) monkeypatch.setattr(module.subprocess, "run", quit) return adapter, state, module @@ -690,8 +699,8 @@ def quit(command, **parameters): @pytest.mark.parametrize("fault,phase", [(None, "quit"), ("unreadable-launch", "probe"), *[(fault, phase) for fault in ("selection", "new-process", "stale-quota") for phase in ("probe", "quit")]]) -def test_force_quit_requires_the_same_observed_launch_and_selection(native, monkeypatch, fault, phase): - adapter, state, module = client_host(native, monkeypatch) +def test_force_quit_requires_the_same_observed_launch_and_selection(native, monkeypatch, tmp_path, fault, phase): + adapter, state, module = client_host(native, monkeypatch, tmp_path) adapter.not_after = None forced = [] @@ -729,8 +738,8 @@ def stalled(command, **parameters): '{"launch_time":"1000"}', '{"launch_time":0}', '{"launch_time":-1}', '{"launch_time":NaN}', '{"launch_time":Infinity}', '{"launch_time":-Infinity}']) -def test_incomplete_or_invalid_probe_cannot_authorize_any_quit(native, monkeypatch, receipt): - adapter, state, module = client_host(native, monkeypatch) +def test_incomplete_or_invalid_probe_cannot_authorize_any_quit(native, monkeypatch, tmp_path, receipt): + adapter, state, module = client_host(native, monkeypatch, tmp_path) calls = [] def probe(command, **parameters): @@ -747,8 +756,8 @@ def probe(command, **parameters): @pytest.mark.parametrize("launch_time", [None, 1000.0]) @pytest.mark.parametrize("outcome", ["graceful", "refused", "timeout"]) -def test_force_permission_only_escalates_a_failed_normal_quit(native, monkeypatch, launch_time, outcome): - adapter, state, module = client_host(native, monkeypatch) +def test_force_permission_only_escalates_a_failed_normal_quit(native, monkeypatch, tmp_path, launch_time, outcome): + adapter, state, module = client_host(native, monkeypatch, tmp_path) adapter.not_after = None calls = [] @@ -785,22 +794,22 @@ def test_force_permission_is_independent_and_survives_reconfiguration(native, tm provider.configure(expected_revision=3, enabled=False, slots=["a", "b"], force_quit_clients="yes") -def test_normal_quit_precedes_native_idle_transaction_and_same_route_reopen(native, monkeypatch): - adapter, state, _ = client_host(native, monkeypatch) +def test_normal_quit_precedes_native_idle_transaction_and_same_route_reopen(native, monkeypatch, tmp_path): + adapter, state, _ = client_host(native, monkeypatch, tmp_path) result = adapter.switch("b", "initial", reopen=True, restart_clients=True) assert state["quit"] and state["switch"] == 1 and state["open"] == 1 assert result["active_slot"] == "b" and result["client_running"] is True -def test_wait_only_mode_does_not_quit_a_client(native, monkeypatch): - adapter, state, _ = client_host(native, monkeypatch) +def test_wait_only_mode_does_not_quit_a_client(native, monkeypatch, tmp_path): + adapter, state, _ = client_host(native, monkeypatch, tmp_path) with pytest.raises(Exception, match="waiting_for_idle"): adapter.switch("b", "initial", reopen=True) assert not state["quit"] and state["active"] == "a" -def test_quit_refusal_and_timeout_do_not_mutate_credentials(native, monkeypatch): - adapter, state, module = client_host(native, monkeypatch) +def test_quit_refusal_and_timeout_do_not_mutate_credentials(native, monkeypatch, tmp_path): + adapter, state, module = client_host(native, monkeypatch, tmp_path) def refuse(*_, **__): raise module.subprocess.CalledProcessError(1, "synthetic") @@ -815,8 +824,8 @@ def refuse(*_, **__): assert state["active"] == "a" -def test_manual_selection_and_new_client_prevent_closing_the_wrong_process(native, monkeypatch): - adapter, state, _ = client_host(native, monkeypatch) +def test_manual_selection_and_new_client_prevent_closing_the_wrong_process(native, monkeypatch, tmp_path): + adapter, state, _ = client_host(native, monkeypatch, tmp_path) with pytest.raises(Exception, match="selection_changed"): adapter.switch("b", "stale", reopen=True, restart_clients=True) assert not state["quit"] @@ -827,16 +836,16 @@ def test_manual_selection_and_new_client_prevent_closing_the_wrong_process(nativ assert not state["quit"] and state["switch"] == 0 and state["open"] == 0 -def test_native_switch_failure_reopens_source_without_replaying_account_selection(native, monkeypatch): - adapter, state, module = client_host(native, monkeypatch) +def test_native_switch_failure_reopens_source_without_replaying_account_selection(native, monkeypatch, tmp_path): + adapter, state, module = client_host(native, monkeypatch, tmp_path) adapter.manager.switch = lambda _: (_ for _ in ()).throw(module.HostError("readback_failed")) with pytest.raises(Exception, match="readback_failed"): adapter.switch("b", "initial", reopen=True, restart_clients=True) assert state["active"] == "a" and state["open"] == 1 -def test_manual_selection_during_quit_wins_before_any_credential_write(native, monkeypatch): - adapter, state, module = client_host(native, monkeypatch) +def test_manual_selection_during_quit_wins_before_any_credential_write(native, monkeypatch, tmp_path): + adapter, state, module = client_host(native, monkeypatch, tmp_path) quit = module.subprocess.run def manual_choice(*args, **kwargs): @@ -849,8 +858,8 @@ def manual_choice(*args, **kwargs): assert state["active"] == "c" and state["switch"] == 0 and state["open"] == 0 -def test_expired_automatic_evidence_does_not_even_quit_the_client(native, monkeypatch): - adapter, state, _ = client_host(native, monkeypatch) +def test_expired_automatic_evidence_does_not_even_quit_the_client(native, monkeypatch, tmp_path): + adapter, state, _ = client_host(native, monkeypatch, tmp_path) with pytest.raises(Exception, match="quota_stale"): adapter.switch("b", "initial", reopen=True, restart_clients=True, not_after=time.time() - 1) assert not state["quit"] and not state["switch"]