diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index bef9ae4ac..c17372cb0 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/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 diff --git a/packages/coding-agent/src/ssh/ssh-executor.ts b/packages/coding-agent/src/ssh/ssh-executor.ts index 3dd6f5b6d..6e3a7be28 100644 --- a/packages/coding-agent/src/ssh/ssh-executor.ts +++ b/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 | undefined; cleanup: () => void } { + if (!signal) { + return { promise: undefined, cleanup: () => {} }; + } + + const { promise, resolve } = Promise.withResolvers(); + const onAbort = () => { + const error = new ptree.AbortError(signal.reason, ""); + 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(); } } diff --git a/packages/coding-agent/test/ssh/ssh-executor.test.ts b/packages/coding-agent/test/ssh/ssh-executor.test.ts new file mode 100644 index 000000000..085c32531 --- /dev/null +++ b/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 { + return new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("started\n")); + }, + }); +} + +function createBlockedChild(): ChildProcess { + const { promise } = Promise.withResolvers(); + + return { + stdout: createNeverClosingStream(), + stderr: undefined, + exited: promise, + [Symbol.dispose]() {}, + } as unknown as ChildProcess; +} + +async function flushMicrotasks(count: number): Promise { + 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(() => createBlockedChild()); + + const chunked = Promise.withResolvers(); + 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 | 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"); + }); +});