From 4968cf04676ff10d2c9603ffeeb813541200ae22 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 26 Oct 2026 11:40:40 +0800 Subject: [PATCH] fix(native-accounts): discover and restart auth consumers by ownership Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../loopx-codex-provider-routing/NATIVE.md | 36 ++- .../extension.toml | 2 +- .../pyproject.toml | 2 +- .../smoke/native_host_clients_smoke.py | 176 +++++++++++- .../loopx_codex_provider_routing/native.py | 10 +- .../native_host.py | 259 +++++++++++++----- .../native_processes.py | 46 ++++ 7 files changed, 451 insertions(+), 80 deletions(-) create mode 100644 packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native_processes.py diff --git a/packages/loopx-codex-provider-routing/NATIVE.md b/packages/loopx-codex-provider-routing/NATIVE.md index e5ffd5d5a3..956c6d766c 100644 --- a/packages/loopx-codex-provider-routing/NATIVE.md +++ b/packages/loopx-codex-provider-routing/NATIVE.md @@ -59,21 +59,47 @@ must still match. The launch probe must explicitly contain `launch_time`: a positive finite timestamp retains the launch-date fence; explicit `null` from 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 +normal quit never escalates. Unmanaged 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`; +With client restart enabled, version 0.12.6 discovers all loaded current-user +LaunchAgents whose observed Codex process trees consume this home. Service names are +not an allowlist: renamed Chat services, additional providers and plists with +different filenames participate through their existing launchd manager. A +trusted plist, its loaded path, PID, executable and kernel start identity must +match. An unrelated home, ambiguous definition or unverifiable owner cannot +authorize a restart. Discovery combines open home files with the private launch +environment of Codex processes, covering warm app-servers that cached auth and +closed their files. Arguments and environment values are never projected or logged. + +Each service's restoration intent is saved privately before unloading. Observed +descendants are retired only while their kernel identity still matches, even if +they have since closed every auth file. The provider waits for their exit and +rechecks native idle under the existing account transaction locks. Up to three +rescans cover services that appear during shutdown. With force permission, +helpers still alive after normal termination can be killed against those same +frozen identities. No arbitrary CLI is relaunched and no submitted job is replayed. + +Restoration is checked even after a partial failure. A loaded job alone is not +success: a previously running service must have a fresh live process. 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. + +`loopx-codex-accounts clients` performs a read-only live inventory with PIDs, +App/service/unmanaged classification and restartability. This inventory also +leaves outstanding service recovery to the existing status/switch owner. It never reads token +contents or changes selection. Unknown file owners (including filesystem/VM +services without a proven restart owner) remain visible holds; installation +does not turn them into termination targets. Discovery is best effort within +the current user's kernel visibility, not a claim to inspect auth cached inside +another user's process or a virtual machine. 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 3737dac713..b6bcef3676 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.4" +version = "0.12.6" 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 e2972f4bbf..7cd0ba987c 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.4" +version = "0.12.6" 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 index 84656cad41..dd78d05304 100644 --- a/packages/loopx-codex-provider-routing/smoke/native_host_clients_smoke.py +++ b/packages/loopx-codex-provider-routing/smoke/native_host_clients_smoke.py @@ -2,6 +2,7 @@ from __future__ import annotations import importlib +import itertools import json import os from pathlib import Path @@ -15,7 +16,8 @@ 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 +from loopx_codex_provider_routing.native import private_read # noqa: E402 +from loopx_codex_provider_routing.native_processes import _codex_home, codex_home # noqa: E402 BUNDLE = "/Applications/Example.app" EXE = BUNDLE + "/Contents/MacOS/Example" @@ -31,9 +33,10 @@ def __init__(self, home): 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)} + 301: (300, "/test/codex", 901, 5)} self.records[os.getpid()] = (1, "/test/python", 800, 8) self.plist = home / "Library/LaunchAgents" / (LABEL + ".plist") + self.service_pid = 300 self.plist.parent.mkdir(parents=True) self.plist.write_bytes(plistlib.dumps({"Label": LABEL, "ProgramArguments": ["/test/chat"]})) self.plist.chmod(0o600) @@ -57,13 +60,14 @@ def close(self, _expected, **kwargs): assert kwargs["inspected"] == (100, self.records[100]) self.app_pid = None self.records[101] = (1, *self.records[101][1:]) + del self.records[100] return True def open_app(self, *_): self.app_pid = 400 def idle(self): - if self.holders: + if self.host._home_owners(): raise host_module.HostError("waiting_for_idle") def switch(self, _slot): @@ -80,10 +84,13 @@ def run(self, command, **_kwargs): 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 "") + return SimpleNamespace(returncode=0 if self.loaded else 113, + stdout=f"path = {self.plist}\npid = {self.service_pid}\n" if self.loaded else "") if verb == "bootout": self.loaded = False self.holders = [p for p in self.holders if p != 301] + self.records.pop(self.service_pid, None) + self.records.pop(301, None) if self.pause_error: raise self.pause_error if verb == "bootstrap": @@ -92,6 +99,8 @@ def run(self, command, **_kwargs): if self.resume_error == "timeout_no_effect": raise subprocess.TimeoutExpired(command, 20) self.loaded = True + self.service_pid = 600 + self.records[600] = (1, "/test/chat", 1400, 12) if self.resume_error == "timeout_after_effect": raise subprocess.TimeoutExpired(command, 20) return SimpleNamespace(returncode=0, stdout="") @@ -102,13 +111,16 @@ def kill(self, pid, _signal): raise ProcessLookupError() return self.killed.append(pid) - self.holders.remove(pid) + if pid in self.holders: + self.holders.remove(pid) + self.records.pop(pid, None) 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, "codex_home", return_value=None)) 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 @@ -135,6 +147,134 @@ def fixture(): home.mkdir(mode=0o700) return Fixture(home) + # Service labels are installation details. A renamed same-home consumer + # must be stopped and restored through its existing launch agent, including + # when the plist filename differs from its launchd label. + f = fixture() + f.holders = [301] + renamed = "org.example.codex.transport" + f.plist.rename(f.plist.with_name("transport-installation.plist")) + f.plist = f.plist.with_name("transport-installation.plist") + f.plist.write_bytes(plistlib.dumps({"Label": renamed, "ProgramArguments": ["/test/chat"]})) + with f.context(): + assert f.handoff()["active_slot"] == "b" + assert any(c[1:3] == ["bootout", f"gui/{os.getuid()}/{renamed}"] for c in f.calls) + assert f.loaded and f.switches == 1 + + # A second independent service participates in the same transaction. Neither + # label is privileged; both must be restored, without a second account swap. + f = fixture() + f.holders = [101, 301, 701] + second = f.plist.with_name("another-provider.plist") + second.write_bytes(plistlib.dumps({"Label": "org.example.second", "ProgramArguments": ["/test/other"]})) + f.records.update({700: (1, "/test/other", 1500, 14), 701: (700, "/test/codex", 1501, 15)}) + second_state = {"loaded": True, "pid": 700} + primary_run = f.run + def multi_run(command, **kwargs): + if (command[0] == "/bin/launchctl" and + (command[-1].endswith("/org.example.second") or command[-1] == str(second))): + f.calls.append(command) + if command[1] == "print": + return SimpleNamespace(returncode=0 if second_state["loaded"] else 113, + stdout=f"path = {second}\npid = {second_state['pid']}\n") + if command[1] == "bootout": + second_state["loaded"] = False + f.records.pop(700) + f.records.pop(701) + f.holders.remove(701) + elif command[1] == "bootstrap": + second_state.update(loaded=True, pid=800) + f.records[800] = (1, "/test/other", 1600, 16) + return SimpleNamespace(returncode=0, stdout="") + return primary_run(command, **kwargs) + f.run = multi_run + with f.context(): + assert f.handoff()["active_slot"] == "b" + assert f.loaded and second_state["loaded"] and f.switches == 1 + assert len([c for c in f.calls if c[1] == "bootout"]) == 2 + + # The app-server has cached auth and closed every file. Its launch + # environment still proves it consumes this home, and its service restarts. + f = fixture() + f.holders = [] + f.records[301] = (300, "/test/codex", 901, 5) + with f.context(), patch.object(host_module, "codex_home", side_effect=lambda p: str(f.home) if p == 301 else None): + inventory = f.host.clients() + assert inventory["complete"] and len(inventory["clients"]) == 1 + assert inventory["clients"][0]["service"] == LABEL and inventory["clients"][0]["restartable"] + assert f.handoff()["active_slot"] == "b" + assert f.loaded and f.switches == 1 + + # The same warm CLI without a recoverable manager remains an explicit hold; + # a different home remains untouched. Neither case grants broad pkill. + for same_home in (False, True): + f = fixture() + f.holders = [] + f.records[200] = (1, "/test/codex", 999, 3) + clock = iter(range(0, 1000, 20)) + with f.context(), patch.object(host_module, "codex_home", return_value=str(f.home if same_home else f.home / "other")), \ + patch.object(host_module.time, "monotonic", side_effect=lambda: next(clock)): + if same_home: + error("other_clients_busy", f.handoff) + else: + assert f.handoff()["active_slot"] == "b" + assert 200 not in f.killed and f.switches == (0 if same_home else 1) + + # Merely opening a directory is not proof that a service consumes auth. + f = fixture() + f.holders = [301] + f.records[301] = (300, "/test/indexer", 901, 5) + with f.context(): + f.host.pause_chat_agents() + assert not any(c[1] == "bootout" for c in f.calls) + + # Closing FDs does not prove that a captured helper has exited. Normal + # termination is tried first; only separate force permission can escalate. + for force in (False, True): + f = fixture() + f.holders = [101] + signals = [] + original_kill = f.kill + def delayed_kill(pid, sig): + signals.append((pid, sig)) + if pid == 101 and sig == host_module.signal.SIGTERM: + if pid in f.holders: + f.holders.remove(pid) + return # The still-live helper could retain auth in memory. + return original_kill(pid, sig) + f.kill = delayed_kill + clock = itertools.count() + with f.context(), patch.object(host_module.time, "monotonic", side_effect=lambda: next(clock)), \ + patch.object(host_module.time, "sleep"): + if force: + assert f.handoff(force_quit_clients=True)["active_slot"] == "b" + else: + error("other_clients_busy", f.handoff) + assert ((101, host_module.signal.SIGKILL) in signals) is force + assert f.switches == int(force) + + # Changed loaded definitions and reused service PIDs cannot authorize stop. + for mutation in ("path", "pid"): + f = fixture() + f.holders = [301] + original_run, prints = f.run, [0] + def changed_run(command, **kwargs): + result = original_run(command, **kwargs) + if command[0] == "/bin/launchctl" and command[1] == "print": + prints[0] += 1 + if mutation == "path": + result.stdout = result.stdout.replace(str(f.plist), "/different/agent.plist") + elif prints[0] == 2: + f.records[300] = (1, "/test/chat", 2000, 20) + return result + f.run = changed_run + with f.context(): + if mutation == "path": + f.host.pause_chat_agents() + else: + error("client_changed", f.host.pause_chat_agents) + assert not any(c[1] == "bootout" for c in f.calls) and f.switches == 0 + # The same bundled binary does not establish launch ownership. Only the # captured descendant is eligible after it becomes orphaned. f = fixture() @@ -197,7 +337,7 @@ def 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.killed and not any(c[0] == "/bin/launchctl" and c[1] != "print" for c in f.calls) assert not (f.root / "chat-recovery.json").exists() # A product Chat label attached to another home is not paused. @@ -293,5 +433,29 @@ def fixture(): if sys.platform == "darwin": identity = host_module.process_identity(os.getpid()) assert identity and identity[2] > 0 and Path(identity[1]).is_absolute(), identity + # Real kernel environment IO, scoped to our disposable child. Secrets and + # argv remain private even when an argument resembles an environment entry. + child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)", "CODEX_HOME=/decoy"], + env={**os.environ, "CODEX_HOME": "/example/native-home"}, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + try: + assert codex_home(child.pid) == "/example/native-home" + finally: + child.terminate() + child.wait(timeout=5) + +# An argv lookalike cannot supply CODEX_HOME; empty argv entries do not shift +# the environment boundary. Duplicate environment keys are unverifiable. +def process_bytes(arguments, environment): + return (len(arguments).to_bytes(4, sys.byteorder, signed=True) + b"/test/codex\0\0" + + b"\0".join(arguments) + b"\0" + b"\0".join(environment) + b"\0") +assert _codex_home(process_bytes([b"codex", b"", b"CODEX_HOME=/decoy"], [b"CODEX_HOME=/real"])) == "/real" +assert _codex_home(process_bytes([b"codex", b"CODEX_HOME=/decoy"], [b"OTHER=value"])) is None +try: + _codex_home(process_bytes([b"codex"], [b"CODEX_HOME=/one", b"CODEX_HOME=/two"])) +except OSError: + pass +else: + raise AssertionError("ambiguous environment accepted") 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.py b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native.py index e3025423fc..0281efa98f 100644 --- a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native.py +++ b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native.py @@ -162,6 +162,10 @@ def status(self) -> dict: native = self.transport(adapter, "status") return self._projection(state, native) + def clients(self) -> dict: + state = self._load() + return self.transport(state.get("adapter") or discover(), "clients") + def configure(self, *, expected_revision: int, enabled: bool, slots: list[str], reconnect: bool = False, restart_clients: bool | None = None, force_quit_clients: bool | None = None) -> dict: if type(enabled) is not bool or not isinstance(slots, list) or len(slots) > 16 or len(slots) != len(set(slots)): @@ -414,7 +418,7 @@ def close(self): def main() -> int: parser = argparse.ArgumentParser(prog="loopx-codex-accounts") - parser.add_argument("operation", choices=("status", "configure", "refresh", "switch", "recover", "rollback", "reopen", "watch")) + parser.add_argument("operation", choices=("status", "clients", "configure", "refresh", "switch", "recover", "rollback", "reopen", "watch")) parser.add_argument("--runtime-root", type=Path, default=Path.home() / ".loopx") args = parser.parse_args() provider = NativeAccounts(args.runtime_root) @@ -426,9 +430,11 @@ def main() -> int: except KeyboardInterrupt: watcher.close() return 0 - request = {} if args.operation in ("status", "refresh") else json.load(sys.stdin) + request = {} if args.operation in ("status", "clients", "refresh") else json.load(sys.stdin) if args.operation == "status": result = provider.status() + elif args.operation == "clients": + result = provider.clients() elif args.operation == "configure": result = provider.configure(**request) elif args.operation in ("refresh", "switch"): 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 5c0d7f967d..4e327f320f 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 @@ -26,6 +26,8 @@ import urllib.request import urllib.parse +from .native_processes import codex_home + class HostError(Exception): def __init__(self, code: str): @@ -33,10 +35,9 @@ 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",) +# Kept only to read recovery intents written before dynamic service discovery. +# New handoffs discover the actual same-home process tree, not service names. +CHAT_AGENTS = ("com.loopx.chat", "com.loopx.personal-bots.chat") def process_identity(pid: int) -> tuple[int, str, int, int] | None: @@ -114,7 +115,7 @@ def discover() -> dict: class NativeHost: - def __init__(self, adapter: dict, *, user_home: Path | None = None): + def __init__(self, adapter: dict, *, user_home: Path | None = None, reconcile_services: bool = True): source = Path(adapter["source"]) for name in ("personal_apps.py", "personal_accounts.py"): if hashlib.sha256(trusted_file(source / name).read_bytes()).hexdigest() != adapter["pins"][name]: @@ -137,6 +138,9 @@ def idle(): self.apps.account_idle(self.home, self.spec) except (ValueError, OSError, subprocess.SubprocessError): raise HostError("waiting_for_idle") from None + # A warm CLI can cache auth without keeping an auth file open. + if self._cached_codex_owners(): + raise HostError("waiting_for_idle") if self.expected is not None and self.revision() != self.expected: raise HostError("selection_changed") @@ -144,7 +148,7 @@ 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(): + if reconcile_services and sys.platform == "darwin" and (self.manager.root / "chat-recovery.json").exists(): self.resume_chat_agents() def revision(self) -> str: @@ -276,12 +280,55 @@ def reopen(self, expected: str) -> dict: return self.status() def _home_owners(self) -> list[int]: + processes = self._processes() 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()] + owners = {int(line[1:]) for line in result.stdout.splitlines() + if line.startswith("p") and line[1:].isdigit()} + owners |= self._cached_codex_owners(processes) + if any(processes.get(p) is None or process_identity(p) != processes[p] for p in owners): + raise HostError("client_changed") + self._owner_identities = {p: processes[p] for p in owners} + return sorted(owners) + + def _processes(self) -> dict: + result = subprocess.run(["/bin/ps", "-axo", "pid="], + capture_output=True, text=True, check=True, timeout=5) + return {int(p): process_identity(int(p)) for p in result.stdout.split() if p.isdigit()} + + def _cached_codex_owners(self, processes: dict | None = None) -> set[int]: + owners = set() + for pid, identity in (self._processes() if processes is None else processes).items(): + # Names select which private environment to inspect. They never + # establish restart/signal ownership; that needs the process tree. + if identity is None or Path(identity[1]).name != "codex": + continue + try: + home = codex_home(pid) + except OSError: + if process_identity(pid) is not None: + raise HostError("ownership_unverified") from None + continue + if process_identity(pid) != identity: + raise HostError("client_changed") + if home: + if not Path(home).is_absolute(): + raise HostError("ownership_unverified") + if Path(home).resolve() == Path(self.spec["home"]).resolve(): + owners.add(pid) + return owners + + @staticmethod + def _descendants(root: int, processes: dict) -> set[int]: + descendants = {root} + 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: + return descendants + descendants.update(added) def inspect_client_helpers(self) -> tuple[tuple, dict]: """Freeze descendants of the inspected launch BEFORE requesting quit.""" @@ -293,16 +340,8 @@ def inspect_client_helpers(self) -> tuple[tuple, dict]: 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) + processes, helpers = self._processes(), {} + descendants = self._descendants(pid, processes) for child in descendants - {pid}: identity = processes[child] if identity[1].startswith(bundle + "/"): @@ -311,14 +350,12 @@ def inspect_client_helpers(self) -> tuple[tuple, dict]: raise HostError("client_changed") return (pid, root), helpers - def retire_app_helpers(self, helpers: dict) -> None: + def retire_app_helpers(self, helpers: dict, *, force: bool = False) -> 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 + for pid in helpers: identity = process_identity(pid) if identity is None or identity[1:] != helpers[pid]: continue @@ -327,18 +364,98 @@ def retire_app_helpers(self, helpers: dict) -> None: if self.not_after is not None and time.time() >= self.not_after: raise HostError("quota_stale") try: - os.kill(pid, signal.SIGTERM) + os.kill(pid, signal.SIGKILL if force else signal.SIGTERM) except OSError: pass + def _helpers_stopped(self, helpers: dict) -> bool: + for pid, frozen in helpers.items(): + current = process_identity(pid) + if current is not None: + if current[1:] == frozen: + return False + else: + try: + os.kill(pid, 0) + except ProcessLookupError: + continue + except OSError: + pass + return False # Unknown identity does not prove exit. + return True + + def _consumer_agents(self, owners: list[int]) -> list[tuple]: + """Resolve real holders to trusted, currently loaded user LaunchAgents.""" + definitions = {} + for plist in sorted((self.home / "Library/LaunchAgents").glob("*.plist")): + try: + value = plistlib.loads(trusted_file(plist).read_bytes()) + label = value.get("Label") if isinstance(value, dict) else None + if not isinstance(label, str) or re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,254}", label) is None: + continue + definitions.setdefault(label, []).append(plist) + except (HostError, OSError, ValueError, plistlib.InvalidFileException): + continue # An unrelated invalid definition grants no effect. + agents, processes = [], self._processes() + if any(processes.get(p) != self._owner_identities.get(p) for p in owners): + raise HostError("client_changed") + auth_owners = {p for p in owners if Path(processes[p][1]).name == "codex"} + worker_tree = self._descendants(os.getpid(), processes) + for label, plists in definitions.items(): + if len(plists) != 1: + continue # Ambiguous definitions cannot be restored safely. + plist = plists[0] + target = f"gui/{os.getuid()}/{label}" + loaded = subprocess.run(["/bin/launchctl", "print", target], + capture_output=True, text=True, timeout=10) + match = re.search(r"^\s*pid = (\d+)\s*$", loaded.stdout, re.MULTILINE) + path = re.search(r"^\s*path = (.+)\s*$", loaded.stdout, re.MULTILINE) + if loaded.returncode or match is None or path is None or Path(path[1].strip()).resolve() != plist.resolve(): + continue + service_pid = int(match[1]) + service = processes.get(service_pid) + if service is None or service_pid in worker_tree: + continue + descendants = self._descendants(service_pid, processes) + if not (descendants & auth_owners - worker_tree): + continue + helpers = {p: processes[p][1:] for p in descendants - worker_tree if processes.get(p) is not None} + if process_identity(service_pid) != service: + raise HostError("client_changed") + agents.append((label, plist, service_pid, service, helpers)) + return agents + + def clients(self) -> dict: + if sys.platform != "darwin": + raise HostError("platform_unavailable") + owners = self._home_owners() + inspected, helpers = self.inspect_client_helpers() + app = set(helpers) | ({inspected[0]} if inspected[0] else set()) + services = {p: label for label, _, _, _, processes in self._consumer_agents(owners) for p in processes} + return {"complete": True, "clients": [ + {"pid": p, "kind": "app" if p in app else "service" if p in services else "unmanaged", + "process": Path(self._owner_identities[p][1]).name, + "restartable": p in app or p in services, + **({"service": services[p]} if p in services else {})} for p in owners]} + 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())): + if not isinstance(value.get("agents"), dict): raise HostError("state_untrusted") + for label, entry in value["agents"].items(): + if isinstance(entry, str) and label in CHAT_AGENTS: + continue # Legacy recovery intent; keep its original path. + if (not isinstance(label, str) or re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,254}", label) is None + or not isinstance(entry, dict) or not isinstance(entry.get("digest"), str) + or re.fullmatch(r"[0-9a-f]{64}", entry["digest"]) is None + or not isinstance(entry.get("plist"), str) or Path(entry["plist"]).name != entry["plist"] + or not entry["plist"].endswith(".plist") or type(entry.get("pid")) is not int or entry["pid"] <= 1 + or not isinstance(entry.get("identity"), list) or len(entry["identity"]) != 3 + or not isinstance(entry["identity"][0], str) + or any(type(v) is not int or v < 0 for v in entry["identity"][1:])): + raise HostError("state_untrusted") if value["agents"]: owner = value.get("owner", {}) identity = owner.get("identity") if isinstance(owner, dict) else None @@ -359,46 +476,23 @@ def _chat_definition(self, plist: Path, label: str) -> str: 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.""" + def pause_chat_agents(self) -> dict: + """Pause every provably related user service through its real manager.""" from .native import atomic self.accounts.private_dir(self.manager.root) path, recovery = self._chat_recovery() - if recovery["agents"]: + if recovery["agents"] and recovery["owner"]["pid"] != os.getpid(): raise HostError("chat_recovery_pending") - for label in CHAT_AGENTS: - plist = self.home / "Library/LaunchAgents" / f"{label}.plist" - if not plist.is_file(): + helpers = {} + for label, plist, service_pid, service, descendants in self._consumer_agents(self._home_owners()): + if label in recovery["agents"]: 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) + fresh = set(self._home_owners()) + if not any(p in fresh and self._owner_identities[p][1:] == identity + for p, identity in descendants.items()): + raise HostError("client_changed") current = subprocess.run(["/bin/launchctl", "print", target], capture_output=True, text=True, timeout=10) if (current.returncode or process_identity(service_pid) != service or @@ -411,7 +505,8 @@ def pause_chat_agents(self) -> None: # 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 + recovery["agents"][label] = {"digest": digest, "plist": plist.name, + "pid": service_pid, "identity": list(service[1:])} owner = process_identity(os.getpid()) if owner is None: raise HostError("ownership_unverified") @@ -424,6 +519,8 @@ def pause_chat_agents(self) -> None: raise HostError("chat_pause_failed") except (OSError, subprocess.SubprocessError): raise HostError("chat_pause_failed") from None + helpers.update(descendants) + return helpers def _chat_loaded(self, target: str) -> bool: return subprocess.run(["/bin/launchctl", "print", target], @@ -448,8 +545,9 @@ def resume_chat_agents(self) -> None: 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" + for label, entry in list(recovery["agents"].items()): + digest = entry if isinstance(entry, str) else entry["digest"] + plist = self.home / "Library/LaunchAgents" / (f"{label}.plist" if isinstance(entry, str) else entry["plist"]) target = f"gui/{os.getuid()}/{label}" try: if self._chat_definition(plist, label) != digest: @@ -465,6 +563,26 @@ def resume_chat_agents(self) -> None: # service state before accepting or retrying restoration. if not self._chat_loaded(target): raise HostError("chat_resume_failed") + if isinstance(entry, dict): + # A registered job is insufficient: the old auth consumer + # must exit, and restoration must produce a fresh process. + old = process_identity(entry["pid"]) + if old is not None and list(old[1:]) == entry["identity"]: + raise HostError("chat_resume_failed") + deadline, started = time.monotonic() + 5, False + while True: + loaded = subprocess.run(["/bin/launchctl", "print", target], + capture_output=True, text=True, timeout=10) + match = re.search(r"^\s*pid = (\d+)\s*$", loaded.stdout, re.MULTILINE) + if loaded.returncode == 0 and match is not None and process_identity(int(match[1])) is not None: + break + if time.monotonic() >= deadline: + raise HostError("chat_resume_failed") + if not started: + subprocess.run(["/bin/launchctl", "kickstart", target], + capture_output=True, text=True, timeout=10) + started = True # Restore a previously running, non-RunAtLoad job too. + time.sleep(0.1) except (HostError, OSError, subprocess.SubprocessError): failed = True continue # Attempt every outstanding restoration. @@ -502,10 +620,14 @@ def switch(self, slot: str, expected: str, *, reopen: bool, restart_clients: boo # an unrelated CLI/file owner remains a hold, never a kill # target. self.retire_app_helpers(helpers) - self.pause_chat_agents() + helpers.update(self.pause_chat_agents()) + self.retire_app_helpers(helpers) deadline = time.monotonic() + 15 + force_after, rescans = deadline - 5, 0 while True: try: + if not self._helpers_stopped(helpers): + raise HostError("waiting_for_idle") self.manager.idle() break except HostError as error: @@ -513,7 +635,12 @@ def switch(self, slot: str, expected: str, *, reopen: bool, restart_clients: boo raise if time.monotonic() >= deadline: raise HostError("other_clients_busy") from None - self.retire_app_helpers(helpers) + # Catch services started during initial shutdown, + # without an unbounded scan/termination loop. + if rescans < 3: + helpers.update(self.pause_chat_agents()) + rescans += 1 + self.retire_app_helpers(helpers, force=force_quit_clients and time.monotonic() >= force_after) time.sleep(0.2) result = self.manager.switch(slot) except Exception: @@ -784,10 +911,12 @@ def main() -> int: {"schema_version": SCHEMA, "pid": os.getpid()}) try: request = json.load(sys.stdin) - host = NativeHost(request["adapter"]) operation = request["operation"] + host = NativeHost(request["adapter"], reconcile_services=operation != "clients") if operation == "status": value = host.status() + elif operation == "clients": + value = host.clients() elif operation == "observe": value = host.observe(request["slots"], request["expected_revision"], request.get("auth_failures")) elif operation == "switch": diff --git a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native_processes.py b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native_processes.py new file mode 100644 index 0000000000..e1260fc086 --- /dev/null +++ b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native_processes.py @@ -0,0 +1,46 @@ +"""Darwin process IO for native account consumer discovery. + +Process arguments/environment never leave this adapter. Only CODEX_HOME is +returned; the caller fences the read with kernel executable/start identity. +This is discovery evidence, not authorization to signal an arbitrary process. +""" +from __future__ import annotations + +import ctypes +import os +import sys + + +def codex_home(pid: int) -> str | None: + """Read the launch environment without putting arguments/secrets in ps output.""" + if sys.platform != "darwin": + raise OSError("process environment unavailable") + libc = ctypes.CDLL(None, use_errno=True) + mib = (ctypes.c_int * 3)(1, 49, pid) # CTL_KERN, KERN_PROCARGS2 + size = ctypes.c_size_t() + if libc.sysctl(mib, 3, None, ctypes.byref(size), None, 0) != 0 or not 4 < size.value <= 1048576: + raise OSError("process environment unavailable") + buffer = ctypes.create_string_buffer(size.value) + if libc.sysctl(mib, 3, buffer, ctypes.byref(size), None, 0) != 0: + raise OSError("process environment unavailable") + return _codex_home(buffer.raw[:size.value]) + + +def _codex_home(raw: bytes) -> str | None: + argc = int.from_bytes(raw[:4], sys.byteorder, signed=True) + if not 0 < argc <= 65536: + raise OSError("invalid process environment") + position = raw.find(b"\0", 4) + if position < 0: + raise OSError("invalid process environment") + while position < len(raw) and raw[position] == 0: + position += 1 + for _ in range(argc): + end = raw.find(b"\0", position) + if end < 0: + raise OSError("invalid process environment") + position = end + 1 + values = [value[11:] for value in raw[position:].split(b"\0") if value.startswith(b"CODEX_HOME=")] + if len(values) > 1: + raise OSError("ambiguous process environment") + return os.fsdecode(values[0]) if values else None