From cea0b53d86d5360d94fd0564bddaeb0ab63b38fe Mon Sep 17 00:00:00 2001
From: huangruiteng <huangrt01@163.com>
Date: Tue, 3 Nov 2026 17:30:21 +0800
Subject: [PATCH 1/4] fix(desktop): converge managed service runtime before
 connecting

Signed-off-by: huangruiteng <huangrt01@163.com>
---
 apps/desktop/loopx-control-plane/README.md    |  54 +++
 .../loopx-control-plane/src-tauri/src/lib.rs  |  13 +
 .../src-tauri/src/maintenance.rs              |  50 +++
 .../src-tauri/src/services.rs                 |  78 +++++
 .../src-tauri/src/services_tests.rs           |  45 +++
 .../loopx-control-plane/static/boot.js        |  25 +-
 .../loopx-control-plane/static/index.html     |   2 +-
 loopx/cli_commands/support_control_update.py  |  13 +-
 loopx/managed_service_runtime.py              | 317 +++++++++++++++++
 loopx/runtime_activation.py                   |  57 +++-
 tests/test_managed_service_runtime.py         | 319 ++++++++++++++++++
 11 files changed, 952 insertions(+), 21 deletions(-)
 create mode 100644 loopx/managed_service_runtime.py
 create mode 100644 tests/test_managed_service_runtime.py

diff --git a/apps/desktop/loopx-control-plane/README.md b/apps/desktop/loopx-control-plane/README.md
index 4096e26da3..6938911f7c 100644
--- a/apps/desktop/loopx-control-plane/README.md
+++ b/apps/desktop/loopx-control-plane/README.md
@@ -194,6 +194,42 @@ cannot qualify a broken executable. Core also qualifies non-editable wheel
 fingerprints: a local reuse fence, not publisher or source attestation. Older
 CLIs without that readback can use the automatic App-owned fallback.
 
+When the App selects a different runtime, the standard macOS status and Chat
+LaunchAgents must load that executable too. Restarting an unchanged LaunchAgent
+only starts its previous runtime again. Before opening the workspace, the App
+uses Core's existing `update activate` owner to align the pair, preserving their
+registry, runtime root, accounts, tool arguments and log configuration. The
+user's CLI command and installation remain unchanged. Already matching services
+are reused without another activation command or restart.
+
+The transition holds the existing Chat ingress fence and waits for accepted
+work before changing either service. It verifies the loaded manager arguments,
+qualifies the selected executable, reloads the pair and checks both HTTP runtime
+identities. A failed start restores the original configuration; a private,
+locked handoff journal recovers interrupted transitions. Custom shell effects,
+changed manager configurations and unrelated service labels are not rewritten.
+Archive CLI updates recognize the runtime binding and move the same pair to the
+new CLI runtime; other managed services retain their existing restart behavior.
+
+版本选择与系统服务启动使用同一个运行时。已有请求会继续执行；只有请求完成且原服务
+来源可验证时才切换。切换保留原数据、账号和启动参数，并确认两个 HTTP 服务都使用
+目标版本；失败恢复原配置，中断由本地记录恢复。已匹配的服务直接复用，不反复重启。
+
+If alignment cannot complete, the startup clock stops and **Recovery & updates**
+shows the cause. **Reconnect** retries through the native supervisor without
+clearing the remembered runtime or reloading only the web page. Persistent
+alignment failures wait for that action or repair; pending Chat work keeps its
+automatic retry. A legacy Core without alignment support needs an update; it
+does not authorize falling back to terminating its services.
+
+开发诊断可使用 `loopx --format json update activate --service all
+--runtime-executable /absolute/path/to/loopx`：这是会切换服务的操作，保留原工作区和
+请求保护，不安装软件或更改终端 CLI。结果中的 `runtime_identity` 与
+`readiness_verified` 分别表示目标身份与真实 HTTP 验证；安装检查成功不等于工作区
+资源齐全。`service_runtime_route_unverified` 表示配置来源无法验证，
+`service_runtime_alignment_failed` 表示切换失败并尝试恢复，
+`service_runtime_rollback_failed` 表示仍需恢复服务。不要删除工作区数据来处理这些错误。
+
 Wheel qualification has a separate ten-second subprocess deadline because it
 starts Core, checks the TypeScript runtime and hashes package contents. The
 one-second environment/version probes remain unchanged. A slow valid wheel can
@@ -623,6 +659,24 @@ disposable registry does not isolate machine-wide usage reporting; this
 synthetic run must not contribute adoption or installation telemetry. The
 opt-out applies only to this test invocation and leaves user consent unchanged.
 
+The managed-service regression additionally exercises real launchd and both
+HTTP backends, switching an explicitly selected older Core to the current
+checkout after `npm ci && npm run build:chat` in the dashboard directory:
+
+```sh
+LOOPX_USAGE_PING=0 LOOPX_TEST_OLD_RUNTIME=/absolute/path/to/older/loopx \
+  uv run --extra test python -m pytest tests/test_managed_service_runtime.py -q -s
+```
+
+Run from the checkout root. Only test service labels are redirected to unique
+disposable labels; the manager, Core processes, HTTP and stores are real. The
+fixture uses temporary registry/account directories and ports, removes its own
+services afterwards, and never unloads the machine's standard services. It
+checks the target identity, original registry and repeated alignment without
+PID changes. Unit cases cover busy Chat, changed configuration, unsupported
+shells, failed startup/rollback and interrupted handoff. Native tests separately
+cover exact identity/readiness receipts and waiting for an explicit reconnect.
+
 This installs only in the disposable root. The second test uses an explicit
 local launch pin to route the production supervisor's two services to its own
 fresh empty registry and data directory. It checks both artifact identities,
