EvanAgent/dsh-desktop/app/service.js

"use strict";
/**
 * Supervisor for a local `dsh web` service.
 *
 * - attach: if a healthy dsh is already listening on the default port (its
 *   HTML shell carries `__DSH_BOOT__`), use it and own nothing.
 * - spawn: otherwise start `dsh web --port <n>` as a supervised child; a busy
 *   default port falls back to `--port 0` (the actual port is parsed from the
 *   child's "dsh web: http://127.0.0.1:<port>" line, with an lsof fallback).
 * - crash: restart with exponential backoff (suppressed while stopping or
 *   upgrading, and never for an instance we did not spawn).
 * - stop: SIGTERM first — dsh shuts down gracefully — then SIGKILL on timeout.
 */
const { spawn, execFile, execFileSync } = require("node:child_process");
const http = require("node:http");
const fs = require("node:fs");
const path = require("node:path");
const readline = require("node:readline");
const { DSH_BIN, resolveDsh, childEnv } = require("./env");

const DEFAULT_PORT = Number(process.env.DSH_DESKTOP_PORT || 3080);
const READY_TIMEOUT_MS = 120_000;
const STOP_GRACE_MS = 8_000;
const BACKOFF_MS = [1_000, 5_000, 30_000, 60_000];

/** Small GET with timeout; never rejects. */
function httpGet(url, timeoutMs = 2000) {
	return new Promise((resolve) => {
		let settled = false;
		const finish = (value) => { if (!settled) { settled = true; resolve(value); } };
		const req = http.get(url, (res) => {
			let body = "";
			res.on("data", (chunk) => {
				body += chunk;
				if (body.length >= 16384) { req.destroy(); finish({ ok: true, status: res.statusCode, body }); }
			});
			res.on("end", () => finish({ ok: true, status: res.statusCode, body }));
			res.on("error", () => finish({ ok: true, status: res.statusCode, body }));
		});
		req.on("error", () => finish({ ok: false }));
		req.setTimeout(timeoutMs, () => { req.destroy(); finish({ ok: false }); });
	});
}

/** A healthy dsh web UI answers 2xx and its HTML shell carries the boot marker. */
async function probeDsh(port) {
	const res = await httpGet(`http://127.0.0.1:${port}/`);
	return Boolean(res.ok && res.status && res.status < 500 && res.body.includes("__DSH_BOOT__"));
}

class Service {
	constructor({ userDataDir, onState, onLog }) {
		this.userDataDir = userDataDir;
		this.onState = onState || (() => {});
		this.onLog = onLog || (() => {});
		this.child = null;
		this.port = null;
		this.spawned = false; // did we start this dsh?
		this.state = "stopped"; // stopped | starting | running | stopping | crashed
		this.detail = "";
		this.upgrading = false; // suppress crash-restart while the updater works
		this.stopping = false;
		this.backoffIndex = 0;
		this.logLines = [];
		this.logFile = path.join(userDataDir, "dsh-web.log");
		this._ver = null;
		this._verAt = 0;
	}

	get url() { return this.port ? `http://127.0.0.1:${this.port}` : null; }
	get running() { return this.state === "running"; }

	setState(state, detail = "") {
		this.state = state;
		this.detail = detail;
		this.onState(this);
	}

	appendLog(line) {
		this.logLines.push(line);
		if (this.logLines.length > 500) this.logLines.shift();
		this.onLog(line);
		try { fs.appendFileSync(this.logFile, line + "\n"); } catch { /* disk issues must not kill the shell */ }
	}

	ensureLogFile() {
		try { fs.closeSync(fs.openSync(this.logFile, "a")); } catch { /* ignore */ }
		return this.logFile;
	}

