From b95e42f1e1b1beed241c900898c03df1c0043bd3 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Thu, 29 Oct 2026 13:23:16 +0800
Subject: [PATCH 1/2] fix(codex): resume quota handoffs after transient client
 holds

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 loopx/runtime_activation.py                   |  53 +++--
 .../loopx-codex-provider-routing/NATIVE.md    |  34 ++-
 .../extension.toml                            |   2 +-
 .../pyproject.toml                            |   2 +-
 .../smoke/native_host_clients_smoke.py        |  38 +--
 .../loopx_codex_provider_routing/__init__.py  |   2 +-
 .../loopx_codex_provider_routing/native.py    |  13 +-
 .../native_host.py                            | 223 +++++++++++-------
 tests/presentation/test_codex_accounts.py     | 106 +++++++++
 9 files changed, 352 insertions(+), 121 deletions(-)

diff --git a/loopx/runtime_activation.py b/loopx/runtime_activation.py
index f7eacb1699..998ee372cf 100644
--- a/loopx/runtime_activation.py
+++ b/loopx/runtime_activation.py
@@ -7,7 +7,7 @@
 
 from __future__ import annotations
 
-from contextlib import ExitStack
+from contextlib import ExitStack, contextmanager
 import os
 from pathlib import Path
 import plistlib
@@ -93,9 +93,9 @@ def _managed_service_plists() -> list[Path]:
     ]
 
 
-def _managed_service_pid(plist: Path) -> int | None:
+def _managed_service_pid(plist: Path, *, label: str | None = None) -> int | None:
     result = subprocess.run(
-        ["launchctl", "print", f"gui/{os.getuid()}/{plist.stem}"],
+        ["launchctl", "print", f"gui/{os.getuid()}/{label or plist.stem}"],
         capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=5,
     )
     if result.returncode != 0:
@@ -113,6 +113,32 @@ def _managed_service_pid(plist: Path) -> int | None:
     return int(pids[0]) if len(pids) == 1 else None
 
 
