fix(dsh): close startup races and validate env updates

integrations/dsh/docs/environment.md

@@ -197,6 +197,9 @@ processes it manages when the Host exits. The plugin owns a small service adapte
   already in flight. Before replacing a recorded launch that never became healthy, Start only kills it when
   ownership is proven; an unproven record is kept and reported, never silently dropped. The selected interpreter's
   dependencies are checked before stopping the previous service.
+  Status reads before spawn acknowledgement keep the launch in flight. If a newer service record appears during
+  probing or spawning, it is preserved and any unrecorded new child is stopped before reporting failure.
+  Service-record read errors abort the operation instead of being treated as an absent record.
 - **Auto-start** (`autoStart`, off by default) pulls the service up when DSH starts, if it is not already running.
   A refused launch is reported in the DSH Host log.
 
@@ -227,7 +230,8 @@ with other projects.
 use `EMPTY` for the embedding key when vector search is off. Emptying a secret field never reaches the file, so
 the stored key is kept; a required key that is not stored at all still fails the save, and the file is left
 unchanged. Only a missing file is treated as a first save; permission or I/O read errors abort the save without
-replacing the active file or its backup.
+replacing the active file or its backup. Names must start with a letter or underscore and contain only letters, digits and underscores; values must be single-line strings;
+invalid updates leave both files unchanged.
 
 **The service reports “Port occupied”.** Something else listens on the endpoint. Change the port in the service
 URL (Memory settings) or stop the other program.

integrations/dsh/docs/environment_ZH.md

@@ -154,6 +154,8 @@ ReMe 在服务启动时读取 workspace 目录下的 `.env`；标签页从不把
   停掉可证明归属的旧启动、或丢弃已证明属于外部的记录。只有本标签页发起的启动仍在进行时才会显示“启动中”。
 - **启动**是串行化的：两个请求不会拉起两个服务，后到的请求会观察已在进行的启动。替换“一直没有就绪”的旧记录时，只有
   归属被证明才会杀掉它；证明不了就保留记录并报错，不会静默丢弃。停止旧服务前会先检查所选解释器的启动依赖。
+  在操作系统确认启动之前，状态查询不会清除启动标记。探测或启动过程中出现更新的服务记录时，会保留新记录，
+  并在报告失败前停止尚未登记的新子进程；读取记录失败会中止操作，不会按“没有记录”继续启动。
 - **自动启动**（`autoStart`，默认关闭）会在 DSH 启动时把尚未运行的服务拉起；被拒绝的启动会记入 DSH Host 日志。
 
 Python 探测与 pip 安装都是有界的 Host 操作；服务是唯一需要比 DSH 活得更久的进程。
@@ -174,7 +176,7 @@ Python 探测与 pip 安装都是有界的 Host 操作；服务是唯一需要
 
 **所选解释器不在列表里。** 标签页会用“不在探测结果中”标注它，但仍然使用它。重新探测，或改选探测到的解释器。
 
-**保存 `.env` 时读取失败。** 只有文件不存在才按首次保存处理；权限或 I/O 错误会中止保存，不会替换原文件或备份。
+**保存 `.env` 时读取失败。** 只有文件不存在才按首次保存处理；权限或 I/O 错误会中止保存，不会替换原文件或备份。变量名须以字母或下划线开头，其余只接受字母、数字和下划线；值必须是单行字符串；无效更新不会改动原文件或备份。
 
 **安装失败。** 先看日志尾部：缺少 wheel、代理配置过期或网络受限通常都能看出来，可以换一个软件源。conda 环境里 `pip` 也可
 用，但独立环境更能避免与其它项目的依赖冲突。

integrations/dsh/src/environment/host/manager.ts