diff --git a/apps/desktop/loopx-control-plane/src-tauri/src/lib.rs b/apps/desktop/loopx-control-plane/src-tauri/src/lib.rs
index 0cfa9583f4..636d5763db 100644
--- a/apps/desktop/loopx-control-plane/src-tauri/src/lib.rs
+++ b/apps/desktop/loopx-control-plane/src-tauri/src/lib.rs
@@ -54,6 +54,15 @@ fn maintenance_origin(url: &Url) -> String {
 // echo verbatim into the boot surface, where it names the recovery panel's
 // actionable diagnostics instead of a misleading fixed message.
 fn boot_failure_message(error: &str) -> String {
+    if matches!(
+        error,
+        "service_runtime_alignment_unavailable"
+            | "service_runtime_route_unverified"
+            | "service_runtime_alignment_failed"
+            | "service_runtime_rollback_failed"
+    ) {
+        return "服务版本尚未对齐；请展开「恢复与更新」查看原因。".to_string();
+    }
     if error == "chat_work_pending" {
         return "管家正在处理请求，完成后会自动连接工作区。".to_string();
     }
@@ -334,6 +343,10 @@ pub fn run() {
                                     | "runtime_pairing_required"
                                     | "runtime_selection_invalid"
                                     | "runtime_selection_unavailable"
+                                    | "service_runtime_alignment_unavailable"
+                                    | "service_runtime_alignment_failed"
+                                    | "service_runtime_route_unverified"
+                                    | "service_runtime_rollback_failed"
                             ) {
                                 150
                             } else {
diff --git a/apps/desktop/loopx-control-plane/src-tauri/src/maintenance.rs b/apps/desktop/loopx-control-plane/src-tauri/src/maintenance.rs
index 471d0956a5..fcda6dd47f 100644
--- a/apps/desktop/loopx-control-plane/src-tauri/src/maintenance.rs
+++ b/apps/desktop/loopx-control-plane/src-tauri/src/maintenance.rs
@@ -396,6 +396,16 @@ impl Maintenance {
         &self,
         start: impl FnOnce() -> Result<T, String>,
     ) -> Result<Option<T>, String> {
+        {
+            let snapshot = self.snapshot.lock().unwrap();
+            if snapshot["phase"] == "service_error"
+                && crate::services::runtime_recovery_required(
+                    snapshot["details"]["code"].as_str().unwrap_or(""),
+                )
+            {
+                return Ok(None);
+            }
+        }
         if self.busy.load(Ordering::Acquire)
             || self.incomplete_app_installation.load(Ordering::Acquire)
             || self.manual_failure_pending.load(Ordering::Acquire)
@@ -436,6 +446,12 @@ impl Maintenance {
                         "chat_work_pending" => "chat_work_pending",
                         "chat_state_unverified" => "chat_state_unverified",
                         "chat_service_unverified" => "chat_service_unverified",
+                        "service_runtime_alignment_unavailable" => {
+                            "service_runtime_alignment_unavailable"
+                        }
+                        "service_runtime_alignment_failed" => "service_runtime_alignment_failed",
+                        "service_runtime_route_unverified" => "service_runtime_route_unverified",
+                        "service_runtime_rollback_failed" => "service_runtime_rollback_failed",
                         _ => "service_start_failed",
                     };
                     self.publish("service_error", json!({"code":code}));
@@ -657,6 +673,17 @@ pub async fn desktop_update(
     action: String,
     channel: String,
 ) -> Result<Value, String> {
+    if action == "reconnect" {
+        let state = app.state::<Maintenance>();
+        let _guard = state.acquire()?;
+        allow_action(
+            state.snapshot.lock().unwrap()["phase"]
+                .as_str()
+                .unwrap_or(""),
+            &action,
+        )?;
+        return Ok(state.publish("connecting", json!({})));
+    }
     if cfg!(dev) || !app_update_platform_supported() {
         return Err("platform_update_not_supported".into());
     }
@@ -2536,6 +2563,29 @@ mod tests {
         }
     }
 
+    #[test]
+    fn failed_alignment_waits_for_reconnect_without_restarting_services() {
+        let state = Maintenance::default();
+        for code in [
+            "service_runtime_alignment_unavailable",
+            "service_runtime_route_unverified",
+            "service_runtime_alignment_failed",
+            "service_runtime_rollback_failed",
+        ] {
+            state.publish("service_error", json!({"code":code}));
+            for _ in 0..3 {
+                assert!(state
+                    .reconcile_services::<()>(|| panic!(
+                        "unchanged failure must not restart services"
+                    ))
+                    .unwrap()
+                    .is_none());
+            }
+            state.publish("connecting", json!({}));
+            assert_eq!(state.reconcile_services(|| Ok(42)).unwrap(), Some(42));
+        }
+    }
+
     #[test]
     fn only_verified_previous_apps_keep_the_safe_restart_promise() {
         // Verified App bundle + pairing runtime: safe-restart class, journal
diff --git a/apps/desktop/loopx-control-plane/src-tauri/src/services.rs b/apps/desktop/loopx-control-plane/src-tauri/src/services.rs
index 979e54e36a..e7b6bc3b87 100644
--- a/apps/desktop/loopx-control-plane/src-tauri/src/services.rs
+++ b/apps/desktop/loopx-control-plane/src-tauri/src/services.rs
@@ -20,6 +20,16 @@ const STARTUP_TIMEOUT: Duration = Duration::from_secs(15);
 const PROBE_TIMEOUT: Duration = Duration::from_millis(500);
 const MAX_PROBE_RESPONSE_BYTES: u64 = 1024 * 1024;
 
+pub(crate) fn runtime_recovery_required(code: &str) -> bool {
+    matches!(
+        code,
+        "service_runtime_alignment_unavailable"
+            | "service_runtime_route_unverified"
+            | "service_runtime_alignment_failed"
+            | "service_runtime_rollback_failed"
+    )
+}
+
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub enum ServiceKind {
     Status,
@@ -167,6 +177,10 @@ impl ServiceSet {
         endpoints: &ServiceEndpoints,
         progress: impl Fn(&[ServiceKind]) + Sync,
     ) -> Result<Self, ServiceError> {
+        #[cfg(target_os = "macos")]
+        if !endpoints.isolated {
+            align_managed_runtime(runtime)?;
+        }
         Self::collect(
             connect_all(
                 SERVICE_KINDS,
@@ -263,6 +277,70 @@ impl ServiceSet {
     }
 }
 
+#[cfg(target_os = "macos")]
+fn align_managed_runtime(runtime: &SelectedRuntime) -> Result<(), ServiceError> {
+    let managed = SERVICE_KINDS
+        .iter()
+        .any(|kind| platform_managed_service_declared(env::var_os("HOME").as_deref(), *kind));
+    if !managed
+        || SERVICE_KINDS
+            .iter()
+            .all(|kind| probe(*kind, runtime.identity.as_ref()) == Probe::Matching)
+    {
+        return Ok(());
+    }
+    let mut command = Command::new(&runtime.executable);
+    configure_runtime_environment(&mut command);
+    command.args([
+        "--format",
+        "json",
+        "update",
+        "activate",
+        "--service",
+        "all",
+        "--runtime-executable",
+        &runtime.executable,
+    ]);
+    managed_runtime_alignment_result(
+        timed_output_with_timeout(command, Duration::from_secs(60)),
+        runtime.identity.as_ref(),
+    )
+}
+
+#[cfg(any(target_os = "macos", test))]
+fn managed_runtime_alignment_result(
+    output: Option<std::process::Output>,
+    expected: Option<&serde_json::Value>,
+) -> Result<(), ServiceError> {
+    let payload = output
+        .as_ref()
+        .and_then(|result| serde_json::from_slice::<serde_json::Value>(&result.stdout).ok());
+    let Some(payload) = payload.filter(|value| {
+        value["schema_version"] == "loopx_runtime_service_activation_v1"
+            && value["service"] == "all"
+    }) else {
+        return Err(ServiceError("service_runtime_alignment_unavailable".into()));
+    };
+    if payload["ok"] == true
+        && payload["status"] == "aligned"
+        && payload["readiness_verified"] == true
+        && expected.is_some_and(|identity| payload.get("runtime_identity") == Some(identity))
+        && output
+            .as_ref()
+            .is_some_and(|result| result.status.success())
+    {
+        return Ok(());
+    }
+    let code = match payload["reason"].as_str() {
+        Some("chat_work_pending") => "chat_work_pending",
+        Some("chat_state_unverified") => "chat_state_unverified",
+        Some("service_runtime_route_unverified") => "service_runtime_route_unverified",
+        Some("service_runtime_rollback_failed") => "service_runtime_rollback_failed",
+        _ => "service_runtime_alignment_failed",
+    };
+    Err(ServiceError(code.into()))
+}
+
 // Ownership transfer (adopt, update handoff, stop) is exercisable without a
 // real runtime: the set owns exactly the groups a test hands it, so a reclaim
 // is observable on the same job objects production uses.
diff --git a/apps/desktop/loopx-control-plane/src-tauri/src/services_tests.rs b/apps/desktop/loopx-control-plane/src-tauri/src/services_tests.rs
index ec41b987ff..1e1c26a263 100644
--- a/apps/desktop/loopx-control-plane/src-tauri/src/services_tests.rs
+++ b/apps/desktop/loopx-control-plane/src-tauri/src/services_tests.rs
@@ -984,3 +984,48 @@ fn declared_machine_source_remains_authoritative_while_unloaded() {
         ServiceKind::Status
     ));
 }
+#[test]
+#[cfg(unix)]
+fn runtime_alignment_requires_exact_identity_and_keeps_recovery_codes() {
+    use std::os::unix::process::ExitStatusExt;
+    let identity = serde_json::json!({"release_id":"selected"});
+    let output = |payload: serde_json::Value, exit| {
+        Some(std::process::Output {
+            status: std::process::ExitStatus::from_raw(exit),
+            stdout: serde_json::to_vec(&payload).unwrap(),
+            stderr: vec![],
+        })
+    };
+    let mut payload = serde_json::json!({"schema_version":"loopx_runtime_service_activation_v1",
+        "service":"all", "ok":true, "status":"aligned", "readiness_verified":true, "runtime_identity":identity});
+    assert!(managed_runtime_alignment_result(output(payload.clone(), 0), Some(&identity)).is_ok());
+    assert!(managed_runtime_alignment_result(output(payload.clone(), 1), Some(&identity)).is_err());
+    payload["readiness_verified"] = false.into();
+    assert!(managed_runtime_alignment_result(output(payload.clone(), 0), Some(&identity)).is_err());
+    payload["readiness_verified"] = true.into();
+    payload["runtime_identity"] = serde_json::json!({"release_id":"old"});
+    assert!(managed_runtime_alignment_result(output(payload.clone(), 0), Some(&identity)).is_err());
+    for reason in [
+        "chat_work_pending",
+        "chat_state_unverified",
+        "service_runtime_route_unverified",
+        "service_runtime_alignment_failed",
+        "service_runtime_rollback_failed",
+    ] {
+        payload["ok"] = false.into();
+        payload["status"] = "deferred".into();
+        payload["reason"] = reason.into();
+        assert_eq!(
+            managed_runtime_alignment_result(output(payload.clone(), 1), Some(&identity))
+                .unwrap_err()
+                .to_string(),
+            reason
+        );
+    }
+    assert_eq!(
+        managed_runtime_alignment_result(None, Some(&identity))
+            .unwrap_err()
+            .to_string(),
+        "service_runtime_alignment_unavailable"
+    );
+}
diff --git a/apps/desktop/loopx-control-plane/static/boot.js b/apps/desktop/loopx-control-plane/static/boot.js
index 2f9eaf1264..390b56331b 100644
--- a/apps/desktop/loopx-control-plane/static/boot.js
+++ b/apps/desktop/loopx-control-plane/static/boot.js
@@ -7,6 +7,15 @@ let lastTiming = null;
 let timingObservedAt = pageStarted;
 let bootState = null;
 let observedEnvironment = null;
+const runtimeRecoveryTitles = {
+  service_runtime_alignment_unavailable: "请更新 LoopX 以连接现有服务",
+  service_runtime_route_unverified: "系统服务配置需要恢复",
+  service_runtime_alignment_failed: "服务版本切换未完成",
+  service_runtime_rollback_failed: "系统服务需要恢复",
+};
+function serviceRecoveryRequired(state) {
+  return state?.phase === "service_error" && Object.hasOwn(runtimeRecoveryTitles, state.details?.code);
+}
 function renderStartup(result) {
   const state = result?.state;
   bootState = state;
@@ -24,6 +33,7 @@ function renderStartup(result) {
     service_error: "本地服务连接失败，正在等待重试",
   };
   const serviceTitles = {
+    ...runtimeRecoveryTitles,
     chat_work_pending: "管家正在处理请求，完成后自动连接",
     chat_state_unverified: "正在核验管家任务状态，稍后自动重试",
     chat_service_unverified: "正在确认管家服务切换，稍后自动重试",
@@ -38,7 +48,7 @@ function renderStartup(result) {
   updateStartupElapsed();
 }
 function updateStartupElapsed() {
-  if (["error", "runtime_required"].includes(bootState?.phase)) {
+  if (["error", "runtime_required"].includes(bootState?.phase) || serviceRecoveryRequired(bootState)) {
     bootElapsed.textContent = "等待恢复";
     bootDetail.textContent = "启动已停止等待。可在下方恢复，不必反复重开 App。";
     return;
@@ -83,7 +93,7 @@ let channelInitialized = false;
 let runtimeExplicit = false;
 let bundledRepairAvailable = true;
 const updateWorkingPhases = ["checking","downloading","installing_app","installing_runtime","connecting"];
-document.querySelector("#retry").onclick = () => location.reload();
+document.querySelector("#retry").onclick = () => window.__TAURI__ ? void run("reconnect") : location.reload();
 channel.onchange = () => { channelInitialized = true; render({phase:"idle"}); };
 const labels = {
   idle: "检查当前通道，不会自动安装。",
@@ -101,6 +111,10 @@ const labels = {
   error: "更新未完成。请重试检查，或修复当前版本。Goal 数据不会被删除。",
 };
 const errors = {
+  service_runtime_alignment_unavailable: "当前运行时尚不支持服务版本对齐。请更新 LoopX CLI 后重新连接。现有服务未被停止。",
+  service_runtime_route_unverified: "系统服务配置已变化或使用自定义启动方式，无法自动对齐。请通过原安装方式恢复服务后重新连接。",
+  service_runtime_alignment_failed: "服务版本切换未完成，已尝试恢复原配置。可点击「重新连接」重试。",
+  service_runtime_rollback_failed: "服务版本切换与恢复均未完成。请保留诊断信息，通过原安装方式恢复系统服务。",
   chat_work_pending: "管家正在处理已有请求，完成后会自动切换服务并连接工作区。无需强退或重复发送。",
   chat_state_unverified: "暂时无法核验管家服务的任务状态。连接会自动重试；请保留诊断信息。",
   chat_service_unverified: "管家服务切换尚未确认。连接会自动重试；可保留诊断信息，或通过原安装方式更新 LoopX。",
@@ -239,7 +253,7 @@ async function run(action) {
   // Match the phase the backend publishes for each action (rollback restores
   // the previous app; restart keeps the required-restart state) instead of
   // previewing a download that is not happening.
-  render({phase: action === "check" ? "checking" : action === "repair" || action === "align_runtime" ? "installing_runtime" : action === "forget_runtime_selection" ? "connecting" : action === "rollback" ? "installing_app" : action === "restart" ? "restart_required" : "downloading"});
+  render({phase: action === "check" ? "checking" : action === "repair" || action === "align_runtime" ? "installing_runtime" : action === "forget_runtime_selection" || action === "reconnect" ? "connecting" : action === "rollback" ? "installing_app" : action === "restart" ? "restart_required" : "downloading"});
   try {
     const result = await window.__TAURI__.core.invoke("desktop_update", {action,channel:channel.value});
     statusGeneration++;
@@ -292,14 +306,15 @@ async function refresh() {
   }
 }
 function escalateFromSnapshot(state) {
-  const terminal = TERMINAL_PHASES.includes(state?.phase);
+  const terminal = TERMINAL_PHASES.includes(state?.phase) || serviceRecoveryRequired(state);
   terminalRounds = terminal ? terminalRounds + 1 : 0;
   if (terminal && terminalRounds >= ERROR_ESCALATION_ROUNDS) escalated = true;
   if (!escalated) return;
   if (terminal) {
     panel.dataset.state = "error";
     panel.setAttribute("aria-busy", "false");
-    status.textContent = `${codeText(state?.details?.code, state?.phase)} 详见下方「恢复与更新」面板，可复制诊断信息反馈。`;
+    status.textContent = serviceRecoveryRequired(state) ? runtimeRecoveryTitles[state.details.code]
+      : `${codeText(state?.details?.code, state?.phase)} 详见下方「恢复与更新」面板，可复制诊断信息反馈。`;
     bootDetail.textContent = "启动已停止等待。可在下方恢复，不必反复重开 App。";
     document.querySelector(".recovery").open = true;
   } else {
diff --git a/apps/desktop/loopx-control-plane/static/index.html b/apps/desktop/loopx-control-plane/static/index.html
index 4eebcedc2a..25b4a7b7ef 100644
--- a/apps/desktop/loopx-control-plane/static/index.html
+++ b/apps/desktop/loopx-control-plane/static/index.html
@@ -25,7 +25,7 @@ <h1 id="startup-title">LoopX</h1>
       <details class="recovery">
         <summary>恢复与更新 / Recovery & updates</summary>
         <p>重新连接不会执行任务。更新 App 与匹配的运行时可能短暂重启服务。</p>
-        <button id="retry" type="button">重新加载 / Reload</button>
+        <button id="retry" type="button">重新连接 / Reconnect</button>
         <label>更新通道 <select id="channel"><option value="stable">稳定版 / Stable</option><option value="main">main 开发版</option></select></label>
         <button id="update" type="button">检查更新 / Check for updates</button>
         <p id="update-status" role="status"></p>
diff --git a/loopx/cli_commands/support_control_update.py b/loopx/cli_commands/support_control_update.py
index 4e7c8a0b7c..6e36819a8b 100644
--- a/loopx/cli_commands/support_control_update.py
+++ b/loopx/cli_commands/support_control_update.py
@@ -108,6 +108,8 @@ def register_update_command(
         help="With update activate, retry installed managed services (default: all).")
     update_parser.add_argument("--port", type=int,
         help="With update activate --service chat, select its loopback port (default: 8767).")
+    update_parser.add_argument("--runtime-executable",
+        help="With update activate --service all, align managed services with this qualified runtime while retaining their data and account routes.")
     update_parser.add_argument(
         "--archive-url",
         help="Explicit tarball URL passed to the installer as LOOPX_ARCHIVE_URL.",
@@ -148,16 +150,19 @@ def handle_update_command(
             if upgrade or any((args.check, args.dry_run, args.execute, args.rollback,
                     args.repo, args.ref, args.archive_url, args.installed_doctor_json,
                     args.operation_id, args.source, args.allow_downgrade, args.background)):
-                raise ValueError("use update activate only with --service and --port; no installation is performed")
+                raise ValueError("use update activate only with --service, --port and --runtime-executable; no installation is performed")
             if args.port is not None and args.service != "chat":
                 raise ValueError("--port requires update activate --service chat")
+            if args.runtime_executable is not None and args.service == "chat":
+                raise ValueError("--runtime-executable requires update activate --service all")
             from ..runtime_activation import activate_runtime_services
-            payload = activate_runtime_services(service=args.service or "all", port=args.port if args.port is not None else 8767)
+            options = {"runtime_executable": args.runtime_executable} if args.runtime_executable is not None else {}
+            payload = activate_runtime_services(service=args.service or "all", port=args.port if args.port is not None else 8767, **options)
             print_payload(payload, output_format(args), lambda value:
                 f"Managed service activation: {value['status']} ({value.get('reason') or 'verify service readiness'})")
             return 0 if payload.get("ok") else 1
-        if getattr(args, "service", None) is not None or getattr(args, "port", None) is not None:
-            raise ValueError("--service and --port require update activate")
+        if any(getattr(args, name, None) is not None for name in ("service", "port", "runtime_executable")):
+            raise ValueError("--service, --port and --runtime-executable require update activate")
         if args.update_action == "status":
             if any((args.check, args.dry_run, args.execute, args.rollback, args.repo, args.ref, args.archive_url,
                     getattr(args, "source", False), getattr(args, "allow_downgrade", False), getattr(args, "background", False))):
diff --git a/loopx/managed_service_runtime.py b/loopx/managed_service_runtime.py
new file mode 100644
index 0000000000..a9be6bc1ea
--- /dev/null
+++ b/loopx/managed_service_runtime.py
@@ -0,0 +1,317 @@
+"""macOS adapter for aligning the existing service manager with a qualified Core.
+
+Keep the CLI installation, service arguments and Chat store under their existing
+owners. A restart alone cannot adopt an App-private release: launchd must load
+the new executable before it can start that release.
+"""
+
+from __future__ import annotations
+
+import copy
+import base64
+import json
+import os
+from pathlib import Path
+import plistlib
+import shlex
+import subprocess
+import time
+import urllib.error
+import urllib.request
+
+from .file_lock import exclusive_file_lock, write_bytes_atomic, write_json_atomic
+from . import runtime_activation as activation
+
+
+def _literal_arguments(command: str) -> list[str]:
+    # Only the installer's final literal exec is editable. Never evaluate a
+    # custom shell expansion, pipeline, redirection or command substitution.
+    quote = None
+    escaped = False
+    for char in command:
+        if escaped:
+            escaped = False
+        elif char == "\\" and quote != "'":
+            escaped = True
+        elif char == quote:
+            quote = None
+        elif char in "\"'" and quote is None:
+            quote = char
+        elif (char in "$`" and quote != "'") or (quote is None and char in ";&|()<>\n"):
+            raise ValueError("service_runtime_route_unverified")
+    words = shlex.split(command)
+    if not words or quote or escaped:
+        raise ValueError("service_runtime_route_unverified")
+    return words
+
+
+def _retarget(payload: dict, executable: str, python: str) -> dict:
+    result = copy.deepcopy(payload)
+    if "Program" in result:
+        raise ValueError("service_runtime_route_unverified")
+    args = result.get("ProgramArguments")
+    if not isinstance(args, list) or not args or not all(isinstance(a, str) for a in args):
+        raise ValueError("service_runtime_route_unverified")
+    if Path(args[0]).name in {"sh", "bash", "zsh"}:
+        if len(args) != 3 or args[1] != "-c":
+            raise ValueError("service_runtime_route_unverified")
+        prefix, separator, command = args[2].rpartition("; exec ")
+        if not separator:
+            if not args[2].startswith("exec "):
+                raise ValueError("service_runtime_route_unverified")
+            prefix, command = "", args[2][5:]
+        # A split inside a quoted argument is not the generated exec boundary.
+        shlex.split(prefix)
+        words = _literal_arguments(command)
+        if words[:5] == ["/usr/bin/env", "-u", "PYTHONPATH", "-u", "LOOPX_RELEASE_ROOT"]:
+            words = [words[0], *words[5:]]
+        if words[0] == "/usr/bin/env" and len(words) > 2 and words[1].startswith("LOOPX_PYTHON="):
+            words = words[2:]
+        if not words or not Path(words[0]).is_absolute():
+            raise ValueError("service_runtime_route_unverified")
+        launch = ["/usr/bin/env", "-u", "PYTHONPATH", "-u", "LOOPX_RELEASE_ROOT",
+            f"LOOPX_PYTHON={python}", executable, *words[1:]]
+        result["ProgramArguments"] = [args[0], "-c", (prefix + "; " if prefix else "") + "exec " + shlex.join(launch)]
+    else:
+        if not Path(args[0]).is_absolute() or "Program" in result:
+            raise ValueError("service_runtime_route_unverified")
+        result["ProgramArguments"] = [executable, *args[1:]]
+        result.setdefault("EnvironmentVariables", {})["LOOPX_PYTHON"] = python
+        result["EnvironmentVariables"].pop("PYTHONPATH", None)
+        result["EnvironmentVariables"].pop("LOOPX_RELEASE_ROOT", None)
+    result["LoopXRuntimeBinding"] = {"schema_version": "loopx_managed_runtime_binding_v1"}
+    return result
+
+
+def _launchctl(*args: str) -> None:
+    result = subprocess.run(["launchctl", *args], capture_output=True, timeout=5)
+    if result.returncode:
+        raise RuntimeError("service_runtime_alignment_failed")
+
+
+def has_runtime_binding() -> bool:
+    for kind in ("status", "chat"):
+        path = Path.home() / "Library/LaunchAgents" / f"com.loopx.{kind}.plist"
+        if not path.exists():
+            continue
+        payload = plistlib.loads(path.read_bytes())
+        if not isinstance(payload, dict):
+            raise ValueError("service_runtime_route_unverified")
+        if payload.get("LoopXRuntimeBinding") == {"schema_version": "loopx_managed_runtime_binding_v1"}:
+            return True
+    return False
+
+
+def _endpoint(payload: dict) -> int:
+    arguments = payload["ProgramArguments"]
+    if Path(arguments[0]).name in {"sh", "bash", "zsh"}:
+        command = arguments[2].rpartition("; exec ")[2]
+        if arguments[2].startswith("exec ") and "; exec " not in arguments[2]:
+            command = arguments[2][5:]
+        arguments = _literal_arguments(command)
+    def option(flag):
+        values = [word.split("=", 1)[1] if word.startswith(flag + "=")
+            else arguments[i + 1] if i + 1 < len(arguments) else ""
+            for i, word in enumerate(arguments) if word == flag or word.startswith(flag + "=")]
+        if len(values) != 1:
+            raise ValueError("service_runtime_route_unverified")
+        return values[0]
+    if option("--host") != "127.0.0.1":
+        raise ValueError("service_runtime_route_unverified")
+    try:
+        port = int(option("--port"))
+    except ValueError as exc:
+        raise ValueError("service_runtime_route_unverified") from exc
+    if not 1 <= port <= 65535:
+        raise ValueError("service_runtime_route_unverified")
+    return port
+
+
+def _verify_services(services: dict[Path, int], identity: dict) -> None:
+    # A working CLI does not prove its Chat assets can serve. Read back the
+    # exact HTTP runtime before committing the handoff; a failed start restores
+    # the original manager configuration instead of leaving the App spinning.
+    pending = dict(services)
+    opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
+    deadline = time.monotonic() + 15
+    while pending and time.monotonic() < deadline:
+        for path, port in list(pending.items()):
+            status = path.stem.endswith(".status")
+            url = f"http://127.0.0.1:{port}" + ("/?readiness=1" if status else "/api/chat/readiness")
+            try:
+                try:
+                    response = opener.open(url, timeout=0.5)
+                except urllib.error.HTTPError as exc:
+                    if status or exc.code != 404:
+                        raise
+                    response = opener.open(f"http://127.0.0.1:{port}/api/chat/capabilities", timeout=0.5)
+                with response:
+                    body = response.read(1024 * 1024 + 1)
+                    if len(body) > 1024 * 1024:
+                        continue
+                    observed = json.loads(body)
+                if not isinstance(observed, dict):
+                    continue
+                readiness = observed.get("readiness")
+                ready = (observed.get("source") == "serve-status"
+                    and isinstance(readiness, dict) and readiness.get("state") == "ready") if status else (
+                    observed.get("schema_version") == "loopx_chat_capabilities_v1" or
+                    (observed.get("schema_version") == "loopx_chat_readiness_v1"
+                        and observed.get("ok") is True and observed.get("state") == "ready"))
+                if ready and observed.get("runtime_identity") == identity:
+                    pending.pop(path)
+            except (OSError, ValueError):
+                pass
+        if pending:
+            time.sleep(0.1)
+    if pending:
+        raise RuntimeError("service_runtime_alignment_failed")
+
+
+def align_managed_runtime(executable: str) -> dict:
+    """Align the standard desktop services once, with rollback and Chat fencing.
+
+    This is an explicit activation operation, not an installer or a change to
+    the user's default CLI. Unsupported custom launchers remain untouched.
+    """
+    directory = Path.home() / "Library/LaunchAgents"
+    directory.mkdir(parents=True, exist_ok=True)
+    journal = directory / ".loopx-runtime-activation.json"
+    with exclusive_file_lock(directory / ".loopx-runtime-activation.lock", timeout_seconds=2):
+        _recover_alignment(journal)
+        return _align_managed_runtime(executable, journal)
+
+
+def _recover_alignment(journal: Path) -> None:
+    if not journal.exists():
+        return
+    saved = json.loads(journal.read_text())
+    if not isinstance(saved, dict) or saved.get("schema_version") != "loopx_managed_runtime_handoff_v1":
+        raise ValueError("service_runtime_route_unverified")
+    entries = saved.get("services")
+    if not isinstance(entries, dict) or not entries or not set(entries) <= {"com.loopx.status", "com.loopx.chat"}:
+        raise ValueError("service_runtime_route_unverified")
+    originals, candidates = {}, {}
+    for label, item in entries.items():
+        if (not isinstance(item, dict) or set(item) != {"original", "candidate"}
+                or not all(isinstance(value, str) for value in item.values())):
+            raise ValueError("service_runtime_route_unverified")
+        path = journal.parent / f"{label}.plist"
+        originals[path] = base64.b64decode(item["original"], validate=True)
+        candidates[path] = base64.b64decode(item["candidate"], validate=True)
+        if path.is_symlink() or path.read_bytes() not in {originals[path], candidates[path]}:
+            raise ValueError("service_runtime_route_unverified")
+    chat = journal.parent / "com.loopx.chat.plist"
+    with activation.managed_chat_activation_guard([chat] if chat.exists() else []):
+        domain = f"gui/{os.getuid()}"
+        loaded = []
+        for path in originals:
+            old, _ = activation._managed_service_state(path, payload=plistlib.loads(originals[path]))
+            new, _ = activation._managed_service_state(path, payload=plistlib.loads(candidates[path]))
+            if old is False and new is False:
+                raise ValueError("service_runtime_route_unverified")
+            if old or new:
+                loaded.append(path)
+        try:
+            for path in loaded:
+                _launchctl("bootout", f"{domain}/{path.stem}")
+            for path, content in originals.items():
+                write_bytes_atomic(path, content, preserve_mode=True, fsync_directory=True)
+            for path in originals:
+                _launchctl("bootstrap", domain, str(path))
+        except (OSError, RuntimeError, subprocess.SubprocessError) as exc:
+            raise RuntimeError("service_runtime_rollback_failed") from exc
+        journal.unlink()
+
+
+def _align_managed_runtime(executable: str, journal: Path) -> dict:
+    target = Path(executable)
+    if not target.is_absolute() or not target.is_file() or not os.access(target, os.X_OK):
+        raise ValueError("service_runtime_unverified")
+    environment = {k: v for k, v in os.environ.items()
+        if k not in {"LOOPX_PYTHON", "PYTHONPATH", "LOOPX_RELEASE_ROOT"}}
+    doctor = subprocess.run([str(target), "--format", "json", "doctor", "--installation-only"],
+        env=environment, capture_output=True, text=True, timeout=15)
+    qualified = json.loads(doctor.stdout)
+    if not isinstance(qualified, dict):
+        raise ValueError("service_runtime_unverified")
+    identity = qualified.get("service_runtime_identity")
+    python_info = qualified.get("python")
+    python = python_info.get("executable") if isinstance(python_info, dict) else None
+    if (doctor.returncode or qualified.get("ok") is not True
+            or qualified.get("scope") != "installation_only"
+            or not isinstance(identity, dict) or identity.get("schema_version") != "loopx_runtime_identity_v1"
+            or not isinstance(python, str) or not Path(python).is_absolute()):
+        raise ValueError("service_runtime_unverified")
+
+    plists = [Path.home() / "Library/LaunchAgents" / f"com.loopx.{kind}.plist"
+        for kind in ("status", "chat")]
+    originals, updates, pids, endpoints = {}, {}, {}, {}
+    for plist in plists:
+        if not plist.exists():
+            continue
+        if plist.is_symlink():
+            raise ValueError("service_runtime_route_unverified")
+        original = plist.read_bytes()
+        payload = plistlib.loads(original)
+        if not isinstance(payload, dict) or payload.get("Label") != plist.stem:
+            raise ValueError("service_runtime_route_unverified")
+        loaded, pid = activation._managed_service_state(plist)
+        if not loaded:
+            raise ValueError("service_runtime_route_unverified")
+        updated = _retarget(payload, str(target), python)
+        endpoints[plist] = _endpoint(updated)
+        if updated != payload:
+            originals[plist] = original
+            updates[plist] = plistlib.dumps(updated)
+            pids[plist] = pid
+    if not updates:
+        _verify_services(endpoints, identity)
+        return {"status": "aligned", "runtime_identity": identity, "restarted_services": [],
+            "readiness_verified": bool(endpoints)}
+
+    chat = [p for p in plists if p.exists() and p.stem.endswith(".chat")]
+    stopped, written, restarted = [], [], []
+    domain = f"gui/{os.getuid()}"
+    with activation.managed_chat_activation_guard(chat):
+        # Fence the complete pair before the first signal or persistent edit.
+        for plist in updates:
+            if (plist.read_bytes() != originals[plist]
+                    or activation._managed_service_state(plist) != (True, pids[plist])):
+                raise ValueError("service_runtime_route_unverified")
+        write_json_atomic(journal, {"schema_version": "loopx_managed_runtime_handoff_v1",
+            "services": {p.stem: {"original": base64.b64encode(originals[p]).decode(),
+                "candidate": base64.b64encode(updates[p]).decode()} for p in updates}},
+            mode=0o600, fsync_directory=True)
+        try:
+            for plist in updates:
+                _launchctl("bootout", f"{domain}/{plist.stem}")
+                stopped.append(plist)
+            for plist, content in updates.items():
+                if plist.read_bytes() != originals[plist]:
+                    raise RuntimeError("service_runtime_route_unverified")
+                write_bytes_atomic(plist, content, preserve_mode=True, fsync_directory=True)
+                written.append(plist)
+            for plist in updates:
+                _launchctl("bootstrap", domain, str(plist))
+                restarted.append(plist)
+            _verify_services(endpoints, identity)
+        except (OSError, RuntimeError, subprocess.SubprocessError):
+            # Restore the manager's original source as well as its file. Keep
+            # rollback failure explicit; never declare HTTP readiness here.
+            try:
+                for plist in restarted:
+                    _launchctl("bootout", f"{domain}/{plist.stem}")
+                for plist in written:
+                    if plist.read_bytes() != updates[plist]:
+                        raise RuntimeError("service_runtime_route_unverified")
+                    write_bytes_atomic(plist, originals[plist], preserve_mode=True, fsync_directory=True)
+                for plist in stopped:
+                    _launchctl("bootstrap", domain, str(plist))
+            except (OSError, RuntimeError, subprocess.SubprocessError) as exc:
+                raise RuntimeError("service_runtime_rollback_failed") from exc
+            journal.unlink()
+            raise RuntimeError("service_runtime_alignment_failed")
+        journal.unlink()
+    return {"status": "aligned", "runtime_identity": identity, "readiness_verified": True,
+        "restarted_services": [p.stem for p in restarted]}
diff --git a/loopx/runtime_activation.py b/loopx/runtime_activation.py
index cfb404b3db..35d7e88bf2 100644
--- a/loopx/runtime_activation.py
+++ b/loopx/runtime_activation.py
@@ -93,24 +93,29 @@ def _managed_service_plists() -> list[Path]:
     ]
 
 
-def _managed_service_pid(plist: Path, *, label: str | None = None) -> int | None:
+def _managed_service_state(plist: Path, *, label: str | None = None,
+        payload: dict[str, Any] | None = None) -> tuple[bool | None, int | None]:
     result = subprocess.run(
         ["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:
-        return None
-    payload = plistlib.loads(plist.read_bytes())
+        return None, None
+    payload = payload if payload is not None else plistlib.loads(plist.read_bytes())
     if not isinstance(payload, dict):
-        return None
+        return False, 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.
+        return False, 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
+    return True, int(pids[0]) if len(pids) == 1 else None
+
+
+def _managed_service_pid(plist: Path, *, label: str | None = None) -> int | None:
+    return _managed_service_state(plist, label=label)[1]
 
 
 @contextmanager
@@ -222,7 +227,8 @@ def replace_managed_loopx_chat(
         raise RuntimeActivationDeferred("chat_state_unverified", labels) from exc
 
 
-def activate_runtime_services(*, service: str = "all", port: int = 8767) -> dict[str, Any]:
+def activate_runtime_services(*, service: str = "all", port: int = 8767,
+        runtime_executable: str | None = None) -> dict[str, Any]:
     """Retry installed-service adoption without reinstalling or running work.
 
     The desktop Chat caller uses the existing verified listener/manager/store
@@ -239,6 +245,11 @@ def activate_runtime_services(*, service: str = "all", port: int = 8767) -> dict
         return {**payload, "ok": False, "status": "unsupported",
             "reason": "managed_service_activation_unavailable"}
     try:
+        if runtime_executable is not None:
+            if service != "all":
+                raise ValueError("service_runtime_route_unverified")
+            from .managed_service_runtime import align_managed_runtime
+            return {**payload, "ok": True, **align_managed_runtime(runtime_executable)}
         if service == "chat":
             observed = replace_managed_loopx_chat("127.0.0.1", port, restart_managed=True)
             status = "not_running" if observed == "unavailable" else "restart_requested"
@@ -249,13 +260,17 @@ def activate_runtime_services(*, service: str = "all", port: int = 8767) -> dict
         return {**payload, "ok": True, "status": status}
     except RuntimeActivationDeferred as exc:
         return {**payload, "ok": False, "status": "deferred", "reason": str(exc)}
-    except (RuntimeActivationFailed, OSError, ValueError, RuntimeError):
+    except (RuntimeActivationFailed, OSError, ValueError, RuntimeError, subprocess.SubprocessError) as exc:
         # Public diagnostics carry a coarse cause, never manager commands,
         # local paths or the contents of accepted messages.
-        return {**payload, "ok": False, "status": "failed", "reason": "chat_service_unverified"}
+        reason = str(exc) if str(exc) in {
+            "service_runtime_unverified", "service_runtime_route_unverified",
+            "service_runtime_alignment_failed", "service_runtime_rollback_failed",
+        } else "chat_service_unverified"
+        return {**payload, "ok": False, "status": "failed", "reason": reason}
 
 
-def restart_managed_loopx_services() -> list[str]:
+def restart_managed_loopx_services(*, plists: list[Path] | None = None) -> list[str]:
     """Switch managed macOS services only with no accepted Chat work outstanding.
 
     After ``loopx update`` replaces the installed release, running status/chat
@@ -267,7 +282,7 @@ def restart_managed_loopx_services() -> list[str]:
         return []
     labels: list[str] = []
     chat_plists: list[Path] = []
-    for plist in _managed_service_plists():
+    for plist in _managed_service_plists() if plists is None else plists:
         stem = plist.stem.lower()
         labels.append(plist.stem)
         if stem.endswith(".chat"):
@@ -347,6 +362,23 @@ def restart_services_for_runtime_activation(
             "restart_status": "skipped_custom_installation",
         }
     try:
+        from .managed_service_runtime import align_managed_runtime, has_runtime_binding
+        if sys.platform == "darwin" and has_runtime_binding():
+            # Preserve the existing pending-work result even when the new
+            # installation cannot yet be qualified. Alignment rechecks under
+            # its own lock before any manager mutation.
+            with managed_chat_activation_guard([p for p in _managed_service_plists() if p.stem.endswith(".chat")]):
+                pass
+            if installed_bin is None:
+                from .doctor import current_script_invocation_path
+                installed_bin = current_script_invocation_path()
+            if installed_bin is None:
+                raise RuntimeActivationDeferred("service_runtime_unverified", ["com.loopx.status", "com.loopx.chat"])
+            aligned = align_managed_runtime(str(installed_bin))
+            other = [p for p in _managed_service_plists() if p.stem not in {"com.loopx.status", "com.loopx.chat"}]
+            restarted = restart_managed_loopx_services(plists=other) if other else []
+            return {"restarted_services": [*aligned["restarted_services"], *restarted],
+                "restart_status": "restarted", "activation_pending": False}
         return {
             "restarted_services": restart_managed_loopx_services(),
             "restart_status": "restarted",
@@ -368,3 +400,6 @@ def restart_services_for_runtime_activation(
             "restart_reason": str(exc),
             "activation_pending": True,
         }
+    except (OSError, ValueError, RuntimeError, subprocess.SubprocessError):
+        return {"restarted_services": [], "restart_status": "failed",
+            "restart_reason": "service_runtime_alignment_failed", "activation_pending": True}
diff --git a/tests/test_managed_service_runtime.py b/tests/test_managed_service_runtime.py
new file mode 100644
index 0000000000..d0bc9623e2
--- /dev/null
+++ b/tests/test_managed_service_runtime.py
@@ -0,0 +1,319 @@
+"""Service alignment preserves the real Chat admission boundary and manager route."""
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+import plistlib
+import shlex
+import subprocess
+import sys
+import socket
+import time
+import urllib.request
+import uuid
+
+import pytest
+
+from loopx import runtime_activation as activation
+from loopx.managed_service_runtime import _retarget, align_managed_runtime
+from loopx.chat_store import ChatSessionStore
+
+
+@pytest.fixture
+def managed(tmp_path, monkeypatch):
+    monkeypatch.setattr("loopx.managed_service_runtime._verify_services", lambda *_: None)
+    monkeypatch.setattr(Path, "home", staticmethod(lambda: tmp_path))
+    root = tmp_path / "original data"
+    agents = tmp_path / "Library/LaunchAgents"
+    agents.mkdir(parents=True)
+    target = tmp_path / "new runtime's $literal" / "loopx"
+    target.parent.mkdir()
+    identity = {"schema_version": "loopx_runtime_identity_v1", "package_version": "2.2.0",
+        "release_id": "new", "source_revision": "new-revision"}
+    doctor = {"ok": True, "scope": "installation_only", "service_runtime_identity": identity,
+        "python": {"executable": sys.executable}}
+    target.write_text(f"#!{sys.executable}\nimport json,os,sys\n"
+        f"print({json.dumps(json.dumps(doctor))} if 'doctor' in sys.argv else json.dumps({{'args':sys.argv[1:],'home':os.getenv('CODEX_HOME'),'python':os.getenv('LOOPX_PYTHON')}}))\n")
+    target.chmod(0o700)
+    loaded, calls = {}, []
+    for kind in ("status", "chat"):
+        label = f"com.loopx.{kind}"
+        args = ["/old/bin/loopx", "--registry", str(tmp_path / "registry"),
+            "--runtime-root", str(root), "chat" if kind == "chat" else "serve-status",
+            "--host", "127.0.0.1", "--port", "8767" if kind == "chat" else "8766"]
+        shell = "export CODEX_HOME='/original account'; export LOOPX_PYTHON='/old/python'; exec " + shlex.join(args)
+        payload = {"Label": label, "ProgramArguments": ["/bin/sh", "-c", shell],
+            "KeepAlive": True, "ThrottleInterval": 10, "StandardErrorPath": str(tmp_path / f"{kind}.err")}
+        (agents / f"{label}.plist").write_bytes(plistlib.dumps(payload))
+        loaded[label] = payload
+    original_run = subprocess.run
+    fail = []
+
+    def launchctl(args, **kwargs):
+        if args[0] != "launchctl":
+            return original_run(args, **kwargs)
+        verb = args[1]
+        label = Path(args[-1]).stem if verb == "bootstrap" else args[-1].split("/")[-1]
+        if verb == "print":
+            text = "arguments = {\n" + "\n".join(loaded[label]["ProgramArguments"]) + "\n}\npid = 900\n" if label in loaded else ""
+            return subprocess.CompletedProcess(args, 0 if label in loaded else 1, text, "")
+        calls.append((verb, label))
+        if fail and (verb, label) == fail[0]:
+            fail.pop(0)
+            return subprocess.CompletedProcess(args, 1, b"", b"fixture failure")
+        if verb == "bootout":
+            loaded.pop(label)
+        elif verb == "bootstrap":
+            loaded[label] = plistlib.loads(Path(args[-1]).read_bytes())
+        return subprocess.CompletedProcess(args, 0, b"", b"")
+
+    monkeypatch.setattr(activation.subprocess, "run", launchctl)
+    return root, agents, target, loaded, calls, fail
+
+
+def test_pair_adopts_target_without_changing_routes_or_cli_and_is_idempotent(managed):
+    root, agents, target, loaded, calls, _ = managed
+    before = {p.name: plistlib.loads(p.read_bytes()) for p in agents.glob("*.plist")}
+    result = align_managed_runtime(str(target))
+    assert result["status"] == "aligned"
+    assert calls == [("bootout", "com.loopx.status"), ("bootout", "com.loopx.chat"),
+        ("bootstrap", "com.loopx.status"), ("bootstrap", "com.loopx.chat")]
+    for name, payload in loaded.items():
+        assert payload["LoopXRuntimeBinding"]["schema_version"] == "loopx_managed_runtime_binding_v1"
+        assert {k: v for k, v in payload.items() if k not in {"ProgramArguments", "LoopXRuntimeBinding"}} == {
+            k: v for k, v in before[f"{name}.plist"].items() if k != "ProgramArguments"}
+        observed = json.loads(subprocess.check_output(payload["ProgramArguments"], text=True))
+        assert observed["home"] == "/original account"
+        assert observed["python"] == sys.executable
+        assert observed["args"][observed["args"].index("--runtime-root") + 1] == str(root)
+    calls.clear()
+    assert align_managed_runtime(str(target))["restarted_services"] == []
+    assert calls == []
+
+
+def test_pending_chat_prevents_both_service_changes(managed):
+    root, agents, target, loaded, calls, _ = managed
+    store = ChatSessionStore(root)
+    sid = store.create_session(goal_id="fixture", agent_id="fixture", adapter_kind="fake",
+        upstream_thread_id="fixture")["session_id"]
+    turn, _ = store.create_turn(sid, client_turn_id="pending", message="Keep accepted work.")
+    before = {p: p.read_bytes() for p in agents.glob("*.plist")}
+    with pytest.raises(activation.RuntimeActivationDeferred, match="chat_work_pending"):
+        align_managed_runtime(str(target))
+    assert not calls and len(loaded) == 2
+    assert {p: p.read_bytes() for p in before} == before
+    assert store.load_turn(sid, turn["turn_id"])["status"] == "queued"
+
+
+def test_failed_second_bootstrap_restores_original_loaded_pair(managed):
+    _, agents, target, loaded, calls, fail = managed
+    before = {p: p.read_bytes() for p in agents.glob("*.plist")}
+    fail.append(("bootstrap", "com.loopx.chat"))
+    with pytest.raises(RuntimeError, match="service_runtime_alignment_failed"):
+        align_managed_runtime(str(target))
+    assert {p: p.read_bytes() for p in before} == before
+    assert loaded == {p.stem: plistlib.loads(content) for p, content in before.items()}
+    assert calls[-2:] == [("bootstrap", "com.loopx.status"), ("bootstrap", "com.loopx.chat")]
+
+
+def test_startup_readback_failure_restores_both_original_services(managed, monkeypatch):
+    _, agents, target, loaded, _, _ = managed
+    before = {p: p.read_bytes() for p in agents.glob("*.plist")}
+    def unavailable(*_):
+        raise RuntimeError("service_runtime_alignment_failed")
+    monkeypatch.setattr("loopx.managed_service_runtime._verify_services", unavailable)
+    with pytest.raises(RuntimeError, match="service_runtime_alignment_failed"):
+        align_managed_runtime(str(target))
+    assert {p: p.read_bytes() for p in before} == before
+    assert loaded == {p.stem: plistlib.loads(content) for p, content in before.items()}
+
+
+def test_process_interruption_recovers_before_reapplying(managed, monkeypatch):
+    from loopx import managed_service_runtime as runtime
+
+    _, agents, target, loaded, calls, _ = managed
+    original_write = runtime.write_bytes_atomic
+    interrupted = False
+
+    def crash(path, content, **kwargs):
+        nonlocal interrupted
+        original_write(path, content, **kwargs)
+        if path.suffix == ".plist" and not interrupted:
+            interrupted = True
+            raise KeyboardInterrupt("fixture process interruption")
+
+    monkeypatch.setattr(runtime, "write_bytes_atomic", crash)
+    with pytest.raises(KeyboardInterrupt):
+        align_managed_runtime(str(target))
+    assert (agents / ".loopx-runtime-activation.json").exists()
+    assert not loaded
+    monkeypatch.setattr(runtime, "write_bytes_atomic", original_write)
+    calls.clear()
+    assert align_managed_runtime(str(target))["status"] == "aligned"
+    assert calls[:2] == [("bootstrap", "com.loopx.status"), ("bootstrap", "com.loopx.chat")]
+    assert len(loaded) == 2
+    assert not (agents / ".loopx-runtime-activation.json").exists()
+
+
+def test_edited_manager_source_prevents_signals_and_writes(managed):
+    _, agents, target, loaded, calls, _ = managed
+    changed = agents / "com.loopx.chat.plist"
+    payload = plistlib.loads(changed.read_bytes())
+    payload["ProgramArguments"] = ["/some/other/loopx", "chat"]
+    changed.write_bytes(plistlib.dumps(payload))
+    before = {p: p.read_bytes() for p in agents.glob("*.plist")}
+    with pytest.raises(ValueError, match="service_runtime_route_unverified"):
+        align_managed_runtime(str(target))
+    assert not calls and len(loaded) == 2
+    assert {p: p.read_bytes() for p in before} == before
+
+
+@pytest.mark.parametrize("saved", [[], {"schema_version": "unknown"}])
+def test_damaged_handoff_stops_before_any_service_mutation(managed, saved, monkeypatch):
+    _, agents, target, loaded, calls, _ = managed
+    monkeypatch.setattr(activation.sys, "platform", "darwin")
+    journal = agents / ".loopx-runtime-activation.json"
+    journal.write_text(json.dumps(saved))
+    result = activation.activate_runtime_services(runtime_executable=str(target))
+    assert result["ok"] is False
+    assert result["reason"] == "service_runtime_route_unverified"
+    assert not calls and len(loaded) == 2
+    assert journal.exists()
+
+
+def test_cli_update_moves_bound_pair_and_still_restarts_other_services(managed, monkeypatch):
+    _, agents, target, loaded, calls, _ = managed
+    align_managed_runtime(str(target))
+    installed = agents.parents[1] / ".local/bin/loopx"
+    installed.parent.mkdir(parents=True)
+    installed.symlink_to(target)
+    other = agents / "com.loopx.project.status.plist"
+    other.write_bytes(plistlib.dumps({"Label": other.stem, "ProgramArguments": ["/project/loopx", "serve-status"]}))
+    loaded[other.stem] = plistlib.loads(other.read_bytes())
+    monkeypatch.setattr(activation.sys, "platform", "darwin")
+    calls.clear()
+    result = activation.restart_services_for_runtime_activation(
+        changes_applied=True, runtime_ready=True, installed_bin=installed)
+    assert result["activation_pending"] is False
+    assert set(result["restarted_services"]) == {"com.loopx.status", "com.loopx.chat", other.stem}
+    assert calls[-1] == ("kickstart", other.stem)
+    assert loaded[other.stem]["ProgramArguments"] == ["/project/loopx", "serve-status"]
+    for kind in ("status", "chat"):
+        assert str(installed) in loaded[f"com.loopx.{kind}"]["ProgramArguments"][2]
+
+
+@pytest.mark.parametrize("shell", ["exec /old/loopx $(touch /tmp/no)",
+    "exec /old/loopx chat | cat", "exec /old/loopx chat > out", "echo exec /old/loopx chat",
+    'exec /old/loopx "$RUNTIME"', "exec /old/loopx `whoami`", "exec /old/loopx; echo done"])
+def test_custom_shell_effects_are_not_rewritten(shell):
+    with pytest.raises(ValueError):
+        _retarget({"ProgramArguments": ["/bin/sh", "-c", shell]}, "/new/loopx", "/new/python")
+
+
+def test_direct_launcher_retains_environment_and_arguments():
+    original = {"ProgramArguments": ["/old/loopx", "chat", "--runtime-root", "/original"],
+        "EnvironmentVariables": {"CODEX_HOME": "/account"}}
+    updated = _retarget(original, "/new/loopx", "/new/python")
+    assert updated["ProgramArguments"] == ["/new/loopx", "chat", "--runtime-root", "/original"]
+    assert updated["EnvironmentVariables"] == {"CODEX_HOME": "/account", "LOOPX_PYTHON": "/new/python"}
+    assert "LOOPX_PYTHON" not in original["EnvironmentVariables"]
+
+
+@pytest.mark.skipif(sys.platform != "darwin" or not os.environ.get("LOOPX_TEST_OLD_RUNTIME"),
+    reason="opt-in real launchd test requires a selected old Core; see Desktop README")
+def test_real_manager_switches_core_and_preserves_original_workspace(tmp_path, monkeypatch):
+    """Only labels are redirected; manager, processes, HTTP and stores are real."""
+    old = Path(os.environ["LOOPX_TEST_OLD_RUNTIME"]).resolve()
+    target = Path(sys.executable).parent / "loopx"
+    assert old.is_file() and target.is_file()
+    home = tmp_path / "home"
+    agents = home / "Library/LaunchAgents"
+    agents.mkdir(parents=True)
+    root = tmp_path / "original workspace"
+    root.mkdir()
+    registry = root / "registry.json"
+    registry.write_text('{"goals":[]}')
+    monkeypatch.setattr(Path, "home", staticmethod(lambda: home))
+    alias = f"com.loopx.qa.{uuid.uuid4().hex}"
+    domain = f"gui/{os.getuid()}"
+    original_run = subprocess.run
+    ports = {}
+    for kind in ("status", "chat"):
+        with socket.socket() as listener:
+            listener.bind(("127.0.0.1", 0))
+            ports[kind] = listener.getsockname()[1]
+    assert ports["status"] != ports["chat"]
+
+    def launchctl(args, **kwargs):
+        if args[0] != "launchctl":
+            return original_run(args, **kwargs)
+        args = list(args)
+        if args[1] in {"print", "bootout", "kickstart"}:
+            for kind in ("status", "chat"):
+                if args[-1] == f"{domain}/com.loopx.{kind}":
+                    args[-1] = f"{domain}/{alias}.{kind}"
+        if args[1] == "bootstrap":
+            source = Path(args[-1])
+            kind = source.stem.rsplit(".", 1)[-1]
+            payload = plistlib.loads(source.read_bytes())
+            payload["Label"] = f"{alias}.{kind}"
+            mapped = tmp_path / f"{alias}.{kind}.plist"
+            mapped.write_bytes(plistlib.dumps(payload))
+            args[-1] = str(mapped)
+        return original_run(args, **kwargs)
+
+    monkeypatch.setattr(activation.subprocess, "run", launchctl)
+    for kind in ("status", "chat"):
+        args = [str(old), "--registry", str(registry), "--runtime-root", str(root),
+            "serve-status" if kind == "status" else "chat", "--host", "127.0.0.1",
+            "--port", str(ports[kind]), "--scan-path", str(root)]
+        if kind == "chat":
+            args += ["--no-open"]
+        payload = {"Label": f"com.loopx.{kind}", "ProgramArguments": args, "RunAtLoad": True,
+            "KeepAlive": True, "EnvironmentVariables": {"HOME": str(home), "LOOPX_USAGE_PING": "0",
+                "CODEX_HOME": str(home / ".codex"), "PATH": os.environ.get("PATH", "")},
+            "StandardErrorPath": str(tmp_path / f"{kind}.log")}
+        (agents / f"com.loopx.{kind}.plist").write_bytes(plistlib.dumps(payload))
+
+    def read(kind, path):
+        with urllib.request.urlopen(f"http://127.0.0.1:{ports[kind]}{path}", timeout=2) as response:
+            return json.load(response)
+
+    def ready():
+        deadline = time.monotonic() + 30
+        while time.monotonic() < deadline:
+            try:
+                status = read("status", "/?readiness=1")
+                chat = read("chat", "/api/chat/capabilities")
+                assert status["readiness"]["state"] == "ready"
+                return status["runtime_identity"], chat["runtime_identity"]
+            except (OSError, ValueError):
+                time.sleep(0.1)
+        pytest.fail("isolated real manager did not become ready")
+
+    try:
+        for kind in ("status", "chat"):
+            result = launchctl(["launchctl", "bootstrap", domain, str(agents / f"com.loopx.{kind}.plist")], capture_output=True)
+            assert result.returncode == 0, result.stderr
+        before = ready()
+        started = time.monotonic()
+        result = align_managed_runtime(str(target))
+        after = ready()
+        alignment_seconds = time.monotonic() - started
+        assert before[0] != after[0]
+        assert after == (result["runtime_identity"], result["runtime_identity"])
+        assert read("chat", "/api/chat/sessions")["sessions"] == []
+        assert read("status", "/status.json")["registry"] == str(registry)
+        pids = [activation._managed_service_pid(agents / f"com.loopx.{k}.plist") for k in ports]
+        for _ in range(3):
+            assert align_managed_runtime(str(target))["restarted_services"] == []
+            assert ready() == after
+        assert pids == [activation._managed_service_pid(agents / f"com.loopx.{k}.plist") for k in ports]
+        print(json.dumps({"alignment_seconds": round(alignment_seconds, 3),
+            "including_three_repeats_seconds": round(time.monotonic() - started, 3),
+            "old_version": before[0]["package_version"], "new_version": after[0]["package_version"],
+            "repeat_alignment_restarts": 0, "original_registry_retained": True}))
+    finally:
+        for kind in ports:
+            original_run(["launchctl", "bootout", f"{domain}/{alias}.{kind}"], capture_output=True)

From 9f2a02580db25c9155d1a67f805a071fb38eed4e Mon Sep 17 00:00:00 2001
From: huangruiteng <huangrt01@163.com>
Date: Tue, 3 Nov 2026 17:39:23 +0800
Subject: [PATCH 2/4] test(desktop): measure cold projection separately from
 readiness

Signed-off-by: huangruiteng <huangrt01@163.com>
---
 tests/test_managed_service_runtime.py | 14 ++++++++++----
 1 file changed, 10 insertions(+), 4 deletions(-)

diff --git a/tests/test_managed_service_runtime.py b/tests/test_managed_service_runtime.py
index d0bc9623e2..f4528a6782 100644
--- a/tests/test_managed_service_runtime.py
+++ b/tests/test_managed_service_runtime.py
@@ -273,11 +273,12 @@ def launchctl(args, **kwargs):
         payload = {"Label": f"com.loopx.{kind}", "ProgramArguments": args, "RunAtLoad": True,
             "KeepAlive": True, "EnvironmentVariables": {"HOME": str(home), "LOOPX_USAGE_PING": "0",
                 "CODEX_HOME": str(home / ".codex"), "PATH": os.environ.get("PATH", "")},
-            "StandardErrorPath": str(tmp_path / f"{kind}.log")}
+            "StandardErrorPath": str(tmp_path / f"{kind}.log"),
+            "StandardOutPath": str(tmp_path / f"{kind}.out.log")}
         (agents / f"com.loopx.{kind}.plist").write_bytes(plistlib.dumps(payload))
 
-    def read(kind, path):
-        with urllib.request.urlopen(f"http://127.0.0.1:{ports[kind]}{path}", timeout=2) as response:
+    def read(kind, path, *, timeout=2):
+        with urllib.request.urlopen(f"http://127.0.0.1:{ports[kind]}{path}", timeout=timeout) as response:
             return json.load(response)
 
     def ready():
@@ -304,13 +305,18 @@ def ready():
         assert before[0] != after[0]
         assert after == (result["runtime_identity"], result["runtime_identity"])
         assert read("chat", "/api/chat/sessions")["sessions"] == []
-        assert read("status", "/status.json")["registry"] == str(registry)
+        # This full cold projection checks data routing, after readiness was
+        # qualified separately. Its provider cost is not the startup budget.
+        projection_started = time.monotonic()
+        assert read("status", "/status.json", timeout=15)["registry"] == str(registry)
+        projection_seconds = time.monotonic() - projection_started
         pids = [activation._managed_service_pid(agents / f"com.loopx.{k}.plist") for k in ports]
         for _ in range(3):
             assert align_managed_runtime(str(target))["restarted_services"] == []
             assert ready() == after
         assert pids == [activation._managed_service_pid(agents / f"com.loopx.{k}.plist") for k in ports]
         print(json.dumps({"alignment_seconds": round(alignment_seconds, 3),
+            "cold_projection_seconds": round(projection_seconds, 3),
             "including_three_repeats_seconds": round(time.monotonic() - started, 3),
             "old_version": before[0]["package_version"], "new_version": after[0]["package_version"],
             "repeat_alignment_restarts": 0, "original_registry_retained": True}))

From a129f6fff736ccd74f13d7f547614307adf50c32 Mon Sep 17 00:00:00 2001
From: huangruiteng <huangrt01@163.com>
Date: Tue, 3 Nov 2026 18:12:09 +0800
Subject: [PATCH 3/4] fix(desktop): recover runtime alignment automatically
 with backoff

Signed-off-by: huangruiteng <huangrt01@163.com>
---
 apps/desktop/loopx-control-plane/README.md    |  29 ++--
 .../loopx-control-plane/src-tauri/src/lib.rs  |  14 +-
 .../src-tauri/src/maintenance.rs              | 147 +++++++++++++++---
 .../src-tauri/src/services.rs                 |   2 +-
 .../loopx-control-plane/static/boot.css       |   3 +
 .../loopx-control-plane/static/boot.js        |  41 +++--
 .../loopx-control-plane/static/index.html     |   2 +-
 loopx/{ => hosts}/managed_service_runtime.py  |   4 +-
 loopx/runtime_activation.py                   |   4 +-
 tests/test_managed_service_runtime.py         |  28 +++-
 10 files changed, 205 insertions(+), 69 deletions(-)
 rename loopx/{ => hosts}/managed_service_runtime.py (99%)

diff --git a/apps/desktop/loopx-control-plane/README.md b/apps/desktop/loopx-control-plane/README.md
index 6938911f7c..5e88a26f52 100644
--- a/apps/desktop/loopx-control-plane/README.md
+++ b/apps/desktop/loopx-control-plane/README.md
@@ -215,12 +215,21 @@ new CLI runtime; other managed services retain their existing restart behavior.
 来源可验证时才切换。切换保留原数据、账号和启动参数，并确认两个 HTTP 服务都使用
 目标版本；失败恢复原配置，中断由本地记录恢复。已匹配的服务直接复用，不反复重启。
 
-If alignment cannot complete, the startup clock stops and **Recovery & updates**
-shows the cause. **Reconnect** retries through the native supervisor without
-clearing the remembered runtime or reloading only the web page. Persistent
-alignment failures wait for that action or repair; pending Chat work keeps its
-automatic retry. A legacy Core without alignment support needs an update; it
-does not authorize falling back to terminating its services.
+If alignment cannot complete, the native supervisor automatically retries with
+increasing delays (2, 4, 8, 16, 32, then at most 60 seconds between attempts).
+The cooldown starts after an attempt finishes; it does not expire while a slow
+attempt is still running. Each attempt rediscovers the runtime and reuses the
+same manager/Chat fences and interrupted-handoff recovery. There is no permanent
+retry-count latch. Once the services qualify, the workspace opens without a
+click or reload. The startup screen shows the next automatic retry; waiting is
+visually idle, with details available in **Recovery & updates**. **Retry now**
+is optional and resets the delay through the native supervisor, preserving the
+selected runtime. Pending Chat work retains its existing automatic retry.
+
+恢复是默认流程：暂时切换失败会自动退避重试，中断的切换会先恢复原服务。服务就绪后
+自动进入工作区，无需点击按钮；「立即重试」只是跳过等待的可选入口。自定义来源仍需
+可验证，已有请求仍受保护；检查更新、安装重启和显式运行时选择保留原有边界。旧 Core
+尚不支持对齐时会继续检测可用运行时，不回退到反复终止旧服务。
 
 开发诊断可使用 `loopx --format json update activate --service all
 --runtime-executable /absolute/path/to/loopx`：这是会切换服务的操作，保留原工作区和
@@ -673,9 +682,11 @@ disposable labels; the manager, Core processes, HTTP and stores are real. The
 fixture uses temporary registry/account directories and ports, removes its own
 services afterwards, and never unloads the machine's standard services. It
 checks the target identity, original registry and repeated alignment without
-PID changes. Unit cases cover busy Chat, changed configuration, unsupported
-shells, failed startup/rollback and interrupted handoff. Native tests separately
-cover exact identity/readiness receipts and waiting for an explicit reconnect.
+PID changes, including a real manager failure followed by rollback and successful
+recovery. Unit cases cover busy Chat, changed configuration, unsupported shells,
+failed startup/rollback and interrupted handoff. Native tests cover exact
+identity/readiness receipts, automatic recovery without user input, capped
+backoff after slow attempts, and preservation of update admission.
 
 This installs only in the disposable root. The second test uses an explicit
 local launch pin to route the production supervisor's two services to its own
diff --git a/apps/desktop/loopx-control-plane/src-tauri/src/lib.rs b/apps/desktop/loopx-control-plane/src-tauri/src/lib.rs
index 636d5763db..ab887464ca 100644
--- a/apps/desktop/loopx-control-plane/src-tauri/src/lib.rs
+++ b/apps/desktop/loopx-control-plane/src-tauri/src/lib.rs
@@ -54,14 +54,8 @@ fn maintenance_origin(url: &Url) -> String {
 // echo verbatim into the boot surface, where it names the recovery panel's
 // actionable diagnostics instead of a misleading fixed message.
 fn boot_failure_message(error: &str) -> String {
-    if matches!(
-        error,
-        "service_runtime_alignment_unavailable"
-            | "service_runtime_route_unverified"
-            | "service_runtime_alignment_failed"
-            | "service_runtime_rollback_failed"
-    ) {
-        return "服务版本尚未对齐；请展开「恢复与更新」查看原因。".to_string();
+    if services::runtime_alignment_error(error) {
+        return "正在自动恢复本地服务，完成后会打开工作区。".to_string();
     }
     if error == "chat_work_pending" {
         return "管家正在处理请求，完成后会自动连接工作区。".to_string();
@@ -343,10 +337,6 @@ pub fn run() {
                                     | "runtime_pairing_required"
                                     | "runtime_selection_invalid"
                                     | "runtime_selection_unavailable"
-                                    | "service_runtime_alignment_unavailable"
-                                    | "service_runtime_alignment_failed"
-                                    | "service_runtime_route_unverified"
-                                    | "service_runtime_rollback_failed"
                             ) {
                                 150
                             } else {
diff --git a/apps/desktop/loopx-control-plane/src-tauri/src/maintenance.rs b/apps/desktop/loopx-control-plane/src-tauri/src/maintenance.rs
index fcda6dd47f..d46e35e3e3 100644
--- a/apps/desktop/loopx-control-plane/src-tauri/src/maintenance.rs
+++ b/apps/desktop/loopx-control-plane/src-tauri/src/maintenance.rs
@@ -27,6 +27,7 @@ pub struct Maintenance {
     pending: Mutex<Option<(String, Update)>>,
     last_failure: Mutex<Value>,
     runtime_retry: Mutex<RuntimeRetry>,
+    service_retry: Mutex<ServiceRetry>,
     install_journal_discarded: AtomicBool,
     environment_cache: Mutex<Option<(Instant, Value)>>,
     startup_started: std::sync::OnceLock<Instant>,
@@ -53,6 +54,26 @@ struct RuntimeRetry {
     last_attempt: Option<Instant>,
 }
 
+#[derive(Default)]
+struct ServiceRetry {
+    failures: u32,
+    next_attempt: Option<Instant>,
+}
+
+impl ServiceRetry {
+    fn failed(&mut self, now: Instant) {
+        let delay = Duration::from_secs((2_u64 << self.failures.min(5)).min(60));
+        self.failures = self.failures.saturating_add(1);
+        self.next_attempt = Some(now + delay);
+    }
+
+    fn remaining(&self, now: Instant) -> Duration {
+        self.next_attempt
+            .map(|next| next.saturating_duration_since(now))
+            .unwrap_or_default()
+    }
+}
+
 impl RuntimeRetry {
     fn admit(&mut self, now: Instant) -> bool {
         if self.attempts >= 3
@@ -72,7 +93,17 @@ impl Maintenance {
         if self.incomplete_app_installation.load(Ordering::Acquire) {
             json!({"phase":"error", "details":{"code":"app_install_incomplete"}})
         } else {
-            self.snapshot.lock().unwrap().clone()
+            let mut snapshot = self.snapshot.lock().unwrap().clone();
+            if snapshot["phase"] == "service_error"
+                && crate::services::runtime_alignment_error(
+                    snapshot["details"]["code"].as_str().unwrap_or(""),
+                )
+            {
+                let retry = self.service_retry.lock().unwrap();
+                snapshot["details"]["retry_after_ms"] =
+                    json!(retry.remaining(Instant::now()).as_millis() as u64);
+            }
+            snapshot
         }
     }
 
@@ -396,16 +427,14 @@ impl Maintenance {
         &self,
         start: impl FnOnce() -> Result<T, String>,
     ) -> Result<Option<T>, String> {
-        {
-            let snapshot = self.snapshot.lock().unwrap();
-            if snapshot["phase"] == "service_error"
-                && crate::services::runtime_recovery_required(
-                    snapshot["details"]["code"].as_str().unwrap_or(""),
-                )
-            {
-                return Ok(None);
-            }
-        }
+        self.reconcile_services_with_clock(Instant::now, start)
+    }
+
+    fn reconcile_services_with_clock<T>(
+        &self,
+        now: impl Fn() -> Instant,
+        start: impl FnOnce() -> Result<T, String>,
+    ) -> Result<Option<T>, String> {
         if self.busy.load(Ordering::Acquire)
             || self.incomplete_app_installation.load(Ordering::Acquire)
             || self.manual_failure_pending.load(Ordering::Acquire)
@@ -429,8 +458,21 @@ impl Maintenance {
         if matches!(phase.as_str(), "available" | "restart_required") {
             return Ok(None);
         }
+        // A failed alignment owes automatic recovery, not a user click. Bound
+        // repeated attempts without removing the Core's route/Chat fences.
+        if phase == "service_error" {
+            let retry = self.service_retry.lock().unwrap();
+            if !retry.remaining(now()).is_zero() {
+                return Ok(None);
+            }
+            if retry.next_attempt.is_some() {
+                drop(retry);
+                self.publish("connecting", json!({}));
+            }
+        }
         match start() {
             Ok(services) => {
+                *self.service_retry.lock().unwrap() = ServiceRetry::default();
                 self.publish("ready", json!({}));
                 Ok(Some(services))
             }
@@ -454,7 +496,17 @@ impl Maintenance {
                         "service_runtime_rollback_failed" => "service_runtime_rollback_failed",
                         _ => "service_start_failed",
                     };
-                    self.publish("service_error", json!({"code":code}));
+                    let mut details = json!({"code":code});
+                    if crate::services::runtime_alignment_error(code) {
+                        let failed_at = now();
+                        let mut retry = self.service_retry.lock().unwrap();
+                        retry.failed(failed_at);
+                        details["retry_after_ms"] =
+                            json!(retry.remaining(failed_at).as_millis() as u64);
+                    } else {
+                        *self.service_retry.lock().unwrap() = ServiceRetry::default();
+                    }
+                    self.publish("service_error", details);
                 }
                 Err(error)
             }
@@ -676,12 +728,14 @@ pub async fn desktop_update(
     if action == "reconnect" {
         let state = app.state::<Maintenance>();
         let _guard = state.acquire()?;
+        let _supervision = state.supervision.lock().await;
         allow_action(
             state.snapshot.lock().unwrap()["phase"]
                 .as_str()
                 .unwrap_or(""),
             &action,
         )?;
+        *state.service_retry.lock().unwrap() = ServiceRetry::default();
         return Ok(state.publish("connecting", json!({})));
     }
     if cfg!(dev) || !app_update_platform_supported() {
@@ -2564,25 +2618,76 @@ mod tests {
     }
 
     #[test]
-    fn failed_alignment_waits_for_reconnect_without_restarting_services() {
-        let state = Maintenance::default();
+    fn failed_alignment_recovers_automatically_after_backoff() {
         for code in [
             "service_runtime_alignment_unavailable",
             "service_runtime_route_unverified",
             "service_runtime_alignment_failed",
             "service_runtime_rollback_failed",
         ] {
-            state.publish("service_error", json!({"code":code}));
-            for _ in 0..3 {
+            let state = Maintenance::default();
+            let clock = std::cell::Cell::new(Instant::now());
+            for delay in [2, 4, 8, 16, 32, 60, 60] {
+                assert_eq!(
+                    state.reconcile_services_with_clock::<()>(|| clock.get(), || Err(code.into())),
+                    Err(code.into())
+                );
+                assert_eq!(
+                    state.snapshot.lock().unwrap()["details"]["retry_after_ms"],
+                    delay * 1000
+                );
+                clock.set(clock.get() + Duration::from_secs(delay) - Duration::from_millis(1));
                 assert!(state
-                    .reconcile_services::<()>(|| panic!(
-                        "unchanged failure must not restart services"
-                    ))
+                    .reconcile_services_with_clock::<()>(
+                        || clock.get(),
+                        || panic!("cooldown must not touch services")
+                    )
                     .unwrap()
                     .is_none());
+                clock.set(clock.get() + Duration::from_millis(1));
             }
-            state.publish("connecting", json!({}));
-            assert_eq!(state.reconcile_services(|| Ok(42)).unwrap(), Some(42));
+            // No reconnect, reload or user action between failure and success.
+            assert_eq!(
+                state
+                    .reconcile_services_with_clock(|| clock.get(), || Ok(42))
+                    .unwrap(),
+                Some(42)
+            );
+            assert_eq!(state.startup_snapshot()["phase"], "ready");
+            assert!(state.service_retry.lock().unwrap().next_attempt.is_none());
+        }
+    }
+
+    #[test]
+    fn recovery_cooldown_starts_after_attempt_and_keeps_update_gates() {
+        let state = Maintenance::default();
+        let clock = std::cell::Cell::new(Instant::now());
+        assert!(state
+            .reconcile_services_with_clock::<()>(
+                || clock.get(),
+                || {
+                    clock.set(clock.get() + Duration::from_secs(70));
+                    Err("service_runtime_alignment_failed".into())
+                }
+            )
+            .is_err());
+        assert!(state
+            .reconcile_services_with_clock::<()>(
+                || clock.get(),
+                || panic!("no immediate retry after slow IO")
+            )
+            .unwrap()
+            .is_none());
+        clock.set(clock.get() + Duration::from_secs(2));
+        for phase in ["available", "restart_required"] {
+            state.publish(phase, json!({}));
+            assert!(state
+                .reconcile_services_with_clock::<()>(
+                    || clock.get(),
+                    || panic!("automatic recovery must preserve update admission")
+                )
+                .unwrap()
+                .is_none());
         }
     }
 
diff --git a/apps/desktop/loopx-control-plane/src-tauri/src/services.rs b/apps/desktop/loopx-control-plane/src-tauri/src/services.rs
index e7b6bc3b87..b894078a00 100644
--- a/apps/desktop/loopx-control-plane/src-tauri/src/services.rs
+++ b/apps/desktop/loopx-control-plane/src-tauri/src/services.rs
@@ -20,7 +20,7 @@ const STARTUP_TIMEOUT: Duration = Duration::from_secs(15);
 const PROBE_TIMEOUT: Duration = Duration::from_millis(500);
 const MAX_PROBE_RESPONSE_BYTES: u64 = 1024 * 1024;
 
-pub(crate) fn runtime_recovery_required(code: &str) -> bool {
+pub(crate) fn runtime_alignment_error(code: &str) -> bool {
     matches!(
         code,
         "service_runtime_alignment_unavailable"
diff --git a/apps/desktop/loopx-control-plane/static/boot.css b/apps/desktop/loopx-control-plane/static/boot.css
index c097ab3751..5fa9713b9e 100644
--- a/apps/desktop/loopx-control-plane/static/boot.css
+++ b/apps/desktop/loopx-control-plane/static/boot.css
@@ -88,6 +88,9 @@ button:disabled { cursor: wait; opacity: .6; }
 main[data-state="error"] .mark { box-shadow: none; animation: none; }
 main[data-state="error"] .status-dots { display: none; }
 main[data-state="error"] .progress::after { width: 100%; transform: none; background: var(--warning); animation: none; }
+main[data-state="waiting"] .mark { box-shadow: none; animation: none; }
+main[data-state="waiting"] .status-dots { display: none; }
+main[data-state="waiting"] .progress::after { width: 100%; transform: none; background: var(--hairline); animation: none; }
 @keyframes boot-progress {
   0% { transform: translateX(-60px); }
   50% { transform: translateX(86px); }
diff --git a/apps/desktop/loopx-control-plane/static/boot.js b/apps/desktop/loopx-control-plane/static/boot.js
index 390b56331b..a5ff071a4e 100644
--- a/apps/desktop/loopx-control-plane/static/boot.js
+++ b/apps/desktop/loopx-control-plane/static/boot.js
@@ -8,12 +8,12 @@ let timingObservedAt = pageStarted;
 let bootState = null;
 let observedEnvironment = null;
 const runtimeRecoveryTitles = {
-  service_runtime_alignment_unavailable: "请更新 LoopX 以连接现有服务",
-  service_runtime_route_unverified: "系统服务配置需要恢复",
-  service_runtime_alignment_failed: "服务版本切换未完成",
-  service_runtime_rollback_failed: "系统服务需要恢复",
+  service_runtime_alignment_unavailable: "正在等待可用的运行时",
+  service_runtime_route_unverified: "正在核验系统服务",
+  service_runtime_alignment_failed: "正在恢复本地服务",
+  service_runtime_rollback_failed: "正在恢复原服务",
 };
-function serviceRecoveryRequired(state) {
+function serviceRecoveryPending(state) {
   return state?.phase === "service_error" && Object.hasOwn(runtimeRecoveryTitles, state.details?.code);
 }
 function renderStartup(result) {
@@ -42,13 +42,20 @@ function renderStartup(result) {
     status.textContent = state.phase === "service_error"
       ? serviceTitles[state.details?.code] || titles.service_error
       : titles[state.phase];
-    panel.dataset.state = "loading";
-    panel.setAttribute("aria-busy", "true");
+    const waiting = serviceRecoveryPending(state);
+    panel.dataset.state = waiting ? "waiting" : "loading";
+    panel.setAttribute("aria-busy", String(!waiting));
   }
   updateStartupElapsed();
 }
 function updateStartupElapsed() {
-  if (["error", "runtime_required"].includes(bootState?.phase) || serviceRecoveryRequired(bootState)) {
+  if (serviceRecoveryPending(bootState)) {
+    const remaining = Math.max(0, (bootState.details.retry_after_ms ?? 0) - (performance.now() - timingObservedAt));
+    bootElapsed.textContent = remaining > 0 ? `${Math.ceil(remaining / 1000)} 秒后自动重试` : "正在重新连接";
+    bootDetail.textContent = "恢复后会自动打开工作区，无需手动操作。";
+    return;
+  }
+  if (["error", "runtime_required"].includes(bootState?.phase)) {
     bootElapsed.textContent = "等待恢复";
     bootDetail.textContent = "启动已停止等待。可在下方恢复，不必反复重开 App。";
     return;
@@ -68,13 +75,13 @@ function updateStartupElapsed() {
 }
 setInterval(updateStartupElapsed, 1000);
 window.loopxBootFailed = (message) => {
-  if (bootState?.phase === "connecting") return;
+  if (bootState?.phase === "connecting" || serviceRecoveryPending(bootState)) return;
   panel.dataset.state = "error";
   panel.setAttribute("aria-busy", "false");
   status.textContent = message;
 };
 window.loopxBootRetrying = () => {
-  if (["error", "runtime_required"].includes(bootState?.phase)) return;
+  if (["error", "runtime_required"].includes(bootState?.phase) || serviceRecoveryPending(bootState)) return;
   panel.dataset.state = "loading";
   panel.setAttribute("aria-busy", "true");
   status.textContent = "正在重新连接本地控制面";
@@ -111,10 +118,10 @@ const labels = {
   error: "更新未完成。请重试检查，或修复当前版本。Goal 数据不会被删除。",
 };
 const errors = {
-  service_runtime_alignment_unavailable: "当前运行时尚不支持服务版本对齐。请更新 LoopX CLI 后重新连接。现有服务未被停止。",
-  service_runtime_route_unverified: "系统服务配置已变化或使用自定义启动方式，无法自动对齐。请通过原安装方式恢复服务后重新连接。",
-  service_runtime_alignment_failed: "服务版本切换未完成，已尝试恢复原配置。可点击「重新连接」重试。",
-  service_runtime_rollback_failed: "服务版本切换与恢复均未完成。请保留诊断信息，通过原安装方式恢复系统服务。",
+  service_runtime_alignment_unavailable: "当前运行时尚不支持服务版本对齐。App 会自动重新检测可用运行时；现有服务保持运行。",
+  service_runtime_route_unverified: "系统服务来源暂时无法验证。App 会自动复查，来源确认后继续连接；自定义配置保持不变。",
+  service_runtime_alignment_failed: "服务版本切换未完成，已尝试恢复原配置。App 会自动重试。",
+  service_runtime_rollback_failed: "原服务尚未恢复。App 会自动重试恢复，并保留诊断信息。",
   chat_work_pending: "管家正在处理已有请求，完成后会自动切换服务并连接工作区。无需强退或重复发送。",
   chat_state_unverified: "暂时无法核验管家服务的任务状态。连接会自动重试；请保留诊断信息。",
   chat_service_unverified: "管家服务切换尚未确认。连接会自动重试；可保留诊断信息，或通过原安装方式更新 LoopX。",
@@ -306,19 +313,19 @@ async function refresh() {
   }
 }
 function escalateFromSnapshot(state) {
-  const terminal = TERMINAL_PHASES.includes(state?.phase) || serviceRecoveryRequired(state);
+  const terminal = TERMINAL_PHASES.includes(state?.phase);
   terminalRounds = terminal ? terminalRounds + 1 : 0;
   if (terminal && terminalRounds >= ERROR_ESCALATION_ROUNDS) escalated = true;
   if (!escalated) return;
   if (terminal) {
     panel.dataset.state = "error";
     panel.setAttribute("aria-busy", "false");
-    status.textContent = serviceRecoveryRequired(state) ? runtimeRecoveryTitles[state.details.code]
-      : `${codeText(state?.details?.code, state?.phase)} 详见下方「恢复与更新」面板，可复制诊断信息反馈。`;
+    status.textContent = `${codeText(state?.details?.code, state?.phase)} 详见下方「恢复与更新」面板，可复制诊断信息反馈。`;
     bootDetail.textContent = "启动已停止等待。可在下方恢复，不必反复重开 App。";
     document.querySelector(".recovery").open = true;
   } else {
     escalated = false;
+    if (serviceRecoveryPending(state)) return;
     panel.dataset.state = "loading";
     panel.setAttribute("aria-busy", "true");
     status.textContent = "正在重新连接本地控制面";
diff --git a/apps/desktop/loopx-control-plane/static/index.html b/apps/desktop/loopx-control-plane/static/index.html
index 25b4a7b7ef..2bd477b9ec 100644
--- a/apps/desktop/loopx-control-plane/static/index.html
+++ b/apps/desktop/loopx-control-plane/static/index.html
@@ -25,7 +25,7 @@ <h1 id="startup-title">LoopX</h1>
       <details class="recovery">
         <summary>恢复与更新 / Recovery & updates</summary>
         <p>重新连接不会执行任务。更新 App 与匹配的运行时可能短暂重启服务。</p>
-        <button id="retry" type="button">重新连接 / Reconnect</button>
+        <button id="retry" type="button">立即重试 / Retry now</button>
         <label>更新通道 <select id="channel"><option value="stable">稳定版 / Stable</option><option value="main">main 开发版</option></select></label>
         <button id="update" type="button">检查更新 / Check for updates</button>
         <p id="update-status" role="status"></p>
diff --git a/loopx/managed_service_runtime.py b/loopx/hosts/managed_service_runtime.py
similarity index 99%
rename from loopx/managed_service_runtime.py
rename to loopx/hosts/managed_service_runtime.py
index a9be6bc1ea..4dd5915bfc 100644
--- a/loopx/managed_service_runtime.py
+++ b/loopx/hosts/managed_service_runtime.py
@@ -19,8 +19,8 @@
 import urllib.error
 import urllib.request
 
-from .file_lock import exclusive_file_lock, write_bytes_atomic, write_json_atomic
-from . import runtime_activation as activation
+from ..file_lock import exclusive_file_lock, write_bytes_atomic, write_json_atomic
+from .. import runtime_activation as activation
 
 
 def _literal_arguments(command: str) -> list[str]:
diff --git a/loopx/runtime_activation.py b/loopx/runtime_activation.py
index 35d7e88bf2..8fc04514c8 100644
--- a/loopx/runtime_activation.py
+++ b/loopx/runtime_activation.py
@@ -248,7 +248,7 @@ def activate_runtime_services(*, service: str = "all", port: int = 8767,
         if runtime_executable is not None:
             if service != "all":
                 raise ValueError("service_runtime_route_unverified")
-            from .managed_service_runtime import align_managed_runtime
+            from .hosts.managed_service_runtime import align_managed_runtime
             return {**payload, "ok": True, **align_managed_runtime(runtime_executable)}
         if service == "chat":
             observed = replace_managed_loopx_chat("127.0.0.1", port, restart_managed=True)
@@ -362,7 +362,7 @@ def restart_services_for_runtime_activation(
             "restart_status": "skipped_custom_installation",
         }
     try:
-        from .managed_service_runtime import align_managed_runtime, has_runtime_binding
+        from .hosts.managed_service_runtime import align_managed_runtime, has_runtime_binding
         if sys.platform == "darwin" and has_runtime_binding():
             # Preserve the existing pending-work result even when the new
             # installation cannot yet be qualified. Alignment rechecks under
diff --git a/tests/test_managed_service_runtime.py b/tests/test_managed_service_runtime.py
index f4528a6782..9e9a91eae3 100644
--- a/tests/test_managed_service_runtime.py
+++ b/tests/test_managed_service_runtime.py
@@ -16,13 +16,16 @@
 import pytest
 
 from loopx import runtime_activation as activation
-from loopx.managed_service_runtime import _retarget, align_managed_runtime
+from loopx.hosts import managed_service_runtime as managed_runtime
+from loopx.hosts.managed_service_runtime import _retarget, align_managed_runtime
 from loopx.chat_store import ChatSessionStore
 
 
 @pytest.fixture
 def managed(tmp_path, monkeypatch):
-    monkeypatch.setattr("loopx.managed_service_runtime._verify_services", lambda *_: None)
+    if sys.platform == "win32":
+        pytest.skip("macOS launchd fixture executes POSIX launchers")
+    monkeypatch.setattr("loopx.hosts.managed_service_runtime._verify_services", lambda *_: None)
     monkeypatch.setattr(Path, "home", staticmethod(lambda: tmp_path))
     root = tmp_path / "original data"
     agents = tmp_path / "Library/LaunchAgents"
@@ -122,7 +125,7 @@ def test_startup_readback_failure_restores_both_original_services(managed, monke
     before = {p: p.read_bytes() for p in agents.glob("*.plist")}
     def unavailable(*_):
         raise RuntimeError("service_runtime_alignment_failed")
-    monkeypatch.setattr("loopx.managed_service_runtime._verify_services", unavailable)
+    monkeypatch.setattr("loopx.hosts.managed_service_runtime._verify_services", unavailable)
     with pytest.raises(RuntimeError, match="service_runtime_alignment_failed"):
         align_managed_runtime(str(target))
     assert {p: p.read_bytes() for p in before} == before
@@ -130,7 +133,7 @@ def unavailable(*_):
 
 
 def test_process_interruption_recovers_before_reapplying(managed, monkeypatch):
-    from loopx import managed_service_runtime as runtime
+    from loopx.hosts import managed_service_runtime as runtime
 
     _, agents, target, loaded, calls, _ = managed
     original_write = runtime.write_bytes_atomic
@@ -211,6 +214,7 @@ def test_custom_shell_effects_are_not_rewritten(shell):
         _retarget({"ProgramArguments": ["/bin/sh", "-c", shell]}, "/new/loopx", "/new/python")
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="macOS launcher uses POSIX paths")
 def test_direct_launcher_retains_environment_and_arguments():
     original = {"ProgramArguments": ["/old/loopx", "chat", "--runtime-root", "/original"],
         "EnvironmentVariables": {"CODEX_HOME": "/account"}}
@@ -298,6 +302,22 @@ def ready():
             result = launchctl(["launchctl", "bootstrap", domain, str(agents / f"com.loopx.{kind}.plist")], capture_output=True)
             assert result.returncode == 0, result.stderr
         before = ready()
+        # A transient manager failure must restore the running original pair;
+        # the next guarded attempt can then recover without data/config edits.
+        launch = managed_runtime._launchctl
+        interrupted = False
+
+        def fail_once(*args):
+            nonlocal interrupted
+            if args[0] == "bootstrap" and args[-1].endswith("com.loopx.chat.plist") and not interrupted:
+                interrupted = True
+                raise RuntimeError("service_runtime_alignment_failed")
+            return launch(*args)
+
+        monkeypatch.setattr(managed_runtime, "_launchctl", fail_once)
+        with pytest.raises(RuntimeError, match="service_runtime_alignment_failed"):
+            align_managed_runtime(str(target))
+        assert ready() == before
         started = time.monotonic()
         result = align_managed_runtime(str(target))
         after = ready()

From 57f7d971f933f03ae77c5ea44ef6800ada000984 Mon Sep 17 00:00:00 2001
From: huangruiteng <huangrt01@163.com>
Date: Tue, 3 Nov 2026 19:00:38 +0800
Subject: [PATCH 4/4] fix(desktop): verify original managed service roles
 before alignment

Signed-off-by: huangruiteng <huangrt01@163.com>
---
 apps/desktop/loopx-control-plane/README.md |  4 ++
 loopx/hosts/managed_service_runtime.py     | 55 +++++++++++++++------
 tests/test_managed_service_runtime.py      | 56 ++++++++++++++++++++++
 3 files changed, 100 insertions(+), 15 deletions(-)

diff --git a/apps/desktop/loopx-control-plane/README.md b/apps/desktop/loopx-control-plane/README.md
index 5e88a26f52..9e6bef1537 100644
--- a/apps/desktop/loopx-control-plane/README.md
+++ b/apps/desktop/loopx-control-plane/README.md
@@ -225,6 +225,10 @@ click or reload. The startup screen shows the next automatic retry; waiting is
 visually idle, with details available in **Recovery & updates**. **Retry now**
 is optional and resets the delay through the native supervisor, preserving the
 selected runtime. Pending Chat work retains its existing automatic retry.
+Before any stop or write, alignment verifies the original literal launcher,
+the actual `serve-status`/`chat` command position and its loopback endpoint;
+a standard label alone never qualifies a custom service. Interrupted handoffs
+apply the same role and unchanged-data-route checks before restoring services.
 
 恢复是默认流程：暂时切换失败会自动退避重试，中断的切换会先恢复原服务。服务就绪后
 自动进入工作区，无需点击按钮；「立即重试」只是跳过等待的可选入口。自定义来源仍需
diff --git a/loopx/hosts/managed_service_runtime.py b/loopx/hosts/managed_service_runtime.py
index 4dd5915bfc..67fc70bfcf 100644
--- a/loopx/hosts/managed_service_runtime.py
+++ b/loopx/hosts/managed_service_runtime.py
@@ -45,11 +45,10 @@ def _literal_arguments(command: str) -> list[str]:
     return words
 
 
-def _retarget(payload: dict, executable: str, python: str) -> dict:
-    result = copy.deepcopy(payload)
-    if "Program" in result:
+def _launcher(payload: dict) -> tuple[list[str], str | None]:
+    if "Program" in payload:
         raise ValueError("service_runtime_route_unverified")
-    args = result.get("ProgramArguments")
+    args = payload.get("ProgramArguments")
     if not isinstance(args, list) or not args or not all(isinstance(a, str) for a in args):
         raise ValueError("service_runtime_route_unverified")
     if Path(args[0]).name in {"sh", "bash", "zsh"}:
@@ -69,13 +68,21 @@ def _retarget(payload: dict, executable: str, python: str) -> dict:
             words = words[2:]
         if not words or not Path(words[0]).is_absolute():
             raise ValueError("service_runtime_route_unverified")
+        return words, prefix
+    if not Path(args[0]).is_absolute():
+        raise ValueError("service_runtime_route_unverified")
+    return args, None
+
+
+def _retarget(payload: dict, executable: str, python: str) -> dict:
+    result = copy.deepcopy(payload)
+    words, prefix = _launcher(payload)
+    if prefix is not None:
         launch = ["/usr/bin/env", "-u", "PYTHONPATH", "-u", "LOOPX_RELEASE_ROOT",
             f"LOOPX_PYTHON={python}", executable, *words[1:]]
-        result["ProgramArguments"] = [args[0], "-c", (prefix + "; " if prefix else "") + "exec " + shlex.join(launch)]
+        result["ProgramArguments"] = [payload["ProgramArguments"][0], "-c", (prefix + "; " if prefix else "") + "exec " + shlex.join(launch)]
     else:
-        if not Path(args[0]).is_absolute() or "Program" in result:
-            raise ValueError("service_runtime_route_unverified")
-        result["ProgramArguments"] = [executable, *args[1:]]
+        result["ProgramArguments"] = [executable, *words[1:]]
         result.setdefault("EnvironmentVariables", {})["LOOPX_PYTHON"] = python
         result["EnvironmentVariables"].pop("PYTHONPATH", None)
         result["EnvironmentVariables"].pop("LOOPX_RELEASE_ROOT", None)
@@ -103,12 +110,23 @@ def has_runtime_binding() -> bool:
 
 
 def _endpoint(payload: dict) -> int:
-    arguments = payload["ProgramArguments"]
-    if Path(arguments[0]).name in {"sh", "bash", "zsh"}:
-        command = arguments[2].rpartition("; exec ")[2]
-        if arguments[2].startswith("exec ") and "; exec " not in arguments[2]:
-            command = arguments[2][5:]
-        arguments = _literal_arguments(command)
+    arguments, _ = _launcher(payload)
+    # A standard label is not evidence of the service's role. Parse the command
+    # position, rather than finding a word that could be an option's value.
+    # This adapter accepts the installer's global options; custom invocations
+    # remain under their original owner without any signal or persistent edit.
+    expected = {"com.loopx.status": "serve-status", "com.loopx.chat": "chat"}.get(payload.get("Label"))
+    if expected is None or Path(arguments[0]).name != "loopx":
+        raise ValueError("service_runtime_route_unverified")
+    index = 1
+    while index < len(arguments) and arguments[index].startswith("--"):
+        flag, separator, _ = arguments[index].partition("=")
+        if flag not in {"--registry", "--runtime-root", "--format"}:
+            raise ValueError("service_runtime_route_unverified")
+        index += 1 if separator else 2
+    if index >= len(arguments) or arguments[index] != expected:
+        raise ValueError("service_runtime_route_unverified")
+    arguments = arguments[index + 1:]
     def option(flag):
         values = [word.split("=", 1)[1] if word.startswith(flag + "=")
             else arguments[i + 1] if i + 1 < len(arguments) else ""
@@ -199,6 +217,13 @@ def _recover_alignment(journal: Path) -> None:
         path = journal.parent / f"{label}.plist"
         originals[path] = base64.b64decode(item["original"], validate=True)
         candidates[path] = base64.b64decode(item["candidate"], validate=True)
+        original = plistlib.loads(originals[path])
+        candidate = plistlib.loads(candidates[path])
+        if (not isinstance(original, dict) or not isinstance(candidate, dict)
+                or original.get("Label") != label or candidate.get("Label") != label
+                or _endpoint(original) != _endpoint(candidate)
+                or _launcher(original)[0][1:] != _launcher(candidate)[0][1:]):
+            raise ValueError("service_runtime_route_unverified")
         if path.is_symlink() or path.read_bytes() not in {originals[path], candidates[path]}:
             raise ValueError("service_runtime_route_unverified")
     chat = journal.parent / "com.loopx.chat.plist"
@@ -259,8 +284,8 @@ def _align_managed_runtime(executable: str, journal: Path) -> dict:
         loaded, pid = activation._managed_service_state(plist)
         if not loaded:
             raise ValueError("service_runtime_route_unverified")
+        endpoints[plist] = _endpoint(payload)
         updated = _retarget(payload, str(target), python)
-        endpoints[plist] = _endpoint(updated)
         if updated != payload:
             originals[plist] = original
             updates[plist] = plistlib.dumps(updated)
diff --git a/tests/test_managed_service_runtime.py b/tests/test_managed_service_runtime.py
index 9e9a91eae3..85602e670f 100644
--- a/tests/test_managed_service_runtime.py
+++ b/tests/test_managed_service_runtime.py
@@ -109,6 +109,62 @@ def test_pending_chat_prevents_both_service_changes(managed):
     assert store.load_turn(sid, turn["turn_id"])["status"] == "queued"
 
 
+@pytest.mark.parametrize("shell", [False, True])
+@pytest.mark.parametrize("readiness_fails", [False, True])
+@pytest.mark.parametrize("route", ["wrong_role", "option_value", "foreign_executable"])
+def test_custom_status_route_keeps_its_accepted_chat_and_has_no_effects(
+    managed, monkeypatch, shell, readiness_fails, route,
+):
+    root, agents, target, loaded, calls, _ = managed
+    custom_root = root.parent / "custom chat"
+    store = ChatSessionStore(custom_root)
+    sid = store.create_session(goal_id="fixture", agent_id="fixture", adapter_kind="fake",
+        upstream_thread_id="fixture")["session_id"]
+    turn, _ = store.create_turn(sid, client_turn_id="pending", message="Keep accepted work.")
+    args = ["/old/bin/loopx", "--runtime-root", str(custom_root)]
+    if route == "option_value":
+        args += ["--registry", "serve-status"]
+    args += ["chat", "--host", "127.0.0.1", "--port", "8766"]
+    if route == "foreign_executable":
+        args[0], args[3] = "/custom/server", "serve-status"
+    path = agents / "com.loopx.status.plist"
+    payload = plistlib.loads(path.read_bytes())
+    payload["ProgramArguments"] = ["/bin/sh", "-c", "exec " + shlex.join(args)] if shell else args
+    path.write_bytes(plistlib.dumps(payload))
+    loaded[path.stem] = payload
+    before = {p: p.read_bytes() for p in agents.glob("*.plist")}
+    def readiness(*_):
+        if readiness_fails:
+            raise RuntimeError("service_runtime_alignment_failed")
+        pytest.fail("unverified source reached readiness after mutation")
+    monkeypatch.setattr(managed_runtime, "_verify_services", readiness)
+    with pytest.raises(ValueError, match="service_runtime_route_unverified"):
+        align_managed_runtime(str(target))
+    assert calls == []
+    assert {p: p.read_bytes() for p in before} == before
+    assert loaded == {p.stem: plistlib.loads(content) for p, content in before.items()}
+    assert store.load_turn(sid, turn["turn_id"])["status"] == "queued"
+    assert not (agents / ".loopx-runtime-activation.json").exists()
+
+
+def test_recovery_rejects_a_journal_with_changed_role_or_data_route(managed):
+    import base64
+
+    _, agents, target, loaded, calls, _ = managed
+    path = agents / "com.loopx.status.plist"
+    original = path.read_bytes()
+    candidate = plistlib.loads(original)
+    candidate["ProgramArguments"][2] = candidate["ProgramArguments"][2].replace("serve-status", "chat")
+    journal = agents / ".loopx-runtime-activation.json"
+    journal.write_text(json.dumps({"schema_version": "loopx_managed_runtime_handoff_v1",
+        "services": {path.stem: {"original": base64.b64encode(original).decode(),
+            "candidate": base64.b64encode(plistlib.dumps(candidate)).decode()}}}))
+    with pytest.raises(ValueError, match="service_runtime_route_unverified"):
+        align_managed_runtime(str(target))
+    assert calls == [] and len(loaded) == 2
+    assert path.read_bytes() == original and journal.exists()
+
+
 def test_failed_second_bootstrap_restores_original_loaded_pair(managed):
     _, agents, target, loaded, calls, fail = managed
     before = {p: p.read_bytes() for p in agents.glob("*.plist")}