From e230638b1d52127976bec1e7a3713843cb8af02a Mon Sep 17 00:00:00 2001
From: JUN <243035832+lidge-jun@users.noreply.github.com>
Date: Wed, 28 Oct 2026 16:45:05 +0900
Subject: [PATCH 1/2] fix(messaging): reserve SSH forward ports and recover
 lost candidates

---
 .../content/docs/reference/cli/messaging.md   |   6 +-
 .../test-layout/layout-codex-collision.json   |   2 +
 src/messaging/remote-forward-failure.ts       |  35 +++
 src/messaging/remote-ports.ts                 |  32 ++-
 src/messaging/remote-process.ts               |  24 +-
 src/messaging/remote-tunnels.ts               |  86 +++++--
 structure/remote-messaging.md                 |  13 ++
 .../messaging-remote-enrollment-gaps.test.ts  |   2 +-
 .../messaging-remote-forward-failure.test.ts  |  39 ++++
 .../messaging-remote-lifecycle.test.ts        |   6 +-
 .../messaging-remote-port-reservation.test.ts | 213 ++++++++++++++++++
 tests/fixtures/test-layout-expected.json      |   2 +
 12 files changed, 431 insertions(+), 29 deletions(-)
 create mode 100644 src/messaging/remote-forward-failure.ts
 create mode 100644 tests/codex-integration/messaging-remote-forward-failure.test.ts
 create mode 100644 tests/codex-integration/messaging-remote-port-reservation.test.ts

diff --git a/docs-site/src/content/docs/reference/cli/messaging.md b/docs-site/src/content/docs/reference/cli/messaging.md
index b64b25e7255..ef90c846e13 100644
--- a/docs-site/src/content/docs/reference/cli/messaging.md
+++ b/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.
diff --git a/scripts/test-layout/layout-codex-collision.json b/scripts/test-layout/layout-codex-collision.json
index cabb852d35d..3c103f3fc3d 100644
--- a/scripts/test-layout/layout-codex-collision.json
+++ b/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",
diff --git a/src/messaging/remote-forward-failure.ts b/src/messaging/remote-forward-failure.ts
new file mode 100644
index 00000000000..6df150bf8bd
--- /dev/null
+++ b/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;
+      }
+    }
+  } };
+}
diff --git a/src/messaging/remote-ports.ts b/src/messaging/remote-ports.ts
index 36746e36b2a..0f4e6d7be2e 100644
--- a/src/messaging/remote-ports.ts
+++ b/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,
diff --git a/src/messaging/remote-process.ts b/src/messaging/remote-process.ts
index ff5cf2a82a3..8058dd1831f 100644
--- a/src/messaging/remote-process.ts
+++ b/src/messaging/remote-process.ts
@@ -2,9 +2,12 @@ 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;
   close(): Promise<void>;
 }
 /** Narrow subprocess seam for offline lifecycle tests; production always uses owned Bun children. */
@@ -24,9 +27,16 @@ 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;
@@ -65,6 +75,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 +89,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 +119,12 @@ 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; }, close };
 }
 
 /** One bounded batch covers helper startup, stdin, exit and output; errors never include raw output. */
diff --git a/src/messaging/remote-tunnels.ts b/src/messaging/remote-tunnels.ts
index fd45c82b8c4..8da073cd17e 100644
--- a/src/messaging/remote-tunnels.ts
+++ b/src/messaging/remote-tunnels.ts
@@ -3,7 +3,7 @@ import { buildExecArgv, buildTunnelArgv, remoteOcxArgv } from "../link/ssh-argv"
 import { MessageBudget } from "./budget";
 import { exactRecord, RemoteCapacity, remoteError, validPort, type RemotePeer } from "./remote-contract";
 import { withRemoteLock, writeRemoteFile } from "./remote-files";
-import { remotePortCandidate, remotePortIsLoopback } from "./remote-ports";
+import { reserveRemotePort, remotePortIsLoopback, type RemotePortAllocator, type RemotePortReservation } from "./remote-ports";
 import { runRemoteHelper, spawnRemoteHelper, type RemoteHelper } from "./remote-process";
 import type { RemoteMessageStore } from "./remote-store";
 
