diff --git a/loopx/chat_server.py b/loopx/chat_server.py index 4687d7e606..8a8a04e975 100644 --- a/loopx/chat_server.py +++ b/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 diff --git a/packages/loopx-codex-provider-routing/NATIVE.md b/packages/loopx-codex-provider-routing/NATIVE.md index 1222f137c0..55c7d3819b 100644 --- a/packages/loopx-codex-provider-routing/NATIVE.md +++ b/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) diff --git a/packages/loopx-codex-provider-routing/docs/assets/native-policy-owner-desktop.png b/packages/loopx-codex-provider-routing/docs/assets/native-policy-owner-desktop.png new file mode 100644 index 0000000000..9dfa4a9b8b Binary files /dev/null and b/packages/loopx-codex-provider-routing/docs/assets/native-policy-owner-desktop.png differ diff --git a/packages/loopx-codex-provider-routing/docs/assets/native-policy-owner-narrow.png b/packages/loopx-codex-provider-routing/docs/assets/native-policy-owner-narrow.png new file mode 100644 index 0000000000..9cb1ef3ece Binary files /dev/null and b/packages/loopx-codex-provider-routing/docs/assets/native-policy-owner-narrow.png differ diff --git a/packages/loopx-codex-provider-routing/docs/assets/native-policy-owner-unknown-narrow.png b/packages/loopx-codex-provider-routing/docs/assets/native-policy-owner-unknown-narrow.png new file mode 100644 index 0000000000..b2d35c0bec Binary files /dev/null and b/packages/loopx-codex-provider-routing/docs/assets/native-policy-owner-unknown-narrow.png differ diff --git a/packages/loopx-codex-provider-routing/extension.toml b/packages/loopx-codex-provider-routing/extension.toml index b6b15a6a56..c7110dbe1c 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.7" +version = "0.12.8" 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 ccf795dabf..224fe98548 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.7" +version = "0.12.8" 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/test_native_observer.py b/packages/loopx-codex-provider-routing/smoke/test_native_observer.py new file mode 100644 index 0000000000..ad5895e691 --- /dev/null +++ b/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() 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 4fe34bb21c..c08e33c9d0 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.3" +__version__ = "0.12.8" 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 0281efa98f..26bdc643c5 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 @@ -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: diff --git a/tests/test_chat_transport_composition.py b/tests/test_chat_transport_composition.py index e911b46d95..2c1f2512d1 100644 --- a/tests/test_chat_transport_composition.py +++ b/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