From 7584495744511b859bff7c6a49ddf8754c8189c6 Mon Sep 17 00:00:00 2001
From: LiangshouX <xiaoliangshou.bit@gmail.com>
Date: Wed, 28 Oct 2026 16:31:35 +0800
Subject: [PATCH] fix(dsh): probe from the Host and bound every process
 lifetime

Local now means the DSH Host everywhere. The tab no longer fetches the
service URL from the browser: the hero reads the same Host probe the
service card uses, and a remote endpoint is probed through a new
remeEnv/health method, so a browser on another machine cannot disagree
with the Host. The wire namespace, method parameter convention and
service keys are unchanged.

Process lifetimes are explicit and documented in the manager. Python
probes and reme-ai checks stay bounded one-shot commands. pip installs
live in the manager's registry, keep the interpreter locked between a
cancel request and the actual exit, and are stopped by dispose() when
the plugin unloads. The ReMe service stays detached on purpose: it must
outlive DSH, so it is not handed to DSH's subprocess provider, which
kills its children on Host exit; it is stopped only through the
ownership-checked adapter in service.ts.

That adapter now verifies identity before reporting or stopping a
service: the recorded endpoint must match, the pid must be alive, and
its image must still be Python or reme, which protects against pid
reuse. A launch is recorded only after the OS acknowledged the spawn,
and a spawn failure returns an error instead of an unhandled event.
Starts are serialized and re-check the phase inside the lock, so two
requests cannot spawn two services. Health answers are validated, so a
generic HTTP server answering 200 with '{}' is no longer reported as a
running ReMe service, and authToken now rides the Authorization header
of both the client requests and the Host probe.

.env merging rewrites every occurrence of a key (parsing keeps the last
one), keeps the file's line ending, and validates the merged file rather
than just the submitted values.
---
 .../dsh/src/environment/client/rpc.ts         |   8 +
 .../dsh/src/environment/client/service.tsx    |  64 +++---
 .../dsh/src/environment/client/tab.tsx        |   9 +-
 .../dsh/src/environment/host/gateway.ts       |  34 ++-
 .../dsh/src/environment/host/install.ts       |  57 +++--
 .../dsh/src/environment/host/manager.ts       | 210 +++++++++++++-----
 .../dsh/src/environment/host/service.ts       | 201 +++++++++++++----
 .../dsh/src/environment/host/workspace.ts     |  36 +--
 integrations/dsh/src/environment/types.ts     |  10 +
 integrations/dsh/src/index.ts                 |   5 +
 integrations/dsh/src/reme/client.ts           |  11 +-
 integrations/dsh/src/reme/types.ts            |   5 +
 integrations/dsh/tests/env-gateway.test.mjs   |  30 +++
 integrations/dsh/tests/env-manager.test.mjs   | 105 +++++++++
 integrations/dsh/tests/env-reme.test.mjs      |  32 ++-
 integrations/dsh/tests/env-service.test.mjs   | 138 +++++++++---
 integrations/dsh/tests/env-tab.test.mjs       | 111 +++++++--
 integrations/dsh/tests/env-workspace.test.mjs |  26 +++
 18 files changed, 870 insertions(+), 222 deletions(-)
 create mode 100644 integrations/dsh/tests/env-manager.test.mjs

diff --git a/integrations/dsh/src/environment/client/rpc.ts b/integrations/dsh/src/environment/client/rpc.ts
index fd1f233d..c28821e2 100644
--- a/integrations/dsh/src/environment/client/rpc.ts
+++ b/integrations/dsh/src/environment/client/rpc.ts
@@ -1,5 +1,6 @@
 import type { StatusRpc } from "../../client/status-page.js";
 import type {
+  ConnectionHealth,
   EnvSummary,
   InstallStatus,
   LlmTestResult,
@@ -20,6 +21,7 @@ export type EnvMethod =
   | "readEnv"
   | "writeEnv"
   | "testLlm"
+  | "health"
   | "serviceStatus"
   | "serviceStart"
   | "serviceStop"
@@ -83,6 +85,12 @@ export function parseLlmResult(value: unknown): LlmTestResult {
   return record(value, "LLM test") as unknown as LlmTestResult;
 }
 
+export function parseHealth(value: unknown): ConnectionHealth {
+  const body = record(value, "health");
+  if (typeof body.ok !== "boolean") throw new Error("Invalid health response");
+  return value as unknown as ConnectionHealth;
+}
+
 export function parseServiceState(value: unknown): ServiceState {
   const body = record(value, "service status");
   if (typeof body.phase !== "string")
diff --git a/integrations/dsh/src/environment/client/service.tsx b/integrations/dsh/src/environment/client/service.tsx
index 854f3a32..2e4c4c9c 100644
--- a/integrations/dsh/src/environment/client/service.tsx
+++ b/integrations/dsh/src/environment/client/service.tsx
@@ -1,13 +1,13 @@
 import { useCallback, useEffect, useState } from "react";
 import type { StatusRpc } from "../../client/status-page.js";
 import { safeError } from "../../endpoint.js";
-import { ReMeClient } from "../../reme/client.js";
 import type { ServiceState } from "../types.js";
 import type { EnvTranslator } from "./locales.js";
 import { fill } from "./locales.js";
 import { StatusBadge } from "./presentation.js";
 import {
   envCall,
+  parseHealth,
   parseServiceLog,
   parseServiceStart,
   parseServiceState,
@@ -52,9 +52,10 @@ export function useService(
     local: boolean;
     pythonPath: string;
     workspaceDir: string;
+    authToken: string;
   },
 ): ServiceControl {
-  const { t, endpoint, local, pythonPath, workspaceDir } = options;
+  const { t, endpoint, local, pythonPath, workspaceDir, authToken } = options;
   const [state, setState] = useState<ServiceViewState>({ phase: "checking" });
   const [stopping, setStopping] = useState(false);
   const [tick, setTick] = useState(0);
@@ -65,15 +66,17 @@ export function useService(
   });
 
   useEffect(() => {
-    if (!local || !endpoint) {
-      setState({ phase: "checking" });
-      return;
-    }
+    if (!local || !endpoint) return;
     const controller = new AbortController();
     void (async () => {
       try {
         const next = parseServiceState(
-          await envCall(rpc, "serviceStatus", { endpoint }, controller.signal),
+          await envCall(
+            rpc,
+            "serviceStatus",
+            { endpoint, authToken },
+            controller.signal,
+          ),
         );
         if (controller.signal.aborted) return;
         setState((current) =>
@@ -91,7 +94,7 @@ export function useService(
       }
     })();
     return () => controller.abort();
-  }, [rpc, endpoint, local, tick]);
+  }, [rpc, endpoint, local, authToken, tick]);
 
   const refresh = useCallback(() => setTick((current) => current + 1), []);
 
@@ -105,17 +108,13 @@ export function useService(
           endpoint,
           pythonPath,
           workspaceDir,
+          authToken,
         }),
       );
       if (started.error) {
         setState({ phase: "error", error: started.error });
         return;
       }
