fix(process): enforce remote command deadlines through pipe drain

apps/dokploy/__test__/server/exec-async-remote-lifecycle.test.ts

@@ -0,0 +1,211 @@
+import { type ChildProcess, execFileSync, spawn } from "node:child_process";
+import { findServerById } from "@dokploy/server/services/server";
+import {
+	ExecError,
+	execAsyncRemote,
+} from "@dokploy/server/utils/process/execAsync";
+import { type Connection, Server, utils } from "ssh2";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+
+vi.mock("@dokploy/server/services/server", () => ({ findServerById: vi.fn() }));
+
+const alive = (pid: number) => {
+	try {
+		process.kill(pid, 0);
+		return true;
+	} catch {
+		return false;
+	}
+};
+
+describe.skipIf(process.platform === "win32")(
+	"real loopback SSH process lifecycle",
+	() => {
+		const key = utils.generateKeyPairSync("ed25519");
+		let server: Server;
+		const connections: Connection[] = [];
+		const processes: ChildProcess[] = [];
+		const observedPids = new Set<number>();
+
+		const observeGroup = (group: number) => {
+			for (const line of execFileSync("ps", ["-axo", "pid=,pgid="], {
+				encoding: "utf8",
+			}).split("\n")) {
+				const [pid, pgid] = line.trim().split(/\s+/).map(Number);
+				if (pid && pgid === group) observedPids.add(pid);
+			}
+		};
+
+		beforeEach(async () => {
+			observedPids.clear();
+			server = new Server({ hostKeys: [key.private] }, (connection) => {
+				connections.push(connection);
+				connection.on("error", () => {});
+				connection.on("authentication", (context) => context.accept());
+				connection.on("ready", () =>
+					connection.on("session", (accept) => {
+						const session = accept();
+						// Ignore SSH signals deliberately: the remote deadline must survive transport loss.
+						session.on("signal", () => {});
+						session.on("exec", (acceptExec, _reject, info) => {
+							const channel = acceptExec();
+							const child = spawn("/bin/sh", ["-c", info.command], {
+								detached: true,
+								stdio: ["ignore", "pipe", "pipe"],
+							});
+							processes.push(child);
+							if (child.pid) observedPids.add(child.pid);
+							child.stdout?.on("data", (chunk: Buffer) => {
+								if (child.pid) observeGroup(child.pid);
+								if (!channel.destroyed) channel.write(chunk);
+							});
+							child.stderr?.on("data", (chunk: Buffer) => {
+								if (!channel.destroyed) channel.stderr.write(chunk);
+							});
+							child.on("close", (code, signal) => {
+								if (channel.destroyed) return;
+								if (signal) channel.exit(signal);
+								else channel.exit(code ?? 1);
+								channel.end();
+							});
+						});
+					}),
+				);
+			});
+			await new Promise<void>((resolve, reject) => {
+				server.once("error", reject);
+				server.listen(0, "127.0.0.1", resolve);
+			});
+			const address = server.address();
+			if (!address || typeof address === "string")
+				throw new Error("Missing fixture port");
+			vi.mocked(findServerById).mockResolvedValue({
+				sshKeyId: "fixture",
+				ipAddress: "127.0.0.1",
+				port: address.port,
+				username: "fixture",
+				sshKey: { privateKey: key.private },
+			} as never);
+		});
+
+		afterEach(async () => {
+			for (const child of processes.splice(0)) {
+				if (!child.pid) continue;
+				try {
+					process.kill(-child.pid, "SIGKILL");
+				} catch {
+					/* already gone */
+				}
+			}
+			for (const connection of connections.splice(0)) connection.end();
+			await new Promise<void>((resolve) => server.close(() => resolve()));
+		});
+
+		it("kills the remote shell, child, grandchild and watchdog after timeout even when signals are ignored", async () => {
+			const command =
+				"echo shell=$$; sh -c 'echo child=$$; sleep 30 & echo grandchild=$!; echo partial >&2; wait' & wait";
+			const started = Date.now();
+			await expect(
+				execAsyncRemote("fixture", command, undefined, { timeout: 2000 }),
+			).rejects.toSatisfy(
+				(error: unknown) =>
+					error instanceof ExecError &&
+					/timed out/.test(error.message) &&
+					(error.stdout?.includes("grandchild=") ?? false) &&
+					(error.stderr?.includes("partial") ?? false),
+			);
+			expect(Date.now() - started).toBeLessThan(3500);
+			expect(observedPids.size).toBeGreaterThanOrEqual(3);
+			await vi.waitFor(
+				() => {
+					expect([...observedPids].filter(alive)).toEqual([]);
+				},
+				{ timeout: 2000, interval: 30 },
+			);
+		});
+
+		it("retains remote cleanup after the SSH transport disconnects", async () => {
+			let disconnected = false;
+			const pending = execAsyncRemote(
+				"fixture",
+				"echo shell=$$; sleep 30 & echo grandchild=$!; wait",
+				() => {
+					if (disconnected) return;
+					disconnected = true;
+					connections[0]?.end();
+				},
+				{ timeout: 2000 },
+			);
+			await expect(pending).rejects.toBeInstanceOf(ExecError);
+			expect(observedPids.size).toBeGreaterThanOrEqual(3);
+			await vi.waitFor(
+				() => expect([...observedPids].filter(alive)).toEqual([]),
+				{ timeout: 2000, interval: 30 },
+			);
+		});
+
+		it.each(["1", "2"])(
+			"retains the deadline while an orphan grandchild holds output fd %s",
+			async (fd) => {
+				await expect(
+					execAsyncRemote(
+						"fixture",
+						`sleep 30 >&${fd} & echo grandchild=$!`,
+						undefined,
+						{ timeout: 2000 },
+					),
+				).rejects.toSatisfy(
+					(error: unknown) =>
+						error instanceof ExecError && /timed out/.test(error.message),
+				);
+				await vi.waitFor(
+					() => expect([...observedPids].filter(alive)).toEqual([]),
+					{ timeout: 2000, interval: 30 },
+				);
+			},
+		);
+
+		it.each(["exit 0", "exec sh -c 'exit 0'"])(
+			"does not leak the private exit status for %s",
+			async (command) => {
+				await expect(
+					execAsyncRemote("fixture", command, undefined, { timeout: 2000 }),
+				).resolves.toEqual({ stdout: "", stderr: "" });
+			},
+		);
+
+		it("preserves large interleaved stdout and stderr independently", async () => {
+			const command =
+				"i=0; while [ $i -lt 2048 ]; do printf 'stdout\\n'; printf 'stderr\\n' >&2; i=$((i+1)); done";
+			await expect(
+				execAsyncRemote("fixture", command, undefined, { timeout: 5000 }),
+			).resolves.toEqual({
+				stdout: "stdout\n".repeat(2048),
+				stderr: "stderr\n".repeat(2048),
+			});
+		});
+
+		it.each([0, 7])(
+			"clears every watchdog process on normal exit %i",
+			async (code) => {
+				const pending = execAsyncRemote(
+					"fixture",
+					`echo shell=$$; sleep 0.1; exit ${code}`,
+					undefined,
+					{ timeout: 30_000 },
+				);
+				if (code === 0)
+					await expect(pending).resolves.toMatchObject({ stderr: "" });
+				else
+					await expect(pending).rejects.toMatchObject({
+						exitCode: code,
+						stderr: "",
+					});
+				await vi.waitFor(
+					() => expect([...observedPids].filter(alive)).toEqual([]),
+					{ timeout: 1000 },
+				);
+			},
+		);
+	},
+);

