fix(coding-agent/dap): bound writeMessage flush and clean up leaked adapter processes

- Replace the module-level writeMessage with a private DapClient.#writeMessage
  that races sink.flush() against a 30 s cap and this.proc.exited, disposes the
  client, and rethrows on either failure. A wedged adapter stdin previously
  blocked writeMessage forever.
- sendRequest fires the write in the background and attaches a passive
  promise.catch guard so the request timer's rejection cannot become an
  unhandled promise before the caller's await subscribes. This also lets the
  caller's timeoutMs win over a hung flush instead of waiting the full 30 s
  write cap.
- #spawnSocketUnix and #spawnSocketClientAddr wrap the readiness / connect
  race in try/catch that kills proc before rethrowing. Add a private
  socketReadyTimeoutMs hook on DapSpawnOptions so tests do not have to wait
  the full 10 s socket cap.

Regression tests cover all three failure modes: a wedged stdin flush must
reject sendRequest at the caller timeout with no unhandledRejection, and the
Unix + TCP client-addr spawn paths must terminate the detached adapter
process when the socket never appears.

Fixes #4233

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

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<unknown>;
 
 const DEFAULT_REQUEST_TIMEOUT_MS = 30_000;
-
-async function writeMessage(sink: DapWriteSink, message: DapRequestMessage | DapResponseMessage): Promise<void> {
-	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<DapClient> {
+	static async spawn({ adapter, cwd, socketReadyTimeoutMs }: DapSpawnOptions): Promise<DapClient> {
 		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:<path>
 	 * macOS/other: the adapter dials into our TCP listener via --client-addr
 	 */
-	static async #spawnSocket({ adapter, cwd }: DapSpawnOptions): Promise<DapClient> {
+	static async #spawnSocket({ adapter, cwd, socketReadyTimeoutMs }: DapSpawnOptions): Promise<DapClient> {
 		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:<path>, then connect to the socket. */
@@ -126,10 +136,12 @@ export class DapClient {
 		adapter,
 		cwd,
 		env,
+		timeoutMs,
 	}: {
 		adapter: DapResolvedAdapter;
 		cwd: string;
 		env: Record<string, string | undefined>;
+		timeoutMs: number;
 	}): Promise<DapClient> {
 		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<string, string | undefined>;
+		timeoutMs: number;
 	}): Promise<DapClient> {
 		const { promise: connPromise, resolve: resolveConn } = Promise.withResolvers<Bun.Socket<undefined>>();
 
@@ -182,25 +206,32 @@ export class DapClient {
 			detached: true,
 		});
 
-		// Wait for dlv to connect (with timeout)
-		let rawSocket: Bun.Socket<undefined>;
+		// 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<never>();
 		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<TBody>();
+		// 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<void> {
+		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<void>();
+		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<void> {

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<number>();
+		const proc = {
+			exited: procExited.promise,
+			exitCode: null,
+			stdin: { write: () => 0, flush: () => undefined },
+			stdout: new ReadableStream<Uint8Array>(),
+			stderr: new ReadableStream<Uint8Array>(),
+			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<number>(() => {}),
+		};
+		const readable = new ReadableStream<Uint8Array>();
+		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", () => {