-      if (started.alreadyRunning) {
-        setState({ phase: "running" });
-        refresh();
-        return;
-      }
       const deadline = Date.now() + START_DEADLINE_MS;
       for (;;) {
         await new Promise((resolve) => setTimeout(resolve, 1000));
@@ -125,7 +124,7 @@ export function useService(
         }
         try {
           const status = parseServiceState(
-            await envCall(rpc, "serviceStatus", { endpoint }),
+            await envCall(rpc, "serviceStatus", { endpoint, authToken }),
           );
           if (status.phase === "running") {
             setState({ phase: "running", data: status });
@@ -138,7 +137,7 @@ export function useService(
     } catch (error) {
       setState({ phase: "error", error: safeError(error) });
     }
-  }, [endpoint, pythonPath, refresh, rpc, state.phase, t, workspaceDir]);
+  }, [authToken, endpoint, pythonPath, rpc, state.phase, t, workspaceDir]);
 
   const stop = useCallback(async () => {
     if (stopping) return;
@@ -184,14 +183,13 @@ export interface ConnectionState {
  * URL), so the hero still reports whether the configured endpoint answers.
  */
 export function useHealth(options: {
+  rpc: StatusRpc;
   endpoint: string;
   enabled: boolean;
   tick: number;
-  requestTimeoutMs: number;
-  backgroundTimeoutMs: number;
+  authToken: string;
 }): ConnectionState {
-  const { endpoint, enabled, tick, requestTimeoutMs, backgroundTimeoutMs } =
-    options;
+  const { rpc, endpoint, enabled, tick, authToken } = options;
   const [health, setHealth] = useState<ConnectionState>({
     phase: "loading",
     endpoint: "",
@@ -200,27 +198,29 @@ export function useHealth(options: {
   useEffect(() => {
     if (!enabled || !endpoint) return;
     const controller = new AbortController();
-    const client = new ReMeClient({
-      endpoint,
-      requestTimeoutMs,
-      backgroundTimeoutMs,
-    });
-    void client
-      .healthCheck({ signal: controller.signal })
-      .then((result) => {
+    void (async () => {
+      try {
+        const result = parseHealth(
+          await envCall(
+            rpc,
+            "health",
+            { endpoint, authToken },
+            controller.signal,
+          ),
+        );
         if (controller.signal.aborted) return;
         setHealth(
           result.ok
-            ? { phase: "ok", endpoint, version: result.health?.version ?? "" }
+            ? { phase: "ok", endpoint, version: result.version }
             : { phase: "fail", endpoint, error: result.error },
         );
-      })
-      .catch((error: unknown) => {
+      } catch (error) {
         if (!controller.signal.aborted)
           setHealth({ phase: "fail", endpoint, error: safeError(error) });
-      });
+      }
+    })();
     return () => controller.abort();
-  }, [endpoint, enabled, requestTimeoutMs, backgroundTimeoutMs, tick]);
+  }, [authToken, endpoint, enabled, rpc, tick]);
 
   return health;
 }
diff --git a/integrations/dsh/src/environment/client/tab.tsx b/integrations/dsh/src/environment/client/tab.tsx
index a59ee7ce..90257cec 100644
--- a/integrations/dsh/src/environment/client/tab.tsx
+++ b/integrations/dsh/src/environment/client/tab.tsx
@@ -107,6 +107,7 @@ export function ReMeEnvironmentTab({
   const endpoint = value?.endpoint ?? "";
   const local = isLocalEndpoint(endpoint);
   const workspaceDir = value?.remeWorkspaceDir ?? "";
+  const authToken = value?.authToken ?? "";
   const [heroTick, setHeroTick] = useState(0);
 
   const writeSetting = useCallback(
@@ -141,15 +142,17 @@ export function ReMeEnvironmentTab({
     local,
     pythonPath: python.selectedPath,
     workspaceDir,
+    authToken,
   });
   // One health source per endpoint: the service probe already covers a local
-  // endpoint, so only a remote URL needs its own check.
+  // endpoint, so only a remote URL needs its own check. Both run on the Host,
+  // so the browser cannot disagree with it about what "local" means.
   const remote = useHealth({
+    rpc,
     endpoint,
     enabled: !local,
     tick: heroTick,
-    requestTimeoutMs: value?.requestTimeoutMs ?? 10000,
-    backgroundTimeoutMs: value?.backgroundTimeoutMs ?? 3600000,
+    authToken,
   });
   const connection: ConnectionState = local
     ? service.state.phase === "running"
diff --git a/integrations/dsh/src/environment/host/gateway.ts b/integrations/dsh/src/environment/host/gateway.ts
index 3174519a..3a211db0 100644
--- a/integrations/dsh/src/environment/host/gateway.ts
+++ b/integrations/dsh/src/environment/host/gateway.ts
@@ -2,6 +2,7 @@ import type { Context } from "@deepseek-ai/cordis";
 import { Remote, TypertRemoteService } from "@deepseek-ai/dsh-typert-protocol";
 
 import type {
+  ConnectionHealth,
   EnvSummary,
   InstallStatus,
   LlmTestResult,
@@ -12,6 +13,7 @@ import type {
   ServiceState,
 } from "../types.js";
 import type { ReMeEnvManager } from "./manager.js";
+import type { ServiceStopResult } from "./service.js";
 
 declare module "@deepseek-ai/cordis" {
   interface Context {
@@ -115,25 +117,49 @@ export class ReMeEnvGateway extends TypertRemoteService {
     );
   }
 
+  /**
+   * Probe the configured endpoint from this Host. The Environment hero and the
+   * service card both read Host results, so a browser on another machine cannot
+   * disagree with the Host about "local".
+   */
+  @Remote("health")
+  async health(
+    input: { endpoint?: string; authToken?: string } | undefined,
+  ): Promise<ConnectionHealth> {
+    return this.ctx.remeEnv.health(
+      input?.endpoint ?? "",
+      input?.authToken ?? "",
+    );
+  }
+
   /** Phase of the local ReMe service: running, conflict or stopped. */
   @Remote("serviceStatus")
   async serviceStatus(
-    input: { endpoint?: string } | undefined,
+    input: { endpoint?: string; authToken?: string } | undefined,
   ): Promise<ServiceState> {
-    return this.ctx.remeEnv.serviceStatus(input?.endpoint ?? "");
+    return this.ctx.remeEnv.serviceStatus(
+      input?.endpoint ?? "",
+      input?.authToken ?? "",
+    );
   }
 
   /** Spawn a detached `reme start`; poll serviceStatus until healthy. */
   @Remote("serviceStart")
   async serviceStart(
     input:
-      | { endpoint?: string; pythonPath?: string; workspaceDir?: string }
+      | {
+          endpoint?: string;
+          pythonPath?: string;
+          workspaceDir?: string;
+          authToken?: string;
+        }
       | undefined,
   ): Promise<ServiceStartResult> {
     return this.ctx.remeEnv.serviceStart(
       input?.endpoint ?? "",
       input?.pythonPath ?? "",
       input?.workspaceDir ?? "",
+      input?.authToken ?? "",
     );
   }
 
@@ -141,7 +167,7 @@ export class ReMeEnvGateway extends TypertRemoteService {
   @Remote("serviceStop")
   async serviceStop(
     input: { endpoint?: string } | undefined,
-  ): Promise<{ stopped: boolean; pid?: number }> {
+  ): Promise<ServiceStopResult> {
     return this.ctx.remeEnv.serviceStop(input?.endpoint ?? "");
   }
 
diff --git a/integrations/dsh/src/environment/host/install.ts b/integrations/dsh/src/environment/host/install.ts
index 8404591b..aab06b3c 100644
--- a/integrations/dsh/src/environment/host/install.ts
+++ b/integrations/dsh/src/environment/host/install.ts
@@ -2,11 +2,7 @@ import { spawn } from "node:child_process";
 import { closeSync, mkdirSync, openSync, readFileSync } from "node:fs";
 import { isAbsolute, join } from "node:path";
 
-import type {
-  InstallPhase,
-  InstallRequest,
-  InstallStatus,
-} from "../types.js";
+import type { InstallPhase, InstallRequest, InstallStatus } from "../types.js";
 
 export interface InstallChild {
   pid: number | undefined;
@@ -127,6 +123,10 @@ interface Job {
   exitCode?: number | null;
   logPath: string;
   child: InstallChild;
+  /** Set between a cancel request and the pip process actually exiting. */
+  cancelling: boolean;
+  /** Fallback that settles the job if the process ignores the kill. */
+  cancelTimer?: ReturnType<typeof setTimeout>;
 }
 
 /**
@@ -172,15 +172,11 @@ export class InstallJobs {
       startedAt: new Date().toISOString(),
       logPath,
       child,
+      cancelling: false,
     };
     this.jobs.set(id, job);
     this.latest = id;
-    child.onExit((code) => {
-      if (job.phase !== "running") return;
-      job.phase = code === 0 ? "succeeded" : "failed";
-      job.exitCode = code;
-      job.finishedAt = new Date().toISOString();
-    });
+    child.onExit((code) => this.finish(job, code));
     return this.snapshot(job);
   }
 
@@ -190,21 +186,51 @@ export class InstallJobs {
     return job ? this.snapshot(job) : null;
   }
 
-  /** Ask pip to stop; the polling client observes the cancelled phase. */
+  /**
+   * Ask pip to stop. The job keeps the interpreter locked (phase stays
+   * `running`, `cancelling` set) until the process has actually exited, so a
+   * second installation cannot start while the first is still dying.
+   */
   cancel(jobId?: string): boolean {
     const job = this.jobs.get(this.key(jobId));
-    if (!job || job.phase !== "running") return false;
-    job.phase = "cancelled";
-    job.finishedAt = new Date().toISOString();
+    if (!job || job.phase !== "running" || job.cancelling) return false;
+    job.cancelling = true;
     job.child.kill();
+    // A process that ignores the kill must not lock the interpreter forever.
+    job.cancelTimer = setTimeout(() => this.finish(job, null), 15_000);
+    job.cancelTimer.unref();
     return true;
   }
 
+  /** Stop every running installation; called when the plugin unloads. */
+  dispose(): void {
+    for (const job of this.jobs.values()) {
+      if (job.phase !== "running") continue;
+      job.cancelling = true;
+      job.child.kill();
+    }
+  }
+
   private key(jobId?: string): string {
     const trimmed = (jobId ?? "").trim();
     return trimmed || this.latest || "";
   }
 
+  /** Settle a running job exactly once, from an exit or the cancel fallback. */
+  private finish(job: Job, code: number | null): void {
+    if (job.phase !== "running") return;
+    if (job.cancelTimer) clearTimeout(job.cancelTimer);
+    job.cancelTimer = undefined;
+    job.phase = job.cancelling
+      ? "cancelled"
+      : code === 0
+      ? "succeeded"
+      : "failed";
+    job.exitCode = code;
+    job.finishedAt = new Date().toISOString();
+    job.cancelling = false;
+  }
+
   /** Serialize only JSON-safe values; optional fields are omitted when unset. */
   private snapshot(job: Job): InstallStatus {
     return {
@@ -215,6 +241,7 @@ export class InstallJobs {
       startedAt: job.startedAt,
       ...(job.finishedAt === undefined ? {} : { finishedAt: job.finishedAt }),
       ...(job.exitCode === undefined ? {} : { exitCode: job.exitCode }),
+      ...(job.cancelling ? { cancelling: true } : {}),
       logPath: job.logPath,
       logTail: tailLog(job.logPath),
     };
diff --git a/integrations/dsh/src/environment/host/manager.ts b/integrations/dsh/src/environment/host/manager.ts
index 115b05e4..3cc00823 100644
--- a/integrations/dsh/src/environment/host/manager.ts
+++ b/integrations/dsh/src/environment/host/manager.ts
@@ -3,11 +3,7 @@ import { homedir } from "node:os";
 import { join } from "node:path";
 
 import { detectPythonEnvs } from "./detect.js";
-import {
-  createInstallSpawner,
-  InstallJobs,
-  tailLog,
-} from "./install.js";
+import { createInstallSpawner, InstallJobs, tailLog } from "./install.js";
 import { checkReme, remeVersionAtLeast } from "./reme.js";
 import { createNodeRunner } from "./runner.js";
 import {
@@ -17,16 +13,18 @@ import {
   startService,
   stopService,
 } from "./service.js";
-import type { ServiceIO } from "./service.js";
+import type { ServiceIO, ServiceStopResult } from "./service.js";
 import {
   envSummary,
+  mergeEnvFile,
   parseEnvFile,
   resolveWorkspaceDir,
   testLlm,
   validateEnvUpdate,
-  writeEnvFile,
+  writeEnvText,
 } from "./workspace.js";
 import type {
+  ConnectionHealth,
   EnvSummary,
   InstallRequest,
   InstallStatus,
@@ -38,21 +36,43 @@ import type {
   ServiceState,
 } from "../types.js";
 
+const DEFAULT_ENDPOINT = "http://127.0.0.1:2333";
 const CACHE_MS = 60_000;
+/** How long a spawn is treated as "already starting" while it comes up. */
+const START_GRACE_MS = 120_000;
 
 /**
- * Host-side local environment manager: read-only Python detection plus the
- * reme-ai dependency check and the one-click pip install job.
+ * Host-side environment manager: the one place that runs local operations and
+ * hands the results to the gateway.
+ *
+ * Everything here runs on the machine DSH runs on, so the tab and the Host can
+ * never disagree about which Python, which `.env` or which service "local"
+ * means. Lifetimes are explicit:
+ *
+ * - Python probes and reme-ai checks are bounded one-shot commands (`timeoutMs`).
+ * - A pip installation is a job in this manager's registry; {@link dispose}
+ *   stops the ones still running when the plugin unloads, and its log survives
+ *   in `$DSH_HOME/reme-memory`.
+ * - The ReMe service is detached on purpose: it must outlive DSH, so it is not
+ *   handed to DSH's subprocess provider, which kills its children on Host exit.
+ *   It is stopped only through the ownership-checked adapter in `service.ts`.
  */
 export class ReMeEnvManager {
   private cache?: { at: number; detection: PythonDetection };
   private readonly installs: InstallJobs;
   private readonly env: NodeJS.ProcessEnv;
-  private readonly serviceIO: ServiceIO = createServiceIO();
+  private readonly serviceIO: ServiceIO;
   private readonly serviceStatePath: string;
+  /** Serializes starts so two requests cannot spawn two services. */
+  private startQueue: Promise<unknown> = Promise.resolve();
+  private starting?: { endpoint: string; at: number };
 
-  constructor(env: NodeJS.ProcessEnv = process.env) {
+  constructor(
+    env: NodeJS.ProcessEnv = process.env,
+    serviceIO: ServiceIO = createServiceIO(),
+  ) {
     this.env = env;
+    this.serviceIO = serviceIO;
     const home = env.DSH_HOME?.trim() || join(homedir(), ".dsh");
     this.installs = new InstallJobs({
       spawner: createInstallSpawner(env),
@@ -100,6 +120,11 @@ export class ReMeEnvManager {
     return this.installs.cancel(jobId);
   }
 
+  /** Stop the installations this manager owns, on plugin disposal. */
+  dispose(): void {
+    this.installs.dispose();
+  }
+
   /** Masked .env projection for the selected workspace. */
   readEnv(workspaceDir: string): EnvSummary {
     const dir = resolveWorkspaceDir(workspaceDir, homedir());
@@ -107,20 +132,22 @@ export class ReMeEnvManager {
   }
 
   /**
-   * Merge edited values into the workspace .env. Validation covers the whole
-   * effective file, so an untouched real key still satisfies the requirement.
+   * Merge edited values into the workspace .env. The whole merged file is
+   * validated, so an untouched real key still satisfies the requirement and a
+   * cleared one fails here rather than at the next service start.
    */
   writeEnv(workspaceDir: string, values: Record<string, string>): EnvSummary {
     const dir = resolveWorkspaceDir(workspaceDir, homedir());
     const path = join(dir, ".env");
-    let existing: Record<string, string> = {};
+    let existing = "";
     try {
-      existing = parseEnvFile(readFileSync(path, "utf8"));
+      existing = readFileSync(path, "utf8");
     } catch {
       // First write creates the file.
     }
-    validateEnvUpdate({ ...existing, ...values });
-    writeEnvFile(path, values);
+    const merged = mergeEnvFile(existing, values);
+    validateEnvUpdate(parseEnvFile(merged));
+    writeEnvText(path, merged);
     return envSummary(path, dir);
   }
 
@@ -136,31 +163,136 @@ export class ReMeEnvManager {
     );
   }
 
+  /** Whether the configured ReMe endpoint answers, probed from this Host. */
+  async health(endpoint: string, authToken = ""): Promise<ConnectionHealth> {
+    const endpointValue = this.endpoint(endpoint);
+    const health = await this.serviceIO.fetchHealth(endpointValue, authToken);
+    return {
+      endpoint: endpointValue,
+      ok: health.ok,
+      version: health.version,
+      error: health.error,
+      checkedAt: new Date().toISOString(),
+    };
+  }
+
   /** Current service phase (running / conflict / stopped) and ownership. */
-  async serviceStatus(endpoint: string): Promise<ServiceState> {
-    const endpointValue = endpoint.trim() || "http://127.0.0.1:2333";
+  async serviceStatus(endpoint: string, authToken = ""): Promise<ServiceState> {
+    const endpointValue = this.endpoint(endpoint);
     const persisted = readServiceState(this.serviceStatePath);
-    return serviceStatus(this.serviceIO, {
+    const status = await serviceStatus(this.serviceIO, {
       endpoint: endpointValue,
       statePath: this.serviceStatePath,
+      authToken,
       logPath: join(
         persisted?.workspaceDir ?? resolveWorkspaceDir("", homedir()),
         "service.log",
       ),
     });
+    if (status.phase === "running" && this.starting?.endpoint === endpointValue)
+      this.starting = undefined;
+    return status;
   }
 
-  /** Spawn `reme start` detached; the caller polls serviceStatus. */
-  async serviceStart(
+  /**
+   * Spawn `reme start` detached; the caller polls serviceStatus. Starts are
+   * serialized and re-check the phase inside the lock, so two requests cannot
+   * spawn two services and overwrite the one recorded pid.
+   */
+  serviceStart(
     endpoint: string,
     pythonPath: string,
     workspaceDir: string,
+    authToken = "",
   ): Promise<ServiceStartResult> {
-    const endpointValue = endpoint.trim() || "http://127.0.0.1:2333";
+    const run = this.startQueue.then(() =>
+      this.startLocked(endpoint, pythonPath, workspaceDir, authToken),
+    );
+    this.startQueue = run.then(
+      () => undefined,
+      () => undefined,
+    );
+    return run;
+  }
+
+  /** Stop the plugin-managed service; a foreign service stays untouched. */
+  async serviceStop(endpoint: string): Promise<ServiceStopResult> {
+    const result = await stopService(this.serviceIO, {
+      statePath: this.serviceStatePath,
+      endpoint: this.endpoint(endpoint),
+    });
+    if (result.stopped) this.starting = undefined;
+    return result;
+  }
+
+  /** Tail of the workspace service log. */
+  serviceLog(workspaceDir: string): ServiceLog {
+    const dir = resolveWorkspaceDir(workspaceDir, homedir());
+    const logPath = join(dir, "service.log");
+    return { logPath, lines: tailLog(logPath, 40) };
+  }
+
+  /** DSH-startup best effort: start a stopped local service when configured. */
+  async autoStart(options: {
+    endpoint: string;
+    pythonPath: string;
+    workspaceDir: string;
+    authToken?: string;
+  }): Promise<void> {
+    const status = await this.serviceStatus(
+      options.endpoint,
+      options.authToken ?? "",
+    );
+    if (status.phase === "running") return;
+    let pythonPath = options.pythonPath.trim();
+    if (!pythonPath) {
+      const detection = await this.detectPython();
+      pythonPath = detection.suggested;
+    }
+    if (!pythonPath) return;
+    await this.serviceStart(
+      options.endpoint,
+      pythonPath,
+      options.workspaceDir,
+      options.authToken ?? "",
+    );
+  }
+
+  private endpoint(raw: string): string {
+    const value = raw.trim() || DEFAULT_ENDPOINT;
+    let url: URL;
+    try {
+      url = new URL(value);
+    } catch {
+      throw new TypeError("Invalid ReMe endpoint");
+    }
+    if (url.protocol !== "http:" && url.protocol !== "https:")
+      throw new TypeError("ReMe endpoint must be an http(s) URL");
+    return value;
+  }
+
+  private async startLocked(
+    endpoint: string,
+    pythonPath: string,
+    workspaceDir: string,
+    authToken: string,
+  ): Promise<ServiceStartResult> {
+    const endpointValue = this.endpoint(endpoint);
     const dir = resolveWorkspaceDir(workspaceDir, homedir());
     const logPath = join(dir, "service.log");
     const now = new Date().toISOString();
-    const current = await this.serviceStatus(endpointValue);
+    if (
+      this.starting?.endpoint === endpointValue &&
+      Date.now() - this.starting.at < START_GRACE_MS
+    )
+      return {
+        started: false,
+        alreadyRunning: true,
+        command: "",
+        logPath,
+        startedAt: now,
+      };
+    const current = await this.serviceStatus(endpointValue, authToken);
     if (current.phase === "running")
       return {
         started: false,
@@ -190,6 +322,7 @@ export class ReMeEnvManager {
         startedAt: now,
         error: "reme-ai is not installed in the selected interpreter",
       };
+    this.starting = { endpoint: endpointValue, at: Date.now() };
     return startService(this.serviceIO, {
       endpoint: endpointValue,
       workspaceDir: dir,
@@ -202,33 +335,4 @@ export class ReMeEnvManager {
       env: this.env,
     });
   }
-
-  /** Stop the plugin-managed service; a foreign service stays untouched. */
-  serviceStop(_endpoint: string): { stopped: boolean; pid?: number } {
-    return stopService(this.serviceIO, { statePath: this.serviceStatePath });
-  }
-
-  /** Tail of the workspace service log. */
-  serviceLog(workspaceDir: string): ServiceLog {
-    const dir = resolveWorkspaceDir(workspaceDir, homedir());
-    const logPath = join(dir, "service.log");
-    return { logPath, lines: tailLog(logPath, 40) };
-  }
-
-  /** DSH-startup best effort: start a stopped local service when configured. */
-  async autoStart(options: {
-    endpoint: string;
-    pythonPath: string;
-    workspaceDir: string;
-  }): Promise<void> {
-    const status = await this.serviceStatus(options.endpoint);
-    if (status.phase === "running") return;
-    let pythonPath = options.pythonPath.trim();
-    if (!pythonPath) {
-      const detection = await this.detectPython();
-      pythonPath = detection.suggested;
-    }
-    if (!pythonPath) return;
-    await this.serviceStart(options.endpoint, pythonPath, options.workspaceDir);
-  }
 }
diff --git a/integrations/dsh/src/environment/host/service.ts b/integrations/dsh/src/environment/host/service.ts
index 874601dd..0bf73937 100644
--- a/integrations/dsh/src/environment/host/service.ts
+++ b/integrations/dsh/src/environment/host/service.ts
@@ -1,4 +1,4 @@
-import { spawn } from "node:child_process";
+import { execFile, spawn } from "node:child_process";
 import {
   closeSync,
   existsSync,
@@ -11,8 +11,18 @@ import {
 import { createConnection } from "node:net";
 import { dirname } from "node:path";
 
+import { safeError } from "../../endpoint.js";
 import type { ServiceStartResult, ServiceState } from "../types.js";
 
+/**
+ * The local ReMe service is a *detached* process: it outlives DSH, which is the
+ * point of starting it from here, so it is deliberately not owned by DSH's
+ * subprocess provider (that provider terminates its children on Host exit).
+ * This module is the small service adapter that owns that lifetime instead:
+ * it spawns the process, records the pid in a state file, verifies ownership
+ * before reporting or stopping it, and tree-kills it on request.
+ */
+
 export interface PersistedService {
   pid?: number;
   endpoint: string;
@@ -22,8 +32,22 @@ export interface PersistedService {
   startedAt: string;
 }
 
+export interface ServiceHealth {
+  ok: boolean;
+  version: string;
+  /** Non-empty when the endpoint answered but not as a ReMe service. */
+  error: string;
+}
+
+export interface ServiceStopResult {
+  stopped: boolean;
+  pid?: number;
+  /** Why an unmanaged or missing service was left alone. */
+  reason?: string;
+}
+
 export interface ServiceIO {
-  fetchHealth(endpoint: string): Promise<{ ok: boolean; version: string }>;
+  fetchHealth(endpoint: string, authToken: string): Promise<ServiceHealth>;
   portOpen(host: string, port: number): Promise<boolean>;
   spawnService(options: {
     exe: string;
@@ -31,9 +55,11 @@ export interface ServiceIO {
     cwd: string;
     logPath: string;
     env: NodeJS.ProcessEnv;
-  }): { pid: number | undefined };
+  }): Promise<{ pid: number | undefined }>;
   killTree(pid: number): void;
   probePid(pid: number): void;
+  /** OS image name of a pid, or "" when it cannot be read. */
+  processImage(pid: number): Promise<string>;
   exists(path: string): boolean;
   platform: string;
 }
@@ -126,25 +152,54 @@ function alive(io: ServiceIO, pid?: number): boolean {
   }
 }
 
+/**
+ * Whether the recorded process is the service this plugin started for the
+ * endpoint on screen. The saved endpoint must match the requested one, the pid
+ * must still be alive, and its image must still be a Python (or `reme`)
+ * executable, which is what a recycled pid would rarely be.
+ */
+export async function ownsService(
+  io: ServiceIO,
+  persisted: PersistedService | undefined,
+  endpoint: string,
+): Promise<boolean> {
+  if (!persisted || persisted.pid === undefined) return false;
+  if (persisted.endpoint.trim() !== endpoint.trim()) return false;
+  if (!alive(io, persisted.pid)) return false;
+  const image = (await io.processImage(persisted.pid)).toLowerCase();
+  if (image === "") return true;
+  return /^(reme|python|py)/.test(image);
+}
+
 /** Running / conflict / stopped, with ownership of the running process. */
 export async function serviceStatus(
   io: ServiceIO,
-  options: { endpoint: string; statePath: string; logPath: string },
+  options: {
+    endpoint: string;
+    statePath: string;
+    logPath: string;
+    authToken?: string;
+  },
 ): Promise<ServiceState> {
-  const health = await io.fetchHealth(options.endpoint);
+  const health = await io.fetchHealth(
+    options.endpoint,
+    options.authToken ?? "",
+  );
   const persisted = readServiceState(options.statePath);
-  if (health.ok)
+  if (health.ok) {
+    const managed = await ownsService(io, persisted, options.endpoint);
     return {
       phase: "running",
-      managed: alive(io, persisted?.pid),
+      managed,
       endpoint: options.endpoint,
-      ...(persisted?.pid === undefined ? {} : { pid: persisted.pid }),
+      ...(managed && persisted?.pid !== undefined
+        ? { pid: persisted.pid }
+        : {}),
       ...(health.version ? { version: health.version } : {}),
-      ...(persisted?.startedAt === undefined
-        ? {}
-        : { startedAt: persisted.startedAt }),
+      ...(managed && persisted ? { startedAt: persisted.startedAt } : {}),
       logPath: options.logPath,
     };
+  }
   const { host, port } = parseLocalEndpoint(options.endpoint);
   if (await io.portOpen(host, port))
     return {
@@ -162,8 +217,12 @@ export async function serviceStatus(
   };
 }
 
-/** Spawn a detached service; the caller polls status until it is healthy. */
-export function startService(
+/**
+ * Spawn a detached service; the caller polls status until it is healthy. The
+ * launch is recorded only once the OS acknowledged the spawn, so a missing
+ * interpreter leaves no state file behind for a later tab to adopt.
+ */
+export async function startService(
   io: ServiceIO,
   options: {
     endpoint: string;
@@ -175,7 +234,7 @@ export function startService(
     logPath: string;
     env: NodeJS.ProcessEnv;
   },
-): ServiceStartResult {
+): Promise<ServiceStartResult> {
   mkdirSync(options.workspaceDir, { recursive: true });
   const command = buildStartCommand({
     cliPath: options.cliPath,
@@ -185,19 +244,31 @@ export function startService(
     disableDreamCron: options.disableDreamCron,
     exists: io.exists,
   });
-  const child = io.spawnService({
-    exe: command.exe,
-    args: command.args,
-    cwd: options.workspaceDir,
-    logPath: options.logPath,
-    env: {
-      ...options.env,
-      PYTHONUTF8: "1",
-      PYTHONIOENCODING: "utf-8",
-      NO_PROXY: "localhost,127.0.0.1,::1",
-    },
-  });
   const startedAt = new Date().toISOString();
+  let child: { pid: number | undefined };
+  try {
+    child = await io.spawnService({
+      exe: command.exe,
+      args: command.args,
+      cwd: options.workspaceDir,
+      logPath: options.logPath,
+      env: {
+        ...options.env,
+        PYTHONUTF8: "1",
+        PYTHONIOENCODING: "utf-8",
+        NO_PROXY: "localhost,127.0.0.1,::1",
+      },
+    });
+  } catch (error) {
+    return {
+      started: false,
+      alreadyRunning: false,
+      command: command.display,
+      logPath: options.logPath,
+      startedAt,
+      error: `Could not start the ReMe service: ${safeError(error)}`,
+    };
+  }
   writeServiceState(options.statePath, {
     pid: child.pid,
     endpoint: options.endpoint,
@@ -216,13 +287,22 @@ export function startService(
   };
 }
 
-/** Stop the managed service (tree kill) and clear its state file. */
-export function stopService(
+/**
+ * Stop the service this plugin started for the endpoint, and only that one: a
+ * service on another port, or a foreign process, is left running.
+ */
+export async function stopService(
   io: ServiceIO,
-  options: { statePath: string },
-): { stopped: boolean; pid?: number } {
+  options: { statePath: string; endpoint: string },
+): Promise<ServiceStopResult> {
   const persisted = readServiceState(options.statePath);
-  if (!persisted || persisted.pid === undefined) return { stopped: false };
+  if (!persisted || persisted.pid === undefined)
+    return { stopped: false, reason: "No service started by this plugin" };
+  if (!(await ownsService(io, persisted, options.endpoint)))
+    return {
+      stopped: false,
+      reason: "The recorded service does not belong to this endpoint",
+    };
   const pid = persisted.pid;
   try {
     io.killTree(pid);
@@ -240,13 +320,17 @@ export function stopService(
 /** Node-backed IO: fetch health, probe the port, spawn detached, tree-kill. */
 export function createServiceIO(): ServiceIO {
   return {
-    async fetchHealth(endpoint) {
+    async fetchHealth(endpoint, authToken) {
       const controller = new AbortController();
       const timer = setTimeout(() => controller.abort(), 3000);
+      const headers: Record<string, string> = {
+        "Content-Type": "application/json",
+      };
+      if (authToken) headers.Authorization = `Bearer ${authToken}`;
       try {
         const response = await fetch(`${endpoint}/health_check`, {
           method: "POST",
-          headers: { "Content-Type": "application/json" },
+          headers,
           body: "{}",
           signal: controller.signal,
         });
@@ -254,13 +338,26 @@ export function createServiceIO(): ServiceIO {
           success?: boolean;
           metadata?: { health?: { version?: unknown } };
         };
-        const version = body.metadata?.health?.version;
+        if (!response.ok)
+          return { ok: false, version: "", error: `HTTP ${response.status}` };
+        const health = body.metadata?.health;
+        if (body.success === false)
+          return { ok: false, version: "", error: "ReMe reported an error" };
+        // A generic HTTP server answering 200 with `{}` is not ReMe.
+        if (typeof health !== "object" || health === null)
+          return {
+            ok: false,
+            version: "",
+            error: "The endpoint did not answer with a ReMe health report",
+          };
+        const version = health.version;
         return {
-          ok: response.ok && body.success !== false,
+          ok: true,
           version: typeof version === "string" ? version : "",
+          error: "",
         };
-      } catch {
-        return { ok: false, version: "" };
+      } catch (error) {
+        return { ok: false, version: "", error: safeError(error) };
       } finally {
         clearTimeout(timer);
       }
@@ -278,7 +375,7 @@ export function createServiceIO(): ServiceIO {
         socket.once("error", () => finish(false));
       });
     },
-    spawnService({ exe, args, cwd, logPath, env }) {
+    async spawnService({ exe, args, cwd, logPath, env }) {
       const fd = openSync(logPath, "a");
       let child: ReturnType<typeof spawn>;
       try {
@@ -290,9 +387,18 @@ export function createServiceIO(): ServiceIO {
           env,
           stdio: ["ignore", fd, fd],
         });
+        await new Promise<void>((resolve, reject) => {
+          child.once("spawn", () => resolve());
+          child.once("error", (error: Error) => reject(error));
+        });
       } finally {
+        // The child owns its inherited handle; the parent copy is no longer
+        // needed, and an unref'd child survives this Host exiting.
         closeSync(fd);
       }
+      // A failed executable is reported here rather than as an unhandled
+      // `error` event that would exit the Host.
+      child.on("error", () => undefined);
       child.unref();
       return { pid: child.pid };
     },
@@ -313,6 +419,25 @@ export function createServiceIO(): ServiceIO {
     probePid(pid) {
       process.kill(pid, 0);
     },
+    processImage(pid) {
+      return new Promise((resolve) => {
+        if (process.platform === "win32") {
+          execFile(
+            "tasklist",
+            ["/FI", `PID eq ${pid}`, "/FO", "CSV", "/NH"],
+            { windowsHide: true },
+            (error, stdout) => {
+              const match = /^"([^"]+)"/.exec(String(stdout ?? "").trim());
+              resolve(error ? "" : match?.[1] ?? "");
+            },
+          );
+          return;
+        }
+        execFile("ps", ["-p", String(pid), "-o", "comm="], (error, stdout) =>
+          resolve(error ? "" : String(stdout ?? "").trim()),
+        );
+      });
+    },
     exists: existsSync,
     platform: process.platform,
   };
diff --git a/integrations/dsh/src/environment/host/workspace.ts b/integrations/dsh/src/environment/host/workspace.ts
index 9ca850fa..ff8c5d31 100644
--- a/integrations/dsh/src/environment/host/workspace.ts
+++ b/integrations/dsh/src/environment/host/workspace.ts
@@ -56,29 +56,39 @@ export function parseEnvFile(text: string): Record<string, string> {
 
 /**
  * Rewrite only the lines whose keys appear in updates; every other line
- * (comments, unknown variables, ordering) is preserved byte-for-byte.
+ * (comments, unknown variables, ordering) is preserved byte-for-byte, and the
+ * file keeps its own line ending.
+ *
+ * A key that appears more than once is rewritten in every position. Parsing
+ * keeps the last occurrence, so rewriting only the first would leave an older
+ * value in force: after this function every duplicate agrees.
  */
 export function mergeEnvFile(
   existingText: string,
   updates: Record<string, string>,
 ): string {
   const pending = new Map(Object.entries(updates));
+  const eol = /\r\n/.test(existingText) ? "\r\n" : "\n";
   const lines = existingText.length ? existingText.split(/\r?\n/) : [];
   const output: string[] = [];
+  const written = new Set<string>();
   for (const line of lines) {
     const match = /^\s*([A-Za-z_][A-Za-z0-9_]*)\s*=/.exec(line);
     const key = match?.[1] ?? "";
     if (key && pending.has(key)) {
+      // Every occurrence takes the new value: parsing keeps the last one, so
+      // rewriting only the first would leave an older value in force.
       output.push(`${key}=${pending.get(key) ?? ""}`);
-      pending.delete(key);
+      written.add(key);
       continue;
     }
     output.push(line);
   }
   // Normalize the trailing newline before appending new keys.
   while (output.length && output[output.length - 1] === "") output.pop();
-  for (const [key, value] of pending) output.push(`${key}=${value}`);
-  return `${output.join("\n")}\n`;
+  for (const [key, value] of pending)
+    if (!written.has(key)) output.push(`${key}=${value}`);
+  return `${output.join(eol)}${eol}`;
 }
 
 /** "••••" plus the last four characters; empty values stay empty. */
@@ -125,19 +135,13 @@ export function validateEnvUpdate(values: Record<string, string>): void {
     );
 }
 
-/** Merge updates into an existing .env, creating parent directories as needed. */
-export function writeEnvFile(
-  path: string,
-  updates: Record<string, string>,
-): void {
-  let existing = "";
-  try {
-    existing = readFileSync(path, "utf8");
-  } catch {
-    // First write creates the file.
-  }
+/**
+ * Write a fully merged `.env`. The caller merges and validates first, so the
+ * file on disk is exactly what was validated.
+ */
+export function writeEnvText(path: string, text: string): void {
   mkdirSync(dirname(path), { recursive: true });
-  writeFileSync(path, mergeEnvFile(existing, updates), "utf8");
+  writeFileSync(path, text, "utf8");
 }
 
 /**
diff --git a/integrations/dsh/src/environment/types.ts b/integrations/dsh/src/environment/types.ts
index b740a691..ebc09168 100644
--- a/integrations/dsh/src/environment/types.ts
+++ b/integrations/dsh/src/environment/types.ts
@@ -110,6 +110,16 @@ export interface LlmTestResult {
   error: string;
 }
 
+/** Whether the configured endpoint answers, probed from the DSH Host. */
+export interface ConnectionHealth {
+  endpoint: string;
+  ok: boolean;
+  version: string;
+  /** Non-empty when the endpoint answered but not as a ReMe service. */
+  error: string;
+  checkedAt: string;
+}
+
 export interface ServiceState {
   phase: "running" | "stopped" | "conflict";
   /** True when the running service was started by this plugin. */
diff --git a/integrations/dsh/src/index.ts b/integrations/dsh/src/index.ts
index 4755de51..d004a123 100644
--- a/integrations/dsh/src/index.ts
+++ b/integrations/dsh/src/index.ts
@@ -29,6 +29,10 @@ export function apply(ctx: Context, input: ReMeConfigInput = {}): void {
   ctx.provide("remeMemory", runtime);
   const envManager = new ReMeEnvManager();
   ctx.provide("remeEnv", envManager);
+  // The manager owns the bounded operations it started (pip installs), so it
+  // releases them when this plugin unloads. The detached ReMe service is not
+  // part of that lifetime: it is meant to outlive DSH.
+  ctx.effect(() => () => envManager.dispose(), "remeMemory.envLifetime()");
   void ctx.plugin(ReMeStatusGateway);
   void ctx.plugin(ReMeEnvGateway);
   ctx.effect(() => {
@@ -39,6 +43,7 @@ export function apply(ctx: Context, input: ReMeConfigInput = {}): void {
           endpoint: live.endpoint,
           pythonPath: live.pythonPath,
           workspaceDir: live.remeWorkspaceDir,
+          authToken: live.authToken,
         })
         .catch((error: unknown) => {
           ctx.logger.warn("remeMemory.envAutoStart failed", {
diff --git a/integrations/dsh/src/reme/client.ts b/integrations/dsh/src/reme/client.ts
index 10d9050c..50d7f65b 100644
--- a/integrations/dsh/src/reme/client.ts
+++ b/integrations/dsh/src/reme/client.ts
@@ -199,11 +199,16 @@ export class ReMeClient {
         throw new TypeError(
           "Invalid ReMe endpoint: credentials, query, and fragment are unsupported",
         );
+      const headers: Record<string, string> = {
+        "Content-Type": "application/json",
+      };
+      // The token belongs to the endpoint it was configured for, so it is sent
+      // here and not on any other request this plugin makes.
+      if (config.authToken)
+        headers.Authorization = `Bearer ${config.authToken}`;
       const response = await fetch(`${config.endpoint}/${job}`, {
         method: "POST",
-        headers: {
-          "Content-Type": "application/json",
-        },
+        headers,
         body: JSON.stringify(payload),
         signal,
       });
diff --git a/integrations/dsh/src/reme/types.ts b/integrations/dsh/src/reme/types.ts
index cf32dcb2..2341d099 100644
--- a/integrations/dsh/src/reme/types.ts
+++ b/integrations/dsh/src/reme/types.ts
@@ -3,6 +3,11 @@ export interface ReMeClientConfig {
   endpoint: string;
   requestTimeoutMs: number;
   backgroundTimeoutMs: number;
+  /**
+   * Shared secret sent as `Authorization: Bearer` to this endpoint only. Empty
+   * or absent means the service runs without authentication.
+   */
+  authToken?: string;
 }
 
 /** Normalized response returned by the ReMe HTTP client. */
diff --git a/integrations/dsh/tests/env-gateway.test.mjs b/integrations/dsh/tests/env-gateway.test.mjs
index 773c4791..9c02c76f 100644
--- a/integrations/dsh/tests/env-gateway.test.mjs
+++ b/integrations/dsh/tests/env-gateway.test.mjs
@@ -23,6 +23,13 @@ function stubManager() {
       cliPath: "",
       error: pythonPath ? "" : "Python interpreter path must be absolute",
     }),
+    health: async (endpoint, authToken) => ({
+      endpoint,
+      ok: false,
+      version: "",
+      error: authToken ? "auth token forwarded" : "",
+      checkedAt: "",
+    }),
   };
 }
 
@@ -90,3 +97,26 @@ test("gateway methods remain callable through the manager after remount", async
   assert.equal(check.installed, false);
   assert.match(check.error, /absolute/);
 });
+
+test("health keeps the single-input convention and forwards the token", async () => {
+  const { root } = await mounted();
+  const receiver = root.reflect.get("remeEnvGateway", false);
+  const source = Function.prototype.toString.call(
+    Object.getPrototypeOf(receiver).health,
+  );
+  const open = source.indexOf("(");
+  const close = source.indexOf(")", open + 1);
+  const parameters = source
+    .slice(open + 1, close)
+    .split(",")
+    .map((part) => part.trim())
+    .filter(Boolean);
+  assert.deepEqual(parameters, ["input"]);
+
+  const result = await receiver.health({
+    endpoint: "http://127.0.0.1:2334",
+    authToken: "secret",
+  });
+  assert.equal(result.endpoint, "http://127.0.0.1:2334");
+  assert.equal(result.error, "auth token forwarded");
+});
diff --git a/integrations/dsh/tests/env-manager.test.mjs b/integrations/dsh/tests/env-manager.test.mjs
new file mode 100644
index 00000000..ae7cd21d
--- /dev/null
+++ b/integrations/dsh/tests/env-manager.test.mjs
@@ -0,0 +1,105 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+
+import { ReMeEnvManager } from "../dist/environment/host/manager.js";
+
+/** A service adapter that records launches instead of starting processes. */
+function fakeServiceIO() {
+  const calls = { spawn: [] };
+  return {
+    calls,
+    io: {
+      fetchHealth: async () => ({ ok: false, version: "", error: "" }),
+      portOpen: async () => false,
+      async spawnService(options) {
+        calls.spawn.push(options);
+        // A launch is only visible once the OS acknowledged it.
+        await new Promise((resolve) => setTimeout(resolve, 10));
+        return { pid: 5000 + calls.spawn.length };
+      },
+      killTree() {},
+      probePid() {},
+      processImage: async () => "python.exe",
+      exists: () => true,
+      platform: "win32",
+    },
+  };
+}
+
+function managerIn(t) {
+  const directory = mkdtempSync(join(tmpdir(), "reme-manager-"));
+  t.after(() => rmSync(directory, { recursive: true, force: true }));
+  const { io, calls } = fakeServiceIO();
+  const manager = new ReMeEnvManager({ DSH_HOME: directory }, io);
+  // Both probes are Host process work; the test replaces them with facts.
+  manager.checkReme = async () => ({
+    installed: true,
+    version: "0.4.2.1",
+    versionOk: true,
+    hasCore: true,
+    cliPath: "",
+    error: "",
+  });
+  manager.serviceStatus = async (endpoint) => ({
+    phase: "stopped",
+    managed: false,
+    endpoint: endpoint || "http://127.0.0.1:2333",
+    logPath: join(directory, "service.log"),
+  });
+  return { manager, calls, directory };
+}
+
+test("serializes concurrent starts into one service launch", async (t) => {
+  const { manager, calls, directory } = managerIn(t);
+  const options = [
+    "http://127.0.0.1:2334",
+    "D:\\env\\python.exe",
+    join(directory, "workspace"),
+  ];
+  const [first, second] = await Promise.all([
+    manager.serviceStart(...options),
+    manager.serviceStart(...options),
+  ]);
+  assert.equal(calls.spawn.length, 1, "the second request joins the first");
+  assert.equal(first.started, true);
+  assert.equal(second.started, false);
+  assert.equal(second.alreadyRunning, true);
+});
+
+test("probes health from the Host and rejects a non-http endpoint", async (t) => {
+  const { manager } = managerIn(t);
+  const health = await manager.health("http://127.0.0.1:2334");
+  assert.equal(health.endpoint, "http://127.0.0.1:2334");
+  assert.equal(health.ok, false);
+  assert.equal(typeof health.checkedAt, "string");
+  await assert.rejects(() => manager.health("file:///etc/passwd"), /http/);
+});
+
+test("validates the merged .env, not just the submitted values", async (t) => {
+  const { manager, directory } = managerIn(t);
+  const envPath = join(directory, ".env");
+  writeFileSync(
+    envPath,
+    "LLM_MODEL_NAME=qwen3.7-plus\nLLM_API_KEY=sk-keep\nEMBEDDING_API_KEY=EMPTY\n",
+  );
+
+  // An untouched required key still satisfies the requirement.
+  const summary = manager.writeEnv(directory, {
+    LLM_MODEL_NAME: "qwen3.8-plus",
+  });
+  assert.equal(
+    summary.entries.find((entry) => entry.key === "LLM_MODEL_NAME").value,
+    "qwen3.8-plus",
+  );
+  assert.match(readFileSync(envPath, "utf8"), /^LLM_API_KEY=sk-keep$/m);
+
+  // An explicit clear leaves the file alone and reports why.
+  assert.throws(
+    () => manager.writeEnv(directory, { LLM_API_KEY: "" }),
+    /LLM_API_KEY is required/,
+  );
+  assert.match(readFileSync(envPath, "utf8"), /^LLM_API_KEY=sk-keep$/m);
+});
diff --git a/integrations/dsh/tests/env-reme.test.mjs b/integrations/dsh/tests/env-reme.test.mjs
index 63b00eb7..5776c31f 100644
--- a/integrations/dsh/tests/env-reme.test.mjs
+++ b/integrations/dsh/tests/env-reme.test.mjs
@@ -10,7 +10,11 @@ import {
   parseRemeVersion,
   remeVersionAtLeast,
 } from "../dist/environment/host/reme.js";
-import { InstallJobs, pipArgs, tailLog } from "../dist/environment/host/install.js";
+import {
+  InstallJobs,
+  pipArgs,
+  tailLog,
+} from "../dist/environment/host/install.js";
 
 const SNIPPETS = {
   version: "importlib.metadata",
@@ -212,15 +216,33 @@ test("tracks a failed pip job and keeps the log tail", (t) => {
   assert.match(done?.logTail ?? "", /Could not find a version/);
 });
 
-test("cancels a running job without reporting a failure", (t) => {
+test("keeps the interpreter locked until a cancelled job exits", (t) => {
   const { jobs, children } = makeJobs(t);
   const started = jobs.start({ pythonPath: "C:\\env\\python.exe" });
   assert.equal(jobs.cancel(started.id), true);
   assert.equal(children[0].killed, true);
-  assert.equal(jobs.status(started.id)?.phase, "cancelled");
-  children[0].exit(1); // Late exits must not overwrite the cancelled phase.
-  assert.equal(jobs.status(started.id)?.phase, "cancelled");
+  // The process still owns the interpreter, so the job stays running until the
+  // exit is observed and no second installation may start.
+  const cancelling = jobs.status(started.id);
+  assert.equal(cancelling?.phase, "running");
+  assert.equal(cancelling?.cancelling, true);
+  assert.equal(
+    jobs.start({ pythonPath: "C:\\env\\python.exe" }).id,
+    started.id,
+  );
   assert.equal(jobs.cancel(started.id), false);
+  children[0].exit(1); // A late exit settles the cancelled job, not a failure.
+  assert.equal(jobs.status(started.id)?.phase, "cancelled");
+  assert.equal(jobs.status(started.id)?.cancelling, undefined);
+});
+
+test("dispose stops the installations the manager still owns", (t) => {
+  const { jobs, children } = makeJobs(t);
+  const started = jobs.start({ pythonPath: "C:\\env\\python.exe" });
+  jobs.dispose();
+  assert.equal(children[0].killed, true);
+  children[0].exit(null);
+  assert.equal(jobs.status(started.id)?.phase, "cancelled");
 });
 
 test("reuses the running job for the same interpreter and validates paths", (t) => {
diff --git a/integrations/dsh/tests/env-service.test.mjs b/integrations/dsh/tests/env-service.test.mjs
index 43ba8d54..c848e3fb 100644
--- a/integrations/dsh/tests/env-service.test.mjs
+++ b/integrations/dsh/tests/env-service.test.mjs
@@ -6,6 +6,7 @@ import { join } from "node:path";
 
 import {
   buildStartCommand,
+  ownsService,
   parseLocalEndpoint,
   readServiceState,
   serviceStatus,
@@ -14,12 +15,14 @@ import {
   writeServiceState,
 } from "../dist/environment/host/service.js";
 
+const ENDPOINT = "http://127.0.0.1:2334";
+
 function fakeIO(overrides = {}) {
   const calls = { killTree: [], spawn: [] };
   const io = {
-    fetchHealth: async () => ({ ok: false, version: "" }),
+    fetchHealth: async () => ({ ok: false, version: "", error: "" }),
     portOpen: async () => false,
-    spawnService(options) {
+    async spawnService(options) {
       calls.spawn.push(options);
       return { pid: 4321 };
     },
@@ -27,6 +30,7 @@ function fakeIO(overrides = {}) {
       calls.killTree.push(pid);
     },
     probePid() {},
+    processImage: async () => "python.exe",
     exists: () => true,
     platform: "win32",
     ...overrides,
@@ -34,6 +38,24 @@ function fakeIO(overrides = {}) {
   return { io, calls };
 }
 
+function tempState(t, prefix) {
+  const directory = mkdtempSync(join(tmpdir(), prefix));
+  t.after(() => rmSync(directory, { recursive: true, force: true }));
+  return { directory, statePath: join(directory, "service.json") };
+}
+
+function record(statePath, overrides = {}) {
+  writeServiceState(statePath, {
+    pid: 777,
+    endpoint: ENDPOINT,
+    workspaceDir: join(statePath, ".."),
+    pythonPath: "D:\\env\\python.exe",
+    command: "reme start",
+    startedAt: "2026-10-27T00:00:00.000Z",
+    ...overrides,
+  });
+}
+
 test("accepts only loopback endpoints and validates the port", () => {
   assert.deepEqual(parseLocalEndpoint("http://127.0.0.1:2334"), {
     host: "127.0.0.1",
@@ -93,31 +115,22 @@ test("falls back to the interpreter wrapper when the CLI is missing", () => {
 });
 
 test("derives running, conflict and stopped phases", async (t) => {
-  const directory = mkdtempSync(join(tmpdir(), "reme-service-"));
-  t.after(() => rmSync(directory, { recursive: true, force: true }));
-  const statePath = join(directory, "service.json");
+  const { directory, statePath } = tempState(t, "reme-service-");
   const base = {
-    endpoint: "http://127.0.0.1:2334",
+    endpoint: ENDPOINT,
     statePath,
     logPath: join(directory, "service.log"),
   };
 
   const runningIo = fakeIO({
-    fetchHealth: async () => ({ ok: true, version: "0.4.0.9" }),
+    fetchHealth: async () => ({ ok: true, version: "0.4.0.9", error: "" }),
   });
   let status = await serviceStatus(runningIo.io, base);
   assert.equal(status.phase, "running");
   assert.equal(status.managed, false);
   assert.equal(status.version, "0.4.0.9");
 
-  writeServiceState(statePath, {
-    pid: 777,
-    endpoint: base.endpoint,
-    workspaceDir: directory,
-    pythonPath: "D:\\env\\python.exe",
-    command: "reme start",
-    startedAt: "2026-10-27T00:00:00.000Z",
-  });
+  record(statePath);
   status = await serviceStatus(runningIo.io, base);
   assert.equal(status.phase, "running");
   assert.equal(status.managed, true);
@@ -133,13 +146,49 @@ test("derives running, conflict and stopped phases", async (t) => {
   assert.equal(stopped.phase, "stopped");
 });
 
+test("a service on another endpoint is not reported as managed", async (t) => {
+  const { directory, statePath } = tempState(t, "reme-own-");
+  record(statePath, { endpoint: "http://127.0.0.1:2333" });
+  const { io } = fakeIO({
+    fetchHealth: async () => ({ ok: true, version: "0.4.0.9", error: "" }),
+  });
+  const status = await serviceStatus(io, {
+    endpoint: ENDPOINT,
+    statePath,
+    logPath: join(directory, "service.log"),
+  });
+  assert.equal(status.phase, "running");
+  assert.equal(status.managed, false);
+  assert.equal(status.pid, undefined);
+});
+
+test("a recycled pid that is no longer the service is not managed", async (t) => {
+  const { statePath } = tempState(t, "reme-pid-");
+  record(statePath);
+  const { io } = fakeIO({
+    fetchHealth: async () => ({ ok: true, version: "0.4.0.9", error: "" }),
+    processImage: async () => "node.exe",
+  });
+  assert.equal(
+    await ownsService(io, readServiceState(statePath), ENDPOINT),
+    false,
+  );
+  const { io: gone } = fakeIO({
+    probePid() {
+      throw new Error("ESRCH");
+    },
+  });
+  assert.equal(
+    await ownsService(gone, readServiceState(statePath), ENDPOINT),
+    false,
+  );
+});
+
 test("starts detached with the GBK-safe environment and records state", async (t) => {
-  const directory = mkdtempSync(join(tmpdir(), "reme-start-"));
-  t.after(() => rmSync(directory, { recursive: true, force: true }));
-  const statePath = join(directory, "service.json");
+  const { directory, statePath } = tempState(t, "reme-start-");
   const { io, calls } = fakeIO();
-  const result = startService(io, {
-    endpoint: "http://127.0.0.1:2334",
+  const result = await startService(io, {
+    endpoint: ENDPOINT,
     workspaceDir: directory,
     pythonPath: "D:\\env\\python.exe",
     cliPath: "D:\\env\\Scripts\\reme.exe",
@@ -159,22 +208,49 @@ test("starts detached with the GBK-safe environment and records state", async (t
   assert.equal(persisted?.workspaceDir, directory);
 });
 
-test("stops the managed pid and clears the state file", async (t) => {
-  const directory = mkdtempSync(join(tmpdir(), "reme-stop-"));
-  t.after(() => rmSync(directory, { recursive: true, force: true }));
-  const statePath = join(directory, "service.json");
-  writeServiceState(statePath, {
-    pid: 999,
-    endpoint: "http://127.0.0.1:2334",
+test("records nothing when the spawn fails", async (t) => {
+  const { directory, statePath } = tempState(t, "reme-spawn-");
+  const { io } = fakeIO({
+    async spawnService() {
+      throw new Error("spawn ENOENT");
+    },
+  });
+  const result = await startService(io, {
+    endpoint: ENDPOINT,
     workspaceDir: directory,
     pythonPath: "D:\\env\\python.exe",
-    command: "reme start",
-    startedAt: "2026-10-27T00:00:00.000Z",
+    cliPath: "D:\\missing\\reme.exe",
+    disableDreamCron: false,
+    statePath,
+    logPath: join(directory, "service.log"),
+    env: {},
   });
+  assert.equal(result.started, false);
+  assert.match(result.error, /ENOENT/);
+  assert.equal(readServiceState(statePath), undefined);
+});
+
+test("stops the managed pid and clears the state file", async (t) => {
+  const { directory, statePath } = tempState(t, "reme-stop-");
+  record(statePath, { pid: 999, workspaceDir: directory });
   const { io, calls } = fakeIO();
-  const stopped = stopService(io, { statePath });
+  const stopped = await stopService(io, { statePath, endpoint: ENDPOINT });
   assert.deepEqual(stopped, { stopped: true, pid: 999 });
   assert.deepEqual(calls.killTree, [999]);
   assert.equal(readServiceState(statePath), undefined);
-  assert.deepEqual(stopService(io, { statePath }), { stopped: false });
+  assert.deepEqual(await stopService(io, { statePath, endpoint: ENDPOINT }), {
+    stopped: false,
+    reason: "No service started by this plugin",
+  });
+});
+
+test("refuses to stop a service recorded for another endpoint", async (t) => {
+  const { statePath } = tempState(t, "reme-refuse-");
+  record(statePath, { endpoint: "http://127.0.0.1:2333" });
+  const { io, calls } = fakeIO();
+  const stopped = await stopService(io, { statePath, endpoint: ENDPOINT });
+  assert.equal(stopped.stopped, false);
+  assert.match(stopped.reason, /does not belong/);
+  assert.deepEqual(calls.killTree, []);
+  assert.ok(readServiceState(statePath), "the foreign record is left in place");
 });
diff --git a/integrations/dsh/tests/env-tab.test.mjs b/integrations/dsh/tests/env-tab.test.mjs
index 04e9a8fb..c75d3560 100644
--- a/integrations/dsh/tests/env-tab.test.mjs
+++ b/integrations/dsh/tests/env-tab.test.mjs
@@ -107,21 +107,11 @@ function texts(node, found = []) {
   return found;
 }
 
-async function mount() {
-  // The tab probes /health_check on mount; keep that hermetic.
-  const originalFetch = globalThis.fetch;
-  globalThis.fetch = async () =>
-    new Response(
-      JSON.stringify({
-        success: true,
-        metadata: {
-          health: { version: "0.4.0.9", healthy: true, components: {} },
-        },
-      }),
-      { status: 200, headers: { "Content-Type": "application/json" } },
-    );
-  const calls = { mutate: [] };
-  const snapshot = {
+async function mount(overrides = {}) {
+  const calls = { mutate: [], rpc: [] };
+  const listeners = new Set();
+  // A new object per change, the way the Host-backed form publishes a snapshot.
+  let snapshot = {
     status: "ready",
     writable: true,
     revision: 3,
@@ -132,10 +122,14 @@ async function mount() {
       autoStart: false,
       requestTimeoutMs: 10000,
       backgroundTimeoutMs: 3600000,
+      ...overrides,
     },
   };
   const scope = {
-    subscribe: () => () => undefined,
+    subscribe: (listener) => {
+      listeners.add(listener);
+      return () => listeners.delete(listener);
+    },
     getSnapshot: () => snapshot,
     mutate: async (ops, revision) => {
       calls.mutate.push({ ops, revision });
@@ -143,10 +137,10 @@ async function mount() {
     },
   };
   const rpc = {
-    call: async (_path, method) => ({
-      ok: true,
-      value: RESPONSES[method] ?? null,
-    }),
+    call: async (_path, method, payload) => {
+      calls.rpc.push({ method, input: payload?.args?.input });
+      return { ok: true, value: RESPONSES[method] ?? null };
+    },
   };
 
   let renderer;
@@ -161,9 +155,14 @@ async function mount() {
   return {
     renderer,
     calls,
-    restore: () => {
-      globalThis.fetch = originalFetch;
+    setValue(patch) {
+      snapshot = {
+        ...snapshot,
+        value: { ...snapshot.value, ...patch },
+      };
+      for (const listener of listeners) listener();
     },
+    restore: () => listeners.clear(),
   };
 }
 
@@ -250,3 +249,71 @@ test("marks the suggested interpreter and explains why", async (t) => {
     { op: "set", path: ["pythonPath"], value: ENVS[0].exe },
   ]);
 });
+
+test("drops .env drafts when the workspace changes", async (t) => {
+  const { renderer, calls, setValue, restore } = await mount();
+  t.after(restore);
+
+  await act(async () => {
+    buttonByText(renderer, envZh.expand).props.onClick();
+  });
+  const apiKey = () =>
+    renderer.root.findAll(
+      (node) => node.type === "input" && node.props.type === "password",
+    )[0];
+  await act(async () => {
+    apiKey().props.onChange({
+      target: { value: "sk-typed-for-first-workspace" },
+    });
+  });
+  assert.equal(apiKey().props.value, "sk-typed-for-first-workspace");
+
+  await act(async () => {
+    setValue({ remeWorkspaceDir: "/tmp/second-workspace" });
+  });
+  await act(async () => {
+    await new Promise((resolve) => setTimeout(resolve, 10));
+  });
+  // The draft belonged to the first workspace and is gone with it, so a save
+  // cannot carry a key typed for one workspace into another.
+  assert.equal(apiKey().props.value, "");
+
+  await act(async () => {
+    buttonByText(renderer, envZh.saveEnv).props.onClick();
+  });
+  const write = calls.rpc
+    .filter((call) => call.method === "remeEnv/writeEnv")
+    .at(-1);
+  assert.ok(write);
+  assert.equal(write.input.workspaceDir, "/tmp/second-workspace");
+  assert.deepEqual(write.input.values, {});
+});
+
+test("probes the service from the Host, never from the browser", async (t) => {
+  const original = globalThis.fetch;
+  let fetched = 0;
+  globalThis.fetch = async () => {
+    fetched += 1;
+    throw new Error("the tab must not fetch the service URL itself");
+  };
+  t.after(() => {
+    globalThis.fetch = original;
+  });
+  const { renderer, calls, restore } = await mount();
+  t.after(restore);
+  assert.equal(fetched, 0);
+  assert.match(
+    texts(renderer.toJSON()).join(" "),
+    /服务 · ReMe 0\.4\.0\.9/,
+    "the hero reports the version the Host service probe returned",
+  );
+  assert.ok(
+    calls.rpc.some((call) => call.method === "remeEnv/serviceStatus"),
+    "the hero reads its connection from the Host service probe",
+  );
+  assert.equal(
+    calls.rpc.some((call) => call.method === "remeEnv/health"),
+    false,
+    "a local endpoint is covered by the service probe already",
+  );
+});
diff --git a/integrations/dsh/tests/env-workspace.test.mjs b/integrations/dsh/tests/env-workspace.test.mjs
index ce0c672b..2654b2af 100644
--- a/integrations/dsh/tests/env-workspace.test.mjs
+++ b/integrations/dsh/tests/env-workspace.test.mjs
@@ -74,6 +74,32 @@ test("merges updates while preserving comments, order and unknown keys", () => {
   );
 });
 
+test("a duplicated key is rewritten everywhere, so parsing agrees", () => {
+  const existing = [
+    "LLM_API_KEY=old-key",
+    "LLM_MODEL_NAME=qwen3.7-plus",
+    "LLM_API_KEY=older-key",
+  ].join("\n");
+  const merged = mergeEnvFile(existing, { LLM_API_KEY: "new-key" });
+  assert.equal((merged.match(/^LLM_API_KEY=new-key$/gm) ?? []).length, 2);
+  // Parsing keeps the last occurrence, and that occurrence is the new value.
+  assert.equal(parseEnvFile(merged)["LLM_API_KEY"], "new-key");
+  assert.equal((merged.match(/^LLM_API_KEY=/gm) ?? []).length, 2);
+});
+
+test("keeps the file's own line ending", () => {
+  const merged = mergeEnvFile(
+    "LLM_BACKEND=openai\r\nEMBEDDING_API_KEY=EMPTY\r\n",
+    {
+      LLM_BACKEND: "dashscope",
+    },
+  );
+  assert.match(merged, /^LLM_BACKEND=dashscope\r$/m);
+  assert.equal(merged.includes("\n\n"), false);
+  assert.ok(merged.endsWith("\r\n"));
+  assert.equal(mergeEnvFile("", { A: "1" }), "A=1\n");
+});
+
 test("masks secrets down to the last four characters", () => {
   assert.equal(maskSecret("sk-abcdefgh"), "••••efgh");
   assert.equal(maskSecret("abc"), "••••");