+@contextmanager
+def managed_chat_activation_guard(
+    plists: list[Path], *, expected_services: dict[Path, tuple[str, int]] | None = None,
+):
+    """Keep accepted Chat work and new ingress safe through a client restart.
+
+    Account-provider handoffs and runtime updates share the same store owner;
+    an empty Session summary is not sufficient evidence of an idle service.
+    The caller must still verify its loaded service/PID before stopping it.
+    """
+    from .chat_ingress import chat_runtime_activation_lock, chat_runtime_activation_state
+
+    labels = [plist.stem for plist in plists]
+    roots = sorted({_chat_runtime_root(plist) for plist in plists})
+    with ExitStack() as locks:
+        for root in roots:
+            locks.enter_context(chat_runtime_activation_lock(root, timeout_seconds=2.0))
+        for root in roots:
+            if chat_runtime_activation_state(root)["nonterminal_turns"]:
+                raise RuntimeActivationDeferred("chat_work_pending", labels)
+        for plist, (label, pid) in (expected_services or {}).items():
+            if plist not in plists or _managed_service_pid(plist, label=label) != pid:
+                raise ValueError("chat_service_route_unresolved")
+        yield
+
+
 def replace_managed_loopx_chat(host: str, port: int, *, timeout_seconds: float = 5.0) -> str:
     """Replace the exact managed listener only while its accepted work is idle.
 
@@ -190,21 +216,8 @@ def restart_managed_loopx_services() -> list[str]:
             chat_plists.append(plist)
     restarted: list[str] = []
     uid = os.getuid() if hasattr(os, "getuid") else 0
-    from .chat_ingress import (
-        chat_runtime_activation_lock,
-        chat_runtime_activation_state,
-    )
-
     try:
-        roots = sorted({_chat_runtime_root(plist) for plist in chat_plists})
-        with ExitStack() as locks:
-            for root in roots:
-                locks.enter_context(
-                    chat_runtime_activation_lock(root, timeout_seconds=2.0)
-                )
-            for root in roots:
-                if chat_runtime_activation_state(root)["nonterminal_turns"]:
-                    raise RuntimeActivationDeferred("chat_work_pending", labels)
+        with managed_chat_activation_guard(chat_plists):
             # Keep admission fenced until launchctl has replaced the old process.
             # Existing workers need no new lock: any accepted queued/active work
             # was checked above, including completing and orphaned queued Turns.
@@ -227,8 +240,10 @@ def restart_managed_loopx_services() -> list[str]:
                     raise RuntimeActivationFailed(
                         restarted, [s for s in labels if s not in restarted]
                     ) from exc
-    except RuntimeActivationDeferred:
-        raise
+    except RuntimeActivationDeferred as exc:
+        # The shared Chat guard names its participants; an update still defers
+        # the complete original status/chat activation batch.
+        raise RuntimeActivationDeferred(str(exc), labels) from exc
     except (
         OSError,
         ValueError,
diff --git a/packages/loopx-codex-provider-routing/NATIVE.md b/packages/loopx-codex-provider-routing/NATIVE.md
index a81c949abb..b291cafded 100644
--- a/packages/loopx-codex-provider-routing/NATIVE.md
+++ b/packages/loopx-codex-provider-routing/NATIVE.md
@@ -21,8 +21,9 @@ inspected GPT instance, waits for exit and native idle, saves/switches the login
 then reopens that same route. It never addresses a shared bundle id or quits a
 different route. With the separate force permission below, a refused or timed-out
 normal quit can escalate against that same inspected instance. Otherwise those
-failures, or remaining CLI/file owners, leave credentials unchanged and pause
-automatic shutdowns until a check. Changed process ownership leaves credentials
+failures leave credentials unchanged and pause automatic shutdowns until a check.
+Remaining CLI/file owners wait without consuming that failure budget and resume
+automatically after the blocker releases. Changed process ownership leaves credentials
 unchanged and consumes the bounded retry budget.
 Without this additional permission it continues to wait for closed clients.
 It does not implement hot login replacement, job replay or
@@ -106,6 +107,35 @@ unreadable live Codex process into a proven exit.
 只读重扫，再以稳定的新身份清单进入原受控重启流程。PID 复用、仍存活但身份不明及
 持续变化继续等待；重扫不重放退出、凭据事务或业务，也不扩大 VM 的重启权限。
 
+Version 0.12.11 checks for non-restartable holders before quitting the GPT App.
+An existing `other_clients_busy` pause is rechecked automatically; an unchanged
+unmanaged holder waits without repeatedly closing/reopening clients. Normal
+quit refusal, unknown effects and other exhausted failure budgets keep their
+existing recovery boundary. Only numeric read-only directory descriptors are
+excluded from auth consumers. A read-only regular file, missing descriptor type,
+directory writer or warm Codex process still blocks. The native slot owner's idle
+check rechecks these read-only exceptions, followed by another typed handle scan
+under its account locks. This does not restart a VM or claim to inspect a guest's
+cached login; a pinned auth file remains a hold.
+
+Before stopping a discovered Chat service, the provider uses LoopX's existing
+accepted-work reader and ingress lock for its explicit runtime. Queued work,
+including an orphan without an active Session pointer, completing work and
+unreadable state defer the entire handoff before App quit. A new/changed service
+cannot inherit the frozen restart set. The lock remains held through account
+readback and restoration. A standalone observer must run in the same upgraded
+LoopX environment to restart these services; missing Core support waits safely.
+Prefer the serving App/Chat observer so normal LoopX updates also refresh it;
+retire any separately installed legacy watcher through its original manager.
+
+0.12.11 在退出 GPT App 前检查无法受控恢复的占用；原 `other_clients_busy` 暂停会
+自动重新检查，阻塞解除后无需手动换号。仅有数字 FD、只读且类型明确为目录的句柄
+不视为认证消费者；认证文件、写句柄、未知类型及缓存登录的 Codex 进程仍阻塞。
+原账号 owner 在事务锁内复查只读例外，并再次核验句柄；不重启 VM，也不据此声称
+已检查 guest 内的认证缓存。停止 Chat 前复用原在途任务读取与 ingress 锁，排队、
+completing 或不可读状态均在退出 App 前等待；恢复期间保持锁与原失败/未知回执。
+观察器应随实际 LoopX 服务升级；独立旧 watcher 通过原管理器退役，避免占住唯一观察权。
+
 Each service's restoration intent is saved privately before unloading. Version
 0.12.7 waits up to five seconds for launchd to finish an acknowledged asynchronous
 unload; a job that remains loaded retains its recovery intent and fails without
diff --git a/packages/loopx-codex-provider-routing/extension.toml b/packages/loopx-codex-provider-routing/extension.toml
index 4dd7003b31..638f1656e2 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.10"
+version = "0.12.11"
 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 40e7c38e05..55406ab9bf 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.10"
+version = "0.12.11"
 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 6ad7025538..b8548239f5 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
@@ -14,6 +14,7 @@
 from unittest.mock import patch
 
 PACKAGE_ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(PACKAGE_ROOT.parents[1]))  # The actual LoopX service fence.
 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  # noqa: E402
@@ -38,7 +39,8 @@ def __init__(self, home):
         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.service_args = ["/test/chat", "--runtime-root", str(home / "runtime"), "chat"]
+        self.plist.write_bytes(plistlib.dumps({"Label": LABEL, "ProgramArguments": self.service_args}))
         self.plist.chmod(0o600)
         self.root = home / "native-accounts"
         self.root.mkdir(mode=0o700)
@@ -84,8 +86,9 @@ 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":
+            arguments = "arguments = {\n" + "\n".join(self.service_args) + "\n}\n"
             return SimpleNamespace(returncode=0 if self.loaded else 113,
-                                   stdout=f"path = {self.plist}\npid = {self.service_pid}\n" if self.loaded else "")
+                                   stdout=f"path = {self.plist}\npid = {self.service_pid}\n{arguments}" if self.loaded else "")
         if verb == "bootout":
             self.loaded = False
             self.holders = [p for p in self.holders if p != 301]
@@ -129,6 +132,10 @@ def context(self):
     def handoff(self, **kwargs):
         return self.host.switch("b", "before", reopen=True, restart_clients=True, **kwargs)
 
+    def pause(self):
+        with self.host.restart_guard():
+            return self.host.pause_chat_agents()
+
 
 def error(code, operation):
     try:
@@ -229,7 +236,8 @@ def failed_environment(pid):
     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"]}))
+    f.plist.write_bytes(plistlib.dumps({"Label": renamed, "ProgramArguments": [
+        "/test/chat", "--runtime-root", str(f.home / "runtime"), "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)
@@ -268,7 +276,7 @@ def stuck_unload(command, **kwargs):
     f.run = stuck_unload
     with f.context(), patch.object(host_module.time, "monotonic", side_effect=iter(range(0, 100, 2))), \
             patch.object(host_module.time, "sleep"):
-        error("chat_pause_failed", f.host.pause_chat_agents)
+        error("chat_pause_failed", f.pause)
     assert f.switches == 0 and private_read(f.root / "chat-recovery.json")["agents"]
 
     # A second independent service participates in the same transaction. Neither
@@ -276,17 +284,19 @@ def stuck_unload(command, **kwargs):
     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"]}))
+    second.write_bytes(plistlib.dumps({"Label": "org.example.second", "ProgramArguments": [
+        "/test/other", "--runtime-root", str(f.home / "runtime"), "chat"]}))
     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
+        if (Path(command[0]).name == "launchctl" and
                 (command[-1].endswith("/org.example.second") or command[-1] == str(second))):
             f.calls.append(command)
             if command[1] == "print":
+                arguments = "arguments = {\n" + "\n".join(plistlib.loads(second.read_bytes())["ProgramArguments"]) + "\n}\n"
                 return SimpleNamespace(returncode=0 if second_state["loaded"] else 113,
-                                       stdout=f"path = {second}\npid = {second_state['pid']}\n")
+                                       stdout=f"path = {second}\npid = {second_state['pid']}\n{arguments}")
             if command[1] == "bootout":
                 second_state["loaded"] = False
                 f.records.pop(700)
@@ -325,7 +335,7 @@ def multi_run(command, **kwargs):
         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)
+                error("waiting_for_idle", f.handoff)
             else:
                 assert f.handoff()["active_slot"] == "b"
         assert 200 not in f.killed and f.switches == (0 if same_home else 1)
@@ -401,13 +411,13 @@ def changed_run(command, **kwargs):
             f.host, "before", inspected=(100, (1, EXE, 99, 0))))
 
     # A remaining independent CLI blocks the full switch and leaves its work
-    # untouched; the source App is reopened without changing the account.
+    # untouched. Read-only preflight leaves even the source App running.
     f = fixture()
     clock = iter(range(0, 1000, 20))
     with f.context(), patch.object(host_module.time, "monotonic", side_effect=lambda: next(clock)):
-        error("other_clients_busy", f.handoff)
-    assert f.killed == [101] and f.holders == [200] and f.switches == 0
-    assert f.active == "a" and f.app_pid is not None
+        error("waiting_for_idle", f.handoff)
+    assert not f.killed and f.holders == [101, 200] and f.switches == 0
+    assert f.active == "a" and f.app_pid == 100
 
     # PID reuse, a prior orphan, a new launch and unprovable identity stay holds.
     for mutation in ("pid_reused", "unreadable", "new_launch", "selection_changed", "quota_stale"):
@@ -496,7 +506,7 @@ def changed_run(command, **kwargs):
     f = fixture()
     f.holders = [301]
     with f.context():
-        f.host.pause_chat_agents()
+        f.pause()
         f.plist.write_text("changed by user")
         error("chat_resume_failed", f.host.resume_chat_agents)
     assert not f.loaded
@@ -507,7 +517,7 @@ def changed_run(command, **kwargs):
     f = fixture()
     f.holders = [301]
     with f.context():
-        f.host.pause_chat_agents()
+        f.pause()
         journal = f.root / "chat-recovery.json"
         recovery = private_read(journal)
         recovery["owner"]["pid"] = 500
diff --git a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/__init__.py b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/__init__.py
index 964fe497bf..e8886a6460 100644
--- a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/__init__.py
+++ b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/__init__.py
@@ -1,3 +1,3 @@
 __all__ = ["__version__"]
 
-__version__ = "0.12.10"
+__version__ = "0.12.11"
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 26bdc643c5..b1e2df1e6f 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
@@ -279,7 +279,12 @@ def tick(self, *, refresh: bool = False, manual_slot: str | None = None,
                 # Only status/readback runs while paused; unchanged selection
                 # must not retry usage refresh, client shutdown or switching.
                 if state.get("failure_count", 0) >= 3 and not refresh and not manual_slot:
-                    return self._projection(state)
+                    if state.get("code") != "other_clients_busy":
+                        return self._projection(state)
+                    # Legacy busy pauses are safe to recheck: the native host
+                    # now resolves remaining owners BEFORE any client quit.
+                    # A released blocker must not require a manual account swap.
+                    state["failure_count"] = 0
                 # Inspect the active account even when it is not an authorized
                 # destination. Candidates stay strictly inside the allowlist.
                 slots = list(dict.fromkeys([native["active_slot"], *state["policy"]["slots"]]))
@@ -336,10 +341,10 @@ def tick(self, *, refresh: bool = False, manual_slot: str | None = None,
                 code = error.code if isinstance(error, HostError) else "native_unavailable"
                 # Unrelated/remaining owners still require a safe-boundary wait.
                 # Other repeated failures stop automatic attempts until Refresh.
-                failures = state.get("failure_count", 0) + (code not in ("waiting_for_idle", "handoff_pending"))
-                if code in ("other_clients_busy", "client_quit_refused", "client_quit_timeout"):
+                failures = state.get("failure_count", 0) + (code not in ("waiting_for_idle", "other_clients_busy", "handoff_pending"))
+                if code in ("client_quit_refused", "client_quit_timeout"):
                     failures = 3  # Do not repeatedly close/reopen a refused client.
-                state.update(action=code if code in ("waiting_for_idle", "selection_changed", "adapter_changed", "client_quit_refused", "client_quit_timeout", "client_changed", "handoff_pending") else "unavailable",
+                state.update(action="waiting_for_idle" if code == "other_clients_busy" else code if code in ("waiting_for_idle", "selection_changed", "adapter_changed", "client_quit_refused", "client_quit_timeout", "client_changed", "handoff_pending") else "unavailable",
                              code=code, retry_at=self.clock() + (5 if code == "handoff_pending" else min(300, 60 * 2 ** min(failures, 3))), failure_count=failures)
                 if manual_slot:
                     state["last_operation_id"] = operation_id
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 fc2f0d5e7b..467f1886fa 100644
--- a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native_host.py
+++ b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native_host.py
@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 import hashlib
+from contextlib import ExitStack, contextmanager
 import ctypes
 import importlib
 import json
@@ -145,12 +146,18 @@ def __init__(self, adapter: dict, *, user_home: Path | None = None, reconcile_se
         def idle():
             if self.not_after is not None and time.time() >= self.not_after:
                 raise HostError("quota_stale")
+            # A filesystem server holding only read-only directory handles
+            # neither pins auth.json nor proves a cached Codex login. Keep
+            # real file owners and warm Codex processes as independent holds.
+            if self._home_owners():
+                raise HostError("waiting_for_idle")
             try:
-                self.apps.account_idle(self.home, self.spec)
+                self.apps.account_idle(self.home, self.spec,
+                                       read_only_owners=tuple(self._directory_identities))
             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():
+            if self._home_owners():
                 raise HostError("waiting_for_idle")
             if self.expected is not None and self.revision() != self.expected:
                 raise HostError("selection_changed")
@@ -296,13 +303,26 @@ def _home_owners(self) -> list[int]:
         # which was never inspected. Rescan only discovery; never retry effects.
         for _ in range(3):
             processes = self._processes()
-            result = subprocess.run(["/usr/sbin/lsof", "-Fp", "+d", str(self.spec["home"])],
+            result = subprocess.run(["/usr/sbin/lsof", "-Fpfat", "+d", str(self.spec["home"])],
                                     capture_output=True, text=True, timeout=10)
             if result.returncode not in (0, 1):
                 raise HostError("ownership_unverified")
-            owners = {int(line[1:]) for line in result.stdout.splitlines()
-                      if line.startswith("p") and line[1:].isdigit()}
-            owners |= self._cached_codex_owners(processes)
+            handles, owner, descriptor = {}, None, None
+            for line in result.stdout.splitlines():
+                if line.startswith("p") and line[1:].isdigit():
+                    owner, descriptor = int(line[1:]), None
+                    handles.setdefault(owner, [])
+                elif line.startswith("f") and owner is not None:
+                    descriptor = {"fd": line[1:].lstrip("*")}
+                    handles[owner].append(descriptor)
+                elif line[:1] in ("a", "t") and descriptor is not None and line[:1] not in descriptor:
+                    descriptor[line[0]] = line[1:]
+                elif line:
+                    raise HostError("ownership_unverified")
+            cached = self._cached_codex_owners(processes)
+            directories = {pid for pid, files in handles.items() if files and all(
+                f["fd"].isdigit() and f.get("a") == "r" and f.get("t") == "DIR" for f in files)} - cached
+            owners = set(handles) | cached
             unstable = False
             for pid in owners:
                 current, previous = process_identity(pid), processes.get(pid)
@@ -315,7 +335,8 @@ def _home_owners(self) -> list[int]:
                     raise HostError("client_changed")  # Reuse/unknown is a hold.
             if not unstable:
                 self._owner_identities = {p: processes[p] for p in owners}
-                return sorted(owners)
+                self._directory_identities = {p: processes[p] for p in directories}
+                return sorted(owners - directories)
         raise HostError("client_changed")
 
     def _processes(self) -> dict:
@@ -527,6 +548,44 @@ def clients(self) -> dict:
              **({"service": services[p]} if p in services else {}),
              **(self._external_manager(p) if p not in app and p not in services else {})} for p in owners]}
 
+    @contextmanager
+    def restart_guard(self):
+        """Resolve blockers before quitting an App; fence accepted Chat work."""
+        owners = self._home_owners()
+        inspected, helpers = self.inspect_client_helpers()
+        agents = self._consumer_agents(owners)
+        restartable = set(helpers) | ({inspected[0]} if inspected[0] else set())
+        restartable.update(p for _, _, _, _, children in agents for p in children)
+        if set(owners) - restartable:
+            raise HostError("waiting_for_idle")
+        if not agents:
+            yield
+            return
+        try:
+            # Reuse the running product's accepted-work reader and ingress
+            # lock. Missing/unsupported Core cannot authorize a service stop.
+            from loopx.runtime_activation import managed_chat_activation_guard, RuntimeActivationDeferred
+        except ImportError as error:
+            raise HostError("waiting_for_idle") from error
+        frozen = {label: (pid, identity, self._chat_definition(plist, label))
+                  for label, plist, pid, identity, _ in agents}
+        with ExitStack() as guard:
+            try:
+                guard.enter_context(managed_chat_activation_guard(
+                    [plist for _, plist, _, _, _ in agents],
+                    expected_services={plist: (label, pid) for label, plist, pid, _, _ in agents},
+                ))
+                if any(self._chat_definition(plist, label) != frozen[label][2]
+                       for label, plist, _, _, _ in agents):
+                    raise HostError("client_changed")
+            except (RuntimeActivationDeferred, OSError, ValueError, TimeoutError) as error:
+                raise HostError("waiting_for_idle") from error
+            try:
+                self._guarded_agents = frozen
+                yield
+            finally:
+                self._guarded_agents = {}
+
     def _chat_recovery(self) -> tuple[Path, dict]:
         from .native import SCHEMA, private_read
         path = self.manager.root / "chat-recovery.json"
@@ -576,6 +635,9 @@ def pause_chat_agents(self) -> dict:
         for label, plist, service_pid, service, descendants in self._consumer_agents(self._home_owners()):
             if label in recovery["agents"]:
                 continue
+            if getattr(self, "_guarded_agents", {}).get(label) != (
+                    service_pid, service, self._chat_definition(plist, label)):
+                raise HostError("client_changed")
             target = f"gui/{os.getuid()}/{label}"
             digest = self._chat_definition(plist, label)
             fresh = set(self._home_owners())
@@ -694,81 +756,84 @@ def switch(self, slot: str, expected: str, *, reopen: bool, restart_clients: boo
         # idle() is rechecked by the native owner under launch/account locks,
         # and immediately before its first mutation. This is the CAS fence,
         # including against manual selections outside LoopX.
-        result, restore_error = None, None
-        try:
+        with ExitStack() as guard:
+            if restart_clients:
+                guard.enter_context(self.restart_guard())
+            result, restore_error = None, None
             try:
-                if restart_clients:
-                    if not_after is not None and time.time() >= not_after:
-                        raise HostError("quota_stale")
-                    inspected, helpers = self.inspect_client_helpers()
-                    self.close_client(expected, inspected=inspected,
-                                      **({"force": True} if force_quit_clients else {}))
-                    # Orphaned App helpers and idle chat transports are
-                    # product-owned clients of this home. Retire or pause them;
-                    # an unrelated CLI/file owner remains a hold, never a kill
-                    # target.
-                    self.retire_app_helpers(helpers)
-                    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 restart_clients:
+                        if not_after is not None and time.time() >= not_after:
+                            raise HostError("quota_stale")
+                        inspected, helpers = self.inspect_client_helpers()
+                        self.close_client(expected, inspected=inspected,
+                                          **({"force": True} if force_quit_clients else {}))
+                        # Orphaned App helpers and idle chat transports are
+                        # product-owned clients of this home. Retire or pause them;
+                        # an unrelated CLI/file owner remains a hold, never a kill
+                        # target.
+                        self.retire_app_helpers(helpers)
+                        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:
+                                if error.code != "waiting_for_idle":
+                                    raise
+                                if time.monotonic() >= deadline:
+                                    raise HostError("other_clients_busy") from None
+                                # 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:
+                    # A refused/failed transaction must not unnecessarily leave a
+                    # normally closed source App down. Native preflight blocks
+                    # opening an incomplete credential journal; a later manual
+                    # choice wins.
+                    if restart_clients and before.get("client_running"):
                         try:
-                            if not self._helpers_stopped(helpers):
-                                raise HostError("waiting_for_idle")
-                            self.manager.idle()
-                            break
-                        except HostError as error:
-                            if error.code != "waiting_for_idle":
-                                raise
-                            if time.monotonic() >= deadline:
-                                raise HostError("other_clients_busy") from None
-                            # 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:
-                # A refused/failed transaction must not unnecessarily leave a
-                # normally closed source App down. Native preflight blocks
-                # opening an incomplete credential journal; a later manual
-                # choice wins.
-                if restart_clients and before.get("client_running"):
-                    try:
-                        if self.status().get("client_running") is False:
-                            self.reopen(expected)
-                    except (HostError, OSError, ValueError, subprocess.SubprocessError):
-                        pass
-                raise
-        finally:
-            # Restore product-owned transports even when the handoff fails.
-            try:
-                self.resume_chat_agents()
-            except HostError as error:
-                if result is None:
+                            if self.status().get("client_running") is False:
+                                self.reopen(expected)
+                        except (HostError, OSError, ValueError, subprocess.SubprocessError):
+                            pass
                     raise
-                # A committed account still needs its App reopened. Report the
-                # retained service recovery after that original continuation.
-                restore_error = error
-        value = self.status()
-        if value["active_slot"] != slot or value["current_login_matches"] is not True:
-            raise HostError("readback_failed")
-        launch_failed = False
-        if reopen:
-            try:
-                self.apps.open_app(self.home, "gpt", [])
-            except (OSError, ValueError, subprocess.SubprocessError):
-                launch_failed = True
-        after = self.status()
-        if after["selection_revision"] != value["selection_revision"]:
-            raise HostError("selection_changed")
-        value = after
-        if restore_error is not None:
-            raise restore_error
-        return value | {"history_untouched": result["history_untouched"], "launch_failed": launch_failed}
+            finally:
+                # Restore product-owned transports even when the handoff fails.
+                try:
+                    self.resume_chat_agents()
+                except HostError as error:
+                    if result is None:
+                        raise
+                    # A committed account still needs its App reopened. Report the
+                    # retained service recovery after that original continuation.
+                    restore_error = error
+            value = self.status()
+            if value["active_slot"] != slot or value["current_login_matches"] is not True:
+                raise HostError("readback_failed")
+            launch_failed = False
+            if reopen:
+                try:
+                    self.apps.open_app(self.home, "gpt", [])
+                except (OSError, ValueError, subprocess.SubprocessError):
+                    launch_failed = True
+            after = self.status()
+            if after["selection_revision"] != value["selection_revision"]:
+                raise HostError("selection_changed")
+            value = after
+            if restore_error is not None:
+                raise restore_error
+            return value | {"history_untouched": result["history_untouched"], "launch_failed": launch_failed}
 
     def recover(self, rollback: bool, expected: str) -> dict:
         if sys.platform != "darwin":
diff --git a/tests/presentation/test_codex_accounts.py b/tests/presentation/test_codex_accounts.py
index 47e31ecdf9..e42c175b03 100644
--- a/tests/presentation/test_codex_accounts.py
+++ b/tests/presentation/test_codex_accounts.py
@@ -348,6 +348,112 @@ def test_paused_unchanged_selection_does_not_repeat_usage_or_shutdown(native, tm
     assert provider._load()["failure_count"] == 3
 
 
+def test_busy_blocker_release_resumes_without_refresh_or_account_choice(native, tmp_path):
+    provider, transport, clock = configured(native, tmp_path)
+    provider.configure(expected_revision=1, enabled=True, slots=["a", "b"], restart_clients=True)
+    attempts = []
+    busy = True
+
+    def handoff(adapter, operation, **parameters):
+        if operation == "switch":
+            attempts.append(parameters)
+            if busy:
+                raise native.HostError("other_clients_busy")
+        return transport(adapter, operation, **parameters)
+
+    for _ in range(4):
+        restarted = native.NativeAccounts(tmp_path, transport=handoff, clock=clock)
+        assert restarted.tick()["action"] == "waiting_for_idle"
+        assert restarted._load()["failure_count"] == 0
+        clock.now += 301
+    busy = False
+    assert restarted.tick()["active_slot"] == "b"
+    clock.now += 301
+    assert restarted.tick()["action"] == "ready"
+    assert len(attempts) == 5  # No second account transaction after success.
+
+
+def test_existing_busy_pause_reenters_new_readonly_preflight(native, tmp_path):
+    provider, transport, clock = configured(native, tmp_path)
+    state = provider._load()
+    state.update(failure_count=3, code="other_clients_busy", action="unavailable")
+    native.atomic(provider.path, state)
+    restarted = native.NativeAccounts(tmp_path, transport=transport, clock=clock)
+    assert restarted.tick()["action"] == "selected"
+    assert restarted._load()["failure_count"] == 0
+    assert len([call for call in transport.calls if call[0] == "switch"]) == 1
+
+
+@pytest.mark.parametrize("status", ["queued", "starting", "running", "completing", "interrupting", "invalid"])
+def test_native_restart_uses_real_accepted_work_fence_before_app_quit(native, monkeypatch, tmp_path, status):
+    from loopx.chat_store import ChatSessionStore
+    host_module = importlib.import_module("loopx_codex_provider_routing.native_host")
+    store = ChatSessionStore(tmp_path / "runtime")
+    sid = store.create_session(goal_id="test", agent_id="test", adapter_kind="fake",
+                               upstream_thread_id="test", upstream_mode="managed")["session_id"]
+    turn, _ = store.create_queued_turn(sid, client_turn_id="original", message="preserve this input")
+    store.update_turn(sid, turn["turn_id"], status=status)
+    before = {p: p.read_bytes() for p in store.root.rglob("*.json")}
+    plist = tmp_path / "renamed.plist"
+    plist.write_bytes(plistlib.dumps({"Label": "renamed", "ProgramArguments": ["loopx", "--runtime-root", str(tmp_path / "runtime"), "chat"]}))
+    host = host_module.NativeHost.__new__(host_module.NativeHost)
+    host.resume_chat_agents = lambda: None
+    host._chat_recovery = lambda: (None, {"agents": {}})
+    host.status = lambda: {"selection_revision": "fixture", "active_slot": "a",
+                          "current_login_matches": True, "recovery_required": False,
+                          "client_running": True}
+    host._home_owners = lambda: [301]
+    host.inspect_client_helpers = lambda: ((100, (1, "/test/app", 1, 1)), {})
+    host._consumer_agents = lambda _: [("renamed", plist, 300, (1, "/test/chat", 1, 1), {301: ("/test/codex", 1, 1)})]
+    monkeypatch.setattr("loopx.runtime_activation._managed_service_pid", lambda p, *, label: 300 if p == plist and label == "renamed" else None)
+
+    class OwnedQuitReached(Exception):
+        pass
+
+    quits = []
+    def quit_client(*_args, **_kwargs):
+        quits.append("quit")
+        from loopx.chat_ingress import chat_runtime_activation_lock
+        with pytest.raises(TimeoutError):
+            with chat_runtime_activation_lock(tmp_path / "runtime", timeout_seconds=0.01):
+                pytest.fail("new input was not fenced during client quit")
+        raise OwnedQuitReached()  # Stop before any synthetic credential effect.
+
+    host.close_client = quit_client
+    with pytest.raises(native.HostError, match="waiting_for_idle"):
+        host.switch("b", "fixture", reopen=True, restart_clients=True)
+    assert not quits
+    assert all(p.read_bytes() == raw for p, raw in before.items())
+    if status != "invalid":
+        store.update_turn(sid, turn["turn_id"], status="completed")
+        with pytest.raises(OwnedQuitReached):
+            host.switch("b", "fixture", reopen=True, restart_clients=True)
+        assert quits == ["quit"]
+
+
+@pytest.mark.parametrize("handles,ignored", [
+    ("f12\nar\ntDIR\n", True),
+    ("f*12\nar\ntDIR\nf13\nar\ntDIR\n", True),
+    ("f12\nar\ntREG\n", False),  # Read-only auth inode is still a hold.
+    ("f12\naw\ntDIR\n", False),
+    ("fcwd\nar\ntDIR\n", False),
+    ("f12\nar\n", False),
+    ("f12\nar\ntDIR\nf13\nar\ntREG\n", False),
+])
+def test_only_proven_readonly_directory_handles_are_not_auth_consumers(native, monkeypatch, tmp_path, handles, ignored):
+    module = importlib.import_module("loopx_codex_provider_routing.native_host")
+    host = module.NativeHost.__new__(module.NativeHost)
+    host.spec = {"home": tmp_path}
+    identity = (1, "/test/filesystem-server", 10, 20)
+    host._processes = lambda: {200: identity}
+    host._cached_codex_owners = lambda _: set()
+    monkeypatch.setattr(module, "process_identity", lambda _: identity)
+    monkeypatch.setattr(module.subprocess, "run", lambda *_a, **_k: SimpleNamespace(returncode=0, stdout="p200\n" + handles))
+    assert host._home_owners() == ([] if ignored else [200])
+    host._cached_codex_owners = lambda _: {200}
+    assert host._home_owners() == [200]  # Closed FDs do not clear a cached CLI.
+
+
 @pytest.mark.parametrize("force", [False, True])
 @pytest.mark.parametrize("code", ["client_changed", "native_unavailable"])
 def test_failed_handoffs_exhaust_a_durable_retry_budget(native, tmp_path, force, code):

From 0389d697282dcb00d49206f2f60888c7d326f295 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Thu, 29 Oct 2026 13:40:56 +0800
Subject: [PATCH 2/2] fix(codex): retain loaded Core owner in native workers

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../loopx-codex-provider-routing/NATIVE.md    |  5 +++
 .../loopx_codex_provider_routing/native.py    | 15 ++++++-
 tests/presentation/test_codex_accounts.py     | 43 +++++++++++++++++++
 3 files changed, 61 insertions(+), 2 deletions(-)

diff --git a/packages/loopx-codex-provider-routing/NATIVE.md b/packages/loopx-codex-provider-routing/NATIVE.md
index b291cafded..752e2ee537 100644
--- a/packages/loopx-codex-provider-routing/NATIVE.md
+++ b/packages/loopx-codex-provider-routing/NATIVE.md
@@ -125,6 +125,9 @@ unreadable state defer the entire handoff before App quit. A new/changed service
 cannot inherit the frozen restart set. The lock remains held through account
 readback and restoration. A standalone observer must run in the same upgraded
 LoopX environment to restart these services; missing Core support waits safely.
+An already loaded archive/bootstrap Core passes its import root to the native
+worker, including detached handoffs; the worker does not rely on its current
+directory to rediscover that owner.
 Prefer the serving App/Chat observer so normal LoopX updates also refresh it;
 retire any separately installed legacy watcher through its original manager.
 
@@ -134,6 +137,8 @@ retire any separately installed legacy watcher through its original manager.
 原账号 owner 在事务锁内复查只读例外，并再次核验句柄；不重启 VM，也不据此声称
 已检查 guest 内的认证缓存。停止 Chat 前复用原在途任务读取与 ingress 锁，排队、
 completing 或不可读状态均在退出 App 前等待；恢复期间保持锁与原失败/未知回执。
+已加载的 archive/bootstrap Core 向原 native 子进程传递其加载路径，包含脱离父进程的
+换号任务，避免依赖工作目录或误用旧安装；没有加载 Core 的独立入口仍安全等待。
 观察器应随实际 LoopX 服务升级；独立旧 watcher 通过原管理器退役，避免占住唯一观察权。
 
 Each service's restoration intent is saved privately before unloading. Version
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 b1e2df1e6f..681ed36fc0 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
@@ -69,13 +69,23 @@ def policy(request: dict) -> dict:
 def host(adapter: dict, operation: str, **parameters) -> dict:
     receipt = parameters.pop("handoff_receipt", None)
     command = [sys.executable, "-m", "loopx_codex_provider_routing.native_host"]
+    environment = None
+    core = sys.modules.get("loopx")
+    if core is not None and getattr(core, "__file__", None):
+        # An archive/private bootstrap can load Core through sys.path alone.
+        # Its detached child must use that same loaded owner, not an older
+        # installation (or no Core) discovered from the child's working dir.
+        environment = os.environ.copy()
+        root = str(Path(core.__file__).resolve().parent.parent)
+        inherited = environment.get("PYTHONPATH")
+        environment["PYTHONPATH"] = root + (os.pathsep + inherited if inherited else "")
     request = json.dumps({"adapter": adapter, "operation": operation, **parameters})
     if receipt:
         # Closing the calling native App must not close this continuation or
         # its result channel. All native sub-operations have bounded waits.
         process = subprocess.Popen([*command, "--result-file", receipt], stdin=subprocess.PIPE,
                                    stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
-                                   text=True, start_new_session=True)
+                                   text=True, start_new_session=True, env=environment)
         try:
             process.communicate(request, timeout=150)
         except subprocess.TimeoutExpired:
@@ -87,7 +97,8 @@ def host(adapter: dict, operation: str, **parameters) -> dict:
         if packet.get("ok") is not True:
             raise HostError(packet.get("code", "native_unavailable"))
         return packet["result"]
-    result = subprocess.run(command, input=request, capture_output=True, text=True, timeout=150)
+    result = subprocess.run(command, input=request, capture_output=True, text=True, timeout=150,
+                            env=environment)
     if len(result.stdout) > 131072:
         raise HostError("native_unavailable")
     try:
diff --git a/tests/presentation/test_codex_accounts.py b/tests/presentation/test_codex_accounts.py
index e42c175b03..411dcf1ae1 100644
--- a/tests/presentation/test_codex_accounts.py
+++ b/tests/presentation/test_codex_accounts.py
@@ -176,6 +176,49 @@ def configured(native, tmp_path):
     return provider, transport, clock
 
 
+@pytest.mark.parametrize("detached", [False, True])
+def test_native_worker_inherits_loaded_core_outside_checkout(native, monkeypatch, tmp_path, detached):
+    import loopx
+    expected = str(Path(loopx.__file__).resolve())
+    monkeypatch.delenv("PYTHONPATH", raising=False)
+    # Probe real child import provenance through host's two launch paths.
+    # -S removes editable-install/site-package help, as in an archive bootstrap.
+    # The child substitutes only the account protocol result; no auth is read.
+    probe = '''
+import json, os, sys
+from pathlib import Path
+import loopx
+from loopx.runtime_activation import managed_chat_activation_guard
+packet = {"ok": True, "result": {"core": str(Path(loopx.__file__).resolve()),
+                               "guard": callable(managed_chat_activation_guard)}}
+if len(sys.argv) == 3 and sys.argv[1] == "--result-file":
+    packet["schema_version"] = "codex_native_accounts_v1"
+    fd = os.open(sys.argv[2], os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+    with os.fdopen(fd, "w") as stream:
+        json.dump(packet, stream)
+else:
+    print(json.dumps(packet))
+'''
+    launch = native.subprocess.Popen if detached else native.subprocess.run
+    def child(command, **kwargs):
+        return launch([command[0], "-S", "-c", probe, *command[3:]], cwd=tmp_path, **kwargs)
+    monkeypatch.setattr(native.subprocess, "Popen" if detached else "run", child)
+    result = native.host({}, "status", **({"handoff_receipt": str(tmp_path / "reply.json")} if detached else {}))
+    assert result == {"core": expected, "guard": True}
+    assert "PYTHONPATH" not in os.environ  # No process-wide environment mutation.
+
+
+def test_native_worker_does_not_invent_a_core_owner(native, monkeypatch):
+    monkeypatch.delitem(sys.modules, "loopx", raising=False)
+    captured = []
+    def child(_command, **kwargs):
+        captured.append(kwargs["env"])
+        return SimpleNamespace(stdout='{"ok": true, "result": {}}')
+    monkeypatch.setattr(native.subprocess, "run", child)
+    assert native.host({}, "status") == {}
+    assert captured == [None]
+
+
 def test_default_off_has_no_native_effect_or_startup_observer(native, tmp_path):
     from loopx.presentation.codex_accounts_api import CodexAccountService