=> {
+ if (!snapshot.writable) return false;
+ try {
+ const applied = await scope.mutate(
+ [{ op: "set", path: [field], value: next }],
+ snapshot.revision,
+ );
+ if (applied === false) {
+ setSettingsError(t("settingsWriteFailed"));
+ return false;
+ }
+ setSettingsError("");
+ return true;
+ } catch (error) {
+ setSettingsError(safeError(error));
+ return false;
+ }
},
- [scope, snapshot.revision, snapshot.writable],
+ [scope, snapshot.revision, snapshot.writable, t],
);
const python = usePython(rpc, {
pythonPath: value?.pythonPath ?? "",
writable: snapshot.writable,
+ enabled: local,
onSelect: (exe) => writeSetting("pythonPath", exe),
});
const dependency = useDependency(rpc, {
pythonPath: python.selectedPath,
+ enabled: local,
onInstalled: () => {
python.refresh();
},
@@ -134,6 +174,7 @@ export function ReMeEnvironmentTab({
const workspace = useWorkspace(rpc, {
workspaceDir,
writable: snapshot.writable,
+ enabled: local,
onWorkspaceDir: (dir) => writeSetting("remeWorkspaceDir", dir),
});
const service = useService(rpc, {
@@ -154,18 +195,30 @@ export function ReMeEnvironmentTab({
tick: heroTick,
authToken,
});
+ // The Host names the endpoint its answer belongs to; comparing that with the
+ // settings endpoint is what makes a stale answer fail the `current` test.
+ const serviceEndpoint = service.state.data?.endpoint ?? endpoint;
const connection: ConnectionState = local
? service.state.phase === "running"
- ? { phase: "ok", endpoint, version: service.state.data?.version ?? "" }
+ ? {
+ phase: "ok",
+ endpoint: serviceEndpoint,
+ version: service.state.data?.version ?? "",
+ }
: service.state.phase === "checking" || service.state.phase === "starting"
- ? { phase: "loading", endpoint }
- : { phase: "fail", endpoint, error: service.state.error }
+ ? { phase: "loading", endpoint: serviceEndpoint }
+ : {
+ phase: "fail",
+ endpoint: serviceEndpoint,
+ error: service.state.error,
+ }
: remote;
if (snapshot.status === "unavailable") return null;
const current =
- connection.phase !== "loading" && connection.endpoint === endpoint;
+ connection.phase !== "loading" &&
+ endpointKey(connection.endpoint) === endpointKey(endpoint);
const ready =
current &&
connection.phase === "ok" &&
@@ -241,6 +294,11 @@ export function ReMeEnvironmentTab({
{connection.error}
) : null}
+ {settingsError ? (
+
+ {settingsError}
+
+ ) : null}
{!local ? (
{t("remoteNotice")}
diff --git a/integrations/dsh/src/environment/client/workspace.tsx b/integrations/dsh/src/environment/client/workspace.tsx
index 9e423ee1..751d76d5 100644
--- a/integrations/dsh/src/environment/client/workspace.tsx
+++ b/integrations/dsh/src/environment/client/workspace.tsx
@@ -144,10 +144,12 @@ export function useWorkspace(
options: {
workspaceDir: string;
writable: boolean;
+ /** False for a remote endpoint, whose local steps are hidden. */
+ enabled?: boolean;
onWorkspaceDir(dir: string): Promise;
},
): WorkspaceState {
- const { workspaceDir, writable, onWorkspaceDir } = options;
+ const { workspaceDir, writable, enabled = true, onWorkspaceDir } = options;
const [env, setEnv] = useState({ phase: "loading" });
const [draft, setDraft] = useState({ dir: "", values: {} });
const [saving, setSaving] = useState(false);
@@ -158,12 +160,18 @@ export function useWorkspace(
// Read by a save that resolves after a workspace switch, so the answer for
// the previous workspace is dropped instead of shown against the new one.
const latestDir = useRef(workspaceDir);
+ const llmController = useRef(undefined);
useEffect(() => {
latestDir.current = workspaceDir;
+ // A probe for the previous workspace must not land on the new one.
+ llmController.current?.abort();
}, [workspaceDir]);
+ useEffect(() => () => llmController.current?.abort(), []);
+
useEffect(() => {
+ if (!enabled) return;
const controller = new AbortController();
void (async () => {
try {
@@ -173,6 +181,7 @@ export function useWorkspace(
{ workspaceDir },
controller.signal,
);
+ if (controller.signal.aborted) return;
setEnv({
phase: "ready",
data: parseEnvSummary(value),
@@ -188,7 +197,7 @@ export function useWorkspace(
}
})();
return () => controller.abort();
- }, [rpc, workspaceDir, tick]);
+ }, [rpc, workspaceDir, tick, enabled]);
// A workspace switch drops drafts typed for the previous workspace, so a key
// typed for one workspace can never be written into another. The read is
@@ -262,13 +271,24 @@ export function useWorkspace(
const testLlm = useCallback(
async (pythonPath: string) => {
if (!pythonPath || llmTest.phase === "running") return;
+ const dir = workspaceDir;
+ llmController.current?.abort();
+ const controller = new AbortController();
+ llmController.current = controller;
setLlmTest({ phase: "running" });
try {
const data = parseLlmResult(
- await envCall(rpc, "testLlm", { workspaceDir, pythonPath }),
+ await envCall(
+ rpc,
+ "testLlm",
+ { workspaceDir: dir, pythonPath },
+ controller.signal,
+ ),
);
+ if (controller.signal.aborted || latestDir.current !== dir) return;
setLlmTest({ phase: data.ok ? "ok" : "fail", result: data });
} catch (error) {
+ if (controller.signal.aborted || latestDir.current !== dir) return;
setLlmTest({
phase: "fail",
result: {
@@ -287,7 +307,10 @@ export function useWorkspace(
if (!writable) return;
const next = workspaceInput ?? workspaceDir;
setWorkspaceInput(undefined);
- await onWorkspaceDir(next.trim()).catch(() => undefined);
+ const saved = await onWorkspaceDir(next.trim()).catch(() => false);
+ // A rejected save is reported by the settings writer; do not reload the
+ // old directory as if the new one had been accepted.
+ if (saved === false) return;
setTick((current) => current + 1);
}, [onWorkspaceDir, workspaceDir, workspaceInput, writable]);
diff --git a/integrations/dsh/src/environment/host/detect.ts b/integrations/dsh/src/environment/host/detect.ts
index dadd66bc..a2a7ed3a 100644
--- a/integrations/dsh/src/environment/host/detect.ts
+++ b/integrations/dsh/src/environment/host/detect.ts
@@ -225,6 +225,24 @@ export function makeCondaRunner(
};
}
+/**
+ * Unix interpreters that a `which -a python3` can miss when the shell has a
+ * trimmed PATH: the system install, Homebrew on both Apple Silicon and Intel,
+ * and pyenv/local shims. Windows has `where` and the registry for this.
+ */
+function commonPythonPaths(io: DetectIO, windows: boolean): string[] {
+ if (windows) return [];
+ const path = pathOf(io.homedir);
+ return [
+ "/usr/bin/python3",
+ "/usr/bin/python",
+ "/usr/local/bin/python3",
+ "/opt/homebrew/bin/python3",
+ path.join(io.homedir, ".pyenv", "shims", "python3"),
+ path.join(io.homedir, ".local", "bin", "python3"),
+ ];
+}
+
function commonCondaPaths(io: DetectIO, windows: boolean): string[] {
const programData = io.env["ProgramData"] ?? "";
const path = windows ? win32 : pathOf(io.homedir);
@@ -340,7 +358,11 @@ function classifyOrigin(
};
if ([...candidate.sources].some((source) => source.startsWith("registry")))
return { kind: "python.org" };
- if (candidate.sources.has("PATH") || candidate.sources.has("which"))
+ if (
+ candidate.sources.has("PATH") ||
+ candidate.sources.has("which") ||
+ candidate.sources.has("well-known")
+ )
return { kind: "system" };
return { kind: "unknown" };
}
@@ -472,6 +494,10 @@ export async function detectPythonEnvs(io: DetectIO): Promise {
if (isWindowsAppsAlias(exe)) continue;
addCandidate(candidates, exe, windows ? "PATH" : "which", windows);
}
+ // Well-known Unix install locations, independent of the Host PATH.
+ for (const exe of commonPythonPaths(io, windows))
+ if (io.exists(exe))
+ addCandidate(candidates, exe, "well-known", windows);
})(),
// conda: PATH first, then well-known install roots.
(async () => {
diff --git a/integrations/dsh/src/environment/host/install.ts b/integrations/dsh/src/environment/host/install.ts
index f274c90a..662b0508 100644
--- a/integrations/dsh/src/environment/host/install.ts
+++ b/integrations/dsh/src/environment/host/install.ts
@@ -1,14 +1,15 @@
import { spawn } from "node:child_process";
-import { closeSync, mkdirSync, openSync, readFileSync } from "node:fs";
+import { closeSync, fstatSync, mkdirSync, openSync, readSync } from "node:fs";
import { join } from "node:path";
import type { InstallPhase, InstallRequest, InstallStatus } from "../types.js";
-import { pathOf } from "./paths.js";
+import { isAbsolutePath } from "./paths.js";
export interface InstallChild {
pid: number | undefined;
onExit(callback: (code: number | null) => void): void;
- kill(): void;
+ /** `force` escalates to SIGKILL on POSIX; Windows always force-kills. */
+ kill(force?: boolean): void;
}
export interface InstallSpawner {
@@ -33,18 +34,32 @@ export function pipArgs(request: InstallRequest): string[] {
if (indexUrl) {
if (!/^https?:\/\/\S+$/i.test(indexUrl))
throw new TypeError("Package index must be an http(s) URL");
+ // Credentials embedded in the URL would be echoed in the job command, the
+ // install log and the browser; pip's own config is the safe place for them.
+ const parsed = new URL(indexUrl);
+ if (parsed.username || parsed.password)
+ throw new TypeError(
+ "Package index must not embed credentials; configure them in pip.conf",
+ );
args.push("--index-url", indexUrl);
}
args.push("reme-ai[core]");
return args;
}
-/** Last `maxLines` non-empty lines of a log file, normalized to "\n". */
+/**
+ * Last `maxLines` non-empty lines of a log file, normalized to "\n". Only the
+ * tail is read: this runs on every 2 s poll and the log can grow for minutes.
+ */
export function tailLog(path: string, maxLines = 30, maxBytes = 8192): string {
+ let fd: number | undefined;
try {
- const buffer = readFileSync(path);
+ fd = openSync(path, "r");
+ const size = fstatSync(fd).size;
+ const length = Math.min(size, maxBytes);
+ const buffer = Buffer.alloc(length);
+ readSync(fd, buffer, 0, length, Math.max(0, size - length));
const text = buffer
- .subarray(Math.max(0, buffer.length - maxBytes))
.toString("utf8")
.replaceAll("\r\n", "\n")
.replaceAll("\r", "\n");
@@ -55,6 +70,14 @@ export function tailLog(path: string, maxLines = 30, maxBytes = 8192): string {
.join("\n");
} catch {
return "";
+ } finally {
+ if (fd !== undefined) {
+ try {
+ closeSync(fd);
+ } catch {
+ // Already closed.
+ }
+ }
}
}
@@ -70,6 +93,9 @@ export function createInstallSpawner(
child = spawn(exe, args, {
shell: false,
windowsHide: true,
+ // POSIX: lead a process group so a cancel can stop pip's build
+ // backends too, not only pip itself.
+ detached: process.platform !== "win32",
env: {
...base,
PYTHONIOENCODING: "utf-8",
@@ -94,17 +120,30 @@ export function createInstallSpawner(
child.on("close", finish);
child.on("error", () => finish(null));
},
- kill() {
+ kill(force = false) {
if (child.pid !== undefined && process.platform === "win32") {
// Tree-kill: pip's build backends are child processes.
- spawn("taskkill", ["/PID", String(child.pid), "/T", "/F"], {
- windowsHide: true,
- stdio: "ignore",
- });
+ const killer = spawn(
+ "taskkill",
+ ["/PID", String(child.pid), "/T", "/F"],
+ { windowsHide: true, stdio: "ignore" },
+ );
+ // Without this listener an async spawn failure would exit the Host.
+ killer.on("error", () => undefined);
+ killer.unref();
return;
}
+ const signal = force ? "SIGKILL" : "SIGTERM";
+ // The install process leads its own group on POSIX, so pip's build
+ // backends receive the signal as well.
try {
- child.kill("SIGTERM");
+ if (child.pid !== undefined) process.kill(-child.pid, signal);
+ return;
+ } catch {
+ // Not a group leader (e.g. spawn failed); signal the process only.
+ }
+ try {
+ child.kill(signal);
} catch {
// Already gone.
}
@@ -148,7 +187,7 @@ export class InstallJobs {
const pythonPath = (request.pythonPath ?? "").trim();
// The interpreter may be written for another platform than the host, so its
// own flavor decides whether the path is absolute.
- if (!pythonPath || !pathOf(pythonPath).isAbsolute(pythonPath))
+ if (!pythonPath || !isAbsolutePath(pythonPath))
throw new TypeError(
"Install requires an absolute Python interpreter path",
);
@@ -199,8 +238,13 @@ export class InstallJobs {
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);
+ // A process that ignores the graceful kill must not keep the interpreter
+ // locked, so escalate before the last-resort fallback settles the job.
+ job.cancelTimer = setTimeout(() => {
+ job.child.kill(true);
+ job.cancelTimer = setTimeout(() => this.finish(job, null), 15_000);
+ job.cancelTimer.unref();
+ }, 5_000);
job.cancelTimer.unref();
return true;
}
diff --git a/integrations/dsh/src/environment/host/manager.ts b/integrations/dsh/src/environment/host/manager.ts
index 3cc00823..b7b71400 100644
--- a/integrations/dsh/src/environment/host/manager.ts
+++ b/integrations/dsh/src/environment/host/manager.ts
@@ -2,9 +2,10 @@ import { existsSync, readFileSync } from "node:fs";
import { homedir } from "node:os";
import { join } from "node:path";
+import { safeError } from "../../endpoint.js";
import { detectPythonEnvs } from "./detect.js";
import { createInstallSpawner, InstallJobs, tailLog } from "./install.js";
-import { checkReme, remeVersionAtLeast } from "./reme.js";
+import { checkReme, MINIMUM_REME, remeVersionAtLeast } from "./reme.js";
import { createNodeRunner } from "./runner.js";
import {
createServiceIO,
@@ -189,8 +190,16 @@ export class ReMeEnvManager {
"service.log",
),
});
- if (status.phase === "running" && this.starting?.endpoint === endpointValue)
- this.starting = undefined;
+ 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.
+ if (
+ status.phase === "running" ||
+ (status.phase === "stopped" && status.pid === undefined)
+ )
+ this.starting = undefined;
+ }
return status;
}
@@ -268,7 +277,15 @@ export class ReMeEnvManager {
}
if (url.protocol !== "http:" && url.protocol !== "https:")
throw new TypeError("ReMe endpoint must be an http(s) URL");
- return value;
+ if (url.username || url.password)
+ throw new TypeError("ReMe endpoint must not contain credentials");
+ if (url.search || url.hash)
+ throw new TypeError(
+ "ReMe endpoint must not contain a query string or fragment",
+ );
+ // The client appends job names to the endpoint, so it never keeps a
+ // trailing slash; normalising here is what makes records comparable.
+ return `${url.protocol}//${url.host}${url.pathname.replace(/\/+$/, "")}`;
}
private async startLocked(
@@ -281,17 +298,30 @@ export class ReMeEnvManager {
const dir = resolveWorkspaceDir(workspaceDir, homedir());
const logPath = join(dir, "service.log");
const now = new Date().toISOString();
- if (
- this.starting?.endpoint === endpointValue &&
- Date.now() - this.starting.at < START_GRACE_MS
- )
+ const starting = this.starting;
+ if (starting && Date.now() - starting.at < START_GRACE_MS) {
+ if (starting.endpoint === endpointValue)
+ 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
+ // orphan it.
return {
started: false,
- alreadyRunning: true,
+ alreadyRunning: false,
command: "",
logPath,
startedAt: now,
+ error: `A service start for ${starting.endpoint} is still in progress`,
};
+ }
+ this.starting = undefined;
const current = await this.serviceStatus(endpointValue, authToken);
if (current.phase === "running")
return {
@@ -302,6 +332,35 @@ 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) {
+ const previous = await stopService(this.serviceIO, {
+ statePath: this.serviceStatePath,
+ endpoint: endpointValue,
+ });
+ if (!previous.stopped)
+ return {
+ started: false,
+ alreadyRunning: false,
+ command: "",
+ logPath,
+ startedAt: now,
+ error: `A previous launch could not be stopped: ${
+ previous.reason ?? "unknown reason"
+ }`,
+ };
+ }
const path = pythonPath.trim();
if (!path)
return {
@@ -322,17 +381,45 @@ export class ReMeEnvManager {
startedAt: now,
error: "reme-ai is not installed in the selected interpreter",
};
+ if (!check.versionOk)
+ return {
+ started: false,
+ alreadyRunning: false,
+ command: "",
+ logPath,
+ startedAt: now,
+ error: `reme-ai ${
+ check.version || "(unknown version)"
+ } is older than ${MINIMUM_REME}; upgrade it first`,
+ };
+ // Set before the spawn so a second request arriving while the service is
+ // 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() };
- return startService(this.serviceIO, {
- endpoint: endpointValue,
- workspaceDir: dir,
- pythonPath: path,
- cliPath: check.cliPath,
- // 0.4.0.x has no jobs..enabled; only newer releases accept it.
- disableDreamCron: remeVersionAtLeast(check.version, "0.4.1"),
- statePath: this.serviceStatePath,
- logPath,
- env: this.env,
- });
+ try {
+ const result = await startService(this.serviceIO, {
+ endpoint: endpointValue,
+ workspaceDir: dir,
+ pythonPath: path,
+ cliPath: check.cliPath,
+ // 0.4.0.x has no jobs..enabled; only newer releases accept it.
+ disableDreamCron: remeVersionAtLeast(check.version, "0.4.1"),
+ statePath: this.serviceStatePath,
+ logPath,
+ env: this.env,
+ });
+ if (!result.started) this.starting = undefined;
+ return result;
+ } catch (error) {
+ this.starting = undefined;
+ return {
+ started: false,
+ alreadyRunning: false,
+ command: "",
+ logPath,
+ startedAt: now,
+ error: `Could not start the ReMe service: ${safeError(error)}`,
+ };
+ }
}
}
diff --git a/integrations/dsh/src/environment/host/paths.ts b/integrations/dsh/src/environment/host/paths.ts
index 3f61756b..942f1d35 100644
--- a/integrations/dsh/src/environment/host/paths.ts
+++ b/integrations/dsh/src/environment/host/paths.ts
@@ -20,3 +20,16 @@ export function isWindowsPath(value: string): boolean {
export function pathOf(value: string): PlatformPath {
return isWindowsPath(value) ? win32 : posix;
}
+
+/**
+ * Whether a path is absolute in its own platform's semantics.
+ *
+ * `win32.isAbsolute("\\python.exe")` is true, but such a path is relative to
+ * the current drive, which is not what the interpreter and install checks must
+ * accept. Windows therefore needs a drive letter or a UNC prefix.
+ */
+export function isAbsolutePath(value: string): boolean {
+ if (isWindowsPath(value))
+ return /^[A-Za-z]:[\\/]/.test(value) || value.startsWith("\\\\");
+ return value.startsWith("/");
+}
diff --git a/integrations/dsh/src/environment/host/reme.ts b/integrations/dsh/src/environment/host/reme.ts
index 5e6ea244..df30cd13 100644
--- a/integrations/dsh/src/environment/host/reme.ts
+++ b/integrations/dsh/src/environment/host/reme.ts
@@ -1,7 +1,7 @@
import { win32 } from "node:path";
import type { RemeCheck, RemeProbe } from "../types.js";
-import { pathOf } from "./paths.js";
+import { isAbsolutePath, pathOf } from "./paths.js";
import type { RunCommand } from "./runner.js";
/** Minimum reme-ai version this DSH integration is tested against. */
@@ -54,15 +54,25 @@ export async function probeReme(
timeoutMs: 10_000,
});
if (result.error || result.code !== 0)
- return { installed: false, version: "", versionOk: false };
+ return {
+ installed: false,
+ version: "",
+ versionOk: false,
+ ...(result.error ? { error: result.error } : {}),
+ };
const version = parseRemeVersion(result.stdout);
return {
installed: version !== "",
version,
versionOk: remeVersionAtLeast(version),
};
- } catch {
- return { installed: false, version: "", versionOk: false };
+ } catch (error) {
+ return {
+ installed: false,
+ version: "",
+ versionOk: false,
+ error: String(error),
+ };
}
}
@@ -95,7 +105,7 @@ export async function checkReme(
pythonPath: string,
): Promise {
const path = pythonPath.trim();
- if (!path || !pathOf(path).isAbsolute(path)) {
+ if (!path || !isAbsolutePath(path)) {
return {
installed: false,
version: "",
@@ -112,7 +122,7 @@ export async function checkReme(
versionOk: false,
hasCore: false,
cliPath: "",
- error: "",
+ error: probe.error ?? "",
};
}
let hasCore = false;
diff --git a/integrations/dsh/src/environment/host/runner.ts b/integrations/dsh/src/environment/host/runner.ts
index 57346a8a..5e568adf 100644
--- a/integrations/dsh/src/environment/host/runner.ts
+++ b/integrations/dsh/src/environment/host/runner.ts
@@ -53,10 +53,42 @@ export function createNodeRunner(
}
let stdout = "";
let stderr = "";
+ let settled = false;
+ let timedOut = false;
+ const timers: ReturnType[] = [];
+ const settle = (result: RunResult): void => {
+ if (settled) return;
+ settled = true;
+ for (const timer of timers) clearTimeout(timer);
+ resolve(result);
+ };
+ const timeoutMs = options.timeoutMs ?? 10_000;
const timer = setTimeout(() => {
+ // The caller must see an explicit error: an exit caused by our own
+ // timeout is not "the interpreter has no reme-ai".
+ timedOut = true;
child.kill();
- }, options.timeoutMs ?? 10_000);
+ // A child that ignores SIGTERM must not make the probe hang forever:
+ // force it and settle, rather than waiting for a close that may never
+ // arrive.
+ const grace = setTimeout(() => {
+ try {
+ child.kill("SIGKILL");
+ } catch {
+ // Already exited.
+ }
+ settle({
+ code: null,
+ stdout,
+ stderr,
+ error: `Command timed out after ${timeoutMs} ms`,
+ });
+ }, 1_000);
+ grace.unref();
+ timers.push(grace);
+ }, timeoutMs);
timer.unref();
+ timers.push(timer);
child.stdout?.on("data", (chunk) => {
stdout += String(chunk);
});
@@ -64,12 +96,17 @@ export function createNodeRunner(
stderr += String(chunk);
});
child.on("error", (error) => {
- clearTimeout(timer);
- resolve({ code: null, stdout, stderr, error: String(error) });
+ settle({ code: null, stdout, stderr, error: String(error) });
});
child.on("close", (code) => {
- clearTimeout(timer);
- resolve({ code, stdout, stderr });
+ settle({
+ code,
+ stdout,
+ stderr,
+ ...(timedOut
+ ? { error: `Command timed out after ${timeoutMs} ms` }
+ : {}),
+ });
});
});
}
diff --git a/integrations/dsh/src/environment/host/service.ts b/integrations/dsh/src/environment/host/service.ts
index 0bf73937..f7bd370d 100644
--- a/integrations/dsh/src/environment/host/service.ts
+++ b/integrations/dsh/src/environment/host/service.ts
@@ -5,11 +5,12 @@ import {
mkdirSync,
openSync,
readFileSync,
+ renameSync,
rmSync,
writeFileSync,
} from "node:fs";
import { createConnection } from "node:net";
-import { dirname } from "node:path";
+import { basename, dirname } from "node:path";
import { safeError } from "../../endpoint.js";
import type { ServiceStartResult, ServiceState } from "../types.js";
@@ -56,7 +57,11 @@ export interface ServiceIO {
logPath: string;
env: NodeJS.ProcessEnv;
}): Promise<{ pid: number | undefined }>;
- killTree(pid: number): void;
+ /**
+ * Ask the process, and its process group on POSIX, to stop. `force`
+ * escalates to `SIGKILL` (always `/F` on Windows).
+ */
+ killTree(pid: number, force?: boolean): void;
probePid(pid: number): void;
/** OS image name of a pid, or "" when it cannot be read. */
processImage(pid: number): Promise;
@@ -64,7 +69,30 @@ export interface ServiceIO {
platform: string;
}
-/** Only loopback endpoints may be managed as a local service. */
+/** A pid we are willing to signal: > 1, so kill(0)/kill(-1) is impossible. */
+export function isValidPid(pid: unknown): pid is number {
+ return typeof pid === "number" && Number.isInteger(pid) && pid > 1;
+}
+
+/**
+ * Canonical identity of an endpoint for service-record matching. Loopback
+ * aliases and default ports compare equal, so saving `localhost:2333` does not
+ * make the service this plugin started on `127.0.0.1:2333` look foreign.
+ */
+export function serviceEndpointKey(endpoint: string): string {
+ try {
+ const url = new URL(endpoint.trim());
+ const hostname = url.hostname.toLowerCase();
+ const host = hostname === "localhost" ? "127.0.0.1" : hostname;
+ const port = url.port || (url.protocol === "https:" ? "443" : "80");
+ const path = url.pathname.replace(/\/+$/, "");
+ return `${url.protocol}//${host}:${port}${path}`;
+ } catch {
+ return endpoint.trim();
+ }
+}
+
+/** Only loopback HTTP endpoints may be managed as a local service. */
export function parseLocalEndpoint(endpoint: string): {
host: string;
port: number;
@@ -75,13 +103,17 @@ export function parseLocalEndpoint(endpoint: string): {
} 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");
+ // The service this plugin launches is plain HTTP, so an https loopback URL
+ // could never become healthy and must not be offered as manageable.
+ if (url.protocol !== "http:")
+ throw new TypeError(
+ "Local service management requires an http:// endpoint",
+ );
if (!["127.0.0.1", "localhost", "[::1]", "::1"].includes(url.hostname))
throw new TypeError(
"Local service management requires a loopback endpoint",
);
- const port = Number(url.port || (url.protocol === "https:" ? "443" : "80"));
+ const port = Number(url.port || "80");
if (!Number.isInteger(port) || port < 1 || port > 65535)
throw new TypeError("Invalid ReMe endpoint port");
return { host: "127.0.0.1", port };
@@ -129,21 +161,57 @@ export function buildStartCommand(options: {
};
}
+function isRecord(value: unknown): value is Record {
+ return typeof value === "object" && value !== null && !Array.isArray(value);
+}
+
+/** Read a record defensively: a malformed file is "no record", not a crash. */
export function readServiceState(path: string): PersistedService | undefined {
+ let value: unknown;
try {
- return JSON.parse(readFileSync(path, "utf8")) as PersistedService;
+ value = JSON.parse(readFileSync(path, "utf8"));
} catch {
return undefined;
}
+ if (!isRecord(value)) return undefined;
+ const endpoint = typeof value.endpoint === "string" ? value.endpoint : "";
+ if (!endpoint.trim()) return undefined;
+ return {
+ ...(isValidPid(value.pid) ? { pid: value.pid } : {}),
+ endpoint,
+ workspaceDir:
+ typeof value.workspaceDir === "string" ? value.workspaceDir : "",
+ pythonPath: typeof value.pythonPath === "string" ? value.pythonPath : "",
+ command: typeof value.command === "string" ? value.command : "",
+ startedAt: typeof value.startedAt === "string" ? value.startedAt : "",
+ };
}
export function writeServiceState(path: string, state: PersistedService): void {
mkdirSync(dirname(path), { recursive: true });
- writeFileSync(path, JSON.stringify(state, null, 2), "utf8");
+ const temporary = `${path}.${process.pid}.tmp`;
+ writeFileSync(temporary, JSON.stringify(state, null, 2), {
+ encoding: "utf8",
+ mode: 0o600,
+ });
+ try {
+ renameSync(temporary, path);
+ } catch (error) {
+ rmSync(temporary, { force: true });
+ throw error;
+ }
+}
+
+function clearServiceState(path: string): void {
+ try {
+ rmSync(path, { force: true });
+ } catch {
+ // A stale state file is retried by the next status read.
+ }
}
function alive(io: ServiceIO, pid?: number): boolean {
- if (pid === undefined) return false;
+ if (!isValidPid(pid)) return false;
try {
io.probePid(pid);
return true;
@@ -152,23 +220,37 @@ function alive(io: ServiceIO, pid?: number): boolean {
}
}
+/**
+ * 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
+ * would accept e.g. `pycharm64.exe`).
+ */
+export function isServiceImage(image: string): boolean {
+ const name = basename(image.trim().replaceAll("\\", "/")).toLowerCase();
+ return /^(python|pythonw|reme)([0-9.]*)?(\.exe)?$/.test(name);
+}
+
/**
* 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.
+ *
+ * This guards a destructive operation, so it fails *closed*: an unreadable
+ * image is not proof of ownership.
*/
export async function ownsService(
io: ServiceIO,
persisted: PersistedService | undefined,
endpoint: string,
): Promise {
- if (!persisted || persisted.pid === undefined) return false;
- if (persisted.endpoint.trim() !== endpoint.trim()) return false;
+ 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)).toLowerCase();
- if (image === "") return true;
- return /^(reme|python|py)/.test(image);
+ const image = await io.processImage(persisted.pid);
+ if (!image.trim()) return false;
+ return isServiceImage(image);
}
/** Running / conflict / stopped, with ownership of the running process. */
@@ -186,8 +268,19 @@ export async function serviceStatus(
options.authToken ?? "",
);
const persisted = readServiceState(options.statePath);
+ const recordForEndpoint =
+ persisted !== undefined &&
+ serviceEndpointKey(persisted.endpoint) ===
+ serviceEndpointKey(options.endpoint);
+ const recordAlive = alive(io, persisted?.pid);
+ // 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);
if (health.ok) {
- const managed = await ownsService(io, persisted, options.endpoint);
+ const managed =
+ recordForEndpoint &&
+ recordAlive &&
+ (await ownsService(io, persisted, options.endpoint));
return {
phase: "running",
managed,
@@ -200,8 +293,14 @@ export async function serviceStatus(
logPath: options.logPath,
};
}
- const { host, port } = parseLocalEndpoint(options.endpoint);
- if (await io.portOpen(host, port))
+ let portOpen = false;
+ try {
+ const { host, port } = parseLocalEndpoint(options.endpoint);
+ portOpen = await io.portOpen(host, port);
+ } catch {
+ // A remote or https endpoint is health-checked only; it has no local port.
+ }
+ if (portOpen)
return {
phase: "conflict",
managed: false,
@@ -212,7 +311,11 @@ export async function serviceStatus(
phase: "stopped",
managed: false,
endpoint: options.endpoint,
- ...(persisted?.pid === undefined ? {} : { pid: persisted.pid }),
+ // A live record for this endpoint means a launch that never became
+ // healthy; the manager stops it before spawning a replacement.
+ ...(recordForEndpoint && recordAlive && persisted?.pid !== undefined
+ ? { pid: persisted.pid }
+ : {}),
logPath: options.logPath,
};
}
@@ -235,16 +338,28 @@ export async function startService(
env: NodeJS.ProcessEnv;
},
): Promise {
- mkdirSync(options.workspaceDir, { recursive: true });
- const command = buildStartCommand({
- cliPath: options.cliPath,
- pythonPath: options.pythonPath,
- workspaceDir: options.workspaceDir,
- endpoint: options.endpoint,
- disableDreamCron: options.disableDreamCron,
- exists: io.exists,
- });
const startedAt = new Date().toISOString();
+ let command: { exe: string; args: string[]; display: string };
+ try {
+ mkdirSync(options.workspaceDir, { recursive: true });
+ command = buildStartCommand({
+ cliPath: options.cliPath,
+ pythonPath: options.pythonPath,
+ workspaceDir: options.workspaceDir,
+ endpoint: options.endpoint,
+ disableDreamCron: options.disableDreamCron,
+ exists: io.exists,
+ });
+ } catch (error) {
+ return {
+ started: false,
+ alreadyRunning: false,
+ command: "",
+ logPath: options.logPath,
+ startedAt,
+ error: `Could not start the ReMe service: ${safeError(error)}`,
+ };
+ }
let child: { pid: number | undefined };
try {
child = await io.spawnService({
@@ -269,14 +384,32 @@ export async function startService(
error: `Could not start the ReMe service: ${safeError(error)}`,
};
}
- writeServiceState(options.statePath, {
- pid: child.pid,
- endpoint: options.endpoint,
- workspaceDir: options.workspaceDir,
- pythonPath: options.pythonPath,
- command: command.display,
- startedAt,
- });
+ try {
+ writeServiceState(options.statePath, {
+ pid: child.pid,
+ endpoint: options.endpoint,
+ workspaceDir: options.workspaceDir,
+ pythonPath: options.pythonPath,
+ command: command.display,
+ startedAt,
+ });
+ } 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.
+ }
+ return {
+ started: false,
+ alreadyRunning: false,
+ command: command.display,
+ logPath: options.logPath,
+ startedAt,
+ error: `Could not record the ReMe service: ${safeError(error)}`,
+ };
+ }
return {
started: true,
alreadyRunning: false,
@@ -287,34 +420,97 @@ export async function startService(
};
}
+const STOP_TIMEOUT_MS = 5_000;
+const STOP_FORCE_TIMEOUT_MS = 3_000;
+const STOP_POLL_MS = 100;
+
+function delay(ms: number): Promise {
+ return new Promise((resolve) => {
+ setTimeout(resolve, ms);
+ });
+}
+
+/** Poll until the pid disappears; true when it exited within `timeoutMs`. */
+async function waitForExit(
+ io: ServiceIO,
+ pid: number,
+ timeoutMs: number,
+ pollMs: number,
+): Promise {
+ const deadline = Date.now() + timeoutMs;
+ for (;;) {
+ if (!alive(io, pid)) return true;
+ const remaining = deadline - Date.now();
+ if (remaining <= 0) return false;
+ await delay(Math.max(1, Math.min(pollMs, remaining)));
+ }
+}
+
/**
* 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.
+ *
+ * The record is removed only after the process is confirmed gone. Otherwise a
+ * SIGTERM the service ignored would delete the pid, leaving a running process
+ * that the tab can never stop again.
*/
export async function stopService(
io: ServiceIO,
- options: { statePath: string; endpoint: string },
+ options: {
+ statePath: string;
+ endpoint: string;
+ /** Grace period before escalating to SIGKILL (tests shorten it). */
+ timeoutMs?: number;
+ forceTimeoutMs?: number;
+ pollMs?: number;
+ },
): Promise {
const persisted = readServiceState(options.statePath);
- if (!persisted || persisted.pid === undefined)
+ if (!persisted || !isValidPid(persisted.pid)) {
+ if (persisted) clearServiceState(options.statePath);
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;
+ const pollMs = options.pollMs ?? STOP_POLL_MS;
try {
io.killTree(pid);
} catch {
// Already exited.
}
- try {
- rmSync(options.statePath, { force: true });
- } catch {
- // A stale state file is harmless; the next start overwrites it.
+ if (
+ await waitForExit(io, pid, options.timeoutMs ?? STOP_TIMEOUT_MS, pollMs)
+ ) {
+ clearServiceState(options.statePath);
+ return { stopped: true, pid };
}
- return { stopped: true, pid };
+ // The service ignored (or slowly handled) SIGTERM: escalate, then only drop
+ // the record once the process is really gone.
+ try {
+ io.killTree(pid, true);
+ } catch {
+ // Already exited.
+ }
+ if (
+ await waitForExit(
+ io,
+ pid,
+ options.forceTimeoutMs ?? STOP_FORCE_TIMEOUT_MS,
+ pollMs,
+ )
+ ) {
+ clearServiceState(options.statePath);
+ return { stopped: true, pid };
+ }
+ return {
+ stopped: false,
+ pid,
+ reason: "The service did not exit after SIGKILL",
+ };
}
/** Node-backed IO: fetch health, probe the port, spawn detached, tree-kill. */
@@ -402,30 +598,50 @@ export function createServiceIO(): ServiceIO {
child.unref();
return { pid: child.pid };
},
- killTree(pid) {
+ killTree(pid, force = false) {
+ if (!isValidPid(pid)) return;
if (process.platform === "win32") {
- spawn("taskkill", ["/PID", String(pid), "/T", "/F"], {
+ const child = spawn("taskkill", ["/PID", String(pid), "/T", "/F"], {
windowsHide: true,
stdio: "ignore",
});
+ // A spawn failure is an async event; without a listener it would exit
+ // the Host.
+ child.on("error", () => undefined);
+ child.unref();
return;
}
+ // The service was spawned detached, so it leads its own process group;
+ // signalling the group stops pip's children too.
+ const signal = force ? "SIGKILL" : "SIGTERM";
try {
- process.kill(pid, "SIGTERM");
+ process.kill(-pid, signal);
+ return;
+ } catch {
+ // Not a group leader (e.g. a record from an older launch).
+ }
+ try {
+ process.kill(pid, signal);
} catch {
// Already exited.
}
},
probePid(pid) {
+ if (!isValidPid(pid))
+ throw new RangeError("Refusing to probe an invalid pid");
process.kill(pid, 0);
},
processImage(pid) {
return new Promise((resolve) => {
+ if (!isValidPid(pid)) {
+ resolve("");
+ return;
+ }
if (process.platform === "win32") {
execFile(
"tasklist",
["/FI", `PID eq ${pid}`, "/FO", "CSV", "/NH"],
- { windowsHide: true },
+ { windowsHide: true, timeout: 3000 },
(error, stdout) => {
const match = /^"([^"]+)"/.exec(String(stdout ?? "").trim());
resolve(error ? "" : match?.[1] ?? "");
@@ -433,8 +649,11 @@ export function createServiceIO(): ServiceIO {
);
return;
}
- execFile("ps", ["-p", String(pid), "-o", "comm="], (error, stdout) =>
- resolve(error ? "" : String(stdout ?? "").trim()),
+ execFile(
+ "ps",
+ ["-p", String(pid), "-o", "comm="],
+ { timeout: 3000 },
+ (error, stdout) => resolve(error ? "" : String(stdout ?? "").trim()),
);
});
},
diff --git a/integrations/dsh/src/environment/host/workspace.ts b/integrations/dsh/src/environment/host/workspace.ts
index ff8c5d31..d01a534f 100644
--- a/integrations/dsh/src/environment/host/workspace.ts
+++ b/integrations/dsh/src/environment/host/workspace.ts
@@ -1,4 +1,13 @@
-import { mkdirSync, readFileSync, writeFileSync } from "node:fs";
+import {
+ chmodSync,
+ copyFileSync,
+ existsSync,
+ mkdirSync,
+ readFileSync,
+ renameSync,
+ rmSync,
+ writeFileSync,
+} from "node:fs";
import { dirname, isAbsolute, join, resolve } from "node:path";
import { safeError } from "../../endpoint.js";
@@ -6,13 +15,13 @@ import type { EnvSummary, LlmTestResult } from "../types.js";
import type { RunCommand } from "./runner.js";
/** Keys the Environment tab manages; split for the two form groups. */
-export const LLM_KEYS = [
+const LLM_KEYS = [
"LLM_BACKEND",
"LLM_MODEL_NAME",
"LLM_API_KEY",
"LLM_BASE_URL",
] as const;
-export const EMBEDDING_KEYS = [
+const EMBEDDING_KEYS = [
"EMBEDDING_BACKEND",
"EMBEDDING_MODEL_NAME",
"EMBEDDING_API_KEY",
@@ -138,10 +147,33 @@ export function validateEnvUpdate(values: Record): void {
/**
* Write a fully merged `.env`. The caller merges and validates first, so the
* file on disk is exactly what was validated.
+ *
+ * The write is atomic (same-directory temp file plus rename) and keeps the
+ * previous file as `.env.bak`: a crash or a full disk must not take unrelated
+ * keys such as `LLM_API_KEY` with it. New files are owner-only (0600) because
+ * they hold credentials.
*/
export function writeEnvText(path: string, text: string): void {
mkdirSync(dirname(path), { recursive: true });
- writeFileSync(path, text, "utf8");
+ const temporary = `${path}.${process.pid}.tmp`;
+ writeFileSync(temporary, text, { encoding: "utf8", mode: 0o600 });
+ try {
+ if (existsSync(path)) {
+ const backup = `${path}.bak`;
+ copyFileSync(path, backup);
+ // The backup holds the same credentials, so it must not be more readable
+ // than the file it preserves.
+ try {
+ chmodSync(backup, 0o600);
+ } catch {
+ // Windows has no POSIX mode bits; the copy is still user-owned.
+ }
+ }
+ renameSync(temporary, path);
+ } catch (error) {
+ rmSync(temporary, { force: true });
+ throw error;
+ }
}
/**
diff --git a/integrations/dsh/src/environment/types.ts b/integrations/dsh/src/environment/types.ts
index ebc09168..82fceaf1 100644
--- a/integrations/dsh/src/environment/types.ts
+++ b/integrations/dsh/src/environment/types.ts
@@ -47,6 +47,8 @@ export interface RemeProbe {
version: string;
/** True when the version satisfies the minimum this integration targets. */
versionOk: boolean;
+ /** Non-empty when the probe itself failed (spawn error, timeout). */
+ error?: string;
}
export interface RemeCheck extends RemeProbe {
diff --git a/integrations/dsh/src/reme/client.ts b/integrations/dsh/src/reme/client.ts
index 50d7f65b..e00b0998 100644
--- a/integrations/dsh/src/reme/client.ts
+++ b/integrations/dsh/src/reme/client.ts
@@ -52,9 +52,15 @@ export class ReMeClient {
sessionId: string,
options: AutoMemoryOptions = {},
): Promise {
- const config = {
- ...this.configSource(),
- ...(options.endpoint ? { endpoint: options.endpoint } : {}),
+ const base = this.configSource();
+ const endpoint = options.endpoint || base.endpoint;
+ // A recovered batch keeps the endpoint it was queued for. The token is
+ // scoped to the *configured* endpoint, so it must not ride along when the
+ // batch goes somewhere else.
+ const config: ReMeClientConfig = {
+ ...base,
+ endpoint,
+ ...(endpoint === base.endpoint ? {} : { authToken: "" }),
};
return this.request(
"auto_memory",
diff --git a/integrations/dsh/tests/client.test.mjs b/integrations/dsh/tests/client.test.mjs
index 21ed3d3c..5c0d8a81 100644
--- a/integrations/dsh/tests/client.test.mjs
+++ b/integrations/dsh/tests/client.test.mjs
@@ -100,6 +100,32 @@ test("uses the persisted destination and request ID when sending a recovered bat
}
});
+test("scopes the bearer token to the endpoint it was configured for", async () => {
+ const originalFetch = globalThis.fetch;
+ const requests = [];
+ globalThis.fetch = async (url, init) => {
+ requests.push({ url, authorization: init.headers.Authorization });
+ return Response.json({ success: true });
+ };
+ try {
+ const client = new ReMeClient({
+ endpoint: "http://new.test",
+ authToken: "current-token",
+ requestTimeoutMs: 1000,
+ backgroundTimeoutMs: 1000,
+ });
+ // The current endpoint keeps the token.
+ await client.search("fact");
+ assert.equal(requests.at(-1).authorization, "Bearer current-token");
+ // A recovered batch for another endpoint must not receive it.
+ await client.autoMemory([], "session", { endpoint: "http://old.test" });
+ assert.equal(requests.at(-1).url, "http://old.test/auto_memory");
+ assert.equal(requests.at(-1).authorization, undefined);
+ } finally {
+ globalThis.fetch = originalFetch;
+ }
+});
+
test("calls ReMe jobs with their native request and response envelopes", async () => {
const calls = [];
const originalFetch = globalThis.fetch;
diff --git a/integrations/dsh/tests/env-detect.test.mjs b/integrations/dsh/tests/env-detect.test.mjs
index 118ea565..f1749cf9 100644
--- a/integrations/dsh/tests/env-detect.test.mjs
+++ b/integrations/dsh/tests/env-detect.test.mjs
@@ -607,6 +607,49 @@ test("detects Unix interpreters and merges paths that resolve together", async (
assert.equal(named.version, "3.11.9");
});
+test("probes well-known Unix locations beyond the Host PATH", async () => {
+ const homebrew = "/opt/homebrew/bin/python3";
+ const pyenv = "/Users/u/.pyenv/shims/python3";
+ const rules = [
+ {
+ match: (exe) => exe === "which",
+ result: { code: 1, stdout: "", stderr: "" },
+ },
+ verifyRule(homebrew, "3.12.4", "/opt/homebrew"),
+ // pyenv shims report the real interpreter they dispatch to.
+ {
+ match: (actual, args) =>
+ actual === pyenv &&
+ args[0] === VERIFY_ARG0 &&
+ String(args[1]).includes("import sys"),
+ result: {
+ code: 0,
+ stdout:
+ '{"exe":"/Users/u/.pyenv/versions/3.12.4/bin/python3","ver":"3.12.4","base":"/Users/u/.pyenv/versions/3.12.4"}\n',
+ stderr: "",
+ },
+ },
+ remeRule(homebrew, "0.4.1.11"),
+ remeRule(pyenv, null),
+ remeRule("/Users/u/.pyenv/versions/3.12.4/bin/python3", null),
+ ];
+ const detection = await detectPythonEnvs({
+ run: fakeRun(rules),
+ platform: "darwin",
+ env: {},
+ homedir: "/Users/u",
+ exists: (path) => path === homebrew || path === pyenv,
+ });
+ assert.deepEqual(
+ detection.envs.map((env) => env.exe).sort(),
+ [homebrew, "/Users/u/.pyenv/versions/3.12.4/bin/python3"].sort(),
+ );
+ assert.ok(
+ detection.envs.every((env) => env.origin.kind === "system"),
+ "well-known paths are classified as system installations",
+ );
+});
+
test("collects conda failures without rejecting the whole detection", async () => {
const { io, miniconda } = windowsFixture();
io.run = async (exe, args) => {
diff --git a/integrations/dsh/tests/env-manager.test.mjs b/integrations/dsh/tests/env-manager.test.mjs
index ae7cd21d..85603a99 100644
--- a/integrations/dsh/tests/env-manager.test.mjs
+++ b/integrations/dsh/tests/env-manager.test.mjs
@@ -5,10 +5,11 @@ 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";
/** A service adapter that records launches instead of starting processes. */
function fakeServiceIO() {
- const calls = { spawn: [] };
+ const calls = { killTree: [], spawn: [] };
return {
calls,
io: {
@@ -49,7 +50,7 @@ function managerIn(t) {
endpoint: endpoint || "http://127.0.0.1:2333",
logPath: join(directory, "service.log"),
});
- return { manager, calls, directory };
+ return { manager, calls, directory, io };
}
test("serializes concurrent starts into one service launch", async (t) => {
@@ -69,6 +70,84 @@ test("serializes concurrent starts into one service launch", async (t) => {
assert.equal(second.alreadyRunning, true);
});
+test("a failed spawn can be retried immediately", async (t) => {
+ const { manager, calls, directory, io } = managerIn(t);
+ io.spawnService = async (options) => {
+ calls.spawn.push(options);
+ if (calls.spawn.length === 1) throw new Error("spawn ENOENT");
+ return { pid: 7000 };
+ };
+ const options = [
+ "http://127.0.0.1:2334",
+ "D:\\env\\python.exe",
+ join(directory, "workspace"),
+ ];
+ const first = await manager.serviceStart(...options);
+ assert.equal(first.started, false);
+ assert.match(first.error, /ENOENT/);
+
+ // The start marker was cleared with the failure; a retry must really spawn
+ // instead of being reported as "already starting".
+ const second = await manager.serviceStart(...options);
+ assert.equal(second.started, true);
+ assert.equal(calls.spawn.length, 2);
+});
+
+test("refuses to spawn while another process owns the port", async (t) => {
+ const { manager, calls, directory } = managerIn(t);
+ manager.serviceStatus = async (endpoint) => ({
+ phase: "conflict",
+ managed: false,
+ endpoint,
+ logPath: join(directory, "service.log"),
+ });
+ 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, /already listening/);
+ assert.equal(calls.spawn.length, 0);
+});
+
+test("stops a previous launch that never became healthy before replacing it", async (t) => {
+ const { manager, calls, directory, io } = managerIn(t);
+ manager.serviceStatus = async (endpoint) => ({
+ phase: "stopped",
+ managed: false,
+ endpoint,
+ // A live record for this endpoint means a launch that is still booting.
+ pid: 6000,
+ logPath: join(directory, "service.log"),
+ });
+ writeServiceState(join(directory, "reme-memory", "service.json"), {
+ 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",
+ });
+ let running = true;
+ io.probePid = () => {
+ if (!running) throw new Error("ESRCH");
+ };
+ io.killTree = (pid) => {
+ calls.killTree.push(pid);
+ running = false;
+ };
+
+ 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, [6000]);
+ assert.equal(calls.spawn.length, 1);
+});
+
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");
diff --git a/integrations/dsh/tests/env-reme.test.mjs b/integrations/dsh/tests/env-reme.test.mjs
index 3d70a672..ff8b1d0b 100644
--- a/integrations/dsh/tests/env-reme.test.mjs
+++ b/integrations/dsh/tests/env-reme.test.mjs
@@ -148,6 +148,15 @@ test("composes pip arguments for install, upgrade and mirrors", () => {
assert.throws(() =>
pipArgs({ pythonPath: "C:\\env\\python.exe", indexUrl: "ftp://x" }),
);
+ // Credentials would be echoed in the job command, the log and the browser.
+ assert.throws(
+ () =>
+ pipArgs({
+ pythonPath: "C:\\env\\python.exe",
+ indexUrl: "https://user:token@mirror.example/simple",
+ }),
+ /must not embed credentials/,
+ );
});
test("tails a log with CRLF and progress carriage returns", (t) => {
@@ -281,5 +290,8 @@ test("reuses the running job for the same interpreter and validates paths", (t)
const second = jobs.start({ pythonPath: "C:\\env\\python.exe" });
assert.equal(first.id, second.id);
assert.throws(() => jobs.start({ pythonPath: "relative\\python.exe" }));
+ // Root-relative Windows paths pass win32.isAbsolute but are not absolute in
+ // the sense the install needs (they resolve against the current drive).
+ assert.throws(() => jobs.start({ pythonPath: "\\python.exe" }));
assert.equal(jobs.status("unknown-job"), null);
});
diff --git a/integrations/dsh/tests/env-runner.test.mjs b/integrations/dsh/tests/env-runner.test.mjs
new file mode 100644
index 00000000..4e359b7b
--- /dev/null
+++ b/integrations/dsh/tests/env-runner.test.mjs
@@ -0,0 +1,30 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+
+import { createNodeRunner } from "../dist/environment/host/runner.js";
+
+test("a timeout settles with an explicit error instead of hanging", async () => {
+ // The child ignores SIGTERM, so the grace path must force it and settle.
+ const run = createNodeRunner({ PATH: process.env.PATH });
+ const started = Date.now();
+ const result = await run(
+ process.execPath,
+ ["-e", "process.on('SIGTERM', () => {}); setInterval(() => {}, 1000)"],
+ { timeoutMs: 200 },
+ );
+ assert.match(result.error, /timed out after 200 ms/);
+ assert.ok(
+ Date.now() - started < 5_000,
+ "the probe must not wait for a process that never closes",
+ );
+});
+
+test("a child that exits on SIGTERM is still reported as a timeout", async () => {
+ const run = createNodeRunner({ PATH: process.env.PATH });
+ const result = await run(
+ process.execPath,
+ ["-e", "setInterval(() => {}, 1000)"],
+ { timeoutMs: 200 },
+ );
+ assert.match(result.error ?? "", /timed out after 200 ms/);
+});
diff --git a/integrations/dsh/tests/env-service.test.mjs b/integrations/dsh/tests/env-service.test.mjs
index c848e3fb..c7c0504b 100644
--- a/integrations/dsh/tests/env-service.test.mjs
+++ b/integrations/dsh/tests/env-service.test.mjs
@@ -1,11 +1,13 @@
import assert from "node:assert/strict";
import test from "node:test";
+import { createServer } from "node:http";
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import {
buildStartCommand,
+ createServiceIO,
ownsService,
parseLocalEndpoint,
readServiceState,
@@ -18,7 +20,7 @@ import {
const ENDPOINT = "http://127.0.0.1:2334";
function fakeIO(overrides = {}) {
- const calls = { killTree: [], spawn: [] };
+ const calls = { killTree: [], force: [], spawn: [] };
const io = {
fetchHealth: async () => ({ ok: false, version: "", error: "" }),
portOpen: async () => false,
@@ -26,8 +28,9 @@ function fakeIO(overrides = {}) {
calls.spawn.push(options);
return { pid: 4321 };
},
- killTree(pid) {
+ killTree(pid, force = false) {
calls.killTree.push(pid);
+ if (force) calls.force.push(pid);
},
probePid() {},
processImage: async () => "python.exe",
@@ -56,7 +59,7 @@ function record(statePath, overrides = {}) {
});
}
-test("accepts only loopback endpoints and validates the port", () => {
+test("accepts only loopback HTTP endpoints and validates the port", () => {
assert.deepEqual(parseLocalEndpoint("http://127.0.0.1:2334"), {
host: "127.0.0.1",
port: 2334,
@@ -68,6 +71,12 @@ test("accepts only loopback endpoints and validates the port", () => {
assert.throws(() => parseLocalEndpoint("http://10.0.0.5:2333"));
assert.throws(() => parseLocalEndpoint("not-a-url"));
assert.throws(() => parseLocalEndpoint("http://127.0.0.1:0"));
+ // The plugin starts a plain HTTP service, so https on loopback can never be
+ // healthy; it must not be offered as manageable.
+ assert.throws(
+ () => parseLocalEndpoint("https://127.0.0.1:2333"),
+ /http:\/\//,
+ );
});
test("builds reme start with the workspace cwd contract", () => {
@@ -162,6 +171,21 @@ test("a service on another endpoint is not reported as managed", async (t) => {
assert.equal(status.pid, undefined);
});
+test("a loopback alias still identifies the service this plugin started", async (t) => {
+ const { directory, statePath } = tempState(t, "reme-alias-");
+ record(statePath, { endpoint: "http://localhost:2334/" });
+ 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.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);
@@ -182,6 +206,35 @@ test("a recycled pid that is no longer the service is not managed", async (t) =>
await ownsService(gone, readServiceState(statePath), ENDPOINT),
false,
);
+ // Ownership guards a destructive action: an image that cannot be read is
+ // not evidence that the pid is ours, and a prefix match would accept e.g.
+ // pycharm64.exe.
+ for (const image of ["", " ", "pycharm64.exe", "python-config"]) {
+ const { io: unreadable } = fakeIO({ processImage: async () => image });
+ assert.equal(
+ await ownsService(unreadable, readServiceState(statePath), ENDPOINT),
+ false,
+ `image ${JSON.stringify(image)} must not be owned`,
+ );
+ }
+});
+
+test("a dead record is cleared instead of reporting a stale pid", async (t) => {
+ const { directory, statePath } = tempState(t, "reme-dead-");
+ record(statePath);
+ const { io } = fakeIO({
+ probePid() {
+ throw new Error("ESRCH");
+ },
+ });
+ const status = await serviceStatus(io, {
+ endpoint: ENDPOINT,
+ statePath,
+ logPath: join(directory, "service.log"),
+ });
+ assert.equal(status.phase, "stopped");
+ assert.equal(status.pid, undefined);
+ assert.equal(readServiceState(statePath), undefined);
});
test("starts detached with the GBK-safe environment and records state", async (t) => {
@@ -230,11 +283,24 @@ test("records nothing when the spawn fails", async (t) => {
assert.equal(readServiceState(statePath), undefined);
});
-test("stops the managed pid and clears the state file", 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 stopped = await stopService(io, { statePath, endpoint: ENDPOINT });
+ let running = true;
+ io.probePid = () => {
+ if (!running) throw new Error("ESRCH");
+ };
+ io.killTree = (pid) => {
+ calls.killTree.push(pid);
+ running = false;
+ };
+ const stopped = await stopService(io, {
+ statePath,
+ endpoint: ENDPOINT,
+ timeoutMs: 100,
+ pollMs: 5,
+ });
assert.deepEqual(stopped, { stopped: true, pid: 999 });
assert.deepEqual(calls.killTree, [999]);
assert.equal(readServiceState(statePath), undefined);
@@ -244,6 +310,25 @@ test("stops the managed pid and clears the state file", 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 stopped = await stopService(io, {
+ statePath,
+ endpoint: ENDPOINT,
+ timeoutMs: 10,
+ forceTimeoutMs: 10,
+ pollMs: 2,
+ });
+ assert.equal(stopped.stopped, false);
+ assert.match(stopped.reason, /did not exit/);
+ assert.deepEqual(calls.killTree, [999, 999]);
+ assert.deepEqual(calls.force, [999]);
+ // The pid survives in the record, so Stop can be tried again later.
+ assert.equal(readServiceState(statePath)?.pid, 999);
+});
+
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" });
@@ -254,3 +339,80 @@ test("refuses to stop a service recorded for another endpoint", async (t) => {
assert.deepEqual(calls.killTree, []);
assert.ok(readServiceState(statePath), "the foreign record is left in place");
});
+
+/* ------------------------------------------------------------------ */
+/* Real IO: the pieces the fake adapter above cannot vouch for. */
+/* ------------------------------------------------------------------ */
+
+async function listening(handler, t) {
+ const server = createServer(handler);
+ await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
+ t.after(() => server.close());
+ const address = server.address();
+ return `http://127.0.0.1:${address.port}`;
+}
+
+test("real IO accepts only a ReMe health report", async (t) => {
+ const reme = await listening((request, response) => {
+ response.setHeader("Content-Type", "application/json");
+ if (request.url === "/health_check")
+ response.end(
+ JSON.stringify({
+ success: true,
+ metadata: { health: { version: "0.4.2.1", healthy: true } },
+ }),
+ );
+ else response.end("{}");
+ }, t);
+ const generic = await listening((_request, response) => {
+ response.setHeader("Content-Type", "application/json");
+ response.end("{}");
+ }, t);
+
+ const io = createServiceIO();
+ const good = await io.fetchHealth(reme, "");
+ assert.equal(good.ok, true);
+ assert.equal(good.version, "0.4.2.1");
+
+ const empty = await io.fetchHealth(generic, "");
+ assert.equal(empty.ok, false);
+ assert.match(empty.error, /ReMe health report/);
+
+ // The Authorization header is actually sent when a token is configured.
+ let authorization = "";
+ const guarded = await listening((request, response) => {
+ authorization = String(request.headers.authorization ?? "");
+ response.end("{}");
+ }, t);
+ await io.fetchHealth(guarded, "secret-token");
+ assert.equal(authorization, "Bearer secret-token");
+});
+
+test("real IO spawns, identifies and stops a detached process", async (t) => {
+ const directory = mkdtempSync(join(tmpdir(), "reme-real-service-"));
+ t.after(() => rmSync(directory, { recursive: true, force: true }));
+ const io = createServiceIO();
+ const { pid } = await io.spawnService({
+ exe: process.execPath,
+ args: ["-e", "setInterval(() => {}, 1000)"],
+ cwd: directory,
+ logPath: join(directory, "service.log"),
+ env: process.env,
+ });
+ assert.ok(pid !== undefined && pid > 1);
+ assert.doesNotThrow(() => io.probePid(pid));
+ assert.doesNotMatch(await io.processImage(pid), /^$/);
+
+ io.killTree(pid);
+ const deadline = Date.now() + 5_000;
+ let running = true;
+ while (running && Date.now() < deadline) {
+ try {
+ io.probePid(pid);
+ await new Promise((resolve) => setTimeout(resolve, 50));
+ } catch {
+ running = false;
+ }
+ }
+ assert.equal(running, false, "the detached process must actually exit");
+});
diff --git a/integrations/dsh/tests/env-tab.test.mjs b/integrations/dsh/tests/env-tab.test.mjs
index c75d3560..43e0eeff 100644
--- a/integrations/dsh/tests/env-tab.test.mjs
+++ b/integrations/dsh/tests/env-tab.test.mjs
@@ -52,13 +52,21 @@ const RESPONSES = {
},
"remeEnv/installStatus": null,
"remeEnv/readEnv": {
+ workspaceDir: "C:\\Users\\test\\.reme",
exists: true,
envPath: "C:\\Users\\test\\.reme\\.env",
entries: [
- { key: "LLM_BACKEND", value: "dashscope", secret: false, set: true },
+ {
+ key: "LLM_BACKEND",
+ value: "dashscope",
+ masked: "dashscope",
+ secret: false,
+ set: true,
+ },
{
key: "LLM_MODEL_NAME",
value: "qwen3.7-plus",
+ masked: "qwen3.7-plus",
secret: false,
set: true,
},
@@ -70,6 +78,7 @@ const RESPONSES = {
managed: false,
pid: 98820,
endpoint: "http://127.0.0.1:2333",
+ logPath: "C:\\Users\\test\\.reme\\service.log",
},
};
@@ -250,6 +259,32 @@ test("marks the suggested interpreter and explains why", async (t) => {
]);
});
+test("sends the chosen package mirror with the install request", async (t) => {
+ const { renderer, calls, restore } = await mount();
+ t.after(restore);
+
+ const select = renderer.root.findAll((node) => node.type === "select")[0];
+ await act(async () => {
+ select.props.onChange({
+ target: { value: "https://pypi.tuna.tsinghua.edu.cn/simple" },
+ });
+ });
+ await act(async () => {
+ buttonByText(renderer, envZh.reinstall).props.onClick();
+ });
+ await act(async () => {
+ buttonByText(renderer, envZh.confirmInstall).props.onClick();
+ });
+
+ const install = calls.rpc.filter((call) => call.method === "remeEnv/install");
+ assert.ok(install.length > 0, "the install request must be sent");
+ assert.deepEqual(install.at(-1).input, {
+ pythonPath: SUGGESTED,
+ upgrade: false,
+ indexUrl: "https://pypi.tuna.tsinghua.edu.cn/simple",
+ });
+});
+
test("drops .env drafts when the workspace changes", async (t) => {
const { renderer, calls, setValue, restore } = await mount();
t.after(restore);