diff --git a/loopx/hosts/managed_service_runtime.py b/loopx/hosts/managed_service_runtime.py index e54a473d3f..4b279ea885 100644 --- a/loopx/hosts/managed_service_runtime.py +++ b/loopx/hosts/managed_service_runtime.py @@ -9,12 +9,17 @@ import copy import base64 +from contextlib import ExitStack, contextmanager +import importlib import json +import math import os from pathlib import Path import plistlib import shlex +import stat import subprocess +import sys import time import urllib.error import urllib.request @@ -23,6 +28,11 @@ from .. import runtime_activation as activation +_OBSERVER_LABEL = "com.loopx.codex-native-accounts" +_OBSERVER_MODULE = "loopx_codex_provider_routing.native" +_RUNTIME_BINDING = {"schema_version": "loopx_managed_runtime_binding_v1"} + + def _literal_arguments(command: str) -> list[str]: # Only the installer's final literal exec is editable. Never evaluate a # custom shell expansion, pipeline, redirection or command substitution. @@ -86,7 +96,7 @@ def _retarget(payload: dict, executable: str, python: str) -> dict: result.setdefault("EnvironmentVariables", {})["LOOPX_PYTHON"] = python result["EnvironmentVariables"].pop("PYTHONPATH", None) result["EnvironmentVariables"].pop("LOOPX_RELEASE_ROOT", None) - result["LoopXRuntimeBinding"] = {"schema_version": "loopx_managed_runtime_binding_v1"} + result["LoopXRuntimeBinding"] = _RUNTIME_BINDING.copy() return result @@ -119,12 +129,12 @@ def has_runtime_binding() -> bool: payload = plistlib.loads(path.read_bytes()) if not isinstance(payload, dict): raise ValueError("service_runtime_route_unverified") - if payload.get("LoopXRuntimeBinding") == {"schema_version": "loopx_managed_runtime_binding_v1"}: + if payload.get("LoopXRuntimeBinding") == _RUNTIME_BINDING: return True return False -def _endpoint(payload: dict) -> int: +def _service_command(payload: dict) -> tuple[dict[str, str], list[str]]: arguments, _ = _launcher(payload) # A standard label is not evidence of the service's role. Parse the command # position, rather than finding a word that could be an option's value. @@ -134,14 +144,21 @@ def _endpoint(payload: dict) -> int: if expected is None or Path(arguments[0]).name != "loopx": raise ValueError("service_runtime_route_unverified") index = 1 + options = {} while index < len(arguments) and arguments[index].startswith("--"): - flag, separator, _ = arguments[index].partition("=") - if flag not in {"--registry", "--runtime-root", "--format"}: + flag, separator, value = arguments[index].partition("=") + if (flag not in {"--registry", "--runtime-root", "--format"} or flag in options + or not separator and index + 1 >= len(arguments)): raise ValueError("service_runtime_route_unverified") + options[flag] = value if separator else arguments[index + 1] index += 1 if separator else 2 if index >= len(arguments) or arguments[index] != expected: raise ValueError("service_runtime_route_unverified") - arguments = arguments[index + 1:] + return options, arguments[index + 1:] + + +def _endpoint(payload: dict) -> int: + _, arguments = _service_command(payload) def option(flag): values = [word.split("=", 1)[1] if word.startswith(flag + "=") else arguments[i + 1] if i + 1 < len(arguments) else "" @@ -160,6 +177,168 @@ def option(flag): return port +def _observer_route(payload: dict) -> Path: + if not isinstance(payload, dict): + raise ValueError("native_observer_route_unverified") + arguments, shell = _launcher(payload) + rest = arguments[1:] + if (payload.get("Label") != _OBSERVER_LABEL or shell is not None or len(rest) != 5 + or rest[:4] != ["-m", _OBSERVER_MODULE, "watch", "--runtime-root"] + or not Path(rest[4]).is_absolute()): + raise ValueError("native_observer_route_unverified") + return Path(rest[4]) + + +def _retarget_observer(payload: dict, python: str, core: Path) -> dict: + _observer_route(payload) + # The standalone watcher needs the *same* Core's supported Chat fence + # during a native account handoff, including in its detached worker. + root = core.parent.parent + if core.parent.name != "scripts" or not (root / "loopx/extensions/chat_service_guard.py").is_file(): + raise ValueError("native_observer_core_unverified") + env = payload.get("EnvironmentVariables", {}) + if not isinstance(env, dict): + raise ValueError("native_observer_route_unverified") + env = {key: value for key, value in env.items() if key != "PYTHONHOME"} + env.update(PYTHONPATH=str(root), PYTHONSAFEPATH="1", PYTHONNOUSERSITE="1", + LOOPX_PYTHON=python, LOOPX_RELEASE_ROOT=str(root)) + return {**payload, "EnvironmentVariables": env, "LoopXRuntimeBinding": _RUNTIME_BINDING.copy(), + "ProgramArguments": [python, *payload["ProgramArguments"][1:]]} + + +def _provider_pinned(python: str) -> bool: + # A generic Core update must not take over an independently installed + # observer. Only a Core explicitly pinned to a venv's own provider can + # adopt that observer. A venv's inherited system package does not qualify. + bin_dir = Path(python).parent + return (bin_dir.parent / "pyvenv.cfg").is_file() and (bin_dir / "loopx-native-account-provider").is_file() + + +def _observer_source(path: Path, pinned: bool) -> tuple[bytes, dict] | None: + try: + if path.is_symlink(): + raise ValueError("native_observer_route_unverified") + original = path.read_bytes() + payload = plistlib.loads(original) + if not isinstance(payload, dict): + raise ValueError("native_observer_route_unverified") + except (OSError, ValueError, plistlib.InvalidFileException) as exc: + if pinned: + raise ValueError("native_observer_route_unverified") from exc + return None + if pinned or payload.get("LoopXRuntimeBinding") == _RUNTIME_BINDING: + return original, payload + return None + + +def _native_package(python: str): + if Path(python) != Path(sys.executable): + raise activation.RuntimeActivationDeferred("native_observer_requires_target_runtime", [_OBSERVER_LABEL]) + try: + module = importlib.import_module(_OBSERVER_MODULE) + except ImportError as exc: + raise ValueError("native_observer_route_unverified") from exc + prefix = Path(python).parent.parent.resolve() + if not module.__file__ or not Path(module.__file__).resolve().is_relative_to(prefix): + raise ValueError("native_observer_route_unverified") + if not callable(getattr(module, "observer_build_identity", None)): + raise activation.RuntimeActivationDeferred("native_observer_provider_upgrade_required", [_OBSERVER_LABEL]) + return module + + +def _chat_account_root(payload: dict) -> Path: + from ..control_plane.projects.registry_codec import load_project_registry + from ..paths import resolve_runtime_root + + options, _ = _service_command(payload) + registry = Path(options.get("--registry", "")) + if not registry.is_absolute() or not registry.is_file(): + raise ValueError("native_observer_route_unverified") + config = load_project_registry(registry) + override = None if config.get("common_runtime_root") else options.get("--runtime-root") + return resolve_runtime_root(config, override, registry_path=registry).resolve() + + +@contextmanager +def _alignment_guard(chat: list[Path], observer_root: Path | None): + with activation.managed_chat_activation_guard(chat): + if observer_root is None: + yield None + return + try: + module = importlib.import_module(_OBSERVER_MODULE) + except ImportError as exc: + raise ValueError("native_observer_requires_target_runtime") from exc + owner = module.NativeAccounts(observer_root) + with ExitStack() as locks: + try: + locks.enter_context(owner._lock()) + locks.enter_context(owner._lock(name="api-relay-state.lock")) + state = owner._load() + except BlockingIOError as exc: + raise activation.RuntimeActivationDeferred("native_account_work_pending", [_OBSERVER_LABEL]) from exc + except (module.HostError, AttributeError, TypeError) as exc: + raise ValueError("native_observer_owner_unverified") from exc + if state.get("intent") is not None or state.get("action") in {"restarting", "handoff_interrupted"}: + raise activation.RuntimeActivationDeferred("native_account_work_pending", [_OBSERVER_LABEL]) + yield owner + + +def _observer_lock_pid(owner) -> int | None: + lock = owner.root / "observer.lock" + try: + info = lock.lstat() + except FileNotFoundError: + return None + if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or info.st_mode & 0o077: + raise ValueError("native_observer_owner_unverified") + result = subprocess.run(["/usr/sbin/lsof", "-t", str(lock)], + capture_output=True, text=True, encoding="utf-8", timeout=5) + if result.returncode == 1 and not result.stdout and not result.stderr: + return None + pids = result.stdout.splitlines() + if result.returncode or len(pids) != 1 or not pids[0].isdecimal(): + raise ValueError("native_observer_owner_unverified") + return int(pids[0]) + + +def _wait_observer_released(owner) -> None: + deadline = time.monotonic() + 5 + while _observer_lock_pid(owner) is not None: + if time.monotonic() >= deadline: + raise RuntimeError("native_observer_owner_unverified") + time.sleep(0.05) + + +def _verify_observer(path: Path, owner, after: float, expected: dict[str, str] | None) -> None: + module = importlib.import_module(_OBSERVER_MODULE) + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + loaded, pid = activation._managed_service_state(path) + if loaded and pid and _observer_lock_pid(owner) == pid: + try: + value = module.private_read(owner.root / "observer.json", schema=module.OBSERVER_SCHEMA) + checked = value.get("checked_at") + expires = value.get("deadline_at") + now = time.time() + if (type(checked) in (int, float) and after <= checked <= now + and type(expires) in (int, float) and math.isfinite(expires) and expires > now + and value.get("phase") in {"waiting", "retrying"} + and all(value.get(key) == build for key, build in (expected or {}).items())): + return + except FileNotFoundError: + pass + except (module.HostError, AttributeError, TypeError) as exc: + raise ValueError("native_observer_readiness_unverified") from exc + time.sleep(0.1) + raise RuntimeError("native_observer_readiness_unverified") + + +def _start_order(paths): + # Start the registered lock owner before the hosted Chat contender. + return sorted(paths, key=lambda path: path.stem != _OBSERVER_LABEL) + + def _verify_services(services: dict[Path, int], identity: dict) -> None: # A working CLI does not prove its Chat assets can serve. Read back the # exact HTTP runtime before committing the handoff; a failed start restores @@ -222,9 +401,10 @@ def _recover_alignment(journal: Path) -> None: if not isinstance(saved, dict) or saved.get("schema_version") != "loopx_managed_runtime_handoff_v1": raise ValueError("service_runtime_route_unverified") entries = saved.get("services") - if not isinstance(entries, dict) or not entries or not set(entries) <= {"com.loopx.status", "com.loopx.chat"}: + if not isinstance(entries, dict) or not entries or not set(entries) <= {"com.loopx.status", "com.loopx.chat", _OBSERVER_LABEL}: raise ValueError("service_runtime_route_unverified") originals, candidates = {}, {} + observer_root = None for label, item in entries.items(): if (not isinstance(item, dict) or set(item) != {"original", "candidate"} or not all(isinstance(value, str) for value in item.values())): @@ -235,21 +415,39 @@ def _recover_alignment(journal: Path) -> None: original = plistlib.loads(originals[path]) candidate = plistlib.loads(candidates[path]) if (not isinstance(original, dict) or not isinstance(candidate, dict) - or original.get("Label") != label or candidate.get("Label") != label - or _endpoint(original) != _endpoint(candidate) - or _launcher(original)[0][1:] != _launcher(candidate)[0][1:]): + or original.get("Label") != label or candidate.get("Label") != label): + raise ValueError("service_runtime_route_unverified") + if label == _OBSERVER_LABEL: + observer_root = _observer_route(original) + if _observer_route(candidate) != observer_root: + raise ValueError("native_observer_route_unverified") + elif _endpoint(original) != _endpoint(candidate) or _launcher(original)[0][1:] != _launcher(candidate)[0][1:]: raise ValueError("service_runtime_route_unverified") if path.is_symlink() or path.read_bytes() not in {originals[path], candidates[path]}: raise ValueError("service_runtime_route_unverified") chat = journal.parent / "com.loopx.chat.plist" - with activation.managed_chat_activation_guard([chat] if chat.exists() else []): + if observer_root is not None: + if chat not in originals or _chat_account_root(plistlib.loads(originals[chat])) != observer_root.resolve(): + raise ValueError("native_observer_route_unverified") + observer = journal.parent / f"{_OBSERVER_LABEL}.plist" + candidate = plistlib.loads(candidates[observer]) + core = Path(_launcher(plistlib.loads(candidates[chat]))[0][0]) + if _retarget_observer(plistlib.loads(originals[observer]), candidate["ProgramArguments"][0], core) != candidate: + raise ValueError("native_observer_route_unverified") + with _alignment_guard([chat] if chat.exists() else [], observer_root) as owner: domain = f"gui/{os.getuid()}" loaded = [] for path in originals: - old, _ = activation._managed_service_state(path, payload=plistlib.loads(originals[path])) - new, _ = activation._managed_service_state(path, payload=plistlib.loads(candidates[path])) + old, old_pid = activation._managed_service_state(path, payload=plistlib.loads(originals[path])) + new, new_pid = activation._managed_service_state(path, payload=plistlib.loads(candidates[path])) if old is False and new is False: raise ValueError("service_runtime_route_unverified") + if path.stem == _OBSERVER_LABEL: + pid = old_pid or new_pid + if ((old or new) and not pid) or _observer_lock_pid(owner) != pid: + raise ValueError("native_observer_owner_unverified") + if not pid: + _wait_unloaded(f"{domain}/{path.stem}") if old or new: loaded.append(path) try: @@ -258,10 +456,15 @@ def _recover_alignment(journal: Path) -> None: _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: + for path in _start_order(originals): _wait_unloaded(f"{domain}/{path.stem}") + if path.stem == _OBSERVER_LABEL: + _wait_observer_released(owner) + started = time.time() _launchctl("bootstrap", domain, str(path)) - except (OSError, RuntimeError, subprocess.SubprocessError) as exc: + if path.stem == _OBSERVER_LABEL: + _verify_observer(path, owner, started, None) + except (OSError, RuntimeError, ValueError, subprocess.SubprocessError) as exc: raise RuntimeError("service_runtime_rollback_failed") from exc journal.unlink() @@ -311,20 +514,53 @@ def _align_managed_runtime(executable: str, journal: Path) -> dict: originals[plist] = original updates[plist] = plistlib.dumps(updated) pids[plist] = pid + chat = [p for p in plists if p.exists() and p.stem.endswith(".chat")] + observer = journal.parent / f"{_OBSERVER_LABEL}.plist" + observer_root = None + native = expected_observer = None + pinned = _provider_pinned(python) + source = _observer_source(observer, pinned) if chat and observer.exists() else None + if source: + original, payload = source + root = _observer_route(payload) + if root.resolve() == _chat_account_root(plistlib.loads(chat[0].read_bytes())): + if not pinned: + raise activation.RuntimeActivationDeferred("native_observer_provider_binding_required", [_OBSERVER_LABEL]) + native = _native_package(python) + expected_observer = native.observer_build_identity() + loaded, pid = activation._managed_service_state(observer) + if not loaded or not pid: + raise ValueError("native_observer_owner_unverified") + observer_root = root + candidate = _retarget_observer(payload, python, target) + if candidate != payload: + originals[observer] = original + updates[observer] = plistlib.dumps(candidate) + pids[observer] = pid + # The hosted contender must stop before the standalone owner. + if chat[0] not in updates: + originals[chat[0]] = updates[chat[0]] = chat[0].read_bytes() + pids[chat[0]] = activation._managed_service_pid(chat[0]) if not updates: _verify_services(endpoints, identity) + if observer_root is not None: + _verify_observer(observer, native.NativeAccounts(observer_root), time.time() - 480, expected_observer) return {"status": "aligned", "runtime_identity": identity, "restarted_services": [], "readiness_verified": bool(endpoints)} - chat = [p for p in plists if p.exists() and p.stem.endswith(".chat")] stopped, written, restarted = [], [], [] domain = f"gui/{os.getuid()}" - with activation.managed_chat_activation_guard(chat): - # Fence the complete pair before the first signal or persistent edit. + with _alignment_guard(chat, observer_root) as owner: + # Fence all participating services before the first signal or write. for plist in updates: if (plist.read_bytes() != originals[plist] or activation._managed_service_state(plist) != (True, pids[plist])): raise ValueError("service_runtime_route_unverified") + if observer_root is not None: + loaded, pid = activation._managed_service_state(observer) + if (not loaded or not pid or _observer_lock_pid(owner) != pid + or _chat_account_root(plistlib.loads(chat[0].read_bytes())) != observer_root.resolve()): + raise ValueError("native_observer_owner_unverified") write_json_atomic(journal, {"schema_version": "loopx_managed_runtime_handoff_v1", "services": {p.stem: {"original": base64.b64encode(originals[p]).decode(), "candidate": base64.b64encode(updates[p]).decode()} for p in updates}}, @@ -339,25 +575,38 @@ def _align_managed_runtime(executable: str, journal: Path) -> dict: raise RuntimeError("service_runtime_route_unverified") write_bytes_atomic(plist, content, preserve_mode=True, fsync_directory=True) written.append(plist) - for plist in updates: + observer_after = time.time() - 480 + for plist in _start_order(updates): + if plist == observer: + _wait_observer_released(owner) + observer_after = time.time() _launchctl("bootstrap", domain, str(plist)) restarted.append(plist) + if plist == observer: + _verify_observer(plist, owner, observer_after, expected_observer) _verify_services(endpoints, identity) - except (OSError, RuntimeError, subprocess.SubprocessError): + if observer_root is not None: + _verify_observer(observer, owner, observer_after, expected_observer) + except (OSError, RuntimeError, ValueError, subprocess.SubprocessError): # Restore the manager's original source as well as its file. Keep # rollback failure explicit; never declare HTTP readiness here. try: - for plist in restarted: + for plist in sorted(restarted, key=lambda path: path == observer): _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: + for plist in _start_order(stopped): _wait_unloaded(f"{domain}/{plist.stem}") + if plist == observer: + _wait_observer_released(owner) + started = time.time() _launchctl("bootstrap", domain, str(plist)) - except (OSError, RuntimeError, subprocess.SubprocessError) as exc: + if plist == observer: + _verify_observer(plist, owner, started, None) + except (OSError, RuntimeError, ValueError, subprocess.SubprocessError) as exc: raise RuntimeError("service_runtime_rollback_failed") from exc journal.unlink() raise RuntimeError("service_runtime_alignment_failed") diff --git a/packages/loopx-codex-provider-routing/NATIVE.md b/packages/loopx-codex-provider-routing/NATIVE.md index e3af60c5ff..8b84af8a90 100644 --- a/packages/loopx-codex-provider-routing/NATIVE.md +++ b/packages/loopx-codex-provider-routing/NATIVE.md @@ -42,6 +42,20 @@ boundary to adopt it. Do not retain custom launch scripts or editable imports in temporary worktree. A missing optional channel or its credentials remains an explicit unavailable channel; never reconstruct credentials from chat history. +When an existing macOS `com.loopx.codex-native-accounts` LaunchAgent watches +the same account root as managed Chat, adopt the qualified Core and its pinned +provider environment together. Run the installed target's +`update activate --service all --runtime-executable `. +The supported direct `watch` job participates in the same handoff as Chat and +Status. The handoff waits for accepted Chat and account work, starts the +observer first, and verifies its lock and both loaded builds from the 0.15.5 +observer heartbeat before committing. +It restores the previous job routes on failure. Once the watcher has been +adopted, a Core update without a pinned provider defers before replacing any +service. Core-only updates still leave unbound standalone watchers in place. +Run the provider's separate relay reconciliation after the managed processes +have adopted the same build. + `loopx-native-account-provider` reads this JSON protocol on stdin: ```json diff --git a/packages/loopx-codex-provider-routing/pyproject.toml b/packages/loopx-codex-provider-routing/pyproject.toml index 945ac14367..afbe14138f 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.15.4" +version = "0.15.5" 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/src/loopx_codex_provider_routing/native-extension.toml b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native-extension.toml index 67d6d014c9..6d60618467 100644 --- a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native-extension.toml +++ b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native-extension.toml @@ -1,6 +1,6 @@ schema_version = "loopx_extension_manifest_v0" id = "loopx-native-account-provider" -version = "0.15.4" +version = "0.15.5" requires_loopx_api = ">=1,<2" permissions = ["model-provider.local-read", "model-provider.transport-manage"] 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 d7750a2343..abfcc3718b 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 @@ -30,6 +30,11 @@ for name in ("native.py", "native_policy.ts", "native_policy_bridge.ts"))).hexdigest() +def observer_build_identity() -> dict[str, str]: + from .openai_relay import BUILD + return {"build": OBSERVER_BUILD, "relay_build": BUILD} + + def private_read(path: Path, *, schema: str = SCHEMA) -> dict: if any(p.is_symlink() for p in (path, *path.parents)): raise HostError("state_untrusted") @@ -644,7 +649,7 @@ def _observe_failed(self): def _heartbeat(self, phase: str, *, deadline: float, next_check: float | None = None): now = time.time() atomic(self.provider.root / "observer.json", {"schema_version": OBSERVER_SCHEMA, - "build": OBSERVER_BUILD, "checked_at": now, "deadline_at": now + deadline, + **observer_build_identity(), "checked_at": now, "deadline_at": now + deadline, "next_check_at": next_check, "phase": phase}) def close(self): diff --git a/tests/test_managed_service_runtime.py b/tests/test_managed_service_runtime.py index 6b41c530ed..1241314dd8 100644 --- a/tests/test_managed_service_runtime.py +++ b/tests/test_managed_service_runtime.py @@ -1,6 +1,7 @@ """Service alignment preserves the real Chat admission boundary and manager route.""" from __future__ import annotations +from contextlib import contextmanager import json import os from pathlib import Path @@ -10,6 +11,7 @@ import sys import socket import time +from types import SimpleNamespace import urllib.request import uuid @@ -95,6 +97,191 @@ def test_pair_adopts_target_without_changing_routes_or_cli_and_is_idempotent(man assert calls == [] +@pytest.fixture +def managed_with_observer(managed, monkeypatch): + root, agents, target, loaded, calls, fail = managed + # Chat's message store and the shared native account owner are different. + account = agents.parent.parent / "shared native accounts" + (account / "extensions/codex-native-accounts").mkdir(parents=True) + (agents.parent.parent / "registry").write_text(json.dumps({ + "goals": [], "common_runtime_root": str(account), + })) + core = target.parent + (core / "scripts").mkdir() + target = target.rename(core / "scripts/loopx") + (core / "loopx/extensions").mkdir(parents=True) + (core / "loopx/extensions/chat_service_guard.py").touch() + label = "com.loopx.codex-native-accounts" + path = agents / f"{label}.plist" + payload = {"Label": label, "ProgramArguments": ["/old/python", "-m", + "loopx_codex_provider_routing.native", "watch", "--runtime-root", str(account)], + "KeepAlive": True, "EnvironmentVariables": {"PATH": "/original/bin", "PYTHONPATH": "/old/core", + "PYTHONHOME": "/old/python/home"}} + path.write_bytes(plistlib.dumps(payload)) + loaded[label] = payload + state = {"intent": None} + held = set() + + class Owner: + def __init__(self, runtime_root): + assert runtime_root == account + self.root = runtime_root / "extensions/codex-native-accounts" + + @contextmanager + def _lock(self, *, name="lock"): + assert name not in held + held.add(name) + try: + yield + finally: + held.remove(name) + + def _load(self): + assert held == {"lock", "api-relay-state.lock"} + return state + + def observer_heartbeat(*_, **__): + now = time.time() + candidate = loaded[label]["ProgramArguments"][0] == sys.executable + return {"build": "candidate" if candidate else "old", + "relay_build": "candidate-relay" if candidate else "old-relay", + "checked_at": now, "deadline_at": now + 60, "phase": "retrying"} + + native = SimpleNamespace(NativeAccounts=Owner, HostError=RuntimeError, + OBSERVER_SCHEMA="observer-fixture", private_read=observer_heartbeat, + observer_build_identity=lambda: {"build": "candidate", "relay_build": "candidate-relay"}) + monkeypatch.setitem(sys.modules, "loopx_codex_provider_routing.native", native) + monkeypatch.setattr(managed_runtime, "_provider_pinned", lambda _: True) + monkeypatch.setattr(managed_runtime, "_native_package", lambda _: native) + monkeypatch.setattr(managed_runtime, "_observer_lock_pid", lambda _: 900 if label in loaded else None) + return agents, target, loaded, calls, fail, state, held, path + + +def test_provider_bound_core_adopts_original_observer_and_keeps_its_route(managed_with_observer): + _, target, loaded, calls, _, _, held, path = managed_with_observer + original = plistlib.loads(path.read_bytes()) + result = align_managed_runtime(str(target)) + assert result["readiness_verified"] + assert calls == [("bootout", "com.loopx.status"), ("bootout", "com.loopx.chat"), + ("bootout", path.stem), ("bootstrap", path.stem), + ("bootstrap", "com.loopx.status"), ("bootstrap", "com.loopx.chat")] + observer = loaded[path.stem] + assert observer["ProgramArguments"] == [sys.executable, *original["ProgramArguments"][1:]] + env = observer["EnvironmentVariables"] + assert env["PYTHONPATH"] == env["LOOPX_RELEASE_ROOT"] == str(target.parent.parent) + assert env["PATH"] == "/original/bin" and "PYTHONHOME" not in env + assert not held + calls.clear() + assert align_managed_runtime(str(target))["restarted_services"] == [] + assert calls == [] + + +@pytest.mark.parametrize("blocker", ["active_intent", "foreign_owner"]) +def test_native_observer_hold_leaves_all_services_running(managed_with_observer, monkeypatch, blocker): + agents, target, loaded, calls, _, state, held, path = managed_with_observer + before = {p: p.read_bytes() for p in agents.glob("*.plist")} + if blocker == "active_intent": + state["intent"] = {"target": "fixture"} + expected = activation.RuntimeActivationDeferred + reason = "native_account_work_pending" + else: + monkeypatch.setattr(managed_runtime, "_observer_lock_pid", lambda _: 901) + expected = ValueError + reason = "native_observer_owner_unverified" + with pytest.raises(expected, match=reason): + align_managed_runtime(str(target)) + assert set(loaded) == {"com.loopx.status", "com.loopx.chat", path.stem} + assert not calls and not held + assert {p: p.read_bytes() for p in before} == before + + +def test_core_only_activation_does_not_retarget_standalone_observer(managed_with_observer, monkeypatch): + _, target, _, calls, _, _, _, path = managed_with_observer + original = path.read_bytes() + monkeypatch.setattr(managed_runtime, "_provider_pinned", lambda _: False) + align_managed_runtime(str(target)) + assert path.read_bytes() == original + assert all(label != path.stem for _, label in calls) + + +def test_core_only_update_preserves_an_adopted_provider_binding(managed_with_observer, monkeypatch): + agents, target, _, calls, _, _, _, _ = managed_with_observer + align_managed_runtime(str(target)) + calls.clear() + before = {p: p.read_bytes() for p in agents.glob("*.plist")} + monkeypatch.setattr(managed_runtime, "_provider_pinned", lambda _: False) + with pytest.raises(activation.RuntimeActivationDeferred, match="native_observer_provider_binding_required"): + align_managed_runtime(str(target)) + assert not calls + assert {p: p.read_bytes() for p in before} == before + + +def test_provider_activation_leaves_another_account_root_alone(managed_with_observer): + agents, target, _, calls, _, _, _, path = managed_with_observer + registry = agents.parent.parent / "registry" + registry.write_text(json.dumps({"goals": [], "common_runtime_root": str(agents / "other owner")})) + original = path.read_bytes() + align_managed_runtime(str(target)) + assert path.read_bytes() == original + assert all(label != path.stem for _, label in calls) + + +def test_handoff_reads_relay_build_even_when_observer_build_matches(managed_with_observer, monkeypatch): + _, target, _, _, _, _, _, _ = managed_with_observer + native = sys.modules["loopx_codex_provider_routing.native"] + read_heartbeat = native.private_read + builds = [] + + def readback(*args, **kwargs): + value = read_heartbeat(*args, **kwargs) + assert value["build"] == "candidate" + if not builds: + value["relay_build"] = "old-relay" + builds.append(value["relay_build"]) + return value + + monkeypatch.setattr(native, "private_read", readback) + assert align_managed_runtime(str(target))["readiness_verified"] + assert builds[0] == "old-relay" and builds[1:] and set(builds[1:]) == {"candidate-relay"} + + +def test_failed_chat_readiness_rolls_back_observer_and_core(managed_with_observer, monkeypatch): + agents, target, loaded, calls, _, _, held, path = managed_with_observer + before = {p: p.read_bytes() for p in agents.glob("*.plist")} + 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()} + assert calls[-3:] == [("bootstrap", path.stem), ("bootstrap", "com.loopx.status"), + ("bootstrap", "com.loopx.chat")] + assert not held and not (agents / ".loopx-runtime-activation.json").exists() + + +def test_interrupted_observer_handoff_restores_original_routes(managed_with_observer, monkeypatch): + agents, target, loaded, calls, _, _, held, path = managed_with_observer + before = {p: p.read_bytes() for p in agents.glob("*.plist")} + write = managed_runtime.write_bytes_atomic + def crash(plist, content, **kwargs): + write(plist, content, **kwargs) + if plist == path: + raise KeyboardInterrupt("fixture process interruption") + monkeypatch.setattr(managed_runtime, "write_bytes_atomic", crash) + with pytest.raises(KeyboardInterrupt): + align_managed_runtime(str(target)) + journal = agents / ".loopx-runtime-activation.json" + assert journal.exists() and not loaded and not held + monkeypatch.setattr(managed_runtime, "write_bytes_atomic", write) + managed_runtime._recover_alignment(journal) + assert {p: p.read_bytes() for p in before} == before + assert loaded == {p.stem: plistlib.loads(content) for p, content in before.items()} + assert calls[-3:] == [("bootstrap", path.stem), ("bootstrap", "com.loopx.status"), + ("bootstrap", "com.loopx.chat")] + assert not held and not journal.exists() + + def test_mutable_cli_alias_cannot_hide_a_new_runtime_from_the_pair(managed): _, agents, target, loaded, calls, _ = managed alias = target.parent.parent / "installed alias" / "loopx"