@@ -16,6 +16,7 @@ import {
   processAlive,
   readServiceState,
   removeServiceState,
+  sameServiceRecord,
   serviceEndpointKey,
   serviceOriginKey,
   serviceOwnership,
@@ -82,7 +83,7 @@ export class ReMeEnvManager {
   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 };
+  private starting?: { endpoint: string; at: number; pid?: number };
 
   constructor(
     env: NodeJS.ProcessEnv = process.env,
@@ -210,6 +211,8 @@ export class ReMeEnvManager {
   /** Current service phase (running / conflict / stopped) and ownership. */
   async serviceStatus(endpoint: string, authToken = ""): Promise<ServiceState> {
     const endpointValue = this.endpoint(endpoint);
+    const starting = this.starting;
+    const acknowledged = starting?.pid !== undefined;
     // A record left under a path prefix of the same origin describes the
     // service this endpoint talks to. It is re-keyed to the endpoint on screen
     // first, so the tab can manage (stop) it and the start path can replace
@@ -232,7 +235,11 @@ export class ReMeEnvManager {
       authToken,
       logPath: join(logDir, "service.log"),
     });
-    if (this.starting?.endpoint === endpointValue) {
+    if (
+      starting &&
+      this.starting === starting &&
+      starting.endpoint === endpointValue
+    ) {
       // Running and conflict both mean the port answered: the launch is no
       // longer "in flight". Stopped without a live pid means it died before it
       // ever listened, so a retry must not be blocked either. A stopped status
@@ -240,7 +247,7 @@ export class ReMeEnvManager {
       if (
         status.phase === "running" ||
         status.phase === "conflict" ||
-        (status.phase === "stopped" && status.pid === undefined)
+        (acknowledged && status.phase === "stopped" && status.pid === undefined)
       )
         this.starting = undefined;
     }
@@ -510,15 +517,10 @@ export class ReMeEnvManager {
       return;
     if (!this.sameOrigin(record.endpoint, endpointValue)) return;
     const current = readServiceState(this.serviceStatePath);
-    if (
-      !current ||
-      current.pid !== record.pid ||
-      current.endpoint !== record.endpoint
-    )
-      return;
+    if (!sameServiceRecord(current, record)) return;
     try {
       writeServiceState(this.serviceStatePath, {
-        ...current,
+        ...record,
         endpoint: endpointValue,
       });
     } catch {
@@ -636,6 +638,15 @@ export class ReMeEnvManager {
       };
       return this.refused(logPath, now, code, errors[code]);
     }
+    const inspected = readServiceState(this.serviceStatePath);
+    if (!sameServiceRecord(inspected, record))
+      return this.refused(
+        logPath,
+        now,
+        "previous-not-stopped",
+        "The service record changed during the dependency check; retry with its current state",
+        { record: serviceRecordRef(inspected) },
+      );
     // All replacement cases share this path. Validate the new interpreter
     // before stopping anything, then re-verify ownership at the stop boundary.
     if (replace && record) {
@@ -663,6 +674,7 @@ export class ReMeEnvManager {
     // booting observes the first launch instead of spawning another one. It is
     // cleared again when the spawn fails, so a retry is possible immediately.
     this.starting = { endpoint: endpointValue, at: Date.now() };
+    const launch = this.starting;
     try {
       const result = await startService(this.serviceIO, {
         endpoint: endpointValue,
@@ -676,6 +688,7 @@ export class ReMeEnvManager {
         env: this.env,
       });
       if (!result.started) this.starting = undefined;
+      else if (result.pid !== undefined) launch.pid = result.pid;
       return result;
     } catch (error) {
       this.starting = undefined;

integrations/dsh/src/environment/host/service.ts

@@ -137,9 +137,16 @@ function isRecord(value: unknown): value is Record<string, unknown> {
 
 /** Read a record defensively: a malformed file is "no record", not a crash. */
 export function readServiceState(path: string): PersistedService | undefined {
+  let text: string;
+  try {
+    text = readFileSync(path, "utf8");
+  } catch (error) {
+    if ((error as NodeJS.ErrnoException).code === "ENOENT") return undefined;
+    throw error;
+  }
   let value: unknown;
   try {
-    value = JSON.parse(readFileSync(path, "utf8"));
+    value = JSON.parse(text);
   } catch {
     return undefined;
   }
@@ -157,6 +164,21 @@ export function readServiceState(path: string): PersistedService | undefined {
   };
 }
 
+/** Compare observed records before changing ownership metadata after an await. */
+export function sameServiceRecord(
+  left: PersistedService | undefined,
+  right: PersistedService | undefined,
+): boolean {
+  return (
+    left?.pid === right?.pid &&
+    left?.endpoint === right?.endpoint &&
+    left?.startedAt === right?.startedAt &&
+    left?.workspaceDir === right?.workspaceDir &&
+    left?.pythonPath === right?.pythonPath &&
+    left?.command === right?.command
+  );
+}
+
 export function writeServiceState(path: string, state: PersistedService): void {
   mkdirSync(dirname(path), { recursive: true });
   const temporary = `${path}.${process.pid}.tmp`;
@@ -522,6 +544,12 @@ export async function startService(
     };
   }
   try {
+    // Another manager may have published a launch while spawn acknowledgement
+    // was pending. Keep its record and terminate only our unrecorded child.
+    if (readServiceState(options.statePath))
+      throw new Error(
+        "Another service was recorded while this launch was starting",
+      );
     writeServiceState(options.statePath, {
       pid: child.pid,
       endpoint: options.endpoint,
@@ -533,10 +561,19 @@ export async function startService(
   } catch (error) {
     // The process exists but could not be recorded; leaving it running would
     // make it unmanageable, so stop it before reporting the failure.
-    try {
-      if (child.pid !== undefined) io.killTree(child.pid, true);
-    } catch {
-      // Best effort: the error below already explains the launch failed.
+    let exited = child.pid === undefined;
+    if (child.pid !== undefined) {
+      try {
+        io.killTree(child.pid, true);
+      } catch {
+        // Still confirm whether it exited before reporting the failure.
+      }
+      exited = await waitForExit(
+        io,
+        child.pid,
+        STOP_FORCE_TIMEOUT_MS,
+        STOP_POLL_MS,
+      );
     }
     return {
       started: false,
@@ -544,7 +581,10 @@ export async function startService(
       command: command.display,
       logPath: options.logPath,
       startedAt,
-      error: `Could not record the ReMe service: ${safeError(error)}`,
+      error: `Could not record the ReMe service: ${safeError(error)}${
+        exited ? "" : `; process ${child.pid} could not be confirmed stopped`
+      }`,
+      ...(!exited && child.pid !== undefined ? { pid: child.pid } : {}),
       code: "start-failed",
     };
   }
@@ -639,11 +679,7 @@ export async function stopService(
   const pid = persisted.pid;
   const isCurrentRecord = (): boolean => {
     const current = readServiceState(options.statePath);
-    return (
-      current?.pid === pid &&
-      current.endpoint === persisted.endpoint &&
-      current.startedAt === persisted.startedAt
-    );
+    return sameServiceRecord(current, persisted);
   };
   // Identity reads yield to the event loop. Never signal a record that was
   // replaced while those reads were in flight, or erase a newer launch later.

integrations/dsh/src/environment/host/workspace.ts

@@ -63,6 +63,12 @@ export function mergeEnvFile(
   updates: Record<string, string>,
 ): string {
   const pending = new Map(Object.entries(updates));
+  for (const [key, value] of pending) {
+    if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(key))
+      throw new TypeError("Invalid environment variable name");
+    if (typeof value !== "string" || /[\r\n\0]/.test(value))
+      throw new TypeError("Environment values must be single-line strings");
+  }
   const eol = /\r\n/.test(existingText) ? "\r\n" : "\n";
   const lines = existingText.length ? existingText.split(/\r?\n/) : [];
   const output: string[] = [];

integrations/dsh/tests/env-manager.test.mjs

@@ -66,6 +66,176 @@ function managerIn(t) {
   return { manager, calls, directory, io };
 }
 
+function deferred() {
+  let resolve;
+  const promise = new Promise((done) => {
+    resolve = done;
+  });
+  return { promise, resolve };
+}
+
+for (const delayed of [false, true]) {
+  test(`a ${
+    delayed ? "delayed" : "completed"
+  } status read before spawn acknowledgement preserves the start`, async (t) => {
+    const { manager, calls, directory, io } = managerIn(t);
+    manager.serviceStatus = ReMeEnvManager.prototype.serviceStatus;
+    const spawned = deferred(),
+      acknowledgement = deferred();
+    const statusEntered = deferred(),
+      statusReply = deferred();
+    t.after(() => {
+      acknowledgement.resolve();
+      statusReply.resolve();
+    });
+    let pauseStatus = false;
+    let running = true;
+    io.portOpen = async () => {
+      if (pauseStatus) {
+        pauseStatus = false;
+        statusEntered.resolve();
+        await statusReply.promise;
+      }
+      return false;
+    };
+    io.spawnService = async (options) => {
+      calls.spawn.push(options);
+      spawned.resolve();
+      if (calls.spawn.length === 1) await acknowledgement.promise;
+      running = true;
+      return { pid: 5000 + calls.spawn.length };
+    };
+    io.probePid = () => {
+      if (!running) throw Object.assign(new Error("gone"), { code: "ESRCH" });
+    };
+    io.killTree = (pid) => {
+      calls.killTree.push(pid);
+      running = false;
+    };
+    io.processCommandLine = async () =>
+      `reme start workspace_dir=${directory} service.port=2334`;
+    const options = ["http://127.0.0.1:2334", "/env/bin/python", directory];
+    const first = manager.serviceStart(...options);
+    await spawned.promise;
+    const second = manager.serviceStart(...options);
+    pauseStatus = delayed;
+    const status = manager.serviceStatus(options[0]);
+    if (delayed) {
+      await statusEntered.promise;
+      acknowledgement.resolve();
+      await first;
+      statusReply.resolve();
+    } else {
+      await status;
+      acknowledgement.resolve();
+    }
+    const [started, repeated] = await Promise.all([first, second, status]);
+    assert.equal(started.started, true);
+    assert.equal(repeated.alreadyRunning, true);
+    assert.equal(calls.spawn.length, 1);
+    assert.deepEqual(calls.killTree, []);
+    assert.equal(
+      readServiceState(join(directory, "reme-memory", "service.json"))?.pid,
+      5001,
+    );
+  });
+}
+
+test("an acknowledged launch that exits before listening can be retried", async (t) => {
+  const { manager, calls, directory, io } = managerIn(t);
+  manager.serviceStatus = ReMeEnvManager.prototype.serviceStatus;
+  io.probePid = () => {
+    throw Object.assign(new Error("gone"), { code: "ESRCH" });
+  };
+  const options = ["http://127.0.0.1:2334", "/env/bin/python", directory];
+  assert.equal((await manager.serviceStart(...options)).started, true);
+  const status = await manager.serviceStatus(options[0]);
+  assert.equal(status.phase, "stopped");
+  assert.equal(status.pid, undefined);
+  assert.equal((await manager.serviceStart(...options)).started, true);
+  assert.equal(calls.spawn.length, 2);
+});
+
+for (const initial of ["missing", "dead", "foreign"]) {
+  test(`a ${initial} record snapshot cannot overwrite a launch published during the dependency check`, async (t) => {
+    const { manager, calls, directory, io } = managerIn(t);
+    const statePath = join(directory, "reme-memory", "service.json");
+    const record = {
+      pid: 7000,
+      endpoint: "http://127.0.0.1:2334",
+      workspaceDir: directory,
+      pythonPath: "/other/bin/python",
+      command: "reme start",
+      startedAt: "new",
+    };
+    if (initial !== "missing")
+      writeServiceState(statePath, { ...record, pid: 6000, startedAt: "old" });
+    io.probePid = () => {
+      if (initial === "dead")
+        throw Object.assign(new Error("gone"), { code: "ESRCH" });
+    };
+    io.processImage = async () => "node";
+    const check = manager.checkReme;
+    manager.checkReme = async (path) => {
+      writeServiceState(statePath, record);
+      return check(path);
+    };
+    const result = await manager.serviceStart(
+      record.endpoint,
+      "/env/bin/python",
+      directory,
+    );
+    assert.equal(result.started, false);
+    assert.equal(result.code, "previous-not-stopped");
+    assert.deepEqual(result.record, {
+      endpoint: record.endpoint,
+      pid: record.pid,
+    });
+    assert.deepEqual(calls.spawn, []);
+    assert.deepEqual(calls.killTree, []);
+    assert.deepEqual(readServiceState(statePath), record);
+  });
+}
+
+for (const code of ["EIO", "EACCES"]) {
+  test(`a ${code} service-record read failure cannot discard ownership`, async (t) => {
+    const { manager, calls, directory } = managerIn(t);
+    manager.serviceStatus = ReMeEnvManager.prototype.serviceStatus;
+    const statePath = join(directory, "reme-memory", "service.json");
+    writeServiceState(statePath, {
+      pid: 7000,
+      endpoint: "http://127.0.0.1:2333",
+      workspaceDir: directory,
+      pythonPath: "/env/bin/python",
+      command: "reme start",
+      startedAt: "old",
+    });
+    const original = readFileSync(statePath, "utf8");
+    const read = fs.readFileSync;
+    const stub = t.mock.method(fs, "readFileSync", (path, ...args) => {
+      if (path === statePath) throw Object.assign(new Error(code), { code });
+      return read(path, ...args);
+    });
+    syncBuiltinESMExports();
+    try {
+      await assert.rejects(
+        manager.serviceStart(
+          "http://127.0.0.1:2334",
+          "/env/bin/python",
+          directory,
+        ),
+        { code },
+      );
+    } finally {
+      stub.mock.restore();
+      syncBuiltinESMExports();
+    }
+    assert.equal(readFileSync(statePath, "utf8"), original);
+    assert.deepEqual(calls.spawn, []);
+    assert.deepEqual(calls.killTree, []);
+  });
+}
+
 test("serializes concurrent starts into one service launch", async (t) => {
   const { manager, calls, directory } = managerIn(t);
   const options = [
@@ -1029,3 +1199,21 @@ test("a missing .env can still be created on the first save", (t) => {
     /^LLM_API_KEY=new$/m,
   );
 });
+
+test(".env updates cannot inject additional configuration lines", (t) => {
+  const { manager, directory } = managerIn(t);
+  const envPath = join(directory, ".env");
+  const original = "UNRELATED=keep\nLLM_API_KEY=old\nEMBEDDING_API_KEY=EMPTY\n";
+  writeFileSync(envPath, original);
+  writeFileSync(`${envPath}.bak`, "previous backup");
+  for (const values of [
+    { LLM_API_KEY: "new\nUNRELATED=changed" },
+    { LLM_API_KEY: "new\r\nUNRELATED=changed" },
+    { "LLM_API_KEY\nUNRELATED": "changed" },
+    { LLM_API_KEY: 123 },
+  ]) {
+    assert.throws(() => manager.writeEnv(directory, values), TypeError);
+    assert.equal(readFileSync(envPath, "utf8"), original);
+    assert.equal(readFileSync(`${envPath}.bak`, "utf8"), "previous backup");
+  }
+});

integrations/dsh/tests/env-service.test.mjs

@@ -1,7 +1,7 @@
 import assert from "node:assert/strict";
 import test from "node:test";
 import { createServer } from "node:http";
-import { mkdtempSync, readFileSync, rmSync } from "node:fs";
+import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 
@@ -476,6 +476,62 @@ test("records nothing when the spawn fails", async (t) => {
   assert.equal(readServiceState(statePath), undefined);
 });
 
+test("missing and malformed service records remain rebuildable", (t) => {
+  const { statePath } = tempState(t, "reme-malformed-record-");
+  assert.equal(readServiceState(statePath), undefined);
+  for (const text of ["{", "{}", '{"endpoint":7}']) {
+    writeFileSync(statePath, text);
+    assert.equal(readServiceState(statePath), undefined);
+  }
+});
+
+test("a launch never overwrites a service recorded before its spawn acknowledgement", async (t) => {
+  const { directory, statePath } = tempState(t, "reme-publish-race-");
+  let running = true;
+  let group = true;
+  const { io, calls } = fakeIO({
+    async spawnService() {
+      record(statePath, { pid: 777 });
+      return { pid: 4321 };
+    },
+    killTree(pid, force) {
+      calls.killTree.push(pid);
+      assert.equal(force, true);
+      running = false;
+      setTimeout(() => {
+        group = false;
+      }, 5);
+    },
+    probePid() {
+      if (!running) throw Object.assign(new Error("gone"), { code: "ESRCH" });
+    },
+    groupAlive: () => group,
+  });
+  const result = await startService(io, {
+    endpoint: ENDPOINT,
+    workspaceDir: directory,
+    pythonPath: "/env/bin/python",
+    cliPath: "",
+    disableDreamCron: false,
+    statePath,
+    logPath: join(directory, "service.log"),
+    env: {},
+  });
+  assert.equal(result.started, false);
+  assert.equal(result.code, "start-failed");
+  assert.equal(
+    group,
+    false,
+    "the unrecorded process group must exit before failure is reported",
+  );
+  assert.deepEqual(
+    calls.killTree,
+    [4321],
+    "only the new unrecorded launch is stopped",
+  );
+  assert.equal(readServiceState(statePath)?.pid, 777);
+});
+
 test("stops the managed pid and clears the state file after it exits", async (t) => {
   const { directory, statePath } = tempState(t, "reme-stop-");
   record(statePath, { pid: 999, workspaceDir: directory });