fix(ssh): identify concurrent terminal joins

electron/bridges/sshBridge.connectionReuse.test.cjs

@@ -20,7 +20,12 @@ const {
 // Load sshBridge with a mocked ssh2 module so we can observe whether a *new*
 // SSH client is constructed (a fresh connection) versus an existing connection
 // being reused for a new shell channel (issue #1204).
-function loadBridgeWithMockedSsh2(t, { connectReady = false, remoteVer = "OpenSSH_9.0" } = {}) {
+function loadBridgeWithMockedSsh2(t, {
+  connectReady = false,
+  remoteVer = "OpenSSH_9.0",
+  trackShellPids = false,
+  beforeShellPidScanResult,
+} = {}) {
   const bridgePath = require.resolve("./sshBridge.cjs");
   const authHelperPath = require.resolve("./sshAuthHelper.cjs");
   const originalLoad = Module._load;
@@ -54,7 +59,24 @@ function loadBridgeWithMockedSsh2(t, { connectReady = false, remoteVer = "OpenSS
     }
     end() { this.ended += 1; }
     destroy() {}
-    exec(_command, callback) { callback?.(new Error("exec unavailable")); }
+    exec(_command, callback) {
+      if (!trackShellPids) {
+        callback?.(new Error("exec unavailable"));
+        return;
+      }
+      const stream = new EventEmitter();
+      stream.stderr = new EventEmitter();
+      stream.close = () => {};
+      const pids = this.openedShells
+        .map((_shell, index) => `${(index + 1) * 111} 1`)
+        .join("\n");
+      setImmediate(() => {
+        beforeShellPidScanResult?.();
+        stream.emit("data", Buffer.from(`${pids}\n__NETCATTY_SHELL_SCAN_COMPLETE__\n`));
+        stream.emit("close", 0);
+      });
+      callback(null, stream);
+    }
     shell(_pty, _options, callback) {
       const stream = makeStream();
       this.openedShells.push(stream);
@@ -114,10 +136,13 @@ function loadBridgeWithMockedSsh2(t, { connectReady = false, remoteVer = "OpenSS
   return { bridge, getClientConstructCount: () => clientConstructCount };
 }
 
-test("simultaneous normal opens of the same host make one physical SSH dial", async (t) => {
+test("simultaneous normal opens identify both shells before sharing one SSH connection", async (t) => {
   resetSshTransportRegistryForTests({ defaultIdleTtlMs: 0 });
   t.after(() => resetSshTransportRegistryForTests({ defaultIdleTtlMs: 0 }));
-  const { bridge, getClientConstructCount } = loadBridgeWithMockedSsh2(t, { connectReady: true });
+  const { bridge, getClientConstructCount } = loadBridgeWithMockedSsh2(t, {
+    connectReady: true,
+    trackShellPids: true,
+  });
   const sessions = new Map();
   const start = registerStartHandler(bridge, sessions);
   const options = {
@@ -140,6 +165,70 @@ test("simultaneous normal opens of the same host make one physical SSH dial", as
   assert.equal(getClientConstructCount(), 1);
   assert.equal(sessions.get("normal-1").conn, sessions.get("normal-2").conn);
   assert.equal(sessions.get("normal-1").connRef.count, 2);
+  assert.equal(sessions.get("normal-1").shellPid, "111");
+  assert.equal(sessions.get("normal-2").shellPid, "222");
+});
+
+test("simultaneous normal opens use separate connections when shell identity is unavailable", async (t) => {
+  resetSshTransportRegistryForTests({ defaultIdleTtlMs: 0 });
+  t.after(() => resetSshTransportRegistryForTests({ defaultIdleTtlMs: 0 }));
+  const { bridge, getClientConstructCount } = loadBridgeWithMockedSsh2(t, {
+    connectReady: true,
+  });
+  const sessions = new Map();
+  const start = registerStartHandler(bridge, sessions);
+  const options = {
+    hostname: "10.0.0.54",
+    username: "alice",
+    port: 22,
+    authMethod: "password",
+    password: "secret",
+    useSshAgent: false,
+    verifyHostKeys: false,
+  };
+
+  await Promise.all([
+    start({ sender: makeSender() }, { ...options, sessionId: "normal-1" }),
+    start({ sender: makeSender() }, { ...options, sessionId: "normal-2" }),
+  ]);
+
+  assert.equal(getClientConstructCount(), 2);
+  assert.notEqual(sessions.get("normal-1").conn, sessions.get("normal-2").conn);
+});
+
+test("simultaneous normal opens do not trust a replacement session after identity discovery", async (t) => {
+  resetSshTransportRegistryForTests({ defaultIdleTtlMs: 0 });
+  t.after(() => resetSshTransportRegistryForTests({ defaultIdleTtlMs: 0 }));
+  const sessions = new Map();
+  let replaced = false;
+  const { bridge, getClientConstructCount } = loadBridgeWithMockedSsh2(t, {
+    connectReady: true,
+    trackShellPids: true,
+    beforeShellPidScanResult: () => {
+      if (replaced) return;
+      replaced = true;
+      const original = sessions.get("normal-1");
+      sessions.set("normal-1", { ...original, shellPid: "999" });
+    },
+  });
+  const start = registerStartHandler(bridge, sessions);
+  const options = {
+    hostname: "10.0.0.55",
+    username: "alice",
+    port: 22,
+    authMethod: "password",
+    password: "secret",
+    useSshAgent: false,
+    verifyHostKeys: false,
+  };
+
+  await Promise.all([
+    start({ sender: makeSender() }, { ...options, sessionId: "normal-1" }),
+    start({ sender: makeSender() }, { ...options, sessionId: "normal-2" }),
+  ]);
+
+  assert.equal(getClientConstructCount(), 2);
+  assert.notEqual(sessions.get("normal-1").conn, sessions.get("normal-2").conn);
 });
 
 test("reuseTransport false makes simultaneous same-host opens dial independently", async (t) => {

electron/bridges/sshBridge/startSession.cjs

@@ -217,6 +217,38 @@ function createStartSessionApi(ctx) {
       return last;
     };
 
+    const ensureConcurrentJoinShellIdentity = async (connRef, options) => {
+      if (!connRef || options.skipShellPidDiscovery) return true;
+
+      const current = [...sessions.entries()].filter(([, candidate]) => (
+        candidate?.connRef === connRef
+        && candidate?.stream
+      ));
+      if (current.length === 0) return false;
+      if (current.every(([, candidate]) => candidate.shellPid)) return true;
+      if (current.length !== 1) return false;
+
+      const [[sessionId, candidate]] = current;
+      const discovery = await listInteractiveShellPids(connRef.conn);
+      const live = sessions.get(sessionId);
+      if (
+        discovery.available
+        && discovery.pids.length === 1
+        && live === candidate
+        && live.connRef === connRef
+        && live.stream === candidate.stream
+        && !live.shellPid
+      ) {
+        live.shellPid = String(discovery.pids[0]);
+      }
+      return Boolean(
+        live === candidate
+        && live.connRef === connRef
+        && live.stream === candidate.stream
+        && live.shellPid
+      );
+    };
+
     const waitForNewInteractiveShellPid = async (conn, previousPids, opts = {}) => {
       const previous = new Set(previousPids);
       const initialDelayMs = Math.max(0, Number(opts.initialDelayMs) || 0);
@@ -1124,6 +1156,10 @@ function createStartSessionApi(ctx) {
 
     function reuseShellSession(event, options, sourceSession, sessionId, log, reuseOpts = {}) {
       const connRef = sourceSession.connRef;
+      const {
+        ensureConcurrentJoinIdentity = false,
+        ...openReuseOpts
+      } = reuseOpts;
       const refHolder = {};
       // Pin while queued as well as while opening: the source tab may close
       // before this copy reaches the front of the per-connection queue.
@@ -1132,16 +1168,31 @@ function createStartSessionApi(ctx) {
       const previous = connRef.shellOpenQueue || Promise.resolve();
       const operation = previous
         .catch(() => {})
-        .then(() => openReusedShellSerialized(
-          event,
-          options,
-          sourceSession,
-          sessionId,
-          log,
-          connRef,
-          refHolder,
-          reuseOpts,
-        ));
+        .then(async () => {
+          try {
+            if (
+              ensureConcurrentJoinIdentity
+              && !await ensureConcurrentJoinShellIdentity(connRef, options)
+            ) {
+              const error = new Error("Concurrent terminal shell identity is unavailable");
+              error.code = "SSH_CONCURRENT_SHELL_IDENTITY_UNSAFE";
+              throw error;
+            }
+          } catch (error) {
+            releaseConnectionRef(refHolder);
+            throw error;
+          }
+          return openReusedShellSerialized(
+            event,
+            options,
+            sourceSession,
+            sessionId,
+            log,
+            connRef,
+            refHolder,
+            openReuseOpts,
+          );
+        });
       const tail = operation.then(() => undefined, () => undefined);
       connRef.shellOpenQueue = tail;
 
@@ -1301,22 +1352,33 @@ function createStartSessionApi(ctx) {
               { conn: transport.conn, connRef: transport, stream: {} },
               sessionId,
               log,
+              {
+                ensureConcurrentJoinIdentity: coordination.role === "join"
+                  && coordination._record?.kind === "shell",
+              },
             );
           } catch (coordinationErr) {
-            if (options._passphraseSignal?.aborted) {
-              throw options._passphraseSignal.reason instanceof Error
-                ? options._passphraseSignal.reason
-                : coordinationErr;
+            if (coordinationErr?.code === "SSH_CONCURRENT_SHELL_IDENTITY_UNSAFE") {
+              log("concurrent transport shell identity unavailable; connecting fresh", {
+                sessionId,
+                hostname: options.hostname,
+              });
+            } else {
+              if (options._passphraseSignal?.aborted) {
+                throw options._passphraseSignal.reason instanceof Error
+                  ? options._passphraseSignal.reason
+                  : coordinationErr;
+              }
+              // A waiter must observe the leader's real failure instead of
+              // immediately starting a second authentication prompt. Existing
+              // transport reuse keeps its historical fresh-dial fallback.
+              if (coordination.role === "join") throw coordinationErr;
+              log("coordinated transport reuse failed, connecting fresh", {
+                sessionId,
+                hostname: options.hostname,
+                error: coordinationErr?.message,
+              });
             }
-            // A waiter must observe the leader's real failure instead of
-            // immediately starting a second authentication prompt. Existing
-            // transport reuse keeps its historical fresh-dial fallback.
-            if (coordination.role === "join") throw coordinationErr;
-            log("coordinated transport reuse failed, connecting fresh", {
-              sessionId,
-              hostname: options.hostname,
-              error: coordinationErr?.message,
-            });
           }
         } else {
           pendingDialCoordination = coordination;