fix(accounts): share native policy and elect one observer

loopx/chat_server.py

@@ -1830,7 +1830,13 @@ def serve_chat(
     server = ChatHTTPServer((host, port), ChatRequestHandler)
     server.registry_path = resolved_registry_path
     server.runtime_root = runtime_root
-    server.codex_account_service = CodexAccountService(runtime_root)
+    # Account policy follows the registry's native owner. A separate Chat
+    # history directory must not silently create another account policy.
+    account_runtime_root = (
+        resolve_runtime_root(registry, registry_path=resolved_registry_path)
+        if registry.get("common_runtime_root") else runtime_root
+    )
+    server.codex_account_service = CodexAccountService(account_runtime_root)
     server.runtime_root_override = resolved_runtime_root_override
     server.scan_roots = resolved_scan_roots
     server.limit = limit

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

@@ -48,6 +48,24 @@ TypeScript selector. Enrol accounts through the native owner's official login.
 Click **Switch automatically when quota runs out** to authorize the listed slots
 as destinations. Installation alone grants no account effect. The App observer
 runs with its serving LoopX process and rehydrates the saved opt-in on restart.
+Version 0.12.8 uses the registry's declared runtime owner for this policy even when Chat
+history uses a separate `--runtime-root`. The settings page and the native CLI
+therefore read and update the same policy; no history or credential data is
+copied. A registry without `common_runtime_root` retains the explicit Chat
+runtime for its policy, so an isolated setup cannot fall back to the user's
+global policy. Explicit separate registry/runtime owners remain isolated. For a CLI
+watch, pass that registry owner's runtime to `--runtime-root` as usual.
+
+An App observer and CLI `watch` sharing that policy elect one active observer
+with a private process lock. Contenders perform no native account observations
+or switches; they wait up to five seconds between ownership checks and can take
+over when the active owner exits. Stopping a contender releases its thread
+without stopping the active owner. The existing transaction lock still serializes
+configuration, explicit checks and switches. Turning the policy off affects the
+actual observer, while an already dispatched credential transaction retains its
+normal completion/recovery contract. Upgrade and restart all existing observers
+before relying on this ownership fence; an older process does not acquire it.
+
 Client restart is a separate permission, off by default and absent from older
 policies. Enabling it also permits a subsequent explicit manual switch to quit
 and reopen the GPT instance. Neither permission is enabled by installation.
@@ -290,3 +308,33 @@ exercised, followed by a fresh successful check without changing the account.
 ![Desktop force-quit settings / 桌面强关设置](docs/assets/native-accounts-force-desktop.png)
 
 ![Narrow settings / 窄屏设置](docs/assets/native-accounts-force-narrow.png)
+
+### Shared policy owner / 共用策略来源
+
+The settings page and CLI use the registry-declared native account owner even
+when Chat history has a separate runtime directory. Without a declared owner,
+an explicit isolated Chat runtime remains isolated. Version 0.12.8 also holds a
+process lock for the observer lifetime: another App or CLI stays passive until
+the owner exits. Upgrade and restart every existing observer before relying on
+this guarantee; an older observer does not participate in the new lock.
+
+设置页与 CLI 共用 registry 声明的原生账号策略来源，会话历史仍保留在原目录。
+未声明公共来源时，显式隔离的 Chat runtime 保持隔离。0.12.8 通过跨进程锁保证同一
+策略来源只有一个自动观察器；原观察器退出后，等待者接管。升级时需要重启所有旧
+观察器，旧版本不会自动遵守新锁。
+
+The actual packaged page and HTTP service below use synthetic A/B/C accounts.
+Turning automatic switching off and back on persisted to the shared owner and
+survived reload without changing the separate conversation policy. Unknown
+allowance kept account A selected; a fresh successful check recovered. These
+images do not qualify a real credential switch.
+
+以下为实际打包网页与 HTTP 服务，账号 A/B/C 为合成数据。关闭、重新开启和刷新后
+均读回共用策略，会话目录中的独立旧策略未被改写。额度未知时保留 A，重新检查后
+恢复；不代表已完成真实账号切换。
+
+![Shared account policy, desktop / 桌面](docs/assets/native-policy-owner-desktop.png)
+
+![Shared account policy, narrow / 窄屏](docs/assets/native-policy-owner-narrow.png)
+
+![Unknown allowance preserves selection / 额度未知保留当前账号](docs/assets/native-policy-owner-unknown-narrow.png)

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.7"
+version = "0.12.8"
 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.7"
+version = "0.12.8"
 description = "Native Codex account slots and exhausted-account handoff for LoopX"
 readme = "README.md"
 requires-python = ">=3.11"

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

@@ -0,0 +1,145 @@
+"""Real process ownership; synthetic allowance never accesses native credentials."""
+from __future__ import annotations
+
+import importlib
+import os
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+
+SOURCE = Path(__file__).resolve().parents[1] / "src"
+sys.path.insert(0, str(SOURCE))
+native = importlib.import_module("loopx_codex_provider_routing.native")
+
+CHILD = r'''
+import json, os, sys, time
+from pathlib import Path
+from loopx_codex_provider_routing.native import NativeAccounts, Watcher
+root = Path(sys.argv[1])
+def transport(adapter, operation, **parameters):
+    if operation == "status":
+        return {"slots": ["a", "b"], "active_slot": "a", "selection_revision": "fixture",
+                "current_login_matches": True, "recovery_required": False}
+    if operation == "observe":
+        with (root / "observations").open("a") as stream:
+            stream.write(str(os.getpid()) + "\n")
+        return {"observations": [{"slot": s, "status": "ok", "observed_at": time.time(),
+                "usage": {"rate_limit": {"primary_window": {"used_percent": 0,
+                          "reset_at": time.time() + 3600}}, "credits": {"has_credits": False}}}
+                for s in parameters["slots"]]}
+    raise AssertionError("No native account effect is authorized: " + operation)
+provider = NativeAccounts(root, transport=transport)
+tick = provider.tick
+def observed_tick():
+    with (root / "ticks").open("a") as stream:
+        stream.write(str(os.getpid()) + "\n")
+    value = tick()
+    value["retry_at"] = time.time() + 5  # Bound the fixture's scheduling wait only.
+    return value
+provider.tick = observed_tick
+Watcher(provider).run()
+'''
+
+
+@unittest.skipUnless(os.name == "posix", "Native account ownership uses POSIX locks")
+class NativeObserverOwnership(unittest.TestCase):
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory(prefix="native-observer-")
+        self.root = Path(self.temporary.name).resolve()
+        self.children = []
+
+    def tearDown(self):
+        for process in self.children:
+            if process.poll() is None:
+                process.terminate()
+            try:
+                process.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                process.kill()
+                process.wait(timeout=5)
+        self.temporary.cleanup()
+
+    def policy(self, name):
+        root = self.root / name
+        provider = native.NativeAccounts(root)
+        provider.root.mkdir(parents=True, mode=0o700)
+        native.atomic(provider.path, {
+            "schema_version": native.SCHEMA, "revision": 1,
+            "policy": {"enabled": True, "slots": ["a", "b"], "cooldown_seconds": 300,
+                       "restart_clients": False, "force_quit_clients": False},
+            "adapter": {"synthetic": True}, "observations": [], "last_switch_at": None,
+            "action": "checking",
+        })
+        return root, provider
+
+    def start(self, root):
+        env = {**os.environ, "PYTHONPATH": str(SOURCE)}
+        process = subprocess.Popen([sys.executable, "-c", CHILD, str(root)], env=env,
+                                   stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
+                                   stderr=subprocess.DEVNULL)
+        self.children.append(process)
+        return process
+
+    def ticks(self, root):
+        path = root / "ticks"
+        return path.read_text().splitlines() if path.exists() else []
+
+    def wait_for_ticks(self, root, count):
+        deadline = time.monotonic() + 9
+        while len(self.ticks(root)) < count and time.monotonic() < deadline:
+            time.sleep(0.025)
+        self.assertEqual(len(self.ticks(root)), count)
+
+    def test_only_one_process_observes_and_owner_exit_allows_takeover(self):
+        root, _ = self.policy("shared")
+        first = self.start(root)
+        self.wait_for_ticks(root, 1)
+        second = self.start(root)
+        time.sleep(0.4)
+        self.assertIsNone(first.poll())
+        self.assertIsNone(second.poll())
+        self.assertEqual(self.ticks(root), [str(first.pid)])
+        first.terminate()
+        first.wait(timeout=5)
+        self.wait_for_ticks(root, 2)
+        self.assertEqual(self.ticks(root), [str(first.pid), str(second.pid)])
+
+    def test_explicit_separate_policy_owners_do_not_block_each_other(self):
+        left, _ = self.policy("left")
+        right, _ = self.policy("right")
+        first, second = self.start(left), self.start(right)
+        self.wait_for_ticks(left, 1)
+        self.wait_for_ticks(right, 1)
+        self.assertEqual(self.ticks(left), [str(first.pid)])
+        self.assertEqual(self.ticks(right), [str(second.pid)])
+
+    def test_disable_from_another_process_stops_native_observation(self):
+        root, provider = self.policy("disable")
+        process = self.start(root)
+        deadline = time.monotonic() + 9
+        observations = root / "observations"
+        while not observations.exists() and time.monotonic() < deadline:
+            time.sleep(0.025)
+        self.assertEqual(observations.read_text().splitlines(), [str(process.pid)])
+        provider.configure(expected_revision=1, enabled=False, slots=["a", "b"])
+        self.wait_for_ticks(root, 2)
+        self.assertEqual(observations.read_text().splitlines(), [str(process.pid)])
+        self.assertFalse(provider._load()["policy"]["enabled"])
+
+    def test_untrusted_observer_lock_is_not_followed(self):
+        root, provider = self.policy("untrusted")
+        target = self.root / "unrelated"
+        target.write_text("untouched")
+        (provider.root / "observer.lock").symlink_to(target)
+        process = self.start(root)
+        time.sleep(0.4)
+        self.assertIsNone(process.poll())
+        self.assertEqual(self.ticks(root), [])
+        self.assertEqual(target.read_text(), "untouched")
+
+
+if __name__ == "__main__":
+    unittest.main()

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

@@ -1,3 +1,3 @@
 __all__ = ["__version__"]
 
-__version__ = "0.12.3"
+__version__ = "0.12.8"

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

@@ -115,7 +115,7 @@ def _load(self) -> dict:
                 "observations": [], "last_switch_at": None, "action": "disabled"}
 
     @contextmanager
