diff --git a/src/mcp/client.ts b/src/mcp/client.ts index 69b113a5..86fe88fb 100644 --- a/src/mcp/client.ts +++ b/src/mcp/client.ts @@ -15,14 +15,8 @@ import { resolveHunkMcpConfig, type ResolvedHunkMcpConfig, } from "./config"; -import { - isHunkDaemonHealthy, - isLoopbackPortReachable, - launchHunkDaemon, - waitForHunkDaemonHealth, -} from "./daemonLauncher"; +import { ensureHunkDaemonAvailable } from "./daemonLauncher"; -const DAEMON_LAUNCH_COOLDOWN_MS = 5_000; const DAEMON_STARTUP_TIMEOUT_MS = 3_000; const RECONNECT_DELAY_MS = 3_000; const HEARTBEAT_INTERVAL_MS = 10_000; @@ -54,7 +48,6 @@ export class HunkHostClient { private heartbeatTimer: Timer | null = null; private stopped = false; private startupPromise: Promise | null = null; - private lastDaemonLaunchStartedAt = 0; private lastConnectionWarning: string | null = null; constructor( @@ -122,39 +115,11 @@ export class HunkHostClient { } private async ensureDaemonAvailable(config: ResolvedHunkMcpConfig) { - if (await isHunkDaemonHealthy(config)) { - this.lastConnectionWarning = null; - return; - } - - const shouldLaunch = Date.now() - this.lastDaemonLaunchStartedAt >= DAEMON_LAUNCH_COOLDOWN_MS; - if (shouldLaunch) { - this.lastDaemonLaunchStartedAt = Date.now(); - launchHunkDaemon(); - } - - const ready = await waitForHunkDaemonHealth({ + await ensureHunkDaemonAvailable({ config, - timeoutMs: shouldLaunch ? DAEMON_STARTUP_TIMEOUT_MS : 1_500, + timeoutMs: DAEMON_STARTUP_TIMEOUT_MS, }); - - if (ready) { - this.lastConnectionWarning = null; - return; - } - - const portReachable = await isLoopbackPortReachable(config); - if (portReachable) { - throw new Error( - `Hunk MCP port ${config.host}:${config.port} is already in use by another process. ` + - `Stop the conflicting process or set HUNK_MCP_PORT to a different loopback port.`, - ); - } - - throw new Error( - `Timed out waiting for the Hunk MCP daemon on ${config.host}:${config.port}. ` + - `Hunk will retry in the background.`, - ); + this.lastConnectionWarning = null; } setBridge(bridge: HunkAppBridge | null) { @@ -180,7 +145,6 @@ export class HunkHostClient { this.websocket = websocket; websocket.onopen = () => { - this.lastDaemonLaunchStartedAt = 0; this.lastConnectionWarning = null; this.startHeartbeat(); this.send({ diff --git a/src/mcp/daemonLauncher.ts b/src/mcp/daemonLauncher.ts index 7b37a59f..714016ad 100644 --- a/src/mcp/daemonLauncher.ts +++ b/src/mcp/daemonLauncher.ts @@ -1,18 +1,243 @@ import { spawn } from "node:child_process"; import type { ChildProcess } from "node:child_process"; +import { existsSync, mkdirSync, readFileSync, rmSync, statSync, writeFileSync } from "node:fs"; import { connect } from "node:net"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; import { resolveHunkMcpConfig, type ResolvedHunkMcpConfig } from "./config"; const SCRIPT_ENTRYPOINT_PATTERN = /[\\/]|\.(?:[cm]?js|tsx?)$/; +const DEFAULT_DAEMON_LOCK_STALE_MS = 15_000; +const DEFAULT_DAEMON_STARTUP_TIMEOUT_MS = 3_000; +const DEFAULT_DAEMON_HEALTH_POLL_INTERVAL_MS = 100; export interface DaemonLaunchCommand { command: string; args: string[]; } +export interface HunkDaemonRuntimePaths { + runtimeDir: string; + lockPath: string; + metadataPath: string; +} + +interface HunkDaemonLaunchLockFile { + ownerPid: number; + host: string; + port: number; + acquiredAt: string; +} + +interface HunkDaemonLaunchMetadata { + pid: number; + host: string; + port: number; + command: string; + args: string[]; + launchedAt: string; + launchedByPid: number; + launchCwd: string; +} + +interface HunkDaemonLaunchLock { + release: () => void; +} + +export interface EnsureHunkDaemonAvailableOptions { + config?: ResolvedHunkMcpConfig; + cwd?: string; + env?: NodeJS.ProcessEnv; + argv?: string[]; + execPath?: string; + timeoutMs?: number; + intervalMs?: number; + lockStaleMs?: number; + timeoutMessage?: string; + isHealthy?: (config: ResolvedHunkMcpConfig) => Promise; + isPortReachable?: ( + config: Pick, + timeoutMs?: number, + ) => Promise; + launchDaemon?: (options?: { + cwd?: string; + env?: NodeJS.ProcessEnv; + argv?: string[]; + execPath?: string; + }) => ChildProcess; +} + /** Detect Bun's virtual filesystem prefix used inside compiled single-file executables. */ const BUNFS_PREFIX = "/$bunfs/"; +function safeRuntimeToken(value: string) { + return value.replace(/[^a-z0-9]+/gi, "-").replace(/^-+|-+$/g, "") || "default"; +} + +function resolveRuntimeBaseDir(env: NodeJS.ProcessEnv = process.env) { + return env.XDG_RUNTIME_DIR?.trim() || tmpdir(); +} + +function isRunningPid(pid: number) { + if (!Number.isInteger(pid) || pid <= 0) { + return false; + } + + try { + process.kill(pid, 0); + return true; + } catch (error) { + return (error as NodeJS.ErrnoException).code === "EPERM"; + } +} + +function readJsonFile(path: string) { + try { + return JSON.parse(readFileSync(path, "utf8")) as T; + } catch { + return null; + } +} + +function removeFileIfPresent(path: string) { + try { + rmSync(path, { force: true }); + } catch { + // Ignore best-effort cleanup failures. + } +} + +function cleanStaleDaemonMetadata(paths: HunkDaemonRuntimePaths) { + const metadata = readJsonFile(paths.metadataPath); + if (!metadata) { + return; + } + + if (!isRunningPid(metadata.pid)) { + removeFileIfPresent(paths.metadataPath); + } +} + +function tryAcquireDaemonLaunchLock({ + config, + env, + staleAfterMs, +}: { + config: ResolvedHunkMcpConfig; + env: NodeJS.ProcessEnv; + staleAfterMs: number; +}): HunkDaemonLaunchLock | null { + const paths = resolveHunkDaemonRuntimePaths(config, env); + mkdirSync(paths.runtimeDir, { recursive: true }); + + const payload: HunkDaemonLaunchLockFile = { + ownerPid: process.pid, + host: config.host, + port: config.port, + acquiredAt: new Date().toISOString(), + }; + + try { + writeFileSync(paths.lockPath, JSON.stringify(payload, null, 2), { + encoding: "utf8", + flag: "wx", + mode: 0o600, + }); + + return { + release: () => { + const current = readJsonFile(paths.lockPath); + if (current?.ownerPid === payload.ownerPid) { + removeFileIfPresent(paths.lockPath); + } + }, + }; + } catch (error) { + const code = (error as NodeJS.ErrnoException).code; + if (code !== "EEXIST") { + throw error; + } + } + + const existing = readJsonFile(paths.lockPath); + if (!existing) { + if (existsSync(paths.lockPath)) { + try { + const stat = statSync(paths.lockPath); + if (Date.now() - stat.mtimeMs > staleAfterMs) { + removeFileIfPresent(paths.lockPath); + return tryAcquireDaemonLaunchLock({ config, env, staleAfterMs }); + } + } catch { + // Ignore racing readers while another process still owns the lock. + } + } + + return null; + } + + const ownerAlive = isRunningPid(existing.ownerPid); + + if (!ownerAlive) { + removeFileIfPresent(paths.lockPath); + return tryAcquireDaemonLaunchLock({ config, env, staleAfterMs }); + } + + return null; +} + +function writeDaemonLaunchMetadata( + paths: HunkDaemonRuntimePaths, + metadata: HunkDaemonLaunchMetadata, +) { + writeFileSync(paths.metadataPath, JSON.stringify(metadata, null, 2), { + encoding: "utf8", + mode: 0o600, + }); +} + +function daemonPortConflictError(config: Pick) { + return new Error( + `Hunk MCP port ${config.host}:${config.port} is already in use by another process. ` + + `Stop the conflicting process or set HUNK_MCP_PORT to a different loopback port.`, + ); +} + +function daemonStartupTimeoutError( + config: Pick, + timeoutMessage?: string, +) { + return new Error( + timeoutMessage ?? + `Timed out waiting for the Hunk MCP daemon on ${config.host}:${config.port}. ` + + `Hunk will retry in the background.`, + ); +} + +async function waitForDaemonHealthWithCheck({ + config, + timeoutMs, + intervalMs, + isHealthy, +}: { + config: ResolvedHunkMcpConfig; + timeoutMs: number; + intervalMs: number; + isHealthy: (config: ResolvedHunkMcpConfig) => Promise; +}) { + const deadline = Date.now() + timeoutMs; + + while (Date.now() < deadline) { + if (await isHealthy(config)) { + return true; + } + + await Bun.sleep(intervalMs); + } + + return false; +} + /** Resolve how the current Hunk process should launch a sibling `hunk mcp serve` daemon. */ export function resolveDaemonLaunchCommand( argv = process.argv, @@ -46,6 +271,21 @@ export function resolveDaemonLaunchCommand( }; } +/** Resolve the runtime paths Hunk uses to coordinate one daemon per loopback host/port. */ +export function resolveHunkDaemonRuntimePaths( + config: Pick = resolveHunkMcpConfig(), + env: NodeJS.ProcessEnv = process.env, +): HunkDaemonRuntimePaths { + const runtimeDir = join(resolveRuntimeBaseDir(env), "hunk-mcp"); + const fileStem = `${safeRuntimeToken(config.host)}-${config.port}`; + + return { + runtimeDir, + lockPath: join(runtimeDir, `daemon-${fileStem}.lock`), + metadataPath: join(runtimeDir, `daemon-${fileStem}.json`), + }; +} + /** Check whether the loopback Hunk daemon already answers health probes. */ export async function isHunkDaemonHealthy( config: ResolvedHunkMcpConfig = resolveHunkMcpConfig(), @@ -101,24 +341,19 @@ export function isLoopbackPortReachable( /** Wait briefly for a just-launched daemon to become reachable on its health endpoint. */ export async function waitForHunkDaemonHealth({ config = resolveHunkMcpConfig(), - timeoutMs = 3_000, - intervalMs = 100, + timeoutMs = DEFAULT_DAEMON_STARTUP_TIMEOUT_MS, + intervalMs = DEFAULT_DAEMON_HEALTH_POLL_INTERVAL_MS, }: { config?: ResolvedHunkMcpConfig; timeoutMs?: number; intervalMs?: number; }) { - const deadline = Date.now() + timeoutMs; - - while (Date.now() < deadline) { - if (await isHunkDaemonHealthy(config)) { - return true; - } - - await Bun.sleep(intervalMs); - } - - return false; + return waitForDaemonHealthWithCheck({ + config, + timeoutMs, + intervalMs, + isHealthy: (resolvedConfig) => isHunkDaemonHealthy(resolvedConfig), + }); } /** Launch the Hunk daemon in the background without tying it to the current TTY session. */ @@ -144,3 +379,93 @@ export function launchHunkDaemon({ child.unref(); return child; } + +/** Ensure one healthy local Hunk daemon exists, coordinating launch attempts across processes. */ +export async function ensureHunkDaemonAvailable({ + config = resolveHunkMcpConfig(), + cwd = process.cwd(), + env = process.env, + argv = process.argv, + execPath = process.execPath, + timeoutMs = DEFAULT_DAEMON_STARTUP_TIMEOUT_MS, + intervalMs = DEFAULT_DAEMON_HEALTH_POLL_INTERVAL_MS, + lockStaleMs = DEFAULT_DAEMON_LOCK_STALE_MS, + timeoutMessage, + isHealthy = (resolvedConfig) => isHunkDaemonHealthy(resolvedConfig), + isPortReachable = isLoopbackPortReachable, + launchDaemon = launchHunkDaemon, +}: EnsureHunkDaemonAvailableOptions = {}) { + const paths = resolveHunkDaemonRuntimePaths(config, env); + cleanStaleDaemonMetadata(paths); + + if (await isHealthy(config)) { + return; + } + + const deadline = Date.now() + timeoutMs; + + while (Date.now() < deadline) { + const lock = tryAcquireDaemonLaunchLock({ + config, + env, + staleAfterMs: lockStaleMs, + }); + + if (lock) { + try { + cleanStaleDaemonMetadata(paths); + if (await isHealthy(config)) { + return; + } + + const launchCommand = resolveDaemonLaunchCommand(argv, execPath); + const child = launchDaemon({ cwd, env, argv, execPath }); + writeDaemonLaunchMetadata(paths, { + pid: child.pid ?? 0, + host: config.host, + port: config.port, + command: launchCommand.command, + args: launchCommand.args, + launchedAt: new Date().toISOString(), + launchedByPid: process.pid, + launchCwd: cwd, + }); + + const ready = await waitForDaemonHealthWithCheck({ + config, + timeoutMs, + intervalMs, + isHealthy, + }); + if (ready) { + return; + } + } finally { + lock.release(); + } + } + + const remainingMs = deadline - Date.now(); + if (remainingMs <= 0) { + break; + } + + const ready = await waitForDaemonHealthWithCheck({ + config, + timeoutMs: Math.min(remainingMs, intervalMs), + intervalMs, + isHealthy, + }); + if (ready) { + return; + } + + cleanStaleDaemonMetadata(paths); + } + + if (await isPortReachable(config)) { + throw daemonPortConflictError(config); + } + + throw daemonStartupTimeoutError(config, timeoutMessage); +} diff --git a/src/session/commands.ts b/src/session/commands.ts index a90b0143..6269796b 100644 --- a/src/session/commands.ts +++ b/src/session/commands.ts @@ -11,10 +11,9 @@ import type { SessionSelectorInput, } from "../core/types"; import { + ensureHunkDaemonAvailable, isHunkDaemonHealthy, isLoopbackPortReachable, - launchHunkDaemon, - waitForHunkDaemonHealth, } from "../mcp/daemonLauncher"; import { resolveHunkMcpConfig } from "../mcp/config"; import type { @@ -308,13 +307,12 @@ async function restartDaemonForMissingAction( ); } - launchHunkDaemon(); - const config = resolveHunkMcpConfig(); - const ready = await waitForHunkDaemonHealth({ config, timeoutMs: 3_000 }); - if (!ready) { - throw new Error("Timed out waiting for the refreshed Hunk session daemon to start."); - } + await ensureHunkDaemonAvailable({ + config, + timeoutMs: 3_000, + timeoutMessage: "Timed out waiting for the refreshed Hunk session daemon to start.", + }); if (selector || hadSessions) { const registered = await waitForSessionRegistration(selector); diff --git a/test/daemon-launcher.test.ts b/test/daemon-launcher.test.ts index b6679fad..ab304c06 100644 --- a/test/daemon-launcher.test.ts +++ b/test/daemon-launcher.test.ts @@ -1,6 +1,38 @@ -import { describe, expect, test } from "bun:test"; +import { afterEach, describe, expect, test } from "bun:test"; +import type { ChildProcess } from "node:child_process"; +import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import { createServer } from "node:net"; -import { isLoopbackPortReachable, resolveDaemonLaunchCommand } from "../src/mcp/daemonLauncher"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + ensureHunkDaemonAvailable, + isLoopbackPortReachable, + resolveDaemonLaunchCommand, + resolveHunkDaemonRuntimePaths, +} from "../src/mcp/daemonLauncher"; + +const tempDirs: string[] = []; +const testConfig = { + host: "127.0.0.1", + port: 47657, + httpOrigin: "http://127.0.0.1:47657", + wsOrigin: "ws://127.0.0.1:47657", +}; + +function createRuntimeDir() { + const dir = mkdtempSync(join(tmpdir(), "hunk-mcp-launcher-test-")); + tempDirs.push(dir); + return dir; +} + +afterEach(() => { + while (tempDirs.length > 0) { + const dir = tempDirs.pop(); + if (dir) { + rmSync(dir, { recursive: true, force: true }); + } + } +}); describe("MCP daemon launcher", () => { test("reuses the current script entrypoint when Hunk is running from source or a JS wrapper", () => { @@ -58,4 +90,112 @@ describe("MCP daemon launcher", () => { await expect(isLoopbackPortReachable({ host: "127.0.0.1", port })).resolves.toBe(false); }); + + test("coordinates concurrent ensure calls so only one launcher runs", async () => { + const runtimeDir = createRuntimeDir(); + const env = { ...process.env, XDG_RUNTIME_DIR: runtimeDir }; + let healthy = false; + let launchCount = 0; + + const ensureCalls = Array.from({ length: 6 }, () => + ensureHunkDaemonAvailable({ + config: testConfig, + env, + cwd: "/repo", + argv: ["bun", "src/main.tsx", "diff"], + execPath: "/usr/bin/bun", + timeoutMs: 300, + intervalMs: 10, + isHealthy: async () => healthy, + isPortReachable: async () => false, + launchDaemon: () => { + launchCount += 1; + const timer = setTimeout(() => { + healthy = true; + }, 25); + timer.unref?.(); + return { pid: process.pid } as ChildProcess; + }, + }), + ); + + await expect(Promise.all(ensureCalls)).resolves.toHaveLength(6); + expect(launchCount).toBe(1); + + const paths = resolveHunkDaemonRuntimePaths(testConfig, env); + expect(existsSync(paths.lockPath)).toBe(false); + expect(JSON.parse(readFileSync(paths.metadataPath, "utf8"))).toMatchObject({ + pid: process.pid, + host: "127.0.0.1", + port: 47657, + command: "/usr/bin/bun", + args: ["src/main.tsx", "mcp", "serve"], + }); + }); + + test("recovers a stale launch lock from a dead launcher and overwrites stale metadata", async () => { + const runtimeDir = createRuntimeDir(); + const env = { ...process.env, XDG_RUNTIME_DIR: runtimeDir }; + const paths = resolveHunkDaemonRuntimePaths(testConfig, env); + mkdirSync(paths.runtimeDir, { recursive: true }); + + writeFileSync( + paths.lockPath, + JSON.stringify( + { + ownerPid: 999999, + host: testConfig.host, + port: testConfig.port, + acquiredAt: new Date().toISOString(), + }, + null, + 2, + ), + ); + writeFileSync( + paths.metadataPath, + JSON.stringify( + { + pid: 999999, + host: testConfig.host, + port: testConfig.port, + command: "/usr/bin/bun", + args: ["src/main.tsx", "mcp", "serve"], + launchedAt: new Date(0).toISOString(), + launchedByPid: 999999, + launchCwd: "/stale", + }, + null, + 2, + ), + ); + + let healthy = false; + let launchCount = 0; + + await ensureHunkDaemonAvailable({ + config: testConfig, + env, + cwd: "/repo", + argv: ["bun", "src/main.tsx", "diff"], + execPath: "/usr/bin/bun", + timeoutMs: 300, + intervalMs: 10, + isHealthy: async () => healthy, + isPortReachable: async () => false, + launchDaemon: () => { + launchCount += 1; + healthy = true; + return { pid: 54321 } as ChildProcess; + }, + }); + + expect(launchCount).toBe(1); + expect(existsSync(paths.lockPath)).toBe(false); + expect(JSON.parse(readFileSync(paths.metadataPath, "utf8"))).toMatchObject({ + pid: 54321, + launchedByPid: process.pid, + launchCwd: "/repo", + }); + }); });