	/** Current @deepseek-ai/dsh version, read from its package.json (cached). */
	version() {
		const now = Date.now();
		if (this._ver && now - this._verAt < 10_000) return this._ver;
		let version = null;
		try {
			let dir = path.dirname(fs.realpathSync(DSH_BIN));
			for (let i = 0; i < 8 && dir !== path.dirname(dir); i++) {
				const manifest = path.join(dir, "package.json");
				if (fs.existsSync(manifest)) {
					const pkg = JSON.parse(fs.readFileSync(manifest, "utf8"));
					if (pkg.name === "@deepseek-ai/dsh") { version = pkg.version; break; }
				}
				dir = path.dirname(dir);
			}
		} catch { /* fall through to the CLI */ }
		if (!version) {
			try { version = execFileSync(DSH_BIN, ["-V"], { encoding: "utf8" }).trim(); } catch { /* unknown */ }
		}
		this._ver = version;
		this._verAt = now;
		return version;
	}

	/** Attach to a healthy instance, else spawn (fixed port first, then a free one).
	 * ENOENT retries cover the `npm install -g` window: while npm rewrites the
	 * bin symlink the spawn target briefly does not exist. */
	async ensure() {
		if (process.env.DSH_DESKTOP_NO_ATTACH !== "1" && await probeDsh(DEFAULT_PORT)) {
			this.port = DEFAULT_PORT;
			this.spawned = false;
			this.setState("running", "attached");
			this.appendLog(`[desktop] attached to an existing dsh at ${this.url}`);
			return true;
		}
		for (let attempt = 0; attempt < 6; attempt++) {
			const wanted = attempt === 0 ? DEFAULT_PORT : 0;
			if (await this.start(wanted)) return true;
			// ENOENT (bin link mid-rewrite) is worth waiting out; other failures
			// stop after the usual port fallback.
			if (!this.lastSpawnEnoent && attempt >= 1) return false;
			this.lastSpawnEnoent = false;
			this.appendLog(`[desktop] spawn target missing; retry ${attempt + 1}/5 in 3s`);
			await new Promise((r) => setTimeout(r, 3000));
		}
		return false;
	}

	start(wanted = DEFAULT_PORT) {
		if (this.child || this.running) return Promise.resolve(true);
		this.stopping = false;
		this.setState("starting", wanted === 0 ? "spawn · free port" : `spawn · :${wanted}`);
		return new Promise((resolve) => {
			let settled = false;
			let portHint = null;
			const done = (ok) => { if (!settled) { settled = true; resolve(ok); } };
			const bin = resolveDsh(); // re-resolved per spawn: npm upgrades replace the link
			let child;
			try {
				child = spawn(bin, ["web", "--port", String(wanted)], {
					stdio: ["ignore", "pipe", "pipe"],
					env: childEnv(),
				});
			} catch (err) {
				this.lastSpawnEnoent = err.code === "ENOENT";
				this.setState("crashed", err.message);
				done(false);
				return;
			}
			this.child = child;
			this.spawned = true;
			this.appendLog(`[desktop] spawn: ${bin} web --port ${wanted} (pid ${child.pid})`);
			for (const stream of [child.stdout, child.stderr]) {
				readline.createInterface({ input: stream }).on("line", (line) => {
					if (line) this.appendLog(line);
					const m = line.match(/http:\/\/127\.0\.0\.1:(\d+)/);
					if (m) portHint = Number(m[1]);
				});
			}
			child.on("error", (err) => {
				this.appendLog(`[desktop] spawn error: ${err.message}`);
				this.child = null;
				this.lastSpawnEnoent = err.code === "ENOENT";
				if (!settled) { this.setState("crashed", err.message); done(false); }
			});
			child.on("exit", (code, signal) => {
				this.child = null;
				this.appendLog(`[desktop] dsh exited (code=${code} signal=${signal})`);
				if (settled) { this.handleCrash(); return; }
				this.setState("crashed", `启动即退出（code=${code}）`);
				done(false);
			});
			const startedAt = Date.now();
			let lastLsof = 0;
			const timer = setInterval(async () => {
				if (this.child !== child) { clearInterval(timer); return; }
				// Readiness requires the child's OWN port — its "dsh web: http://…"
				// line above, or lsof on its pid. Never probe `wanted` directly:
				// a foreign dsh already on that port would fake readiness.
				if (!portHint && child.pid && Date.now() - lastLsof > 2000) {
					lastLsof = Date.now();
					portHint = await this.listenPortOf(child.pid);
				}
				if (portHint && (await probeDsh(portHint))) {
					clearInterval(timer);
					this.port = portHint;
					this.setState("running", `pid ${child.pid}`);
					this.appendLog(`[desktop] service ready at ${this.url}`);
					this.armBackoffReset();
					done(true);
					return;
				}
				if (Date.now() - startedAt > READY_TIMEOUT_MS) {
					clearInterval(timer);
					this.appendLog("[desktop] ready timeout; killing child");
					try { child.kill("SIGKILL"); } catch { /* already gone */ }
					this.setState("crashed", "就绪超时");
					done(false);
				}
			}, 500);
		});
	}

