fix(native-accounts): discover and restart auth consumers by ownership

packages/loopx-codex-provider-routing/NATIVE.md

@@ -59,21 +59,47 @@ must still match. The launch probe must explicitly contain `launch_time`: a
 positive finite timestamp retains the launch-date fence; explicit `null` from
 Launch Services uses the inspected PID and executable instead. Missing fields,
 invalid types and non-finite timestamps cannot authorize force quit. A successful
-normal quit never escalates. Other CLI/file owners remain holds. Force quit can interrupt
+normal quit never escalates. Unmanaged CLI/file owners remain holds. Force quit can interrupt
 work in that instance; it does not authorize replay of submitted jobs.
 Version 0.12.4 also records the inspected App's descendants before quitting.
 Only those helpers, with unchanged kernel executable and microsecond start
 identity, can be retired after the App exits. An older orphan with no observed
 parentage, a standalone CLI using the same bundled binary, or a later App launch
 remains a hold. Bundle paths alone never authorize termination.
-With client restart enabled, the known LoopX Chat launch agent can also be
-paused when its observed process tree actually holds this home. Its restoration
-intent is saved privately before unloading, and restoration is checked even
-after a partial failure. A failed restore is reported as `chat_resume_failed`;
+With client restart enabled, version 0.12.6 discovers all loaded current-user
+LaunchAgents whose observed Codex process trees consume this home. Service names are
+not an allowlist: renamed Chat services, additional providers and plists with
+different filenames participate through their existing launchd manager. A
+trusted plist, its loaded path, PID, executable and kernel start identity must
+match. An unrelated home, ambiguous definition or unverifiable owner cannot
+authorize a restart. Discovery combines open home files with the private launch
+environment of Codex processes, covering warm app-servers that cached auth and
+closed their files. Arguments and environment values are never projected or logged.
+
+Each service's restoration intent is saved privately before unloading. Observed
+descendants are retired only while their kernel identity still matches, even if
+they have since closed every auth file. The provider waits for their exit and
+rechecks native idle under the existing account transaction locks. Up to three
+rescans cover services that appear during shutdown. With force permission,
+helpers still alive after normal termination can be killed against those same
+frozen identities. No arbitrary CLI is relaunched and no submitted job is replayed.
+
+Restoration is checked even after a partial failure. A loaded job alone is not
+success: a previously running service must have a fresh live process. A failed
+restore is reported as `chat_resume_failed`;
 the next native request reconciles that intent without repeating the account
 transaction. A changed service definition is preserved and requires local
 repair rather than bootstrapping unchecked bytes. Installation does not enable
 client restart or start a new Chat service.
