From 53d6bc50580f152cd4e3383191ddc39578e843e1 Mon Sep 17 00:00:00 2001 From: Samuel Ainsworth Date: Sun, 30 Aug 2026 17:28:21 -0400 Subject: [PATCH] fix(session): report daemon health probe failures --- .changeset/quiet-daemons-report.md | 5 + .../src/session/agent/commands.daemon.test.ts | 45 ++++- packages/hunk/src/session/agent/commands.ts | 30 ++- .../src/session/broker/brokerLauncher.test.ts | 179 +++++++++++++++++ .../hunk/src/session/broker/brokerLauncher.ts | 184 ++++++++++++++++-- 5 files changed, 416 insertions(+), 27 deletions(-) create mode 100644 .changeset/quiet-daemons-report.md diff --git a/.changeset/quiet-daemons-report.md b/.changeset/quiet-daemons-report.md new file mode 100644 index 000000000..30ba292dc --- /dev/null +++ b/.changeset/quiet-daemons-report.md @@ -0,0 +1,5 @@ +--- +"hunkdiff": patch +--- + +Report the underlying health-probe failure when a reachable session daemon port cannot be verified. diff --git a/packages/hunk/src/session/agent/commands.daemon.test.ts b/packages/hunk/src/session/agent/commands.daemon.test.ts index 93f59b68d..fbeb96e2b 100644 --- a/packages/hunk/src/session/agent/commands.daemon.test.ts +++ b/packages/hunk/src/session/agent/commands.daemon.test.ts @@ -76,9 +76,17 @@ describe("resolveDaemonAvailability with no daemon listening", () => { }); describe("resolveDaemonAvailability with a foreign process on the port", () => { - probeTest("throws a port-conflict error when the port is reachable but unhealthy", async () => { - // A non-broker server occupies the port: reachable (TCP connects) but not health-OK. - const server = Bun.serve({ port: 0, fetch: () => new Response("nope", { status: 404 }) }); + probeTest("throws a port-conflict error with the terminal health failure", async () => { + // The second response proves the diagnostic comes from the probe after TCP reachability, + // rather than stale evidence captured before a listener generation could change. + let healthRequests = 0; + const server = Bun.serve({ + port: 0, + fetch: () => { + healthRequests += 1; + return new Response("nope", { status: healthRequests === 1 ? 503 : 404 }); + }, + }); process.env.HUNK_MCP_PORT = String(server.port); try { await expect( @@ -87,7 +95,36 @@ describe("resolveDaemonAvailability with a foreign process on the port", () => { action: "list", output: "json", } satisfies SessionCommandInput), - ).rejects.toThrow(/already in use/); + ).rejects.toThrow( + /already in use.*Hunk health probe returned HTTP 404 after \d+ms.*busy Hunk daemon or another process/, + ); + expect(healthRequests).toBe(2); + } finally { + server.stop(true); + } + }); + + probeTest("returns an empty list when the listener disappears during health checks", async () => { + let healthRequests = 0; + let server!: ReturnType; + server = Bun.serve({ + port: 0, + fetch: () => { + healthRequests += 1; + if (healthRequests === 2) queueMicrotask(() => server.stop(true)); + return new Response("unavailable", { status: 503 }); + }, + }); + process.env.HUNK_MCP_PORT = String(server.port); + + try { + const output = await runSessionCommand({ + kind: "session", + action: "list", + output: "json", + } satisfies SessionCommandInput); + expect(JSON.parse(output)).toEqual({ sessions: [] }); + expect(healthRequests).toBe(2); } finally { server.stop(true); } diff --git a/packages/hunk/src/session/agent/commands.ts b/packages/hunk/src/session/agent/commands.ts index 714c6df02..33ccfdcba 100644 --- a/packages/hunk/src/session/agent/commands.ts +++ b/packages/hunk/src/session/agent/commands.ts @@ -1,7 +1,11 @@ import type { SessionCommandInput, SessionCommandOutput } from "../../core/run/commandInputs"; import type { SessionLiveCommentSummary, SessionReviewNoteSummary } from "../types"; import { NO_ACTIVE_SESSIONS_MESSAGE } from "./errors"; -import { isSessionBrokerHealthy, isLoopbackPortReachable } from "../broker/brokerLauncher"; +import { + describeSessionBrokerHealthProbeFailure, + isLoopbackPortReachable, + probeSessionBrokerHealth, +} from "../broker/brokerLauncher"; import { resolveSessionBrokerConfig } from "../broker/brokerConfig"; import { normalizeSessionSelector } from "@hunk/session-broker-core"; import { SessionBrokerClientAuthenticationError } from "@hunk/session-broker"; @@ -79,17 +83,29 @@ async function ensureRequiredAction(action: SessionDaemonAction, client = create async function resolveDaemonAvailability(action: SessionCommandInput["action"]) { const config = resolveSessionBrokerConfig(); - const healthy = await isSessionBrokerHealthy(config); - if (healthy) { + const initialHealthProbe = await probeSessionBrokerHealth(config); + if (initialHealthProbe.kind === "healthy") { return true; } const portReachable = await isLoopbackPortReachable(config); if (portReachable) { - throw new Error( - `Hunk session daemon 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.`, - ); + // Probe again so the diagnostic describes the listener that accepted the reachability check, + // rather than a daemon generation that may have stopped or recovered in the meantime. + const terminalHealthProbe = await probeSessionBrokerHealth(config); + if (terminalHealthProbe.kind === "healthy") { + return true; + } + // If the listener disappeared during the probes, preserve the ordinary no-daemon behavior. + // A remaining listener makes this the terminal probe evidence included in the CLI error. + if (await isLoopbackPortReachable(config)) { + const diagnostic = describeSessionBrokerHealthProbeFailure(terminalHealthProbe); + throw new Error( + `Hunk session daemon port ${config.host}:${config.port} is already in use, and the listener's ` + + `Hunk health probe ${diagnostic}. The listener may be a busy Hunk daemon or another process. ` + + `Retry, stop the conflicting process, or set HUNK_MCP_PORT to a different loopback port.`, + ); + } } if (action === "list") { diff --git a/packages/hunk/src/session/broker/brokerLauncher.test.ts b/packages/hunk/src/session/broker/brokerLauncher.test.ts index bb9eeaf55..c597e9609 100644 --- a/packages/hunk/src/session/broker/brokerLauncher.test.ts +++ b/packages/hunk/src/session/broker/brokerLauncher.test.ts @@ -1,12 +1,16 @@ 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, type Socket } from "node:net"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { + describeSessionBrokerHealthProbeFailure, ensureSessionBrokerAvailable, isLoopbackPortReachable, parseSessionBrokerHealth, + probeSessionBrokerHealth, + readSessionBrokerHealth, readSessionBrokerLaunchFingerprint, resolveDaemonLaunchCommand, resolveSessionBrokerRuntimePaths, @@ -21,6 +25,16 @@ const testConfig = { wsOrigin: "ws://127.0.0.1:47657", }; +/** Create a broker config for one ephemeral test listener. */ +function createTestBrokerConfig(port: number) { + return { + host: "127.0.0.1", + port, + httpOrigin: `http://127.0.0.1:${port}`, + wsOrigin: `ws://127.0.0.1:${port}`, + }; +} + /** Create manually settled foreign work for launcher commit-fence tests. */ function createDeferredTest() { let resolve!: (value: T | PromiseLike) => void; @@ -119,6 +133,171 @@ describe("session daemon launcher", () => { expect(parseSessionBrokerHealth(value)).toBeNull(); } }); + + test("retains the health payload for a successful probe", async () => { + const server = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch: () => Response.json({ ok: true, pid: 123, sessions: 1 }), + }); + const config = createTestBrokerConfig(server.port!); + + try { + await expect(probeSessionBrokerHealth(config)).resolves.toMatchObject({ + kind: "healthy", + health: { ok: true, pid: 123, sessions: 1 }, + }); + } finally { + server.stop(true); + } + }); + + test("retains HTTP status while nullable health readers stay compatible", async () => { + const server = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch: () => new Response("unavailable", { status: 503 }), + }); + const config = createTestBrokerConfig(server.port!); + + try { + await expect(probeSessionBrokerHealth(config)).resolves.toMatchObject({ + kind: "http-status", + status: 503, + }); + await expect(readSessionBrokerHealth(config)).resolves.toBeNull(); + } finally { + server.stop(true); + } + }); + + test("distinguishes invalid JSON, incompatible payloads, and oversized bodies", async () => { + let response: "json" | "utf8" | "payload" | "large" = "json"; + const server = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch: () => { + if (response === "json") { + return new Response("not-json", { headers: { "content-type": "application/json" } }); + } + if (response === "utf8") { + return new Response( + new Uint8Array([ + ...new TextEncoder().encode('{"ok":true,"startedAt":"'), + 0xff, + ...new TextEncoder().encode('"}'), + ]), + ); + } + if (response === "payload") return Response.json({ ok: "yes" }); + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array(32 * 1024)); + controller.enqueue(new Uint8Array(32 * 1024 + 1)); + controller.close(); + }, + }), + ); + }, + }); + const config = createTestBrokerConfig(server.port!); + + try { + await expect(probeSessionBrokerHealth(config)).resolves.toMatchObject({ + kind: "invalid-json", + }); + response = "utf8"; + await expect(probeSessionBrokerHealth(config)).resolves.toMatchObject({ + kind: "invalid-json", + }); + response = "payload"; + await expect(probeSessionBrokerHealth(config)).resolves.toMatchObject({ + kind: "invalid-response", + }); + response = "large"; + await expect(probeSessionBrokerHealth(config)).resolves.toMatchObject({ + kind: "response-too-large", + limitBytes: 64 * 1024, + }); + } finally { + server.stop(true); + } + }); + + test("classifies truncated response bodies as request errors", async () => { + const sockets = new Set(); + const server = createServer((socket) => { + sockets.add(socket); + socket.once("close", () => sockets.delete(socket)); + socket.end("HTTP/1.1 200 OK\r\nContent-Length: 20\r\nConnection: close\r\n\r\n{"); + }); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolve); + }); + const address = server.address(); + const port = typeof address === "object" && address ? address.port : 0; + const config = createTestBrokerConfig(port); + + try { + await expect(probeSessionBrokerHealth(config)).resolves.toMatchObject({ + kind: "request-error", + }); + } finally { + const closed = new Promise((resolve) => server.close(() => resolve())); + for (const socket of sockets) socket.destroy(); + await closed; + } + }); + + test("enforces the timeout across response body consumption", async () => { + const server = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch: () => + new Response( + new ReadableStream({ + async pull(controller) { + await Bun.sleep(100); + controller.enqueue(new TextEncoder().encode('{"ok":true}')); + controller.close(); + }, + }), + ), + }); + const config = createTestBrokerConfig(server.port!); + + try { + const result = await probeSessionBrokerHealth(config, 10); + expect(result).toMatchObject({ kind: "timeout", timeoutMs: 10 }); + if (result.kind !== "timeout") throw new Error("Expected a timeout health probe result."); + expect(result.elapsedMs).toBeGreaterThanOrEqual(10); + expect(describeSessionBrokerHealthProbeFailure(result)).toMatch( + /^timed out after 10ms \(probe elapsed \d+ms\)$/, + ); + } finally { + server.stop(true); + } + }); + + test("rejects health redirects instead of following a foreign listener", async () => { + const server = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch: () => Response.redirect("https://example.com/health"), + }); + const config = createTestBrokerConfig(server.port!); + + try { + await expect(probeSessionBrokerHealth(config)).resolves.toMatchObject({ + kind: "request-error", + }); + } finally { + server.stop(true); + } + }); + test("reuses the current script entrypoint when Hunk is running from source or a JS wrapper", () => { expect(resolveDaemonLaunchCommand(["bun", "src/main.tsx", "diff"], "/usr/bin/bun")).toEqual({ command: "/usr/bin/bun", diff --git a/packages/hunk/src/session/broker/brokerLauncher.ts b/packages/hunk/src/session/broker/brokerLauncher.ts index 7fd6452b0..3d32893f4 100644 --- a/packages/hunk/src/session/broker/brokerLauncher.ts +++ b/packages/hunk/src/session/broker/brokerLauncher.ts @@ -19,6 +19,7 @@ const DEFAULT_DAEMON_LOCK_STALE_MS = 15_000; const DEFAULT_DAEMON_STARTUP_TIMEOUT_MS = 3_000; const DEFAULT_DAEMON_HEALTH_POLL_INTERVAL_MS = 100; const MAX_DAEMON_LAUNCH_METADATA_BYTES = 16 * 1024; +const MAX_DAEMON_HEALTH_RESPONSE_BYTES = 64 * 1024; export interface DaemonLaunchCommand { command: string; @@ -375,6 +376,100 @@ export interface SessionBrokerHealth { staleSessionTtlMs?: number; } +type SessionBrokerHealthProbeResult = + | { kind: "healthy"; health: SessionBrokerHealth } + | { kind: "http-status"; status: number; elapsedMs: number } + | { kind: "invalid-json"; elapsedMs: number } + | { kind: "invalid-response"; elapsedMs: number } + | { kind: "response-too-large"; limitBytes: number; elapsedMs: number } + | { kind: "timeout"; timeoutMs: number; elapsedMs: number } + | { kind: "request-error"; message: string; elapsedMs: number }; + +type SessionBrokerHealthProbeFailure = Exclude; + +class SessionBrokerHealthResponseTooLargeError extends Error {} +class SessionBrokerHealthInvalidJsonError extends Error {} + +/** Round one failed probe duration for stable, human-readable diagnostics. */ +function healthProbeElapsedMs(startedAt: number) { + return Math.max(0, Math.round(performance.now() - startedAt)); +} + +/** Bound one runtime-generated transport error before it reaches a terminal. */ +function healthProbeErrorMessage(error: unknown) { + const raw = error instanceof Error ? error.message || error.name : String(error); + return ( + raw + .replace(/[\u0000-\u001f\u007f-\u009f]/g, " ") + .replace(/\s+/g, " ") + .trim() + .slice(0, 240) || "Unknown request error" + ); +} + +/** Read one health response body without allowing a foreign listener to stream unbounded data. */ +async function readSessionBrokerHealthJson(response: Response) { + const declaredLength = response.headers.get("content-length"); + if ( + declaredLength !== null && + /^(?:0|[1-9][0-9]*)$/.test(declaredLength) && + Number(declaredLength) > MAX_DAEMON_HEALTH_RESPONSE_BYTES + ) { + await response.body?.cancel().catch(() => undefined); + throw new SessionBrokerHealthResponseTooLargeError(); + } + + if (!response.body) throw new SessionBrokerHealthInvalidJsonError(); + const reader = response.body.getReader(); + const chunks: Uint8Array[] = []; + let totalBytes = 0; + try { + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + if (!value) continue; + totalBytes += value.byteLength; + if (totalBytes > MAX_DAEMON_HEALTH_RESPONSE_BYTES) { + await reader.cancel().catch(() => undefined); + throw new SessionBrokerHealthResponseTooLargeError(); + } + chunks.push(value); + } + } finally { + reader.releaseLock(); + } + + const bytes = new Uint8Array(totalBytes); + let offset = 0; + for (const chunk of chunks) { + bytes.set(chunk, offset); + offset += chunk.byteLength; + } + try { + return JSON.parse(new TextDecoder("utf-8", { fatal: true }).decode(bytes)) as unknown; + } catch { + throw new SessionBrokerHealthInvalidJsonError(); + } +} + +/** Describe one failed health probe for the final user-facing CLI error. */ +export function describeSessionBrokerHealthProbeFailure(failure: SessionBrokerHealthProbeFailure) { + switch (failure.kind) { + case "http-status": + return `returned HTTP ${failure.status} after ${failure.elapsedMs}ms`; + case "invalid-json": + return `returned invalid JSON after ${failure.elapsedMs}ms`; + case "invalid-response": + return `returned an incompatible health payload after ${failure.elapsedMs}ms`; + case "response-too-large": + return `exceeded ${failure.limitBytes} bytes after ${failure.elapsedMs}ms`; + case "timeout": + return `timed out after ${failure.timeoutMs}ms (probe elapsed ${failure.elapsedMs}ms)`; + case "request-error": + return `failed after ${failure.elapsedMs}ms (${failure.message})`; + } +} + /** Parse the minimal or legacy-rich health response without trusting cross-process JSON. */ export function parseSessionBrokerHealth(value: unknown): SessionBrokerHealth | null { try { @@ -450,37 +545,94 @@ export function readSessionBrokerLaunchFingerprint( } } -/** Read the daemon's health payload when one is reachable on the configured loopback port. */ -export async function readSessionBrokerHealth( +/** Probe daemon health while retaining bounded failure evidence for a terminal CLI error. */ +export async function probeSessionBrokerHealth( config: ResolvedSessionBrokerConfig = resolveSessionBrokerConfig(), timeoutMs = 500, -) { +): Promise { + const startedAt = performance.now(); const controller = new AbortController(); - const timeout = setTimeout(() => controller.abort(), timeoutMs); - timeout.unref?.(); + let timeout: ReturnType; + const timeoutResult = new Promise((resolveTimeout) => { + // Keep this timer referenced: Bun 1.3.x on Windows can skip unref'ed timeout guards. + timeout = setTimeout(() => { + resolveTimeout({ + kind: "timeout", + timeoutMs, + elapsedMs: healthProbeElapsedMs(startedAt), + }); + controller.abort(); + }, timeoutMs); + }); + const requestResult = (async (): Promise => { + try { + const response = await fetch(`${config.httpOrigin}/health`, { + redirect: "error", + signal: controller.signal, + }); + if (!response.ok) { + await response.body?.cancel().catch(() => undefined); + return { + kind: "http-status", + status: response.status, + elapsedMs: healthProbeElapsedMs(startedAt), + }; + } - try { - const response = await fetch(`${config.httpOrigin}/health`, { - signal: controller.signal, - }); - if (!response.ok) { - return null; + let payload: unknown; + try { + payload = await readSessionBrokerHealthJson(response); + } catch (error) { + if (error instanceof SessionBrokerHealthResponseTooLargeError) { + return { + kind: "response-too-large", + limitBytes: MAX_DAEMON_HEALTH_RESPONSE_BYTES, + elapsedMs: healthProbeElapsedMs(startedAt), + }; + } + if (error instanceof SessionBrokerHealthInvalidJsonError) { + return { kind: "invalid-json", elapsedMs: healthProbeElapsedMs(startedAt) }; + } + throw error; + } + + const health = parseSessionBrokerHealth(payload); + return health + ? { kind: "healthy", health } + : { kind: "invalid-response", elapsedMs: healthProbeElapsedMs(startedAt) }; + } catch (error) { + return { + kind: "request-error", + message: healthProbeErrorMessage(error), + elapsedMs: healthProbeElapsedMs(startedAt), + }; } + })(); - return parseSessionBrokerHealth(await response.json()); - } catch { - return null; + try { + return await Promise.race([requestResult, timeoutResult]); } finally { - clearTimeout(timeout); + clearTimeout(timeout!); + // A runtime may ignore abort; consume any later rejection after the timeout result wins. + requestResult.catch(() => undefined); } } +/** Read the daemon's health payload while preserving the nullable compatibility contract. */ +export async function readSessionBrokerHealth( + config: ResolvedSessionBrokerConfig = resolveSessionBrokerConfig(), + timeoutMs = 500, +) { + const result = await probeSessionBrokerHealth(config, timeoutMs); + return result.kind === "healthy" ? result.health : null; +} + /** Check whether the loopback session broker already answers health probes. */ export async function isSessionBrokerHealthy( config: ResolvedSessionBrokerConfig = resolveSessionBrokerConfig(), timeoutMs = 500, ) { - return (await readSessionBrokerHealth(config, timeoutMs))?.ok === true; + return (await probeSessionBrokerHealth(config, timeoutMs)).kind === "healthy"; } /** Check whether some local process is already accepting TCP connections on the daemon port. */