fix(ssh): cancelled controlmaster stream waits

Return promptly when the ssh executor receives a user interrupt while ControlMaster-owned streams remain open.

Fixes #2180

packages/coding-agent/CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## [Unreleased]
 
+### Fixed
+
+- Fixed SSH tool cancellation hanging behind OpenSSH ControlMaster streams that stayed open after an Esc/user interrupt ([#2180](https://github.com/can1357/oh-my-pi/issues/2180)).
+
 ## [15.10.8] - 2026-06-09
 
 ### Added

packages/coding-agent/src/ssh/ssh-executor.ts

@@ -42,6 +42,42 @@ export interface SSHResult {
 	artifactId?: string;
 }
 
+type SSHExitEvent = { kind: "exit"; exitCode: number } | { kind: "error"; error: unknown };
+
+function sshExitEvent(exitCode: number): SSHExitEvent {
+	return { kind: "exit", exitCode };
+}
+
+function sshErrorEvent(error: unknown): SSHExitEvent {
+	return { kind: "error", error };
+}
+
+function createAbortWaiter(
+	signal: AbortSignal | undefined,
+	streamAbort: AbortController,
+): { promise: Promise<ptree.AbortError> | undefined; cleanup: () => void } {
+	if (!signal) {
+		return { promise: undefined, cleanup: () => {} };
+	}
+
+	const { promise, resolve } = Promise.withResolvers<ptree.AbortError>();
+	const onAbort = () => {
+		const error = new ptree.AbortError(signal.reason, "<cancelled>");
+		if (!streamAbort.signal.aborted) {
+			streamAbort.abort(error);
+		}
+		resolve(error);
+	};
+
+	if (signal.aborted) {
+		onAbort();
+		return { promise, cleanup: () => {} };
+	}
+
+	signal.addEventListener("abort", onAbort, { once: true });
+	return { promise, cleanup: () => signal.removeEventListener("abort", onAbort) };
+}
+
 function quoteForCompatShell(command: string): string {
 	if (command.length === 0) {
 		return "''";
@@ -94,19 +130,34 @@ export async function executeSSH(
 		maxColumns: resolveOutputMaxColumns(settings),
 	});
 
-	const streams = [child.stdout.pipeTo(sink.createInput())];
+	const streamAbort = new AbortController();
+	const abortWaiter = createAbortWaiter(options?.signal, streamAbort);
+	const streamOptions = { signal: streamAbort.signal };
+	const streams = [child.stdout.pipeTo(sink.createInput(), streamOptions)];
 	if (child.stderr) {
-		streams.push(child.stderr.pipeTo(sink.createInput()));
+		streams.push(child.stderr.pipeTo(sink.createInput(), streamOptions));
 	}
-	await Promise.allSettled(streams).catch(() => {});
+	const streamsSettled = Promise.allSettled(streams).then(() => {});
 
 	try {
+		const exitEvent = child.exited.then(sshExitEvent, sshErrorEvent);
+		const abortEvent = abortWaiter.promise?.then(sshErrorEvent);
+		const event = await (abortEvent ? Promise.race([exitEvent, abortEvent]) : exitEvent);
+		if (event.kind === "error") {
+			throw event.error;
+		}
+
+		await streamsSettled;
 		return {
-			exitCode: await child.exited,
+			exitCode: event.exitCode,
 			cancelled: false,
 			...(await sink.dump()),
 		};
 	} catch (err) {
+		if (!streamAbort.signal.aborted) {
+			streamAbort.abort(err);
+		}
+		void streamsSettled;
 		if (err instanceof ptree.Exception) {
 			if (err instanceof ptree.TimeoutError) {
 				return {
@@ -129,5 +180,7 @@ export async function executeSSH(
 			};
 		}
 		throw err;
+	} finally {
+		abortWaiter.cleanup();
 	}
 }

packages/coding-agent/test/ssh/ssh-executor.test.ts

@@ -0,0 +1,65 @@
+import { afterEach, describe, expect, it, vi } from "bun:test";
+import * as connectionManager from "@oh-my-pi/pi-coding-agent/ssh/connection-manager";
+import { executeSSH } from "@oh-my-pi/pi-coding-agent/ssh/ssh-executor";
+import * as sshfsMount from "@oh-my-pi/pi-coding-agent/ssh/sshfs-mount";
+import { type ChildProcess, ptree } from "@oh-my-pi/pi-utils";
+
+type TestStdin = "pipe" | "ignore" | Buffer | Uint8Array | null;
+
+function createNeverClosingStream(): ReadableStream<Uint8Array> {
+	return new ReadableStream<Uint8Array>({
+		start(controller) {
+			controller.enqueue(new TextEncoder().encode("started\n"));
+		},
+	});
+}
+
+function createBlockedChild<In extends TestStdin>(): ChildProcess<In> {
+	const { promise } = Promise.withResolvers<number>();
+
+	return {
+		stdout: createNeverClosingStream(),
+		stderr: undefined,
+		exited: promise,
+		[Symbol.dispose]() {},
+	} as unknown as ChildProcess<In>;
+}
+
+async function flushMicrotasks(count: number): Promise<void> {
+	for (let i = 0; i < count; i++) {
+		await Promise.resolve();
+	}
+}
+
+describe("executeSSH", () => {
+	afterEach(() => {
+		vi.restoreAllMocks();
+	});
+
+	it("returns promptly when an abort races a ControlMaster stream that stays open", async () => {
+		vi.spyOn(connectionManager, "ensureConnection").mockResolvedValue();
+		vi.spyOn(connectionManager, "buildRemoteCommand").mockResolvedValue(["remote", "sleep 60"]);
+		vi.spyOn(sshfsMount, "hasSshfs").mockReturnValue(false);
+		vi.spyOn(ptree, "spawn").mockImplementation(<In extends TestStdin>() => createBlockedChild<In>());
+
+		const chunked = Promise.withResolvers<void>();
+		const controller = new AbortController();
+		const resultPromise = executeSSH({ name: "remote", host: "remote" }, "sleep 60", {
+			signal: controller.signal,
+			onChunk: () => chunked.resolve(),
+		});
+		await chunked.promise;
+
+		let result: Awaited<typeof resultPromise> | undefined;
+		resultPromise.then(value => {
+			result = value;
+		});
+		controller.abort("user interrupt");
+		await flushMicrotasks(20);
+		expect(result).toBeDefined();
+		if (!result) return;
+		expect(result.cancelled).toBe(true);
+		expect(result.exitCode).toBeUndefined();
+		expect(result.output).toContain("Command aborted");
+	});
+});