	/** The TCP port a pid listens on, via lsof (macOS). */
	listenPortOf(pid) {
		return new Promise((resolve) => {
			execFile("lsof", ["-nP", "-a", "-iTCP", "-sTCP:LISTEN", "-p", String(pid)], (err, stdout) => {
				if (err || !stdout) return resolve(null);
				const m = stdout.match(/127\.0\.0\.1:(\d+)/);
				resolve(m ? Number(m[1]) : null);
			});
		});
	}

	handleCrash() {
		if (this.stopping || this.upgrading || !this.spawned) { this.setState("stopped"); return; }
		const delay = BACKOFF_MS[Math.min(this.backoffIndex, BACKOFF_MS.length - 1)];
		this.backoffIndex += 1;
		this.setState("crashed", `${Math.round(delay / 1000)}s 后自动重启`);
		this.appendLog(`[desktop] crashed; restart in ${delay}ms`);
		setTimeout(() => {
			if (!this.stopping && !this.upgrading && !this.child) this.start(0);
		}, delay);
	}

	armBackoffReset() {
		setTimeout(() => { if (this.running) this.backoffIndex = 0; }, 60_000);
	}

	stop() {
		this.stopping = true;
		if (!this.child) { this.setState("stopped"); return Promise.resolve(true); }
		const child = this.child;
		this.setState("stopping");
		this.appendLog("[desktop] stopping service (SIGTERM)");
		return new Promise((resolve) => {
			const killTimer = setTimeout(() => {
				this.appendLog("[desktop] grace elapsed; SIGKILL");
				try { child.kill("SIGKILL"); } catch { /* already gone */ }
			}, STOP_GRACE_MS);
			child.once("exit", () => { clearTimeout(killTimer); resolve(true); });
			try { child.kill("SIGTERM"); } catch { clearTimeout(killTimer); resolve(false); }
		}).then(() => {
			this.child = null;
			this.port = null;
			this.spawned = false;
			this.setState("stopped");
			return true;
		});
	}

	/** Restart: our child stops gracefully; an attached foreign instance is taken over. */
	async restart() {
		if (this.child) {
			await this.stop();
			return this.start(DEFAULT_PORT);
		}
		if (this.port) {
			this.appendLog(`[desktop] taking over the attached service on :${this.port}`);
			await this.killByPort(this.port);
			this.port = null;
			this.spawned = false;
			return this.start(DEFAULT_PORT);
		}
		return this.start(DEFAULT_PORT);
	}

	/** SIGTERM whatever listens on a port (used only for explicit take-over). */
	killByPort(port) {
		return new Promise((resolve) => {
			execFile("lsof", ["-ti", `tcp:${port}`], (err, stdout) => {
				const pids = (stdout || "").trim().split(/\s+/).filter(Boolean).map(Number);
				if (err || pids.length === 0) return resolve(false);
				this.appendLog(`[desktop] SIGTERM pid ${pids.join(", ")} (port ${port})`);
				for (const pid of pids) { try { process.kill(pid, "SIGTERM"); } catch { /* gone */ } }
				const startedAt = Date.now();
				const iv = setInterval(async () => {
					const alive = await probeDsh(port);
					if (!alive || Date.now() - startedAt > 10_000) {
						clearInterval(iv);
						if (alive) for (const pid of pids) { try { process.kill(pid, "SIGKILL"); } catch { /* gone */ } }
						resolve(true);
					}
				}, 400);
			});
		});
	}
}

module.exports = { Service, probeDsh, DEFAULT_PORT, DSH_BIN };