-    def _lock(self, *, wait_seconds: float = 0):
+    def _lock(self, *, wait_seconds: float = 0, name: str = "lock"):
         if os.name != "posix":
             raise HostError("platform_unavailable")
         import fcntl
@@ -125,7 +125,7 @@ def _lock(self, *, wait_seconds: float = 0):
         self.root.mkdir(mode=0o700, parents=True, exist_ok=True)
         if self.root.stat().st_uid != os.getuid() or self.root.stat().st_mode & 0o077:
             raise HostError("state_untrusted")
-        path = self.root / "lock"
+        path = self.root / name
         if path.is_symlink():
             raise HostError("state_untrusted")
         with open(path, "a", opener=lambda p, flags: os.open(p, flags, 0o600)) as stream:
@@ -386,7 +386,7 @@ def reopen(self, *, expected_selection: str) -> dict:
 
 
 class Watcher:
-    """Owned by the serving App (or explicit CLI watch), never an import effect."""
+    """One observer per policy owner; App/CLI contenders stay passive."""
     def __init__(self, provider: NativeAccounts):
         self.provider = provider
         self.stop = threading.Event()
@@ -400,6 +400,20 @@ def start(self):
         return self
 
     def run(self):
+        while not self.stop.is_set():
+            try:
+                # Keep ownership across sleeps, not just one transaction. An
+                # App restart and an existing CLI watch can share policy without
+                # duplicating native observations. The kernel releases this
+                # lock on exit; a waiting owner can then take over safely.
+                with self.provider._lock(name="observer.lock"):
+                    self._observe()
+            except BlockingIOError:
+                self.stop.wait(5)
+            except (HostError, OSError, ValueError):
+                self.stop.wait(60)
+
+    def _observe(self):
         while not self.stop.is_set():
             delay = 60.0
             try:

