fix(ssh): propagate output closure and bound exit observations

packages/ssh/subprocess-ssh/README.i18n.yaml

@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/ssh/subprocess-ssh/README.md
-README.md: a22f237d3c84485c3ae42e8ff86cbcd3550aa77c
-README.zh.md: 993ef5fe6bd1e86a8ab5e712c088470af0dbda7c
+README.md: 018919acd3aa6c898cfd7bcfa11cf3af6cf3b8f8
+README.zh.md: af26dbdd11ee62cb3469d5cfe4133f868eea04ad

packages/ssh/subprocess-ssh/README.md

@@ -29,7 +29,7 @@ Mount this provider with [`dsh-ssh`](../ssh/README.md) and its filesystem provid
 
 Ordinary spawn returns a handle while remote allocation proceeds. Piped stdin and the optional duplex control endpoint accept writes during allocation. Terminal allocation, writes, foreground inspection, signals and termination retain their asynchronous interfaces.
 
-A direct exit reported by `done` does not prove managed-range quiescence; `waitForExit()` observes that separately. `terminate()` targets the same remote owner. Connection loss rejects unconfirmed work, and automatic replay never follows an ambiguous launch or mutation.
+A direct exit reported by `done` does not prove managed-range quiescence; `waitForExit()` observes that separately. Its optional signal bounds the entire observation, including pending allocation and termination, and cancellation returns `false` without stopping the process. An already confirmed empty range returns `true`; genuine observation failures reject. `terminate()` targets the same remote owner. Connection loss rejects unconfirmed work, and automatic replay never follows an ambiguous launch or mutation.
 
 -----
 
@@ -69,7 +69,7 @@ This provider contributes no request-prefix content. Its consumers own model-vis
 <a id="known-limitations-and-deferred-work"></a>
 
 - Completed spill files survive connection disposal and remain available until external temporary-file cleanup.
-- Callers must consume or close raw output streams. Independent channels allow control progress when stdout is paused, but do not remove per-stream backpressure or shared-network congestion.
+- Callers must consume or close raw output streams. Destroying public stdout or stderr closes its matching transport, including when destruction precedes remote allocation. Independent channels allow control progress when stdout is paused, but do not remove per-stream backpressure or shared-network congestion.
 - Loss of the SSH connection leaves remote termination unconfirmed from the client; lease cleanup is a remote action, not a successful client acknowledgement.
 
 <a id="dev-note"></a>

packages/ssh/subprocess-ssh/README.zh.md

@@ -29,7 +29,7 @@ kind: "package-reference"
 
 普通 spawn 在远端分配过程中返回句柄。管道 stdin 和可选双工控制端点在分配过程中接受写入。终端分配、写入、前台检查、信号及终止均保留异步接口。
 
-`done` 报告的直接进程退出并不证明托管进程范围已经静止；`waitForExit()` 单独观察该状态。`terminate()` 针对同一个远端管理器。连接丢失会拒绝未确认的操作；启动或修改结果不明确时绝不自动重放。
+`done` 报告的直接进程退出并不证明托管进程范围已经静止；`waitForExit()` 单独观察该状态。其可选信号约束整个观察过程，包括等待分配和终止；取消返回 `false`，不会停止进程。已确认清空的范围返回 `true`，实际观察故障则拒绝。`terminate()` 针对同一个远端管理器。连接丢失会拒绝未确认的操作；启动或修改结果不明确时绝不自动重放。
 
 -----
 
@@ -69,7 +69,7 @@ kind: "package-reference"
 <a id="known-limitations-and-deferred-work"></a>
 
 - 已完成的 spill 文件在连接释放后继续保留，直至外部临时文件清理。
-- 调用方必须消费或关闭原始输出流。独立通道允许在 stdout 暂停时继续传输控制消息，但不会消除逐流背压或共享网络拥塞。
+- 调用方必须消费或关闭原始输出流。销毁公开的 stdout 或 stderr 会关闭对应传输，即使销毁发生在远端分配之前也是如此。独立通道允许在 stdout 暂停时继续传输控制消息，但不会消除逐流背压或共享网络拥塞。
 - SSH 连接丢失后，客户端无法确认远端终止；租期清理是远端动作，不等于客户端收到了成功确认。
 
 <a id="dev-note"></a>

packages/ssh/subprocess-ssh/src/index.ts

@@ -132,7 +132,12 @@ class RemoteProcess implements SubprocessHandle {
             }))
           } else {
             socket.end()
-            socket.pipe(name === 'stdout' ? this.out : this.err)
+            const output = name === 'stdout' ? this.out : this.err
+            const closeSocket = (): void => { socket.destroy() }
+            output.once('close', closeSocket)
+            socket.once('close', () => { output.off('close', closeSocket) })
+            if (output.destroyed) closeSocket()
+            else socket.pipe(output)
           }
         }
         if (name === 'stdin') this.inbound.pipe(socket)
