diff --git a/docs/guides/installing-loopx.md b/docs/guides/installing-loopx.md index bc653c9bf8..c8ecdcfeea 100644 --- a/docs/guides/installing-loopx.md +++ b/docs/guides/installing-loopx.md @@ -139,6 +139,21 @@ Chat listener before switching to launchd, and read back the original Sessions and one consumer per App. A generated plist alone does not qualify recovery after logout or reboot. +The managed Chat replacement entry also checks accepted work before sending +SIGTERM. It resolves the existing listener's runtime from its live macOS +LaunchAgent and holds the same ingress fence used by update activation; a new +command's empty runtime cannot authorize stopping a different busy store. +Queued, running or completing work defers replacement with `chat_work_pending`. +An unknown manager, route or unreadable store returns `chat_state_unverified` +and leaves the listener running. A free port still starts normally. This check +does not intercept external `launchctl bootout` or operating-system termination. + +中文：受管 Chat 的显式替换入口同样先检查已受理任务。它按现有 listener 的真实 +macOS LaunchAgent 确定运行目录，并持有升级激活共用的准入锁；新命令的空目录不能 +作为停止另一份忙碌状态的依据。排队、运行中或结算中的任务使替换延期；管理器、路由 +或状态无法核验时保留原进程。空闲端口仍可正常启动；外部 `launchctl bootout` 和操作 +系统终止不受此入口拦截。 + The helper reads the full `/status.json` contract with a one-second connection deadline and a 15-second total deadline. A responding feed that takes longer than five seconds can therefore still show its schema and write-API setting. diff --git a/loopx/cli_commands/support_control.py b/loopx/cli_commands/support_control.py index 4a08edd18a..e26bb6e400 100644 --- a/loopx/cli_commands/support_control.py +++ b/loopx/cli_commands/support_control.py @@ -16,7 +16,8 @@ serve_chat, ) from ..control_plane.scheduler.execution_context import SchedulerRuntimeProfile -from ..dashboard_launcher import launch_dashboard, replace_existing_loopx_chat +from ..dashboard_launcher import launch_dashboard +from ..runtime_activation import replace_managed_loopx_chat as replace_existing_loopx_chat from ..execution_profile import execution_profile_turn_granularity from ..heartbeat_prequota import ( render_heartbeat_pre_quota_markdown, diff --git a/loopx/runtime_activation.py b/loopx/runtime_activation.py index c016c1323f..f7eacb1699 100644 --- a/loopx/runtime_activation.py +++ b/loopx/runtime_activation.py @@ -11,9 +11,12 @@ import os from pathlib import Path import plistlib +import re import shlex +import signal import subprocess import sys +import time from typing import Any from xml.parsers.expat import ExpatError @@ -82,6 +85,92 @@ def _chat_runtime_root(plist: Path) -> Path: return root.resolve() +def _managed_service_plists() -> list[Path]: + return [ + plist for plist in sorted((Path.home() / "Library/LaunchAgents").glob("*.plist")) + if ("loopx" in plist.stem.lower() or "goal-harness" in plist.stem.lower()) + and plist.stem.lower().endswith((".status", ".chat")) + ] + + +def _managed_service_pid(plist: Path) -> int | None: + result = subprocess.run( + ["launchctl", "print", f"gui/{os.getuid()}/{plist.stem}"], + capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=5, + ) + if result.returncode != 0: + return None + payload = plistlib.loads(plist.read_bytes()) + if not isinstance(payload, dict): + return None + configured = payload.get("ProgramArguments") + arguments = re.findall(r"^\s*arguments = \{\n(.*?)^\s*\}", result.stdout, re.MULTILINE | re.DOTALL) + if (not isinstance(configured, list) or not all(isinstance(arg, str) for arg in configured) + or len(arguments) != 1 + or [line.strip() for line in arguments[0].splitlines()] != configured): + return None # An edited plist is not proof of the running process's route. + pids = re.findall(r"^\s*pid = ([0-9]+)\s*$", result.stdout, re.MULTILINE) + return int(pids[0]) if len(pids) == 1 else None + + +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. + + The listener's live manager supplies its route, never the new command's + potentially different/empty runtime. Keep the existing ingress fence through + SIGTERM and port release. An unresolved manager or store leaves it running. + """ + from .chat_server import is_loopback_host + from .chat_ingress import chat_runtime_activation_lock, chat_runtime_activation_state + from .dashboard_launcher import ( + _is_same_user_loopx_chat_process, _listener_pids, _probe_existing_chat, + ) + + if not is_loopback_host(host): + raise ValueError("managed LoopX Chat replacement requires an explicit loopback host") + existing = _probe_existing_chat(host, port) + if existing == "unavailable": + return existing + if existing == "foreign": + raise RuntimeError(f"port {port} is owned by an unverified service; stopped none") + pids = _listener_pids(port) + if len(pids) != 1 or pids[0] == os.getpid() or not _is_same_user_loopx_chat_process(pids[0]): + raise RuntimeError(f"expected one verified same-user LoopX Chat listener on port {port}; stopped none") + pid = pids[0] + labels: list[str] = [] + try: + owners = [ + plist for plist in _managed_service_plists() + if sys.platform == "darwin" and plist.stem.lower().endswith(".chat") + and _managed_service_pid(plist) == pid + ] + if len(owners) != 1: + raise ValueError("chat_service_route_unresolved") + plist = owners[0] + labels = [plist.stem] + root = _chat_runtime_root(plist) + with chat_runtime_activation_lock(root, timeout_seconds=2.0): + if chat_runtime_activation_state(root)["nonterminal_turns"]: + raise RuntimeActivationDeferred("chat_work_pending", labels) + if (_managed_service_pid(plist) != pid or _chat_runtime_root(plist) != root + or _listener_pids(port) != [pid] + or not _is_same_user_loopx_chat_process(pid)): + raise ValueError("chat_service_route_unresolved") + # Signal the pinned PID, not a second lookup's possible replacement. + os.kill(pid, signal.SIGTERM) + deadline = time.monotonic() + max(0.1, timeout_seconds) + while time.monotonic() < deadline: + if pid not in _listener_pids(port): + return existing + time.sleep(0.1) + raise RuntimeError("the managed Chat listener did not stop after SIGTERM; no stronger signal was sent") + except RuntimeActivationDeferred: + raise + except (OSError, ValueError, plistlib.InvalidFileException, ExpatError, + TimeoutError, subprocess.SubprocessError) as exc: + raise RuntimeActivationDeferred("chat_state_unverified", labels) from exc + + def restart_managed_loopx_services() -> list[str]: """Switch managed macOS services only with no accepted Chat work outstanding. @@ -92,17 +181,10 @@ def restart_managed_loopx_services() -> list[str]: """ if sys.platform != "darwin": return [] - agents_dir = Path.home() / "Library" / "LaunchAgents" - if not agents_dir.is_dir(): - return [] labels: list[str] = [] chat_plists: list[Path] = [] - for plist in sorted(agents_dir.glob("*.plist")): + for plist in _managed_service_plists(): stem = plist.stem.lower() - if "loopx" not in stem and "goal-harness" not in stem: - continue - if not (stem.endswith(".status") or stem.endswith(".chat")): - continue labels.append(plist.stem) if stem.endswith(".chat"): chat_plists.append(plist) diff --git a/tests/test_chat_startup_options.py b/tests/test_chat_startup_options.py index c853c951c4..220a0da6fd 100644 --- a/tests/test_chat_startup_options.py +++ b/tests/test_chat_startup_options.py @@ -1,6 +1,12 @@ from __future__ import annotations import inspect +import hashlib +from pathlib import Path +import plistlib +import signal +import subprocess +import threading import pytest @@ -8,6 +14,209 @@ from loopx.cli import build_parser from loopx.cli_commands import support_control from loopx.cli_commands.support_control_chat import chat_server_options +from loopx.chat_store import ChatSessionStore +from loopx import dashboard_launcher, runtime_activation + + +@pytest.fixture +def managed_listener(tmp_path, monkeypatch): + monkeypatch.setattr(Path, "home", staticmethod(lambda: tmp_path)) + monkeypatch.setattr(runtime_activation.sys, "platform", "darwin") + agents = tmp_path / "Library/LaunchAgents" + agents.mkdir(parents=True) + runtime = tmp_path / "original history with spaces" + plist = agents / "com.loopx.chat.plist" + plist.write_bytes(plistlib.dumps({"ProgramArguments": [ + "loopx", "--runtime-root", str(runtime), "chat", + ]})) + store = ChatSessionStore(runtime) + sid = store.create_session(goal_id="fixture", agent_id="fixture", adapter_kind="fake", + upstream_thread_id="fixture", upstream_mode="managed")["session_id"] + listeners, signals, starts = [4321], [], [] + monkeypatch.setattr(dashboard_launcher, "_probe_existing_chat", lambda *_args: "stale") + monkeypatch.setattr(dashboard_launcher, "_listener_pids", lambda _port: list(listeners)) + monkeypatch.setattr(dashboard_launcher, "_is_same_user_loopx_chat_process", lambda pid: pid == 4321) + monkeypatch.setattr(runtime_activation, "_managed_service_pid", lambda _plist: 4321, raising=False) + + def stop(pid, sig): + if sig != 0: # File-lock liveness probes are reads, not a stop attempt. + signals.append((pid, sig)) + listeners.clear() + + monkeypatch.setattr(runtime_activation.os, "kill", stop) + monkeypatch.setattr(support_control, "serve_chat", lambda **options: starts.append(options)) + + def invoke(): + # This fresh empty directory must never mask the listener's original work. + args = build_parser().parse_args(["--runtime-root", str(tmp_path / "new empty root"), + "chat", "--replace-existing-loopx-chat", "--no-open"]) + payloads = [] + result = support_control.handle_support_control_command(args, registry_path=tmp_path / "registry.json", + registry_was_supplied=True, print_payload=lambda payload, *_args: payloads.append(payload), + output_format=lambda *_args: "json") + return result, payloads + + return store, sid, plist, listeners, signals, starts, invoke + + +def history_bytes(store): + return {str(p.relative_to(store.root)): hashlib.sha256(p.read_bytes()).hexdigest() + for p in store.root.rglob("*") if p.suffix in {".json", ".jsonl"}} + + +@pytest.mark.parametrize("state", ["queued", "starting", "running", "completing", "interrupting"]) +def test_explicit_replacement_preserves_all_accepted_work(managed_listener, state): + store, sid, _plist, _listeners, signals, starts, invoke = managed_listener + turn, _ = store.create_turn(sid, client_turn_id="accepted", message="keep original input") + store.update_turn(sid, turn["turn_id"], status=state) + before = history_bytes(store) + result, payloads = invoke() + assert result == 1 and payloads[0]["error"] == "chat_work_pending" + assert signals == starts == [] and history_bytes(store) == before + + +def test_explicit_replacement_waits_for_orphaned_queue_then_replaces_same_pid(managed_listener): + store, sid, _plist, _listeners, signals, starts, invoke = managed_listener + turn, _ = store.create_queued_turn(sid, client_turn_id="queued", message="original queue") + assert not store.load_session(sid).get("active_turn_id") + assert invoke()[0] == 1 and not signals + store.update_turn(sid, turn["turn_id"], status="completed") + before = history_bytes(store) + assert invoke() == (0, []) + assert signals == [(4321, signal.SIGTERM)] and len(starts) == 1 + assert history_bytes(store) == before + + +@pytest.mark.parametrize("damage", ["missing_route", "relative_route", "bad_plist", "bad_session", "bad_turn", "missing_active", "unknown_state", "unknown_manager"]) +def test_explicit_replacement_keeps_unverified_state_untouched(managed_listener, monkeypatch, damage): + store, sid, plist, _listeners, signals, starts, invoke = managed_listener + turn, _ = store.create_turn(sid, client_turn_id="accepted", message="preserve") + store.update_turn(sid, turn["turn_id"], status="completed") + if damage == "missing_route": + plist.write_bytes(plistlib.dumps({"ProgramArguments": ["loopx", "chat"]})) + elif damage == "relative_route": + plist.write_bytes(plistlib.dumps({"ProgramArguments": ["loopx", "--runtime-root", "relative", "chat"]})) + elif damage == "bad_plist": + plist.write_text("{broken") + elif damage == "bad_session": + store._session_path(sid).write_text("{broken") + elif damage == "bad_turn": + store._turn_path(sid, turn["turn_id"]).write_text("{broken") + elif damage == "missing_active": + store.update_turn(sid, turn["turn_id"], status="running") + store._turn_path(sid, turn["turn_id"]).unlink() + elif damage == "unknown_state": + store.update_turn(sid, turn["turn_id"], status="unknown") + else: + monkeypatch.setattr(runtime_activation, "_managed_service_pid", lambda _plist: None) + before = history_bytes(store) + result, payloads = invoke() + assert result == 1 and payloads[0]["error"] == "chat_state_unverified" + assert not signals and not starts and history_bytes(store) == before + + +@pytest.mark.parametrize("drift", ["manager", "listener", "route", "identity"]) +def test_explicit_replacement_rechecks_the_pinned_target_under_fence(managed_listener, monkeypatch, drift): + store, _sid, plist, listeners, signals, starts, invoke = managed_listener + if drift == "manager": + pids = iter([4321, 9876]) + monkeypatch.setattr(runtime_activation, "_managed_service_pid", lambda _plist: next(pids)) + elif drift == "listener": + reads = iter([[4321], [9876]]) + monkeypatch.setattr(dashboard_launcher, "_listener_pids", lambda _port: next(reads)) + elif drift == "identity": + reads = iter([True, False]) + monkeypatch.setattr(dashboard_launcher, "_is_same_user_loopx_chat_process", lambda _pid: next(reads)) + else: + original = runtime_activation._chat_runtime_root + reads = iter([store.root.parent, store.root.parent / "different"]) + monkeypatch.setattr(runtime_activation, "_chat_runtime_root", lambda _plist: next(reads)) + assert original(plist) == store.root.parent + assert invoke()[0] == 1 and not signals and not starts + + +def test_explicit_replacement_holds_the_real_ingress_fence_through_stop(managed_listener, monkeypatch): + store, sid, _plist, listeners, signals, _starts, invoke = managed_listener + stopping, release, attempted, admitted = [threading.Event() for _ in range(4)] + results, errors = [], [] + + def stop(pid, sig): + if sig == 0: + return + signals.append((pid, sig)) + stopping.set() + assert release.wait(5) + listeners.clear() + + def accept(): + attempted.set() + try: + store.create_queued_turn(sid, client_turn_id="during-replacement", message="keep next input") + admitted.set() + except Exception as error: + errors.append(error) + + monkeypatch.setattr(runtime_activation.os, "kill", stop) + replacer = threading.Thread(target=lambda: results.append(invoke())) + writer = threading.Thread(target=accept) + replacer.start() + try: + assert stopping.wait(3) + writer.start() + assert attempted.wait(2) + assert not admitted.wait(0.15) + assert store.turn_for_client(sid, "during-replacement") is None + finally: + release.set() + replacer.join(5) + if writer.ident is not None: + writer.join(5) + assert not replacer.is_alive() and not writer.is_alive() and not errors + assert results == [(0, [])] and admitted.is_set() + assert len(signals) == 1 + + +def test_managed_service_pid_requires_one_exact_native_pid_field(monkeypatch, tmp_path): + monkeypatch.setattr(runtime_activation.os, "getuid", lambda: 501, raising=False) + args = "arguments = {\n loopx\n chat\n}\n" + outputs = iter([args + "pid = 4321", args + "other_pid = 4321", args + "pid = 4321\npid = 4321", + "arguments = {\n loopx\n other\n}\npid = 4321"]) + monkeypatch.setattr(runtime_activation.subprocess, "run", lambda args, **kwargs: + subprocess.CompletedProcess(args, 0, next(outputs), "")) + plist = tmp_path / "com.loopx.chat.plist" + plist.write_bytes(plistlib.dumps({"ProgramArguments": ["loopx", "chat"]})) + assert [runtime_activation._managed_service_pid(plist) for _ in range(4)] == [4321, None, None, None] + + +@pytest.mark.parametrize("probe, expected", [("unavailable", 0), ("foreign", 1)]) +def test_explicit_replacement_keeps_free_port_and_foreign_service_behavior(managed_listener, monkeypatch, probe, expected): + _store, _sid, _plist, _listeners, signals, starts, invoke = managed_listener + monkeypatch.setattr(dashboard_launcher, "_probe_existing_chat", lambda *_args: probe) + assert invoke()[0] == expected and not signals + assert len(starts) == (expected == 0) + + +def test_explicit_replacement_ignores_another_managed_services_busy_store(managed_listener, monkeypatch): + store, _sid, plist, _listeners, signals, starts, invoke = managed_listener + other = plist.with_name("com.loopx.other.chat.plist") + root = store.root.parent.parent / "other busy store" + other.write_bytes(plistlib.dumps({"ProgramArguments": ["loopx", "--runtime-root", str(root), "chat"]})) + busy = ChatSessionStore(root) + sid = busy.create_session(goal_id="other", agent_id="other", adapter_kind="fake", + upstream_thread_id="other", upstream_mode="managed")["session_id"] + busy.create_turn(sid, client_turn_id="other", message="leave another service running") + monkeypatch.setattr(runtime_activation, "_managed_service_pid", lambda selected: 4321 if selected == plist else 9876) + before = history_bytes(busy) + assert invoke() == (0, []) and len(signals) == len(starts) == 1 + assert history_bytes(busy) == before + + +def test_explicit_replacement_never_escalates_a_stop_timeout(managed_listener, monkeypatch): + _store, _sid, _plist, _listeners, signals, _starts, _invoke = managed_listener + monkeypatch.setattr(runtime_activation.os, "kill", lambda pid, sig: signals.append((pid, sig)) if sig else None) + with pytest.raises(RuntimeError, match="no stronger signal"): + runtime_activation.replace_managed_loopx_chat("127.0.0.1", 8767, timeout_seconds=0.1) + assert signals == [(4321, signal.SIGTERM)] @pytest.mark.parametrize("configured, expected", [(None, None), ("0", 0.1), ("12.5", 12.5), ("45", 45.0)])