fix(chat): preserve accepted work during explicit managed replacement

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.

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,

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)

tests/test_chat_startup_options.py

@@ -1,13 +1,222 @@
 from __future__ import annotations
 
 import inspect
+import hashlib
+from pathlib import Path
+import plistlib
+import signal
+import subprocess
+import threading
 
 import pytest
 
 from loopx.chat_server import serve_chat
 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)])