apps/dokploy/__test__/server/exec-async-timeout.test.ts

@@ -11,6 +11,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 
 const { ssh, MockClient } = vi.hoisted(() => {
 	class MockStream {
+		signal = vi.fn();
 		close = vi.fn(() => {
 			this.closed = true;
 		});
@@ -27,14 +28,25 @@ const { ssh, MockClient } = vi.hoisted(() => {
 			return this;
 		}
 
+		stderrHandlers: Array<(...args: unknown[]) => void> = [];
 		stderr = {
-			on: (_event: string, _cb: (...args: unknown[]) => void) => this,
+			on: (_event: string, cb: (...args: unknown[]) => void) => {
+				this.stderrHandlers.push(cb);
+				return this;
+			},
 		};
+		emitStderr(data: string) {
+			for (const cb of this.stderrHandlers) cb(data);
+		}
 
 		emitClose(code: number) {
 			for (const cb of this.handlers.close ?? []) cb(code, null);
 		}
 
+		emitError(error: Error) {
+			for (const cb of this.handlers.error ?? []) cb(error);
+		}
+
 		emitData(data: string) {
 			for (const cb of this.handlers.data ?? []) cb(data);
 		}
@@ -45,6 +57,9 @@ const { ssh, MockClient } = vi.hoisted(() => {
 		destroyed = false;
 		execCommand = "";
 		execShouldFail: Error | null = null;
+		execShouldThrow: Error | null = null;
+		deferExec = false;
+		execCallback?: (err: Error | null, stream: MockStream) => void;
 		stream = new MockStream();
 		end = vi.fn(() => {
 			this.ended = true;
@@ -69,11 +84,17 @@ const { ssh, MockClient } = vi.hoisted(() => {
 		}
 
 		connect() {
+			if (ssh.connectError) throw ssh.connectError;
 			return this;
 		}
 
 		exec(command: string, cb: (err: Error | null, stream: MockStream) => void) {
 			this.execCommand = command;
+			if (this.execShouldThrow) throw this.execShouldThrow;
+			if (this.deferExec) {
+				this.execCallback = cb;
+				return;
+			}
 			if (this.execShouldFail) {
 				cb(this.execShouldFail, this.stream);
 				return;
@@ -90,7 +111,10 @@ const { ssh, MockClient } = vi.hoisted(() => {
 		}
 	}
 
-	const ssh = { instances: [] as MockClient[] };
+	const ssh = {
+		instances: [] as MockClient[],
+		connectError: null as Error | null,
+	};
 	return { ssh, MockClient };
 });
 
@@ -173,6 +197,7 @@ describe("execAsyncRemote timeout", () => {
 
 	beforeEach(() => {
 		ssh.instances.length = 0;
+		ssh.connectError = null;
 		vi.mocked(findServerById).mockReset();
 		vi.mocked(findServerById).mockResolvedValue(server as never);
 		vi.useFakeTimers();
@@ -212,9 +237,10 @@ describe("execAsyncRemote timeout", () => {
 		await flush();
 		const client = ssh.instances[0];
 		client?.emitReady();
-		expect(client?.execCommand).toBe("nvidia-smi");
+		expect(client?.execCommand).toContain("nvidia-smi");
 		await vi.advanceTimersByTimeAsync(1000);
 		await rejected;
+		expect(client?.stream.signal).toHaveBeenCalledWith("KILL");
 		expect(client?.stream.close).toHaveBeenCalledTimes(1);
 		expect(client?.end).toHaveBeenCalledTimes(1);
 		client?.stream.emitClose(0);
@@ -231,6 +257,7 @@ describe("execAsyncRemote timeout", () => {
 		client?.stream.emitClose(0);
 		await expect(pending).resolves.toEqual({ stdout: "", stderr: "" });
 		expect(vi.getTimerCount()).toBeLessThan(timersBeforeClose);
+		expect(vi.getTimerCount()).toBe(0);
 	});
 
 	it("resolves on success and does not reject later when the timer would have fired", async () => {
@@ -303,6 +330,103 @@ describe("execAsyncRemote timeout", () => {
 		expect(client?.end).toHaveBeenCalled();
 	});
 
+	it("cleans timers and clients when connect throws synchronously", async () => {
+		ssh.connectError = new Error("invalid key");
+		await expect(startRemote()).rejects.toMatchObject({
+			message: "SSH connection failed",
+		});
+		expect(vi.getTimerCount()).toBe(0);
+		expect(ssh.instances[0]?.destroy).toHaveBeenCalledOnce();
+	});
+
+	it("cleans timers and clients when exec throws synchronously", async () => {
+		const pending = startRemote();
+		const rejected = expect(pending).rejects.toMatchObject({
+			message: "Remote command execution failed",
+		});
+		await flush();
+		const client = ssh.instances[0];
+		if (client) client.execShouldThrow = new Error("channel unavailable");
+		client?.emitReady();
+		await rejected;
+		expect(vi.getTimerCount()).toBe(0);
+		expect(client?.destroy).toHaveBeenCalledOnce();
+	});
+
+	it("closes a late channel even when its signal method throws", async () => {
+		const pending = startRemote();
+		const rejected = expect(pending).rejects.toBeInstanceOf(ExecError);
+		await flush();
+		const client = ssh.instances[0];
+		if (client) client.deferExec = true;
+		client?.emitReady();
+		await vi.advanceTimersByTimeAsync(1000);
+		await rejected;
+		client?.stream.signal.mockImplementation(() => {
+			throw new Error("closed transport");
+		});
+		if (client) client.execCallback?.(null, client.stream);
+		expect(client?.stream.close).toHaveBeenCalledOnce();
+		expect(client?.stream.destroy).toHaveBeenCalledOnce();
+		expect(vi.getTimerCount()).toBe(0);
+	});
+
+	it("preserves both partial streams and ignores late output and errors", async () => {
+		const onData = vi.fn();
+		const pending = startRemote(onData);
+		const rejected = expect(pending).rejects.toMatchObject({
+			stdout: "out",
+			stderr: "err",
+		});
+		await flush();
+		const client = ssh.instances[0];
+		client?.emitReady();
+		client?.stream.emitData("out");
+		client?.stream.emitStderr("err");
+		await vi.advanceTimersByTimeAsync(1000);
+		await rejected;
+		client?.stream.emitData("late out");
+		client?.stream.emitStderr("late err");
+		client?.emitError(new Error("late error"));
+		client?.stream.emitError(new Error("late stream error"));
+		client?.stream.emitClose(0);
+		expect(onData.mock.calls).toEqual([["out"], ["err"]]);
+		expect(client?.end).toHaveBeenCalledTimes(1);
+		expect(client?.destroy).toHaveBeenCalledTimes(1);
+		expect(vi.getTimerCount()).toBe(0);
+	});
+
+	it.each([undefined, 0, -1, Number.NaN, Number.POSITIVE_INFINITY])(
+		"does not wrap commands without a positive finite timeout: %s",
+		async (timeout) => {
+			const pending = execAsyncRemote(
+				"server-1",
+				"original command",
+				undefined,
+				{ timeout },
+			);
+			await flush();
+			const client = ssh.instances[0];
+			client?.emitReady();
+			expect(client?.execCommand).toBe("original command");
+			expect(vi.getTimerCount()).toBe(0);
+			client?.stream.emitClose(0);
+			await pending;
+			expect(client?.stream.signal).not.toHaveBeenCalled();
+		},
+	);
+
+	it("keeps a normal nonzero exit distinct from timeout", async () => {
+		const pending = startRemote();
+		const rejected = expect(pending).rejects.toMatchObject({ exitCode: 7 });
+		await flush();
+		const client = ssh.instances[0];
+		client?.emitReady();
+		client?.stream.emitClose(7);
+		await rejected;
+		expect(vi.getTimerCount()).toBe(0);
+	});
+
 	it("still delivers onData chunks if the command hangs afterward", async () => {
 		const chunks: string[] = [];
 		const pending = startRemote((data) => {

packages/server/src/utils/process/execAsync.ts

@@ -1,8 +1,9 @@
 import { type ChildProcess, exec, execFile, spawn } from "node:child_process";
 import util from "node:util";
 import { findServerById } from "@dokploy/server/services/server";
-import { Client } from "ssh2";
+import { Client, type ClientChannel } from "ssh2";
 import { ExecError } from "./ExecError";
+import { withRemoteDeadline } from "./remote-deadline";
 
 export class WriteFileRemoteError extends Error {
 	constructor(
@@ -129,7 +130,7 @@ export const execAsync = async (
 	options?: ExecAsyncOptions,
 ): Promise<{ stdout: string; stderr: string }> => {
 	const timeout = options?.timeout;
-	if (timeout != null && timeout > 0) {
+	if (timeout != null && Number.isFinite(timeout) && timeout > 0) {
 		return execAsyncTimed(command, { ...options, timeout });
 	}
 
@@ -271,13 +272,40 @@ export const execAsyncRemote = async (
 
 	let stdout = "";
 	let stderr = "";
-	const timeoutMs = options?.timeout;
+	const requestedTimeout = options?.timeout;
+	const timeoutMs =
+		requestedTimeout != null &&
+		Number.isFinite(requestedTimeout) &&
+		requestedTimeout > 0
+			? requestedTimeout
+			: undefined;
+	const started = Date.now();
 	return new Promise((resolve, reject) => {
 		const conn = new Client();
-		let stream: { close: () => void; destroy: () => void } | undefined;
+		let stream: ClientChannel | undefined;
 		let settled = false;
 		let timer: ReturnType<typeof setTimeout> | undefined;
 
+		const closeStream = (channel?: ClientChannel) => {
+			if (timeoutMs !== undefined) {
+				try {
+					channel?.signal("KILL");
+				} catch {
+					/* remote watchdog owns cleanup */
+				}
+			}
+			try {
+				channel?.close();
+			} catch {
+				/* already closed */
+			}
+			try {
+				channel?.destroy();
+			} catch {
+				/* already destroyed */
+			}
+		};
+
 		const settle = (
 			error?: ExecError,
 			value?: { stdout: string; stderr: string },
@@ -289,16 +317,7 @@ export const execAsyncRemote = async (
 				timer = undefined;
 			}
 			if (error) {
-				try {
-					stream?.close();
-				} catch {
-					// already closed
-				}
-				try {
-					stream?.destroy();
-				} catch {
-					// already destroyed
-				}
+				closeStream(stream);
 				try {
 					conn.end();
 				} catch {
@@ -333,110 +352,158 @@ export const execAsyncRemote = async (
 			}, timeoutMs);
 		}
 
-		sleep(1000);
-		conn
-			.once("ready", () => {
-				if (settled) return;
-				conn.exec(command, (err, commandStream) => {
-					if (settled) {
-						try {
-							commandStream?.close();
-						} catch {
-							// already closed
-						}
-						return;
-					}
-					if (err) {
-						onData?.(err.message);
+		try {
+			conn
+				.once("ready", () => {
+					if (settled) return;
+					const remoteCommand =
+						timeoutMs === undefined
+							? command
+							: withRemoteDeadline(
+									command,
+									Math.max(1, timeoutMs - (Date.now() - started)),
+								);
+					try {
+						conn.exec(remoteCommand, (err, commandStream) => {
+							if (settled) {
+								closeStream(commandStream);
+								return;
+							}
+							if (err) {
+								onData?.(err.message);
+								settle(
+									new ExecError(
+										`Remote command execution failed: ${err.message}`,
+										{
+											command,
+											serverId,
+											originalError: err,
+										},
+									),
+								);
+								return;
+							}
+							stream = commandStream;
+							commandStream
+								.on("error", (error: Error) => {
+									if (settled) return;
+									settle(
+										new ExecError(
+											`Remote command stream failed: ${error.message}`,
+											{
+												command,
+												stdout,
+												stderr,
+												serverId,
+												originalError: error,
+											},
+										),
+									);
+								})
+								.on("close", (code: number, _signal: string) => {
+									if (code === 0) {
+										settle(undefined, { stdout, stderr });
+									} else {
+										settle(
+											new ExecError(
+												`Remote command failed with exit code ${code}`,
+												{
+													command,
+													stdout,
+													stderr,
+													exitCode: code,
+													serverId,
+												},
+											),
+										);
+									}
+								})
+								.on("data", (data: string) => {
+									if (settled) return;
+									stdout += data.toString();
+									onData?.(data.toString());
+								})
+								.stderr.on("data", (data) => {
+									if (settled) return;
+									stderr += data.toString();
+									onData?.(data.toString());
+								});
+						});
+					} catch (error) {
 						settle(
-							new ExecError(`Remote command execution failed: ${err.message}`, {
+							new ExecError("Remote command execution failed", {
+								command,
+								stdout,
+								stderr,
+								serverId,
+								originalError: error instanceof Error ? error : undefined,
+							}),
+						);
+					}
+				})
+				.on("error", (err) => {
+					if (settled) return;
+					if (err.level === "client-authentication") {
+						const technicalDetail = `Error: ${err.message} ${err.level}`;
+						const friendlyMessage = [
+							"",
+							"❌ Couldn't connect to your server — the SSH key was not accepted.",
+							"",
+							"This usually means the key doesn't match what's on the server, or the key format is invalid.",
+							"",
+							`Technical details: ${technicalDetail}`,
+							"",
+							"💡 Hints:",
+							"  • Check that the SSH key you added in Dokploy is the same one installed on the server (e.g. in ~/.ssh/authorized_keys).",
+							"  • Try generating a new SSH key in Dokploy and add only the public key to the server, then try again.",
+							"  • Make sure to follow the instructions on the Setup Server Button on the SSH Keys tab and then click on deployments tab and check the logs for more details.",
+						].join("\n");
+						onData?.(friendlyMessage);
+						settle(
+							new ExecError(
+								`Authentication failed: Invalid SSH private key. ${friendlyMessage}`,
+								{
+									command,
+									serverId,
+									originalError: err,
+								},
+							),
+						);
+					} else {
+						const errorMsg = `SSH connection error: ${err.message}`;
+						onData?.(errorMsg);
+						settle(
+							new ExecError(errorMsg, {
+								stdout,
+								stderr,
 								command,
 								serverId,
 								originalError: err,
 							}),
 						);
-						return;
 					}
-					stream = commandStream;
-					commandStream
-						.on("close", (code: number, _signal: string) => {
-							if (code === 0) {
-								settle(undefined, { stdout, stderr });
-							} else {
-								settle(
-									new ExecError(
-										`Remote command failed with exit code ${code}`,
-										{
-											command,
-											stdout,
-											stderr,
-											exitCode: code,
-											serverId,
-										},
-									),
-								);
-							}
-						})
-						.on("data", (data: string) => {
-							stdout += data.toString();
-							onData?.(data.toString());
-						})
-						.stderr.on("data", (data) => {
-							stderr += data.toString();
-							onData?.(data.toString());
-						});
+				})
+				.connect({
+					host: server.ipAddress,
+					port: server.port,
+					username: server.username,
+					privateKey: server.sshKey?.privateKey,
+					timeout: timeoutMs != null && timeoutMs > 0 ? timeoutMs : 99999,
+					...(timeoutMs != null && timeoutMs > 0
+						? { readyTimeout: timeoutMs }
+						: {}),
 				});
-			})
-			.on("error", (err) => {
-				if (settled) return;
-				if (err.level === "client-authentication") {
-					const technicalDetail = `Error: ${err.message} ${err.level}`;
-					const friendlyMessage = [
-						"",
-						"❌ Couldn't connect to your server — the SSH key was not accepted.",
-						"",
-						"This usually means the key doesn't match what's on the server, or the key format is invalid.",
-						"",
-						`Technical details: ${technicalDetail}`,
-						"",
-						"💡 Hints:",
-						"  • Check that the SSH key you added in Dokploy is the same one installed on the server (e.g. in ~/.ssh/authorized_keys).",
-						"  • Try generating a new SSH key in Dokploy and add only the public key to the server, then try again.",
-						"  • Make sure to follow the instructions on the Setup Server Button on the SSH Keys tab and then click on deployments tab and check the logs for more details.",
-					].join("\n");
-					onData?.(friendlyMessage);
-					settle(
-						new ExecError(
-							`Authentication failed: Invalid SSH private key. ${friendlyMessage}`,
-							{
-								command,
-								serverId,
-								originalError: err,
-							},
-						),
-					);
-				} else {
-					const errorMsg = `SSH connection error: ${err.message}`;
-					onData?.(errorMsg);
-					settle(
-						new ExecError(errorMsg, {
-							command,
-							serverId,
-							originalError: err,
-						}),
-					);
-				}
-			})
-			.connect({
-				host: server.ipAddress,
-				port: server.port,
-				username: server.username,
-				privateKey: server.sshKey?.privateKey,
-				timeout: timeoutMs != null && timeoutMs > 0 ? timeoutMs : 99999,
-				...(timeoutMs != null && timeoutMs > 0
-					? { readyTimeout: timeoutMs }
-					: {}),
-			});
+		} catch (error) {
+			settle(
+				new ExecError("SSH connection failed", {
+					command,
+					stdout,
+					stderr,
+					serverId,
+					originalError: error instanceof Error ? error : undefined,
+				}),
+			);
+		}
 	});
 };
 

packages/server/src/utils/process/remote-deadline.ts

@@ -0,0 +1,39 @@
+const quote = (value: string) => `'${value.replaceAll("'", "'\\''")}'`;
+
+/** OpenSSH gives each non-PTY command its own process group, including its children. */
+export const withRemoteDeadline = (
+	command: string,
+	timeout: number,
+): string => {
+	const seconds = timeout / 1000;
+	// The status pipe is private; output relays keep the deadline alive until inherited pipes drain.
+	return `exec sh -c ${quote(`
+kill -0 -$$ 2>/dev/null || { echo 'Timed SSH command requires an isolated process group' >&2; exit 125; }
+command -v cat >/dev/null 2>&1 || { echo 'Timed SSH command requires cat' >&2; exit 125; }
+if sleep 0.001 2>/dev/null; then
+ duration=${seconds}
+elif sleep 0 2>/dev/null; then
+ duration=${Math.ceil(seconds)}
+else
+ echo 'Timed SSH command requires sleep' >&2
+ exit 125
+fi
+sleep "$duration" &
+deadline=$!
+(
+ result=$(
+  exec 3>&1
+  {
+   { sh -c ${quote(command)} 3>&- 4>&- 5>&-; printf '%s' "$?" >&3; } | cat >&4
+  } 2>&1 | cat >&5
+ )
+ kill "$deadline" 2>/dev/null
+ exit "$result"
+) 4>&1 5>&2 &
+command_pid=$!
+if wait "$deadline" 2>/dev/null; then
+ kill -s KILL 0
+fi
+wait "$command_pid"
+`)}`;
+};