From ad58340ae95cd68d8a48edbb1cdb5e5b54a05e73 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 9 Nov 2026 07:41:29 +0800 Subject: [PATCH 1/2] fix(runtime): adopt the native observer with its Core release Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/hosts/managed_service_runtime.py | 272 ++++++++++++++++-- .../loopx-codex-provider-routing/NATIVE.md | 12 + .../pyproject.toml | 2 +- .../native-extension.toml | 2 +- .../loopx_codex_provider_routing/native.py | 7 +- tests/test_managed_service_runtime.py | 175 +++++++++++ 6 files changed, 446 insertions(+), 24 deletions(-) diff --git a/loopx/hosts/managed_service_runtime.py b/loopx/hosts/managed_service_runtime.py index e54a473d3f..cea89b3129 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,10 @@ from .. import runtime_activation as activation +_OBSERVER_LABEL = "com.loopx.codex-native-accounts" +_OBSERVER_MODULE = "loopx_codex_provider_routing.native" + + 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. @@ -124,7 +133,7 @@ def has_runtime_binding() -> bool: 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 +143,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 +176,151 @@ 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, + "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 _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 +383,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 +397,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 +438,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 +496,52 @@ 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 + if chat and observer.exists() and _provider_pinned(python): + if observer.is_symlink(): + raise ValueError("native_observer_route_unverified") + original = observer.read_bytes() + payload = plistlib.loads(original) + root = _observer_route(payload) + if root.resolve() == _chat_account_root(plistlib.loads(chat[0].read_bytes())): + 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 +556,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..77e0babfa9 100644 --- a/packages/loopx-codex-provider-routing/NATIVE.md +++ b/packages/loopx-codex-provider-routing/NATIVE.md @@ -42,6 +42,18 @@ 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. Core-only updates leave the +independent watcher 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..f9cd49d0a9 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,179 @@ 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_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" From 7efecf65e65bc94c6be15d70a2b334f3c33d9afc Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 9 Nov 2026 07:45:46 +0800 Subject: [PATCH 2/2] fix(runtime): preserve an adopted provider across Core updates Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/hosts/managed_service_runtime.py | 35 ++++++++++++++----- .../loopx-codex-provider-routing/NATIVE.md | 8 +++-- tests/test_managed_service_runtime.py | 12 +++++++ 3 files changed, 44 insertions(+), 11 deletions(-) diff --git a/loopx/hosts/managed_service_runtime.py b/loopx/hosts/managed_service_runtime.py index cea89b3129..4b279ea885 100644 --- a/loopx/hosts/managed_service_runtime.py +++ b/loopx/hosts/managed_service_runtime.py @@ -30,6 +30,7 @@ _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]: @@ -95,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 @@ -128,7 +129,7 @@ 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 @@ -201,7 +202,7 @@ def _retarget_observer(payload: dict, python: str, core: Path) -> dict: 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, + return {**payload, "EnvironmentVariables": env, "LoopXRuntimeBinding": _RUNTIME_BINDING.copy(), "ProgramArguments": [python, *payload["ProgramArguments"][1:]]} @@ -213,6 +214,23 @@ def _provider_pinned(python: str) -> bool: 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]) @@ -500,13 +518,14 @@ def _align_managed_runtime(executable: str, journal: Path) -> dict: observer = journal.parent / f"{_OBSERVER_LABEL}.plist" observer_root = None native = expected_observer = None - if chat and observer.exists() and _provider_pinned(python): - if observer.is_symlink(): - raise ValueError("native_observer_route_unverified") - original = observer.read_bytes() - payload = plistlib.loads(original) + 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) diff --git a/packages/loopx-codex-provider-routing/NATIVE.md b/packages/loopx-codex-provider-routing/NATIVE.md index 77e0babfa9..8b84af8a90 100644 --- a/packages/loopx-codex-provider-routing/NATIVE.md +++ b/packages/loopx-codex-provider-routing/NATIVE.md @@ -50,9 +50,11 @@ 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. Core-only updates leave the -independent watcher in place. Run the provider's separate relay reconciliation -after the managed processes have adopted the same build. +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: diff --git a/tests/test_managed_service_runtime.py b/tests/test_managed_service_runtime.py index f9cd49d0a9..1241314dd8 100644 --- a/tests/test_managed_service_runtime.py +++ b/tests/test_managed_service_runtime.py @@ -204,6 +204,18 @@ def test_core_only_activation_does_not_retarget_standalone_observer(managed_with 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"