From 4c79fe9f8d84375f74fa03a9f4a320e086eeff10 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 25 Oct 2026 01:02:19 +0800 Subject: [PATCH 1/2] fix(chat): defer runtime activation while accepted work is pending Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/product/periodic-reports.md | 12 + loopx/chat_ingress.py | 2 +- loopx/chat_store.py | 56 +++- loopx/cli_commands/support_control_upgrade.py | 8 + loopx/runtime_activation.py | 152 ++++++++- loopx/self_update.py | 30 ++ tests/test_chat_runtime_activation.py | 298 ++++++++++++++++++ tests/test_self_update_runtime_activation.py | 5 +- tests/test_self_upgrade.py | 11 + 9 files changed, 557 insertions(+), 17 deletions(-) create mode 100644 tests/test_chat_runtime_activation.py diff --git a/docs/product/periodic-reports.md b/docs/product/periodic-reports.md index 19031622e7..d19db650de 100644 --- a/docs/product/periodic-reports.md +++ b/docs/product/periodic-reports.md @@ -84,6 +84,18 @@ eligible idle boundary. If host tools require a replacement upstream thread, preserve the visible conversation and history, and record that transport change. Do not claim to preserve a physical thread that was actually replaced. +The macOS CLI update path checks every configured Chat store before switching +its LaunchAgent services. Accepted, queued, completing or unreadable work defers +the whole service switch; an empty Session summary is insufficient. Updated +consumers and the updater share an admission lock through process replacement, +so a new request cannot pass the idle check and begin on the retiring process. +The first adoption from an older consumer without that fence still requires an +independently established idle boundary. Installation readback remains visible +when service activation is pending, including partial restart failures. The +same-source update retry performs a fresh check; this adds no second consumer, +automatic business replay or separate activation scheduler. The combined App +upgrade also stops before restarting the App when this service switch is pending. + Keep the steward's existing portfolio-read authority separate from its execution actor. Resolve both through their native owners on every admission. Registering the actor does not grant new project writes or broaden the original audience. diff --git a/loopx/chat_ingress.py b/loopx/chat_ingress.py index f5e041051d..a596cd0f1a 100644 --- a/loopx/chat_ingress.py +++ b/loopx/chat_ingress.py @@ -36,7 +36,7 @@ def create_queued_turn( if normalized_attachments: validate_chat_turn_envelope({"message": message, "attachments": normalized_attachments}) session_path = self._session_path(session_id) - with self._session_lock(session_id): + with self.runtime_activation_lock(), self._session_lock(session_id): with exclusive_file_lock( session_path, agent_id="loopx-chat", diff --git a/loopx/chat_store.py b/loopx/chat_store.py index 4ed8a85c61..e9941aacc0 100644 --- a/loopx/chat_store.py +++ b/loopx/chat_store.py @@ -59,6 +59,57 @@ def utc_now() -> str: return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") +def chat_runtime_activation_lock(runtime_root: Path, *, timeout_seconds: float | None = None): + """Fence new durable input against a managed-service runtime switch.""" + return exclusive_file_lock( + runtime_root.expanduser().resolve() / "chat" / "runtime-activation", + agent_id="loopx-chat", + operation="chat_runtime_activation", + timeout_seconds=timeout_seconds, + ) + + +def chat_runtime_activation_state(runtime_root: Path) -> dict[str, int]: + """Strict read of all accepted work; never initialize or recover a store. + + The caller holds ``chat_runtime_activation_lock`` through the service switch. + Session summaries alone omit queued input and are insufficient for this check. + Unknown or unreadable records must not authorize stopping their consumer. + """ + sessions = runtime_root.expanduser().resolve() / "chat" / "sessions" + counts = {"sessions": 0, "nonterminal_turns": 0} + if not sessions.exists(): + return counts + for directory in sessions.iterdir(): + if not directory.is_dir(): + continue + session = json.loads((directory / "session.json").read_text(encoding="utf-8")) + if not isinstance(session, dict) or session.get("schema_version") != CHAT_SESSION_SCHEMA_VERSION: + raise ValueError("unreadable Chat Session state") + counts["sessions"] += 1 + active_id = session.get("active_turn_id") + active_found = not active_id + turns = directory / "turns" + for path in turns.iterdir() if turns.exists() else (): + if path.suffix != ".json": + continue + turn = json.loads(path.read_text(encoding="utf-8")) + if (not isinstance(turn, dict) + or turn.get("schema_version") != CHAT_TURN_SCHEMA_VERSION + or not isinstance(turn.get("status"), str) + or turn.get("status") not in TERMINAL_TURN_STATES | { + "queued", "starting", "running", "completing", "interrupting", + }): + raise ValueError("unreadable Chat Turn state") + if turn.get("turn_id") == active_id: + active_found = True + if turn["status"] not in TERMINAL_TURN_STATES: + counts["nonterminal_turns"] += 1 + if not active_found: + raise ValueError("active Chat Turn state is missing") + return counts + + def _instant(value: Any) -> datetime | None: try: return datetime.fromisoformat(str(value).replace("Z", "+00:00")) @@ -228,6 +279,9 @@ def _session_dir(self, session_id: str) -> Path: raise ValueError("session_id must be a compact opaque id") return self.sessions_root / token + def runtime_activation_lock(self): + return chat_runtime_activation_lock(self.root.parent) + def _session_path(self, session_id: str) -> Path: return self._session_dir(session_id) / "session.json" @@ -1006,7 +1060,7 @@ def accept_managed_turn( candidate_message_id = uuid.uuid4().hex accepted_at = utc_now() session_path = self._session_path(session_token) - with self._session_lock(session_token): + with self.runtime_activation_lock(), self._session_lock(session_token): with exclusive_file_lock( session_path, agent_id="loopx-chat", diff --git a/loopx/cli_commands/support_control_upgrade.py b/loopx/cli_commands/support_control_upgrade.py index 3b18cf07d8..84f579fd31 100644 --- a/loopx/cli_commands/support_control_upgrade.py +++ b/loopx/cli_commands/support_control_upgrade.py @@ -269,6 +269,14 @@ def finish_upgrade( "CLI install finished but its runtime identity could not be read back. Run `loopx doctor --installation-only`." ) desktop = payload["desktop_upgrade"] + if payload.get("execution", {}).get("activation_pending"): + payload["desktop_upgrade"] = { + **desktop, + "status": "skipped_service_activation_pending", + } + payload["upgrade_complete"] = False + payload["ok"] = False + return payload if desktop["status"] == "not_installed": payload["ok"] = bool( payload.get("ok") and payload.get("upgrade_complete", payload.get("ok")) diff --git a/loopx/runtime_activation.py b/loopx/runtime_activation.py index b0c08ab28b..f1983436b6 100644 --- a/loopx/runtime_activation.py +++ b/loopx/runtime_activation.py @@ -7,15 +7,83 @@ from __future__ import annotations +from contextlib import ExitStack import os from pathlib import Path +import plistlib +import shlex import subprocess import sys from typing import Any +from xml.parsers.expat import ExpatError + + +class RuntimeActivationDeferred(Exception): + """The installed runtime is ready, but its existing consumers must survive.""" + + def __init__(self, reason: str, labels: list[str]) -> None: + super().__init__(reason) + self.labels = labels + + +class RuntimeActivationFailed(Exception): + def __init__(self, restarted: list[str], pending: list[str]) -> None: + super().__init__("service_restart_unverified") + self.restarted = restarted + self.pending = pending + + +def _chat_runtime_root(plist: Path) -> Path: + """Read an explicit LaunchAgent route, including the installer's shell wrapper. + + Never execute a plist's shell or expand arbitrary shell expressions while + discovering the state protected by an update. Unresolved routes stay running. + """ + payload = plistlib.loads(plist.read_bytes()) + if not isinstance(payload, dict): + raise ValueError("chat_service_route_unresolved") + arguments = payload.get("ProgramArguments") + if not isinstance(arguments, list) or not all( + isinstance(arg, str) for arg in arguments + ): + raise ValueError("chat_service_route_unresolved") + if arguments and Path(arguments[0]).name in {"sh", "bash", "zsh"}: + for index, argument in enumerate(arguments[1:], 1): + if ( + argument.startswith("-") + and "c" in argument[1:] + and index + 1 < len(arguments) + ): + arguments = shlex.split(arguments[index + 1]) + break + else: + raise ValueError("chat_service_route_unresolved") + if "exec" not in arguments: + raise ValueError("chat_service_route_unresolved") + arguments = arguments[arguments.index("exec") + 1 :] + candidates = [ + argument.split("=", 1)[1] + if argument.startswith("--runtime-root=") + else arguments[index + 1] + if index + 1 < len(arguments) + else "" + for index, argument in enumerate(arguments) + if argument == "--runtime-root" or argument.startswith("--runtime-root=") + ] + if ( + len(candidates) != 1 + or not candidates[0] + or any(c in candidates[0] for c in "$`\n") + ): + raise ValueError("chat_service_route_unresolved") + root = Path(candidates[0]).expanduser() + if not root.is_absolute(): + raise ValueError("chat_service_route_unresolved") + return root.resolve() def restart_managed_loopx_services() -> list[str]: - """Best-effort restart of user LaunchAgent-managed LoopX services on macOS. + """Switch managed macOS services only with no accepted Chat work outstanding. After ``loopx update`` replaces the installed release, running status/chat services still belong to the previous release. Restarting the managed @@ -28,6 +96,7 @@ def restart_managed_loopx_services() -> list[str]: if not agents_dir.is_dir(): return [] labels: list[str] = [] + chat_plists: list[Path] = [] for plist in sorted(agents_dir.glob("*.plist")): stem = plist.stem.lower() if "loopx" not in stem and "goal-harness" not in stem: @@ -35,17 +104,54 @@ def restart_managed_loopx_services() -> list[str]: if not (stem.endswith(".status") or stem.endswith(".chat")): continue labels.append(plist.stem) + if stem.endswith(".chat"): + chat_plists.append(plist) restarted: list[str] = [] uid = os.getuid() if hasattr(os, "getuid") else 0 - for label in labels: - result = subprocess.run( - ["launchctl", "kickstart", "-k", f"gui/{uid}/{label}"], - capture_output=True, - text=True, encoding="utf-8", errors="replace", - timeout=30, - ) - if result.returncode == 0: - restarted.append(label) + from .chat_store 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) + # 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. + for label in labels: + try: + result = subprocess.run( + ["launchctl", "kickstart", "-k", f"gui/{uid}/{label}"], + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=30, + ) + if result.returncode != 0: + raise RuntimeActivationFailed( + restarted, [s for s in labels if s not in restarted] + ) + restarted.append(label) + except (OSError, subprocess.SubprocessError) as exc: + raise RuntimeActivationFailed( + restarted, [s for s in labels if s not in restarted] + ) from exc + except RuntimeActivationDeferred: + raise + except ( + OSError, + ValueError, + plistlib.InvalidFileException, + ExpatError, + TimeoutError, + ) as exc: + raise RuntimeActivationDeferred("chat_state_unverified", labels) from exc return restarted @@ -82,7 +188,25 @@ def restart_services_for_runtime_activation( "restarted_services": [], "restart_status": "skipped_custom_installation", } - return { - "restarted_services": restart_managed_loopx_services(), - "restart_status": "restarted", - } + try: + return { + "restarted_services": restart_managed_loopx_services(), + "restart_status": "restarted", + "activation_pending": False, + } + except RuntimeActivationDeferred as exc: + return { + "restarted_services": [], + "deferred_services": exc.labels, + "restart_status": "deferred", + "restart_reason": str(exc), + "activation_pending": True, + } + except RuntimeActivationFailed as exc: + return { + "restarted_services": exc.restarted, + "deferred_services": exc.pending, + "restart_status": "failed", + "restart_reason": str(exc), + "activation_pending": True, + } diff --git a/loopx/self_update.py b/loopx/self_update.py index 5522a54b11..cc4d9db19e 100644 --- a/loopx/self_update.py +++ b/loopx/self_update.py @@ -946,6 +946,27 @@ def build_update_plan( return (_windows_console_update_refusal(payload) if execute else None) or payload +def _retain_pending_service_activation(updated: dict[str, Any]) -> bool: + """Keep installation success distinct from switching its live consumers.""" + if not updated["execution"].get("activation_pending"): + return False + updated["ok"] = False + updated["recommended_action"] = ( + "the runtime is installed, but managed-service activation is incomplete. " + "Let original Chat work finish, or inspect the reported service-state or " + "restart failure, then retry the same update source" + ) + updated["next_action"] = { + "kind": "retry_runtime_activation", + "command": (updated.get("commands") or {}).get("apply") + or _update_action_command(UpdateAction.APPLY, updated.get("source") or {}), + "mutating": True, + "requires_explicit_approval": True, + "reason": updated["recommended_action"], + } + return True + + def _execute_python_distribution_update( payload: dict[str, Any], *, @@ -1084,6 +1105,8 @@ def _execute_python_distribution_update( runtime_ready=runtime_ready, ) ) + if _retain_pending_service_activation(updated): + return updated if updated["ok"]: updated["recommended_action"] = ( "PyPI update and host-material readback passed; use the new LoopX process" @@ -1259,6 +1282,8 @@ def execute_update_plan( installed_bin=loopx_bin, ) ) + if _retain_pending_service_activation(updated): + return updated if not updated["ok"] and runtime_ready: updated["recommended_action"] = ( "the updated runtime is installed; repair the blocked " @@ -1539,6 +1564,11 @@ def render_update_plan_markdown(payload: dict[str, Any]) -> str: f"- Version: `{installed_identity.get('package_version')}`", f"- Source revision: `{installed_identity.get('source_revision')}`", "- Fresh process installation check passed; the plan below describes the pre-update installation."]) + if (payload.get("execution") or {}).get("activation_pending"): + lines.extend(["", "## Service activation pending", "", + "- The CLI is installed; managed-service activation is incomplete.", + f"- Reason: `{payload['execution'].get('restart_reason')}`", + "- Accepted Chat work stays with its original consumer. Retry the same update after it finishes or its unreadable state is repaired."]) lines.extend( [ "", diff --git a/tests/test_chat_runtime_activation.py b/tests/test_chat_runtime_activation.py new file mode 100644 index 0000000000..dc15d5daec --- /dev/null +++ b/tests/test_chat_runtime_activation.py @@ -0,0 +1,298 @@ +"""Update the real store/activation boundary without stopping a real service.""" + +from __future__ import annotations + +import hashlib +import json +from pathlib import Path +import plistlib +import shlex +import subprocess +import threading +from unittest import mock + +import pytest + +from loopx.chat_store import ChatSessionStore, chat_runtime_activation_lock +from loopx.runtime_activation import restart_services_for_runtime_activation +from loopx.self_update import execute_update_plan, render_update_plan_markdown + + +@pytest.fixture +def services(tmp_path, monkeypatch): + monkeypatch.setattr("loopx.runtime_activation.sys.platform", "darwin") + monkeypatch.setattr(Path, "home", staticmethod(lambda: tmp_path)) + monkeypatch.delenv("LOOPX_BIN_DIR", raising=False) + agents = tmp_path / "Library/LaunchAgents" + agents.mkdir(parents=True) + runtime = tmp_path / "runtime with spaces" + route = ["loopx", "--runtime-root", str(runtime), "chat"] + (agents / "com.loopx.chat.plist").write_bytes( + plistlib.dumps({"ProgramArguments": route}) + ) + (agents / "com.loopx.status.plist").write_bytes( + plistlib.dumps({"ProgramArguments": ["loopx", "serve-status"]}) + ) + calls = [] + original_run = subprocess.run + + def launchctl(args, **kwargs): + if args[0] != "launchctl": + return original_run(args, **kwargs) + calls.append(args) + return subprocess.CompletedProcess(args, 0, "", "") + + monkeypatch.setattr("loopx.runtime_activation.subprocess.run", launchctl) + return runtime, agents, calls + + +def session(store): + return store.create_session( + goal_id="fixture", + agent_id="fixture", + adapter_kind="fake", + upstream_thread_id="fixture", + upstream_mode="managed", + )["session_id"] + + +def activate(): + return restart_services_for_runtime_activation( + changes_applied=True, runtime_ready=True + ) + + +def content_hashes(root): + return { + str(p.relative_to(root)): hashlib.sha256(p.read_bytes()).hexdigest() + for p in root.rglob("*") + if p.is_file() and p.suffix in {".json", ".jsonl"} + } + + +@pytest.mark.parametrize( + "state", ["queued", "starting", "running", "completing", "interrupting"] +) +def test_update_never_restarts_accepted_work_or_rewrites_its_input(services, state): + runtime, _agents, calls = services + store = ChatSessionStore(runtime) + sid = session(store) + turn, _ = store.create_turn( + sid, client_turn_id="original", message="original request" + ) + store.update_turn(sid, turn["turn_id"], status=state) + before = content_hashes(store.root) + result = activate() + assert ( + result["activation_pending"] and result["restart_reason"] == "chat_work_pending" + ) + assert result["restarted_services"] == [] and len(result["deferred_services"]) == 2 + assert not calls + assert content_hashes(store.root) == before + + +def test_queue_without_an_active_pointer_still_defers_until_original_work_finishes( + services, +): + runtime, _agents, calls = services + store = ChatSessionStore(runtime) + sid = session(store) + queued, _ = store.create_queued_turn( + sid, client_turn_id="queued", message="later input" + ) + assert not store.load_session(sid).get("active_turn_id") + assert activate()["restart_status"] == "deferred" + assert not calls + # The original consumer, represented here by the real store's terminal write, + # finishes the same Turn. Activation never creates or replays a business Turn. + store.update_turn(sid, queued["turn_id"], status="completed") + assert activate()["restart_status"] == "restarted" + assert len(calls) == 2 + assert store.load_turn(sid, queued["turn_id"])["message"] == "later input" + + +@pytest.mark.parametrize( + "damage", ["session", "turn", "active_missing", "unknown_status", "invalid_status"] +) +def test_unreadable_or_inconsistent_work_fails_closed(services, damage): + runtime, _agents, calls = services + store = ChatSessionStore(runtime) + sid = session(store) + turn, _ = store.create_turn(sid, client_turn_id="original", message="preserve me") + if damage == "session": + store._session_path(sid).write_text("{broken") + elif damage == "turn": + store._turn_path(sid, turn["turn_id"]).write_text("{broken") + elif damage == "active_missing": + store._turn_path(sid, turn["turn_id"]).unlink() + elif damage == "unknown_status": + store.update_turn(sid, turn["turn_id"], status="unrecognized") + else: + store.update_turn(sid, turn["turn_id"], status=[]) + before = content_hashes(store.root) + result = activate() + assert result["restart_status"] == "deferred" + assert result["restart_reason"] == "chat_state_unverified" + assert not calls and content_hashes(store.root) == before + + +@pytest.mark.parametrize("wrapper", [False, True]) +def test_idle_service_uses_its_explicit_route_including_shell_wrapper( + services, wrapper +): + runtime, agents, calls = services + store = ChatSessionStore(runtime) + session(store) + if wrapper: + command = shlex.join( + ["/synthetic/loopx", "--runtime-root", str(runtime), "chat"] + ) + (agents / "com.loopx.chat.plist").write_bytes( + plistlib.dumps( + { + "ProgramArguments": [ + "/bin/zsh", + "-c", + f"export PATH=/synthetic; exec {command}", + ], + } + ) + ) + assert activate()["activation_pending"] is False + assert {call[-1].rsplit("/", 1)[-1] for call in calls} == { + "com.loopx.chat", + "com.loopx.status", + } + + +@pytest.mark.parametrize( + "arguments", + [ + [], + ["loopx", "chat"], + ["loopx", "--runtime-root", "$ROOT", "chat"], + ["loopx", "--runtime-root", "relative", "chat"], + ], +) +def test_unknown_service_route_never_falls_back_to_an_empty_default_store( + services, arguments +): + _runtime, agents, calls = services + (agents / "com.loopx.chat.plist").write_bytes( + plistlib.dumps({"ProgramArguments": arguments}) + ) + assert activate()["restart_reason"] == "chat_state_unverified" + assert not calls + + +@pytest.mark.parametrize("queue", [False, True]) +def test_restart_holds_admission_fence_until_process_replacement( + services, monkeypatch, queue +): + runtime, _agents, calls = services + store = ChatSessionStore(runtime) + sid = session(store) + replacing, release, admitted, attempted = [threading.Event() for _ in range(4)] + results, errors = [], [] + original_run = subprocess.run + + def launchctl(args, **kwargs): + if args[0] != "launchctl": + return original_run(args, **kwargs) + calls.append(args) + replacing.set() + assert release.wait(5) + return subprocess.CompletedProcess(args, 0, "", "") + + def accept_input(): + attempted.set() + try: + method = store.create_queued_turn if queue else store.create_turn + method(sid, client_turn_id="during-update", message="do not lose input") + admitted.set() + except Exception as error: + errors.append(error) + + monkeypatch.setattr("loopx.runtime_activation.subprocess.run", launchctl) + update = threading.Thread(target=lambda: results.append(activate())) + writer = threading.Thread(target=accept_input) + update.start() + try: + assert replacing.wait(3) + writer.start() + assert attempted.wait(2) + assert not admitted.wait(0.15) + assert store.turn_for_client(sid, "during-update") is None + finally: + release.set() + update.join(5) + if writer.ident is not None: + writer.join(5) + assert not update.is_alive() and not writer.is_alive() + assert not errors and admitted.is_set() + assert results[0]["restart_status"] == "restarted" + assert activate()["restart_reason"] == "chat_work_pending" + + +def test_contended_admission_is_not_treated_as_idle(services, monkeypatch): + runtime, _agents, calls = services + store = ChatSessionStore(runtime) + session(store) + original = chat_runtime_activation_lock + monkeypatch.setattr( + "loopx.chat_store.chat_runtime_activation_lock", + lambda root, **_kwargs: original(root, timeout_seconds=0), + ) + with original(runtime): + assert activate()["restart_reason"] == "chat_state_unverified" + assert not calls + + +@pytest.mark.parametrize("unknown", [False, True]) +def test_partial_restart_failure_keeps_the_actual_success_and_unknown_outcome( + services, monkeypatch, unknown +): + runtime, _agents, calls = services + ChatSessionStore(runtime) + + def launchctl(args, **_kwargs): + calls.append(args) + if len(calls) == 2 and unknown: + raise subprocess.TimeoutExpired(args, 30) + return subprocess.CompletedProcess(args, 1 if len(calls) == 2 else 0, "", "") + + monkeypatch.setattr("loopx.runtime_activation.subprocess.run", launchctl) + result = activate() + assert result["restart_status"] == "failed" and result["activation_pending"] + assert result["restarted_services"] == ["com.loopx.chat"] + assert result["deferred_services"] == ["com.loopx.status"] + assert len(calls) == 2 + + +@pytest.mark.parametrize("driver", ["archive_snapshot", "python_pip", "python_pipx"]) +def test_update_keeps_install_readback_but_does_not_claim_deferred_service_is_serving( + services, driver +): + runtime, _agents, _calls = services + store = ChatSessionStore(runtime) + sid = session(store) + store.create_turn(sid, client_turn_id="original", message="original work") + payload = { + "source": {"repo": "loopx-project/loopx", "ref": "f" * 40}, + "plan": {}, + "install_lifecycle": {"execution_driver": driver}, + } + passed = subprocess.CompletedProcess([], 0, json.dumps({"ok": True}), "") + with ( + mock.patch( + "loopx.self_update.run_archive_installer", return_value=(passed, {}) + ), + mock.patch("loopx.self_update.subprocess.run", return_value=passed), + ): + result = execute_update_plan(payload) + assert result["changes_applied"] and not result["ok"] + assert result["execution"]["doctor_returncode"] == 0 + assert result["execution"]["restart_status"] == "deferred" + assert result["next_action"]["kind"] == "retry_runtime_activation" + assert "f" * 40 in result["next_action"]["command"] + assert "Service activation pending" in render_update_plan_markdown(result) diff --git a/tests/test_self_update_runtime_activation.py b/tests/test_self_update_runtime_activation.py index 0e018c8154..00c6c04363 100644 --- a/tests/test_self_update_runtime_activation.py +++ b/tests/test_self_update_runtime_activation.py @@ -3,6 +3,7 @@ import json import os from pathlib import Path +import plistlib import subprocess import sys import threading @@ -810,7 +811,9 @@ def test_restart_managed_loopx_services_restarts_only_loopx_launchagents( "com.goal-harness.status.plist", "unrelated.plist", ): - (agents / name).write_text("", encoding="utf-8") + (agents / name).write_bytes(plistlib.dumps({ + "ProgramArguments": ["loopx", "--runtime-root", str(tmp_path / "runtime"), "chat"], + })) calls: list[list[str]] = [] diff --git a/tests/test_self_upgrade.py b/tests/test_self_upgrade.py index 318120ea3d..8e7ea35d0b 100644 --- a/tests/test_self_upgrade.py +++ b/tests/test_self_upgrade.py @@ -47,6 +47,17 @@ def test_app_restart_keeps_an_explicit_terminal_runtime_pin(monkeypatch): assert "--loopx-upgrade-channel=main" in launch_command +def test_upgrade_never_restarts_the_app_while_service_activation_is_deferred(): + payload = {"ok": False, "changes_applied": True, + "execution": {"install_returncode": 0, "doctor_returncode": 0, "activation_pending": True}, + "desktop_upgrade": {"status": "planned"}} + with mock.patch("loopx.cli_commands.support_control_upgrade._upgrade_desktop") as upgrade: + result = finish_upgrade(payload, timeout_seconds=10) + upgrade.assert_not_called() + assert not result["ok"] and not result["upgrade_complete"] + assert result["desktop_upgrade"]["status"] == "skipped_service_activation_pending" + + def app_fixture(tmp_path: Path, version: str = "0.0.0-main.2.1") -> Path: app = tmp_path / "LoopX.app" runtime = app / "Contents/Resources/runtime" From 3e73580d181ad3f23a0e22b0c4e4ba8ab3261e36 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 25 Oct 2026 01:06:15 +0800 Subject: [PATCH 2/2] refactor(chat): keep activation fencing with durable ingress Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/chat_ingress.py | 58 +++++++++++++++++++++++++++ loopx/chat_store.py | 54 ------------------------- loopx/runtime_activation.py | 5 ++- tests/test_chat_runtime_activation.py | 5 ++- 4 files changed, 65 insertions(+), 57 deletions(-) diff --git a/loopx/chat_ingress.py b/loopx/chat_ingress.py index a596cd0f1a..5a5c57a904 100644 --- a/loopx/chat_ingress.py +++ b/loopx/chat_ingress.py @@ -3,6 +3,8 @@ from datetime import datetime, timedelta, timezone import os +import json +from pathlib import Path import uuid from typing import Any from .chat import require_matching_message_replay, require_matching_replay @@ -13,7 +15,63 @@ SESSION_QUEUE_TTL_SECONDS = 3600 +def chat_runtime_activation_lock(runtime_root: Path, *, timeout_seconds: float | None = None): + """Fence new durable input against a managed-service runtime switch.""" + return exclusive_file_lock( + runtime_root.expanduser().resolve() / "chat" / "runtime-activation", + agent_id="loopx-chat", + operation="chat_runtime_activation", + timeout_seconds=timeout_seconds, + ) + + +def chat_runtime_activation_state(runtime_root: Path) -> dict[str, int]: + """Strict read of all accepted work; never initialize or recover a store. + + The caller holds ``chat_runtime_activation_lock`` through the service switch. + Session summaries alone omit queued input and are insufficient for this check. + Unknown or unreadable records must not authorize stopping their consumer. + """ + from .chat_store import CHAT_SESSION_SCHEMA_VERSION, CHAT_TURN_SCHEMA_VERSION, TERMINAL_TURN_STATES + + sessions = runtime_root.expanduser().resolve() / "chat" / "sessions" + counts = {"sessions": 0, "nonterminal_turns": 0} + if not sessions.exists(): + return counts + for directory in sessions.iterdir(): + if not directory.is_dir(): + continue + session = json.loads((directory / "session.json").read_text(encoding="utf-8")) + if not isinstance(session, dict) or session.get("schema_version") != CHAT_SESSION_SCHEMA_VERSION: + raise ValueError("unreadable Chat Session state") + counts["sessions"] += 1 + active_id = session.get("active_turn_id") + active_found = not active_id + turns = directory / "turns" + for path in turns.iterdir() if turns.exists() else (): + if path.suffix != ".json": + continue + turn = json.loads(path.read_text(encoding="utf-8")) + if (not isinstance(turn, dict) + or turn.get("schema_version") != CHAT_TURN_SCHEMA_VERSION + or not isinstance(turn.get("status"), str) + or turn.get("status") not in TERMINAL_TURN_STATES | { + "queued", "starting", "running", "completing", "interrupting", + }): + raise ValueError("unreadable Chat Turn state") + if turn.get("turn_id") == active_id: + active_found = True + if turn["status"] not in TERMINAL_TURN_STATES: + counts["nonterminal_turns"] += 1 + if not active_found: + raise ValueError("active Chat Turn state is missing") + return counts + + class ChatIngressStore: + def runtime_activation_lock(self): + return chat_runtime_activation_lock(self.root.parent) + def create_queued_turn( self, session_id: str, diff --git a/loopx/chat_store.py b/loopx/chat_store.py index e9941aacc0..b9268ed372 100644 --- a/loopx/chat_store.py +++ b/loopx/chat_store.py @@ -59,57 +59,6 @@ def utc_now() -> str: return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") -def chat_runtime_activation_lock(runtime_root: Path, *, timeout_seconds: float | None = None): - """Fence new durable input against a managed-service runtime switch.""" - return exclusive_file_lock( - runtime_root.expanduser().resolve() / "chat" / "runtime-activation", - agent_id="loopx-chat", - operation="chat_runtime_activation", - timeout_seconds=timeout_seconds, - ) - - -def chat_runtime_activation_state(runtime_root: Path) -> dict[str, int]: - """Strict read of all accepted work; never initialize or recover a store. - - The caller holds ``chat_runtime_activation_lock`` through the service switch. - Session summaries alone omit queued input and are insufficient for this check. - Unknown or unreadable records must not authorize stopping their consumer. - """ - sessions = runtime_root.expanduser().resolve() / "chat" / "sessions" - counts = {"sessions": 0, "nonterminal_turns": 0} - if not sessions.exists(): - return counts - for directory in sessions.iterdir(): - if not directory.is_dir(): - continue - session = json.loads((directory / "session.json").read_text(encoding="utf-8")) - if not isinstance(session, dict) or session.get("schema_version") != CHAT_SESSION_SCHEMA_VERSION: - raise ValueError("unreadable Chat Session state") - counts["sessions"] += 1 - active_id = session.get("active_turn_id") - active_found = not active_id - turns = directory / "turns" - for path in turns.iterdir() if turns.exists() else (): - if path.suffix != ".json": - continue - turn = json.loads(path.read_text(encoding="utf-8")) - if (not isinstance(turn, dict) - or turn.get("schema_version") != CHAT_TURN_SCHEMA_VERSION - or not isinstance(turn.get("status"), str) - or turn.get("status") not in TERMINAL_TURN_STATES | { - "queued", "starting", "running", "completing", "interrupting", - }): - raise ValueError("unreadable Chat Turn state") - if turn.get("turn_id") == active_id: - active_found = True - if turn["status"] not in TERMINAL_TURN_STATES: - counts["nonterminal_turns"] += 1 - if not active_found: - raise ValueError("active Chat Turn state is missing") - return counts - - def _instant(value: Any) -> datetime | None: try: return datetime.fromisoformat(str(value).replace("Z", "+00:00")) @@ -279,9 +228,6 @@ def _session_dir(self, session_id: str) -> Path: raise ValueError("session_id must be a compact opaque id") return self.sessions_root / token - def runtime_activation_lock(self): - return chat_runtime_activation_lock(self.root.parent) - def _session_path(self, session_id: str) -> Path: return self._session_dir(session_id) / "session.json" diff --git a/loopx/runtime_activation.py b/loopx/runtime_activation.py index f1983436b6..c016c1323f 100644 --- a/loopx/runtime_activation.py +++ b/loopx/runtime_activation.py @@ -108,7 +108,10 @@ 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_store import chat_runtime_activation_lock, chat_runtime_activation_state + from .chat_ingress import ( + chat_runtime_activation_lock, + chat_runtime_activation_state, + ) try: roots = sorted({_chat_runtime_root(plist) for plist in chat_plists}) diff --git a/tests/test_chat_runtime_activation.py b/tests/test_chat_runtime_activation.py index dc15d5daec..59d70c1f10 100644 --- a/tests/test_chat_runtime_activation.py +++ b/tests/test_chat_runtime_activation.py @@ -13,7 +13,8 @@ import pytest -from loopx.chat_store import ChatSessionStore, chat_runtime_activation_lock +from loopx.chat_store import ChatSessionStore +from loopx.chat_ingress import chat_runtime_activation_lock from loopx.runtime_activation import restart_services_for_runtime_activation from loopx.self_update import execute_update_plan, render_update_plan_markdown @@ -240,7 +241,7 @@ def test_contended_admission_is_not_treated_as_idle(services, monkeypatch): session(store) original = chat_runtime_activation_lock monkeypatch.setattr( - "loopx.chat_store.chat_runtime_activation_lock", + "loopx.chat_ingress.chat_runtime_activation_lock", lambda root, **_kwargs: original(root, timeout_seconds=0), ) with original(runtime):