fix(messaging): reserve SSH forward ports until handoff and recover lost candidates

docs-site/src/content/docs/reference/cli/messaging.md

@@ -246,7 +246,11 @@ hold twelve authenticated connections, four more stay reserved for the owner's o
 control, outgoing probes and renewal, and at most eight tunnel
 children and two transient control/inspection helpers, ten total. Setup is bounded by
 30 seconds across the whole selected peer batch. A final lease renewal precedes startup
-success; final and periodic renewal batches have five-second deadlines, within the setup
+success. If SSH reports that a selected forward port was lost, setup closes the owned
+pair and selects fresh ports, up to three attempts within that same deadline. SSH forward
+confirmation, loopback checks and authenticated peer readiness must all succeed. Other
+setup failures stop immediately; published routes do not reconnect automatically. Final
+and periodic renewal batches have five-second deadlines, within the setup
 deadline where applicable, and periodic maintenance runs every ten seconds.
 Cancellation stops only owned
 SSH helpers/listeners; it leaves Codex daemons, threads and unrelated proxies alone.

scripts/test-layout/layout-codex-collision.json

@@ -1,4 +1,6 @@
 {
+  "messaging-remote-port-reservation.test.ts": "codex-integration",
+  "messaging-remote-forward-failure.test.ts": "codex-integration",
   "codex-cli-windows-child.test.ts": "codex-integration",
   "codex-cli-update-windows-layout.test.ts": "codex-integration",
   "codex-cli-windows-plan.test.ts": "codex-integration",

src/messaging/remote-forward-failure.ts

@@ -0,0 +1,35 @@
+/** Match fixed SSH bind-failure tokens without retaining any diagnostic bytes. */
+export function remoteForwardFailure(argv: readonly string[], onReady: () => void) {
+  const direction = argv.includes("-L") ? "L" : "R";
+  const index = argv.indexOf(`-${direction}`), port = argv[index + 1]?.split(":")[1];
+  const enabled = index >= 0 && argv.includes("ExitOnForwardFailure=yes") && /^\d+$/.test(port ?? "");
+  const failures = !enabled ? []
+    : direction === "L" ? ["Could not request local forwarding."]
+    : [`remote port forwarding failed for listen port ${port}\r`, `remote port forwarding failed for listen port ${port}\n`];
+  const target = argv[index + 1]?.split(":").slice(2).join(":");
+  // OpenSSH emits the local-listening announcement BEFORE bind(), so it is not a success receipt.
+  const success = direction === "L" ? "debug1: Entering interactive session."
+    : `debug1: remote forward success for: listen 127.0.0.1:${port}, connect ${target}`;
+  const successes = !enabled ? [] : [`${success}\r`, `${success}\n`];
+  const matches = [...failures, ...successes].map((token, index) => {
+    const bytes = new TextEncoder().encode(token), prefixes = new Uint32Array(bytes.length);
+    for (let i = 1, length = 0; i < bytes.length; i++) {
+      while (length && bytes[i] !== bytes[length]) length = prefixes[length - 1]!;
+      if (bytes[i] === bytes[length]) length++;
+      prefixes[i] = length;
+    }
+    return { bytes, prefixes, length: 0, success: index >= failures.length };
+  });
+  let failed = false, ready = false;
+  return { enabled, get failed() { return failed; }, get ready() { return ready; }, write(chunk: Uint8Array) {
+    if (!enabled || failed || ready) return;
+    for (const byte of chunk) for (const match of matches) {
+      while (match.length && byte !== match.bytes[match.length]) match.length = match.prefixes[match.length - 1]!;
+      if (byte === match.bytes[match.length]) match.length++;
+      if (match.length === match.bytes.length) {
+        if (match.success) { ready = true; onReady(); } else failed = true;
+        return;
+      }
+    }
+  } };
+}

src/messaging/remote-ports.ts

@@ -8,12 +8,34 @@ type RemoteProcFile = {
   close(): Promise<void>;
 };
 
-/** Select a candidate only; actual forwarding and authenticated readiness must still succeed. */
+export type RemotePortAllocator = () => { port: number; stop(force?: boolean): void };
+export interface RemotePortReservation { port: number; handoff(): void; release(): void }
+const pendingPorts = new Set<number>();
+const PORT_ALLOCATION_ATTEMPTS = 8;
+const allocateRemotePort: RemotePortAllocator = () => Bun.listen({ hostname: "127.0.0.1", port: 0,
+  exclusive: true, socket: { data(socket) { socket.end(); } } });
+
+/** Hold the socket until handoff, and the process-local claim until the owned attempt closes. */
+export function reserveRemotePort(allocate = allocateRemotePort, excluded: ReadonlySet<number> = new Set()): RemotePortReservation {
+  for (let attempt = 0; attempt < PORT_ALLOCATION_ATTEMPTS; attempt++) {
+    const listener = allocate(), port = listener.port;
+    if (!validPort(port)) { listener.stop(true); throw remoteError("invalid_port", "Cannot allocate an unprivileged messaging port candidate."); }
+    if (pendingPorts.has(port) || excluded.has(port)) { listener.stop(true); continue; }
+    pendingPorts.add(port);
+    let held = true, released = false;
+    const handoff = () => { if (held) { held = false; listener.stop(true); } };
+    return { port, handoff, release() {
+      if (released) return;
+      released = true; handoff(); pendingPorts.delete(port);
+    } };
+  }
+  throw remoteError("forward_port_lost", "Cannot allocate a distinct messaging forward port.");
+}
+
+/** A remote stdio helper cannot transfer its socket to sshd; the initiating owner checks SSH failure. */
 export function remotePortCandidate(): number {
-  const listener = Bun.listen({ hostname: "127.0.0.1", port: 0, socket: { data(socket) { socket.end(); } } });
-  const port = listener.port; listener.stop(true);
-  if (!validPort(port)) throw remoteError("invalid_port", "Cannot allocate an unprivileged messaging port candidate.");
-  return port;
+  const reservation = reserveRemotePort();
+  try { return reservation.port; } finally { reservation.release(); }
 }
 /** Refuse GatewayPorts/public binds; neither port existence nor this scan proves endpoint identity. */
 export async function remotePortIsLoopback(port: number, budget: MessageBudget, capacity: RemoteCapacity,

src/messaging/remote-process.ts

@@ -2,9 +2,14 @@ import { linkSshSpawnEnv } from "../link/ssh-runner";
 import { MessageBudget } from "./budget";
 import { RemoteCapacity, remoteError, REMOTE_LIMITS } from "./remote-contract";
 import { LocalMessagingError } from "./types";
+import { remoteForwardFailure } from "./remote-forward-failure";
 
 export interface RemoteHelper {
   readonly pid: number; readonly exited: Promise<number>; readonly output: Promise<string>;
+  readonly forwardReady: Promise<void>;
+  readonly forwardFailed?: boolean;
+  /** After cleanup: intrinsic I/O or pre-cleanup exit failure, excluding cleanup-induced exits. */
+  readonly setupFailure?: Error;
   close(): Promise<void>;
 }
 /** Narrow subprocess seam for offline lifecycle tests; production always uses owned Bun children. */
@@ -24,13 +29,21 @@ export function spawnRemoteHelper(argv: readonly string[], capacity: RemoteCapac
   stdin?: string, env = linkSshSpawnEnv(), options: RemoteHelperOptions = {}): RemoteHelper {
   if (signal?.aborted) throw remoteError("cancelled", "Messaging helper was cancelled before spawn.");
   const release = capacity.reserveHelper(options.purpose ?? "persistent");
+  let ready!: () => void, notReady!: (reason: unknown) => void;
+  const forwardReady = new Promise<void>((resolve, reject) => { ready = resolve; notReady = reject; });
+  void forwardReady.catch(() => {});
+  const forwardFailure = remoteForwardFailure(argv, ready);
+  if (!forwardFailure.enabled) ready();
+  // Command-line options are first-value-wins; only owned forwards request bounded token inspection.
+  const spawnArgv = forwardFailure.enabled ? [argv[0]!, "-o", "LogLevel=DEBUG1", ...argv.slice(1)] : [...argv];
   let child: RemoteHelperProcess;
-  try { child = options.spawn ? options.spawn(argv, env)
-    : Bun.spawn([...argv], { stdin: "pipe", stdout: "pipe", stderr: "pipe", detached: true, env }); }
+  try { child = options.spawn ? options.spawn(spawnArgv, env)
+    : Bun.spawn(spawnArgv, { stdin: "pipe", stdout: "pipe", stderr: "pipe", detached: true, env }); }
   catch { release(); throw remoteError("ssh_unavailable", "The messaging SSH helper could not start."); }
   const readers = [child.stdout.getReader(), child.stderr.getReader()];
   let closing: Promise<void> | undefined;
   let failure: Error | undefined;
+  let interrupted = false;
   const cancelledPipes = [false, false];
   const pipeEnds: (() => void)[] = [];
   const naturalPipes = readers.map(() => new Promise<void>(resolve => { pipeEnds.push(resolve); }));
@@ -45,6 +58,7 @@ export function spawnRemoteHelper(argv: readonly string[], capacity: RemoteCapac
   };
   const close = (): Promise<void> => closing ??= (async () => {
     signal?.removeEventListener("abort", abort);
+    interrupted = child.exitCode === null;
     signalGroup("SIGTERM");
     const force = setTimeout(() => signalGroup("SIGKILL"), 500);
     let timer: ReturnType<typeof setTimeout> | undefined;
@@ -65,6 +79,7 @@ export function spawnRemoteHelper(argv: readonly string[], capacity: RemoteCapac
         cancelledPipes[index] = true; void reader.cancel().catch(() => {});
       }
       stopOutput(cleanupFailure ?? failure ?? remoteError("ssh_incomplete", "Messaging helper output capture was stopped."));
+      notReady(cleanupFailure ?? failure ?? remoteError(child.exitCode !== null && child.exitCode !== 0 && forwardFailure.failed ? "forward_port_lost" : "ssh_failed", "Messaging SSH forward was not confirmed."));
       // Cancellation cannot prove natural EOF; unresolved ownership keeps its helper capacity charged.
     }
   })();
@@ -78,7 +93,8 @@ export function spawnRemoteHelper(argv: readonly string[], capacity: RemoteCapac
       for (;;) {
         const next = await readers[index]!.read();
         if (next.done) { if (!cancelledPipes[index]) pipeEnds[index]!(); break; }
-        if (index === 1 || discarding) continue; // stderr is never retained or returned.
+        if (index === 1) { forwardFailure.write(next.value); continue; } // No diagnostic bytes are retained or returned.
+        if (discarding) continue;
         try {
           bytes += next.value.byteLength;
           if (bytes > 64 * 1024) throw remoteError("ssh_output_limit", "Messaging helper output exceeded its limit.");
@@ -107,8 +123,18 @@ export function spawnRemoteHelper(argv: readonly string[], capacity: RemoteCapac
     if (failure) throw failure; return out;
   })]);
   void output.catch(() => {});
+  void output.then(async () => {
+    const code = await child.exited;
+    if (!forwardFailure.ready) notReady(remoteError(code !== 0 && forwardFailure.failed ? "forward_port_lost" : "ssh_failed", "Messaging SSH forward was not confirmed."));
+  }, notReady).catch(notReady);
   void child.exited.then(() => { signal?.removeEventListener("abort", abort); }).catch(() => {});
-  return { pid: child.pid, exited: child.exited, output, close };
+  return { pid: child.pid, exited: child.exited, output, forwardReady, get forwardFailed() { return forwardFailure.failed; },
+    get setupFailure() {
+      if (failure) return failure;
+      if (!interrupted && child.exitCode !== null && (child.exitCode !== 0 || !forwardFailure.ready)) {
+        return remoteError(child.exitCode !== 0 && forwardFailure.failed ? "forward_port_lost" : "ssh_failed", "Messaging SSH forward setup failed.");
+      }
+    }, close };
 }
 
 /** One bounded batch covers helper startup, stdin, exit and output; errors never include raw output. */