fix(chat): preserve accepted requests during runtime activation

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.

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,
@@ -36,7 +94,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",

loopx/chat_store.py

@@ -1006,7 +1006,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",

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"))

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,24 +96,65 @@ 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:
             continue
         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_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)
+            # 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 +191,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,
+        }

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(
         [
             "",