Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 34 additions & 19 deletions loopx/runtime_activation.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@

from __future__ import annotations

from contextlib import ExitStack
from contextlib import ExitStack, contextmanager
import os
from pathlib import Path
import plistlib
Expand Down Expand Up @@ -93,9 +93,9 @@ def _managed_service_plists() -> list[Path]:
]


def _managed_service_pid(plist: Path) -> int | None:
def _managed_service_pid(plist: Path, *, label: str | None = None) -> int | None:
result = subprocess.run(
["launchctl", "print", f"gui/{os.getuid()}/{plist.stem}"],
["launchctl", "print", f"gui/{os.getuid()}/{label or plist.stem}"],
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=5,
)
if result.returncode != 0:
Expand All @@ -113,6 +113,32 @@ def _managed_service_pid(plist: Path) -> int | None:
return int(pids[0]) if len(pids) == 1 else None


@contextmanager
def managed_chat_activation_guard(
plists: list[Path], *, expected_services: dict[Path, tuple[str, int]] | None = None,
):
"""Keep accepted Chat work and new ingress safe through a client restart.

Account-provider handoffs and runtime updates share the same store owner;
an empty Session summary is not sufficient evidence of an idle service.
The caller must still verify its loaded service/PID before stopping it.
"""
from .chat_ingress import chat_runtime_activation_lock, chat_runtime_activation_state

labels = [plist.stem for plist in plists]
roots = sorted({_chat_runtime_root(plist) for plist in plists})
with ExitStack() as locks:
for root in roots:
locks.enter_context(chat_runtime_activation_lock(root, timeout_seconds=2.0))
for root in roots:
if chat_runtime_activation_state(root)["nonterminal_turns"]:
raise RuntimeActivationDeferred("chat_work_pending", labels)
for plist, (label, pid) in (expected_services or {}).items():
if plist not in plists or _managed_service_pid(plist, label=label) != pid:
raise ValueError("chat_service_route_unresolved")
yield


