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