tests/test_chat_transport_composition.py

@@ -219,6 +219,10 @@ def __init__(self, *args, **kwargs):
             assert json.load(response)["ok"] is True
         assert server.runtime_controller.startup_timeout_sec == 30.0
         assert server.runtime_controller.acp_startup_timeout_sec is None
+        if server.codex_account_service.provider is not None:
+            assert server.codex_account_service.provider.root == (
+                server.runtime_root / "extensions" / "codex-native-accounts"
+            )
         assert server.manager_return_service.args[2] is server.chat_store
         if installed:
             assert server.manager_return_service.args[3] is server.conversation_transports
@@ -236,3 +240,97 @@ def __init__(self, *args, **kwargs):
         worker.join(8)
     assert not worker.is_alive()
     assert external.calls == [("start", server), "close"] if installed else external.calls == []
+
+
+def test_account_settings_control_registry_owner_with_separate_conversation_storage(tmp_path, monkeypatch):
+    """The settings toggle must control the same policy the native CLI reads."""
+    import importlib
+    from pathlib import Path
+    from urllib.error import HTTPError
+    from urllib.request import Request
+
+    import loopx.chat_server as chat
+    from loopx.extensions.lark.cli_resolution import LarkCliResolution
+
+    monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[1] /
+                                  "packages/loopx-codex-provider-routing/src"))
+    native = importlib.import_module("loopx_codex_provider_routing.native")
+    monkeypatch.setattr(native, "discover", lambda: {"synthetic": True})
+    effects = []
+
+    def transport(adapter, operation, **parameters):
+        effects.append(operation)
+        if operation == "status":
+            return {"slots": ["a", "b"], "active_slot": "a", "selection_revision": "fixture",
+                    "current_login_matches": True, "recovery_required": False}
+        if operation == "observe":
+            import time
+            return {"observations": [{"slot": slot, "status": "ok", "observed_at": time.time(),
+                    "usage": {"rate_limit": {"primary_window": {"used_percent": 0,
+                              "reset_at": time.time() + 3600}}, "credits": {"has_credits": False}}}
+                    for slot in parameters["slots"]]}
+        raise AssertionError("No account effect is authorized in this fixture: " + operation)
+
+    provider_type = native.NativeAccounts
+    monkeypatch.setattr(native, "NativeAccounts", lambda root: provider_type(root, transport=transport))
+    owner_root, conversation_root = tmp_path / "owner", tmp_path / "conversations"
+    owner = provider_type(owner_root, transport=transport)
+    owner.configure(expected_revision=0, enabled=True, slots=["a", "b"], restart_clients=True)
+    separate = provider_type(conversation_root, transport=transport)
+    separate.configure(expected_revision=0, enabled=False, slots=[])
+    separate_before = separate.path.read_bytes()
+    registry = tmp_path / "registry.json"
+    registry.write_text(json.dumps({"goals": [], "common_runtime_root": str(owner_root)}))
+    assets = tmp_path / "assets"
+    assets.mkdir()
+    (assets / "index.html").write_text("fixture")
+    monkeypatch.setattr(chat, "build_lark_goal_topic_runtime_snapshot", lambda **kwargs: {})
+    monkeypatch.setattr(chat, "resolve_lark_cli_for_runtime", lambda **kwargs:
+                        LarkCliResolution(None, False, "missing", None, "lark_cli_not_installed"))
+    servers, ready, failures = [], threading.Event(), []
+
+    class Server(chat.ChatHTTPServer):
+        def __init__(self, *args, **kwargs):
+            super().__init__(*args, **kwargs)
+            servers.append(self)
+            ready.set()
+
+    monkeypatch.setattr(chat, "ChatHTTPServer", Server)
+
+    def serve():
+        try:
+            chat.serve_chat(registry_path=registry, runtime_root_override=conversation_root,
+                            port=0, assets_dir=assets, codex_bin="unavailable-fixture")
+        except BaseException as error:
+            failures.append(error)
+
+    worker = threading.Thread(target=serve, daemon=True)
+    worker.start()
+    try:
+        assert ready.wait(8)
+        server = servers[0]
+        url = f"http://127.0.0.1:{server.server_port}/api/chat/codex-accounts"
+        with urlopen(url, timeout=8) as response:
+            before = json.load(response)
+        assert before["policy"]["enabled"] is True
+        assert before["policy"]["slots"] == ["a", "b"]
+        assert server.chat_store.root == conversation_root / "chat"
+        request = {"operation": "configure", "parameters": {
+            "expected_revision": before["revision"], "enabled": False, "slots": ["a", "b"]}}
+        headers = {"Content-Type": "application/json"}
+        with urlopen(Request(url, data=json.dumps(request).encode(), headers=headers), timeout=8) as response:
+            disabled = json.load(response)
+        assert disabled["policy"]["enabled"] is False
+        assert owner.status()["policy"]["enabled"] is False
+        assert owner.status()["policy"]["restart_clients"] is True
+        with pytest.raises(HTTPError) as conflict:
+            urlopen(Request(url, data=json.dumps(request).encode(), headers=headers), timeout=8)
+        assert conflict.value.code == 409
+        assert json.load(conflict.value)["code"] == "configuration_changed"
+        assert separate.path.read_bytes() == separate_before
+        assert "switch" not in effects
+    finally:
+        if servers:
+            servers[0].shutdown()
+        worker.join(8)
+    assert not worker.is_alive() and not failures