Handle SSH input pipe failures without crashing the controller

packages/adapter-utils/src/ssh-fixture.test.ts

@@ -6,6 +6,7 @@ import path from "node:path";
 import { afterAll, afterEach, describe, expect, it } from "vitest";
 import {
   buildSshSpawnTarget,
+  createSshCommandManagedRuntimeRunner,
   buildSshEnvLabFixtureConfig,
   getSshEnvLabSupport,
   prepareWorkspaceForSshExecution,
@@ -181,6 +182,28 @@ describe("ssh env-lab fixture", () => {
   // this drains it too instead of stranding a listener until the process exits.
   afterAll(drainFixtureTeardowns);
 
+  it.each([0, 7])("handles a closed SSH input pipe without losing exit diagnostics (exit %i)", async (exitCode) => {
+    const rootDir = await createFixtureRootDir();
+    const previousPath = process.env.PATH;
+    await writeFile(path.join(rootDir, "ssh"), `#!/bin/sh
+exec 0<&-
+printf 'remote rejected upload' >&2
+sleep 0.05
+exit ${exitCode}
+`, { mode: 0o700 });
+    process.env.PATH = `${rootDir}${path.delimiter}${previousPath ?? ""}`;
+    try {
+      const runner = createSshCommandManagedRuntimeRunner({ spec: UNREACHABLE_SSH_SPEC });
+      const result = await runner.execute({ command: "cat", stdin: "x".repeat(8 * 1024 * 1024), timeoutMs: 5000 });
+      expect(result.exitCode).toBe(exitCode === 0 ? null : exitCode);
+      expect(result.stderr).toBe("remote rejected upload");
+      expect(result.timedOut).toBe(false);
+    } finally {
+      if (previousPath === undefined) delete process.env.PATH;
+      else process.env.PATH = previousPath;
+    }
+  });
+
   it("starts an isolated sshd fixture and executes commands through it", async () => {
     const rootDir = await createFixtureRootDir();
     const statePath = path.join(rootDir, "state.json");

packages/adapter-utils/src/ssh.ts

@@ -239,6 +239,7 @@ async function spawnText(
     let stderr = "";
     let settled = false;
     let timedOut = false;
+    let stdinError: Error | null = null;
 
     const finishReject = (error: Error & { stdout?: string; stderr?: string; code?: number | null; killed?: boolean }) => {
       if (settled) return;
@@ -308,21 +309,32 @@ async function spawnText(
       clearTimers();
       if (settled) return;
       settled = true;
-      if (code === 0) {
+      if (code === 0 && !stdinError) {
         resolve({ stdout, stderr });
         return;
       }
-      reject(Object.assign(new Error(stderr.trim() || stdout.trim() || `Process exited with code ${code ?? -1}`), {
+      const inputFailure = stdinError ? "Process closed stdin before command input completed." : "";
+      reject(Object.assign(new Error(stderr.trim() || inputFailure || stdout.trim() || `Process exited with code ${code ?? -1}`), {
         stdout,
-        stderr,
-        code,
+        stderr: stderr || inputFailure,
+        code: code === 0 && stdinError ? null : code,
         signal,
         killed: timedOut,
       }));
     });
 
     if (options.stdin != null && child.stdin) {
-      child.stdin.end(options.stdin);
+      // SSH may exit or close its pipe while a multi-megabyte upload is still
+      // queued. A stream error is independent of the ChildProcess error event.
+      // Observe it before writing, then preserve the remote exit diagnostics
+      // when close arrives instead of letting EPIPE crash the controller.
+      child.stdin.on("error", (error) => { stdinError = error; });
+      try {
+        child.stdin.end(options.stdin);
+      } catch (error) {
+        stdinError = error instanceof Error ? error : new Error(String(error));
+        child.stdin.destroy();
+      }
     }
   });
 }