@@ -14,6 +14,8 @@ export interface RemoteTunnelPair {
 export interface RemoteTunnelDeps {
   run?: typeof runRemoteHelper;
   spawn?: typeof spawnRemoteHelper;
+  allocatePort?: RemotePortAllocator;
+  inspect?: typeof remotePortIsLoopback;
 }
 /** Own one -L/-R pair with shared SSH policy and one setup deadline; never adopt/kill existing tunnels. */
 export async function startRemoteTunnels(store: RemoteMessageStore, peer: RemotePeer, budget: MessageBudget,
@@ -21,35 +23,84 @@ export async function startRemoteTunnels(store: RemoteMessageStore, peer: Remote
   if (!peer.ssh || !peer.hostKey) throw remoteError("return_route_only", "This peer has no initiating SSH enrollment; wait for its return route.");
   const knownHosts = join(store.directory, `${peer.machine.id}.known_hosts`);
   withRemoteLock(store.directory, () => writeRemoteFile(knownHosts, peer.hostKey!));
-  const result = exactRecord(JSON.parse(await (deps.run ?? runRemoteHelper)(buildExecArgv({ alias: peer.ssh,
-    knownHostsFile: knownHosts, argv: remoteOcxArgv(["message", "_port"]) }), budget, capacity)), ["port"]);
-  if (!validPort(result.port)) throw remoteError("remote_incompatible", "Remote node did not offer an unprivileged return port.");
-  const localPort = remotePortCandidate(), returnPort = result.port, helpers: RemoteHelper[] = [];
+  return startRemoteForwardPair({ ssh: peer.ssh, knownHosts, peerPort: peer.port, ownerPort: store.requireEnabled().port },
+    budget, capacity, ownerSignal, deps);
+}
+
+interface RemoteForwardOptions { ssh: string; knownHosts: string; peerPort: number; ownerPort: number }
+const FORWARD_SETUP_ATTEMPTS = 3;
+
+/** Transport-only setup: no messages or authenticated-ready state are published here. */
+export async function startRemoteForwardPair(options: RemoteForwardOptions, budget: MessageBudget,
+  capacity: RemoteCapacity, ownerSignal: AbortSignal, deps: RemoteTunnelDeps = {}): Promise<RemoteTunnelPair> {
+  const localPorts = new Set<number>(), returnPorts = new Set<number>();
+  for (let attempt = 0; attempt < FORWARD_SETUP_ATTEMPTS; attempt++) {
+    budget.throwIfEnded();
+    if (ownerSignal.aborted) throw remoteError("cancelled", "Messaging tunnel setup was cancelled.");
+    const result = exactRecord(JSON.parse(await (deps.run ?? runRemoteHelper)(buildExecArgv({ alias: options.ssh,
+      knownHostsFile: options.knownHosts, argv: remoteOcxArgv(["message", "_port"]) }), budget, capacity)), ["port"]);
+    budget.throwIfEnded();
+    if (!validPort(result.port)) throw remoteError("remote_incompatible", "Remote node did not offer an unprivileged return port.");
+    if (returnPorts.has(result.port)) continue;
+    returnPorts.add(result.port);
+    const reservation = reserveRemotePort(deps.allocatePort, localPorts);
+    localPorts.add(reservation.port);
+    try { return await startAttempt(options, reservation, result.port, budget, capacity, ownerSignal, deps); }
+    catch (error) { if ((error as { code?: string })?.code !== "forward_port_lost") throw error; }
+  }
+  throw remoteError("forward_port_lost", "Messaging SSH forward ports were lost during bounded setup.");
+}
+
+async function startAttempt(options: RemoteForwardOptions, reservation: RemotePortReservation, returnPort: number,
+  budget: MessageBudget, capacity: RemoteCapacity, ownerSignal: AbortSignal, deps: RemoteTunnelDeps): Promise<RemoteTunnelPair> {
+  const localPort = reservation.port, helpers: RemoteHelper[] = [], exits: Promise<void>[] = [];
+  let lostPort = false;
+  let cancelReadiness!: (reason: unknown) => void;
+  const cancelled = new Promise<never>((_, reject) => { cancelReadiness = reject; });
+  void cancelled.catch(() => {});
   let alive = true, closing: Promise<void> | undefined;
   const close = (): Promise<void> => closing ??= (async () => {
     alive = false; ownerSignal.removeEventListener("abort", abort);
     const results = await Promise.allSettled(helpers.map(helper => helper.close()));
+    reservation.release();
     const failure = results.find(result => result.status === "rejected");
     if (failure?.status === "rejected") throw failure.reason;
   })();
-  const abort = () => { void close().catch(() => {}); };
+  const abort = () => {
+    cancelReadiness(remoteError("cancelled", "Messaging tunnel setup was cancelled."));
+    void close().catch(() => {});
+  };
   ownerSignal.addEventListener("abort", abort, { once: true });
-  const cancelSetup = () => { void close().catch(() => {}); };
+  const cancelSetup = () => { cancelReadiness(budget.signal.reason); void close().catch(() => {}); };
   budget.signal.addEventListener("abort", cancelSetup, { once: true });
   try {
-    budget.throwIfEnded(); if (ownerSignal.aborted) throw new Error();
-    for (const [direction, bindPort, targetPort] of [["L", localPort, peer.port], ["R", returnPort, store.requireEnabled().port]] as const) {
+    budget.throwIfEnded(); if (ownerSignal.aborted) throw remoteError("cancelled", "Messaging tunnel setup was cancelled.");
+    for (const [direction, bindPort, targetPort] of [["L", localPort, options.peerPort], ["R", returnPort, options.ownerPort]] as const) {
       budget.throwIfEnded();
-      const helper = (deps.spawn ?? spawnRemoteHelper)(buildTunnelArgv({ alias: peer.ssh, knownHostsFile: knownHosts,
-        direction, bindPort, targetPort }), capacity, ownerSignal);
+      const argv = buildTunnelArgv({ alias: options.ssh, knownHostsFile: options.knownHosts, direction, bindPort, targetPort });
+      if (direction === "L") reservation.handoff();
+      const helper = (deps.spawn ?? spawnRemoteHelper)(argv, capacity, ownerSignal);
       helpers.push(helper);
-      void helper.exited.then(() => close()).catch(() => {});
+      const exit = helper.exited.then(async code => {
+        await close();
+        await helper.output.catch(() => {});
+        if (code !== 0 && helper.forwardFailed) lostPort = true;
+      });
+      exits.push(exit); void exit.catch(() => {});
     }
+    // A competing loopback listener can exist before SSH even starts. Its socket is not a bind receipt.
+    await Promise.race([Promise.all(helpers.map(helper => helper.forwardReady)), cancelled]);
+    const checkAlive = async () => {
+      if (alive) return;
+      await Promise.all(exits);
+      throw remoteError(lostPort ? "forward_port_lost" : "ssh_failed", "Messaging SSH forward setup failed.");
+    };
     let ready = false;
     while (!ready) {
       budget.throwIfEnded();
-      if (!alive) throw new Error();
-      ready = await remotePortIsLoopback(localPort, budget, capacity);
+      await checkAlive();
+      ready = await (deps.inspect ?? remotePortIsLoopback)(localPort, budget, capacity);
+      await checkAlive();
       if (!ready) await new Promise<void>((resolve, reject) => {
         const timer = setTimeout(() => { budget.signal.removeEventListener("abort", cancel); resolve(); }, budget.remainingMs(100));
         const cancel = () => { clearTimeout(timer); budget.signal.removeEventListener("abort", cancel); reject(budget.signal.reason); };
@@ -57,9 +108,14 @@ export async function startRemoteTunnels(store: RemoteMessageStore, peer: Remote
         if (budget.signal.aborted) cancel();
       });
     }
+    budget.throwIfEnded(); if (ownerSignal.aborted) throw remoteError("cancelled", "Messaging tunnel setup was cancelled.");
     const exited = Promise.race(helpers.map(helper => helper.exited));
     void exited.then(() => close()).catch(() => {});
     return { localPort, returnPort, get alive() { return alive; }, exited, close };
-  } catch (error) { await close(); throw error; }
+  } catch (error) {
+    await close(); budget.throwIfEnded();
+    if (ownerSignal.aborted) throw remoteError("cancelled", "Messaging tunnel setup was cancelled.");
+    throw error;
+  }
   finally { budget.signal.removeEventListener("abort", cancelSetup); }
 }
diff --git a/structure/remote-messaging.md b/structure/remote-messaging.md
index 059ee8ff04d..726e582d2d0 100644
--- a/structure/remote-messaging.md
+++ b/structure/remote-messaging.md
@@ -153,6 +153,16 @@ Transient old-endpoint unreachability can allow takeover; lease generation is no
 session authority. Initiated routes remain conflicts.
 `src/messaging/remote-ports.ts` refuses public reverse binds, including sshd
 GatewayPorts configurations. Port selection alone does not prove readiness.
+Local candidates use exclusive socket reservations until the synchronous SSH handoff;
+their process-local claims remain held until the owned pair closes. The remote `_port`
+helper cannot transfer a socket to sshd, so its exclusive candidate is released on return.
+SSH bind receipts precede loopback inspection: owned forwards request DEBUG1 diagnostics,
+but inspect only fixed success/failure tokens and retain no diagnostic bytes. The local
+pre-bind listening announcement is not a receipt; local session entry and the exact remote
+forward success are. Only explicit unsuccessful forward binds reselect both candidates,
+at most three attempts within the original setup deadline, after joined pair cleanup.
+Repeated lost candidates are excluded. Authentication, inspection and cleanup failures
+are not retried. Authenticated peer/native readiness remains required in the owner.
 Linux inspection requires the IPv4 listener table; only an absent IPv6 table is treated as
 empty. Permission, read and other inspection failures remain fail-closed.
 
@@ -225,6 +235,9 @@ Offline contracts live in `tests/codex-integration/messaging-remote-contract.tes
 `tests/codex-integration/messaging-remote-lifecycle.test.ts`.
 Helper settlement and persistent/transient capacity contracts live in
 `tests/codex-integration/messaging-remote-process.test.ts`.
+Exclusive reservations, SSH receipts and bounded lost-port recovery are exercised by
+`tests/codex-integration/messaging-remote-forward-failure.test.ts` and
+`tests/codex-integration/messaging-remote-port-reservation.test.ts` without a live SSH service.
 Opt-in `tests/codex-integration/messaging-remote-interop.test.ts` qualifies a
 disposable loopback sshd with generated keys, isolated homes and strict native
 fixtures. `tests/codex-integration/messaging-remote-native.test.ts` qualifies an
diff --git a/tests/codex-integration/messaging-remote-enrollment-gaps.test.ts b/tests/codex-integration/messaging-remote-enrollment-gaps.test.ts
index 9534cacaa99..987d3762955 100644
--- a/tests/codex-integration/messaging-remote-enrollment-gaps.test.ts
+++ b/tests/codex-integration/messaging-remote-enrollment-gaps.test.ts
@@ -122,7 +122,7 @@ test.skipIf(process.platform === "win32")("messaging SSH consumers pass hostile
         }
         let finish!: (code: number) => void, closed = false;
         const exited = new Promise<number>(resolve => { finish = resolve; });
-        return { pid: 0, exited, output: Promise.resolve(""), async close() {
+        return { pid: 0, exited, output: Promise.resolve(""), forwardReady: Promise.resolve(), async close() {
           if (closed) return;
           closed = true; retired.push(direction);
           if (direction === "L") listener?.stop(true);
diff --git a/tests/codex-integration/messaging-remote-forward-failure.test.ts b/tests/codex-integration/messaging-remote-forward-failure.test.ts
new file mode 100644
index 00000000000..dc826ff823c
--- /dev/null
+++ b/tests/codex-integration/messaging-remote-forward-failure.test.ts
@@ -0,0 +1,39 @@
+import { expect, spyOn, test } from "bun:test";
+import { remotePortCandidate } from "../../src/messaging/remote-ports";
+import { RemoteCapacity } from "../../src/messaging/remote-contract";
+import { spawnRemoteHelper } from "../../src/messaging/remote-process";
+
+test("remote candidate allocation requests an exclusive listener", () => {
+  const listen = spyOn(Bun, "listen");
+  try {
+    remotePortCandidate();
+    expect(listen.mock.calls[0]![0]).toMatchObject({ hostname: "127.0.0.1", port: 0, exclusive: true });
+  } finally { listen.mockRestore(); }
+});
+
+for (const [direction, diagnostic, detected] of [
+  ["L", "channel_setup_fwd_listener_tcpip: cannot listen to port: 23456\r\nCould not request local forwarding.\r\n", true],
+  ["R", "Error: remote port forwarding failed for listen port 23456\r\n", true],
+  ["R", "Error: remote port forwarding failed for listen port 23456\n", true],
+  ["R", "Error: remote port forwarding failed for listen port 234560\n", false],
+  ["R", "Permission denied (publickey).\n", false],
+] as const) {
+  test(`SSH ${direction} diagnostic ${JSON.stringify(diagnostic)} is classified without returning stderr`, async () => {
+    const capacity = new RemoteCapacity();
+    const helper = spawnRemoteHelper(["ssh", "-o", "ExitOnForwardFailure=yes", `-${direction}`,
+      "127.0.0.1:23456:127.0.0.1:34567"], capacity, undefined, undefined, {}, {
+      spawn: () => ({ pid: 12345, exitCode: 255, exited: Promise.resolve(255),
+        stdout: new ReadableStream({ start(controller) { controller.close(); } }),
+        stderr: new ReadableStream({ start(controller) {
+          for (const byte of new TextEncoder().encode(`private diagnostic\n${diagnostic}`)) controller.enqueue(Uint8Array.of(byte));
+          controller.close();
+        } }), stdin: { write: () => 0, end: () => 0 } }),
+      signalGroup() { throw new Error("an exited helper must not be signalled"); },
+    });
+    try {
+      expect(await helper.output).toBe("");
+      expect((helper as typeof helper & { forwardFailed?: boolean }).forwardFailed === true).toBe(detected);
+    } finally { await helper.close(); }
+    expect(capacity.snapshot()).toEqual({ connections: 0, requests: 0, helpers: 0, outputBytes: 0 });
+  });
+}
diff --git a/tests/codex-integration/messaging-remote-lifecycle.test.ts b/tests/codex-integration/messaging-remote-lifecycle.test.ts
index 95ac734164e..834f32661a5 100644
--- a/tests/codex-integration/messaging-remote-lifecycle.test.ts
+++ b/tests/codex-integration/messaging-remote-lifecycle.test.ts
@@ -44,7 +44,7 @@ for (const rejectCleanup of [false, true]) {
           })();
           closes.push(close);
           if (index === 1) pair.aStore.mutate(state => { state.generation = crypto.randomUUID(); });
-          return { pid: 0, exited, output: Promise.resolve(""), close };
+          return { pid: 0, exited, output: Promise.resolve(""), forwardReady: Promise.resolve(), close };
         },
       });
       await expect(result).rejects.toThrow(rejectCleanup ? "fixture cleanup incomplete" : "configuration changed");
@@ -90,7 +90,7 @@ for (const [rejectCleanup, deliberateRemoval] of [[false, false], [true, false],
             if (rejectCleanup && index === 1) throw remoteError("cleanup_incomplete", "fixture cleanup incomplete");
           })();
           closes.push(close); exits.push(() => finish(0));
-          return { pid: 0, exited, output: Promise.resolve(""), close };
+          return { pid: 0, exited, output: Promise.resolve(""), forwardReady: Promise.resolve(), close };
         },
       });
       await resolveRemoteRoute(pair.aStore, "worker", budget);
@@ -228,7 +228,7 @@ for (const failRenewal of [false, true]) test.skipIf(process.platform === "win32
           if (listener) await new Promise<void>((resolve, reject) => listener.close(error => error ? reject(error) : resolve()));
           free(); finish(0);
         })();
-        closes.push(close); return { pid: 0, exited, output: Promise.resolve(""), close };
+        closes.push(close); return { pid: 0, exited, output: Promise.resolve(""), forwardReady: Promise.resolve(), close };
       },
     });
     if (failRenewal) {
diff --git a/tests/codex-integration/messaging-remote-port-reservation.test.ts b/tests/codex-integration/messaging-remote-port-reservation.test.ts
new file mode 100644
index 00000000000..d8458158bb0
--- /dev/null
+++ b/tests/codex-integration/messaging-remote-port-reservation.test.ts
@@ -0,0 +1,213 @@
+import { expect, test } from "bun:test";
+import { join } from "node:path";
+import { reserveRemotePort } from "../../src/messaging/remote-ports";
+import { RemoteCapacity, remoteError } from "../../src/messaging/remote-contract";
+import { MessageBudget } from "../../src/messaging/budget";
+import { startRemoteForwardPair, type RemoteTunnelDeps } from "../../src/messaging/remote-tunnels";
+import { remoteForwardFailure } from "../../src/messaging/remote-forward-failure";
+import { spawnRemoteHelper } from "../../src/messaging/remote-process";
+
+test("a pending exclusive reservation physically excludes another listener", () => {
+  const first = reserveRemotePort();
+  let second: ReturnType<typeof reserveRemotePort> | undefined;
+  try {
+    expect(() => Bun.listen({ hostname: "127.0.0.1", port: first.port, exclusive: true,
+      socket: { data(socket) { socket.end(); } } })).toThrow();
+    second = reserveRemotePort();
+    expect(second.port).not.toBe(first.port);
+  } finally { first.release(); second?.release(); }
+});
+
+test("only post-bind SSH receipts confirm forwards, including byte-split remote confirmations", () => {
+  for (const direction of ["L", "R"] as const) {
+    let ready = 0;
+    const state = remoteForwardFailure(["ssh", "-o", "ExitOnForwardFailure=yes", `-${direction}`,
+      "127.0.0.1:23456:127.0.0.1:34567"], () => { ready++; });
+    state.write(new TextEncoder().encode("debug1: Local forwarding listening on 127.0.0.1 port 23456.\n"));
+    state.write(new TextEncoder().encode("debug1: remote forward success for: listen 127.0.0.1:234560, connect 127.0.0.1:34567\n"));
+    state.write(new TextEncoder().encode("debug1: remote forward success for: listen 127.0.0.1:23456, connect 127.0.0.1:345670\n"));
+    expect(state.ready).toBe(false); expect(ready).toBe(0);
+    const success = direction === "L" ? "debug1: Entering interactive session.\r\n"
+      : "debug1: remote forward success for: listen 127.0.0.1:23456, connect 127.0.0.1:34567\r\n";
+    for (const byte of new TextEncoder().encode(success)) state.write(Uint8Array.of(byte));
+    expect(state.ready).toBe(true); expect(ready).toBe(1); expect(state.failed).toBe(false);
+  }
+});
+
+test("owned SSH forwards request diagnostic receipts without returning their diagnostic output", async () => {
+  const capacity = new RemoteCapacity();
+  let spawned: readonly string[] = [];
+  const helper = spawnRemoteHelper(["ssh", "-o", "LogLevel=ERROR", "-o", "ExitOnForwardFailure=yes", "-L",
+    "127.0.0.1:23456:127.0.0.1:34567"], capacity, undefined, undefined, {}, {
+    spawn(argv) {
+      spawned = argv;
+      return { pid: 12345, exitCode: 0, exited: Promise.resolve(0),
+        stdout: new ReadableStream({ start(controller) { controller.close(); } }),
+        stderr: new ReadableStream({ start(controller) {
+          controller.enqueue(new TextEncoder().encode("private fixture diagnostic\ndebug1: Entering interactive session.\n"));
+          controller.close();
+        } }), stdin: { write: () => 0, end: () => 0 } };
+    }, signalGroup() { throw new Error("exited process"); },
+  });
+  try {
+    expect(spawned.slice(0, 3)).toEqual(["ssh", "-o", "LogLevel=DEBUG1"]);
+    await helper.forwardReady;
+    expect(await helper.output).toBe("");
+  } finally { await helper.close(); }
+  expect(capacity.snapshot()).toEqual({ connections: 0, requests: 0, helpers: 0, outputBytes: 0 });
+});
+
+test("pending local reservations stay distinct through socket handoff and release exactly once", () => {
+  const stopped: number[] = [], ports = [23456, 23456, 23457];
+  const allocate = () => {
+    const port = ports.shift()!;
+    return { port, stop() { stopped.push(port); } };
+  };
+  const first = reserveRemotePort(allocate);
+  expect(stopped).toEqual([]);
+  first.handoff();
+  const second = reserveRemotePort(allocate);
+  try {
+    expect(first.port).not.toBe(second.port);
+    expect(stopped).toEqual([23456, 23456]);
+    first.release(); first.release(); second.release(); second.release();
+    expect(stopped).toEqual([23456, 23456, 23457]);
+    const reused = reserveRemotePort(() => ({ port: 23456, stop() {} }));
+    reused.release();
+  } finally { first.release(); second.release(); }
+});
+
+test("an allocator repeatedly offering a pending port is bounded and cannot steal the claim", () => {
+  const first = reserveRemotePort(() => ({ port: 23456, stop() {} }));
+  let allocations = 0, stops = 0;
+  try {
+    expect(() => reserveRemotePort(() => { allocations++; return { port: first.port, stop() { stops++; } }; }))
+      .toThrow("distinct");
+    expect(allocations).toBe(8); expect(stops).toBe(8);
+  } finally { first.release(); }
+});
+
+const options = { ssh: "fixture", knownHosts: join(process.cwd(), ".tmp", "fixture.known_hosts"), peerPort: 34567, ownerPort: 45678 };
+function forwardFixture(direction: "L" | "R", failures: number, forwardFailed = true) {
+  const events: string[] = [], localPorts: number[] = [], remotePorts: number[] = [];
+  let runs = 0, allocated = 0, closed = 0, handedOff = false;
+  const deps: RemoteTunnelDeps = {
+    async run() { runs++; return JSON.stringify({ port: 25000 + runs }); },
+    allocatePort() {
+      const port = 24000 + ++allocated;
+      handedOff = false; localPorts.push(port);
+      return { port, stop() { handedOff = true; events.push(`handoff:${port}`); } };
+    },
+    spawn(argv) {
+      const kind = argv.includes("-L") ? "L" : "R";
+      expect(handedOff).toBe(true);
+      expect(argv).toContain("ExitOnForwardFailure=yes");
+      if (kind === "R") remotePorts.push(Number(argv[argv.indexOf("-R") + 1]!.split(":")[1]));
+      events.push(`spawn:${runs}:${kind}`);
+      const failed = kind === direction && runs <= failures;
+      let finish!: (code: number) => void, closing = false;
+      const exited = failed ? Promise.resolve(255) : new Promise<number>(resolve => { finish = resolve; });
+      const forwardReady = failed ? Promise.reject(remoteError(forwardFailed ? "forward_port_lost" : "ssh_failed", "fixture forward failed")) : Promise.resolve();
+      return { pid: 0, exited, output: Promise.resolve(""), forwardReady, forwardFailed: failed && forwardFailed, async close() {
+        if (closing) return;
+        closing = true; closed++; events.push(`close:${runs}:${kind}`);
+        if (!failed) finish(0);
+      } };
+    },
+    async inspect() { return true; },
+  };
+  return { deps, events, localPorts, remotePorts, get runs() { return runs; }, get closed() { return closed; } };
+}
+
+for (const direction of ["L", "R"] as const) {
+  test(`lost ${direction} candidate reselects both ports only after joining the failed pair`, async () => {
+    const fixture = forwardFixture(direction, 1), budget = new MessageBudget(), capacity = new RemoteCapacity();
+    try {
+      const pair = await startRemoteForwardPair(options, budget, capacity, new AbortController().signal, fixture.deps);
+      try {
+        expect(fixture.runs).toBe(2); expect(fixture.closed).toBe(2);
+        expect(fixture.localPorts).toEqual([24001, 24002]); expect(fixture.remotePorts).toEqual([25001, 25002]);
+        expect(fixture.events.indexOf("close:1:R")).toBeLessThan(fixture.events.indexOf("spawn:2:L"));
+        expect(pair.localPort).toBe(24002); expect(pair.returnPort).toBe(25002); expect(pair.alive).toBe(true);
+      } finally { await pair.close(); }
+      expect(fixture.closed).toBe(4);
+    } finally { budget.dispose(); }
+  });
+}
+
+test("three explicit forward failures exhaust setup without leaking an owned pair", async () => {
+  const fixture = forwardFixture("R", Infinity), budget = new MessageBudget();
+  try {
+    await expect(startRemoteForwardPair(options, budget, new RemoteCapacity(), new AbortController().signal, fixture.deps))
+      .rejects.toMatchObject({ code: "forward_port_lost" });
+    expect(fixture.runs).toBe(3); expect(fixture.closed).toBe(6);
+  } finally { budget.dispose(); }
+});
+
+test("an SSH authentication or generic exit failure does not reselect a port", async () => {
+  const fixture = forwardFixture("L", Infinity, false), budget = new MessageBudget();
+  try {
+    await expect(startRemoteForwardPair(options, budget, new RemoteCapacity(), new AbortController().signal, fixture.deps))
+      .rejects.toMatchObject({ code: "ssh_failed" });
+    expect(fixture.runs).toBe(1); expect(fixture.closed).toBe(2);
+  } finally { budget.dispose(); }
+});
+
+test("cancellation after losing a candidate prevents reselection", async () => {
+  const fixture = forwardFixture("R", 1), controller = new AbortController(), budget = new MessageBudget();
+  const spawn = fixture.deps.spawn!;
+  fixture.deps.spawn = (...args) => {
+    const helper = spawn(...args), close = helper.close;
+    return { ...helper, async close() { await close(); controller.abort(); } };
+  };
+  try {
+    await expect(startRemoteForwardPair(options, budget, new RemoteCapacity(), controller.signal, fixture.deps))
+      .rejects.toMatchObject({ code: "cancelled" });
+    expect(fixture.runs).toBe(1); expect(fixture.closed).toBe(2);
+  } finally { budget.dispose(); }
+});
+
+test("an occupied loopback candidate cannot pass inspection before both SSH bind receipts", async () => {
+  const fixture = forwardFixture("L", 0), budget = new MessageBudget();
+  let release!: () => void, spawned!: () => void, inspections = 0;
+  const receipts = new Promise<void>(resolve => { release = resolve; });
+  const bothSpawned = new Promise<void>(resolve => { spawned = resolve; });
+  const spawn = fixture.deps.spawn!;
+  fixture.deps.spawn = (...args) => {
+    const helper = spawn(...args);
+    if (args[0].includes("-R")) spawned();
+    return { ...helper, forwardReady: receipts };
+  };
+  fixture.deps.inspect = async () => { inspections++; return true; };
+  const pending = startRemoteForwardPair(options, budget, new RemoteCapacity(), new AbortController().signal, fixture.deps);
+  try {
+    await bothSpawned;
+    expect(inspections).toBe(0);
+    release();
+    const pair = await pending;
+    try { expect(inspections).toBe(1); } finally { await pair.close(); }
+  } finally { release(); budget.dispose(); }
+});
+
+test("loopback inspection failure is joined and never treated as a lost candidate", async () => {
+  const fixture = forwardFixture("L", 0), budget = new MessageBudget();
+  fixture.deps.inspect = async () => { throw remoteError("inspection_failed", "fixture inspection failed"); };
+  try {
+    await expect(startRemoteForwardPair(options, budget, new RemoteCapacity(), new AbortController().signal, fixture.deps))
+      .rejects.toMatchObject({ code: "inspection_failed" });
+    expect(fixture.runs).toBe(1); expect(fixture.closed).toBe(2);
+  } finally { budget.dispose(); }
+});
+
+test("a remote helper repeatedly offering the same lost port is bounded without respawning it", async () => {
+  const fixture = forwardFixture("R", 1), budget = new MessageBudget();
+  let runs = 0;
+  const run = fixture.deps.run!;
+  fixture.deps.run = async (...args) => { runs++; await run(...args); return JSON.stringify({ port: 25001 }); };
+  // The remote candidate is the same on every offer; only the first pair may be spawned.
+  try {
+    await expect(startRemoteForwardPair(options, budget, new RemoteCapacity(), new AbortController().signal, fixture.deps))
+      .rejects.toMatchObject({ code: "forward_port_lost" });
+    expect(runs).toBe(3); expect(fixture.localPorts).toHaveLength(1); expect(fixture.closed).toBe(2);
+  } finally { budget.dispose(); }
+});
diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json
index 0b5ed542af8..90b5f04fd03 100644
--- a/tests/fixtures/test-layout-expected.json
+++ b/tests/fixtures/test-layout-expected.json
@@ -103,6 +103,8 @@
   "messaging-remote-interop.test.ts": "codex-integration",
   "messaging-remote-native.test.ts": "codex-integration",
   "messaging-remote-process.test.ts": "codex-integration",
+  "messaging-remote-port-reservation.test.ts": "codex-integration",
+  "messaging-remote-forward-failure.test.ts": "codex-integration",
   "messaging-remote-admission.test.ts": "codex-integration",
   "messaging-remote-replies.test.ts": "codex-integration",
   "messaging-remote-scratch.test.ts": "codex-integration",

From 1ef9557599b31e0cd501439674e8c2b816fc11b0 Mon Sep 17 00:00:00 2001
From: JUN <243035832+lidge-jun@users.noreply.github.com>
Date: Wed, 28 Oct 2026 18:37:19 +0900
Subject: [PATCH 2/2] fix(messaging): preserve initiating forward failures
 through pair shutdown

Keep intrinsic I/O and pre-cleanup exit failure provenance on each helper.
Join diagnostic output and pair cleanup before classifying readiness failures,
so cleanup-induced sibling rejection cannot mask an explicit lost forward port.
Independent failures, cancellation and incomplete cleanup remain terminal.

Add deterministic L/R delayed-stderr recovery and exhaustion regressions plus
terminal-failure controls. Keep reservation handoff, setup deadline and the
three-attempt ceiling unchanged.

Validation: regression 11 pass/4 fail before, 15 pass/0 fail after; requested
eight focused files 85 pass/15 skip/0 fail; test:changed 769 pass/112 skip/0 fail;
typecheck, structure:check, privacy:scan and git diff --check passed.
---
 src/messaging/remote-process.ts               |  12 +-
 src/messaging/remote-tunnels.ts               |  13 ++-
 structure/remote-messaging.md                 |   6 +-
 .../messaging-remote-forward-failure.test.ts  | 107 ++++++++++++++++++
 4 files changed, 134 insertions(+), 4 deletions(-)

diff --git a/src/messaging/remote-process.ts b/src/messaging/remote-process.ts
index 8058dd1831f..31583ebe240 100644
--- a/src/messaging/remote-process.ts
+++ b/src/messaging/remote-process.ts
@@ -8,6 +8,8 @@ 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. */
@@ -41,6 +43,7 @@ export function spawnRemoteHelper(argv: readonly string[], capacity: RemoteCapac
   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); }));
@@ -55,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;
@@ -124,7 +128,13 @@ export function spawnRemoteHelper(argv: readonly string[], capacity: RemoteCapac
     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, forwardReady, get forwardFailed() { return forwardFailure.failed; }, 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. */
diff --git a/src/messaging/remote-tunnels.ts b/src/messaging/remote-tunnels.ts
index 8da073cd17e..5e32b0e58c5 100644
--- a/src/messaging/remote-tunnels.ts
+++ b/src/messaging/remote-tunnels.ts
@@ -55,6 +55,11 @@ async function startAttempt(options: RemoteForwardOptions, reservation: RemotePo
   budget: MessageBudget, capacity: RemoteCapacity, ownerSignal: AbortSignal, deps: RemoteTunnelDeps): Promise<RemoteTunnelPair> {
   const localPort = reservation.port, helpers: RemoteHelper[] = [], exits: Promise<void>[] = [];
   let lostPort = false;
+  let awaitingReceipts = false;
+  const setupFailure = (fallback: unknown) => {
+    const failures = helpers.flatMap(helper => helper.setupFailure ? [helper.setupFailure] : []);
+    return failures.find(error => (error as { code?: string }).code !== "forward_port_lost") ?? failures[0] ?? fallback;
+  };
   let cancelReadiness!: (reason: unknown) => void;
   const cancelled = new Promise<never>((_, reject) => { cancelReadiness = reject; });
   void cancelled.catch(() => {});
@@ -62,6 +67,7 @@ async function startAttempt(options: RemoteForwardOptions, reservation: RemotePo
   const close = (): Promise<void> => closing ??= (async () => {
     alive = false; ownerSignal.removeEventListener("abort", abort);
     const results = await Promise.allSettled(helpers.map(helper => helper.close()));
+    await Promise.allSettled(helpers.map(helper => helper.output));
     reservation.release();
     const failure = results.find(result => result.status === "rejected");
     if (failure?.status === "rejected") throw failure.reason;
@@ -89,11 +95,13 @@ async function startAttempt(options: RemoteForwardOptions, reservation: RemotePo
       exits.push(exit); void exit.catch(() => {});
     }
     // A competing loopback listener can exist before SSH even starts. Its socket is not a bind receipt.
+    awaitingReceipts = true;
     await Promise.race([Promise.all(helpers.map(helper => helper.forwardReady)), cancelled]);
+    awaitingReceipts = false;
     const checkAlive = async () => {
       if (alive) return;
       await Promise.all(exits);
-      throw remoteError(lostPort ? "forward_port_lost" : "ssh_failed", "Messaging SSH forward setup failed.");
+      throw setupFailure(remoteError(lostPort ? "forward_port_lost" : "ssh_failed", "Messaging SSH forward setup failed."));
     };
     let ready = false;
     while (!ready) {
@@ -115,7 +123,8 @@ async function startAttempt(options: RemoteForwardOptions, reservation: RemotePo
   } catch (error) {
     await close(); budget.throwIfEnded();
     if (ownerSignal.aborted) throw remoteError("cancelled", "Messaging tunnel setup was cancelled.");
-    throw error;
+    // Sibling shutdown can reject first; classify only after every helper's diagnostics and cleanup settle.
+    throw awaitingReceipts ? setupFailure(error) : error;
   }
   finally { budget.signal.removeEventListener("abort", cancelSetup); }
 }
diff --git a/structure/remote-messaging.md b/structure/remote-messaging.md
index 726e582d2d0..fb068607a96 100644
--- a/structure/remote-messaging.md
+++ b/structure/remote-messaging.md
@@ -162,7 +162,11 @@ pre-bind listening announcement is not a receipt; local session entry and the ex
 forward success are. Only explicit unsuccessful forward binds reselect both candidates,
 at most three attempts within the original setup deadline, after joined pair cleanup.
 Repeated lost candidates are excluded. Authentication, inspection and cleanup failures
-are not retried. Authenticated peer/native readiness remains required in the owner.
+are not retried. Helpers preserve intrinsic I/O and pre-cleanup exit failures through
+diagnostic settlement; pair classification after joined cleanup ignores sibling exits
+caused by shutdown, so an initiating bind failure survives a sibling readiness rejection.
+Independent failures, cancellation and incomplete cleanup remain terminal.
+Authenticated peer/native readiness remains required in the owner.
 Linux inspection requires the IPv4 listener table; only an absent IPv6 table is treated as
 empty. Permission, read and other inspection failures remain fail-closed.
 
diff --git a/tests/codex-integration/messaging-remote-forward-failure.test.ts b/tests/codex-integration/messaging-remote-forward-failure.test.ts
index dc826ff823c..d12b43a1254 100644
--- a/tests/codex-integration/messaging-remote-forward-failure.test.ts
+++ b/tests/codex-integration/messaging-remote-forward-failure.test.ts
@@ -1,7 +1,11 @@
 import { expect, spyOn, test } from "bun:test";
+import { join } from "node:path";
 import { remotePortCandidate } from "../../src/messaging/remote-ports";
 import { RemoteCapacity } from "../../src/messaging/remote-contract";
 import { spawnRemoteHelper } from "../../src/messaging/remote-process";
+import { MessageBudget } from "../../src/messaging/budget";
+import { remoteError } from "../../src/messaging/remote-contract";
+import { startRemoteForwardPair, type RemoteTunnelDeps } from "../../src/messaging/remote-tunnels";
 
 test("remote candidate allocation requests an exclusive listener", () => {
   const listen = spyOn(Bun, "listen");
@@ -37,3 +41,106 @@ for (const [direction, diagnostic, detected] of [
     expect(capacity.snapshot()).toEqual({ connections: 0, requests: 0, helpers: 0, outputBytes: 0 });
   });
 }
+
+type RaceMode = "recover" | "exhaust" | "input" | "independent" | "cleanup" | "cancel" | "missing-marker";
+const forwardOptions = { ssh: "fixture", knownHosts: join(process.cwd(), ".tmp", "fixture.known_hosts"), peerPort: 34567, ownerPort: 45678 };
+function delayedBindFixture(failedDirection: "L" | "R", mode: RaceMode) {
+  let runs = 0, allocations = 0, inspections = 0, siblingRejections = 0;
+  let finishDiagnostic!: () => void;
+  const events: string[] = [], capacity = new RemoteCapacity(), owner = new AbortController();
+  const deps: RemoteTunnelDeps = {
+    async run() {
+      if (runs) expect(capacity.snapshot().helpers).toBe(0);
+      return JSON.stringify({ port: 25000 + ++runs });
+    },
+    allocatePort() { return { port: 24000 + ++allocations, stop() {} }; },
+    async inspect() { inspections++; expect(runs).toBe(2); return true; },
+    spawn(argv, caps, signal) {
+      const direction = argv.includes("-L") ? "L" : "R", attempt = runs;
+      const failingAttempt = attempt === 1 || mode === "exhaust";
+      const initiating = failingAttempt && direction === failedDirection;
+      const sibling = failingAttempt && !initiating;
+      let code: number | null = initiating || (sibling && mode === "independent") ? 255 : null;
+      let finish!: (value: number) => void;
+      const exited = new Promise<number>(resolve => { finish = resolve; });
+      if (code !== null) finish(code);
+      let stdout!: ReadableStreamDefaultController<Uint8Array>, stderr!: ReadableStreamDefaultController<Uint8Array>;
+      const port = argv[argv.indexOf(`-${direction}`) + 1]!.split(":")[1];
+      const helper = spawnRemoteHelper(argv, caps, signal, undefined, {}, {
+        spawn() { return { pid: direction === "L" ? 12345 : 12346, get exitCode() { return code; }, exited,
+          stdout: new ReadableStream({ start(controller) { stdout = controller; } }),
+          stderr: new ReadableStream({ start(controller) {
+            stderr = controller;
+            if (!failingAttempt) controller.enqueue(new TextEncoder().encode(direction === "L"
+              ? "debug1: Entering interactive session.\r\n"
+              : `debug1: remote forward success for: listen 127.0.0.1:${port}, connect 127.0.0.1:45678\r\n`));
+          } }),
+          stdin: { write() { return 0; }, end() {
+            if (sibling && mode === "input") throw new Error("private fixture input failure");
+            return 0;
+          } },
+        }; },
+        signalGroup() { code = 143; stdout.close(); stderr.close(); finish(143); },
+      });
+      if (initiating) finishDiagnostic = () => {
+        events.push(`diagnostic:${attempt}`);
+        if (mode !== "missing-marker") stderr.enqueue(new TextEncoder().encode(direction === "L"
+          ? "Could not request local forwarding.\r\n"
+          : `Error: remote port forwarding failed for listen port ${port}\r\n`));
+        stderr.close(); stdout.close();
+      };
+      if (sibling) {
+        if (code !== null) { stdout.close(); stderr.close(); }
+        void helper.forwardReady.catch(() => {
+          siblingRejections++; events.push(`sibling-rejected:${attempt}`);
+          if (mode === "cancel") owner.abort();
+          finishDiagnostic();
+        });
+      }
+      const close = helper.close;
+      helper.close = async () => {
+        await close(); events.push(`closed:${attempt}:${direction}`);
+        if (sibling && mode === "cleanup") throw remoteError("cleanup_incomplete", "fixture cleanup failed");
+      };
+      return helper;
+    },
+  };
+  return { deps, capacity, owner, events, get runs() { return runs; }, get inspections() { return inspections; },
+    get siblingRejections() { return siblingRejections; } };
+}
+
+for (const direction of ["L", "R"] as const) for (const mode of ["recover", "exhaust"] as const) {
+  test(`delayed ${direction} bind diagnostic survives sibling rejection (${mode})`, async () => {
+    const fixture = delayedBindFixture(direction, mode), budget = new MessageBudget();
+    try {
+      const pending = startRemoteForwardPair(forwardOptions,
+        budget, fixture.capacity, fixture.owner.signal, fixture.deps);
+      if (mode === "recover") {
+        const pair = await pending;
+        try {
+          expect(pair.alive).toBe(true); expect(pair.localPort).toBe(24002); expect(pair.returnPort).toBe(25002);
+          expect(fixture.runs).toBe(2); expect(fixture.inspections).toBe(1);
+        } finally { await pair.close(); }
+      } else {
+        await expect(pending).rejects.toMatchObject({ code: "forward_port_lost" });
+        expect(fixture.runs).toBe(3); expect(fixture.inspections).toBe(0);
+      }
+      expect(fixture.siblingRejections).toBe(mode === "recover" ? 1 : 3);
+      expect(fixture.events.indexOf("sibling-rejected:1")).toBeLessThan(fixture.events.indexOf("diagnostic:1"));
+      expect(fixture.capacity.snapshot()).toEqual({ connections: 0, requests: 0, helpers: 0, outputBytes: 0 });
+    } finally { budget.dispose(); }
+  });
+}
+
+for (const [mode, code] of [["input", "ssh_incomplete"], ["independent", "ssh_failed"],
+  ["cleanup", "cleanup_incomplete"], ["cancel", "cancelled"], ["missing-marker", "ssh_failed"]] as const) {
+  test(`a delayed bind marker cannot hide ${mode} failure`, async () => {
+    const fixture = delayedBindFixture("L", mode), budget = new MessageBudget();
+    try {
+      await expect(startRemoteForwardPair(forwardOptions,
+        budget, fixture.capacity, fixture.owner.signal, fixture.deps)).rejects.toMatchObject({ code });
+      expect(fixture.runs).toBe(1); expect(fixture.inspections).toBe(0); expect(fixture.siblingRejections).toBe(1);
+      expect(fixture.capacity.snapshot()).toEqual({ connections: 0, requests: 0, helpers: 0, outputBytes: 0 });
+    } finally { budget.dispose(); }
+  });
+}