@@ -172,6 +177,16 @@ class RemoteProcess implements SubprocessHandle {
 
   async waitForExit(signal?: AbortSignal): Promise<boolean> {
     if (this.quiescent) return true
+    if (signal?.aborted) return false
+    if (signal === undefined) return this.observeExit()
+    const cancelled = Promise.withResolvers<boolean>()
+    const abort = (): void => { cancelled.resolve(false) }
+    signal.addEventListener('abort', abort, { once: true })
+    try { return await Promise.race([this.observeExit(signal), cancelled.promise]) }
+    finally { signal.removeEventListener('abort', abort) }
+  }
+
+  private async observeExit(signal?: AbortSignal): Promise<boolean> {
     try { await this.started } catch {
       // Startup errors remain on done; a prepared process must first finish termination.
       if (this.termination !== undefined) await this.termination

packages/ssh/subprocess-ssh/tests/stream-wait-lifecycle.spec.ts

@@ -0,0 +1,190 @@
+/** Caller-owned stream closure and bounded observations across remote allocation. */
+import { duplexPair, type Duplex } from 'node:stream'
+import { getEventListeners } from 'node:events'
+import { Context } from '@deepseek-ai/cordis'
+import { describe, expect, it, onTestFinished } from 'vitest'
+import { z } from 'zod'
+import type { SubprocessSpawnSpec } from '@deepseek-ai/dsh-subprocess'
+import { SshSubprocessRuntime } from '../src/index.ts'
+
+type Stage = 'prepare' | 'connect' | 'start' | 'wait' | 'terminate'
+const id = '8fbfb59c-fbce-44f7-bd87-5ab5a0ea02aa'
+const spec: SubprocessSpawnSpec = {
+  argv: ['node'], cwd: '/workspace', graceMs: 5,
+  stdio: { stdin: 'ignore', stdout: 'pipe', stderr: 'pipe' },
+}
+const finishedValue = { outcome: { exitCode: 0, signal: null }, spills: {}, collected: {} }
+
+async function setup(options: { pause?: Stage; failStart?: Error; failWait?: Error; failTerminate?: Error } = {}) {
+  const ctx = new Context()
+  const gate = Promise.withResolvers<undefined>()
+  const entered = Promise.withResolvers<undefined>()
+  const started = Promise.withResolvers<undefined>()
+  const finished = Promise.withResolvers<typeof finishedValue>()
+  const peers = new Map<string, { host: Duplex; remote: Duplex }>()
+  const calls: string[] = []
+  const errors: unknown[] = []
+  ctx.logger.error = ((error: unknown) => { errors.push(error) }) as typeof ctx.logger.error
+  const stage = async (name: Stage, signal?: AbortSignal) => {
+    if (options.pause !== name) { signal?.throwIfAborted(); return }
+    entered.resolve(undefined)
+    signal?.throwIfAborted()
+    const cancelled = Promise.withResolvers<never>()
+    const abort = (): void => { cancelled.reject(new Error('SSH operation cancelled')) }
+    signal?.addEventListener('abort', abort, { once: true })
+    try { await Promise.race([gate.promise, cancelled.promise]) }
+    finally { signal?.removeEventListener('abort', abort) }
+  }
+  const connection = {
+    request: async <T>(method: string, _params: unknown, schema: z.ZodType<T>, signal?: AbortSignal): Promise<T> => {
+      calls.push(method)
+      let value: unknown
+      if (method === 'process.prepare') {
+        await stage('prepare', signal)
+        value = { id, streams: {
+          stdout: { path: '/tmp/stdout', capability: 'a'.repeat(64) },
+          stderr: { path: '/tmp/stderr', capability: 'b'.repeat(64) },
+        } }
+      } else if (method === 'process.start') {
+        await stage('start', signal)
+        if (options.failStart !== undefined) throw options.failStart
+        started.resolve(undefined)
+        value = {}
+      } else if (method === 'process.done') value = await finished.promise
+      else if (method === 'process.wait') {
+        await stage('wait', signal)
+        if (options.failWait !== undefined) throw options.failWait
+        value = true
+      } else if (method === 'process.terminate') {
+        await stage('terminate', signal)
+        if (options.failTerminate !== undefined) throw options.failTerminate
+        value = null
+      } else throw new Error(`Unexpected request ${method}`)
+      return schema.parse(value)
+    },
+    connectStream: async (endpoint: { path: string }, signal?: AbortSignal) => {
+      await stage('connect', signal)
+      const [host, remote] = duplexPair({ allowHalfOpen: true })
+      host.on('error', () => {})
+      remote.on('error', () => {})
+      peers.set(endpoint.path, { host, remote })
+      return host
+    },
+    dispose: () => Promise.resolve(),
+  }
+  ctx.provide('ssh', connection as never)
+  const fiber = await ctx.plugin(SshSubprocessRuntime)
+  onTestFinished(async () => {
+    gate.resolve(undefined)
+    finished.resolve(finishedValue)
+    for (const peer of peers.values()) { peer.host.destroy(); peer.remote.destroy() }
+    await fiber.dispose()
+    if (options.failTerminate === undefined) expect(errors).toEqual([])
+  })
+  const peer = (name: string) => {
+    const entry = peers.get(`/tmp/${name}`)
+    if (entry === undefined) throw new Error('transport was not allocated')
+    return entry
+  }
+  return { runtime: ctx.subprocess, calls, peer, entered: entered.promise, started: started.promise,
+    release: () => { gate.resolve(undefined) }, finished }
+}
+
+describe('SSH public output streams', () => {
+  it.each(['stdout', 'stderr'] as const)('closes only the matching transport when %s is destroyed', async (name) => {
+    const test = await setup()
+    const handle = test.runtime.spawn(spec)
+    await test.started
+    handle[name]!.destroy()
+    await new Promise(resolve => setImmediate(resolve))
+    expect(test.peer(name).host.destroyed).toBe(true)
+    const other = name === 'stdout' ? 'stderr' : 'stdout'
+    expect(test.peer(other).host.destroyed).toBe(false)
+    expect(test.calls).not.toContain('process.terminate')
+  })
+
+  it.each(['stdout', 'stderr'] as const)('remembers %s closure before remote allocation finishes', async (name) => {
+    const test = await setup({ pause: 'prepare' })
+    const handle = test.runtime.spawn(spec)
+    await test.entered
+    handle[name]!.destroy()
+    test.release()
+    await test.started
+    expect(test.peer(name).host.destroyed).toBe(true)
+    expect(test.peer(name === 'stdout' ? 'stderr' : 'stdout').host.destroyed).toBe(false)
+  })
+})
+
+describe('SSH range-observation cancellation', () => {
+  it.each(['prepare', 'connect', 'start', 'wait', 'terminate'] as const)('bounds the whole observation while %s is pending', async (pause) => {
+    const test = await setup({ pause })
+    const handle = test.runtime.spawn(spec)
+    if (pause === 'terminate') { await test.started; handle.terminate() }
+    const controller = new AbortController()
+    let settled = false
+    const observed = handle.waitForExit(controller.signal).then(
+      value => ({ value }), (error: unknown) => ({ error }),
+    ).finally(() => { settled = true })
+    await test.entered
+    controller.abort('wait deadline')
+    await new Promise(resolve => setImmediate(resolve))
+    expect(settled).toBe(true)
+    expect(await observed).toEqual({ value: false })
+    expect(getEventListeners(controller.signal, 'abort')).toHaveLength(0)
+    if (pause !== 'terminate') expect(test.calls).not.toContain('process.terminate')
+    test.release()
+    expect(await handle.waitForExit()).toBe(true)
+  })
+
+  it('bounds termination after failed startup while retaining real termination failure', async () => {
+    const cleanup = new Error('range owner unavailable')
+    const test = await setup({ pause: 'terminate', failStart: new Error('start refused'), failTerminate: cleanup })
+    const handle = test.runtime.spawn(spec)
+    const controller = new AbortController()
+    let settled = false
+    const observed = handle.waitForExit(controller.signal).then(
+      value => ({ value }), (error: unknown) => ({ error }),
+    ).finally(() => { settled = true })
+    await test.entered
+    controller.abort()
+    await new Promise(resolve => setImmediate(resolve))
+    expect(settled).toBe(true)
+    expect(await observed).toEqual({ value: false })
+    test.release()
+    await expect(handle.waitForExit()).rejects.toBe(cleanup)
+  })
+
+  it('returns false for a pre-aborted live observation and true once quiescence is known', async () => {
+    const test = await setup({ pause: 'prepare' })
+    const handle = test.runtime.spawn(spec)
+    const signal = AbortSignal.abort()
+    let settled = false
+    const observed = handle.waitForExit(signal).then(
+      value => ({ value }), (error: unknown) => ({ error }),
+    ).finally(() => { settled = true })
+    await new Promise(resolve => setImmediate(resolve))
+    expect(settled).toBe(true)
+    expect(await observed).toEqual({ value: false })
+    test.release()
+    expect(await handle.waitForExit()).toBe(true)
+    expect(await handle.waitForExit(signal)).toBe(true)
+  })
+
+  it('preserves real observation failure and detaches the caller cancellation listener', async () => {
+    const failure = new Error('remote range observation failed')
+    const test = await setup({ failWait: failure })
+    const handle = test.runtime.spawn(spec)
+    await test.started
+    const controller = new AbortController()
+    await expect(handle.waitForExit(controller.signal)).rejects.toBe(failure)
+    expect(getEventListeners(controller.signal, 'abort')).toHaveLength(0)
+  })
+
+  it('returns true when an uncancelled bounded observation completes', async () => {
+    const test = await setup()
+    const handle = test.runtime.spawn(spec)
+    const controller = new AbortController()
+    expect(await handle.waitForExit(controller.signal)).toBe(true)
+    expect(getEventListeners(controller.signal, 'abort')).toHaveLength(0)
+  })
+})