+
+`loopx-codex-accounts clients` performs a read-only live inventory with PIDs,
+App/service/unmanaged classification and restartability. This inventory also
+leaves outstanding service recovery to the existing status/switch owner. It never reads token
+contents or changes selection. Unknown file owners (including filesystem/VM
+services without a proven restart owner) remain visible holds; installation
+does not turn them into termination targets. Discovery is best effort within
+the current user's kernel visibility, not a claim to inspect auth cached inside
+another user's process or a virtual machine.
 An operator-owned service independent of the App can run the same observer:
 
 ```sh

packages/loopx-codex-provider-routing/extension.toml

@@ -1,6 +1,6 @@
 schema_version = "loopx_extension_manifest_v0"
 id = "loopx-codex-provider-routing"
-version = "0.12.4"
+version = "0.12.6"
 requires_loopx_api = ">=1,<2"
 permissions = []
 

packages/loopx-codex-provider-routing/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "loopx-codex-provider-routing"
-version = "0.12.4"
+version = "0.12.6"
 description = "Native Codex account slots and exhausted-account handoff for LoopX"
 readme = "README.md"
 requires-python = ">=3.11"

packages/loopx-codex-provider-routing/smoke/native_host_clients_smoke.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import importlib
+import itertools
 import json
 import os
 from pathlib import Path
@@ -15,7 +16,8 @@
 PACKAGE_ROOT = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(PACKAGE_ROOT / "src"))
 host_module = importlib.import_module("loopx_codex_provider_routing.native_host")
-from loopx_codex_provider_routing.native import private_read
+from loopx_codex_provider_routing.native import private_read  # noqa: E402
+from loopx_codex_provider_routing.native_processes import _codex_home, codex_home  # noqa: E402
 
 BUNDLE = "/Applications/Example.app"
 EXE = BUNDLE + "/Contents/MacOS/Example"
@@ -31,9 +33,10 @@ def __init__(self, home):
         self.loaded, self.pause_error, self.resume_error = True, None, None
         self.records = {100: (1, EXE, 1000, 1), 101: (100, HELPER, 1001, 2),
                         200: (1, HELPER, 999, 3), 300: (1, "/test/chat", 900, 4),
-                        301: (300, "/test/app-server", 901, 5)}
+                        301: (300, "/test/codex", 901, 5)}
         self.records[os.getpid()] = (1, "/test/python", 800, 8)
         self.plist = home / "Library/LaunchAgents" / (LABEL + ".plist")
+        self.service_pid = 300
         self.plist.parent.mkdir(parents=True)
         self.plist.write_bytes(plistlib.dumps({"Label": LABEL, "ProgramArguments": ["/test/chat"]}))
         self.plist.chmod(0o600)
@@ -57,13 +60,14 @@ def close(self, _expected, **kwargs):
         assert kwargs["inspected"] == (100, self.records[100])
         self.app_pid = None
         self.records[101] = (1, *self.records[101][1:])
+        del self.records[100]
         return True
 
     def open_app(self, *_):
         self.app_pid = 400
 
     def idle(self):
-        if self.holders:
+        if self.host._home_owners():
             raise host_module.HostError("waiting_for_idle")
 
     def switch(self, _slot):
@@ -80,10 +84,13 @@ def run(self, command, **_kwargs):
             return SimpleNamespace(returncode=0, stdout="".join(f"p{p}\n" for p in self.holders))
         verb = command[1]
         if verb == "print":
-            return SimpleNamespace(returncode=0 if self.loaded else 113, stdout="pid = 300\n" if self.loaded else "")
+            return SimpleNamespace(returncode=0 if self.loaded else 113,
+                                   stdout=f"path = {self.plist}\npid = {self.service_pid}\n" if self.loaded else "")
         if verb == "bootout":
             self.loaded = False
             self.holders = [p for p in self.holders if p != 301]
+            self.records.pop(self.service_pid, None)
+            self.records.pop(301, None)
             if self.pause_error:
                 raise self.pause_error
         if verb == "bootstrap":
@@ -92,6 +99,8 @@ def run(self, command, **_kwargs):
             if self.resume_error == "timeout_no_effect":
                 raise subprocess.TimeoutExpired(command, 20)
             self.loaded = True
+            self.service_pid = 600
+            self.records[600] = (1, "/test/chat", 1400, 12)
             if self.resume_error == "timeout_after_effect":
                 raise subprocess.TimeoutExpired(command, 20)
         return SimpleNamespace(returncode=0, stdout="")
@@ -102,13 +111,16 @@ def kill(self, pid, _signal):
                 raise ProcessLookupError()
             return
         self.killed.append(pid)
-        self.holders.remove(pid)
+        if pid in self.holders:
+            self.holders.remove(pid)
+        self.records.pop(pid, None)
 
     def context(self):
         from contextlib import ExitStack
         stack = ExitStack()
         stack.enter_context(patch.object(host_module.sys, "platform", "darwin"))
         stack.enter_context(patch.object(host_module, "process_identity", side_effect=lambda p: self.records.get(p)))
+        stack.enter_context(patch.object(host_module, "codex_home", return_value=None))
         stack.enter_context(patch.object(host_module.subprocess, "run", side_effect=self.run))
         stack.enter_context(patch.object(host_module.os, "kill", side_effect=self.kill))
         return stack
@@ -135,6 +147,134 @@ def fixture():
         home.mkdir(mode=0o700)
         return Fixture(home)
 
+    # Service labels are installation details. A renamed same-home consumer
+    # must be stopped and restored through its existing launch agent, including
+    # when the plist filename differs from its launchd label.
+    f = fixture()
+    f.holders = [301]
+    renamed = "org.example.codex.transport"
+    f.plist.rename(f.plist.with_name("transport-installation.plist"))
+    f.plist = f.plist.with_name("transport-installation.plist")
+    f.plist.write_bytes(plistlib.dumps({"Label": renamed, "ProgramArguments": ["/test/chat"]}))
+    with f.context():
+        assert f.handoff()["active_slot"] == "b"
+    assert any(c[1:3] == ["bootout", f"gui/{os.getuid()}/{renamed}"] for c in f.calls)
+    assert f.loaded and f.switches == 1
+
+    # A second independent service participates in the same transaction. Neither
+    # label is privileged; both must be restored, without a second account swap.
+    f = fixture()
+    f.holders = [101, 301, 701]
+    second = f.plist.with_name("another-provider.plist")
+    second.write_bytes(plistlib.dumps({"Label": "org.example.second", "ProgramArguments": ["/test/other"]}))
+    f.records.update({700: (1, "/test/other", 1500, 14), 701: (700, "/test/codex", 1501, 15)})
+    second_state = {"loaded": True, "pid": 700}
+    primary_run = f.run
+    def multi_run(command, **kwargs):
+        if (command[0] == "/bin/launchctl" and
+                (command[-1].endswith("/org.example.second") or command[-1] == str(second))):
+            f.calls.append(command)
+            if command[1] == "print":
+                return SimpleNamespace(returncode=0 if second_state["loaded"] else 113,
+                                       stdout=f"path = {second}\npid = {second_state['pid']}\n")
+            if command[1] == "bootout":
+                second_state["loaded"] = False
+                f.records.pop(700)
+                f.records.pop(701)
+                f.holders.remove(701)
+            elif command[1] == "bootstrap":
+                second_state.update(loaded=True, pid=800)
+                f.records[800] = (1, "/test/other", 1600, 16)
+            return SimpleNamespace(returncode=0, stdout="")
+        return primary_run(command, **kwargs)
+    f.run = multi_run
+    with f.context():
+        assert f.handoff()["active_slot"] == "b"
+    assert f.loaded and second_state["loaded"] and f.switches == 1
+    assert len([c for c in f.calls if c[1] == "bootout"]) == 2
+
+    # The app-server has cached auth and closed every file. Its launch
+    # environment still proves it consumes this home, and its service restarts.
+    f = fixture()
+    f.holders = []
+    f.records[301] = (300, "/test/codex", 901, 5)
+    with f.context(), patch.object(host_module, "codex_home", side_effect=lambda p: str(f.home) if p == 301 else None):
+        inventory = f.host.clients()
+        assert inventory["complete"] and len(inventory["clients"]) == 1
+        assert inventory["clients"][0]["service"] == LABEL and inventory["clients"][0]["restartable"]
+        assert f.handoff()["active_slot"] == "b"
+    assert f.loaded and f.switches == 1
+
+    # The same warm CLI without a recoverable manager remains an explicit hold;
+    # a different home remains untouched. Neither case grants broad pkill.
+    for same_home in (False, True):
+        f = fixture()
+        f.holders = []
+        f.records[200] = (1, "/test/codex", 999, 3)
+        clock = iter(range(0, 1000, 20))
+        with f.context(), patch.object(host_module, "codex_home", return_value=str(f.home if same_home else f.home / "other")), \
+                patch.object(host_module.time, "monotonic", side_effect=lambda: next(clock)):
+            if same_home:
+                error("other_clients_busy", f.handoff)
+            else:
+                assert f.handoff()["active_slot"] == "b"
+        assert 200 not in f.killed and f.switches == (0 if same_home else 1)
+
+    # Merely opening a directory is not proof that a service consumes auth.
+    f = fixture()
+    f.holders = [301]
+    f.records[301] = (300, "/test/indexer", 901, 5)
+    with f.context():
+        f.host.pause_chat_agents()
+    assert not any(c[1] == "bootout" for c in f.calls)
+
+    # Closing FDs does not prove that a captured helper has exited. Normal
+    # termination is tried first; only separate force permission can escalate.
+    for force in (False, True):
+        f = fixture()
+        f.holders = [101]
+        signals = []
+        original_kill = f.kill
+        def delayed_kill(pid, sig):
+            signals.append((pid, sig))
+            if pid == 101 and sig == host_module.signal.SIGTERM:
+                if pid in f.holders:
+                    f.holders.remove(pid)
+                return  # The still-live helper could retain auth in memory.
+            return original_kill(pid, sig)
+        f.kill = delayed_kill
+        clock = itertools.count()
+        with f.context(), patch.object(host_module.time, "monotonic", side_effect=lambda: next(clock)), \
+                patch.object(host_module.time, "sleep"):
+            if force:
+                assert f.handoff(force_quit_clients=True)["active_slot"] == "b"
+            else:
+                error("other_clients_busy", f.handoff)
+        assert ((101, host_module.signal.SIGKILL) in signals) is force
+        assert f.switches == int(force)
+
+    # Changed loaded definitions and reused service PIDs cannot authorize stop.
+    for mutation in ("path", "pid"):
+        f = fixture()
+        f.holders = [301]
+        original_run, prints = f.run, [0]
+        def changed_run(command, **kwargs):
+            result = original_run(command, **kwargs)
+            if command[0] == "/bin/launchctl" and command[1] == "print":
+                prints[0] += 1
+                if mutation == "path":
+                    result.stdout = result.stdout.replace(str(f.plist), "/different/agent.plist")
+                elif prints[0] == 2:
+                    f.records[300] = (1, "/test/chat", 2000, 20)
+            return result
+        f.run = changed_run
+        with f.context():
+            if mutation == "path":
+                f.host.pause_chat_agents()
+            else:
+                error("client_changed", f.host.pause_chat_agents)
+        assert not any(c[1] == "bootout" for c in f.calls) and f.switches == 0
+
     # The same bundled binary does not establish launch ownership. Only the
     # captured descendant is eligible after it becomes orphaned.
     f = fixture()
@@ -197,7 +337,7 @@ def fixture():
     f.holders = []
     with f.context():
         f.host.switch("b", "before", reopen=False, restart_clients=False)
-    assert not f.killed and not f.calls
+    assert not f.killed and not any(c[0] == "/bin/launchctl" and c[1] != "print" for c in f.calls)
     assert not (f.root / "chat-recovery.json").exists()
 
     # A product Chat label attached to another home is not paused.
@@ -293,5 +433,29 @@ def fixture():
 if sys.platform == "darwin":
     identity = host_module.process_identity(os.getpid())
     assert identity and identity[2] > 0 and Path(identity[1]).is_absolute(), identity
+    # Real kernel environment IO, scoped to our disposable child. Secrets and
+    # argv remain private even when an argument resembles an environment entry.
+    child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)", "CODEX_HOME=/decoy"],
+                             env={**os.environ, "CODEX_HOME": "/example/native-home"},
+                             stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+    try:
+        assert codex_home(child.pid) == "/example/native-home"
+    finally:
+        child.terminate()
+        child.wait(timeout=5)
+
+# An argv lookalike cannot supply CODEX_HOME; empty argv entries do not shift
+# the environment boundary. Duplicate environment keys are unverifiable.
+def process_bytes(arguments, environment):
+    return (len(arguments).to_bytes(4, sys.byteorder, signed=True) + b"/test/codex\0\0"
+            + b"\0".join(arguments) + b"\0" + b"\0".join(environment) + b"\0")
+assert _codex_home(process_bytes([b"codex", b"", b"CODEX_HOME=/decoy"], [b"CODEX_HOME=/real"])) == "/real"
+assert _codex_home(process_bytes([b"codex", b"CODEX_HOME=/decoy"], [b"OTHER=value"])) is None
+try:
+    _codex_home(process_bytes([b"codex"], [b"CODEX_HOME=/one", b"CODEX_HOME=/two"]))
+except OSError:
+    pass
+else:
+    raise AssertionError("ambiguous environment accepted")
 
 print("native host client handoff smoke passed (identity, scope, switch and recovery)")

packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native.py

@@ -162,6 +162,10 @@ def status(self) -> dict:
         native = self.transport(adapter, "status")
         return self._projection(state, native)
 
+    def clients(self) -> dict:
+        state = self._load()
+        return self.transport(state.get("adapter") or discover(), "clients")
+
     def configure(self, *, expected_revision: int, enabled: bool, slots: list[str], reconnect: bool = False,
                   restart_clients: bool | None = None, force_quit_clients: bool | None = None) -> dict:
         if type(enabled) is not bool or not isinstance(slots, list) or len(slots) > 16 or len(slots) != len(set(slots)):
@@ -414,7 +418,7 @@ def close(self):
 
 def main() -> int:
     parser = argparse.ArgumentParser(prog="loopx-codex-accounts")
-    parser.add_argument("operation", choices=("status", "configure", "refresh", "switch", "recover", "rollback", "reopen", "watch"))
+    parser.add_argument("operation", choices=("status", "clients", "configure", "refresh", "switch", "recover", "rollback", "reopen", "watch"))
     parser.add_argument("--runtime-root", type=Path, default=Path.home() / ".loopx")
     args = parser.parse_args()
     provider = NativeAccounts(args.runtime_root)
@@ -426,9 +430,11 @@ def main() -> int:
             except KeyboardInterrupt:
                 watcher.close()
             return 0
-        request = {} if args.operation in ("status", "refresh") else json.load(sys.stdin)
+        request = {} if args.operation in ("status", "clients", "refresh") else json.load(sys.stdin)
         if args.operation == "status":
             result = provider.status()
+        elif args.operation == "clients":
+            result = provider.clients()
         elif args.operation == "configure":
             result = provider.configure(**request)
         elif args.operation in ("refresh", "switch"):