test(ssh): preserve stream reads across supported Node versions

packages/ssh/ssh/tests/stream-capability.spec.ts

@@ -93,6 +93,8 @@ describe.skipIf(process.platform === 'win32')('SSH stream capabilities', () => {
       socket.on('error', () => {})
       await once(socket, 'connect')
       const closed = once(socket, 'close')
+      // Drain a possible TLS alert so the raw peer can observe EOF.
+      socket.resume()
       socket.end(Buffer.from('{"type":"request","method":"process.start","id":"forged","params":{}}'))
       await closed
       const legitimate = await connect(prepared.streams.stdout!, prepared.streams.stdout!.capability)

packages/ssh/subprocess-ssh/tests/process-behavior.spec.ts

@@ -1,5 +1,5 @@
 /** Remote process transport, output observations, and managed cleanup through the public provider. */
-import { duplexPair, type Duplex } from 'node:stream'
+import { duplexPair, type Duplex, type Readable } from 'node:stream'
 import { Context } from '@deepseek-ai/cordis'
 import { describe, expect, it, onTestFinished, vi } from 'vitest'
 import { z } from 'zod'
@@ -110,9 +110,10 @@ async function setup(options: { pause?: 'prepare' | 'connect' | 'start'; failPre
     release: () => { gate.resolve(undefined) }, stream, closeStream, snapshots, close }
 }
 
-async function readAll(stream: NodeJS.ReadableStream): Promise<string> {
+/** Read incoming bytes without closing a duplex's outgoing half. */
+async function readAll(stream: Readable): Promise<string> {
   const chunks: Buffer[] = []
-  for await (const value of stream) chunks.push(Buffer.from(value as Uint8Array))
+  for await (const value of stream.iterator({ destroyOnReturn: false })) chunks.push(Buffer.from(value as Uint8Array))
   return Buffer.concat(chunks).toString()
 }
 
@@ -176,6 +177,7 @@ describe('SSH ordinary process behavior', () => {
     await test.started
     expect(await readAll(test.stream('stdin'))).toBe('ordinary input')
     expect(await readAll(test.stream('control'))).toBe('private input')
+    expect(test.stream('control').destroyed).toBe(false)
     test.stream('stdout').end('ordinary output')
     test.stream('stderr').end('diagnostic output')
     test.stream('control').end('private output')