From a91022a5bc4d881fda4789ed5756dd014c5545f6 Mon Sep 17 00:00:00 2001 From: LiangshouX Date: Wed, 28 Oct 2026 23:42:02 +0800 Subject: [PATCH] fix(dsh): verify service identity before touching a recorded pid - require command-line evidence for owned service processes - discard unprovable stale records instead of blocking startup - identify and stop our own unhealthy conflict; serialize stop with start - reject IPv6 loopback endpoints - rotate oversized logs and fix tailing multibyte cuts - add tests for env writes, client failures and real spawner --- integrations/dsh/docs/environment.md | 25 ++- integrations/dsh/docs/environment_ZH.md | 20 +- .../dsh/src/environment/client/service.tsx | 14 +- .../dsh/src/environment/client/tab.tsx | 2 +- .../dsh/src/environment/host/detect.ts | 4 + .../dsh/src/environment/host/install.ts | 18 +- integrations/dsh/src/environment/host/log.ts | 24 +++ .../dsh/src/environment/host/manager.ts | 147 +++++++++++--- .../dsh/src/environment/host/service.ts | 182 +++++++++++++++--- integrations/dsh/tests/env-manager.test.mjs | 123 +++++++++++- integrations/dsh/tests/env-reme.test.mjs | 3 +- integrations/dsh/tests/env-runner.test.mjs | 46 +++++ integrations/dsh/tests/env-service.test.mjs | 84 +++++++- integrations/dsh/tests/env-tab.test.mjs | 94 ++++++++- 14 files changed, 693 insertions(+), 93 deletions(-) create mode 100644 integrations/dsh/src/environment/host/log.ts diff --git a/integrations/dsh/docs/environment.md b/integrations/dsh/docs/environment.md index 08ad8eab..bb81c112 100644 --- a/integrations/dsh/docs/environment.md +++ b/integrations/dsh/docs/environment.md @@ -70,7 +70,8 @@ The installation is a Host-owned job: - Progress is polled by job id, so a second installation for another interpreter cannot replace what you see. - **Cancel** stops pip and its build children, and the interpreter stays locked until the process has actually exited. -- The log is written to `$DSH_HOME/reme-memory/install-.log` and its tail is shown in place. +- The log is written to `$DSH_HOME/reme-memory/install-.log` and its tail is shown in place; install and + service logs are rotated to `.log.1` once they pass 5 MB. - When the plugin unloads, running jobs are stopped. ## Workspace `.env` @@ -117,18 +118,22 @@ processes it manages when the Host exits. The plugin owns a small service adapte - The launch is recorded in `$DSH_HOME/reme-memory/service.json` (pid, endpoint, workspace, command, start time) **after** the OS acknowledged the spawn, so a missing interpreter leaves no state behind. - A service is reported as managed — and offered a **Stop** button — only when the record matches the endpoint on - screen, the pid is still alive, and its image is still a Python or `reme` executable. A record for port 2333 - never claims a service on port 2444, and a recycled pid is not mistaken for this service. When the image cannot - be read, ownership fails closed: the process is shown as foreign rather than offered a Stop button. + screen, the pid is still alive, its image is still a Python or `reme` executable, **and** its command line still + carries the recorded `workspace_dir` and `service.port`. The command line is what makes a recycled pid safe: an + image name alone cannot prove which launch a pid belongs to. A record for port 2333 never claims a service on + port 2444. When the evidence cannot be read, ownership stays unproven: the process is shown as foreign, Stop + refuses to touch it, and a later Start drops the stale record instead of being blocked forever. +- A service that occupies the port but fails the health check is reported as a conflict; if it is provably this + plugin's service, the tab still offers **Stop**. - **Stop** tree-kills the recorded process (process group on POSIX, `taskkill /T /F` on Windows), waits for it to - exit, escalates to `SIGKILL` if it ignores the graceful signal, and clears the record only after the pid is - gone. A service started elsewhere stays untouched and is only observed. -- **Start** is serialized: two requests cannot spawn two services, and a request that arrives while a launch is - still in flight is told so instead of silently pretending to start. + exit, escalates (POSIX: `SIGTERM` then `SIGKILL`; Windows: force `taskkill` again), and clears the record only + after the pid is gone. A service started elsewhere stays untouched and is only observed. +- **Start** is serialized: two requests cannot spawn two services, and a second request observes the launch + already in flight. Before replacing a recorded launch that never became healthy, Start only kills it when + ownership is proven; otherwise it drops the record without touching the process. - **Auto-start** (`autoStart`, off by default) pulls the service up when DSH starts, if it is not already running. -Python probes and pip installs are bounded Host operations; the service is the only intentionally detached -process. +Python probes and pip installs are bounded Host operations; the service is the only process that must outlive DSH. ## Platform support diff --git a/integrations/dsh/docs/environment_ZH.md b/integrations/dsh/docs/environment_ZH.md index 2e9a2945..ef4ac4f3 100644 --- a/integrations/dsh/docs/environment_ZH.md +++ b/integrations/dsh/docs/environment_ZH.md @@ -60,7 +60,7 @@ ReMe 需要 **Python 3.11 或更新版本**。标签页在 Host 上探测下列 - 关闭配置页后仍在运行；重新打开标签页会为同一个解释器恢复进度。 - 进度按任务 ID 轮询，为另一个解释器发起的安装不会顶替当前显示。 - **取消**会同时终止 pip 及其构建子进程，且在进程真正退出前该解释器保持占用。 -- 日志写入 `$DSH_HOME/reme-memory/install-.log`，页面直接显示尾部。 +- 日志写入 `$DSH_HOME/reme-memory/install-.log`，页面直接显示尾部；安装与服务日志超过 5 MB 会轮转为 `.log.1`。 - 插件卸载时，仍在运行的安装会被停止。 ## Workspace `.env` @@ -97,15 +97,19 @@ ReMe 在服务启动时读取 workspace 目录下的 `.env`；标签页从不把 - 启动在操作系统确认 spawn 之后才记录到 `$DSH_HOME/reme-memory/service.json`（pid、地址、workspace、命令行、启动时间）， 所以解释器缺失时不会留下残留状态。 -- 只有记录与当前地址一致、pid 仍存活、且进程映像仍是 Python 或 `reme` 时，服务才被判定为“由本插件托管”并给出**停止**按钮。 - 2333 端口上的记录绝不会声称管理 2444 端口上的服务，被复用的 pid 也不会被误认。读不到进程映像时按失败关闭处理： - 显示为外部进程，不给出停止按钮。 -- **停止**会树杀记录的进程（POSIX 上按进程组、Windows 上为 `taskkill /T /F`），等待其退出，忽略优雅信号时升级为 - `SIGKILL`，并**在 pid 真正消失之后**才清除记录。别处启动的服务不受影响，只会被观察。 -- **启动**是串行化的：两个请求不会拉起两个服务；启动仍在进行时到达的请求会被告知，而不是假装正在启动。 +- 只有记录与当前地址一致、pid 仍存活、进程映像仍是 Python 或 `reme`，**且命令行仍包含记录的 `workspace_dir` 与 + `service.port`** 时，服务才被判定为“由本插件托管”并给出**停止**按钮。命令行是防止 pid 复用的关键证据：只看映像名 + 无法证明 pid 属于哪次启动。2333 端口上的记录绝不会声称管理 2444 端口上的服务。证据读不到时按失败关闭处理：显示为 + 外部进程，停止操作不会碰它；之后点启动会丢弃这条过期记录，而不是永久卡住。 +- 服务占用端口但健康检查失败时显示为“端口被占用”；若能证明是本插件启动的，标签页仍给出**停止**按钮。 +- **停止**会树杀记录的进程（POSIX 上按进程组、Windows 上为 `taskkill /T /F`），等待其退出，必要时升级（POSIX： + `SIGTERM` 后 `SIGKILL`；Windows：再次强制 `taskkill`），并**在 pid 真正消失之后**才清除记录。别处启动的服务不受影响， + 只会被观察。 +- **启动**是串行化的：两个请求不会拉起两个服务，后到的请求会观察已在进行的启动。替换“一直没有就绪”的旧记录时，只有 + 归属被证明才会杀掉它；证明不了就只丢弃记录，不碰进程。 - **自动启动**（`autoStart`，默认关闭）会在 DSH 启动时把尚未运行的服务拉起。 -Python 探测与 pip 安装都是有界的 Host 操作；服务是唯一被刻意分离出去的进程。 +Python 探测与 pip 安装都是有界的 Host 操作；服务是唯一需要比 DSH 活得更久的进程。 ## 平台支持 diff --git a/integrations/dsh/src/environment/client/service.tsx b/integrations/dsh/src/environment/client/service.tsx index a15d88e7..cafe02d9 100644 --- a/integrations/dsh/src/environment/client/service.tsx +++ b/integrations/dsh/src/environment/client/service.tsx @@ -289,6 +289,12 @@ export function ServiceSection({ }): JSX.Element { const { state, stopping, stopError, log } = control; const [confirming, setConfirming] = useState(false); + // A conflict can still be our own unhealthy service (booting, wrong token); + // offer the same Stop affordance instead of leaving the tab without a way + // out other than editing the record by hand. + const canStop = + state.phase === "running" || + (state.phase === "conflict" && state.data?.managed === true); return (
@@ -338,11 +344,13 @@ export function ServiceSection({ {t("recheck")} - {state.phase === "running" ? ( + {state.phase === "running" || state.phase === "conflict" ? (

{state.data?.managed ? fill(t("managedNote"), { pid: state.data?.pid ?? "" }) - : t("foreignNote")} + : state.phase === "running" + ? t("foreignNote") + : null}

) : null} {state.phase === "conflict" ? ( @@ -359,7 +367,7 @@ export function ServiceSection({

) : null}
- {state.phase === "running" ? ( + {canStop ? ( confirming ? (

{t("stopConfirm")}

diff --git a/integrations/dsh/src/environment/client/tab.tsx b/integrations/dsh/src/environment/client/tab.tsx index b9d570e4..ac2e8c2c 100644 --- a/integrations/dsh/src/environment/client/tab.tsx +++ b/integrations/dsh/src/environment/client/tab.tsx @@ -32,7 +32,7 @@ function isLocalEndpoint(endpoint: string): boolean { const url = new URL(endpoint); return ( url.protocol === "http:" && - ["127.0.0.1", "localhost", "[::1]", "::1"].includes(url.hostname) + ["127.0.0.1", "localhost"].includes(url.hostname) ); } catch { return false; diff --git a/integrations/dsh/src/environment/host/detect.ts b/integrations/dsh/src/environment/host/detect.ts index a2a7ed3a..42231540 100644 --- a/integrations/dsh/src/environment/host/detect.ts +++ b/integrations/dsh/src/environment/host/detect.ts @@ -483,11 +483,15 @@ export async function detectPythonEnvs(io: DetectIO): Promise { const first = windows ? await io.run("where.exe", ["python"], { timeoutMs: 5_000 }) : await io.run("which", ["-a", "python3"], { timeoutMs: 5_000 }); + // A non-zero exit just means "no match"; a spawn/timeout error is a real + // diagnostic and must not disappear from the report. + if (first.error) errors.push(`python: ${safeError(first.error)}`); let stdout = first.stdout; if (!windows) { const extra = await io.run("which", ["-a", "python"], { timeoutMs: 5_000, }); + if (extra.error) errors.push(`python: ${safeError(extra.error)}`); stdout = `${stdout}\n${extra.stdout}`; } for (const exe of parsePathList(stdout)) { diff --git a/integrations/dsh/src/environment/host/install.ts b/integrations/dsh/src/environment/host/install.ts index 662b0508..860c5f91 100644 --- a/integrations/dsh/src/environment/host/install.ts +++ b/integrations/dsh/src/environment/host/install.ts @@ -3,6 +3,7 @@ import { closeSync, fstatSync, mkdirSync, openSync, readSync } from "node:fs"; import { join } from "node:path"; import type { InstallPhase, InstallRequest, InstallStatus } from "../types.js"; +import { openAppendLog } from "./log.js"; import { isAbsolutePath } from "./paths.js"; export interface InstallChild { @@ -57,12 +58,17 @@ export function tailLog(path: string, maxLines = 30, maxBytes = 8192): string { fd = openSync(path, "r"); const size = fstatSync(fd).size; const length = Math.min(size, maxBytes); + const start = size - length; const buffer = Buffer.alloc(length); - readSync(fd, buffer, 0, length, Math.max(0, size - length)); - const text = buffer - .toString("utf8") - .replaceAll("\r\n", "\n") - .replaceAll("\r", "\n"); + readSync(fd, buffer, 0, length, start); + let text = buffer.toString("utf8"); + if (start > 0) { + // The cut may be mid-line and even mid-character; drop that partial line + // so the tail never starts with a replacement character. + const firstBreak = text.indexOf("\n"); + text = firstBreak >= 0 ? text.slice(firstBreak + 1) : text; + } + text = text.replaceAll("\r\n", "\n").replaceAll("\r", "\n"); return text .split("\n") .filter((line) => line.trim().length > 0) @@ -87,7 +93,7 @@ export function createInstallSpawner( ): InstallSpawner { return { start({ exe, args, logPath }) { - const fd = openSync(logPath, "a"); + const fd = openAppendLog(logPath); let child: ReturnType; try { child = spawn(exe, args, { diff --git a/integrations/dsh/src/environment/host/log.ts b/integrations/dsh/src/environment/host/log.ts new file mode 100644 index 00000000..c4bb2317 --- /dev/null +++ b/integrations/dsh/src/environment/host/log.ts @@ -0,0 +1,24 @@ +import { mkdirSync, openSync, renameSync, statSync } from "node:fs"; +import { dirname } from "node:path"; + +/** Rotate a log once it grows past this size; only one backup is kept. */ +export const LOG_ROTATE_BYTES = 5 * 1024 * 1024; + +/** + * Open a log for appending, rotating an oversized one to `.1` first. + * Install and service logs otherwise grow without bound because both are + * append-only and outlive many sessions. + */ +export function openAppendLog( + path: string, + maxBytes = LOG_ROTATE_BYTES, +): number { + mkdirSync(dirname(path), { recursive: true }); + try { + if (statSync(path).size > maxBytes) renameSync(path, `${path}.1`); + } catch { + // A missing log has nothing to rotate; a locked one (Windows, while the + // service still writes it) is left alone rather than failing the launch. + } + return openSync(path, "a"); +} diff --git a/integrations/dsh/src/environment/host/manager.ts b/integrations/dsh/src/environment/host/manager.ts index b7b71400..7992cf26 100644 --- a/integrations/dsh/src/environment/host/manager.ts +++ b/integrations/dsh/src/environment/host/manager.ts @@ -9,7 +9,11 @@ import { checkReme, MINIMUM_REME, remeVersionAtLeast } from "./reme.js"; import { createNodeRunner } from "./runner.js"; import { createServiceIO, + processAlive, readServiceState, + removeServiceState, + serviceEndpointKey, + serviceOwnership, serviceStatus, startService, stopService, @@ -181,21 +185,30 @@ export class ReMeEnvManager { async serviceStatus(endpoint: string, authToken = ""): Promise { const endpointValue = this.endpoint(endpoint); const persisted = readServiceState(this.serviceStatePath); + // A malformed record must not turn the log path into a relative one. + let logDir = resolveWorkspaceDir("", homedir()); + const recordedDir = persisted?.workspaceDir?.trim() ?? ""; + if (recordedDir) { + try { + logDir = resolveWorkspaceDir(recordedDir, homedir()); + } catch { + // Keep the default log directory. + } + } const status = await serviceStatus(this.serviceIO, { endpoint: endpointValue, statePath: this.serviceStatePath, authToken, - logPath: join( - persisted?.workspaceDir ?? resolveWorkspaceDir("", homedir()), - "service.log", - ), + logPath: join(logDir, "service.log"), }); if (this.starting?.endpoint === endpointValue) { - // Running is the happy end; stopped without a live pid means the launch - // died before it ever listened, so a retry must not be blocked by the - // grace period. A stopped status *with* a pid is a service still booting. + // 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 + // *with* a pid is a service still booting. if ( status.phase === "running" || + status.phase === "conflict" || (status.phase === "stopped" && status.pid === undefined) ) this.starting = undefined; @@ -224,14 +237,20 @@ export class ReMeEnvManager { return run; } - /** Stop the plugin-managed service; a foreign service stays untouched. */ - async serviceStop(endpoint: string): Promise { - const result = await stopService(this.serviceIO, { - statePath: this.serviceStatePath, - endpoint: this.endpoint(endpoint), - }); - if (result.stopped) this.starting = undefined; - return result; + /** + * Stop the plugin-managed service; a foreign service stays untouched. Stop + * joins the start queue, so a request that arrives while a launch is still + * being recorded waits for it and then stops that launch instead of + * reporting "nothing started". + */ + serviceStop(endpoint: string): Promise { + const endpointValue = this.endpoint(endpoint); + const run = this.startQueue.then(() => this.stopLocked(endpointValue)); + this.startQueue = run.then( + () => undefined, + () => undefined, + ); + return run; } /** Tail of the workspace service log. */ @@ -288,6 +307,16 @@ export class ReMeEnvManager { return `${url.protocol}//${url.host}${url.pathname.replace(/\/+$/, "")}`; } + private async stopLocked(endpointValue: string): Promise { + const result = await stopService(this.serviceIO, { + statePath: this.serviceStatePath, + endpoint: endpointValue, + }); + if (result.stopped && this.starting?.endpoint === endpointValue) + this.starting = undefined; + return result; + } + private async startLocked( endpoint: string, pythonPath: string, @@ -301,13 +330,14 @@ export class ReMeEnvManager { const starting = this.starting; if (starting && Date.now() - starting.at < START_GRACE_MS) { if (starting.endpoint === endpointValue) + // A genuine launch is in flight. Observing it is not an error: the + // client keeps polling just as the first request does. return { started: false, alreadyRunning: true, command: "", logPath, startedAt: now, - error: "The service is already starting for this endpoint", }; // The service record is per install, not per endpoint; spawning a second // endpoint's service now would overwrite the first one's record and @@ -332,22 +362,23 @@ export class ReMeEnvManager { startedAt: now, ...(current.version ? { version: current.version } : {}), }; - if (current.phase === "conflict") - return { - started: false, - alreadyRunning: false, - command: "", - logPath, - startedAt: now, - error: "Another process is already listening on this endpoint", - }; - // A previous launch that never became healthy may still be alive. Stop it - // before spawning a replacement, so it cannot be orphaned when the state - // record is overwritten. - if (current.pid !== undefined) { + if (current.phase === "conflict") { + // The port is answered by something that is not a healthy service. A + // provable launch of ours (still booting or rejecting our token) is + // stopped and replaced; anything else is left alone. + if (!current.managed) + return { + started: false, + alreadyRunning: false, + command: "", + logPath, + startedAt: now, + error: "Another process is already listening on this endpoint", + }; const previous = await stopService(this.serviceIO, { statePath: this.serviceStatePath, endpoint: endpointValue, + ownership: "owned", }); if (!previous.stopped) return { @@ -361,6 +392,64 @@ export class ReMeEnvManager { }`, }; } + // A record for a *different* endpoint that is still alive would be + // overwritten by the new launch, leaving that endpoint unable to stop it. + // Only a provably owned process blocks the new start; an unprovable record + // is stale and is discarded, just like the same-endpoint case below. + const recorded = readServiceState(this.serviceStatePath); + if ( + recorded && + recorded.pid !== undefined && + serviceEndpointKey(recorded.endpoint) !== + serviceEndpointKey(endpointValue) && + processAlive(this.serviceIO, recorded.pid) + ) { + const other = await serviceOwnership( + this.serviceIO, + recorded, + recorded.endpoint, + ); + if (other === "owned") + return { + started: false, + alreadyRunning: false, + command: "", + logPath, + startedAt: now, + error: `A service for ${recorded.endpoint} is still running; stop it from that endpoint first`, + }; + removeServiceState(this.serviceStatePath); + } + // A previous launch that never became healthy may still be alive. Stop it + // when ownership is proven; when it cannot be proven, drop the stale record + // without killing anything so a recycled pid is never touched. + if (current.pid !== undefined) { + const ownership = await serviceOwnership( + this.serviceIO, + recorded, + endpointValue, + ); + if (ownership === "owned") { + const previous = await stopService(this.serviceIO, { + statePath: this.serviceStatePath, + endpoint: endpointValue, + ownership, + }); + if (!previous.stopped) + return { + started: false, + alreadyRunning: false, + command: "", + logPath, + startedAt: now, + error: `A previous launch could not be stopped: ${ + previous.reason ?? "unknown reason" + }`, + }; + } else { + removeServiceState(this.serviceStatePath); + } + } const path = pythonPath.trim(); if (!path) return { diff --git a/integrations/dsh/src/environment/host/service.ts b/integrations/dsh/src/environment/host/service.ts index f7bd370d..fbb7c91f 100644 --- a/integrations/dsh/src/environment/host/service.ts +++ b/integrations/dsh/src/environment/host/service.ts @@ -3,7 +3,6 @@ import { closeSync, existsSync, mkdirSync, - openSync, readFileSync, renameSync, rmSync, @@ -14,6 +13,7 @@ import { basename, dirname } from "node:path"; import { safeError } from "../../endpoint.js"; import type { ServiceStartResult, ServiceState } from "../types.js"; +import { openAppendLog } from "./log.js"; /** * The local ReMe service is a *detached* process: it outlives DSH, which is the @@ -65,6 +65,8 @@ export interface ServiceIO { probePid(pid: number): void; /** OS image name of a pid, or "" when it cannot be read. */ processImage(pid: number): Promise; + /** Full command line of a pid, or "" when it cannot be read. */ + processCommandLine?(pid: number): Promise; exists(path: string): boolean; platform: string; } @@ -109,9 +111,11 @@ export function parseLocalEndpoint(endpoint: string): { throw new TypeError( "Local service management requires an http:// endpoint", ); - if (!["127.0.0.1", "localhost", "[::1]", "::1"].includes(url.hostname)) + // The service is launched with `service.host=127.0.0.1`; accepting an IPv6 + // loopback would spawn a process that can never answer that endpoint. + if (!["127.0.0.1", "localhost"].includes(url.hostname)) throw new TypeError( - "Local service management requires a loopback endpoint", + "Local service management requires an IPv4 loopback endpoint", ); const port = Number(url.port || "80"); if (!Number.isInteger(port) || port < 1 || port > 65535) @@ -202,7 +206,8 @@ export function writeServiceState(path: string, state: PersistedService): void { } } -function clearServiceState(path: string): void { +/** Drop the service record without touching the recorded process. */ +export function removeServiceState(path: string): void { try { rmSync(path, { force: true }); } catch { @@ -210,6 +215,8 @@ function clearServiceState(path: string): void { } } +const clearServiceState = removeServiceState; + function alive(io: ServiceIO, pid?: number): boolean { if (!isValidPid(pid)) return false; try { @@ -220,6 +227,11 @@ function alive(io: ServiceIO, pid?: number): boolean { } } +/** Whether a pid is alive, rejecting pids this plugin never records. */ +export function processAlive(io: ServiceIO, pid?: number): boolean { + return alive(io, pid); +} + /** * Whether an OS image name is a process that could be running the ReMe * service: an exact Python-family name or `reme`, never a prefix match (which @@ -231,26 +243,89 @@ export function isServiceImage(image: string): 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. + * Ownership of the recorded process, as far as this plugin can prove it. * - * This guards a destructive operation, so it fails *closed*: an unreadable - * image is not proof of ownership. + * - `owned`: endpoint, pid, image and command line all match the launch this + * plugin recorded. + * - `foreign`: at least one piece of evidence contradicts the record. + * - `unknown`: the process exists but its image or command line could not be + * read (localized `tasklist`, `ps` failure, missing permissions). + * + * Stop is destructive, so it only accepts `owned`; starting can discard an + * `unknown` record (without killing it) instead of being blocked forever. + */ +export type ServiceOwnership = "owned" | "foreign" | "unknown"; + +/** Compare paths and ports with separators and case normalized. */ +function normalizeCommandText(value: string): string { + return value.replaceAll("\\\\", "\\").replaceAll("/", "\\").toLowerCase(); +} + +/** + * The recorded launch always passes `workspace_dir=` and + * `service.port=` as argv entries (also inside the Python wrapper's + * `-c` script), so their presence is strong evidence that this pid is the + * process the record describes — not merely another Python that reused the pid. + */ +function commandMatches(command: string, persisted: PersistedService): boolean { + let port: number; + try { + port = parseLocalEndpoint(persisted.endpoint).port; + } catch { + return false; + } + const workspace = persisted.workspaceDir.trim(); + if (!workspace) return false; + const text = normalizeCommandText(command); + const marker = `workspace_dir=${normalizeCommandText(workspace)}`; + const index = text.indexOf(marker); + if (index < 0) return false; + // Require a boundary so `C:\a` does not match another service using `C:\ab`. + const next = text[index + marker.length] ?? ""; + if (next && !/[\s"',\]]/.test(next)) return false; + return new RegExp(`service\\.port=${port}(?!\\d)`).test(text); +} + +export async function serviceOwnership( + io: ServiceIO, + persisted: PersistedService | undefined, + endpoint: string, +): Promise { + if (!persisted || !isValidPid(persisted.pid)) return "foreign"; + if (serviceEndpointKey(persisted.endpoint) !== serviceEndpointKey(endpoint)) + return "foreign"; + if (!alive(io, persisted.pid)) return "foreign"; + let image = ""; + try { + image = await io.processImage(persisted.pid); + } catch { + image = ""; + } + if (!image.trim()) return "unknown"; + if (!isServiceImage(image)) return "foreign"; + let command = ""; + try { + command = (await io.processCommandLine?.(persisted.pid)) ?? ""; + } catch { + command = ""; + } + if (!command.trim()) return "unknown"; + return commandMatches(command, persisted) ? "owned" : "foreign"; +} + +/** + * Whether the recorded process is provably the service this plugin started + * for the endpoint on screen. + * + * This guards a destructive operation, so it fails *closed*: unreadable + * evidence is not proof of ownership. */ export async function ownsService( io: ServiceIO, persisted: PersistedService | undefined, endpoint: string, ): Promise { - if (!persisted || !isValidPid(persisted.pid)) return false; - if (serviceEndpointKey(persisted.endpoint) !== serviceEndpointKey(endpoint)) - return false; - if (!alive(io, persisted.pid)) return false; - const image = await io.processImage(persisted.pid); - if (!image.trim()) return false; - return isServiceImage(image); + return (await serviceOwnership(io, persisted, endpoint)) === "owned"; } /** Running / conflict / stopped, with ownership of the running process. */ @@ -276,11 +351,15 @@ export async function serviceStatus( // A record whose process is gone must not be reported, or a later start // would keep seeing a stale pid. Remove it while nobody can be relying on it. if (persisted && !recordAlive) clearServiceState(options.statePath); + // Ownership is expensive (image plus command line). During boot the service + // is polled every second while `stopped`, where it is not needed, so it is + // computed lazily only for the phases that report it. + const ownership = async (): Promise => + recordForEndpoint && recordAlive + ? serviceOwnership(io, persisted, options.endpoint) + : "foreign"; if (health.ok) { - const managed = - recordForEndpoint && - recordAlive && - (await ownsService(io, persisted, options.endpoint)); + const managed = (await ownership()) === "owned"; return { phase: "running", managed, @@ -300,19 +379,26 @@ export async function serviceStatus( } catch { // A remote or https endpoint is health-checked only; it has no local port. } - if (portOpen) + if (portOpen) { + // Our own service can occupy the port without being healthy (still + // booting, or a rejected token); identify it so the tab can stop it. + const managed = (await ownership()) === "owned"; return { phase: "conflict", - managed: false, + managed, endpoint: options.endpoint, + ...(managed && persisted?.pid !== undefined + ? { pid: persisted.pid } + : {}), logPath: options.logPath, }; + } return { phase: "stopped", managed: false, endpoint: options.endpoint, // A live record for this endpoint means a launch that never became - // healthy; the manager stops it before spawning a replacement. + // healthy; the manager stops or discards it before spawning a replacement. ...(recordForEndpoint && recordAlive && persisted?.pid !== undefined ? { pid: persisted.pid } : {}), @@ -463,6 +549,8 @@ export async function stopService( timeoutMs?: number; forceTimeoutMs?: number; pollMs?: number; + /** Precomputed ownership from the caller; avoids a second process probe. */ + ownership?: ServiceOwnership; }, ): Promise { const persisted = readServiceState(options.statePath); @@ -470,7 +558,16 @@ export async function stopService( if (persisted) clearServiceState(options.statePath); return { stopped: false, reason: "No service started by this plugin" }; } - if (!(await ownsService(io, persisted, options.endpoint))) + const ownership = + options.ownership ?? + (await serviceOwnership(io, persisted, options.endpoint)); + if (ownership === "unknown") + return { + stopped: false, + reason: + "The recorded process identity could not be verified; refusing to stop it", + }; + if (ownership !== "owned") return { stopped: false, reason: "The recorded service does not belong to this endpoint", @@ -509,7 +606,7 @@ export async function stopService( return { stopped: false, pid, - reason: "The service did not exit after SIGKILL", + reason: "The service did not exit after the force-kill", }; } @@ -572,7 +669,7 @@ export function createServiceIO(): ServiceIO { }); }, async spawnService({ exe, args, cwd, logPath, env }) { - const fd = openSync(logPath, "a"); + const fd = openAppendLog(logPath); let child: ReturnType; try { child = spawn(exe, args, { @@ -657,6 +754,37 @@ export function createServiceIO(): ServiceIO { ); }); }, + processCommandLine(pid) { + return new Promise((resolve) => { + if (!isValidPid(pid)) { + resolve(""); + return; + } + if (process.platform === "win32") { + // wmic is absent from recent Windows builds, so ask CIM. The encoding + // prefix keeps non-ASCII workspace paths intact through the pipe. + execFile( + "powershell.exe", + [ + "-NoProfile", + "-NonInteractive", + "-Command", + `[Console]::OutputEncoding=[Text.Encoding]::UTF8; (Get-CimInstance Win32_Process -Filter "ProcessId=${pid}").CommandLine`, + ], + { windowsHide: true, timeout: 5000 }, + (error, stdout) => + resolve(error ? "" : String(stdout ?? "").trim()), + ); + return; + } + execFile( + "ps", + ["-p", String(pid), "-o", "args="], + { timeout: 3000 }, + (error, stdout) => resolve(error ? "" : String(stdout ?? "").trim()), + ); + }); + }, exists: existsSync, platform: process.platform, }; diff --git a/integrations/dsh/tests/env-manager.test.mjs b/integrations/dsh/tests/env-manager.test.mjs index 85603a99..6f31d04c 100644 --- a/integrations/dsh/tests/env-manager.test.mjs +++ b/integrations/dsh/tests/env-manager.test.mjs @@ -1,11 +1,20 @@ import assert from "node:assert/strict"; import test from "node:test"; -import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { + mkdtempSync, + readFileSync, + rmSync, + statSync, + writeFileSync, +} from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { ReMeEnvManager } from "../dist/environment/host/manager.js"; -import { writeServiceState } from "../dist/environment/host/service.js"; +import { + readServiceState, + writeServiceState, +} from "../dist/environment/host/service.js"; /** A service adapter that records launches instead of starting processes. */ function fakeServiceIO() { @@ -24,6 +33,7 @@ function fakeServiceIO() { killTree() {}, probePid() {}, processImage: async () => "python.exe", + processCommandLine: async () => "", exists: () => true, platform: "win32", }, @@ -137,6 +147,8 @@ test("stops a previous launch that never became healthy before replacing it", as calls.killTree.push(pid); running = false; }; + io.processCommandLine = async () => + `reme start workspace_dir=${directory} service.port=2334`; const result = await manager.serviceStart( "http://127.0.0.1:2334", @@ -148,6 +160,96 @@ test("stops a previous launch that never became healthy before replacing it", as assert.equal(calls.spawn.length, 1); }); +test("an unprovable previous record is discarded without killing it", async (t) => { + const { manager, calls, directory, io } = managerIn(t); + manager.serviceStatus = async (endpoint) => ({ + phase: "stopped", + managed: false, + endpoint, + pid: 6000, + logPath: join(directory, "service.log"), + }); + const statePath = join(directory, "reme-memory", "service.json"); + writeServiceState(statePath, { + pid: 6000, + endpoint: "http://127.0.0.1:2334", + workspaceDir: directory, + pythonPath: "D:\\env\\python.exe", + command: "reme start", + startedAt: "2026-10-27T00:00:00.000Z", + }); + // The process exists, but its command line cannot be read, so ownership is + // unproven: never kill it, just drop the stale record and start. + io.processCommandLine = async () => ""; + const result = await manager.serviceStart( + "http://127.0.0.1:2334", + "D:\\env\\python.exe", + join(directory, "workspace"), + ); + assert.equal(result.started, true); + assert.deepEqual(calls.killTree, []); + assert.equal(calls.spawn.length, 1); + assert.equal(readServiceState(statePath)?.pid, 5001); +}); + +test("refuses to overwrite a live record owned by another endpoint", async (t) => { + const { manager, calls, directory, io } = managerIn(t); + manager.serviceStatus = async (endpoint) => ({ + phase: "stopped", + managed: false, + endpoint, + logPath: join(directory, "service.log"), + }); + writeServiceState(join(directory, "reme-memory", "service.json"), { + pid: 6000, + endpoint: "http://127.0.0.1:2333", + workspaceDir: directory, + pythonPath: "D:\\env\\python.exe", + command: "reme start", + startedAt: "2026-10-27T00:00:00.000Z", + }); + io.processCommandLine = async () => + `reme start workspace_dir=${directory} service.port=2333`; + const result = await manager.serviceStart( + "http://127.0.0.1:2334", + "D:\\env\\python.exe", + join(directory, "workspace"), + ); + assert.equal(result.started, false); + assert.match(result.error, /still running/); + assert.equal(calls.spawn.length, 0); +}); + +test("stop waits for an in-flight start and then stops that launch", async (t) => { + const { manager, calls, directory, io } = managerIn(t); + let running = true; + io.killTree = (pid) => { + calls.killTree.push(pid); + running = false; + }; + io.probePid = () => { + if (!running) throw new Error("ESRCH"); + }; + io.processCommandLine = async () => + `reme start workspace_dir=${join( + directory, + "workspace", + )} service.port=2334`; + + const start = manager.serviceStart( + "http://127.0.0.1:2334", + "D:\\env\\python.exe", + join(directory, "workspace"), + ); + // Arrives while the start is still being recorded, so it must join the queue + // rather than answering "No service started by this plugin". + const stop = manager.serviceStop("http://127.0.0.1:2334"); + const [started, stopped] = await Promise.all([start, stop]); + assert.equal(started.started, true); + assert.equal(stopped.stopped, true); + assert.deepEqual(calls.killTree, [5001]); +}); + 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"); @@ -182,3 +284,20 @@ test("validates the merged .env, not just the submitted values", async (t) => { ); assert.match(readFileSync(envPath, "utf8"), /^LLM_API_KEY=sk-keep$/m); }); + +test("writes .env atomically with a backup and owner-only mode", (t) => { + const { manager, directory } = managerIn(t); + const envPath = join(directory, ".env"); + writeFileSync( + envPath, + "LLM_API_KEY=sk-old\nEMBEDDING_API_KEY=EMPTY\nLLM_MODEL_NAME=old\n", + ); + + manager.writeEnv(directory, { LLM_MODEL_NAME: "new" }); + assert.match(readFileSync(envPath, "utf8"), /^LLM_MODEL_NAME=new$/m); + const backup = readFileSync(`${envPath}.bak`, "utf8"); + assert.match(backup, /^LLM_MODEL_NAME=old$/m); + assert.match(backup, /^LLM_API_KEY=sk-old$/m); + if (process.platform !== "win32") + assert.equal(statSync(envPath).mode & 0o777, 0o600); +}); diff --git a/integrations/dsh/tests/env-reme.test.mjs b/integrations/dsh/tests/env-reme.test.mjs index ff8b1d0b..31174e4f 100644 --- a/integrations/dsh/tests/env-reme.test.mjs +++ b/integrations/dsh/tests/env-reme.test.mjs @@ -181,8 +181,9 @@ function fakeSpawner() { onExit(callback) { child.exit = callback; }, - kill() { + kill(force = false) { child.killed = true; + child.force = force; }, }; children.push(child); diff --git a/integrations/dsh/tests/env-runner.test.mjs b/integrations/dsh/tests/env-runner.test.mjs index 4e359b7b..51140692 100644 --- a/integrations/dsh/tests/env-runner.test.mjs +++ b/integrations/dsh/tests/env-runner.test.mjs @@ -1,6 +1,18 @@ import assert from "node:assert/strict"; import test from "node:test"; +import { + closeSync, + mkdtempSync, + readFileSync, + rmSync, + statSync, + writeFileSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { createInstallSpawner } from "../dist/environment/host/install.js"; +import { openAppendLog } from "../dist/environment/host/log.js"; import { createNodeRunner } from "../dist/environment/host/runner.js"; test("a timeout settles with an explicit error instead of hanging", async () => { @@ -28,3 +40,37 @@ test("a child that exits on SIGTERM is still reported as a timeout", async () => ); assert.match(result.error ?? "", /timed out after 200 ms/); }); + +test("the real install spawner starts and stops a process", async (t) => { + const directory = mkdtempSync(join(tmpdir(), "reme-spawner-")); + t.after(() => rmSync(directory, { recursive: true, force: true })); + const spawner = createInstallSpawner({ PATH: process.env.PATH }); + const logPath = join(directory, "install.log"); + const child = spawner.start({ + exe: process.execPath, + args: ["-e", "setInterval(() => {}, 1000)"], + logPath, + }); + assert.ok(child.pid !== undefined && child.pid > 1); + + const exited = new Promise((resolve) => child.onExit(resolve)); + child.kill(); + const outcome = await Promise.race([ + exited, + new Promise((resolve) => setTimeout(() => resolve("timeout"), 5_000)), + ]); + assert.notEqual(outcome, "timeout", "the install process must actually exit"); + assert.match(readFileSync(logPath, "utf8"), /^$/); +}); + +test("an oversized log is rotated before the next append", (t) => { + const directory = mkdtempSync(join(tmpdir(), "reme-log-")); + t.after(() => rmSync(directory, { recursive: true, force: true })); + const path = join(directory, "install.log"); + writeFileSync(path, "x".repeat(200)); + + const fd = openAppendLog(path, 100); + closeSync(fd); + assert.equal(readFileSync(`${path}.1`, "utf8").length, 200); + assert.equal(statSync(path).size, 0); +}); diff --git a/integrations/dsh/tests/env-service.test.mjs b/integrations/dsh/tests/env-service.test.mjs index c7c0504b..10f8ddce 100644 --- a/integrations/dsh/tests/env-service.test.mjs +++ b/integrations/dsh/tests/env-service.test.mjs @@ -11,6 +11,7 @@ import { ownsService, parseLocalEndpoint, readServiceState, + serviceOwnership, serviceStatus, startService, stopService, @@ -34,6 +35,7 @@ function fakeIO(overrides = {}) { }, probePid() {}, processImage: async () => "python.exe", + processCommandLine: async () => "", exists: () => true, platform: "win32", ...overrides, @@ -41,6 +43,12 @@ function fakeIO(overrides = {}) { return { io, calls }; } +/** The command line a launch for `statePath` would actually have. */ +function commandFor(statePath, port = 2334) { + const state = readServiceState(statePath); + return `reme start workspace_dir=${state.workspaceDir} service.port=${port}`; +} + function tempState(t, prefix) { const directory = mkdtempSync(join(tmpdir(), prefix)); t.after(() => rmSync(directory, { recursive: true, force: true })); @@ -77,6 +85,9 @@ test("accepts only loopback HTTP endpoints and validates the port", () => { () => parseLocalEndpoint("https://127.0.0.1:2333"), /http:\/\//, ); + // The launch binds service.host=127.0.0.1, so an IPv6 loopback endpoint + // would start a service that can never answer it. + assert.throws(() => parseLocalEndpoint("http://[::1]:2333"), /IPv4/); }); test("builds reme start with the workspace cwd contract", () => { @@ -140,6 +151,7 @@ test("derives running, conflict and stopped phases", async (t) => { assert.equal(status.version, "0.4.0.9"); record(statePath); + runningIo.io.processCommandLine = async () => commandFor(statePath); status = await serviceStatus(runningIo.io, base); assert.equal(status.phase, "running"); assert.equal(status.managed, true); @@ -176,6 +188,7 @@ test("a loopback alias still identifies the service this plugin started", async record(statePath, { endpoint: "http://localhost:2334/" }); const { io } = fakeIO({ fetchHealth: async () => ({ ok: true, version: "0.4.0.9", error: "" }), + processCommandLine: async () => commandFor(statePath), }); const status = await serviceStatus(io, { endpoint: ENDPOINT, @@ -186,6 +199,23 @@ test("a loopback alias still identifies the service this plugin started", async assert.equal(status.pid, 777); }); +test("a conflict still identifies a provably owned service", async (t) => { + const { directory, statePath } = tempState(t, "reme-conflict-"); + record(statePath); + const { io } = fakeIO({ + portOpen: async () => true, + processCommandLine: async () => commandFor(statePath), + }); + const status = await serviceStatus(io, { + endpoint: ENDPOINT, + statePath, + logPath: join(directory, "service.log"), + }); + assert.equal(status.phase, "conflict"); + assert.equal(status.managed, true); + assert.equal(status.pid, 777); +}); + test("a recycled pid that is no longer the service is not managed", async (t) => { const { statePath } = tempState(t, "reme-pid-"); record(statePath); @@ -217,6 +247,40 @@ test("a recycled pid that is no longer the service is not managed", async (t) => `image ${JSON.stringify(image)} must not be owned`, ); } + // A Python image whose command line describes another launch is foreign, + // not ours; an unreadable command line is unproven, not ownership. + const { io: otherLaunch } = fakeIO({ + processCommandLine: async () => "python -m http.server 2334", + }); + assert.equal( + await serviceOwnership(otherLaunch, readServiceState(statePath), ENDPOINT), + "foreign", + ); + const { io: unproven } = fakeIO(); + assert.equal( + await serviceOwnership(unproven, readServiceState(statePath), ENDPOINT), + "unknown", + ); + assert.equal( + await ownsService(unproven, readServiceState(statePath), ENDPOINT), + false, + ); + // Matching is boundary-aware: a longer workspace path or port is another + // launch, not this one. + const state = readServiceState(statePath); + for (const command of [ + `reme start workspace_dir=${state.workspaceDir}2 service.port=2334`, + `reme start workspace_dir=${state.workspaceDir} service.port=23340`, + ]) { + const { io: longer } = fakeIO({ + processCommandLine: async () => command, + }); + assert.equal( + await serviceOwnership(longer, state, ENDPOINT), + "foreign", + command, + ); + } }); test("a dead record is cleared instead of reporting a stale pid", async (t) => { @@ -286,7 +350,9 @@ test("records nothing when the spawn fails", async (t) => { 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 }); - const { io, calls } = fakeIO(); + const { io, calls } = fakeIO({ + processCommandLine: async () => commandFor(statePath), + }); let running = true; io.probePid = () => { if (!running) throw new Error("ESRCH"); @@ -313,7 +379,9 @@ test("stops the managed pid and clears the state file after it exits", async (t) test("escalates to SIGKILL and keeps the record when the service never exits", async (t) => { const { directory, statePath } = tempState(t, "reme-stubborn-"); record(statePath, { pid: 999, workspaceDir: directory }); - const { io, calls } = fakeIO(); + const { io, calls } = fakeIO({ + processCommandLine: async () => commandFor(statePath), + }); const stopped = await stopService(io, { statePath, endpoint: ENDPOINT, @@ -340,6 +408,17 @@ test("refuses to stop a service recorded for another endpoint", async (t) => { assert.ok(readServiceState(statePath), "the foreign record is left in place"); }); +test("refuses to stop a process whose identity cannot be verified", async (t) => { + const { statePath } = tempState(t, "reme-unknown-"); + record(statePath); + const { io, calls } = fakeIO(); // image readable, command line not + const stopped = await stopService(io, { statePath, endpoint: ENDPOINT }); + assert.equal(stopped.stopped, false); + assert.match(stopped.reason, /could not be verified/); + assert.deepEqual(calls.killTree, []); + assert.equal(readServiceState(statePath)?.pid, 777); +}); + /* ------------------------------------------------------------------ */ /* Real IO: the pieces the fake adapter above cannot vouch for. */ /* ------------------------------------------------------------------ */ @@ -402,6 +481,7 @@ test("real IO spawns, identifies and stops a detached process", async (t) => { assert.ok(pid !== undefined && pid > 1); assert.doesNotThrow(() => io.probePid(pid)); assert.doesNotMatch(await io.processImage(pid), /^$/); + assert.match(await io.processCommandLine(pid), /node/i); io.killTree(pid); const deadline = Date.now() + 5_000; diff --git a/integrations/dsh/tests/env-tab.test.mjs b/integrations/dsh/tests/env-tab.test.mjs index 43e0eeff..1336b452 100644 --- a/integrations/dsh/tests/env-tab.test.mjs +++ b/integrations/dsh/tests/env-tab.test.mjs @@ -116,8 +116,9 @@ function texts(node, found = []) { return found; } -async function mount(overrides = {}) { +async function mount(overrides = {}, options = {}) { const calls = { mutate: [], rpc: [] }; + const responses = { ...RESPONSES, ...(options.responses ?? {}) }; const listeners = new Set(); // A new object per change, the way the Host-backed form publishes a snapshot. let snapshot = { @@ -142,13 +143,13 @@ async function mount(overrides = {}) { getSnapshot: () => snapshot, mutate: async (ops, revision) => { calls.mutate.push({ ops, revision }); - return true; + return options.mutate ? options.mutate() : true; }, }; const rpc = { call: async (_path, method, payload) => { calls.rpc.push({ method, input: payload?.args?.input }); - return { ok: true, value: RESPONSES[method] ?? null }; + return { ok: true, value: responses[method] ?? null }; }, }; @@ -171,7 +172,13 @@ async function mount(overrides = {}) { }; for (const listener of listeners) listener(); }, - restore: () => listeners.clear(), + restore: () => { + // Unmount so the effect cleanup paths (aborts, timers) actually run. + act(() => { + renderer.unmount(); + }); + listeners.clear(); + }, }; } @@ -352,3 +359,82 @@ test("probes the service from the Host, never from the browser", async (t) => { "a local endpoint is covered by the service probe already", ); }); + +test("a remote endpoint never runs the local setup steps", async (t) => { + const { calls, restore } = await mount( + { endpoint: "http://10.0.0.5:2333" }, + { + responses: { + "remeEnv/health": { + endpoint: "http://10.0.0.5:2333", + ok: false, + version: "", + error: "unreachable", + checkedAt: new Date().toISOString(), + }, + }, + }, + ); + t.after(restore); + const methods = calls.rpc.map((call) => call.method); + assert.ok( + methods.includes("remeEnv/health"), + "a remote endpoint is only health-checked", + ); + for (const local of [ + "remeEnv/python", + "remeEnv/checkReme", + "remeEnv/readEnv", + "remeEnv/serviceStatus", + ]) + assert.equal(methods.includes(local), false, `${local} must not run`); +}); + +test("shows why Stop left the service alone", async (t) => { + const { renderer, restore } = await mount( + {}, + { + responses: { + "remeEnv/serviceStatus": { + phase: "running", + managed: true, + pid: 98820, + version: "0.4.0.9", + endpoint: "http://127.0.0.1:2333", + logPath: "C:\\Users\\test\\.reme\\service.log", + }, + "remeEnv/serviceStop": { + stopped: false, + reason: "The recorded process identity could not be verified", + }, + }, + }, + ); + t.after(restore); + await act(async () => { + buttonByText(renderer, envZh.stopService).props.onClick(); + }); + await act(async () => { + buttonByText(renderer, envZh.stopService).props.onClick(); + }); + await act(async () => { + await new Promise((resolve) => setTimeout(resolve, 10)); + }); + assert.match(texts(renderer.toJSON()).join(" "), /could not be verified/); +}); + +test("surfaces a rejected settings write", async (t) => { + const { renderer, restore } = await mount({}, { mutate: async () => false }); + t.after(restore); + const checkbox = renderer.root.findAll( + (node) => node.type === "input" && node.props.type === "checkbox", + )[0]; + assert.ok(checkbox); + await act(async () => { + checkbox.props.onChange({ target: { checked: true } }); + }); + await act(async () => { + await new Promise((resolve) => setTimeout(resolve, 10)); + }); + assert.match(texts(renderer.toJSON()).join(" "), /设置未保存/); +});