def replace_managed_loopx_chat(host: str, port: int, *, timeout_seconds: float = 5.0) -> str:
"""Replace the exact managed listener only while its accepted work is idle.

Expand Down Expand Up @@ -190,21 +216,8 @@ def restart_managed_loopx_services() -> list[str]:
chat_plists.append(plist)
restarted: list[str] = []
uid = os.getuid() if hasattr(os, "getuid") else 0
from .chat_ingress import (
chat_runtime_activation_lock,
chat_runtime_activation_state,
)

try:
roots = sorted({_chat_runtime_root(plist) for plist in chat_plists})
with ExitStack() as locks:
for root in roots:
locks.enter_context(
chat_runtime_activation_lock(root, timeout_seconds=2.0)
)
for root in roots:
if chat_runtime_activation_state(root)["nonterminal_turns"]:
raise RuntimeActivationDeferred("chat_work_pending", labels)
with managed_chat_activation_guard(chat_plists):
# Keep admission fenced until launchctl has replaced the old process.
# Existing workers need no new lock: any accepted queued/active work
# was checked above, including completing and orphaned queued Turns.
Expand All @@ -227,8 +240,10 @@ def restart_managed_loopx_services() -> list[str]:
raise RuntimeActivationFailed(
restarted, [s for s in labels if s not in restarted]
) from exc
except RuntimeActivationDeferred:
raise
except RuntimeActivationDeferred as exc:
# The shared Chat guard names its participants; an update still defers
# the complete original status/chat activation batch.
raise RuntimeActivationDeferred(str(exc), labels) from exc
except (
OSError,
ValueError,
Expand Down
39 changes: 37 additions & 2 deletions packages/loopx-codex-provider-routing/NATIVE.md
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,9 @@ inspected GPT instance, waits for exit and native idle, saves/switches the login
then reopens that same route. It never addresses a shared bundle id or quits a
different route. With the separate force permission below, a refused or timed-out
normal quit can escalate against that same inspected instance. Otherwise those
failures, or remaining CLI/file owners, leave credentials unchanged and pause
automatic shutdowns until a check. Changed process ownership leaves credentials
failures leave credentials unchanged and pause automatic shutdowns until a check.
Remaining CLI/file owners wait without consuming that failure budget and resume
automatically after the blocker releases. Changed process ownership leaves credentials
unchanged and consumes the bounded retry budget.
Without this additional permission it continues to wait for closed clients.
It does not implement hot login replacement, job replay or
Expand Down Expand Up @@ -106,6 +107,40 @@ unreadable live Codex process into a proven exit.
只读重扫,再以稳定的新身份清单进入原受控重启流程。PID 复用、仍存活但身份不明及
持续变化继续等待;重扫不重放退出、凭据事务或业务,也不扩大 VM 的重启权限。

Version 0.12.11 checks for non-restartable holders before quitting the GPT App.
An existing `other_clients_busy` pause is rechecked automatically; an unchanged
unmanaged holder waits without repeatedly closing/reopening clients. Normal
quit refusal, unknown effects and other exhausted failure budgets keep their
existing recovery boundary. Only numeric read-only directory descriptors are
excluded from auth consumers. A read-only regular file, missing descriptor type,
directory writer or warm Codex process still blocks. The native slot owner's idle
check rechecks these read-only exceptions, followed by another typed handle scan
under its account locks. This does not restart a VM or claim to inspect a guest's
cached login; a pinned auth file remains a hold.

Before stopping a discovered Chat service, the provider uses LoopX's existing
accepted-work reader and ingress lock for its explicit runtime. Queued work,
including an orphan without an active Session pointer, completing work and
unreadable state defer the entire handoff before App quit. A new/changed service
cannot inherit the frozen restart set. The lock remains held through account
readback and restoration. A standalone observer must run in the same upgraded
LoopX environment to restart these services; missing Core support waits safely.
An already loaded archive/bootstrap Core passes its import root to the native
worker, including detached handoffs; the worker does not rely on its current
directory to rediscover that owner.
Prefer the serving App/Chat observer so normal LoopX updates also refresh it;
retire any separately installed legacy watcher through its original manager.

0.12.11 在退出 GPT App 前检查无法受控恢复的占用;原 `other_clients_busy` 暂停会
自动重新检查,阻塞解除后无需手动换号。仅有数字 FD、只读且类型明确为目录的句柄
不视为认证消费者;认证文件、写句柄、未知类型及缓存登录的 Codex 进程仍阻塞。
原账号 owner 在事务锁内复查只读例外,并再次核验句柄;不重启 VM,也不据此声称
已检查 guest 内的认证缓存。停止 Chat 前复用原在途任务读取与 ingress 锁,排队、
completing 或不可读状态均在退出 App 前等待;恢复期间保持锁与原失败/未知回执。
已加载的 archive/bootstrap Core 向原 native 子进程传递其加载路径,包含脱离父进程的
换号任务,避免依赖工作目录或误用旧安装;没有加载 Core 的独立入口仍安全等待。
观察器应随实际 LoopX 服务升级;独立旧 watcher 通过原管理器退役,避免占住唯一观察权。

Each service's restoration intent is saved privately before unloading. Version
0.12.7 waits up to five seconds for launchd to finish an acknowledged asynchronous
unload; a job that remains loaded retains its recovery intent and fails without
Expand Down
2 changes: 1 addition & 1 deletion packages/loopx-codex-provider-routing/extension.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
schema_version = "loopx_extension_manifest_v0"
id = "loopx-codex-provider-routing"
version = "0.12.10"
version = "0.12.11"
requires_loopx_api = ">=1,<2"
permissions = []

Expand Down
2 changes: 1 addition & 1 deletion packages/loopx-codex-provider-routing/pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"

[project]
name = "loopx-codex-provider-routing"
version = "0.12.10"
version = "0.12.11"
description = "Native Codex account slots and exhausted-account handoff for LoopX"
readme = "README.md"
requires-python = ">=3.11"
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
from unittest.mock import patch

PACKAGE_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(PACKAGE_ROOT.parents[1])) # The actual LoopX service fence.
sys.path.insert(0, str(PACKAGE_ROOT / "src"))
host_module = importlib.import_module("loopx_codex_provider_routing.native_host")
from loopx_codex_provider_routing.native import private_read # noqa: E402
Expand All @@ -38,7 +39,8 @@ def __init__(self, home):
self.plist = home / "Library/LaunchAgents" / (LABEL + ".plist")
self.service_pid = 300
self.plist.parent.mkdir(parents=True)
self.plist.write_bytes(plistlib.dumps({"Label": LABEL, "ProgramArguments": ["/test/chat"]}))
self.service_args = ["/test/chat", "--runtime-root", str(home / "runtime"), "chat"]
self.plist.write_bytes(plistlib.dumps({"Label": LABEL, "ProgramArguments": self.service_args}))
self.plist.chmod(0o600)
self.root = home / "native-accounts"
self.root.mkdir(mode=0o700)
Expand Down Expand Up @@ -84,8 +86,9 @@ def run(self, command, **_kwargs):
return SimpleNamespace(returncode=0, stdout="".join(f"p{p}\n" for p in self.holders))
verb = command[1]
if verb == "print":
arguments = "arguments = {\n" + "\n".join(self.service_args) + "\n}\n"
return SimpleNamespace(returncode=0 if self.loaded else 113,
stdout=f"path = {self.plist}\npid = {self.service_pid}\n" if self.loaded else "")
stdout=f"path = {self.plist}\npid = {self.service_pid}\n{arguments}" if self.loaded else "")
if verb == "bootout":
self.loaded = False
self.holders = [p for p in self.holders if p != 301]
Expand Down Expand Up @@ -129,6 +132,10 @@ def context(self):
def handoff(self, **kwargs):
return self.host.switch("b", "before", reopen=True, restart_clients=True, **kwargs)

def pause(self):
with self.host.restart_guard():
return self.host.pause_chat_agents()


def error(code, operation):
try:
Expand Down Expand Up @@ -229,7 +236,8 @@ def failed_environment(pid):
renamed = "org.example.codex.transport"
f.plist.rename(f.plist.with_name("transport-installation.plist"))
f.plist = f.plist.with_name("transport-installation.plist")
f.plist.write_bytes(plistlib.dumps({"Label": renamed, "ProgramArguments": ["/test/chat"]}))
f.plist.write_bytes(plistlib.dumps({"Label": renamed, "ProgramArguments": [
"/test/chat", "--runtime-root", str(f.home / "runtime"), "chat"]}))
with f.context():
assert f.handoff()["active_slot"] == "b"
assert any(c[1:3] == ["bootout", f"gui/{os.getuid()}/{renamed}"] for c in f.calls)
Expand Down Expand Up @@ -268,25 +276,27 @@ def stuck_unload(command, **kwargs):
f.run = stuck_unload
with f.context(), patch.object(host_module.time, "monotonic", side_effect=iter(range(0, 100, 2))), \
patch.object(host_module.time, "sleep"):
error("chat_pause_failed", f.host.pause_chat_agents)
error("chat_pause_failed", f.pause)
assert f.switches == 0 and private_read(f.root / "chat-recovery.json")["agents"]

# A second independent service participates in the same transaction. Neither
# 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"]}))
second.write_bytes(plistlib.dumps({"Label": "org.example.second", "ProgramArguments": [
"/test/other", "--runtime-root", str(f.home / "runtime"), "chat"]}))
f.records.update({700: (1, "/test/other", 1500, 14), 701: (700, "/test/codex", 1501, 15)})
second_state = {"loaded": True, "pid": 700}
primary_run = f.run
def multi_run(command, **kwargs):
if (command[0] == "/bin/launchctl" and
if (Path(command[0]).name == "launchctl" and
(command[-1].endswith("/org.example.second") or command[-1] == str(second))):
f.calls.append(command)
if command[1] == "print":
arguments = "arguments = {\n" + "\n".join(plistlib.loads(second.read_bytes())["ProgramArguments"]) + "\n}\n"
return SimpleNamespace(returncode=0 if second_state["loaded"] else 113,
stdout=f"path = {second}\npid = {second_state['pid']}\n")
stdout=f"path = {second}\npid = {second_state['pid']}\n{arguments}")
if command[1] == "bootout":
second_state["loaded"] = False
f.records.pop(700)
Expand Down Expand Up @@ -325,7 +335,7 @@ def multi_run(command, **kwargs):
with f.context(), patch.object(host_module, "codex_home", return_value=str(f.home if same_home else f.home / "other")), \
patch.object(host_module.time, "monotonic", side_effect=lambda: next(clock)):
if same_home:
error("other_clients_busy", f.handoff)
error("waiting_for_idle", f.handoff)
else:
assert f.handoff()["active_slot"] == "b"
assert 200 not in f.killed and f.switches == (0 if same_home else 1)
Expand Down Expand Up @@ -401,13 +411,13 @@ def changed_run(command, **kwargs):
f.host, "before", inspected=(100, (1, EXE, 99, 0))))

# A remaining independent CLI blocks the full switch and leaves its work
# untouched; the source App is reopened without changing the account.
# untouched. Read-only preflight leaves even the source App running.
f = fixture()
clock = iter(range(0, 1000, 20))
with f.context(), patch.object(host_module.time, "monotonic", side_effect=lambda: next(clock)):
error("other_clients_busy", f.handoff)
assert f.killed == [101] and f.holders == [200] and f.switches == 0
assert f.active == "a" and f.app_pid is not None
error("waiting_for_idle", f.handoff)
assert not f.killed and f.holders == [101, 200] and f.switches == 0
assert f.active == "a" and f.app_pid == 100

# PID reuse, a prior orphan, a new launch and unprovable identity stay holds.
for mutation in ("pid_reused", "unreadable", "new_launch", "selection_changed", "quota_stale"):
Expand Down Expand Up @@ -496,7 +506,7 @@ def changed_run(command, **kwargs):
f = fixture()
f.holders = [301]
with f.context():
f.host.pause_chat_agents()
f.pause()
f.plist.write_text("changed by user")
error("chat_resume_failed", f.host.resume_chat_agents)
assert not f.loaded
Expand All @@ -507,7 +517,7 @@ def changed_run(command, **kwargs):
f = fixture()
f.holders = [301]
with f.context():
f.host.pause_chat_agents()
f.pause()
journal = f.root / "chat-recovery.json"
recovery = private_read(journal)
recovery["owner"]["pid"] = 500
Expand Down
Original file line number Diff line number Diff line change
@@ -1,3 +1,3 @@
__all__ = ["__version__"]

__version__ = "0.12.10"
__version__ = "0.12.11"
Loading
Loading