diff --git a/integrations/dsh/src/client/env-tab.tsx b/integrations/dsh/src/client/env-tab.tsx index 7f6d2539..b9359924 100644 --- a/integrations/dsh/src/client/env-tab.tsx +++ b/integrations/dsh/src/client/env-tab.tsx @@ -4,6 +4,11 @@ import type { ConfigForm } from "@deepseek-ai/dsh-client-ui-settings/client"; import { safeError } from "../endpoint.js"; import type { InstallStatus } from "../pyenv/install.js"; import type { RemeCheck } from "../pyenv/reme.js"; +import type { + ServiceLog, + ServiceStartResult, + ServiceState, +} from "../pyenv/service.js"; import type { PythonDetection, PythonEnv, @@ -111,6 +116,26 @@ export const envEn = { testingLlm: "Testing…", llmOk: "LLM reachable ({ms} ms)", llmFailed: "LLM test failed", + serviceHint: + "Start or stop the local ReMe service. The service keeps running after DSH closes.", + serviceBadgeStarting: "Starting…", + serviceBadgeConflict: "Port occupied", + serviceBadgeStopped: "Stopped", + serviceRunning: "Running", + startService: "Start service", + stopService: "Stop service", + startTimeout: "The service did not become healthy within 90 seconds", + stopConfirm: + "Stop the ReMe service? Automatic memory in flight may need a retry.", + foreignNote: + "This service was started outside DSH; its state is shown but it is not managed here.", + conflictNote: + "Another program occupies this port. Change the service URL port in Memory settings.", + viewLogs: "View log", + logEmpty: "The service log is empty.", + autoStartLabel: "Start the service automatically with DSH", + managedNote: "Managed by this plugin · PID {pid}", + serviceError: "Service operation failed", configTitle: "ReMe parameters", controlTitle: "Service control", upcoming: @@ -209,6 +234,22 @@ export const envZh: typeof envEn = { testingLlm: "正在测试…", llmOk: "LLM 连接正常({ms} ms)", llmFailed: "LLM 测试失败", + serviceHint: "启动或停止本地 ReMe 服务;DSH 关闭后服务仍会保持运行。", + serviceBadgeStarting: "启动中…", + serviceBadgeConflict: "端口被占用", + serviceBadgeStopped: "已停止", + serviceRunning: "运行中", + startService: "启动服务", + stopService: "停止服务", + startTimeout: "服务在 90 秒内未就绪", + stopConfirm: "停止 ReMe 服务?进行中的自动记忆可能需要重试。", + foreignNote: "该服务由 DSH 之外的方式启动,这里只显示状态、不由本插件管理。", + conflictNote: "端口被其他程序占用;请在『记忆设置』中修改服务地址的端口。", + viewLogs: "查看日志", + logEmpty: "服务日志为空。", + autoStartLabel: "随 DSH 启动时自动拉起服务", + managedNote: "由本插件托管 · PID {pid}", + serviceError: "服务操作失败", configTitle: "ReMe 参数", controlTitle: "服务控制", upcoming: "该分区将随环境管理特性的后续里程碑提供。", @@ -266,6 +307,18 @@ interface LlmTestState { result?: LlmTestResult; } +interface ServiceViewState { + phase: "checking" | "running" | "starting" | "stopped" | "conflict" | "error"; + data?: ServiceState; + error?: string; +} + +interface ServiceLogView { + open: boolean; + lines: string; + logPath: string; +} + function isLocalEndpoint(endpoint: string): boolean { try { const url = new URL(endpoint); @@ -364,6 +417,17 @@ export function ReMeEnvironmentTab({ const [workspaceInput, setWorkspaceInput] = useState( undefined, ); + const [service, setService] = useState({ + phase: "checking", + }); + const [serviceTick, setServiceTick] = useState(0); + const [stopping, setStopping] = useState(false); + const [confirmingStop, setConfirmingStop] = useState(false); + const [logView, setLogView] = useState({ + open: false, + lines: "", + logPath: "", + }); // Derived values must precede hook dependency arrays. const detected = detection.phase === "ready" ? detection.data : undefined; @@ -612,6 +676,41 @@ export function ReMeEnvironmentTab({ return () => controller.abort(); }, [loadEnv, workspaceDir, envTick]); + const loadService = useCallback( + async (signal: AbortSignal) => { + try { + const result = await rpc.call( + "/api", + "remeEnv/serviceStatus", + { args: { input: { endpoint } } }, + signal, + ); + if (!result.ok) + throw new Error(result.error.message || result.error.code); + const state = parseServiceState(result.value); + setService((current) => + current.phase === "starting" + ? current + : { phase: state.phase, data: state }, + ); + } catch (error) { + if (!signal.aborted) + setService((current) => + current.phase === "starting" + ? current + : { phase: "error", error: safeError(error) }, + ); + } + }, + [rpc, endpoint], + ); + + useEffect(() => { + const controller = new AbortController(); + void loadService(controller.signal); + return () => controller.abort(); + }, [loadService, serviceTick]); + if (snapshot.status === "unavailable") return null; const select = (env: PythonEnv): void => { @@ -750,6 +849,81 @@ export function ReMeEnvironmentTab({ setEnvTick((tick) => tick + 1); }; + const startService = async (): Promise => { + if (!selectedPath || service.phase === "starting") return; + setService({ phase: "starting" }); + setLogView({ open: false, lines: "", logPath: "" }); + try { + const result = await rpc.call("/api", "remeEnv/serviceStart", { + args: { + input: { endpoint, pythonPath: selectedPath, workspaceDir }, + }, + }); + if (!result.ok) + throw new Error(result.error.message || result.error.code); + const started = parseServiceStart(result.value); + if (started.error) { + setService({ phase: "error", error: started.error }); + return; + } + if (started.alreadyRunning) { + setService({ phase: "running" }); + setServiceTick((tick) => tick + 1); + return; + } + const deadline = Date.now() + 90_000; + for (;;) { + await new Promise((resolve) => setTimeout(resolve, 1000)); + if (Date.now() > deadline) { + setService({ phase: "error", error: t("startTimeout") }); + return; + } + const statusResult = await rpc.call("/api", "remeEnv/serviceStatus", { + args: { input: { endpoint } }, + }); + if (!statusResult.ok) continue; + const status = parseServiceState(statusResult.value); + if (status.phase === "running") { + setService({ phase: "running", data: status }); + return; + } + } + } catch (error) { + setService({ phase: "error", error: safeError(error) }); + } + }; + + const stopService = async (): Promise => { + if (stopping) return; + setStopping(true); + setConfirmingStop(false); + try { + const result = await rpc.call("/api", "remeEnv/serviceStop", { + args: { input: { endpoint } }, + }); + if (!result.ok) + throw new Error(result.error.message || result.error.code); + setServiceTick((tick) => tick + 1); + } catch (error) { + setService({ phase: "error", error: safeError(error) }); + } finally { + setStopping(false); + } + }; + + const viewLog = async (): Promise => { + try { + const result = await rpc.call("/api", "remeEnv/serviceLog", { + args: { input: { workspaceDir } }, + }); + if (!result.ok) return; + const log = parseServiceLog(result.value); + setLogView({ open: true, lines: log.lines, logPath: log.logPath }); + } catch { + // Log viewing is diagnostic only. + } + }; + return (
@@ -1322,9 +1496,157 @@ export function ReMeEnvironmentTab({

{t("controlTitle")}

-

- {t("upcoming")} -

+

{t("serviceHint")}

+
+ {t("endpointLabel")} + {endpoint} + {service.phase === "checking" || service.phase === "starting" ? ( + + + {t( + service.phase === "starting" + ? "serviceBadgeStarting" + : "checking", + )} + + ) : service.phase === "running" ? ( + + + {service.data?.version + ? `ReMe ${service.data.version}` + : t("serviceRunning")} + + ) : service.phase === "conflict" ? ( + + + {t("serviceBadgeConflict")} + + ) : ( + + + {t("serviceBadgeStopped")} + + )} + +
+ {service.phase === "running" ? ( +

+ {service.data?.managed + ? t("managedNote").replace( + "{pid}", + String(service.data?.pid ?? ""), + ) + : t("foreignNote")} +

+ ) : null} + {service.phase === "conflict" ? ( +

{t("conflictNote")}

+ ) : null} + {service.phase === "error" && service.error ? ( +

+ {`${t("serviceError")}: ${service.error}`} +

+ ) : null} +
+ {service.phase === "running" ? ( + confirmingStop ? ( +
+

{t("stopConfirm")}

+ + +
+ ) : ( + + ) + ) : ( + + )} + + +
+ {logView.open ? ( +
+

+ {`${t("envPath")}: ${logView.logPath}`} +

+
+                  {logView.lines || t("logEmpty")}
+                
+ +
+ ) : null}
)} @@ -1387,3 +1709,33 @@ function parseLlmResult(value: unknown): LlmTestResult { throw new Error("Invalid LLM test response"); return value as unknown as LlmTestResult; } + +function parseServiceState(value: unknown): ServiceState { + if ( + typeof value !== "object" || + value === null || + typeof (value as { phase?: unknown }).phase !== "string" + ) + throw new Error("Invalid service status response"); + return value as unknown as ServiceState; +} + +function parseServiceStart(value: unknown): ServiceStartResult { + if ( + typeof value !== "object" || + value === null || + typeof (value as { started?: unknown }).started !== "boolean" + ) + throw new Error("Invalid service start response"); + return value as unknown as ServiceStartResult; +} + +function parseServiceLog(value: unknown): ServiceLog { + if ( + typeof value !== "object" || + value === null || + typeof (value as { lines?: unknown }).lines !== "string" + ) + throw new Error("Invalid service log response"); + return value as unknown as ServiceLog; +} diff --git a/integrations/dsh/src/client/index.tsx b/integrations/dsh/src/client/index.tsx index 24e6f26e..7ea7f89c 100644 --- a/integrations/dsh/src/client/index.tsx +++ b/integrations/dsh/src/client/index.tsx @@ -358,6 +358,7 @@ function ReMeSettingsCard({ typeof value.remeWorkspaceDir === "string" ? value.remeWorkspaceDir : "", + autoStart: value.autoStart === true, } : undefined, ); @@ -761,7 +762,7 @@ function parseDraft( draft: Draft, passthrough?: Pick< ReMeSettings, - "authToken" | "pythonPath" | "remeWorkspaceDir" + "authToken" | "pythonPath" | "remeWorkspaceDir" | "autoStart" >, ): ReMeSettings | undefined { const requestTimeoutMs = boundedInteger(draft.requestTimeoutMs, 1000, 120000); @@ -807,6 +808,7 @@ function parseDraft( authToken: passthrough?.authToken ?? "", pythonPath: passthrough?.pythonPath ?? "", remeWorkspaceDir: passthrough?.remeWorkspaceDir ?? "", + autoStart: passthrough?.autoStart ?? false, requestTimeoutMs: requestTimeoutMs!, backgroundTimeoutMs: backgroundTimeoutMs!, shutdownTimeoutMs: shutdownTimeoutMs!, diff --git a/integrations/dsh/src/client/styles.ts b/integrations/dsh/src/client/styles.ts index 5a8c6d97..5abb7639 100644 --- a/integrations/dsh/src/client/styles.ts +++ b/integrations/dsh/src/client/styles.ts @@ -70,5 +70,6 @@ export const styles = ` .reme-env-inline input{flex:1;min-width:0} .reme-env-config{padding:4px 0 0} .reme-env-group{margin:14px 0 0;color:var(--dsw-alias-label-secondary);font-size:12px;font-weight:600;letter-spacing:.02em} +.reme-env-check-inline{padding:0;border:0} @media(max-width:760px){.reme-settings-health,.reme-settings-component-grid,.reme-status-component-grid,.reme-status-capability-grid{grid-template-columns:1fr}.reme-settings-result{display:none}.reme-status-overview-hero{grid-template-columns:1fr}.reme-status-overview-actions{grid-column:1;grid-row:auto;flex-direction:row}.reme-status-overview-actions .reme-status-button{flex:1}.reme-status-file-browser{grid-template-columns:1fr}.reme-status-file-browser>aside{max-height:210px;border-right:0;border-bottom:1px solid var(--dsw-alias-border-l2)}.reme-status-file-browser>article{min-height:300px}.reme-status-task-list summary{grid-template-columns:1fr}} `; diff --git a/integrations/dsh/src/config.ts b/integrations/dsh/src/config.ts index 884d287a..8cdeb819 100644 --- a/integrations/dsh/src/config.ts +++ b/integrations/dsh/src/config.ts @@ -36,6 +36,11 @@ export const Config = z.object({ .default("") .description("ReMe workspace directory (empty means ~/.reme)") .volatile(), + autoStart: z + .boolean() + .default(false) + .description("Start a stopped local ReMe service when DSH starts") + .volatile(), requestTimeoutMs: z.natural().min(1000).max(120000).default(10000).volatile(), backgroundTimeoutMs: z .natural() @@ -106,6 +111,7 @@ const DEFAULT_CONFIG: Readonly = Object.freeze({ authToken: "", pythonPath: "", remeWorkspaceDir: "", + autoStart: false, requestTimeoutMs: 10000, backgroundTimeoutMs: 3600000, shutdownTimeoutMs: 5000, @@ -170,6 +176,7 @@ export function resolveConfig( typeof config.remeWorkspaceDir === "string" ? config.remeWorkspaceDir.trim() : ""; + config.autoStart = config.autoStart === true; const home = env.DSH_HOME?.trim() || `${homedir()}/.dsh`; const queueDir = config.autoMemoryQueueDir || `${home}/reme-memory/outbox`; config.autoMemoryQueueDir = resolve( diff --git a/integrations/dsh/src/env-gateway.ts b/integrations/dsh/src/env-gateway.ts index d1263144..30fd0def 100644 --- a/integrations/dsh/src/env-gateway.ts +++ b/integrations/dsh/src/env-gateway.ts @@ -3,6 +3,11 @@ import { Remote, TypertRemoteService } from "@deepseek-ai/dsh-typert-protocol"; import type { InstallStatus } from "./pyenv/install.js"; import type { RemeCheck } from "./pyenv/reme.js"; +import type { + ServiceLog, + ServiceStartResult, + ServiceState, +} from "./pyenv/service.js"; import type { PythonDetection } from "./pyenv/types.js"; import type { EnvSummary, LlmTestResult } from "./pyenv/workspace.js"; import type { ReMeEnvManager } from "./env-manager.js"; @@ -104,6 +109,44 @@ export class ReMeEnvGateway extends TypertRemoteService { input?.pythonPath ?? "", ); } + + /** Phase of the local ReMe service: running, conflict or stopped. */ + @Remote("serviceStatus") + async serviceStatus( + input: { endpoint?: string } | undefined, + ): Promise { + return this.ctx.remeEnv.serviceStatus(input?.endpoint ?? ""); + } + + /** Spawn a detached `reme start`; poll serviceStatus until healthy. */ + @Remote("serviceStart") + async serviceStart( + input: + | { endpoint?: string; pythonPath?: string; workspaceDir?: string } + | undefined, + ): Promise { + return this.ctx.remeEnv.serviceStart( + input?.endpoint ?? "", + input?.pythonPath ?? "", + input?.workspaceDir ?? "", + ); + } + + /** Stop the plugin-managed service; foreign instances are untouched. */ + @Remote("serviceStop") + async serviceStop( + input: { endpoint?: string } | undefined, + ): Promise<{ stopped: boolean; pid?: number }> { + return this.ctx.remeEnv.serviceStop(input?.endpoint ?? ""); + } + + /** Tail of the workspace service log for start-failure diagnosis. */ + @Remote("serviceLog") + async serviceLog( + input: { workspaceDir?: string } | undefined, + ): Promise { + return this.ctx.remeEnv.serviceLog(input?.workspaceDir ?? ""); + } } export default ReMeEnvGateway; diff --git a/integrations/dsh/src/env-manager.ts b/integrations/dsh/src/env-manager.ts index a7bd27cb..6814c268 100644 --- a/integrations/dsh/src/env-manager.ts +++ b/integrations/dsh/src/env-manager.ts @@ -6,11 +6,23 @@ import { detectPythonEnvs } from "./pyenv/detect.js"; import { createInstallSpawner, InstallJobs, + tailLog, type InstallRequest, type InstallStatus, } from "./pyenv/install.js"; -import { checkReme, type RemeCheck } from "./pyenv/reme.js"; +import { checkReme, remeVersionAtLeast, type RemeCheck } from "./pyenv/reme.js"; import { createNodeRunner } from "./pyenv/runner.js"; +import { + createServiceIO, + readServiceState, + serviceStatus, + startService, + stopService, + type ServiceIO, + type ServiceLog, + type ServiceStartResult, + type ServiceState, +} from "./pyenv/service.js"; import type { PythonDetection } from "./pyenv/types.js"; import { envSummary, @@ -33,6 +45,8 @@ 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 serviceStatePath: string; constructor(env: NodeJS.ProcessEnv = process.env) { this.env = env; @@ -41,6 +55,7 @@ export class ReMeEnvManager { spawner: createInstallSpawner(env), logDir: join(home, "reme-memory"), }); + this.serviceStatePath = join(home, "reme-memory", "service.json"); } async detectPython( @@ -117,4 +132,100 @@ export class ReMeEnvManager { { workspaceDir: dir, pythonPath }, ); } + + /** Current service phase (running / conflict / stopped) and ownership. */ + async serviceStatus(endpoint: string): Promise { + const endpointValue = endpoint.trim() || "http://127.0.0.1:2333"; + const persisted = readServiceState(this.serviceStatePath); + return serviceStatus(this.serviceIO, { + endpoint: endpointValue, + statePath: this.serviceStatePath, + logPath: join( + persisted?.workspaceDir ?? resolveWorkspaceDir("", homedir()), + "service.log", + ), + }); + } + + /** Spawn `reme start` detached; the caller polls serviceStatus. */ + async serviceStart( + endpoint: string, + pythonPath: string, + workspaceDir: string, + ): Promise { + const endpointValue = endpoint.trim() || "http://127.0.0.1:2333"; + const dir = resolveWorkspaceDir(workspaceDir, homedir()); + const logPath = join(dir, "service.log"); + const now = new Date().toISOString(); + const current = await this.serviceStatus(endpointValue); + if (current.phase === "running") + return { + started: false, + alreadyRunning: true, + command: "", + logPath, + startedAt: now, + ...(current.version ? { version: current.version } : {}), + }; + const path = pythonPath.trim(); + if (!path) + return { + started: false, + alreadyRunning: false, + command: "", + logPath, + startedAt: now, + error: "Select a Python interpreter first", + }; + const check = await this.checkReme(path); + if (!check.installed) + return { + started: false, + alreadyRunning: false, + command: "", + logPath, + startedAt: now, + error: "reme-ai is not installed in the selected interpreter", + }; + 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, + }); + } + + /** 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/index.ts b/integrations/dsh/src/index.ts index a0eb7d82..079fb764 100644 --- a/integrations/dsh/src/index.ts +++ b/integrations/dsh/src/index.ts @@ -27,9 +27,26 @@ export function apply(ctx: Context, input: ReMeConfigInput = {}): void { return () => runtime.disposeAll(); }, "remeMemory.lifecycle()"); ctx.provide("remeMemory", runtime); - ctx.provide("remeEnv", new ReMeEnvManager()); + const envManager = new ReMeEnvManager(); + ctx.provide("remeEnv", envManager); void ctx.plugin(ReMeStatusGateway); void ctx.plugin(ReMeEnvGateway); + ctx.effect(() => { + const live = current(); + if (live.autoStart) + void envManager + .autoStart({ + endpoint: live.endpoint, + pythonPath: live.pythonPath, + workspaceDir: live.remeWorkspaceDir, + }) + .catch((error: unknown) => { + ctx.logger.warn("remeMemory.envAutoStart failed", { + error: String(error), + }); + }); + return () => {}; + }, "remeMemory.envAutoStart()"); ctx.effect( () => registerReMeTools(ctx, client, current), "remeMemory.tools()", diff --git a/integrations/dsh/src/pyenv/service.ts b/integrations/dsh/src/pyenv/service.ts new file mode 100644 index 00000000..08a7acc5 --- /dev/null +++ b/integrations/dsh/src/pyenv/service.ts @@ -0,0 +1,344 @@ +import { spawn } from "node:child_process"; +import { + closeSync, + existsSync, + mkdirSync, + openSync, + readFileSync, + rmSync, + writeFileSync, +} from "node:fs"; +import { createConnection } from "node:net"; +import { dirname } from "node:path"; + +export interface ServiceState { + phase: "running" | "stopped" | "conflict"; + /** True when the running service was started by this plugin. */ + managed: boolean; + endpoint: string; + pid?: number; + version?: string; + startedAt?: string; + logPath: string; +} + +export interface ServiceStartResult { + started: boolean; + alreadyRunning: boolean; + pid?: number; + version?: string; + command: string; + logPath: string; + startedAt: string; + error?: string; +} + +export interface ServiceLog { + logPath: string; + lines: string; +} + +export interface PersistedService { + pid?: number; + endpoint: string; + workspaceDir: string; + pythonPath: string; + command: string; + startedAt: string; +} + +export interface ServiceIO { + fetchHealth(endpoint: string): Promise<{ ok: boolean; version: string }>; + portOpen(host: string, port: number): Promise; + spawnService(options: { + exe: string; + args: string[]; + cwd: string; + logPath: string; + env: NodeJS.ProcessEnv; + }): { pid: number | undefined }; + killTree(pid: number): void; + probePid(pid: number): void; + exists(path: string): boolean; + platform: string; +} + +/** Only loopback endpoints may be managed as a local service. */ +export function parseLocalEndpoint(endpoint: string): { + host: string; + port: number; +} { + let url: URL; + try { + url = new URL(endpoint.trim()); + } 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"); + 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")); + if (!Number.isInteger(port) || port < 1 || port > 65535) + throw new TypeError("Invalid ReMe endpoint port"); + return { host: "127.0.0.1", port }; +} + +/** + * Compose the `reme start` invocation. The working directory must be the + * workspace so ReMe's `.env` discovery (cwd plus five parents) finds the file + * the Environment tab writes. + */ +export function buildStartCommand(options: { + cliPath: string; + pythonPath: string; + workspaceDir: string; + endpoint: string; + disableDreamCron: boolean; + exists(path: string): boolean; +}): { exe: string; args: string[]; display: string } { + const { port } = parseLocalEndpoint(options.endpoint); + const args = [ + "start", + `workspace_dir=${options.workspaceDir}`, + "service.host=127.0.0.1", + `service.port=${port}`, + // Keeps the rich logo from crashing GBK consoles in the background. + "enable_logo=false", + ...(options.disableDreamCron ? ["jobs.dream_cron.enabled=false"] : []), + ]; + if (options.cliPath && options.exists(options.cliPath)) + return { + exe: options.cliPath, + args, + display: `${options.cliPath} ${args.join(" ")}`, + }; + // The console script is missing (e.g. a bare venv without Scripts entry): + // run the same CLI entry through the interpreter itself. + const wrapper = `import sys;sys.argv=${JSON.stringify([ + "reme", + ...args, + ])};from reme.reme import main;sys.exit(main())`; + return { + exe: options.pythonPath, + args: ["-c", wrapper], + display: `${options.pythonPath} -c `, + }; +} + +export function readServiceState(path: string): PersistedService | undefined { + try { + return JSON.parse(readFileSync(path, "utf8")) as PersistedService; + } catch { + return undefined; + } +} + +export function writeServiceState(path: string, state: PersistedService): void { + mkdirSync(dirname(path), { recursive: true }); + writeFileSync(path, JSON.stringify(state, null, 2), "utf8"); +} + +function alive(io: ServiceIO, pid?: number): boolean { + if (pid === undefined) return false; + try { + io.probePid(pid); + return true; + } catch { + return false; + } +} + +/** Running / conflict / stopped, with ownership of the running process. */ +export async function serviceStatus( + io: ServiceIO, + options: { endpoint: string; statePath: string; logPath: string }, +): Promise { + const health = await io.fetchHealth(options.endpoint); + const persisted = readServiceState(options.statePath); + if (health.ok) + return { + phase: "running", + managed: alive(io, persisted?.pid), + endpoint: options.endpoint, + ...(persisted?.pid === undefined ? {} : { pid: persisted.pid }), + ...(health.version ? { version: health.version } : {}), + ...(persisted?.startedAt === undefined + ? {} + : { startedAt: persisted.startedAt }), + logPath: options.logPath, + }; + const { host, port } = parseLocalEndpoint(options.endpoint); + if (await io.portOpen(host, port)) + return { + phase: "conflict", + managed: false, + endpoint: options.endpoint, + logPath: options.logPath, + }; + return { + phase: "stopped", + managed: false, + endpoint: options.endpoint, + ...(persisted?.pid === undefined ? {} : { pid: persisted.pid }), + logPath: options.logPath, + }; +} + +/** Spawn a detached service; the caller polls status until it is healthy. */ +export function startService( + io: ServiceIO, + options: { + endpoint: string; + workspaceDir: string; + pythonPath: string; + cliPath: string; + disableDreamCron: boolean; + statePath: string; + logPath: string; + env: NodeJS.ProcessEnv; + }, +): ServiceStartResult { + 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 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(); + writeServiceState(options.statePath, { + pid: child.pid, + endpoint: options.endpoint, + workspaceDir: options.workspaceDir, + pythonPath: options.pythonPath, + command: command.display, + startedAt, + }); + return { + started: true, + alreadyRunning: false, + ...(child.pid === undefined ? {} : { pid: child.pid }), + command: command.display, + logPath: options.logPath, + startedAt, + }; +} + +/** Stop the managed service (tree kill) and clear its state file. */ +export function stopService( + io: ServiceIO, + options: { statePath: string }, +): { stopped: boolean; pid?: number } { + const persisted = readServiceState(options.statePath); + if (!persisted || persisted.pid === undefined) return { stopped: false }; + const pid = persisted.pid; + 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. + } + return { stopped: true, pid }; +} + +/** Node-backed IO: fetch health, probe the port, spawn detached, tree-kill. */ +export function createServiceIO(): ServiceIO { + return { + async fetchHealth(endpoint) { + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), 3000); + try { + const response = await fetch(`${endpoint}/health_check`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: "{}", + signal: controller.signal, + }); + const body = (await response.json().catch(() => ({}))) as { + success?: boolean; + metadata?: { health?: { version?: unknown } }; + }; + const version = body.metadata?.health?.version; + return { + ok: response.ok && body.success !== false, + version: typeof version === "string" ? version : "", + }; + } catch { + return { ok: false, version: "" }; + } finally { + clearTimeout(timer); + } + }, + portOpen(host, port) { + return new Promise((resolve) => { + const socket = createConnection({ host, port }); + const finish = (value: boolean): void => { + socket.destroy(); + resolve(value); + }; + socket.setTimeout(800); + socket.once("connect", () => finish(true)); + socket.once("timeout", () => finish(false)); + socket.once("error", () => finish(false)); + }); + }, + spawnService({ exe, args, cwd, logPath, env }) { + const fd = openSync(logPath, "a"); + let child: ReturnType; + try { + child = spawn(exe, args, { + shell: false, + windowsHide: true, + detached: true, + cwd, + env, + stdio: ["ignore", fd, fd], + }); + } finally { + closeSync(fd); + } + child.unref(); + return { pid: child.pid }; + }, + killTree(pid) { + if (process.platform === "win32") { + spawn("taskkill", ["/PID", String(pid), "/T", "/F"], { + windowsHide: true, + stdio: "ignore", + }); + return; + } + try { + process.kill(pid, "SIGTERM"); + } catch { + // Already exited. + } + }, + probePid(pid) { + process.kill(pid, 0); + }, + exists: existsSync, + platform: process.platform, + }; +} diff --git a/integrations/dsh/src/types.ts b/integrations/dsh/src/types.ts index 8fae0403..3984f3c7 100644 --- a/integrations/dsh/src/types.ts +++ b/integrations/dsh/src/types.ts @@ -5,6 +5,7 @@ export interface ReMeConfigInput { authToken?: string; pythonPath?: string; remeWorkspaceDir?: string; + autoStart?: boolean; requestTimeoutMs?: number; backgroundTimeoutMs?: number; shutdownTimeoutMs?: number; @@ -28,6 +29,7 @@ export interface ReMeConfig extends ReMeClientConfig { authToken: string; pythonPath: string; remeWorkspaceDir: string; + autoStart: boolean; autoMemoryEnabled: boolean; autoMemoryInterval: number; autoMemoryQueueDir: string; diff --git a/integrations/dsh/tests/env-service.test.mjs b/integrations/dsh/tests/env-service.test.mjs new file mode 100644 index 00000000..98318a25 --- /dev/null +++ b/integrations/dsh/tests/env-service.test.mjs @@ -0,0 +1,180 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { + buildStartCommand, + parseLocalEndpoint, + readServiceState, + serviceStatus, + startService, + stopService, + writeServiceState, +} from "../dist/pyenv/service.js"; + +function fakeIO(overrides = {}) { + const calls = { killTree: [], spawn: [] }; + const io = { + fetchHealth: async () => ({ ok: false, version: "" }), + portOpen: async () => false, + spawnService(options) { + calls.spawn.push(options); + return { pid: 4321 }; + }, + killTree(pid) { + calls.killTree.push(pid); + }, + probePid() {}, + exists: () => true, + platform: "win32", + ...overrides, + }; + return { io, calls }; +} + +test("accepts only loopback endpoints and validates the port", () => { + assert.deepEqual(parseLocalEndpoint("http://127.0.0.1:2334"), { + host: "127.0.0.1", + port: 2334, + }); + assert.deepEqual(parseLocalEndpoint("http://localhost:2333"), { + host: "127.0.0.1", + port: 2333, + }); + assert.throws(() => parseLocalEndpoint("http://10.0.0.5:2333")); + assert.throws(() => parseLocalEndpoint("not-a-url")); + assert.throws(() => parseLocalEndpoint("http://127.0.0.1:0")); +}); + +test("builds reme start with the workspace cwd contract", () => { + const command = buildStartCommand({ + cliPath: "D:\\env\\Scripts\\reme.exe", + pythonPath: "D:\\env\\python.exe", + workspaceDir: "C:\\Users\\u\\.reme", + endpoint: "http://127.0.0.1:2334", + disableDreamCron: false, + exists: () => true, + }); + assert.equal(command.exe, "D:\\env\\Scripts\\reme.exe"); + assert.ok(command.args.includes("start")); + assert.ok(command.args.includes("workspace_dir=C:\\Users\\u\\.reme")); + assert.ok(command.args.includes("service.host=127.0.0.1")); + assert.ok(command.args.includes("service.port=2334")); + assert.ok(command.args.includes("enable_logo=false")); + assert.ok(!command.args.includes("jobs.dream_cron.enabled=false")); + + const newer = buildStartCommand({ + cliPath: "D:\\env\\Scripts\\reme.exe", + pythonPath: "D:\\env\\python.exe", + workspaceDir: "C:\\Users\\u\\.reme", + endpoint: "http://127.0.0.1:2334", + disableDreamCron: true, + exists: () => true, + }); + assert.ok(newer.args.includes("jobs.dream_cron.enabled=false")); +}); + +test("falls back to the interpreter wrapper when the CLI is missing", () => { + const command = buildStartCommand({ + cliPath: "", + pythonPath: "D:\\env\\python.exe", + workspaceDir: "C:\\Users\\u\\.reme", + endpoint: "http://127.0.0.1:2334", + disableDreamCron: false, + exists: () => false, + }); + assert.equal(command.exe, "D:\\env\\python.exe"); + assert.equal(command.args[0], "-c"); + assert.match(String(command.args[1]), /sys\.argv/); + assert.match(String(command.args[1]), /from reme\.reme import main/); + assert.match(String(command.args[1]), /service\.port=2334/); +}); + +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 base = { + endpoint: "http://127.0.0.1:2334", + statePath, + logPath: join(directory, "service.log"), + }; + + const runningIo = fakeIO({ + fetchHealth: async () => ({ ok: true, version: "0.4.0.9" }), + }); + 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", + }); + status = await serviceStatus(runningIo.io, base); + assert.equal(status.phase, "running"); + assert.equal(status.managed, true); + assert.equal(status.pid, 777); + + const conflict = await serviceStatus( + fakeIO({ portOpen: async () => true }).io, + base, + ); + assert.equal(conflict.phase, "conflict"); + + const stopped = await serviceStatus(fakeIO().io, base); + assert.equal(stopped.phase, "stopped"); +}); + +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 { io, calls } = fakeIO(); + const result = startService(io, { + endpoint: "http://127.0.0.1:2334", + workspaceDir: directory, + pythonPath: "D:\\env\\python.exe", + cliPath: "D:\\env\\Scripts\\reme.exe", + disableDreamCron: true, + statePath, + logPath: join(directory, "service.log"), + env: { PATH: "x" }, + }); + assert.equal(result.started, true); + assert.equal(result.pid, 4321); + const spawned = calls.spawn[0]; + assert.equal(spawned.cwd, directory); + assert.equal(spawned.env.PYTHONUTF8, "1"); + assert.equal(spawned.env.NO_PROXY, "localhost,127.0.0.1,::1"); + const persisted = readServiceState(statePath); + assert.equal(persisted?.pid, 4321); + 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", + workspaceDir: directory, + pythonPath: "D:\\env\\python.exe", + command: "reme start", + startedAt: "2026-10-27T00:00:00.000Z", + }); + const { io, calls } = fakeIO(); + const stopped = stopService(io, { statePath }); + assert.deepEqual(stopped, { stopped: true, pid: 999 }); + assert.deepEqual(calls.killTree, [999]); + assert.equal(readServiceState(statePath), undefined); + assert.deepEqual(stopService(io, { statePath }), { stopped: false }); +});