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