diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 64c3c57ae..7dfc6d9ee 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Fixed + +- Fixed the DAP client hanging forever when the debug adapter's stdin stops draining. `writeMessage` now races the flush against a 30 s cap and adapter exit and disposes the client on either failure, and `sendRequest` fires the write in the background with a passive unhandled-rejection guard so the request-timer error is never orphaned before the caller subscribes. Socket-mode spawn helpers (`#spawnSocketUnix`, `#spawnSocketClientAddr`) now kill the detached adapter process when the readiness/connect race fails, closing an orphan-process leak on socket connect timeouts ([#4233](https://github.com/can1357/oh-my-pi/issues/4233)). + ## [16.3.1] - 2026-07-02 ### Breaking Changes diff --git a/packages/coding-agent/src/dap/client.ts b/packages/coding-agent/src/dap/client.ts index 46740d80b..eb4230e23 100644 --- a/packages/coding-agent/src/dap/client.ts +++ b/packages/coding-agent/src/dap/client.ts @@ -17,6 +17,14 @@ import type { interface DapSpawnOptions { adapter: DapResolvedAdapter; cwd: string; + /** + * Cap on how long the socket-mode helpers wait for the adapter to open its + * socket (unix) or dial back into our listener (TCP). Exposed for tests; + * production callers rely on the default. + * + * @internal + */ + socketReadyTimeoutMs?: number; } /** Minimal write interface shared by Bun.FileSink and Bun TCP sockets. */ @@ -29,13 +37,14 @@ type DapEventHandler = (body: unknown, event: DapEventMessage) => void | Promise type DapReverseRequestHandler = (args: unknown) => unknown | Promise; const DEFAULT_REQUEST_TIMEOUT_MS = 30_000; - -async function writeMessage(sink: DapWriteSink, message: DapRequestMessage | DapResponseMessage): Promise { - const content = JSON.stringify(message); - sink.write(`Content-Length: ${Buffer.byteLength(content, "utf-8")}\r\n\r\n`); - sink.write(content); - await sink.flush(); -} +/** + * Hard cap on a single message write. A wedged adapter stdin used to hang the + * whole client forever; on hitting this cap the client disposes itself so the + * next request fails fast instead of piling more work onto a broken adapter. + */ +const WRITE_MESSAGE_TIMEOUT_MS = 30_000; +/** Default wait for socket-mode adapters to become reachable. */ +const SOCKET_READY_TIMEOUT_MS = 10_000; function toErrorMessage(value: unknown): string { if (value instanceof Error) return value.message; @@ -77,9 +86,9 @@ export class DapClient { this.#socket = options?.socket; } - static async spawn({ adapter, cwd }: DapSpawnOptions): Promise { + static async spawn({ adapter, cwd, socketReadyTimeoutMs }: DapSpawnOptions): Promise { if (adapter.connectMode === "socket") { - return DapClient.#spawnSocket({ adapter, cwd }); + return DapClient.#spawnSocket({ adapter, cwd, socketReadyTimeoutMs }); } // Merge non-interactive env and start in a new session (detached → setsid) // so the adapter process tree has no controlling terminal. Without this, @@ -108,17 +117,18 @@ export class DapClient { * Linux: connect to a unix domain socket via --listen=unix: * macOS/other: the adapter dials into our TCP listener via --client-addr */ - static async #spawnSocket({ adapter, cwd }: DapSpawnOptions): Promise { + static async #spawnSocket({ adapter, cwd, socketReadyTimeoutMs }: DapSpawnOptions): Promise { const env = { ...Bun.env, ...NON_INTERACTIVE_ENV, }; + const timeoutMs = socketReadyTimeoutMs ?? SOCKET_READY_TIMEOUT_MS; const isLinux = process.platform === "linux"; if (isLinux) { - return DapClient.#spawnSocketUnix({ adapter, cwd, env }); + return DapClient.#spawnSocketUnix({ adapter, cwd, env, timeoutMs }); } - return DapClient.#spawnSocketClientAddr({ adapter, cwd, env }); + return DapClient.#spawnSocketClientAddr({ adapter, cwd, env, timeoutMs }); } /** Linux: spawn adapter with --listen=unix:, then connect to the socket. */ @@ -126,10 +136,12 @@ export class DapClient { adapter, cwd, env, + timeoutMs, }: { adapter: DapResolvedAdapter; cwd: string; env: Record; + timeoutMs: number; }): Promise { const socketPath = `/tmp/dap-${adapter.name}-${Date.now()}-${Math.random().toString(36).slice(2)}.sock`; const proc = ptree.spawn([adapter.resolvedCommand, ...adapter.args, `--listen=unix:${socketPath}`], { @@ -139,13 +151,23 @@ export class DapClient { detached: true, }); - await waitForCondition(() => isUnixSocketReady(socketPath), 10_000, proc); - - const { readable, writeSink, socket } = await connectSocket({ unix: socketPath }); - const client = new DapClient(adapter, cwd, proc, { readable, writeSink, socket }); - proc.exited.then(() => client.#handleProcessExit()); - void client.#startMessageReader(); - return client; + // If waitForCondition throws (timeout, or adapter exited early) or the + // socket connect fails, we must not leak the detached adapter process. + try { + await waitForCondition(() => isUnixSocketReady(socketPath), timeoutMs, proc); + const { readable, writeSink, socket } = await connectSocket({ unix: socketPath }); + const client = new DapClient(adapter, cwd, proc, { readable, writeSink, socket }); + proc.exited.then(() => client.#handleProcessExit()); + void client.#startMessageReader(); + return client; + } catch (error) { + try { + proc.kill(); + } catch { + /* proc may already be dead */ + } + throw error; + } } /** macOS/other: listen on a random TCP port, spawn adapter with --client-addr, accept connection. */ @@ -153,10 +175,12 @@ export class DapClient { adapter, cwd, env, + timeoutMs, }: { adapter: DapResolvedAdapter; cwd: string; env: Record; + timeoutMs: number; }): Promise { const { promise: connPromise, resolve: resolveConn } = Promise.withResolvers>(); @@ -182,25 +206,32 @@ export class DapClient { detached: true, }); - // Wait for dlv to connect (with timeout) - let rawSocket: Bun.Socket; + // Wait for the adapter to dial back. On timeout (or any other failure + // before we've wired up the client) kill `proc` — otherwise the detached + // adapter process is orphaned. const { promise: timeoutPromise, reject: rejectTimeout } = Promise.withResolvers(); const connectTimeout = setTimeout( - () => rejectTimeout(new Error(`${adapter.name} did not connect within 10s`)), - 10_000, + () => rejectTimeout(new Error(`${adapter.name} did not connect within ${timeoutMs}ms`)), + timeoutMs, ); try { - rawSocket = await Promise.race([connPromise, timeoutPromise]); + const rawSocket = await Promise.race([connPromise, timeoutPromise]); + const { readable, writeSink, socket } = wrapBunSocket(rawSocket); + const client = new DapClient(adapter, cwd, proc, { readable, writeSink, socket }); + proc.exited.then(() => client.#handleProcessExit()); + void client.#startMessageReader(); + return client; + } catch (error) { + try { + proc.kill(); + } catch { + /* proc may already be dead */ + } + throw error; } finally { clearTimeout(connectTimeout); server.stop(); } - - const { readable, writeSink, socket } = wrapBunSocket(rawSocket); - const client = new DapClient(adapter, cwd, proc, { readable, writeSink, socket }); - proc.exited.then(() => client.#handleProcessExit()); - void client.#startMessageReader(); - return client; } get capabilities(): DapCapabilities | undefined { @@ -309,6 +340,12 @@ export class DapClient { arguments: args, }; const { promise, resolve, reject } = Promise.withResolvers(); + // Suppress "unhandled rejection" if the request timer or abort fires + // before the caller's `await` subscribes — e.g. while #writeMessage is + // still racing a wedged stdin flush. The caller's own `await` still + // receives the rejection normally; this handler is a passive guard. + promise.catch(() => {}); + let timeout: NodeJS.Timeout | undefined; const cleanup = () => { if (timeout) clearTimeout(timeout); @@ -342,13 +379,15 @@ export class DapClient { }, }); this.#lastActivity = Date.now(); - try { - await writeMessage(this.#writeSink, request); - } catch (error) { + // Fire the write in the background. Awaiting it here would let a wedged + // stdin flush block the caller's `timeoutMs`; if it fails, propagate the + // failure into `promise` — the timer or abort may still win the race. + void this.#writeMessage(request).catch(error => { + if (!this.#pendingRequests.has(requestSeq)) return; this.#pendingRequests.delete(requestSeq); cleanup(); - throw error; - } + reject(error); + }); return promise; } @@ -362,7 +401,50 @@ export class DapClient { ...(message ? { message } : {}), ...(body !== undefined ? { body } : {}), }; - await writeMessage(this.#writeSink, response); + await this.#writeMessage(response); + } + + /** + * Framed write to the adapter, bounded by {@link WRITE_MESSAGE_TIMEOUT_MS} + * and by adapter exit. Without this bound a wedged adapter stdin used to + * hang the whole client forever. On timeout or exit-before-flush the client + * disposes itself and rethrows. + */ + async #writeMessage(message: DapRequestMessage | DapResponseMessage): Promise { + const content = JSON.stringify(message); + this.#writeSink.write(`Content-Length: ${Buffer.byteLength(content, "utf-8")}\r\n\r\n`); + this.#writeSink.write(content); + const flushResult = this.#writeSink.flush(); + if (!(flushResult instanceof Promise)) return; + + const { promise: guardPromise, reject: guardReject, resolve: guardResolve } = Promise.withResolvers(); + const timer = setTimeout( + () => + guardReject( + new Error(`DAP adapter ${this.adapter.name} write timed out after ${WRITE_MESSAGE_TIMEOUT_MS}ms`), + ), + WRITE_MESSAGE_TIMEOUT_MS, + ); + // If the adapter exits mid-write, fail fast rather than blocking forever + // on a stdin that will never drain. `proc.exited` may resolve normally + // (clean exit) or reject (non-zero); either way the write is doomed. + const onExit = () => { + guardReject(new Error(`DAP adapter ${this.adapter.name} exited before write completed`)); + }; + this.proc.exited.then(onExit, onExit); + + try { + await Promise.race([flushResult, guardPromise]); + } catch (error) { + // The client is now known-broken. Kick off dispose in the background; + // callers will see subsequent sendRequest calls fail fast. + void this.dispose(); + throw error; + } finally { + clearTimeout(timer); + // Release the guard so any late onExit call becomes a no-op. + guardResolve(); + } } async dispose(): Promise { diff --git a/packages/coding-agent/test/debug/dap-launch-failures.test.ts b/packages/coding-agent/test/debug/dap-launch-failures.test.ts index b39f93f86..3664844c2 100644 --- a/packages/coding-agent/test/debug/dap-launch-failures.test.ts +++ b/packages/coding-agent/test/debug/dap-launch-failures.test.ts @@ -364,6 +364,129 @@ describe("DAP launch failure handling", () => { await removeWithRetries(cwd); } }); + + it("times out promptly and does not emit an unhandled rejection when the stdin flush is wedged", async () => { + const procExited = Promise.withResolvers(); + const proc = { + exited: procExited.promise, + exitCode: null, + stdin: { write: () => 0, flush: () => undefined }, + stdout: new ReadableStream(), + stderr: new ReadableStream(), + peekStderr: () => "", + kill: () => { + procExited.resolve(-1); + return true; + }, + } as unknown as DapClientState["proc"]; + // flush() returns a promise that never resolves — models an adapter whose + // stdin has stopped draining (the failure mode in issue #4233). + const writeSink = { + write: (_data: string | Uint8Array) => 0, + flush: () => new Promise(() => {}), + }; + const readable = new ReadableStream(); + const client = new DapClient(TEST_ADAPTER, process.cwd(), proc, { readable, writeSink }); + + const unhandled: unknown[] = []; + const onUnhandled = (reason: unknown) => unhandled.push(reason); + process.on("unhandledRejection", onUnhandled); + + try { + const start = Date.now(); + await expect(client.sendRequest("initialize", {}, undefined, 50)).rejects.toThrow(/timed out/i); + // Must respect the caller's timeoutMs, not the internal 30 s write cap. + expect(Date.now() - start).toBeLessThan(500); + // Let any queued unhandled-rejection microtask fire. + await Bun.sleep(50); + expect(unhandled).toEqual([]); + } finally { + process.off("unhandledRejection", onUnhandled); + // Let writeMessage's exit-guard resolve so no promise leaks past the test. + await client.dispose(); + await Bun.sleep(20); + } + }); + + it("kills the detached adapter process when the Unix socket never appears (Linux)", async () => { + if (process.platform !== "linux") return; + const cwd = await fs.mkdtemp(path.join(os.tmpdir(), "omp-debug-unix-leak-")); + try { + const adapterPath = path.join(cwd, "wedged-unix-adapter.mjs"); + const pidFilePath = path.join(cwd, "adapter.pid"); + // Adapter records its pid and stays alive without ever creating the + // socket, forcing #spawnSocketUnix's readiness wait to time out. + await fs.writeFile( + adapterPath, + `await Bun.write(${JSON.stringify(pidFilePath)}, String(process.pid));\nawait Bun.sleep(60_000);\n`, + ); + const adapter: DapResolvedAdapter = { + ...TEST_ADAPTER, + name: "wedged-unix-adapter", + command: process.execPath, + args: [adapterPath], + resolvedCommand: process.execPath, + connectMode: "socket", + }; + await expect(DapClient.spawn({ adapter, cwd, socketReadyTimeoutMs: 300 })).rejects.toThrow( + /Socket not ready/, + ); + // Wait for the kill signal to propagate to the detached adapter. + await Bun.sleep(500); + const adapterPid = Number(await Bun.file(pidFilePath).text()); + expect(Number.isFinite(adapterPid)).toBe(true); + let alive = true; + try { + process.kill(adapterPid, 0); + } catch { + alive = false; + } + expect(alive).toBe(false); + } finally { + await removeWithRetries(cwd); + } + }); + + it("kills the detached adapter process when it never dials back on the TCP client-addr path", async () => { + const originalPlatform = process.platform; + Object.defineProperty(process, "platform", { value: "darwin", configurable: true }); + try { + const cwd = await fs.mkdtemp(path.join(os.tmpdir(), "omp-debug-tcp-leak-")); + try { + const adapterPath = path.join(cwd, "wedged-tcp-adapter.mjs"); + const pidFilePath = path.join(cwd, "adapter.pid"); + await fs.writeFile( + adapterPath, + `await Bun.write(${JSON.stringify(pidFilePath)}, String(process.pid));\nawait Bun.sleep(60_000);\n`, + ); + const adapter: DapResolvedAdapter = { + ...TEST_ADAPTER, + name: "wedged-tcp-adapter", + command: process.execPath, + args: [adapterPath], + resolvedCommand: process.execPath, + connectMode: "socket", + }; + await expect(DapClient.spawn({ adapter, cwd, socketReadyTimeoutMs: 300 })).rejects.toThrow( + /did not connect within/, + ); + await Bun.sleep(500); + const adapterPid = Number(await Bun.file(pidFilePath).text()); + expect(Number.isFinite(adapterPid)).toBe(true); + let alive = true; + try { + process.kill(adapterPid, 0); + } catch { + alive = false; + } + expect(alive).toBe(false); + } finally { + await removeWithRetries(cwd); + } + } finally { + Object.defineProperty(process, "platform", { value: originalPlatform, configurable: true }); + } + }); }); describe("DebugTool launch validation", () => {